LCOV - code coverage report
Current view: top level - disco/net/sock - fd_sock_tile.c (source / functions) Hit Total Coverage
Test: cov.lcov Lines: 0 544 0.0 %
Date: 2026-10-31 04:25:24 Functions: 0 18 0.0 %

          Line data    Source code
       1             : #define _GNU_SOURCE /* dup3 */
       2             : #include "fd_sock_tile_private.h"
       3             : #include "../../fd_disco_base.h"
       4             : #include "../../topo/fd_topo.h"
       5             : #include "../../../util/net/fd_eth.h"
       6             : #include "../../../util/net/fd_ip4.h"
       7             : #include "../../../util/net/fd_udp.h"
       8             : 
       9             : #include <stdalign.h> /* alignof */
      10             : #include <errno.h>
      11             : #include <fcntl.h> /* fcntl */
      12             : #include <unistd.h> /* dup3, close */
      13             : #include <netinet/in.h> /* sockaddr_in */
      14             : #include <net/if.h> /* if_nametoindex */
      15             : #include <linux/filter.h> /* SO_ATTACH_FILTER */
      16             : #include <sys/socket.h> /* socket */
      17             : #include "../../metrics/fd_metrics.h"
      18             : 
      19             : #include "generated/fd_sock_tile_seccomp.h"
      20             : 
      21             : /* recv/sendmmsg packet count in batch and tango burst depth
      22             :    FIXME make configurable in the future?
      23             :    FIXME keep in sync with fd_net_tile_topo.c */
      24           0 : #define STEM_BURST (64UL)
      25             : 
      26             : /* Place RX socket file descriptors in contiguous integer range. */
      27           0 : #define RX_SOCK_FD_MIN (128)
      28             : 
      29             : /* Controls max ancillary data size.
      30             :    Must be aligned by alignof(struct cmsghdr) */
      31           0 : #define FD_SOCK_CMSG_MAX (64UL)
      32             : 
      33             : /* Musl's CMSG_NXTHDR macro compares size_t against ptrdiff_t,
      34             :    triggering -Wsign-compare. */
      35             : #pragma GCC diagnostic push
      36             : #pragma GCC diagnostic ignored "-Wsign-compare"
      37             : #pragma GCC diagnostic ignored "-Wsign-conversion"
      38             : static inline struct cmsghdr *
      39             : fd_cmsg_nxthdr( struct msghdr *   mhdr,
      40           0 :                 struct cmsghdr * cmsg ) {
      41           0 :   return CMSG_NXTHDR( mhdr, cmsg );
      42           0 : }
      43             : #pragma GCC diagnostic pop
      44             : 
      45             : static ulong
      46             : populate_allowed_seccomp( fd_topo_t const *      topo,
      47             :                           fd_topo_tile_t const * tile,
      48             :                           ulong                  out_cnt,
      49           0 :                           struct sock_filter *   out ) {
      50           0 :   FD_SCRATCH_ALLOC_INIT( l, fd_topo_obj_laddr( topo, tile->tile_obj_id ) );
      51           0 :   fd_sock_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_sock_tile_t), sizeof(fd_sock_tile_t) );
      52             : 
      53           0 :   populate_sock_filter_policy_fd_sock_tile( out_cnt, out, (uint)fd_log_private_logfile_fd(), (uint)ctx->tx_sock, RX_SOCK_FD_MIN, RX_SOCK_FD_MIN+(uint)ctx->sock_cnt );
      54           0 :   return sock_filter_policy_fd_sock_tile_instr_cnt;
      55           0 : }
      56             : 
      57             : static ulong
      58             : populate_allowed_fds( fd_topo_t const *      topo,
      59             :                       fd_topo_tile_t const * tile,
      60             :                       ulong                  out_fds_cnt,
      61           0 :                       int *                  out_fds ) {
      62           0 :   void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
      63           0 :   FD_SCRATCH_ALLOC_INIT( l, scratch );
      64           0 :   fd_sock_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_sock_tile_t), sizeof(fd_sock_tile_t) );
      65             : 
      66           0 :   ulong sock_cnt = ctx->sock_cnt;
      67           0 :   if( FD_UNLIKELY( out_fds_cnt<sock_cnt+3UL ) ) {
      68           0 :     FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
      69           0 :   }
      70             : 
      71           0 :   ulong out_cnt = 0UL;
      72             : 
      73           0 :   out_fds[ out_cnt++ ] = 2; /* stderr */
      74           0 :   if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) ) {
      75           0 :     out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
      76           0 :   }
      77           0 :   if( -1!=ctx->tx_sock ) out_fds[ out_cnt++ ] = ctx->tx_sock;
      78           0 :   for( ulong j=0UL; j<sock_cnt; j++ ) {
      79           0 :     out_fds[ out_cnt++ ] = ctx->pollfd[ j ].fd;
      80           0 :   }
      81           0 :   return out_cnt;
      82           0 : }
      83             : 
      84             : FD_FN_CONST static inline ulong
      85           0 : tx_scratch_footprint( void ) {
      86           0 :   return STEM_BURST * fd_ulong_align_up( FD_NET_MTU, FD_CHUNK_ALIGN );
      87           0 : }
      88             : 
      89             : FD_FN_CONST static inline ulong
      90           0 : scratch_align( void ) {
      91           0 :   return 4096UL;
      92           0 : }
      93             : 
      94             : FD_FN_PURE static inline ulong
      95           0 : scratch_footprint( fd_topo_tile_t const * tile ) {
      96           0 :   ulong l = FD_LAYOUT_INIT;
      97           0 :   l = FD_LAYOUT_APPEND( l, alignof(fd_sock_tile_t),     sizeof(fd_sock_tile_t)                );
      98           0 :   l = FD_LAYOUT_APPEND( l, alignof(struct iovec),       STEM_BURST*sizeof(struct iovec)       );
      99           0 :   l = FD_LAYOUT_APPEND( l, alignof(struct cmsghdr),     STEM_BURST*FD_SOCK_CMSG_MAX           );
     100           0 :   l = FD_LAYOUT_APPEND( l, alignof(struct sockaddr_in), STEM_BURST*sizeof(struct sockaddr_in) );
     101           0 :   l = FD_LAYOUT_APPEND( l, alignof(struct mmsghdr),     STEM_BURST*sizeof(struct mmsghdr)     );
     102           0 :   l = FD_LAYOUT_APPEND( l, alignof(ulong),              STEM_BURST*sizeof(ulong)              );
     103           0 :   if( !tile->sock.only_recv_lo ) l = FD_LAYOUT_APPEND( l, FD_CHUNK_ALIGN, tx_scratch_footprint() );
     104           0 :   return FD_LAYOUT_FINI( l, scratch_align() );
     105           0 : }
     106             : 
     107             : /* create_udp_socket creates and configures a new UDP socket for the
     108             :    sock tile at the given file descriptor ID. */
     109             : 
     110             : static void
     111             : create_udp_socket( int    sock_fd,
     112             :                    uint   bind_addr,
     113             :                    ushort udp_port,
     114             :                    int    so_rcvbuf,
     115           0 :                    uint   lo_ifindex ) {
     116             : 
     117           0 :   if( fcntl( sock_fd, F_GETFD, 0 )!=-1 ) {
     118           0 :     FD_LOG_ERR(( "file descriptor %d already exists", sock_fd ));
     119           0 :   } else if( errno!=EBADF ) {
     120           0 :     FD_LOG_ERR(( "fcntl(F_GETFD) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
     121           0 :   }
     122             : 
     123           0 :   int orig_fd = socket( AF_INET, SOCK_DGRAM, IPPROTO_UDP );
     124           0 :   if( FD_UNLIKELY( orig_fd<0 ) ) {
     125           0 :     FD_LOG_ERR(( "socket(AF_INET,SOCK_DGRAM,IPPROTO_UDP) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
     126           0 :   }
     127             : 
     128           0 :   int reuseport = 1;
     129           0 :   if( FD_UNLIKELY( setsockopt( orig_fd, SOL_SOCKET, SO_REUSEPORT, &reuseport, sizeof(int) )<0 ) ) {
     130           0 :     FD_LOG_ERR(( "setsockopt(SOL_SOCKET,SO_REUSEPORT,1) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
     131           0 :   }
     132             : 
     133           0 :   int ip_pktinfo = 1;
     134           0 :   if( FD_UNLIKELY( setsockopt( orig_fd, IPPROTO_IP, IP_PKTINFO, &ip_pktinfo, sizeof(int) )<0 ) ) {
     135           0 :     FD_LOG_ERR(( "setsockopt(IPPROTO_IP,IP_PKTINFO,1) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
     136           0 :   }
     137             : 
     138           0 :   if( FD_UNLIKELY( 0!=setsockopt( orig_fd, SOL_SOCKET, SO_RCVBUF, &so_rcvbuf, sizeof(int) ) ) ) {
     139           0 :     FD_LOG_ERR(( "setsockopt(SOL_SOCKET,SO_RCVBUF,%i) failed (%i-%s)", so_rcvbuf, errno, fd_io_strerror( errno ) ));
     140           0 :   }
     141             : 
     142           0 :   struct sockaddr_in saddr = {
     143           0 :     .sin_family      = AF_INET,
     144           0 :     .sin_addr.s_addr = bind_addr,
     145           0 :     .sin_port        = fd_ushort_bswap( udp_port ),
     146           0 :   };
     147           0 :   if( lo_ifindex ) {
     148             :     /* SKF_AD_IFINDEX identifies the receive device, including local
     149             :        traffic addressed to a non loopback interface's IP address. */
     150           0 :     struct sock_filter filter[] = {
     151           0 :       BPF_STMT( BPF_LD|BPF_W|BPF_ABS, (uint)(SKF_AD_OFF+SKF_AD_IFINDEX) ),
     152           0 :       BPF_JUMP( BPF_JMP|BPF_JEQ|BPF_K, lo_ifindex, 0, 1 ),
     153           0 :       BPF_STMT( BPF_RET|BPF_K, UINT_MAX ),
     154           0 :       BPF_STMT( BPF_RET|BPF_K, 0 )
     155           0 :     };
     156           0 :     struct sock_fprog prog = { .len = 4, .filter = filter };
     157           0 :     if( FD_UNLIKELY( 0!=setsockopt( orig_fd, SOL_SOCKET, SO_ATTACH_FILTER, &prog, sizeof(prog) ) ) ) {
     158           0 :       FD_LOG_ERR(( "setsockopt(SO_ATTACH_FILTER) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
     159           0 :     }
     160           0 :   }
     161           0 :   if( FD_UNLIKELY( 0!=bind( orig_fd, fd_type_pun_const( &saddr ), sizeof(struct sockaddr_in) ) ) ) {
     162           0 :     FD_LOG_ERR(( "bind(0.0.0.0:%i) failed (%i-%s)", udp_port, errno, fd_io_strerror( errno ) ));
     163           0 :   }
     164             : 
     165           0 :   if( orig_fd==sock_fd ) {
     166           0 :     if( FD_UNLIKELY( fcntl( sock_fd, F_SETFD, FD_CLOEXEC ) ) ) {
     167           0 :       FD_LOG_ERR(( "fcntl(F_SETFD) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
     168           0 :     }
     169           0 :     return;
     170           0 :   }
     171           0 :   int dup_res = dup3( orig_fd, sock_fd, O_CLOEXEC );
     172           0 :   if( FD_UNLIKELY( dup_res!=sock_fd ) ) {
     173           0 :     FD_LOG_ERR(( "dup3 returned %i (%i-%s)", dup_res, errno, fd_io_strerror( errno ) ));
     174           0 :   }
     175             : 
     176           0 :   if( FD_UNLIKELY( 0!=close( orig_fd ) ) ) {
     177           0 :     FD_LOG_ERR(( "close(%d) failed (%i-%s)", orig_fd, errno, fd_io_strerror( errno ) ));
     178           0 :   }
     179           0 : }
     180             : 
     181             : static void
     182             : privileged_init( fd_topo_t const *      topo,
     183           0 :                  fd_topo_tile_t const * tile ) {
     184           0 :   void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
     185           0 :   FD_SCRATCH_ALLOC_INIT( l, scratch );
     186           0 :   fd_sock_tile_t *     ctx        = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_sock_tile_t),     sizeof(fd_sock_tile_t)                );
     187           0 :   struct iovec   *     batch_iov  = FD_SCRATCH_ALLOC_APPEND( l, alignof(struct iovec),       STEM_BURST*sizeof(struct iovec)       );
     188           0 :   void *               batch_cmsg = FD_SCRATCH_ALLOC_APPEND( l, alignof(struct cmsghdr),     STEM_BURST*FD_SOCK_CMSG_MAX           );
     189           0 :   struct sockaddr_in * batch_sa   = FD_SCRATCH_ALLOC_APPEND( l, alignof(struct sockaddr_in), STEM_BURST*sizeof(struct sockaddr_in) );
     190           0 :   struct mmsghdr *     batch_msg  = FD_SCRATCH_ALLOC_APPEND( l, alignof(struct mmsghdr),     STEM_BURST*sizeof(struct mmsghdr)     );
     191           0 :   ulong *              batch_fsz  = FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong),              STEM_BURST*sizeof(ulong)              );
     192           0 :   ulong                tx_sz      = tile->sock.only_recv_lo ? 0UL : tx_scratch_footprint();
     193           0 :   uchar *              tx_scratch = FD_SCRATCH_ALLOC_APPEND( l, FD_CHUNK_ALIGN,              tx_sz );
     194           0 :   FD_DCHECK_CRIT( scratch==ctx, "invalid layout" );
     195             : 
     196           0 :   fd_memset( ctx,       0, sizeof(fd_sock_tile_t)                );
     197           0 :   fd_memset( batch_iov, 0, STEM_BURST*sizeof(struct iovec)       );
     198           0 :   fd_memset( batch_sa,  0, STEM_BURST*sizeof(struct sockaddr_in) );
     199           0 :   fd_memset( batch_msg, 0, STEM_BURST*sizeof(struct mmsghdr)     );
     200             : 
     201           0 :   ctx->batch_cnt   = 0UL;
     202           0 :   ctx->batch_iov   = batch_iov;
     203           0 :   ctx->batch_cmsg  = batch_cmsg;
     204           0 :   ctx->batch_sa    = batch_sa;
     205           0 :   ctx->batch_msg   = batch_msg;
     206           0 :   ctx->batch_frame_sz = batch_fsz;
     207           0 :   ctx->tx_scratch0 = tx_scratch;
     208           0 :   ctx->tx_scratch1 = tx_scratch + tx_sz;
     209           0 :   ctx->tx_ptr      = tx_scratch;
     210           0 :   ctx->tx_sock = -1;
     211           0 :   ctx->bind_address = tile->sock.net.bind_address;
     212           0 :   ctx->tick_per_ns  = fd_tempo_tick_per_ns( NULL );
     213           0 :   if( tile->sock.only_recv_lo ) {
     214           0 :     ctx->lo_ifindex = if_nametoindex( "lo" );
     215           0 :     if( FD_UNLIKELY( !ctx->lo_ifindex ) ) {
     216           0 :       FD_LOG_ERR(( "if_nametoindex(lo) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
     217           0 :     }
     218           0 :   }
     219             : 
     220             :   /* Create receive sockets.  Incrementally assign them to file
     221             :      descriptors starting at sock_fd_min. */
     222             : 
     223           0 :   int sock_fd_min = RX_SOCK_FD_MIN;
     224           0 :   ushort udp_port_candidates[] = {
     225           0 :     (ushort)tile->sock.net.legacy_transaction_listen_port,
     226           0 :     (ushort)tile->sock.net.quic_transaction_listen_port,
     227           0 :     (ushort)tile->sock.net.shred_listen_port,
     228           0 :     (ushort)tile->sock.net.gossip_listen_port,
     229           0 :     (ushort)tile->sock.net.repair_client_listen_port,
     230           0 :     (ushort)tile->sock.net.repair_serve_listen_port,
     231           0 :     (ushort)tile->sock.net.votor_quic_client_listen_port,
     232           0 :     (ushort)tile->sock.net.votor_quic_server_listen_port
     233           0 :   };
     234           0 :   static char const * udp_port_links[] = {
     235           0 :     "net_quic",   /* legacy_transaction_listen_port */
     236           0 :     "net_quic",   /* quic_transaction_listen_port */
     237           0 :     "net_shred",  /* shred_listen_port (turbine) */
     238           0 :     "net_gossvf", /* gossip_listen_port */
     239           0 :     "net_shred",  /* shred_listen_port (repair) */
     240           0 :     "net_rserve", /* repair_serve_listen_port */
     241           0 :     "net_votor",  /* votor_quic_client_listen_port */
     242           0 :     "net_votor"   /* votor_quic_server_listen_port */
     243           0 :   };
     244           0 :   static uchar const udp_port_protos[] = {
     245           0 :     DST_PROTO_TPU_UDP,  /* legacy_transaction_listen_port */
     246           0 :     DST_PROTO_TPU_QUIC, /* quic_transaction_listen_port */
     247           0 :     DST_PROTO_SHRED,    /* shred_listen_port (turbine) */
     248           0 :     DST_PROTO_GOSSIP,   /* gossip_listen_port */
     249           0 :     DST_PROTO_REPAIR,   /* shred_listen_port (repair) */
     250           0 :     DST_PROTO_RSERVE,   /* repair_serve_listen_port */
     251           0 :     DST_PROTO_VOTOR,    /* votor_quic_client_listen_port */
     252           0 :     DST_PROTO_VOTOR     /* votor_quic_server_listen_port */
     253           0 :   };
     254           0 :   for( uint candidate_idx=0U; candidate_idx<sizeof(udp_port_candidates)/sizeof(udp_port_candidates[0]); candidate_idx++ ) {
     255           0 :     if( !udp_port_candidates[ candidate_idx ] ) continue;
     256           0 :     uint sock_idx = ctx->sock_cnt;
     257           0 :     if( sock_idx>=FD_SOCK_TILE_MAX_SOCKETS ) FD_LOG_ERR(( "too many sockets" ));
     258           0 :     ushort port = (ushort)udp_port_candidates[ candidate_idx ];
     259             : 
     260           0 :     char const * target_link = udp_port_links[ candidate_idx ];
     261           0 :     ctx->link_rx_map[ sock_idx ] = 0xFF;
     262           0 :     for( ulong j=0UL; j<(tile->out_cnt); j++ ) {
     263           0 :       if( 0==strcmp( topo->links[ tile->out_link_id[ j ] ].name, target_link ) ) {
     264           0 :         ctx->proto_id    [ sock_idx ] = (uchar)udp_port_protos[ candidate_idx ];
     265           0 :         ctx->link_rx_map [ sock_idx ] = (uchar)j;
     266           0 :         ctx->rx_sock_port[ sock_idx ] = (ushort)port;
     267           0 :         break;
     268           0 :       }
     269           0 :     }
     270           0 :     if( ctx->link_rx_map[ sock_idx ]==0xFF ) {
     271             :       /* listen port number has no associated links,
     272             :          i.e. the repair server is disabled, then no net_rserve link. */
     273           0 :       continue;
     274           0 :     }
     275             : 
     276           0 :     int sock_fd = sock_fd_min + (int)sock_idx;
     277           0 :     create_udp_socket( sock_fd, tile->sock.net.bind_address, port, tile->sock.so_rcvbuf, ctx->lo_ifindex );
     278           0 :     ctx->pollfd[ sock_idx ].fd     = sock_fd;
     279           0 :     ctx->pollfd[ sock_idx ].events = POLLIN;
     280           0 :     ctx->sock_cnt++;
     281           0 :   }
     282             : 
     283           0 :   if( tile->sock.only_recv_lo ) return;
     284             : 
     285             :   /* Create transmit socket */
     286             : 
     287           0 :   int tx_sock = socket( AF_INET, SOCK_RAW|SOCK_CLOEXEC, FD_IP4_HDR_PROTOCOL_UDP );
     288           0 :   if( FD_UNLIKELY( tx_sock<0 ) ) {
     289           0 :     FD_LOG_ERR(( "socket(AF_INET,SOCK_RAW|SOCK_CLOEXEC,17) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
     290           0 :   }
     291             : 
     292           0 :   if( FD_UNLIKELY( 0!=setsockopt( tx_sock, SOL_SOCKET, SO_SNDBUF, &tile->sock.so_sndbuf, sizeof(int) ) ) ) {
     293           0 :     FD_LOG_ERR(( "setsockopt(SOL_SOCKET,SO_SNDBUF,%i) failed (%i-%s)", tile->sock.so_sndbuf, errno, fd_io_strerror( errno ) ));
     294           0 :   }
     295             : 
     296           0 :   uchar mcast_ttl = 64;
     297           0 :   if( FD_UNLIKELY( 0!=setsockopt( tx_sock, IPPROTO_IP, IP_MULTICAST_TTL, &mcast_ttl, sizeof(mcast_ttl) ) ) ) {
     298           0 :     FD_LOG_ERR(( "setsockopt(IPPROTO_IP,IP_MULTICAST_TTL,%u) failed (%i-%s)", (uint)mcast_ttl, errno, fd_io_strerror( errno ) ));
     299           0 :   }
     300             : 
     301           0 :   ctx->tx_sock      = tx_sock;
     302           0 :   ctx->bind_address = tile->sock.net.bind_address;
     303           0 : }
     304             : 
     305             : static void
     306             : unprivileged_init( fd_topo_t const *      topo,
     307           0 :                    fd_topo_tile_t const * tile ) {
     308           0 :   fd_sock_tile_t * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
     309             : 
     310           0 :   if( FD_UNLIKELY( tile->out_cnt > MAX_NET_OUTS ) ) {
     311           0 :     FD_LOG_ERR(( "sock tile has %lu out links which exceeds the max (%lu)", tile->out_cnt, MAX_NET_OUTS ));
     312           0 :   }
     313             : 
     314           0 :   ctx->link_rx_cnt = tile->out_cnt;
     315           0 :   for( ulong i=0UL; i<(tile->out_cnt); i++ ) {
     316           0 :     if( 0!=strncmp( topo->links[ tile->out_link_id[ i ] ].name, "net_", 4 ) ) {
     317           0 :       FD_LOG_ERR(( "out link %lu is not a net RX link", i ));
     318           0 :     }
     319           0 :     fd_topo_link_t const * link = &topo->links[ tile->out_link_id[ i ] ];
     320           0 :     ctx->link_rx[ i ].base   = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
     321           0 :     ctx->link_rx[ i ].chunk0 = fd_dcache_compact_chunk0( ctx->link_rx[ i ].base, link->dcache );
     322           0 :     ctx->link_rx[ i ].wmark  = fd_dcache_compact_wmark(  ctx->link_rx[ i ].base, link->dcache, link->mtu );
     323           0 :     ctx->link_rx[ i ].chunk  = ctx->link_rx[ i ].chunk0;
     324           0 :     if( FD_UNLIKELY( link->burst < STEM_BURST ) ) {
     325           0 :       FD_LOG_ERR(( "link %lu dcache burst is too low (%lu<%lu)",
     326           0 :                    tile->out_link_id[ i ], link->burst, STEM_BURST ));
     327           0 :     }
     328           0 :   }
     329             : 
     330           0 :   FD_CHECK_ERR( tile->in_cnt<=MAX_NET_INS, "too many input links" );
     331             : 
     332             :   /* The loopback-only sock tile has no TX links and is not counted */
     333           0 :   ctx->tx_tile_idx = 0UL;
     334           0 :   ctx->tx_tile_cnt = 0UL;
     335           0 :   for( ulong i=0UL; i<topo->tile_cnt; i++ ) {
     336           0 :     fd_topo_tile_t const * t = &topo->tiles[ i ];
     337           0 :     if( strcmp( t->name, "sock" ) || t->sock.only_recv_lo ) continue;
     338           0 :     if( t->id==tile->id ) ctx->tx_tile_idx = ctx->tx_tile_cnt;
     339           0 :     ctx->tx_tile_cnt++;
     340           0 :   }
     341           0 :   ctx->tx_tile_cnt = fd_ulong_max( ctx->tx_tile_cnt, 1UL );
     342             : 
     343           0 :   for( ulong i=0UL; i<(tile->in_cnt); i++ ) {
     344           0 :     if( !strstr( topo->links[ tile->in_link_id[ i ] ].name, "_net" ) ) {
     345           0 :       FD_LOG_ERR(( "in link %lu is not a net TX link", i ));
     346           0 :     }
     347           0 :     fd_topo_link_t const * link = &topo->links[ tile->in_link_id[ i ] ];
     348           0 :     ctx->link_tx[ i ].base   = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
     349           0 :     ctx->link_tx[ i ].chunk0 = fd_dcache_compact_chunk0( ctx->link_tx[ i ].base, link->dcache );
     350           0 :     ctx->link_tx[ i ].wmark  = fd_dcache_compact_wmark(  ctx->link_tx[ i ].base, link->dcache, link->mtu );
     351           0 :   }
     352             : 
     353           0 : }
     354             : 
     355             : /* RX PATH (socket->tango) ********************************************/
     356             : 
     357             : /* FIXME Pace RX polling and interleave it with TX jobs to reduce TX
     358             :          tail latency */
     359             : 
     360             : /* poll_rx_socket does one recvmmsg batch receive on the given socket
     361             :    index.  Returns the number of packets returned by recvmmsg. */
     362             : 
     363             : static ulong
     364             : poll_rx_socket( fd_sock_tile_t *    ctx,
     365             :                 fd_stem_context_t * stem,
     366             :                 uint                sock_idx,
     367             :                 int                 sock_fd,
     368           0 :                 ushort              proto ) {
     369           0 :   ulong  hdr_sz      = sizeof(fd_eth_hdr_t) + sizeof(fd_ip4_hdr_t) + sizeof(fd_udp_hdr_t);
     370           0 :   ulong  payload_max = FD_NET_MTU-hdr_sz;
     371           0 :   uchar  rx_link     = ctx->link_rx_map[ sock_idx ];
     372           0 :   ushort dport       = ctx->rx_sock_port[ sock_idx ];
     373             : 
     374           0 :   fd_sock_link_rx_t * link = ctx->link_rx + rx_link;
     375           0 :   void * const base       = link->base;
     376           0 :   ulong  const chunk0     = link->chunk0;
     377           0 :   ulong  const wmark      = link->wmark;
     378           0 :   ulong        chunk_next = link->chunk;
     379           0 :   uchar *      cmsg_next  = ctx->batch_cmsg;
     380             : 
     381           0 :   for( ulong j=0UL; j<STEM_BURST; j++ ) {
     382           0 :     ctx->batch_iov[ j ].iov_base = (uchar *)fd_chunk_to_laddr( base, chunk_next ) + hdr_sz;
     383           0 :     ctx->batch_iov[ j ].iov_len  = payload_max;
     384           0 :     ctx->batch_msg[ j ].msg_hdr  = (struct msghdr) {
     385           0 :       .msg_iov        = ctx->batch_iov+j,
     386           0 :       .msg_iovlen     = 1,
     387           0 :       .msg_name       = ctx->batch_sa+j,
     388           0 :       .msg_namelen    = sizeof(struct sockaddr_in),
     389           0 :       .msg_control    = cmsg_next,
     390           0 :       .msg_controllen = FD_SOCK_CMSG_MAX,
     391           0 :     };
     392           0 :     cmsg_next += FD_SOCK_CMSG_MAX;
     393             :     /* Speculatively prepare all chunk indexes for a receive.
     394             :        At function exit, chunks into which a packet was received are
     395             :        committed, all others are freed. */
     396           0 :     chunk_next = fd_dcache_compact_next( chunk_next, FD_NET_MTU, chunk0, wmark );
     397           0 :   }
     398             : 
     399           0 :   int msg_cnt = recvmmsg( sock_fd, ctx->batch_msg, STEM_BURST, MSG_DONTWAIT, NULL );
     400           0 :   if( FD_UNLIKELY( msg_cnt<0 ) ) {
     401           0 :     if( FD_UNLIKELY( errno==EINTR ) ) return 0UL;
     402           0 :     if( FD_LIKELY( errno==EAGAIN ) ) return 0UL;
     403             :     /* unreachable if socket is in a valid state */
     404           0 :     FD_LOG_ERR(( "recvmmsg failed (%i-%s)", errno, fd_io_strerror( errno ) ));
     405           0 :   }
     406           0 :   long ts = fd_tickcount();
     407           0 :   ctx->metrics.sys_recvmmsg_cnt++;
     408             : 
     409           0 :   if( FD_UNLIKELY( msg_cnt==0 ) ) return 0UL;
     410             : 
     411             :   /* Only frags published to rx_link may consume rx_link dcache space.
     412             :      Packets that are dropped or redirected leave a hole, so subsequent
     413             :      packets are compacted down to chunk_pub. */
     414           0 :   ulong chunk_pub = link->chunk;
     415             : 
     416           0 :   for( ulong j=0; j<(ulong)msg_cnt; j++ ) {
     417           0 :     uchar * payload         = ctx->batch_iov[ j ].iov_base;
     418           0 :     ulong   payload_sz      = ctx->batch_msg[ j ].msg_len;
     419           0 :     struct sockaddr_in * sa = ctx->batch_msg[ j ].msg_hdr.msg_name;
     420           0 :     ulong frame_sz          = payload_sz + hdr_sz;
     421           0 :     if( FD_UNLIKELY( ctx->lo_ifindex &&
     422           0 :                      (ctx->batch_msg[ j ].msg_hdr.msg_flags & (MSG_TRUNC|MSG_CTRUNC)) ) ) continue;
     423           0 :     ctx->metrics.rx_bytes_total += frame_sz;
     424           0 :     if( FD_UNLIKELY( sa->sin_family!=AF_INET ) ) {
     425             :       /* unreachable */
     426           0 :       FD_LOG_ERR(( "Received packet with unexpected sin_family %i", sa->sin_family ));
     427           0 :     }
     428             : 
     429           0 :     long daddr = -1;
     430           0 :     struct cmsghdr * cmsg = CMSG_FIRSTHDR( &ctx->batch_msg[ j ].msg_hdr );
     431           0 :     if( FD_LIKELY( cmsg ) ) {
     432           0 :       do {
     433           0 :         if( FD_LIKELY( (cmsg->cmsg_level==IPPROTO_IP) &
     434           0 :                        (cmsg->cmsg_type ==IP_PKTINFO) ) ) {
     435           0 :           struct in_pktinfo const * pi = (struct in_pktinfo const *)CMSG_DATA( cmsg );
     436           0 :           daddr = pi->ipi_addr.s_addr;
     437           0 :         }
     438           0 :         cmsg = fd_cmsg_nxthdr( &ctx->batch_msg[ j ].msg_hdr, cmsg );
     439           0 :       } while( FD_UNLIKELY( cmsg ) ); /* optimize for 1 cmsg */
     440           0 :     }
     441           0 :     if( FD_UNLIKELY( daddr<0L ) ) {
     442             :       /* unreachable because IP_PKTINFO was set */
     443           0 :       FD_LOG_ERR(( "Missing IP_PKTINFO on incoming packet" ));
     444           0 :     }
     445             : 
     446           0 :     fd_eth_hdr_t * eth_hdr    = (fd_eth_hdr_t *)( payload-42UL );
     447           0 :     fd_ip4_hdr_t * ip_hdr     = (fd_ip4_hdr_t *)( payload-28UL );
     448           0 :     fd_udp_hdr_t * udp_hdr    = (fd_udp_hdr_t *)( payload- 8UL );
     449           0 :     memset( eth_hdr->dst, 0, 6 );
     450           0 :     memset( eth_hdr->src, 0, 6 );
     451           0 :     eth_hdr->net_type = fd_ushort_bswap( FD_ETH_HDR_TYPE_IP );
     452           0 :     *ip_hdr = (fd_ip4_hdr_t) {
     453           0 :       .verihl      = FD_IP4_VERIHL( 4, 5 ),
     454           0 :       .net_tot_len = fd_ushort_bswap( (ushort)( payload_sz+28UL ) ),
     455           0 :       .ttl         = 1,
     456           0 :       .protocol    = FD_IP4_HDR_PROTOCOL_UDP,
     457           0 :     };
     458           0 :     uint daddr_ = (uint)(ulong)daddr;
     459           0 :     memcpy( ip_hdr->saddr_c, &sa->sin_addr.s_addr, 4 );
     460           0 :     memcpy( ip_hdr->daddr_c, &daddr_,              4 );
     461           0 :     *udp_hdr = (fd_udp_hdr_t) {
     462           0 :       .net_sport = sa->sin_port,
     463           0 :       .net_dport = (ushort)fd_ushort_bswap( (ushort)dport ),
     464           0 :       .net_len   = (ushort)fd_ushort_bswap( (ushort)( payload_sz+8UL ) ),
     465           0 :       .check     = 0
     466           0 :     };
     467             : 
     468           0 :     ctx->metrics.rx_pkt_cnt++;
     469           0 :     ulong chunk = fd_laddr_to_chunk( base, eth_hdr );
     470           0 :     ulong sig   = fd_disco_netmux_sig( sa->sin_addr.s_addr, fd_ushort_bswap( sa->sin_port ), sa->sin_addr.s_addr, proto, hdr_sz );
     471           0 :     ulong tspub = fd_frag_meta_ts_comp( ts );
     472             : 
     473           0 :     if( FD_UNLIKELY( chunk!=chunk_pub ) ) {
     474           0 :       memcpy( fd_chunk_to_laddr( base, chunk_pub ), eth_hdr, frame_sz );
     475           0 :     }
     476           0 :     fd_stem_publish( stem, rx_link, sig, chunk_pub, frame_sz, 0UL, 0UL, tspub );
     477           0 :     chunk_pub = fd_dcache_compact_next( chunk_pub, FD_NET_MTU, chunk0, wmark );
     478           0 :   }
     479             : 
     480             :   /* Rewind the chunk index to the first free index. */
     481           0 :   link->chunk = chunk_pub;
     482           0 :   return (ulong)msg_cnt;
     483           0 : }
     484             : 
     485             : static ulong
     486             : poll_rx( fd_sock_tile_t *    ctx,
     487           0 :          fd_stem_context_t * stem ) {
     488           0 :   ulong pkt_cnt = 0UL;
     489           0 :   if( FD_UNLIKELY( ctx->batch_cnt ) ) {
     490           0 :     FD_LOG_ERR(( "Batch is not clean" ));
     491           0 :   }
     492           0 :   ctx->tx_idle_cnt = 0; /* restart TX polling */
     493           0 :   long wait_ns = ctx->lo_ifindex ? (long)((double)fd_long_max( stem->then-fd_tickcount(), 0L )/ctx->tick_per_ns) : 0L;
     494           0 :   struct timespec ts = { .tv_sec = wait_ns/1000000000L, .tv_nsec = wait_ns%1000000000L };
     495           0 :   if( FD_UNLIKELY( ppoll( ctx->pollfd, ctx->sock_cnt, &ts, NULL )<0 ) ) {
     496           0 :     if( FD_UNLIKELY( errno==EINTR ) ) return 0UL;
     497           0 :     FD_LOG_ERR(( "ppoll failed (%i-%s)", errno, fd_io_strerror( errno ) ));
     498           0 :   }
     499           0 :   for( uint j=0U; j<ctx->sock_cnt; j++ ) {
     500           0 :     if( ctx->pollfd[ j ].revents & (POLLIN|POLLERR) ) {
     501           0 :       pkt_cnt += poll_rx_socket(
     502           0 :         ctx,
     503           0 :         stem,
     504           0 :         j,
     505           0 :         ctx->pollfd[ j ].fd,
     506           0 :         ctx->proto_id[ j ]
     507           0 :       );
     508           0 :     }
     509           0 :     ctx->pollfd[ j ].revents = 0;
     510           0 :   }
     511           0 :   return pkt_cnt;
     512           0 : }
     513             : 
     514             : /* TX PATH (tango->socket) ********************************************/
     515             : 
     516             : static void
     517           0 : flush_tx_batch( fd_sock_tile_t * ctx ) {
     518           0 :   ulong batch_cnt = ctx->batch_cnt;
     519           0 :   for( int j = 0; j < (int)batch_cnt; /* incremented in loop */ ) {
     520           0 :     int remain   = (int)batch_cnt - j;
     521           0 :     int send_cnt = sendmmsg( ctx->tx_sock, ctx->batch_msg + j, (uint)remain, MSG_DONTWAIT );
     522           0 :     if( send_cnt>=0 ) {
     523           0 :       ctx->metrics.sys_sendmmsg_cnt[ FD_METRICS_ENUM_SENDMMSG_RESULT_V_SUCCESS_IDX ]++;
     524           0 :     }
     525           0 :     if( FD_UNLIKELY( send_cnt < remain ) ) {
     526           0 :       ctx->metrics.tx_drop_cnt++;
     527           0 :       if( FD_UNLIKELY( send_cnt < 0 ) ) {
     528           0 :         switch( errno ) {
     529           0 :         case EAGAIN:
     530           0 :         case ENOBUFS:
     531           0 :           ctx->metrics.sys_sendmmsg_cnt[ FD_METRICS_ENUM_SENDMMSG_RESULT_V_BUFFER_FULL_IDX ]++;
     532           0 :           break;
     533           0 :         case EPERM:
     534           0 :           ctx->metrics.sys_sendmmsg_cnt[ FD_METRICS_ENUM_SENDMMSG_RESULT_V_PERMISSION_IDX ]++;
     535           0 :           break;
     536           0 :         case ENETUNREACH:
     537           0 :         case EHOSTUNREACH:
     538           0 :           ctx->metrics.sys_sendmmsg_cnt[ FD_METRICS_ENUM_SENDMMSG_RESULT_V_UNREACHABLE_IDX ]++;
     539           0 :           break;
     540           0 :         case ENONET:
     541           0 :         case ENETDOWN:
     542           0 :         case EHOSTDOWN:
     543           0 :           ctx->metrics.sys_sendmmsg_cnt[ FD_METRICS_ENUM_SENDMMSG_RESULT_V_DOWN_IDX ]++;
     544           0 :           break;
     545           0 :         default:
     546           0 :           ctx->metrics.sys_sendmmsg_cnt[ FD_METRICS_ENUM_SENDMMSG_RESULT_V_OTHER_IDX ]++;
     547             :           /* log with NOTICE, since flushing has a significant negative performance impact */
     548           0 :           FD_LOG_NOTICE(( "sendmmsg failed (%i-%s)", errno, fd_io_strerror( errno ) ));
     549           0 :         }
     550             : 
     551             :         /* first message failed, so skip failing message and continue */
     552           0 :         j++;
     553           0 :       } else {
     554             :         /* send_cnt succeeded, so skip those and also the failing message */
     555           0 :         j += send_cnt + 1;
     556             : 
     557             :         /* add the successful count */
     558           0 :         ctx->metrics.tx_pkt_cnt += (ulong)send_cnt;
     559           0 :         for( int k=j-send_cnt-1; k<j-1; k++ ) ctx->metrics.tx_bytes_total += ctx->batch_frame_sz[ k ];
     560           0 :       }
     561             : 
     562           0 :       continue;
     563           0 :     }
     564             : 
     565             :     /* send_cnt == batch_cnt, so we sent everything */
     566           0 :     ctx->metrics.tx_pkt_cnt += (ulong)send_cnt;
     567           0 :     for( int k=j; k<j+send_cnt; k++ ) ctx->metrics.tx_bytes_total += ctx->batch_frame_sz[ k ];
     568           0 :     break;
     569           0 :   }
     570             : 
     571           0 :   ctx->tx_ptr = ctx->tx_scratch0;
     572           0 :   ctx->batch_cnt = 0;
     573           0 : }
     574             : 
     575             : /* before_frag is called when a new frag has been detected.  Every sock
     576             :    tile reads every TX link, so each sends only its share of packets. */
     577             : 
     578             : static inline int
     579             : before_frag( fd_sock_tile_t * ctx,
     580             :              ulong            in_idx FD_PARAM_UNUSED,
     581             :              ulong            seq    FD_PARAM_UNUSED,
     582           0 :              ulong            sig ) {
     583           0 :   ulong proto = fd_disco_netmux_sig_proto( sig );
     584           0 :   if( FD_UNLIKELY( proto!=DST_PROTO_OUTGOING ) ) return 1;
     585           0 :   if( fd_disco_netmux_sig_hash( sig )%ctx->tx_tile_cnt!=ctx->tx_tile_idx ) return 1;
     586           0 :   return 0; /* continue */
     587           0 : }
     588             : 
     589             : /* during_frag is called when a new frag passed early filtering.
     590             :    Speculatively copies data into a sendmmsg buffer.  (If all tiles
     591             :    respect backpressure could eliminate this copy) */
     592             : 
     593             : static inline void
     594             : during_frag( fd_sock_tile_t * ctx,
     595             :              ulong            in_idx,
     596             :              ulong            seq FD_PARAM_UNUSED,
     597             :              ulong            sig,
     598             :              ulong            chunk,
     599             :              ulong            sz,
     600           0 :              ulong            ctl FD_PARAM_UNUSED ) {
     601           0 :   if( FD_UNLIKELY( chunk<ctx->link_tx[ in_idx ].chunk0 || chunk>ctx->link_tx[ in_idx ].wmark || sz>FD_NET_MTU ) ) {
     602           0 :     FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, ctx->link_tx[ in_idx ].chunk0, ctx->link_tx[ in_idx ].wmark ));
     603           0 :   }
     604             : 
     605           0 :   ctx->parsed.invalid = 0;
     606             : 
     607           0 :   ulong const hdr_min = sizeof(fd_eth_hdr_t)+sizeof(fd_ip4_hdr_t)+sizeof(fd_udp_hdr_t);
     608           0 :   if( FD_UNLIKELY( sz<hdr_min ) ) {
     609             :     /* FIXME support ICMP messages in the future?
     610             :        Defer the error to after_frag, where we know we weren't
     611             :        overrun. */
     612           0 :     ctx->parsed.invalid = 1;
     613           0 :     return;
     614           0 :   }
     615             : 
     616           0 :   uchar const * frame   = fd_chunk_to_laddr_const( ctx->link_tx[ in_idx ].base, chunk );
     617           0 :   ulong         hdr_sz  = fd_disco_netmux_sig_hdr_sz( sig );
     618           0 :   uchar const * payload = frame+hdr_sz;
     619           0 :   if( FD_UNLIKELY( hdr_sz>sz || hdr_sz<hdr_min ) ) {
     620           0 :     FD_LOG_ERR(( "packet from in_idx=%lu corrupt: hdr_sz=%lu total_sz=%lu",
     621           0 :                  in_idx, hdr_sz, sz ));
     622           0 :   }
     623           0 :   ulong payload_sz = sz-hdr_sz;
     624             : 
     625           0 :   fd_ip4_hdr_t const * ip_hdr  = (fd_ip4_hdr_t const *)( frame  +sizeof(fd_eth_hdr_t) );
     626           0 :   fd_udp_hdr_t const * udp_hdr = (fd_udp_hdr_t const *)( payload-sizeof(fd_udp_hdr_t) );
     627           0 :   ctx->parsed.ip_version  = FD_IP4_GET_VERSION( *ip_hdr );
     628           0 :   ctx->parsed.ip_protocol = ip_hdr->protocol;
     629           0 :   if( FD_UNLIKELY( ( ctx->parsed.ip_version !=4                       ) |
     630           0 :                    ( ctx->parsed.ip_protocol!=FD_IP4_HDR_PROTOCOL_UDP ) ) ) {
     631             :     /* sock tile only supports IPv4 UDP for now.  Defer the error to
     632             :        after_frag, where we know we weren't overrun. */
     633           0 :     ctx->parsed.invalid = 1;
     634           0 :     return;
     635           0 :   }
     636             : 
     637           0 :   ulong msg_sz = sizeof(fd_udp_hdr_t) + payload_sz;
     638             : 
     639           0 :   ulong batch_idx = ctx->batch_cnt;
     640           0 :   FD_DCHECK_CRIT( batch_idx<STEM_BURST, "flow control error" );
     641           0 :   struct mmsghdr *     msg  = ctx->batch_msg + batch_idx;
     642           0 :   struct sockaddr_in * sa   = ctx->batch_sa  + batch_idx;
     643           0 :   struct iovec   *     iov  = ctx->batch_iov + batch_idx;
     644           0 :   struct cmsghdr *     cmsg = (void *)( (ulong)ctx->batch_cmsg + batch_idx*FD_SOCK_CMSG_MAX );
     645           0 :   uchar *              buf  = ctx->tx_ptr;
     646             : 
     647           0 :   *iov = (struct iovec) {
     648           0 :     .iov_base = buf,
     649           0 :     .iov_len  = msg_sz,
     650           0 :   };
     651           0 :   sa->sin_family      = AF_INET;
     652           0 :   sa->sin_addr.s_addr = FD_LOAD( uint, ip_hdr->daddr_c );
     653           0 :   sa->sin_port        = 0; /* ignored */
     654             : 
     655           0 :   cmsg->cmsg_level = IPPROTO_IP;
     656           0 :   cmsg->cmsg_type  = IP_PKTINFO;
     657           0 :   cmsg->cmsg_len   = CMSG_LEN( sizeof(struct in_pktinfo) );
     658           0 :   struct in_pktinfo * pi = (struct in_pktinfo *)CMSG_DATA( cmsg );
     659           0 :   pi->ipi_ifindex         = 0;
     660           0 :   pi->ipi_addr.s_addr     = 0;
     661           0 :   pi->ipi_spec_dst.s_addr = fd_uint_if( !!ip_hdr->saddr, ip_hdr->saddr, ctx->bind_address );
     662             : 
     663           0 :   *msg = (struct mmsghdr) {
     664           0 :     .msg_hdr = {
     665           0 :       .msg_name       = sa,
     666           0 :       .msg_namelen    = sizeof(struct sockaddr_in),
     667           0 :       .msg_iov        = iov,
     668           0 :       .msg_iovlen     = 1,
     669           0 :       .msg_control    = cmsg,
     670           0 :       .msg_controllen = CMSG_LEN( sizeof(struct in_pktinfo) )
     671           0 :     }
     672           0 :   };
     673             : 
     674           0 :   memcpy( buf, udp_hdr, sizeof(fd_udp_hdr_t) );
     675           0 :   fd_memcpy( buf+sizeof(fd_udp_hdr_t), payload, payload_sz );
     676           0 :   ctx->batch_frame_sz[ batch_idx ] = sz;
     677           0 : }
     678             : 
     679             : /* after_frag is called when a frag was copied into a sendmmsg buffer. */
     680             : 
     681             : static void
     682             : after_frag( fd_sock_tile_t *    ctx,
     683             :             ulong               in_idx,
     684             :             ulong               seq    FD_PARAM_UNUSED,
     685             :             ulong               sig    FD_PARAM_UNUSED,
     686             :             ulong               sz,
     687             :             ulong               tsorig FD_PARAM_UNUSED,
     688             :             ulong               tspub  FD_PARAM_UNUSED,
     689           0 :             fd_stem_context_t * stem   FD_PARAM_UNUSED ) {
     690             :   /* Commit the packet added in during_frag.  during_frag defers fatal
     691             :      errors to here (where we know we weren't overrun) by setting the
     692             :      invalid flag. */
     693             : 
     694           0 :   if( FD_UNLIKELY( ctx->parsed.invalid ) ) {
     695           0 :     ulong const hdr_min = sizeof(fd_eth_hdr_t)+sizeof(fd_ip4_hdr_t)+sizeof(fd_udp_hdr_t);
     696           0 :     if( FD_UNLIKELY( sz<hdr_min ) ) {
     697             :       /* FIXME support ICMP messages in the future? */
     698           0 :       FD_LOG_ERR(( "packet too small %lu (in_idx=%lu)", sz, in_idx ));
     699           0 :     }
     700           0 :     if( FD_UNLIKELY( ( ctx->parsed.ip_version !=4                       ) |
     701           0 :                      ( ctx->parsed.ip_protocol!=FD_IP4_HDR_PROTOCOL_UDP ) ) ) {
     702           0 :       FD_LOG_ERR(( "packet from in_idx=%lu: sock tile only supports IPv4 UDP for now", in_idx ));
     703           0 :     }
     704           0 :   }
     705             : 
     706           0 :   ctx->tx_idle_cnt = 0;
     707           0 :   ctx->batch_cnt++;
     708             :   /* Technically leaves a gap.  sz is always larger than the payload
     709             :      written to tx_ptr because Ethernet & IPv4 headers were stripped. */
     710           0 :   ctx->tx_ptr += fd_ulong_align_up( sz, FD_CHUNK_ALIGN );
     711             : 
     712           0 :   if( ctx->batch_cnt >= STEM_BURST ) {
     713           0 :     flush_tx_batch( ctx );
     714           0 :   }
     715           0 : }
     716             : 
     717             : /* End TX path ********************************************************/
     718             : 
     719             : static void
     720           0 : metrics_write( fd_sock_tile_t * ctx ) {
     721           0 :   FD_MCNT_SET( SOCK, SYSCALL_RX,              ctx->metrics.sys_recvmmsg_cnt     );
     722           0 :   FD_MCNT_ENUM_COPY( SOCK, SYSCALL_TX,        ctx->metrics.sys_sendmmsg_cnt     );
     723           0 :   FD_MCNT_SET( SOCK, PKT_RX,                  ctx->metrics.rx_pkt_cnt           );
     724           0 :   FD_MCNT_SET( SOCK, PKT_TX,                  ctx->metrics.tx_pkt_cnt           );
     725           0 :   FD_MCNT_SET( SOCK, PKT_TX_FAILED,           ctx->metrics.tx_drop_cnt          );
     726           0 :   FD_MCNT_SET( SOCK, PKT_TX_BYTES,            ctx->metrics.tx_bytes_total       );
     727           0 :   FD_MCNT_SET( SOCK, PKT_RX_BYTES,            ctx->metrics.rx_bytes_total       );
     728           0 : }
     729             : 
     730             : /* after_credit is called every stem iteration when there are enough
     731             :    flow control credits to publish a burst of fragments. */
     732             : 
     733             : static inline void
     734             : after_credit( fd_sock_tile_t *    ctx,
     735             :               fd_stem_context_t * stem,
     736             :               int *               poll_in FD_PARAM_UNUSED,
     737           0 :               int *               charge_busy ) {
     738           0 :   if( ctx->lo_ifindex ) {
     739             :     /* Housekeeping cannot publish progress while the tile sleeps. */
     740           0 :     FD_COMPILER_MFENCE();
     741           0 :     metrics_write( ctx );
     742           0 :     FD_MGAUGE_SET( TILE, HEARTBEAT_TIME, (ulong)fd_log_wallclock() );
     743           0 :     for( ulong i=0UL; i<ctx->link_rx_cnt; i++ ) {
     744           0 :       fd_mcache_seq_update( fd_mcache_seq_laddr( stem->mcaches[ i ] ), stem->seqs[ i ] );
     745           0 :     }
     746           0 :     FD_COMPILER_MFENCE();
     747           0 :     *charge_busy |= poll_rx( ctx, stem )!=0UL;
     748           0 :     return;
     749           0 :   }
     750           0 :   if( ctx->tx_idle_cnt > 512 ) {
     751           0 :     if( ctx->batch_cnt ) {
     752           0 :       flush_tx_batch( ctx );
     753           0 :     }
     754           0 :     ulong pkt_cnt = poll_rx( ctx, stem );
     755           0 :     *charge_busy = pkt_cnt!=0;
     756           0 :   }
     757           0 :   ctx->tx_idle_cnt++;
     758           0 : }
     759             : 
     760             : static ulong
     761             : rlimit_file_cnt( fd_topo_t const *      topo,
     762           0 :                  fd_topo_tile_t const * tile ) {
     763           0 :   fd_sock_tile_t * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
     764           0 :   return RX_SOCK_FD_MIN + ctx->sock_cnt;
     765           0 : }
     766             : 
     767             : #define STEM_NEVER_PARK             1
     768           0 : #define STEM_CALLBACK_CONTEXT_TYPE  fd_sock_tile_t
     769           0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_sock_tile_t)
     770             : 
     771           0 : #define STEM_LAZY ((long)10e6) /* 10ms */
     772             : 
     773           0 : #define STEM_CALLBACK_METRICS_WRITE       metrics_write
     774           0 : #define STEM_CALLBACK_AFTER_CREDIT        after_credit
     775           0 : #define STEM_CALLBACK_BEFORE_FRAG         before_frag
     776           0 : #define STEM_CALLBACK_DURING_FRAG         during_frag
     777           0 : #define STEM_CALLBACK_AFTER_FRAG          after_frag
     778             : 
     779             : #include "../../stem/fd_stem.c"
     780             : 
     781             : fd_topo_run_tile_t fd_tile_sock = {
     782             :   .name                     = "sock",
     783             :   .rlimit_file_cnt_fn       = rlimit_file_cnt,
     784             :   .populate_allowed_seccomp = populate_allowed_seccomp,
     785             :   .populate_allowed_fds     = populate_allowed_fds,
     786             :   .scratch_align            = scratch_align,
     787             :   .scratch_footprint        = scratch_footprint,
     788             :   .privileged_init          = privileged_init,
     789             :   .unprivileged_init        = unprivileged_init,
     790             :   .run                      = stem_run,
     791             : };

Generated by: LCOV version 1.14