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