LCOV - code coverage report
Current view: top level - app/shared_dev/commands/bench - fd_benchs.c (source / functions) Hit Total Coverage
Test: cov.lcov Lines: 0 309 0.0 %
Date: 2026-05-22 08:05:30 Functions: 0 16 0.0 %

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

Generated by: LCOV version 1.14