Line data Source code
1 : /* _GNU_SOURCE for recvmmsg and sendmmsg */
2 : #define _GNU_SOURCE
3 :
4 : #include "../../../disco/tiles.h"
5 : #include "../../../waltz/xdp/fd_xsk_aio.h"
6 : #include "../../../waltz/quic/fd_quic.h"
7 : #include "../../../waltz/tls/test_tls_helper.h"
8 :
9 : #include <errno.h>
10 : #include <linux/unistd.h>
11 : #include <sys/types.h>
12 : #include <sys/socket.h>
13 : #include <netinet/in.h>
14 : #include <string.h>
15 : #include <unistd.h>
16 : #include <poll.h>
17 :
18 : #include <stdio.h>
19 : #include <stdlib.h>
20 :
21 : #include <time.h>
22 :
23 : /* max number of buffers batched for receive */
24 0 : #define IO_VEC_CNT 128
25 :
26 :
27 : struct signer_ctx {
28 : fd_sha512_t sha512[ 1 ];
29 :
30 : uchar public_key[ 32UL ];
31 : uchar private_key[ 32UL ];
32 : };
33 : typedef struct signer_ctx signer_ctx_t;
34 :
35 :
36 : static void
37 : signer( void * _ctx,
38 : uchar signature[ static 64 ],
39 0 : uchar const payload[ static 130 ] ) {
40 0 : fd_tls_test_sign_ctx_t * ctx = (fd_tls_test_sign_ctx_t *)_ctx;
41 0 : fd_ed25519_sign( signature, payload, 130UL, ctx->public_key, ctx->private_key, ctx->sha512 );
42 0 : }
43 :
44 : static FD_FN_UNUSED
45 : signer_ctx_t
46 0 : signer_ctx( fd_rng_t * rng ) {
47 0 : signer_ctx_t ctx[1];
48 0 : FD_TEST( fd_sha512_join( fd_sha512_new( ctx->sha512 ) ) );
49 0 : for( ulong b=0; b<32UL; b++ ) ctx->private_key[b] = fd_rng_uchar( rng );
50 0 : fd_ed25519_public_from_private( ctx->public_key, ctx->private_key, ctx->sha512 );
51 :
52 0 : return *ctx;
53 0 : }
54 :
55 : static int
56 : quic_tx_aio_send( void * _ctx,
57 : fd_aio_pkt_info_t const * batch,
58 : ulong batch_cnt,
59 : ulong * opt_batch_idx,
60 : int flush );
61 :
62 :
63 : /* quic_now is called by the QUIC engine to get the current timestamp in
64 : UNIX time. */
65 :
66 : static ulong
67 0 : quic_now( void * ctx ) {
68 0 : (void)ctx;
69 0 : return (ulong)fd_log_wallclock();
70 0 : }
71 :
72 : typedef struct {
73 : ulong round_robin_cnt;
74 : ulong round_robin_id;
75 :
76 : ulong packet_cnt;
77 :
78 : ulong conn_cnt;
79 : int conn_fd[ 128UL ];
80 : struct pollfd poll_fd[ 128UL ];
81 :
82 : signer_ctx_t signer_ctx;
83 : int no_quic;
84 : fd_quic_t * quic;
85 : ushort quic_port;
86 : fd_quic_conn_t * quic_conn;
87 : const fd_aio_t * quic_rx_aio;
88 : ulong no_stream;
89 : uint service_ratio_idx;
90 :
91 : // vector receive members
92 : struct mmsghdr rx_msgs[IO_VEC_CNT];
93 : struct mmsghdr tx_msgs[IO_VEC_CNT];
94 : struct iovec rx_iovecs[IO_VEC_CNT];
95 : struct iovec tx_iovecs[IO_VEC_CNT];
96 : char rx_bufs[IO_VEC_CNT][2048];
97 : char tx_bufs[IO_VEC_CNT][2048];
98 :
99 : ulong tx_idx;
100 :
101 : fd_quic_stream_t * stream;
102 :
103 : fd_wksp_t * mem;
104 : } fd_benchs_ctx_t;
105 :
106 : void
107 0 : service_quic( fd_benchs_ctx_t * ctx ) {
108 :
109 0 : if( !ctx->no_quic ) {
110 : /* Publishes to mcache via callbacks */
111 :
112 : /* receive from socket, and pass to quic */
113 0 : int poll_rc = poll( ctx->poll_fd, ctx->conn_cnt, 0 );
114 0 : if( FD_LIKELY( poll_rc == 0 ) ) {
115 0 : return;
116 0 : } if( FD_UNLIKELY( poll_rc == -1 ) ) {
117 0 : if( FD_UNLIKELY( errno == EINTR ) ) return; /* will try later */
118 0 : FD_LOG_ERR(( "Error occurred during poll: %d %s", errno,
119 0 : strerror( errno ) ));
120 0 : }
121 :
122 0 : for( ulong j = 0; j < ctx->conn_cnt; ++j ) {
123 0 : int revents = ctx->poll_fd[j].revents;
124 0 : if( FD_LIKELY( revents & POLLIN ) ) {
125 : /* data available - receive up to IO_VEC_CNT buffers */
126 0 : struct timespec timeout = {0};
127 0 : int retval = recvmmsg( ctx->poll_fd[j].fd, ctx->rx_msgs, IO_VEC_CNT, 0, &timeout );
128 0 : if( FD_UNLIKELY( retval < 0 ) ) {
129 0 : FD_LOG_ERR(( "Error occurred on recvmmsg: %d %s", errno, strerror( errno ) ));
130 0 : } else {
131 : /* pass buffers to QUIC */
132 0 : fd_aio_pkt_info_t pkt[IO_VEC_CNT];
133 0 : ulong hdr_sz = 14 + 20 + 8;
134 0 : for( ulong j = 0; j < (ulong)retval; ++j ) {
135 0 : pkt[j].buf = ctx->rx_bufs[j];
136 0 : pkt[j].buf_sz = (ushort)( ctx->rx_msgs[j].msg_len + hdr_sz );
137 :
138 : /* set some required values */
139 0 : uint payload_len = ctx->rx_msgs[j].msg_len;
140 0 : uint udp_len = payload_len + 8;
141 0 : uint ip_len = udp_len + 20;
142 :
143 0 : uchar * buf = (uchar*)pkt[j].buf;
144 :
145 : /* set ethtype */
146 0 : buf[12 + 0] = 0x08;
147 0 : buf[12 + 1] = 0x00;
148 :
149 : /* set ver and len */
150 0 : buf[14] = 0x45;
151 :
152 : /* set protocol */
153 0 : buf[14 + 9] = 17;
154 :
155 : /* set udp length */
156 0 : buf[14 + 20 + 4] = (uchar)( udp_len >> 8 );
157 0 : buf[14 + 20 + 5] = (uchar)( udp_len );
158 :
159 : /* set ip length */
160 0 : buf[14 + 2] = (uchar)( ip_len >> 8 );
161 0 : buf[14 + 3] = (uchar)( ip_len );
162 0 : }
163 0 : fd_aio_send( ctx->quic_rx_aio, pkt, (ulong)retval, NULL, 1 );
164 0 : }
165 0 : } else if( FD_UNLIKELY( revents & POLLERR ) ) {
166 0 : int error = 0;
167 0 : socklen_t errlen = sizeof(error);
168 :
169 0 : if( getsockopt( ctx->poll_fd[j].fd, SOL_SOCKET, SO_ERROR, (void *)&error, &errlen ) == -1 ) {
170 0 : FD_LOG_ERR(( "Unknown error on socket" ));
171 0 : } else {
172 0 : FD_LOG_ERR(( "Error on socket: %d %s", error, strerror( error ) ));
173 0 : }
174 0 : }
175 0 : }
176 0 : }
177 0 : }
178 :
179 : static void
180 : quic_stream_notify( fd_quic_stream_t * stream,
181 : void * stream_ctx,
182 0 : int type ) {
183 0 : (void)stream;
184 0 : (void)stream_ctx;
185 0 : (void)type;
186 :
187 0 : fd_benchs_ctx_t * ctx = (fd_benchs_ctx_t*)stream_ctx;
188 0 : if( FD_LIKELY( ctx ) ) {
189 0 : if( FD_LIKELY( ctx->stream == stream ) ) {
190 0 : ctx->stream = NULL;
191 0 : }
192 0 : } else {
193 0 : FD_LOG_ERR(( "quic_stream_notify - no context" ));
194 0 : }
195 0 : }
196 :
197 : /* quic_conn_new is invoked by the QUIC engine whenever a new connection
198 : is being established. */
199 : static void
200 : quic_conn_new( fd_quic_conn_t * conn,
201 0 : void * _ctx ) {
202 0 : (void)conn;
203 0 : (void)_ctx;
204 0 : }
205 :
206 :
207 : void
208 : handshake_complete( fd_quic_conn_t * conn,
209 0 : void * _ctx ) {
210 0 : (void)conn;
211 0 : (void)_ctx;
212 0 : FD_LOG_NOTICE(( "client handshake complete" ));
213 0 : }
214 :
215 :
216 : /* quic_stream_new is called back by the QUIC engine whenever an open
217 : connection creates a new stream, at the time this is called, both the
218 : client and server must have agreed to open the stream. In case the
219 : client has opened this stream, it is assumed that the QUIC
220 : implementation has verified that the client has the necessary stream
221 : quota to do so. */
222 :
223 : static void
224 : quic_stream_new( fd_quic_stream_t * stream,
225 0 : void * _ctx ) {
226 : /* we don't expect the server to initiate streams */
227 0 : (void)stream;
228 0 : (void)_ctx;
229 0 : }
230 :
231 : /* quic_stream_receive is called back by the QUIC engine when any stream
232 : in any connection being serviced receives new data. Currently we
233 : simply copy received data out of the xsk (network device memory) into
234 : a local dcache. */
235 :
236 : static void
237 : quic_stream_receive( fd_quic_stream_t * stream,
238 : void * stream_ctx,
239 : uchar const * data,
240 : ulong data_sz,
241 : ulong offset,
242 0 : int fin ) {
243 : /* we're not expecting to receive anything */
244 0 : (void)stream;
245 0 : (void)stream_ctx;
246 0 : (void)data;
247 0 : (void)data_sz;
248 0 : (void)offset;
249 0 : (void)fin;
250 0 : }
251 :
252 :
253 : static void
254 : quic_stream_notify( fd_quic_stream_t * stream,
255 : void * stream_ctx,
256 : int type );
257 :
258 : static void
259 : conn_final( fd_quic_conn_t * conn,
260 0 : void * _ctx ) {
261 0 : (void)conn;
262 :
263 0 : fd_benchs_ctx_t * ctx = (fd_benchs_ctx_t *)_ctx;
264 :
265 0 : if( ctx ) {
266 0 : ctx->quic_conn = NULL;
267 0 : ctx->stream = NULL;
268 0 : }
269 0 : }
270 :
271 : FD_FN_CONST static inline ulong
272 0 : scratch_align( void ) {
273 0 : return fd_ulong_max( fd_quic_align(), alignof( fd_benchs_ctx_t ) );
274 0 : }
275 :
276 : void
277 0 : populate_quic_limits( fd_quic_limits_t * limits ) {
278 : //int argc = 0;
279 : //char * args[] = { NULL };
280 : //char ** argv = args;
281 : //fd_quic_limits_from_env( &argc, &argv, limits );
282 0 : limits->conn_cnt = 2;
283 0 : limits->handshake_cnt = limits->conn_cnt;
284 0 : limits->conn_id_cnt = 16;
285 0 : limits->inflight_pkt_cnt = 1500;
286 0 : limits->tx_buf_sz = fd_ulong_pow2_up( FD_TXN_MTU );
287 0 : limits->stream_pool_cnt = 1UL<<16;
288 0 : limits->stream_id_cnt = 1UL<<16;
289 0 : }
290 :
291 : void
292 0 : populate_quic_config( fd_quic_config_t * config ) {
293 0 : config->role = FD_QUIC_ROLE_CLIENT;
294 0 : config->retry = 0;
295 0 : config->initial_rx_max_stream_data = 0; /* we don't expect the server to initiate streams */
296 :
297 0 : config->net.ephem_udp_port.lo = 12000;
298 0 : config->net.ephem_udp_port.hi = 12100;
299 :
300 0 : config->net.dscp = 0;
301 0 : }
302 :
303 : FD_FN_PURE static inline ulong
304 0 : scratch_footprint( fd_topo_tile_t const * tile ) {
305 0 : (void)tile;
306 0 : ulong l = FD_LAYOUT_INIT;
307 0 : l = FD_LAYOUT_APPEND( l, alignof( fd_benchs_ctx_t ), sizeof( fd_benchs_ctx_t ) );
308 0 : if( !tile->benchs.no_quic ) {
309 0 : fd_quic_limits_t quic_limits = {0};
310 0 : populate_quic_limits( &quic_limits );
311 0 : ulong quic_fp = fd_quic_footprint( &quic_limits );
312 0 : l = FD_LAYOUT_APPEND( l, fd_quic_align(), quic_fp );
313 0 : l = FD_LAYOUT_APPEND( l, fd_aio_align(), fd_aio_footprint() );
314 0 : }
315 0 : return FD_LAYOUT_FINI( l, scratch_align() );
316 0 : }
317 :
318 : static inline int
319 : before_frag( fd_benchs_ctx_t * ctx,
320 : ulong in_idx,
321 : ulong seq,
322 0 : ulong sig ) {
323 0 : (void)in_idx;
324 0 : (void)sig;
325 :
326 0 : return (int)( (seq%ctx->round_robin_cnt)!=ctx->round_robin_id );
327 0 : }
328 :
329 : static inline void
330 : during_frag( fd_benchs_ctx_t * ctx,
331 : ulong in_idx,
332 : ulong seq,
333 : ulong sig,
334 : ulong chunk,
335 0 : ulong sz ) {
336 0 : (void)in_idx;
337 0 : (void)seq;
338 0 : (void)sig;
339 :
340 0 : if( ctx->no_quic ) {
341 :
342 0 : if( FD_UNLIKELY( -1==send( ctx->conn_fd[ ctx->packet_cnt % ctx->conn_cnt ], fd_chunk_to_laddr( ctx->mem, chunk ), sz, 0 ) ) )
343 0 : FD_LOG_ERR(( "send() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
344 :
345 0 : ctx->packet_cnt++;
346 0 : } else {
347 : /* allows to accumulate multiple transactions before creating a UDP datagram */
348 : /* make this configurable */
349 0 : if( FD_UNLIKELY( ctx->service_ratio_idx++ == 8 ) ) {
350 0 : ctx->service_ratio_idx = 0;
351 0 : service_quic( ctx );
352 0 : fd_quic_service( ctx->quic );
353 0 : }
354 :
355 0 : if( FD_UNLIKELY( !ctx->quic_conn ) ) {
356 0 : ctx->no_stream = 0;
357 :
358 : /* try to connect */
359 0 : uint dest_ip = 0;
360 0 : ushort dest_port = fd_ushort_bswap( ctx->quic_port );
361 :
362 0 : ctx->quic_conn = fd_quic_connect( ctx->quic, dest_ip, dest_port, "client" );
363 :
364 : /* failed? try later */
365 0 : if( FD_UNLIKELY( !ctx->quic_conn ) ) {
366 0 : service_quic( ctx );
367 0 : fd_quic_service( ctx->quic );
368 0 : return;
369 0 : }
370 :
371 0 : FD_LOG_NOTICE(( "connection created on port %d", (int)dest_port ));
372 :
373 : /* set the context to point to the location
374 : of the quic_conn pointer
375 : this allows the notification to NULL the value when
376 : a connection dies */
377 0 : fd_quic_conn_set_context( ctx->quic_conn, ctx );
378 :
379 0 : service_quic( ctx );
380 0 : fd_quic_service( ctx->quic );
381 :
382 : /* conn and streams may be invalidated by fd_quic_service */
383 :
384 0 : return;
385 0 : }
386 :
387 0 : fd_quic_stream_t * stream = ctx->stream;
388 0 : if( FD_UNLIKELY( !stream ) ) {
389 0 : ctx->stream = stream = fd_quic_conn_new_stream( ctx->quic_conn );
390 0 : if( FD_LIKELY( stream ) ) {
391 0 : fd_quic_stream_set_context( stream, ctx );
392 0 : }
393 0 : }
394 :
395 0 : if( FD_UNLIKELY( !stream ) ) {
396 0 : ctx->no_stream++;
397 0 : service_quic( ctx );
398 0 : fd_quic_service( ctx->quic );
399 :
400 : /* conn and streams may be invalidated by fd_quic_service */
401 :
402 0 : return;
403 0 : } else {
404 0 : int fin = 1;
405 0 : int rtn = fd_quic_stream_send( stream, fd_chunk_to_laddr( ctx->mem, chunk ), sz, fin );
406 0 : ctx->packet_cnt++;
407 :
408 0 : if( FD_LIKELY( rtn == FD_QUIC_SUCCESS ) ) {
409 : /* after using, fetch a new stream */
410 0 : ctx->stream = stream = fd_quic_conn_new_stream( ctx->quic_conn );
411 0 : if( FD_LIKELY( stream ) ) {
412 0 : fd_quic_stream_set_context( stream, ctx );
413 0 : }
414 0 : } else if( FD_UNLIKELY( rtn == 0 ) ) {
415 0 : FD_LOG_NOTICE(( "fd_quic_stream_send returned zero" ));
416 0 : } else {
417 : /* this can happen dring handshaking */
418 0 : if( rtn != FD_QUIC_SEND_ERR_INVAL_CONN ) {
419 0 : FD_LOG_ERR(( "fd_quic_stream_send failed with: %d", rtn ));
420 0 : }
421 0 : }
422 0 : }
423 0 : }
424 0 : }
425 :
426 : static void
427 : privileged_init( fd_topo_t * topo,
428 0 : fd_topo_tile_t * tile ) {
429 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
430 :
431 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
432 0 : fd_benchs_ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof( fd_benchs_ctx_t ), sizeof( fd_benchs_ctx_t ) );
433 :
434 0 : int no_quic = ctx->no_quic = tile->benchs.no_quic;
435 :
436 0 : if( !no_quic ) {
437 0 : fd_quic_limits_t quic_limits = {0};
438 0 : populate_quic_limits( &quic_limits );
439 0 : ulong quic_fp = fd_quic_footprint( &quic_limits );
440 0 : if( FD_UNLIKELY( !quic_fp ) ) FD_LOG_ERR(( "invalid QUIC parameters" ));
441 0 : void * quic_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_quic_align(), quic_fp );
442 0 : fd_quic_t * quic = fd_quic_join( fd_quic_new( quic_mem, &quic_limits ) );
443 :
444 0 : populate_quic_config( &quic->config );
445 :
446 : /* Signer */
447 0 : fd_rng_t _rng[1];
448 0 : uint seed = 4242424242;
449 0 : fd_rng_t * rng = fd_rng_join( fd_rng_new( _rng, seed, 0UL ) );
450 :
451 0 : ctx->signer_ctx = signer_ctx( rng );
452 :
453 0 : quic->config.sign_ctx = &ctx->signer_ctx;
454 0 : quic->config.sign = signer;
455 :
456 0 : fd_memcpy( quic->config.identity_public_key, ctx->signer_ctx.public_key, 32UL );
457 :
458 : /* store the pointer to quic and quic_rx_aio for later use */
459 0 : ctx->quic = quic;
460 0 : ctx->quic_rx_aio = fd_quic_get_aio_net_rx( quic );
461 :
462 0 : ctx->quic_conn = NULL;
463 0 : ctx->stream = NULL;
464 0 : ctx->tx_idx = 0UL;
465 :
466 : /* call wallclock so glibc loads VDSO, which requires calling mmap while
467 : privileged */
468 0 : fd_log_wallclock();
469 0 : }
470 :
471 0 : ushort port = 12000;
472 :
473 0 : ctx->conn_cnt = tile->benchs.conn_cnt;
474 0 : if( !no_quic ) ctx->conn_cnt = 1;
475 0 : FD_TEST( ctx->conn_cnt <=sizeof(ctx->conn_fd)/sizeof(*ctx->conn_fd) );
476 0 : ctx->quic_port = tile->benchs.send_to_port;
477 0 : for( ulong i=0UL; i<ctx->conn_cnt ; i++ ) {
478 0 : int conn_fd = socket( AF_INET, SOCK_DGRAM, 0 );
479 0 : if( FD_UNLIKELY( -1==conn_fd ) ) FD_LOG_ERR(( "socket() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
480 :
481 0 : int recvbuff = 8<<20;
482 :
483 : // Set the buffer size
484 0 : if( setsockopt( conn_fd, SOL_SOCKET, SO_RCVBUF, &recvbuff, sizeof(recvbuff) ) < 0 ) {
485 0 : FD_LOG_ERR(( "Error setting receive buffer size. Error: %d %s", errno, strerror( errno ) ));
486 0 : }
487 :
488 0 : int sendbuff = 8<<20;
489 0 : if( setsockopt( conn_fd, SOL_SOCKET, SO_SNDBUF, &sendbuff, sizeof(sendbuff) ) < 0 ) {
490 0 : FD_LOG_ERR(( "Error setting transmit buffer size. Error: %d %s", errno, strerror( errno ) ));
491 0 : }
492 :
493 0 : ushort found_port = 0;
494 0 : for( ulong j=0UL; j<10UL; j++ ) {
495 0 : struct sockaddr_in addr = {
496 0 : .sin_family = AF_INET,
497 0 : .sin_port = fd_ushort_bswap( port ),
498 0 : .sin_addr.s_addr = fd_uint_bswap( INADDR_ANY ),
499 0 : };
500 0 : if( FD_UNLIKELY( -1!=bind( conn_fd, fd_type_pun( &addr ), sizeof(addr) ) ) ) {
501 0 : found_port = port;
502 0 : break;
503 0 : }
504 0 : if( FD_UNLIKELY( EADDRINUSE!=errno ) ) FD_LOG_ERR(( "bind() failed (%i-%s)", errno, fd_io_strerror( errno ) ) );
505 0 : port = (ushort)(port + ctx->conn_cnt); /* Make sure it round robins to the same tile index */
506 0 : }
507 0 : if( FD_UNLIKELY( !found_port ) ) FD_LOG_ERR(( "bind() failed to find a src port" ));
508 :
509 0 : struct sockaddr_in addr = {
510 0 : .sin_family = AF_INET,
511 0 : .sin_port = fd_ushort_bswap( tile->benchs.send_to_port ),
512 0 : .sin_addr.s_addr = tile->benchs.send_to_ip_addr,
513 0 : };
514 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 ) ));
515 :
516 0 : ctx->conn_fd[ i ] = conn_fd;
517 0 : if( !no_quic ) {
518 0 : ctx->poll_fd[i].fd = conn_fd;
519 0 : ctx->poll_fd[i].events = POLLIN;
520 0 : }
521 0 : port++;
522 0 : }
523 0 : }
524 :
525 : static void
526 : unprivileged_init( fd_topo_t * topo,
527 0 : fd_topo_tile_t * tile ) {
528 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
529 :
530 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
531 0 : fd_benchs_ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof( fd_benchs_ctx_t ), sizeof( fd_benchs_ctx_t ) );
532 :
533 0 : ctx->packet_cnt = 0UL;
534 :
535 0 : ctx->round_robin_id = tile->kind_id;
536 0 : ctx->round_robin_cnt = fd_topo_tile_name_cnt( topo, "benchs" );
537 :
538 0 : ctx->mem = topo->workspaces[ topo->objs[ topo->links[ tile->in_link_id[ 0UL ] ].dcache_obj_id ].wksp_id ].wksp;
539 :
540 0 : void * aio_mem = NULL;
541 :
542 0 : if( !ctx->no_quic ) {
543 0 : fd_quic_limits_t quic_limits = {0};
544 0 : populate_quic_limits( &quic_limits );
545 0 : ulong quic_fp = fd_quic_footprint( &quic_limits );
546 0 : FD_SCRATCH_ALLOC_APPEND( l, fd_quic_align(), quic_fp );
547 :
548 0 : fd_quic_t * quic = ctx->quic;
549 0 : aio_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_aio_align(), fd_aio_footprint() );
550 0 : fd_aio_t * quic_tx_aio = fd_aio_join( fd_aio_new( aio_mem, ctx, quic_tx_aio_send ) );
551 :
552 0 : if( FD_UNLIKELY( !quic_tx_aio ) ) FD_LOG_ERR(( "fd_aio_join failed" ));
553 :
554 0 : uint quic_ip_addr = 0; /* TODO fetch the quic destination ip addr */
555 0 : ulong quic_idle_timeout_millis = 10000; /* idle timeout in milliseconds */
556 0 : uchar quic_src_mac_addr[6] = {0}; /* source MAC address */
557 0 : quic->config.role = FD_QUIC_ROLE_CLIENT;
558 0 : quic->config.net.ip_addr = quic_ip_addr;
559 0 : quic->config.net.listen_udp_port = 42424; /* should be unused */
560 0 : quic->config.idle_timeout = quic_idle_timeout_millis * 1000000UL;
561 0 : quic->config.initial_rx_max_stream_data = 0;
562 0 : quic->config.retry = 0; /* unused on clients */
563 0 : fd_memcpy( quic->config.link.src_mac_addr, quic_src_mac_addr, 6 );
564 :
565 0 : quic->cb.conn_new = quic_conn_new;
566 0 : quic->cb.conn_hs_complete = handshake_complete;
567 0 : quic->cb.conn_final = conn_final;
568 0 : quic->cb.stream_new = quic_stream_new;
569 0 : quic->cb.stream_receive = quic_stream_receive;
570 0 : quic->cb.stream_notify = quic_stream_notify;
571 0 : quic->cb.now = quic_now;
572 0 : quic->cb.now_ctx = NULL;
573 0 : quic->cb.quic_ctx = ctx;
574 :
575 0 : fd_quic_set_aio_net_tx( quic, quic_tx_aio );
576 0 : if( FD_UNLIKELY( !fd_quic_init( quic ) ) ) FD_LOG_ERR(( "fd_quic_init failed" ));
577 :
578 0 : ulong hdr_sz = 14 + 20 + 8;
579 0 : for( ulong i = 0; i < IO_VEC_CNT; i++ ) {
580 : /* leave space for headers */
581 0 : ctx->rx_iovecs[i].iov_base = ctx->rx_bufs[i] + hdr_sz;
582 0 : ctx->rx_iovecs[i].iov_len = sizeof(ctx->rx_bufs[i]) - hdr_sz;
583 0 : ctx->rx_msgs[i].msg_hdr.msg_iov = &ctx->rx_iovecs[i];
584 0 : ctx->rx_msgs[i].msg_hdr.msg_iovlen = 1;
585 0 : }
586 0 : }
587 :
588 0 : ulong scratch_top = FD_SCRATCH_ALLOC_FINI( l, 1UL );
589 0 : if( FD_UNLIKELY( scratch_top > (ulong)scratch + scratch_footprint( tile ) ) )
590 0 : FD_LOG_ERR(( "scratch overflow %lu %lu %lu", scratch_top - (ulong)scratch - scratch_footprint( tile ), scratch_top, (ulong)scratch + scratch_footprint( tile ) ));
591 :
592 0 : }
593 :
594 : static void
595 0 : quic_tx_aio_send_flush( fd_benchs_ctx_t * ctx ) {
596 0 : if( FD_LIKELY( ctx->tx_idx ) ) {
597 0 : int flags = 0;
598 0 : int rtn = sendmmsg( ctx->conn_fd[0], ctx->tx_msgs, (uint)ctx->tx_idx, flags );
599 0 : if( FD_UNLIKELY( rtn < 0 ) ) {
600 0 : FD_LOG_NOTICE(( "Error occurred in sendmmsg. Error: %d %s",
601 0 : errno, strerror( errno ) ));
602 0 : }
603 0 : ctx->tx_idx = 0;
604 0 : }
605 0 : }
606 :
607 : static int
608 : quic_tx_aio_send( void * _ctx,
609 : fd_aio_pkt_info_t const * batch,
610 : ulong batch_cnt,
611 : ulong * opt_batch_idx,
612 0 : int flush ) {
613 0 : (void)batch;
614 0 : (void)batch_cnt;
615 0 : (void)opt_batch_idx;
616 0 : (void)flush;
617 :
618 0 : fd_benchs_ctx_t * ctx = (fd_benchs_ctx_t *)_ctx;
619 :
620 : /* quic adds eth, ip and udp headers which we don't need */
621 : /* assume 14 + 20 + 8 for those */
622 0 : ulong hdr_sz = 14+20+8;
623 :
624 0 : if( FD_LIKELY( batch_cnt ) ) {
625 : /* do we have space? */
626 0 : ulong remain = IO_VEC_CNT - ctx->tx_idx;
627 0 : if( FD_UNLIKELY( remain > batch_cnt ) ) {
628 0 : quic_tx_aio_send_flush( ctx );
629 :
630 : /* tx_idx may have changed */
631 0 : remain = IO_VEC_CNT - ctx->tx_idx;
632 0 : }
633 :
634 0 : ulong cnt = fd_ulong_min( remain, batch_cnt );
635 0 : ulong tx_idx = ctx->tx_idx;
636 0 : for( ulong j = 0; j < cnt; ++j ) {
637 0 : if( FD_UNLIKELY( batch[j].buf_sz < hdr_sz ) ) continue;
638 :
639 0 : char * tx_buf = ctx->tx_bufs[tx_idx];
640 :
641 : /* copy, stripping the header */
642 0 : fd_memcpy( tx_buf, (uchar*)batch[j].buf + hdr_sz, batch[j].buf_sz - hdr_sz );
643 :
644 0 : ctx->tx_iovecs[tx_idx].iov_base = tx_buf;
645 0 : ctx->tx_iovecs[tx_idx].iov_len = batch[j].buf_sz - hdr_sz;
646 0 : ctx->tx_msgs[tx_idx].msg_hdr.msg_iov = &ctx->tx_iovecs[tx_idx];
647 0 : ctx->tx_msgs[tx_idx].msg_hdr.msg_iovlen = 1;
648 :
649 0 : tx_idx++;
650 0 : }
651 :
652 : /* write back */
653 0 : ctx->tx_idx = tx_idx;
654 :
655 : // TODO count drops?
656 : // ctx->dropped += batch_cnt - remain;
657 0 : }
658 :
659 0 : if( FD_UNLIKELY( ctx->tx_idx == IO_VEC_CNT || flush ) ) {
660 0 : quic_tx_aio_send_flush( ctx );
661 0 : }
662 :
663 0 : if( FD_LIKELY( opt_batch_idx ) ) *opt_batch_idx = batch_cnt;
664 :
665 0 : return 0;
666 0 : }
667 :
668 0 : #define STEM_BURST (1UL)
669 :
670 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_benchs_ctx_t
671 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_benchs_ctx_t)
672 :
673 0 : #define STEM_CALLBACK_BEFORE_FRAG before_frag
674 0 : #define STEM_CALLBACK_DURING_FRAG during_frag
675 :
676 : #include "../../../disco/stem/fd_stem.c"
677 :
678 : fd_topo_run_tile_t fd_tile_benchs = {
679 : .name = "benchs",
680 : .scratch_align = scratch_align,
681 : .scratch_footprint = scratch_footprint,
682 : .privileged_init = privileged_init,
683 : .unprivileged_init = unprivileged_init,
684 : .run = stem_run,
685 : };
|
