feat(shred,repair): shred_repair link

Author: lidatong

src/app/fdctl/topos/fd_firedancer.c

@@ -149,9 +149,10 @@ static uchar const fd_bmtree_node_prefix[32UL] __attribute__((aligned(32))) = "\
 struct __attribute__((packed)) fd_bmtree_node {
   uchar hash[ 32 ]; /* Last bytes may not be meaningful */
 };
-
 typedef struct fd_bmtree_node fd_bmtree_node_t;
 
+FD_STATIC_ASSERT( sizeof(fd_bmtree_node_t) == 32, update FD_SHRED_MERKLE_ROOT_SZ );
+
 /* bmtree_hash_leaf computes `SHA-256(prefix|data), where prefix is the
    first prefix_sz bytes of fd_bmtree_leaf_prefix.  prefix_sz is
    typically FD_BMTREE_LONG_PREFIX_SZ or FD_BMTREE_SHORT_PREFIX_SZ.

src/ballet/bmtree/fd_bmtree.h

@@ -149,6 +149,13 @@ typedef uchar fd_shred_merkle_t[FD_SHRED_MERKLE_NODE_SZ];
 /* Maximum number of data shreds in a slot, also maximum number of parity shreds in a slot */
 #define FD_SHRED_MAX_PER_SLOT (1 << 15UL) /* 32,768 shreds */
 
+/* Many static bounds are specified around the assumption that this is a
+   protocol limit on the max number of shreds in a slot. If this limit
+   changes, all the relevant usages in other areas of the Firedancer
+   codebase should be updated before modifying this assertion. */
+
+FD_STATIC_ASSERT( FD_SHRED_MAX_PER_SLOT == 32768, check all usages before changing this limit! );
+
 /* 36,536,320 bytes per slot */
 #define FD_SHRED_DATA_PAYLOAD_MAX_PER_SLOT (FD_SHRED_DATA_PAYLOAD_MAX * FD_SHRED_MAX_PER_SLOT)
 

src/ballet/shred/fd_shred.h

@@ -28,22 +28,32 @@
 /* FD_NET_MTU is the max full packet size, with ethernet, IP, and UDP
    headers that can go in or out of the net tile.  2048 is the maximum
    XSK entry size, so this value follows naturally. */
+
 #define FD_NET_MTU (2048UL)
 
 /* FD_TPU_MTU is the max serialized byte size of a txn sent over TPU.
 
    This is minimum MTU of IPv6 packet - IPv6 header - UDP header
                                  1280 -          40 -          8 */
+
 #define FD_TPU_MTU (1232UL)
 
 /* FD_GOSSIP_MTU is the max sz of a gossip packet which is the same as
    above. */
+
 #define FD_GOSSIP_MTU (FD_TPU_MTU)
 
 /* FD_SHRED_STORE_MTU is the size of an fd_shred34_t (statically
    asserted in fd_shred_tile.c). */
+
 #define FD_SHRED_STORE_MTU (41792UL)
 
+/* FD_SHRED_REPAIR_MTU is the maximum size of a frag on the shred_repair
+   link.  This is the size of a data shred header + merkle root. */
+
+#define FD_SHRED_REPAIR_MTU (FD_SHRED_DATA_HEADER_SZ + FD_SHRED_MERKLE_ROOT_SZ)
+FD_STATIC_ASSERT( FD_SHRED_REPAIR_MTU == 120 , update FD_SHRED_REPAIR_MTU );
+
 #define FD_NETMUX_SIG_MIN_HDR_SZ    ( 42UL) /* The default header size, which means no vlan tags and no IP options. */
 #define FD_NETMUX_SIG_IGNORE_HDR_SZ (102UL) /* Outside the allowable range, but still fits in 4 bits when compressed */
 
@@ -123,29 +133,61 @@ fd_disco_replay_old_sig( ulong slot,
 FD_FN_CONST static inline ulong fd_disco_replay_old_sig_flags( ulong sig ) { return (sig & 0xFFUL); }
 FD_FN_CONST static inline ulong fd_disco_replay_old_sig_slot( ulong sig ) { return (sig >> 8); }
 
+/* fd_disco_shred_repair_sig constructs a sig for the shred_repair link.
+   The encoded fields vary depending on the type of the sig.  The
+   diagram below describes the encoding.
+
+   skip (1) | slot (32) | fec_set_idx (15) | is_code (1) | shred_idx or data_cnt (15)
+   [63]     | [31, 62]  | [16, 30]         | [15]        | [0, 14]
+
+   The first bit of the sig indicates whether it is ok to skip the frag.
+   This is the case when the frag is a shred header and the necessary
+   information for the link is fully encoded in the remaining bits of
+   the sig.  If skip is 0, readers must ignore the rest of the sig and
+   process the frag.
+
+   If skip = 1, the next 32 bits [31, 62] describe the slot number. Note
+   if the slot number saturates 32 bits (ie. slot >= UINT_MAX) then skip
+   would be 0 in the sig, so the slot number is only encoded in 32 bits
+   within the sig when it is correct to do so.
+
+   The following 15 bits [16, 30] describe the fec_set_idx.  This is a
+   15-bit value because shreds are bounded to 2^15 per slot, so in the
+   worst case there is an independent FEC set for every shred, which
+   results in at most 2^15 FEC sets per slot.
+
+   The next bit [15] describes whether it is a coding shred (is_code).
+   If is_code = 0, the sig describes a data shred, and the last 15 bits
+   [0, 14] encode the shred_idx.  If is_code = 1, the sig describes a
+   coding shred, and the last 15 bits encode the data_cnt.
+
+   When type is 1, the sig describes a completed FEC set.  In this case,
+   the second bit describes whether the FEC set completes the entry
+   batch, which will be true if the last data shred in the FEC set is
+   marked with a DATA_COMPLETES flag (FIXME this is not invariant in the
+   protocol yet).  As with coding frags, the last 15 bits describe the
+   data_cnt.  The frag will contain the full shred header of the last
+   data shred in the FEC set, as well the merkle root and chained merkle
+   root of the FEC set. */
+
+/* TODO this shred_repair_sig can be greatly simplified when FEC sets
+   are uniformly coding shreds and fixed size. */
+
 FD_FN_CONST static inline ulong
-fd_disco_shred_replay_sig( ulong slot,
-                           uint  shred_idx,
-                           uint  fec_set_idx,
-                           int   is_code,
-                           int   completes ) {
-
-  /* | 32 LSB of slot | 15 LSB of shred_idx | 15 LSB of fec_idx | 1 bit of shred data/code type | 1 bit if shred completes the fec set |
-     | slot[32,63]    | shred_idx[17,32]     | fec_idx[2,16]    | is_parity[1]                  | is_complete[0]                       | */
-
-  ulong slot_ul        = fd_ulong_min( (ulong)slot,        (ulong)UINT_MAX              );
-  ulong shred_idx_ul   = fd_ulong_min( (ulong)shred_idx,   (ulong)FD_SHRED_MAX_PER_SLOT );
-  ulong fec_set_idx_ul = fd_ulong_min( (ulong)fec_set_idx, (ulong)FD_SHRED_MAX_PER_SLOT );
-  ulong is_code_ul     = (ulong)is_code;
-  ulong completes_ul   = (ulong)completes;
-  return slot_ul << 32 | shred_idx_ul << 17 | fec_set_idx_ul << 2 | is_code_ul << 1 | completes_ul;
+fd_disco_shred_repair_sig( ulong slot, uint fec_set_idx, int is_code, uint shred_idx_or_data_cnt ) {
+  return 1 | slot << 31 | fec_set_idx << 16 | (ulong)is_code << 15 | shred_idx_or_data_cnt;
 }
 
-FD_FN_CONST static inline ulong fd_disco_shred_replay_sig_slot       ( ulong sig ) { return       fd_ulong_extract    ( sig, 32, 63 ); }
-FD_FN_CONST static inline uint  fd_disco_shred_replay_sig_shred_idx  ( ulong sig ) { return (uint)fd_ulong_extract    ( sig, 17, 31 ); }
-FD_FN_CONST static inline uint  fd_disco_shred_replay_sig_fec_set_idx( ulong sig ) { return (uint)fd_ulong_extract    ( sig, 2, 16  ); }
-FD_FN_CONST static inline int   fd_disco_shred_replay_sig_is_code    ( ulong sig ) { return       fd_ulong_extract_bit( sig, 1      ); }
-FD_FN_CONST static inline int   fd_disco_shred_replay_sig_completes  ( ulong sig ) { return       fd_ulong_extract_bit( sig, 0      ); }
+/* fd_disco_shred_repair_sig_{...} are accessors for the fields encoded
+   in the sig described above. */
+
+FD_FN_CONST static inline int    fd_disco_shred_repair_sig_skip          ( ulong sig ) { return         fd_ulong_extract_bit( sig, 63     ); }
+FD_FN_CONST static inline ulong  fd_disco_shred_repair_sig_slot          ( ulong sig ) { return         fd_ulong_extract    ( sig, 31, 62 ); }
+FD_FN_CONST static inline uint   fd_disco_shred_repair_sig_fec_set_idx   ( ulong sig ) { return (uint)  fd_ulong_extract    ( sig, 16, 30 ); }
+FD_FN_CONST static inline int    fd_disco_shred_repair_sig_is_code       ( ulong sig ) { return         fd_ulong_extract_bit( sig, 15     ); }
+FD_FN_CONST static inline uint   fd_disco_shred_repair_sig_shred_idx     ( ulong sig ) { return (uint)  fd_ulong_extract_lsb( sig, 15     ); } /* only when is_code = 0 */
+FD_FN_CONST static inline uint   fd_disco_shred_repair_sig_data_cnt      ( ulong sig ) { return (uint)  fd_ulong_extract_lsb( sig, 15     ); } /* only when is_code = 1 */
+
 
 FD_FN_PURE static inline ulong
 fd_disco_compact_chunk0( void * wksp ) {

src/disco/fd_disco_base.h

@@ -92,7 +92,6 @@
 #define STORE_OUT_IDX   0
 #define NET_OUT_IDX     1
 #define SIGN_OUT_IDX    2
-#define REPLAY_OUT_IDX  3
 
 #define MAX_SLOTS_PER_EPOCH 432000UL
 
@@ -179,10 +178,11 @@ typedef struct {
   ulong       store_out_wmark;
   ulong       store_out_chunk;
 
-  fd_wksp_t * replay_out_mem;
-  ulong       replay_out_chunk0;
-  ulong       replay_out_wmark;
-  ulong       replay_out_chunk;
+  fd_wksp_t * repair_out_mem;
+  ulong       repair_out_chunk0;
+  ulong       repair_out_wmark;
+  ulong       repair_out_chunk;
+  ulong       repair_out_idx;
 
   fd_blockstore_t   blockstore_ljoin;
   fd_blockstore_t * blockstore;
@@ -556,6 +556,7 @@ after_frag( fd_shred_ctx_t *    ctx,
   const ulong fanout = 200UL;
   fd_shred_dest_idx_t _dests[ 200*(FD_REEDSOL_DATA_SHREDS_MAX+FD_REEDSOL_PARITY_SHREDS_MAX) ];
 
+  fd_bmtree_node_t out_merkle_root;
   if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_NET ) ) {
     uchar * shred_buffer    = ctx->shred_buffer;
     ulong   shred_buffer_sz = ctx->shred_buffer_sz;
@@ -571,10 +572,9 @@ after_frag( fd_shred_ctx_t *    ctx,
 
     fd_fec_set_t const * out_fec_set[1];
     fd_shred_t const   * out_shred[1];
-    fd_bmtree_node_t     out_merkle_root[1];
 
     long add_shred_timing  = -fd_tickcount();
-    int rv = fd_fec_resolver_add_shred( ctx->resolver, shred, shred_buffer_sz, slot_leader->uc, out_fec_set, out_shred, out_merkle_root );
+    int rv = fd_fec_resolver_add_shred( ctx->resolver, shred, shred_buffer_sz, slot_leader->uc, out_fec_set, out_shred, &out_merkle_root );
     add_shred_timing      +=  fd_tickcount();
 
     fd_histf_sample( ctx->metrics->add_shred_timing, (ulong)add_shred_timing );
@@ -596,14 +596,22 @@ after_frag( fd_shred_ctx_t *    ctx,
         for( ulong j=0UL; j<*max_dest_cnt; j++ ) send_shred( ctx, *out_shred, sdest, dests[ j ], ctx->tsorig );
       } while( 0 );
 
-      if( FD_LIKELY( ctx->blockstore && rv==FD_FEC_RESOLVER_SHRED_OKAY ) ) { /* optimize for the compiler - branch predictor will still be correct */
-        uchar * buf = fd_chunk_to_laddr( ctx->replay_out_mem, ctx->replay_out_chunk );
-        ulong   sz  = fd_shred_header_sz( shred->variant );
-        fd_memcpy( buf, shred, sz );
-        ulong tspub       = fd_frag_meta_ts_comp( fd_tickcount() );
-        ulong replay_sig  = fd_disco_shred_replay_sig( shred->slot, shred->idx, shred->fec_set_idx, fd_shred_is_code( fd_shred_type( shred->variant ) ), 0 );
-        fd_stem_publish( stem, REPLAY_OUT_IDX, replay_sig, ctx->replay_out_chunk, sz, 0UL, ctx->tsorig, tspub );
-        ctx->replay_out_chunk = fd_dcache_compact_next( ctx->replay_out_chunk, sz, ctx->replay_out_chunk0, ctx->replay_out_wmark );
+      if( FD_LIKELY( ctx->repair_out_idx!=ULONG_MAX ) ) { /* firedancer topo compiler hint */
+
+        /* Construct the sig from the shred. */
+
+        int  is_code               = fd_shred_is_code( fd_shred_type( shred->variant ) );
+        uint shred_idx_or_data_cnt = shred->idx;
+        if( FD_LIKELY( is_code ) ) shred_idx_or_data_cnt = shred->code.data_cnt;  /* optimize for code_cnt >= data_cnt */
+        ulong sig = fd_disco_shred_repair_sig( shred->slot, shred->fec_set_idx, is_code, shred_idx_or_data_cnt );
+
+        /* Copy the shred header into the frag and publish. */
+
+        ulong sz = fd_shred_header_sz( shred->variant );
+        fd_memcpy( fd_chunk_to_laddr( ctx->repair_out_mem, ctx->repair_out_chunk ), shred, sz );
+        ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
+        fd_stem_publish( stem, ctx->repair_out_idx, sig, ctx->repair_out_chunk, sz, 0UL, ctx->tsorig, tspub );
+        ctx->repair_out_chunk = fd_dcache_compact_next( ctx->repair_out_chunk, sz, ctx->repair_out_chunk0, ctx->repair_out_wmark );
       }
     }
     if( FD_LIKELY( rv!=FD_FEC_RESOLVER_SHRED_COMPLETES ) ) return;
@@ -643,31 +651,43 @@ after_frag( fd_shred_ctx_t *    ctx,
   ulong sz2 = sizeof(fd_shred34_t) - (34UL - s34[ 2 ].shred_cnt)*FD_SHRED_MAX_SZ;
   ulong sz3 = sizeof(fd_shred34_t) - (34UL - s34[ 3 ].shred_cnt)*FD_SHRED_MAX_SZ;
 
-  if( FD_LIKELY( ctx->blockstore ) ) {
-    /* If the shred has a completes flag, then in the replay tile it
-       will do immediate polling for shreds in that FEC set, under
-       the assumption that they live in the blockstore. When a shred
-       completes a FEC set, we need to add the shreds to the
-       blockstore before we notify replay of a completed FEC set.
-       Replay does not poll the blockstore for shreds on notifies of
-       a regular non-completing shred. */
+  if( FD_LIKELY( ctx->blockstore ) ) { /* firedancer topo compiler hint */
+
+    /* Insert shreds into the blockstore. Note we do this regardless of
+       whether the shreds are for one of our leader slots or not. Even
+       though there is a separate link that directly connects pack and
+       replay when we are leader, we still need the shreds in the
+       blockstore to, for example, serve repair requests. */
 
     for( ulong i=0UL; i<set->data_shred_cnt; i++ ) {
       fd_shred_t const * data_shred = (fd_shred_t const *)fd_type_pun_const( set->data_shreds[ i ] );
       fd_blockstore_shred_insert( ctx->blockstore, data_shred );
     }
-    if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_NET ) ) {
-      /* Shred came from block we didn't produce. This is not our leader
-         slot. */
-      fd_shred_t const * shred = (fd_shred_t const *)fd_type_pun_const( ctx->shred_buffer );
-      uchar * buf = fd_chunk_to_laddr( ctx->replay_out_mem, ctx->replay_out_chunk );
-      ulong   sz  = fd_shred_header_sz( shred->variant );
-      fd_memcpy( buf, shred, sz );
-      ulong tspub       = fd_frag_meta_ts_comp( fd_tickcount() );
-      ulong replay_sig  = fd_disco_shred_replay_sig( shred->slot, shred->idx, shred->fec_set_idx, fd_shred_is_code( fd_shred_type( shred->variant ) ), 1 );
-      fd_stem_publish( stem, REPLAY_OUT_IDX, replay_sig, ctx->replay_out_chunk, sz, 0UL, ctx->tsorig, tspub );
-      ctx->replay_out_chunk = fd_dcache_compact_next( ctx->replay_out_chunk, sz, ctx->replay_out_chunk0, ctx->replay_out_wmark );
-    }
+  }
+
+  if( FD_LIKELY( ctx->repair_out_idx!=ULONG_MAX ) ) { /* firedancer topo compiler hint */
+
+  /* Additionally, publish a frag to notify repair that the FEC set is
+     complete. Note the ordering wrt blockstore shred insertion above is
+     intentional: shreds are inserted into the blockstore before
+     notifying repair. This is because the replay tile is downstream of
+     repair, and replay assumes the shreds are already in the blockstore
+     when repair notifies it that the FEC set is complete, and we don't
+     know whether shred will finish inserting into blockstore first or
+     repair will finish validating the FEC set first. */
+
+    fd_shred_t const * last = (fd_shred_t const *)fd_type_pun_const( set->data_shreds[ set->data_shred_cnt - 1 ] );
+
+    /* Copy the last shred and merkle root of the FEC set into the frag. */
+
+    ulong   sig   = ULONG_MAX;
+    uchar * chunk = fd_chunk_to_laddr( ctx->repair_out_mem, ctx->repair_out_chunk );
+    memcpy( chunk, last, FD_SHRED_CODE_HEADER_SZ );
+    memcpy( chunk, out_merkle_root.hash, FD_SHRED_MERKLE_ROOT_SZ );
+    ulong sz    = FD_SHRED_CODE_HEADER_SZ + FD_SHRED_MERKLE_ROOT_SZ;
+    ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
+    fd_stem_publish( stem, ctx->repair_out_idx, sig, ctx->repair_out_chunk, sz, 0UL, ctx->tsorig, tspub );
+    ctx->repair_out_chunk = fd_dcache_compact_next( ctx->repair_out_chunk, sz, ctx->repair_out_chunk0, ctx->repair_out_wmark );
   }
 
   /* Send to the blockstore, skipping any empty shred34_t s. */
@@ -734,20 +754,11 @@ fd_shred_signer( void *        signer_ctx,
 static void
 unprivileged_init( fd_topo_t *      topo,
                    fd_topo_tile_t * tile ) {
-  void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
 
-  if( FD_LIKELY( tile->out_cnt==3UL ) ) { /* frankendancer */
-    FD_TEST( 0==strcmp( topo->links[tile->out_link_id[STORE_OUT_IDX]].name,  "shred_store"  ) );
-    FD_TEST( 0==strcmp( topo->links[tile->out_link_id[NET_OUT_IDX]].name,    "shred_net"    ) );
-    FD_TEST( 0==strcmp( topo->links[tile->out_link_id[SIGN_OUT_IDX]].name,   "shred_sign"   ) );
-  } else if( FD_LIKELY( tile->out_cnt==4UL ) ) { /* firedancer */
-    FD_TEST( 0==strcmp( topo->links[tile->out_link_id[STORE_OUT_IDX]].name,  "shred_storei"  ) );
-    FD_TEST( 0==strcmp( topo->links[tile->out_link_id[NET_OUT_IDX]].name,    "shred_net"    ) );
-    FD_TEST( 0==strcmp( topo->links[tile->out_link_id[SIGN_OUT_IDX]].name,   "shred_sign"   ) );
-    FD_TEST( 0==strcmp( topo->links[tile->out_link_id[REPLAY_OUT_IDX]].name, "shred_replay" ) );
-  } else {
-    FD_LOG_ERR(( "shred tile has unexpected cnt of output links %lu", tile->out_cnt ));
-  }
+  char * store_out_name = topo->links[tile->out_link_id[ STORE_OUT_IDX ]].name;
+  FD_TEST( 0==strcmp( store_out_name, "shred_store" ) || 0==strcmp( store_out_name, "shred_storei" ) );
+  FD_TEST( 0==strcmp( topo->links[tile->out_link_id[ NET_OUT_IDX   ]].name, "shred_net"   ) );
+  FD_TEST( 0==strcmp( topo->links[tile->out_link_id[ SIGN_OUT_IDX  ]].name, "shred_sign"  ) );
 
   if( FD_UNLIKELY( !tile->out_cnt ) )
     FD_LOG_ERR(( "shred tile has no primary output link" ));
@@ -757,6 +768,8 @@ unprivileged_init( fd_topo_t *      topo,
     FD_LOG_ERR(( "shred tile out depths are not equal %lu %lu",
                  topo->links[ tile->out_link_id[ 0 ] ].depth, shred_store_mcache_depth ));
 
+  void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
+
   FD_SCRATCH_ALLOC_INIT( l, scratch );
   fd_shred_ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof( fd_shred_ctx_t ), sizeof( fd_shred_ctx_t ) );
 
@@ -899,21 +912,21 @@ unprivileged_init( fd_topo_t *      topo,
   ctx->store_out_wmark  = fd_dcache_compact_wmark ( ctx->store_out_mem, store_out->dcache, store_out->mtu );
   ctx->store_out_chunk  = ctx->store_out_chunk0;
 
-  if( FD_LIKELY( tile->out_cnt==4UL ) ) { /* firedancer */
-    fd_topo_link_t * replay_out = &topo->links[ tile->out_link_id[ REPLAY_OUT_IDX ] ];
+  ctx->repair_out_idx = fd_topo_find_tile_out_link( topo, tile, "shred_repair", 0 /* one repair tile so always kind_id 0 */ );
+  if( FD_LIKELY( ctx->repair_out_idx!=ULONG_MAX ) ) { /* firedancer topo compiler hint */
+    fd_topo_link_t * repair_out = &topo->links[ ctx->repair_out_idx ];
 
-    ctx->replay_out_mem    = topo->workspaces[ topo->objs[ replay_out->dcache_obj_id ].wksp_id ].wksp;
-    ctx->replay_out_chunk0 = fd_dcache_compact_chunk0( ctx->replay_out_mem, replay_out->dcache );
-    ctx->replay_out_wmark  = fd_dcache_compact_wmark ( ctx->replay_out_mem, replay_out->dcache, replay_out->mtu );
-    ctx->replay_out_chunk  = ctx->replay_out_chunk0;
+    ctx->repair_out_mem    = topo->workspaces[ topo->objs[ repair_out->dcache_obj_id ].wksp_id ].wksp;
+    ctx->repair_out_chunk0 = fd_dcache_compact_chunk0( ctx->repair_out_mem, repair_out->dcache );
+    ctx->repair_out_wmark  = fd_dcache_compact_wmark ( ctx->repair_out_mem, repair_out->dcache, repair_out->mtu );
+    ctx->repair_out_chunk  = ctx->repair_out_chunk0;
   }
 
+  ctx->blockstore = NULL;
   ulong blockstore_obj_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "blockstore" );
   if (FD_LIKELY( blockstore_obj_id!=ULONG_MAX )) {
     ctx->blockstore = fd_blockstore_join( &ctx->blockstore_ljoin, fd_topo_obj_laddr( topo, blockstore_obj_id ) );
     FD_TEST( ctx->blockstore->shmem->magic == FD_BLOCKSTORE_MAGIC );
-  } else {
-    ctx->blockstore = NULL;
   }
 
   ctx->poh_in_expect_seq = 0UL;