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 301 0.0 %
Date: 2025-07-01 05:00:49 Functions: 0 15 0.0 %

          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             : /* quic_now is called by the QUIC engine to get the current timestamp in
      34             :    UNIX time.  */
      35             : 
      36             : static ulong
      37           0 : quic_now( void * ctx ) {
      38           0 :   (void)ctx;
      39           0 :   return (ulong)fd_log_wallclock();
      40           0 : }
      41             : 
      42             : typedef struct {
      43             :   ulong round_robin_cnt;
      44             :   ulong round_robin_id;
      45             : 
      46             :   ulong packet_cnt;
      47             : 
      48             :   ulong         conn_cnt;
      49             :   int           conn_fd[ 128UL ];
      50             :   struct pollfd poll_fd[ 128UL ];
      51             : 
      52             :   fd_tls_test_sign_ctx_t test_signer[1];
      53             :   int              no_quic;
      54             :   fd_quic_t *      quic;
      55             :   ushort           quic_port;
      56             :   fd_quic_conn_t * quic_conn;
      57             :   ulong            no_stream;
      58             :   uint             service_ratio_idx;
      59             :   fd_aio_t         tx_aio;
      60             : 
      61             :   // vector receive members
      62             :   struct mmsghdr rx_msgs[IO_VEC_CNT];
      63             :   struct mmsghdr tx_msgs[IO_VEC_CNT];
      64             :   struct iovec   rx_iovecs[IO_VEC_CNT];
      65             :   struct iovec   tx_iovecs[IO_VEC_CNT];
      66             :   uchar          rx_bufs[IO_VEC_CNT][2048];
      67             :   uchar          tx_bufs[IO_VEC_CNT][2048];
      68             : 
      69             :   ulong tx_idx;
      70             : 
      71             :   fd_wksp_t * mem;
      72             : } fd_benchs_ctx_t;
      73             : 
      74             : static void
      75           0 : service_quic( fd_benchs_ctx_t * ctx ) {
      76             : 
      77           0 :   if( !ctx->no_quic ) {
      78             :     /* Publishes to mcache via callbacks */
      79             : 
      80             :     /* receive from socket, and pass to quic */
      81           0 :     int poll_rc = poll( ctx->poll_fd, ctx->conn_cnt, 0 );
      82           0 :     if( FD_LIKELY( poll_rc == 0 ) ) {
      83           0 :       return;
      84           0 :     } if( FD_UNLIKELY( poll_rc == -1 ) ) {
      85           0 :       if( FD_UNLIKELY( errno == EINTR ) ) return; /* will try later */
      86           0 :       FD_LOG_ERR(( "Error occurred during poll: %d %s", errno,
      87           0 :             strerror( errno ) ));
      88           0 :     }
      89             : 
      90           0 :     for( ulong j = 0; j < ctx->conn_cnt; ++j ) {
      91           0 :       int revents = ctx->poll_fd[j].revents;
      92           0 :       if( FD_LIKELY( revents & POLLIN ) ) {
      93             :         /* data available - receive up to IO_VEC_CNT buffers */
      94           0 :         struct timespec timeout = {0};
      95           0 :         int retval = recvmmsg( ctx->poll_fd[j].fd, ctx->rx_msgs, IO_VEC_CNT, 0, &timeout );
      96           0 :         if( FD_UNLIKELY( retval < 0 ) ) {
      97           0 :           FD_LOG_ERR(( "Error occurred on recvmmsg: %d %s", errno, strerror( errno ) ));
      98           0 :         }
      99             :         /* pass buffers to QUIC */
     100           0 :         for( ulong k = 0; k < (ulong)retval; k++ ) {
     101           0 :           uchar * buf = ctx->rx_bufs[k];
     102             : 
     103             :           /* set some required values */
     104           0 :           uint payload_len = ctx->rx_msgs[k].msg_len;
     105           0 :           uint udp_len     = payload_len + 8;
     106           0 :           uint ip_len      = udp_len + 20;
     107             : 
     108             :           /* set ver and len */
     109           0 :           buf[0] = 0x45;
     110             : 
     111             :           /* set protocol */
     112           0 :           buf[9] = 17;
     113             : 
     114             :           /* set udp length */
     115           0 :           buf[20 + 4] = (uchar)( udp_len >> 8 );
     116           0 :           buf[20 + 5] = (uchar)( udp_len      );
     117             : 
     118             :           /* set ip length */
     119           0 :           buf[2] = (uchar)( ip_len >> 8 );
     120           0 :           buf[3] = (uchar)( ip_len      );
     121             : 
     122           0 :           fd_quic_process_packet( ctx->quic, buf, ip_len );
     123           0 :         }
     124           0 :       } else if( FD_UNLIKELY( revents & POLLERR ) ) {
     125           0 :         int error = 0;
     126           0 :         socklen_t errlen = sizeof(error);
     127             : 
     128           0 :         if( getsockopt( ctx->poll_fd[j].fd, SOL_SOCKET, SO_ERROR, (void *)&error, &errlen ) == -1 ) {
     129           0 :           FD_LOG_ERR(( "Unknown error on socket" ));
     130           0 :         } else {
     131           0 :           FD_LOG_ERR(( "Error on socket: %d %s", error, strerror( error ) ));
     132           0 :         }
     133           0 :       }
     134           0 :     }
     135           0 :   }
     136           0 : }
     137             : 
     138             : /* quic_conn_new is invoked by the QUIC engine whenever a new connection
     139             :    is being established. */
     140             : static void
     141             : quic_conn_new( fd_quic_conn_t * conn,
     142           0 :                void *           _ctx ) {
     143           0 :   (void)conn;
     144           0 :   (void)_ctx;
     145           0 : }
     146             : 
     147             : 
     148             : static void
     149             : handshake_complete( fd_quic_conn_t * conn,
     150           0 :                     void *           _ctx ) {
     151           0 :   (void)conn;
     152           0 :   (void)_ctx;
     153           0 :   FD_LOG_NOTICE(( "client handshake complete" ));
     154           0 : }
     155             : 
     156             : static void
     157             : conn_final( fd_quic_conn_t * conn,
     158           0 :             void *           _ctx ) {
     159           0 :   (void)conn;
     160             : 
     161           0 :   fd_benchs_ctx_t * ctx = (fd_benchs_ctx_t *)_ctx;
     162             : 
     163           0 :   if( ctx ) {
     164           0 :     ctx->quic_conn = NULL;
     165           0 :   }
     166           0 : }
     167             : 
     168             : FD_FN_CONST static inline ulong
     169           0 : scratch_align( void ) {
     170           0 :   return fd_ulong_max( fd_quic_align(), alignof( fd_benchs_ctx_t ) );
     171           0 : }
     172             : 
     173             : static void
     174           0 : populate_quic_limits( fd_quic_limits_t * limits ) {
     175           0 :   limits->conn_cnt = 2;
     176           0 :   limits->handshake_cnt = limits->conn_cnt;
     177           0 :   limits->conn_id_cnt = 16;
     178           0 :   limits->inflight_frame_cnt = 1500;
     179           0 :   limits->tx_buf_sz = FD_TXN_MTU;
     180           0 :   limits->stream_pool_cnt = 1UL<<16;
     181           0 :   limits->stream_id_cnt = 1UL<<16;
     182           0 : }
     183             : 
     184             : static void
     185           0 : populate_quic_config( fd_quic_config_t * config ) {
     186           0 :   config->role = FD_QUIC_ROLE_CLIENT;
     187           0 :   config->retry = 0;
     188           0 :   config->initial_rx_max_stream_data = 0; /* we don't expect the server to initiate streams */
     189           0 :   config->net.dscp = 0;
     190           0 : }
     191             : 
     192             : static inline ulong
     193           0 : scratch_footprint( fd_topo_tile_t const * tile ) {
     194           0 :   ulong l = FD_LAYOUT_INIT;
     195           0 :   l = FD_LAYOUT_APPEND( l, alignof( fd_benchs_ctx_t ), sizeof( fd_benchs_ctx_t ) );
     196           0 :   if( !tile->benchs.no_quic ) {
     197           0 :     fd_quic_limits_t quic_limits = {0};
     198           0 :     populate_quic_limits( &quic_limits );
     199           0 :     ulong quic_fp = fd_quic_footprint( &quic_limits );
     200           0 :     l = FD_LAYOUT_APPEND( l, fd_quic_align(), quic_fp );
     201           0 :   }
     202           0 :   return FD_LAYOUT_FINI( l, scratch_align() );
     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 :   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 );
     236           0 :       fd_quic_service( ctx->quic );
     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 );
     247             : 
     248             :       /* failed? try later */
     249           0 :       if( FD_UNLIKELY( !ctx->quic_conn ) ) {
     250           0 :         service_quic( ctx );
     251           0 :         fd_quic_service( ctx->quic );
     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 );
     264           0 :       fd_quic_service( ctx->quic );
     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 );
     275           0 :       fd_quic_service( ctx->quic );
     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               = quic_idle_timeout_millis * 1000000UL;
     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.now              = quic_now;
     411           0 :     quic->cb.now_ctx          = NULL;
     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 : }
     438             : 
     439             : static void
     440           0 : quic_tx_aio_send_flush( fd_benchs_ctx_t * ctx ) {
     441           0 :   if( FD_LIKELY( ctx->tx_idx ) ) {
     442           0 :     int flags = 0;
     443           0 :     int rtn = sendmmsg( ctx->conn_fd[0], ctx->tx_msgs, (uint)ctx->tx_idx, flags );
     444           0 :     if( FD_UNLIKELY( rtn < 0 ) ) {
     445           0 :       FD_LOG_NOTICE(( "Error occurred in sendmmsg. Error: %d %s",
     446           0 :           errno, strerror( errno ) ));
     447           0 :     }
     448           0 :     ctx->tx_idx = 0;
     449           0 :   }
     450           0 : }
     451             : 
     452             : static int
     453             : quic_tx_aio_send( void *                    _ctx,
     454             :                   fd_aio_pkt_info_t const * batch,
     455             :                   ulong                     batch_cnt,
     456             :                   ulong *                   opt_batch_idx,
     457           0 :                   int                       flush ) {
     458           0 :   fd_benchs_ctx_t * ctx = _ctx;
     459             : 
     460             :   /* quic adds ip and udp headers which we don't need */
     461             :   /* assume 20 + 8 for those */
     462           0 :   ulong hdr_sz = 20+8;
     463             : 
     464           0 :   if( FD_LIKELY( batch_cnt ) ) {
     465             :     /* do we have space? */
     466           0 :     ulong remain = IO_VEC_CNT - ctx->tx_idx;
     467           0 :     if( FD_UNLIKELY( remain > batch_cnt ) ) {
     468           0 :       quic_tx_aio_send_flush( ctx );
     469             : 
     470             :       /* tx_idx may have changed */
     471           0 :       remain = IO_VEC_CNT - ctx->tx_idx;
     472           0 :     }
     473             : 
     474           0 :     ulong cnt = fd_ulong_min( remain, batch_cnt );
     475           0 :     ulong tx_idx = ctx->tx_idx;
     476           0 :     for( ulong j = 0; j < cnt; ++j ) {
     477           0 :       if( FD_UNLIKELY( batch[j].buf_sz < hdr_sz ) ) continue;
     478             : 
     479           0 :       uchar * tx_buf = ctx->tx_bufs[tx_idx];
     480             : 
     481             :       /* copy, stripping the header */
     482           0 :       fd_memcpy( tx_buf, (uchar*)batch[j].buf + hdr_sz, batch[j].buf_sz - hdr_sz );
     483             : 
     484           0 :       ctx->tx_iovecs[tx_idx] = (struct iovec) {
     485           0 :         .iov_base = tx_buf,
     486           0 :         .iov_len  = batch[j].buf_sz - hdr_sz
     487           0 :       };
     488           0 :       ctx->tx_msgs[tx_idx] = (struct mmsghdr) {
     489           0 :         .msg_hdr = {
     490           0 :           .msg_iov    = &ctx->tx_iovecs[tx_idx],
     491           0 :           .msg_iovlen = 1,
     492           0 :         }
     493           0 :       };
     494             : 
     495           0 :       tx_idx++;
     496           0 :     }
     497             : 
     498             :     /* write back */
     499           0 :     ctx->tx_idx = tx_idx;
     500             : 
     501             :     // TODO count drops?
     502             :     // ctx->dropped += batch_cnt - remain;
     503           0 :   }
     504             : 
     505           0 :   if( FD_UNLIKELY( ctx->tx_idx == IO_VEC_CNT || flush ) ) {
     506           0 :     quic_tx_aio_send_flush( ctx );
     507           0 :   }
     508             : 
     509           0 :   if( FD_LIKELY( opt_batch_idx ) ) *opt_batch_idx = batch_cnt;
     510             : 
     511           0 :   return 0;
     512           0 : }
     513             : 
     514           0 : #define STEM_BURST (1UL)
     515             : 
     516           0 : #define STEM_CALLBACK_CONTEXT_TYPE  fd_benchs_ctx_t
     517           0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_benchs_ctx_t)
     518             : 
     519           0 : #define STEM_CALLBACK_BEFORE_FRAG before_frag
     520           0 : #define STEM_CALLBACK_DURING_FRAG during_frag
     521             : 
     522             : #include "../../../../disco/stem/fd_stem.c"
     523             : 
     524             : fd_topo_run_tile_t fd_tile_benchs = {
     525             :   .name                     = "benchs",
     526             :   .scratch_align            = scratch_align,
     527             :   .scratch_footprint        = scratch_footprint,
     528             :   .privileged_init          = privileged_init,
     529             :   .unprivileged_init        = unprivileged_init,
     530             :   .run                      = stem_run,
     531             : };

Generated by: LCOV version 1.14