Line data Source code
1 : /* _GNU_SOURCE for recvmmsg and sendmmsg */
2 : #define _GNU_SOURCE
3 :
4 : #include "../../../../disco/metrics/fd_metrics.h"
5 : #include "../../../../disco/fd_clock_tile.h"
6 : #include "../../../../disco/topo/fd_topo.h"
7 : #include "../../../../waltz/quic/fd_quic.h"
8 : #include "../../../../waltz/quic/tests/fd_quic_test_helpers.h"
9 : #include "../../../../waltz/tls/test_tls_helper.h"
10 :
11 : #include <errno.h>
12 : #include <sys/types.h>
13 : #include <sys/socket.h>
14 : #include <netinet/in.h>
15 : #include <string.h>
16 : #include <unistd.h>
17 : #include <poll.h>
18 :
19 : #include <stdio.h>
20 : #include <stdlib.h>
21 : #include <time.h>
22 :
23 : /* max number of buffers batched for receive */
24 0 : #define IO_VEC_CNT 128
25 :
26 : /* longest time a transaction waits for its QUIC packet to fill */
27 0 : #define QUIC_COALESCE_NS (100000L)
28 :
29 : static int
30 : quic_tx_aio_send( void * _ctx,
31 : fd_aio_pkt_info_t const * batch,
32 : ulong batch_cnt,
33 : ulong * opt_batch_idx,
34 : int flush );
35 :
36 : typedef struct {
37 : ulong round_robin_cnt;
38 : ulong round_robin_id;
39 :
40 : ulong packet_cnt;
41 :
42 : ulong conn_cnt;
43 : int conn_fd[ 128UL ];
44 : struct pollfd poll_fd[ 128UL ];
45 :
46 : fd_tls_test_sign_ctx_t test_signer[1];
47 : int no_quic;
48 : fd_quic_t * quic;
49 : uint quic_ip;
50 : ushort quic_port;
51 : fd_quic_conn_t * quic_conn;
52 : ulong stream_id; /* stream of the next transaction */
53 : long flush_ts; /* seal the packet being filled at this time */
54 : uchar stream_pkt[ 2048 ]; /* packet being filled with transactions */
55 : ulong stream_used;
56 : ulong dropped_cnt;
57 : uint service_ratio_idx;
58 : fd_aio_t tx_aio;
59 :
60 : long now; /* current time in ns */
61 : fd_clock_tile_t clock[1];
62 :
63 : /* vector receive members */
64 : struct mmsghdr rx_msgs[IO_VEC_CNT];
65 : struct mmsghdr tx_msgs[IO_VEC_CNT];
66 : struct iovec rx_iovecs[IO_VEC_CNT];
67 : struct iovec tx_iovecs[IO_VEC_CNT];
68 : uchar rx_bufs[IO_VEC_CNT][2048];
69 : uchar tx_bufs[IO_VEC_CNT][2048];
70 :
71 : ulong tx_idx;
72 :
73 : fd_wksp_t * mem;
74 : } fd_benchs_ctx_t;
75 :
76 : static void
77 : service_quic( fd_benchs_ctx_t * ctx,
78 0 : long now ) {
79 :
80 0 : if( !ctx->no_quic ) {
81 : /* Publishes to mcache via callbacks */
82 :
83 : /* receive from socket, and pass to quic */
84 0 : int poll_rc = poll( ctx->poll_fd, ctx->conn_cnt, 0 );
85 0 : if( FD_LIKELY( poll_rc == 0 ) ) {
86 0 : return;
87 0 : } if( FD_UNLIKELY( poll_rc == -1 ) ) {
88 0 : if( FD_UNLIKELY( errno == EINTR ) ) return; /* will try later */
89 0 : FD_LOG_ERR(( "Error occurred during poll: %d %s", errno,
90 0 : strerror( errno ) ));
91 0 : }
92 :
93 0 : for( ulong j = 0; j < ctx->conn_cnt; ++j ) {
94 0 : int revents = ctx->poll_fd[j].revents;
95 0 : if( FD_LIKELY( revents & POLLIN ) ) {
96 : /* data available - receive up to IO_VEC_CNT buffers */
97 0 : struct timespec timeout = {0};
98 0 : int retval = recvmmsg( ctx->poll_fd[j].fd, ctx->rx_msgs, IO_VEC_CNT, 0, &timeout );
99 0 : if( FD_UNLIKELY( retval < 0 ) ) {
100 0 : FD_LOG_ERR(( "Error occurred on recvmmsg: %d %s", errno, strerror( errno ) ));
101 0 : }
102 : /* pass buffers to QUIC */
103 0 : for( ulong k = 0; k < (ulong)retval; k++ ) {
104 0 : uchar * buf = ctx->rx_bufs[k];
105 :
106 : /* set some required values */
107 0 : uint payload_len = ctx->rx_msgs[k].msg_len;
108 0 : uint udp_len = payload_len + 8;
109 0 : uint ip_len = udp_len + 20;
110 :
111 : /* set ver and len */
112 0 : buf[0] = 0x45;
113 :
114 : /* set protocol */
115 0 : buf[9] = 17;
116 :
117 : /* set udp length */
118 0 : buf[20 + 4] = (uchar)( udp_len >> 8 );
119 0 : buf[20 + 5] = (uchar)( udp_len );
120 :
121 : /* set ip length */
122 0 : buf[2] = (uchar)( ip_len >> 8 );
123 0 : buf[3] = (uchar)( ip_len );
124 :
125 : /* set src ip addr */
126 0 : buf[12] = (uchar)( ctx->quic_ip );
127 0 : buf[13] = (uchar)( ctx->quic_ip >> 8 );
128 0 : buf[14] = (uchar)( ctx->quic_ip >> 16 );
129 0 : buf[15] = (uchar)( ctx->quic_ip >> 24 );
130 :
131 0 : fd_quic_process_packet( ctx->quic, buf, ip_len, now );
132 0 : }
133 0 : } else if( FD_UNLIKELY( revents & POLLERR ) ) {
134 0 : int error = 0;
135 0 : socklen_t errlen = sizeof(error);
136 :
137 0 : if( getsockopt( ctx->poll_fd[j].fd, SOL_SOCKET, SO_ERROR, (void *)&error, &errlen ) == -1 ) {
138 0 : FD_LOG_ERR(( "Unknown error on socket" ));
139 0 : } else {
140 0 : FD_LOG_ERR(( "Error on socket: %d %s", error, strerror( error ) ));
141 0 : }
142 0 : }
143 0 : }
144 0 : }
145 0 : }
146 :
147 : /* quic_conn_new is invoked by the QUIC engine whenever a new connection
148 : is being established. */
149 : static void
150 : quic_conn_new( fd_quic_conn_t * conn,
151 0 : void * _ctx ) {
152 0 : (void)conn;
153 0 : (void)_ctx;
154 0 : }
155 :
156 :
157 : static void
158 : handshake_complete( fd_quic_conn_t * conn,
159 0 : void * _ctx ) {
160 0 : (void)conn;
161 0 : (void)_ctx;
162 0 : FD_LOG_NOTICE(( "client handshake complete" ));
163 0 : }
164 :
165 : static void
166 : conn_final( fd_quic_conn_t * conn,
167 0 : void * _ctx ) {
168 0 : (void)conn;
169 :
170 0 : fd_benchs_ctx_t * ctx = (fd_benchs_ctx_t *)_ctx;
171 :
172 0 : if( ctx ) {
173 0 : ctx->quic_conn = NULL;
174 0 : }
175 0 : }
176 :
177 : FD_FN_CONST static inline ulong
178 0 : scratch_align( void ) {
179 0 : return fd_ulong_max( fd_quic_align(), alignof( fd_benchs_ctx_t ) );
180 0 : }
181 :
182 : static void
183 0 : populate_quic_limits( fd_quic_limits_t * limits ) {
184 0 : limits->conn_cnt = 2;
185 0 : limits->handshake_cnt = limits->conn_cnt;
186 0 : limits->conn_id_cnt = 16;
187 0 : limits->inflight_frame_cnt = 1UL<<14;
188 0 : }
189 :
190 : static void
191 0 : populate_quic_config( fd_quic_config_t * config ) {
192 0 : config->role = FD_QUIC_ROLE_CLIENT;
193 0 : config->retry = 0;
194 0 : config->initial_rx_max_stream_data = 0; /* we don't expect the server to initiate streams */
195 0 : config->net.dscp = 0;
196 0 : }
197 :
198 : static inline ulong
199 0 : scratch_footprint( fd_topo_tile_t const * tile ) {
200 0 : ulong l = FD_LAYOUT_INIT;
201 0 : l = FD_LAYOUT_APPEND( l, alignof( fd_benchs_ctx_t ), sizeof( fd_benchs_ctx_t ) );
202 0 : if( !tile->benchs.no_quic ) {
203 0 : fd_quic_limits_t quic_limits = {0};
204 0 : populate_quic_limits( &quic_limits );
205 0 : ulong quic_fp = fd_quic_footprint( &quic_limits );
206 0 : l = FD_LAYOUT_APPEND( l, fd_quic_align(), quic_fp );
207 0 : }
208 0 : return FD_LAYOUT_FINI( l, scratch_align() );
209 0 : }
210 :
211 : static inline void
212 0 : metrics_write( fd_benchs_ctx_t * ctx ) {
213 0 : FD_MCNT_SET( BENCHS, TXN_TX, ctx->packet_cnt );
214 0 : FD_MCNT_SET( BENCHS, TXN_DROPPED, ctx->dropped_cnt );
215 0 : }
216 :
217 : static void
218 0 : quic_tx_send( fd_benchs_ctx_t * ctx ) {
219 0 : if( FD_LIKELY( ctx->tx_idx ) ) {
220 0 : int rtn = sendmmsg( ctx->conn_fd[0], ctx->tx_msgs, (uint)ctx->tx_idx, 0 );
221 0 : if( FD_UNLIKELY( rtn < 0 ) ) {
222 0 : FD_LOG_NOTICE(( "Error occurred in sendmmsg. Error: %d %s",
223 0 : errno, strerror( errno ) ));
224 0 : }
225 0 : ctx->tx_idx = 0;
226 0 : }
227 0 : }
228 :
229 : /* quic_tx_push queues the first sz bytes of tx_bufs[ tx_idx ] for
230 : sending. */
231 :
232 : static void
233 : quic_tx_push( fd_benchs_ctx_t * ctx,
234 0 : ulong sz ) {
235 0 : ulong tx_idx = ctx->tx_idx;
236 0 : ctx->tx_iovecs[tx_idx] = (struct iovec) {
237 0 : .iov_base = ctx->tx_bufs[tx_idx],
238 0 : .iov_len = sz
239 0 : };
240 0 : ctx->tx_msgs[tx_idx] = (struct mmsghdr) {
241 0 : .msg_hdr = {
242 0 : .msg_iov = &ctx->tx_iovecs[tx_idx],
243 0 : .msg_iovlen = 1,
244 0 : }
245 0 : };
246 0 : ctx->tx_idx = tx_idx+1UL;
247 0 : }
248 :
249 : /* quic_tx_push_sealed queues the sealed packet in stream_pkt for
250 : sending. */
251 :
252 : static void
253 : quic_tx_push_sealed( fd_benchs_ctx_t * ctx,
254 0 : ulong pkt_sz ) {
255 0 : if( FD_UNLIKELY( ctx->tx_idx==IO_VEC_CNT ) ) quic_tx_send( ctx );
256 0 : fd_memcpy( ctx->tx_bufs[ ctx->tx_idx ], ctx->stream_pkt, pkt_sz );
257 0 : quic_tx_push( ctx, pkt_sz );
258 0 : }
259 :
260 : /* quic_tx_flush seals the packet of coalesced transactions and sends
261 : all queued packets. */
262 :
263 : static void
264 0 : quic_tx_flush( fd_benchs_ctx_t * ctx ) {
265 0 : if( FD_LIKELY( ctx->quic_conn ) ) {
266 0 : ulong pkt_sz = fd_quic_conn_tx_stream( ctx->quic_conn, ctx->stream_pkt, sizeof(ctx->stream_pkt), &ctx->stream_used, ULONG_MAX, NULL, 0UL, NULL );
267 0 : if( FD_LIKELY( pkt_sz && pkt_sz!=ULONG_MAX ) ) quic_tx_push_sealed( ctx, pkt_sz );
268 0 : }
269 0 : quic_tx_send( ctx );
270 0 : ctx->flush_ts = LONG_MAX;
271 0 : }
272 :
273 : static inline int
274 : before_frag( fd_benchs_ctx_t * ctx,
275 : ulong in_idx,
276 : ulong seq,
277 0 : ulong sig ) {
278 0 : (void)in_idx;
279 0 : (void)sig;
280 :
281 0 : ctx->now = fd_clock_tile_now( ctx->clock );
282 :
283 0 : return (int)( (seq%ctx->round_robin_cnt)!=ctx->round_robin_id );
284 0 : }
285 :
286 : static inline void
287 : during_frag( fd_benchs_ctx_t * ctx,
288 : ulong in_idx FD_PARAM_UNUSED,
289 : ulong seq FD_PARAM_UNUSED,
290 : ulong sig FD_PARAM_UNUSED,
291 : ulong chunk,
292 : ulong sz,
293 0 : ulong ctl FD_PARAM_UNUSED ) {
294 0 : if( ctx->no_quic ) {
295 :
296 0 : if( FD_UNLIKELY( -1==send( ctx->conn_fd[ ctx->packet_cnt % ctx->conn_cnt ], fd_chunk_to_laddr( ctx->mem, chunk ), sz, 0 ) ) )
297 0 : FD_LOG_ERR(( "send() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
298 :
299 0 : ctx->packet_cnt++;
300 0 : } else {
301 : /* allows to accumulate multiple transactions before creating a UDP datagram */
302 : /* make this configurable */
303 0 : if( FD_UNLIKELY( ctx->service_ratio_idx++ == 8 ) ) {
304 0 : ctx->service_ratio_idx = 0;
305 0 : quic_tx_send( ctx );
306 0 : service_quic( ctx, ctx->now );
307 0 : fd_quic_service( ctx->quic, ctx->now );
308 0 : }
309 :
310 0 : if( FD_UNLIKELY( !ctx->quic_conn ) ) {
311 0 : ctx->dropped_cnt++;
312 :
313 : /* try to connect */
314 0 : uint dest_ip = ctx->quic_ip;
315 0 : ushort dest_port = fd_ushort_bswap( ctx->quic_port );
316 :
317 0 : ctx->quic_conn = fd_quic_connect( ctx->quic, dest_ip, dest_port, 0U, 12000, ctx->now );
318 :
319 : /* failed? try later */
320 0 : if( FD_UNLIKELY( !ctx->quic_conn ) ) {
321 0 : service_quic( ctx, ctx->now );
322 0 : fd_quic_service( ctx->quic, ctx->now );
323 0 : return;
324 0 : }
325 :
326 0 : FD_LOG_NOTICE(( "connection created on port %d", (int)dest_port ));
327 0 : ctx->stream_id = FD_QUIC_STREAM_TYPE_UNI_CLIENT;
328 0 : ctx->stream_used = 0UL;
329 :
330 : /* set the context to point to the location
331 : of the quic_conn pointer
332 : this allows the notification to NULL the value when
333 : a connection dies */
334 0 : fd_quic_conn_set_context( ctx->quic_conn, ctx );
335 :
336 0 : service_quic( ctx, ctx->now );
337 0 : fd_quic_service( ctx->quic, ctx->now );
338 :
339 : /* conn may be invalidated by fd_quic_service */
340 :
341 0 : return;
342 0 : }
343 :
344 0 : if( FD_UNLIKELY( sz>FD_QUIC_STREAM_TX_MAX ) ) FD_LOG_ERR(( "transaction too large for one QUIC packet (%lu bytes, max %lu)", sz, FD_QUIC_STREAM_TX_MAX ));
345 0 : uchar const * txn = fd_chunk_to_laddr( ctx->mem, chunk );
346 0 : ulong pkt_sz = fd_quic_conn_tx_stream( ctx->quic_conn, ctx->stream_pkt, sizeof(ctx->stream_pkt), &ctx->stream_used, ctx->stream_id, txn, sz, NULL );
347 0 : if( pkt_sz && pkt_sz!=ULONG_MAX ) {
348 0 : quic_tx_push_sealed( ctx, pkt_sz );
349 0 : ctx->flush_ts = LONG_MAX;
350 0 : pkt_sz = fd_quic_conn_tx_stream( ctx->quic_conn, ctx->stream_pkt, sizeof(ctx->stream_pkt), &ctx->stream_used, ctx->stream_id, txn, sz, NULL );
351 0 : }
352 0 : if( FD_LIKELY( !pkt_sz ) ) {
353 0 : if( ctx->flush_ts==LONG_MAX ) ctx->flush_ts = ctx->now + QUIC_COALESCE_NS;
354 0 : ctx->stream_id += 4UL;
355 0 : ctx->packet_cnt++;
356 0 : } else {
357 0 : ctx->dropped_cnt++;
358 0 : quic_tx_flush( ctx );
359 0 : service_quic( ctx, ctx->now );
360 0 : fd_quic_service( ctx->quic, ctx->now );
361 0 : }
362 0 : }
363 0 : }
364 :
365 : static void
366 : privileged_init( fd_topo_t const * topo,
367 0 : fd_topo_tile_t const * tile ) {
368 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
369 :
370 : /* call wallclock so glibc loads VDSO, which requires calling mmap while
371 : privileged */
372 0 : fd_log_wallclock();
373 :
374 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
375 0 : fd_benchs_ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof( fd_benchs_ctx_t ), sizeof( fd_benchs_ctx_t ) );
376 0 : fd_memset( ctx, 0, sizeof(fd_benchs_ctx_t) );
377 :
378 0 : int no_quic = ctx->no_quic = tile->benchs.no_quic;
379 0 : ushort port = 12000;
380 :
381 0 : ulong quic_cnt = tile->benchs.conn_cnt; /* [layout] quic_tile_count */
382 0 : ctx->conn_cnt = quic_cnt;
383 0 : if( !no_quic ) ctx->conn_cnt = 1;
384 0 : FD_TEST( ctx->conn_cnt <=sizeof(ctx->conn_fd)/sizeof(*ctx->conn_fd) );
385 0 : ctx->quic_ip = tile->benchs.send_to_ip_addr;
386 0 : ctx->quic_port = tile->benchs.send_to_port;
387 0 : for( ulong i=0UL; i<ctx->conn_cnt ; i++ ) {
388 0 : int conn_fd = socket( AF_INET, SOCK_DGRAM, 0 );
389 0 : if( FD_UNLIKELY( -1==conn_fd ) ) FD_LOG_ERR(( "socket() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
390 :
391 0 : int recvbuff = 8<<20;
392 :
393 : // Set the buffer size
394 0 : if( setsockopt( conn_fd, SOL_SOCKET, SO_RCVBUF, &recvbuff, sizeof(recvbuff) ) < 0 ) {
395 0 : FD_LOG_ERR(( "Error setting receive buffer size. Error: %d %s", errno, strerror( errno ) ));
396 0 : }
397 :
398 0 : int sendbuff = 8<<20;
399 0 : if( setsockopt( conn_fd, SOL_SOCKET, SO_SNDBUF, &sendbuff, sizeof(sendbuff) ) < 0 ) {
400 0 : FD_LOG_ERR(( "Error setting transmit buffer size. Error: %d %s", errno, strerror( errno ) ));
401 0 : }
402 :
403 0 : ulong quic_idx = (tile->kind_id*ctx->conn_cnt + i)%quic_cnt;
404 0 : ushort found_port = 0;
405 0 : for( ; port<65535; port++ ) {
406 0 : if( fd_disco_netmux_sig_hash( fd_disco_netmux_sig( ctx->quic_ip, port, 0U, 0UL, 42UL ) )%quic_cnt!=quic_idx ) continue;
407 0 : struct sockaddr_in addr = {
408 0 : .sin_family = AF_INET,
409 0 : .sin_port = fd_ushort_bswap( port ),
410 0 : .sin_addr.s_addr = fd_uint_bswap( INADDR_ANY ),
411 0 : };
412 0 : if( FD_UNLIKELY( -1!=bind( conn_fd, fd_type_pun( &addr ), sizeof(addr) ) ) ) {
413 0 : found_port = port;
414 0 : break;
415 0 : }
416 0 : if( FD_UNLIKELY( EADDRINUSE!=errno ) ) FD_LOG_ERR(( "bind() failed (%i-%s)", errno, fd_io_strerror( errno ) ) );
417 0 : }
418 0 : if( FD_UNLIKELY( !found_port ) ) FD_LOG_ERR(( "bind() failed to find a src port" ));
419 0 : FD_LOG_INFO(( "src port %hu -> quic tile %lu", found_port, quic_idx ));
420 :
421 0 : struct sockaddr_in addr = {
422 0 : .sin_family = AF_INET,
423 0 : .sin_port = fd_ushort_bswap( tile->benchs.send_to_port ),
424 0 : .sin_addr.s_addr = tile->benchs.send_to_ip_addr,
425 0 : };
426 0 : if( FD_UNLIKELY( -1==connect( conn_fd, fd_type_pun( &addr ), sizeof(addr) ) ) ) FD_LOG_ERR(( "connect() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
427 :
428 0 : ctx->conn_fd[ i ] = conn_fd;
429 0 : if( !no_quic ) {
430 0 : ctx->poll_fd[i].fd = conn_fd;
431 0 : ctx->poll_fd[i].events = POLLIN;
432 0 : }
433 0 : port++;
434 0 : }
435 0 : }
436 :
437 : static void
438 : unprivileged_init( fd_topo_t const * topo,
439 0 : fd_topo_tile_t const * tile ) {
440 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
441 :
442 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
443 0 : fd_benchs_ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof( fd_benchs_ctx_t ), sizeof( fd_benchs_ctx_t ) );
444 :
445 0 : ctx->packet_cnt = 0UL;
446 :
447 0 : ctx->round_robin_id = tile->kind_id;
448 0 : ctx->round_robin_cnt = fd_topo_tile_name_cnt( topo, "benchs" );
449 :
450 0 : ctx->mem = topo->workspaces[ topo->objs[ topo->links[ tile->in_link_id[ 0UL ] ].dcache_obj_id ].wksp_id ].wksp;
451 :
452 0 : if( !ctx->no_quic ) {
453 0 : fd_quic_limits_t quic_limits = {0};
454 0 : populate_quic_limits( &quic_limits );
455 :
456 0 : ulong quic_fp = fd_quic_footprint( &quic_limits );
457 0 : if( FD_UNLIKELY( !quic_fp ) ) FD_LOG_ERR(( "invalid QUIC parameters" ));
458 0 : void * quic_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_quic_align(), quic_fp );
459 0 : fd_quic_t * quic = fd_quic_join( fd_quic_new( quic_mem, &quic_limits ) );
460 :
461 0 : populate_quic_config( &quic->config );
462 :
463 : /* FIXME this always results in the same private key */
464 0 : fd_rng_t _rng[1];
465 0 : fd_rng_t * rng = fd_rng_join( fd_rng_new( _rng, 4242424242, 0UL ) );
466 0 : fd_tls_test_sign_ctx( ctx->test_signer, rng );
467 0 : fd_quic_config_test_signer( quic, ctx->test_signer );
468 :
469 0 : ctx->quic = quic;
470 0 : ctx->tx_idx = 0UL;
471 0 : ctx->flush_ts = LONG_MAX;
472 :
473 0 : fd_aio_t * quic_tx_aio = fd_aio_join( fd_aio_new( &ctx->tx_aio, ctx, quic_tx_aio_send ) );
474 0 : if( FD_UNLIKELY( !quic_tx_aio ) ) FD_LOG_ERR(( "fd_aio_join failed" ));
475 :
476 0 : ulong quic_idle_timeout_millis = 10000; /* idle timeout in milliseconds */
477 0 : quic->config.role = FD_QUIC_ROLE_CLIENT;
478 0 : quic->config.idle_timeout = (long)( quic_idle_timeout_millis * 1000000L );
479 0 : quic->config.initial_rx_max_stream_data = 0;
480 0 : quic->config.retry = 0; /* unused on clients */
481 :
482 0 : quic->cb.conn_new = quic_conn_new;
483 0 : quic->cb.conn_hs_complete = handshake_complete;
484 0 : quic->cb.conn_final = conn_final;
485 0 : quic->cb.quic_ctx = ctx;
486 :
487 0 : fd_quic_set_aio_net_tx( quic, quic_tx_aio );
488 0 : if( FD_UNLIKELY( !fd_quic_init( quic ) ) ) FD_LOG_ERR(( "fd_quic_init failed" ));
489 :
490 0 : ulong hdr_sz = 20 + 8;
491 0 : for( ulong i = 0; i < IO_VEC_CNT; i++ ) {
492 : /* leave space for headers */
493 0 : ctx->rx_iovecs[i] = (struct iovec) {
494 0 : .iov_base = ctx->rx_bufs[i] + hdr_sz,
495 0 : .iov_len = sizeof(ctx->rx_bufs[i]) - hdr_sz
496 0 : };
497 0 : ctx->rx_msgs[i] = (struct mmsghdr) {
498 0 : .msg_hdr = {
499 0 : .msg_iov = &ctx->rx_iovecs[i],
500 0 : .msg_iovlen = 1
501 0 : }
502 0 : };
503 0 : }
504 0 : }
505 :
506 0 : ulong scratch_top = FD_SCRATCH_ALLOC_FINI( l, 1UL );
507 0 : if( FD_UNLIKELY( scratch_top > (ulong)scratch + scratch_footprint( tile ) ) )
508 0 : FD_LOG_ERR(( "scratch overflow %lu %lu %lu", scratch_top - (ulong)scratch - scratch_footprint( tile ), scratch_top, (ulong)scratch + scratch_footprint( tile ) ));
509 :
510 0 : fd_clock_tile_t * clock = ctx->clock;
511 0 : fd_clock_tile_init( clock );
512 0 : ctx->now = fd_clock_tile_now( clock );
513 0 : }
514 :
515 : static int
516 : quic_tx_aio_send( void * _ctx,
517 : fd_aio_pkt_info_t const * batch,
518 : ulong batch_cnt,
519 : ulong * opt_batch_idx,
520 0 : int flush ) {
521 0 : fd_benchs_ctx_t * ctx = _ctx;
522 :
523 : /* quic adds ip and udp headers which we don't need */
524 : /* assume 20 + 8 for those */
525 0 : ulong hdr_sz = 20+8;
526 :
527 0 : for( ulong j = 0; j < batch_cnt; ++j ) {
528 0 : if( FD_UNLIKELY( batch[j].buf_sz < hdr_sz ) ) continue;
529 0 : if( FD_UNLIKELY( ctx->tx_idx == IO_VEC_CNT ) ) quic_tx_send( ctx );
530 :
531 : /* copy, stripping the header */
532 0 : fd_memcpy( ctx->tx_bufs[ ctx->tx_idx ], (uchar*)batch[j].buf + hdr_sz, batch[j].buf_sz - hdr_sz );
533 0 : quic_tx_push( ctx, batch[j].buf_sz - hdr_sz );
534 0 : }
535 :
536 0 : if( FD_UNLIKELY( flush ) ) quic_tx_send( ctx );
537 :
538 0 : if( FD_LIKELY( opt_batch_idx ) ) *opt_batch_idx = batch_cnt;
539 :
540 0 : return 0;
541 0 : }
542 :
543 : static void
544 0 : during_housekeeping( fd_benchs_ctx_t * ctx ) {
545 0 : if( !ctx->no_quic && fd_clock_tile_now( ctx->clock )>=ctx->flush_ts ) quic_tx_flush( ctx );
546 0 : if( FD_UNLIKELY( fd_clock_tile_recal_due( ctx->clock ) ) ) {
547 0 : fd_clock_tile_recal( ctx->clock );
548 0 : }
549 0 : }
550 :
551 0 : #define STEM_BURST (1UL)
552 :
553 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_benchs_ctx_t
554 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_benchs_ctx_t)
555 :
556 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
557 0 : #define STEM_CALLBACK_BEFORE_FRAG before_frag
558 0 : #define STEM_CALLBACK_DURING_FRAG during_frag
559 0 : #define STEM_CALLBACK_DURING_HOUSEKEEPING during_housekeeping
560 :
561 : #include "../../../../disco/stem/fd_stem.c"
562 :
563 : fd_topo_run_tile_t fd_tile_benchs = {
564 : .name = "benchs",
565 : .scratch_align = scratch_align,
566 : .scratch_footprint = scratch_footprint,
567 : .privileged_init = privileged_init,
568 : .unprivileged_init = unprivileged_init,
569 : .run = stem_run,
570 : };
|
