net: fd_xdp_ring helpers

Prompted by a bug reported by @colinlyguo

Author: akhinvasara-jumptrading

src/disco/net/xdp/fd_xdp_tile.c

@@ -397,27 +397,19 @@ net_check_gre_interface_exists( fd_net_ctx_t * ctx ) {
 }
 
 
-/* net_tx_ready returns 1 if the current XSK is ready to submit a TX send
-   job.  If the XSK is blocked for sends, returns 0.  Reasons for block
-   include:
-   - No XSK TX buffer is available
-   - XSK TX ring is full */
+/* net_tx_ready returns 1 if we can submit a job to this TX ring, and 0 otherwise.
+   Reasons for block include:
+   - No TX buffer is available (free ring empty)
+   - TX ring is full
+
+   tx_ring: pointer to the XDP TX ring
+   free_ring: pointer to the free TX ring */
 
 static int
-net_tx_ready( fd_net_ctx_t * ctx,
-              uint           xsk_idx ) {
-  fd_xsk_t *           xsk     = &ctx->xsk[ xsk_idx ];
-  fd_xdp_ring_t *      tx_ring = &xsk->ring_tx;
-  fd_net_free_ring_t * free    = &ctx->free_tx;
-  if( free->prod == free->cons ) return 0; /* drop */
-
-  /* If potentially stale cached_cons says there is space,
-     there is definitely space */
-  if( tx_ring->cached_prod - tx_ring->cached_cons >= tx_ring->depth ) return 1;
-
-  /* read the fseq, and update our cache */
-  tx_ring->cached_cons = FD_VOLATILE_CONST( *tx_ring->cons );
-  if( tx_ring->cached_prod - tx_ring->cached_cons >= tx_ring->depth ) return 0; /* drop */
+net_tx_ready( fd_xdp_ring_t *      tx_ring,
+              fd_net_free_ring_t * free_ring ) {
+  if( FD_UNLIKELY( free_ring->prod == free_ring->cons ) ) return 0; /* drop - no free buffers */
+  if( FD_UNLIKELY( fd_xdp_ring_full( tx_ring ) ) )        return 0; /* drop - tx ring full */
   return 1;
 }
 
@@ -483,17 +475,7 @@ net_tx_periodic_wakeup( fd_net_ctx_t * ctx,
                         long           now,
                         int *          charge_busy ) {
   fd_xdp_ring_t * tx_ring        = &ctx->xsk[ xsk_idx ].ring_tx;
-  uint            tx_prod        = tx_ring->cached_prod;
-  uint            tx_cons        = tx_ring->cached_cons;
-
-  int             tx_ring_empty  = tx_prod==tx_cons;
-  /* If we already think tx_ring_empty, it's definitely empty.
-     But if not, we should update our view of what kernel has consumed. */
-  if( FD_LIKELY( !tx_ring_empty ) ) {
-    tx_cons       = tx_ring->cached_cons = FD_VOLATILE_CONST( *tx_ring->cons );
-    tx_ring_empty = tx_prod==tx_cons;
-  }
-
+  int             tx_ring_empty  = fd_xdp_ring_empty( tx_ring, FD_XDP_RING_ROLE_PROD );
   if( fd_net_flusher_check( ctx->tx_flusher+xsk_idx, now, tx_ring_empty ) ) {
     net_tx_wakeup( ctx, &ctx->xsk[ xsk_idx ], charge_busy );
     fd_net_flusher_wakeup( ctx->tx_flusher+xsk_idx, now );
@@ -712,14 +694,14 @@ before_frag( fd_net_ctx_t * ctx,
 
   /* Skip if TX is blocked */
 
-  if( FD_UNLIKELY( !net_tx_ready( ctx, xsk_idx ) ) ) {
+  fd_xsk_t *           xsk  = &ctx->xsk[ xsk_idx ];
+  fd_net_free_ring_t * free = &ctx->free_tx;
+  if( FD_UNLIKELY( !net_tx_ready( &xsk->ring_tx, free ) ) ) {
     ctx->metrics.tx_full_fail_cnt++;
     return 1;
   }
 
   /* Allocate buffer for receive */
-
-  fd_net_free_ring_t * free      = &ctx->free_tx;
   ulong                alloc_seq = free->cons;
   void *               frame     = (void *)free->queue[ alloc_seq % free->depth ];
   free->cons = fd_seq_inc( alloc_seq, 1UL );
@@ -1057,6 +1039,7 @@ net_comp_event( fd_net_ctx_t * ctx,
   uint            comp_mask  = comp_ring->depth - 1U;
   ulong           frame      = FD_VOLATILE_CONST( comp_ring->frame_ring[ comp_seq&comp_mask ] );
   ulong const     frame_mask = FD_NET_MTU - 1UL;
+  FD_STATIC_ASSERT( FD_ULONG_IS_POW2( FD_NET_MTU ), "FD_NET_MTU must be a power of two" );
   if( FD_UNLIKELY( frame+FD_NET_MTU > ctx->umem_sz ) ) {
     FD_LOG_ERR(( "Bounds check failed: frame=0x%lx umem_sz=0x%lx",
                  frame, (ulong)ctx->umem_sz ));
@@ -1101,28 +1084,15 @@ net_rx_event( fd_net_ctx_t * ctx,
   /* Check if we have space in the fill ring to free the frame */
 
   fd_xdp_ring_t * fill_ring  = &xsk->ring_fr;
-  uint            fill_depth = fill_ring->depth;
-  uint            fill_mask  = fill_depth-1U;
-  ulong           frame_mask = FD_NET_MTU - 1UL;
-  uint            fill_prod  = fill_ring->cached_prod;
-  uint            fill_cons  = fill_ring->cached_cons;
-
-  /* If cached_cons suggests there may not be space in the fill ring,
-     refresh from fseq and check again. Else, skip the fseq access */
-
-  if( FD_UNLIKELY( fill_prod-fill_cons >= fill_depth ) ) {
-    fill_cons = fill_ring->cached_cons = FD_VOLATILE_CONST( *fill_ring->cons );
-    if( FD_UNLIKELY( fill_prod-fill_cons >= fill_depth ) ) {
-      ctx->metrics.rx_fill_blocked_cnt++;
-      return; /* blocked */
-    }
+  if( FD_UNLIKELY( fd_xdp_ring_full( fill_ring ) ) ) {
+    ctx->metrics.rx_fill_blocked_cnt++;
+    return; /* blocked */
   }
 
   /* Pass it to the receive handler */
 
   uint freed_chunk = (uint)( ctx->umem_chunk0 + (frame.addr>>FD_CHUNK_LG_SZ) );
   net_rx_packet( ctx, frame.addr, frame.len, &freed_chunk );
-
   FD_COMPILER_MFENCE();
   rx_ring->cached_cons = rx_seq+1U;
 
@@ -1135,7 +1105,12 @@ net_rx_event( fd_net_ctx_t * ctx,
     FD_LOG_CRIT(( "mcache corruption detected: chunk=%u chunk0=%u wmark=%u",
                   freed_chunk, ctx->umem_chunk0, ctx->umem_wmark ));
   }
-  ulong freed_off = (freed_chunk - ctx->umem_chunk0)<<FD_CHUNK_LG_SZ;
+
+  FD_STATIC_ASSERT( FD_ULONG_IS_POW2( FD_NET_MTU ), "FD_NET_MTU must be a power of two" );
+  uint  fill_prod  = fill_ring->cached_prod;
+  uint  fill_mask  = (fill_ring->depth)-1U;
+  ulong frame_mask = FD_NET_MTU - 1UL;
+  ulong freed_off  = (freed_chunk - ctx->umem_chunk0)<<FD_CHUNK_LG_SZ;
   fill_ring->frame_ring[ fill_prod&fill_mask ] = freed_off & (~frame_mask);
   fill_ring->cached_prod = fill_prod+1U;
 }
@@ -1168,31 +1143,20 @@ before_credit( fd_net_ctx_t *      ctx,
 
   net_tx_periodic_wakeup( ctx, rr_idx, fd_tickcount(), charge_busy );
 
-  uint rx_cons = rr_xsk->ring_rx.cached_cons;
-  uint rx_prod = rr_xsk->ring_rx.cached_prod; /* might be stale */
-  if( FD_UNLIKELY( rx_cons==rx_prod ) ) {
-    rx_prod = rr_xsk->ring_rx.cached_prod = FD_VOLATILE_CONST( *rr_xsk->ring_rx.prod );
-  }
-
-  if( rx_cons!=rx_prod ) {
+  /* Fire RX event if we have RX desc avail */
+  if( !fd_xdp_ring_empty( &rr_xsk->ring_rx, FD_XDP_RING_ROLE_CONS ) ) {
     *charge_busy = 1;
-    net_rx_event( ctx, rr_xsk, rx_cons );
+    net_rx_event( ctx, rr_xsk, rr_xsk->ring_rx.cached_cons );
   } else {
     net_rx_wakeup( ctx, rr_xsk, charge_busy );
     ctx->rr_idx++;
     ctx->rr_idx = fd_uint_if( ctx->rr_idx>=ctx->xsk_cnt, 0, ctx->rr_idx );
   }
 
-  uint comp_cons = rr_xsk->ring_cr.cached_cons;
-  uint comp_prod = rr_xsk->ring_cr.cached_prod; /* might be stale */
-  if( FD_UNLIKELY( comp_cons==comp_prod ) ) {
-    comp_prod = rr_xsk->ring_cr.cached_prod = FD_VOLATILE_CONST( *rr_xsk->ring_cr.prod );
-  }
-
-  if( comp_cons!=comp_prod ) {
+  /* Fire comp event if we have comp desc avail */
+  if( !fd_xdp_ring_empty( &rr_xsk->ring_cr, FD_XDP_RING_ROLE_CONS ) ) {
     *charge_busy = 1;
-    rr_xsk->ring_cr.cached_prod = comp_prod;
-    net_comp_event( ctx, rr_xsk, comp_cons );
+    net_comp_event( ctx, rr_xsk, rr_xsk->ring_cr.cached_cons );
   }
 }
 

src/disco/net/xdp/test_xdp_tile.c

@@ -147,6 +147,89 @@ append_netdev( fd_net_ctx_t * ctx,
   ctx->netdev_tbl.hdr->dev_cnt   = (ushort)(idx + (ushort)1);
 }
 
+/* Test net_tx_ready for all code paths */
+static void
+test_net_tx_ready( void ) {
+  fd_xdp_ring_t tx_ring;
+  uint tx_prod_val = 0;
+  uint tx_cons_val = 0;
+
+  fd_net_free_ring_t free_ring;
+  ulong free_queue[128];
+
+  /* Setup tx_ring */
+  tx_ring.prod = &tx_prod_val;
+  tx_ring.cons = &tx_cons_val;
+  tx_ring.depth = 128;
+  tx_ring.cached_prod = 0;
+  tx_ring.cached_cons = 0;
+
+  /* Setup free_ring */
+  free_ring.queue = free_queue;
+  free_ring.depth = 128;
+  free_ring.prod = 0;
+  free_ring.cons = 0;
+
+  /* Both ready - should return 1 */
+  tx_prod_val = tx_ring.cached_prod = 100;
+  tx_cons_val = tx_ring.cached_cons = 90;
+  free_ring.prod = 10;
+  free_ring.cons = 5;
+  FD_TEST( net_tx_ready( &tx_ring, &free_ring ) == 1 );
+
+  /* Free ring empty (prod == cons) - should return 0 */
+  tx_prod_val = tx_ring.cached_prod = 100;
+  tx_cons_val = tx_ring.cached_cons = 90;
+  free_ring.prod = free_ring.cons = 20; /* empty */
+  FD_TEST( net_tx_ready( &tx_ring, &free_ring ) == 0 );
+
+  /* TX ring full - should return 0 */
+  tx_prod_val = tx_ring.cached_prod = 200;
+  tx_cons_val = tx_ring.cached_cons = 72; /* 200 - 72 = 128, exactly depth */
+  free_ring.prod = 30;
+  free_ring.cons = 25;
+  FD_TEST( net_tx_ready( &tx_ring, &free_ring ) == 0 );
+
+  /* TX ring appears full with stale cache, but actually not full */
+  tx_ring.cached_prod = 300;
+  tx_ring.cached_cons = 172; /* 300 - 172 = 128, appears full */
+  tx_prod_val = 300;
+  tx_cons_val = 200; /* consumer made progress, 300 - 200 = 100 */
+  free_ring.prod = 40;
+  free_ring.cons = 35;
+  FD_TEST( net_tx_ready( &tx_ring, &free_ring ) == 1 );
+  FD_TEST( tx_ring.cached_cons == 200 ); /* verify cache was refreshed */
+
+  /* Both blocking conditions (free empty AND tx full) */
+  tx_prod_val = tx_ring.cached_prod = 500;
+  tx_cons_val = tx_ring.cached_cons = 372; /* 500 - 372 = 128, full */
+  free_ring.prod = free_ring.cons = 60; /* empty*/
+  FD_TEST( net_tx_ready( &tx_ring, &free_ring ) == 0 );
+
+  /* Edge case - tx ring with one slot available */
+  tx_prod_val = tx_ring.cached_prod = 600;
+  tx_cons_val = tx_ring.cached_cons = 473; /* 600 - 473 = 127, one slot */
+  free_ring.prod = 70;
+  free_ring.cons = 69;
+  FD_TEST( net_tx_ready( &tx_ring, &free_ring ) == 1 );
+
+  /* Edge case - free ring with one buffer available */
+  tx_prod_val = tx_ring.cached_prod = 700;
+  tx_cons_val = tx_ring.cached_cons = 690;
+  free_ring.prod = 81;
+  free_ring.cons = 80; /* one buffer */
+  FD_TEST( net_tx_ready( &tx_ring, &free_ring ) == 1 );
+
+  /* TX ring empty (should be ready) */
+  tx_prod_val = tx_ring.cached_prod = 800;
+  tx_cons_val = tx_ring.cached_cons = 800;
+  free_ring.prod = 110;
+  free_ring.cons = 105;
+  FD_TEST( net_tx_ready( &tx_ring, &free_ring ) == 1 );
+
+  FD_LOG_NOTICE(( "test_net_tx_ready: pass" ));
+}
+
 static void
 setup_netdev_table( fd_net_ctx_t * ctx ) {
   ctx->netdev_tbl.hdr->dev_cnt = 0;
@@ -191,6 +274,8 @@ main( int     argc,
       char ** argv ) {
   fd_boot( &argc, &argv );
 
+  test_net_tx_ready();
+
   ulong cpu_idx = fd_tile_cpu_id( fd_tile_idx() );
   if( cpu_idx>fd_shmem_cpu_cnt() ) cpu_idx = 0UL;
   ulong part_max = fd_wksp_part_max_est( SCRATCH_MAX, 64UL );

src/waltz/xdp/Local.mk

@@ -4,6 +4,9 @@ ifdef FD_HAS_HOSTED
 ifdef FD_HAS_LINUX
 $(call add-objs,fd_xsk fd_xdp1 fd_xdp_redirect_user,fd_waltz)
 
+$(call make-unit-test,test_xsk,test_xsk,fd_waltz fd_util)
+$(call run-unit-test,test_xsk)
+
 $(call make-unit-test,test_xdp_ebpf,test_xdp_ebpf,fd_waltz fd_util)
 $(call run-unit-test,test_xdp_ebpf)
 endif # FD_HAS_LINUX

src/waltz/xdp/fd_xsk.h

@@ -153,6 +153,40 @@ struct __attribute__((aligned(64UL))) fd_xdp_ring {
 };
 typedef struct fd_xdp_ring fd_xdp_ring_t;
 
+#define FD_XDP_RING_ROLE_PROD 0
+#define FD_XDP_RING_ROLE_CONS 1
+
+/* fd_xdp_ring_empty returns 1 if the ring is empty, 0 otherwise.
+   'role' is FD_XDP_RING_ROLE_PROD if userspace is the producer (fill, tx),
+   and FD_XDP_RING_ROLE_CONS if userspace is the consumer (rx, completion). */
+
+static inline int
+fd_xdp_ring_empty( fd_xdp_ring_t * ring, uint role ) {
+  if( role == FD_XDP_RING_ROLE_PROD ) {
+    /* If potentially stale cached_cons says everything consumed,
+       it's definitely empty. Else, refresh cached seq. */
+    if( FD_UNLIKELY( ring->cached_prod == ring->cached_cons ) ) return 1;
+    ring->cached_cons = FD_VOLATILE_CONST( *ring->cons );
+  } else {
+    /* If potentially stale cached_prod says we have more to read,
+       it's definitely non-empty. Else, refresh cached seq. */
+    if( FD_LIKELY( ring->cached_cons < ring->cached_prod ) ) return 0;
+    ring->cached_prod = FD_VOLATILE_CONST( *ring->prod );
+  }
+  return ring->cached_prod == ring->cached_cons;
+}
+
+/* fd_xdp_ring_full returns 1 if the ring is full, 0 otherwise.
+   Assumes caller is the producer to this ring (fill, tx). */
+static inline int
+fd_xdp_ring_full( fd_xdp_ring_t * ring ) {
+  /* If potentially stale cached_cons says we have more space,
+      it's definitely not full. Else, refresh cached seq. */
+  if( FD_LIKELY( ring->cached_prod - ring->cached_cons < ring->depth ) ) return 0;
+  ring->cached_cons = FD_VOLATILE_CONST( *ring->cons );
+  return ring->cached_prod - ring->cached_cons >= ring->depth;
+}
+
 /* fd_xsk_params_t: Memory layout parameters of XSK.
    Can be retrieved using fd_xsk_get_params() */