feat(shred,repair): shred publish to repair

Author: lidatong

src/app/fdctl/topos/fd_firedancer.c

@@ -172,7 +172,7 @@ fd_topo_initialize( config_t * config ) {
   fd_topob_wksp( topo, "dedup_pack"   );
 
   fd_topob_wksp( topo, "shred_storei" );
-  fd_topob_wksp( topo, "shred_replay" );
+  fd_topob_wksp( topo, "shred_repair" );
   fd_topob_wksp( topo, "stake_out"    );
 
   fd_topob_wksp( topo, "poh_shred"    );
@@ -246,6 +246,8 @@ fd_topo_initialize( config_t * config ) {
 
   #define FOR(cnt) for( ulong i=0UL; i<cnt; i++ )
 
+  ulong pending_fec_shreds_depth = fd_ulong_min( fd_ulong_pow2_up( config->tiles.shred.max_pending_shred_sets * FD_REEDSOL_DATA_SHREDS_MAX ), USHORT_MAX + 1 /* dcache max */ );
+
   /*                                  topo, link_name,      wksp_name,      depth,                                    mtu,                           burst */
   FOR(quic_tile_cnt)   fd_topob_link( topo, "quic_net",     "net_quic",     config->tiles.net.send_buffer_size,       FD_NET_MTU,                    1UL );
   FOR(shred_tile_cnt)  fd_topob_link( topo, "shred_net",    "net_shred",    config->tiles.net.send_buffer_size,       FD_NET_MTU,                    1UL );
@@ -259,7 +261,6 @@ fd_topo_initialize( config_t * config ) {
 
   FOR(shred_tile_cnt)  fd_topob_link( topo, "shred_sign",   "shred_sign",   128UL,                                    32UL,                          1UL );
   FOR(shred_tile_cnt)  fd_topob_link( topo, "sign_shred",   "sign_shred",   128UL,                                    64UL,                          1UL );
-  FOR(shred_tile_cnt)  fd_topob_link( topo, "shred_replay", "shred_replay", 128UL,                                    128UL,                         config->tiles.shred.max_pending_shred_sets );
 
   /**/                 fd_topob_link( topo, "gossip_sign",  "gossip_sign",  128UL,                                    2048UL,                        1UL );
   /**/                 fd_topob_link( topo, "sign_gossip",  "sign_gossip",  128UL,                                    64UL,                          1UL );
@@ -280,6 +281,7 @@ fd_topo_initialize( config_t * config ) {
   /**/                 fd_topob_link( topo, "repair_store", "repair_store", 1024UL*1024UL,                            FD_SHRED_MAX_SZ,               128UL );
   /**/                 fd_topob_link( topo, "repair_net",   "net_repair",   config->tiles.net.send_buffer_size,       FD_NET_MTU,                    1UL   );
   /**/                 fd_topob_link( topo, "repair_sign",  "repair_sign",  128UL,                                    2048UL,                        1UL );
+  FOR(shred_tile_cnt)  fd_topob_link( topo, "shred_repair", "shred_repair", pending_fec_shreds_depth,                 FD_SHRED_REPAIR_MTU,           config->tiles.shred.max_pending_shred_sets );
   /**/                 fd_topob_link( topo, "sign_repair",  "sign_repair",  128UL,                                    64UL,                          1UL );
   /**/                 fd_topob_link( topo, "store_replay", "store_replay", 128UL,                                    (4096UL*sizeof(fd_txn_p_t))+sizeof(ulong)+sizeof(fd_hash_t), 16UL  );
   FOR(bank_tile_cnt)   fd_topob_link( topo, "replay_poh",   "replay_poh",   128UL,                                    (4096UL*sizeof(fd_txn_p_t))+sizeof(fd_microblock_trailer_t), 1UL  );
@@ -503,7 +505,7 @@ fd_topo_initialize( config_t * config ) {
     /**/               fd_topob_tile_in(  topo, "shred",  i,             "metric_in", "sign_shred",    i,            FD_TOPOB_UNRELIABLE, FD_TOPOB_UNPOLLED );
     /**/               fd_topob_tile_out( topo, "sign",   0UL,                        "sign_shred",    i                                                    );
   }
-  FOR(shred_tile_cnt)  fd_topob_tile_out( topo, "shred",  i,                          "shred_replay",  i                                                    );
+  FOR(shred_tile_cnt)  fd_topob_tile_out( topo, "shred",  i,                          "shred_repair",  i                                                    );
 
   FOR(net_tile_cnt)    fd_topob_tile_in(  topo, "gossip",   0UL,          "metric_in", "net_gossip",   i,            FD_TOPOB_UNRELIABLE, FD_TOPOB_POLLED   ); /* No reliable consumers of networking fragments, may be dropped or overrun */
   /**/                 fd_topob_tile_out( topo, "gossip",   0UL,                       "gossip_net",   0UL                                                  );
@@ -523,15 +525,17 @@ fd_topo_initialize( config_t * config ) {
   /**/                 fd_topob_tile_in(  topo, "repair",  0UL,          "metric_in", "gossip_repai",  0UL,          FD_TOPOB_UNRELIABLE, FD_TOPOB_POLLED   );
   /**/                 fd_topob_tile_in(  topo, "repair",  0UL,          "metric_in", "stake_out",     0UL,          FD_TOPOB_UNRELIABLE, FD_TOPOB_POLLED   );
   /**/                 fd_topob_tile_in(  topo, "repair",  0UL,          "metric_in", "store_repair",  0UL,          FD_TOPOB_UNRELIABLE, FD_TOPOB_POLLED   );
+  FOR(shred_tile_cnt)  fd_topob_tile_in(  topo, "repair",  0UL,          "metric_in", "shred_repair",  i,            FD_TOPOB_RELIABLE,   FD_TOPOB_POLLED   );
   /**/                 fd_topob_tile_in(  topo, "replay",  0UL,          "metric_in", "store_replay",  0UL,          FD_TOPOB_RELIABLE,   FD_TOPOB_POLLED   );
   /**/                 fd_topob_tile_out( topo, "replay",  0UL,                       "stake_out",     0UL                                                  );
   /**/                 fd_topob_tile_out( topo, "replay",  0UL,                       "replay_notif",  0UL                                                  );
   /**/                 fd_topob_tile_in(  topo, "replay",  0UL,          "metric_in", "pack_replay",   0UL,          FD_TOPOB_RELIABLE,   FD_TOPOB_POLLED   );
   /**/                 fd_topob_tile_in(  topo, "replay",  0UL,          "metric_in", "batch_replay",  0UL,          FD_TOPOB_RELIABLE,   FD_TOPOB_POLLED   );
   /**/                 fd_topob_tile_out( topo, "replay",  0UL,                       "replay_voter",  0UL                                                  );
   FOR(bank_tile_cnt)   fd_topob_tile_out( topo, "replay",  0UL,                       "replay_poh",    i                                                    );
-  FOR(exec_tile_cnt)   fd_topob_tile_out( topo, "replay",  0UL,                       "replay_exec",   i                                                  ); /* TODO check order in fd_replay.c macros*/
-  FOR(shred_tile_cnt)  fd_topob_tile_in(  topo, "replay",  0UL,          "metric_in", "shred_replay",  i,            FD_TOPOB_RELIABLE,     FD_TOPOB_POLLED );
+
+
+  FOR(exec_tile_cnt)   fd_topob_tile_out( topo, "replay",  0UL,                       "replay_exec",   i                                                    ); /* TODO check order in fd_replay.c macros*/
 
   FOR(exec_tile_cnt)   fd_topob_tile_in(  topo, "exec",    i,             "metric_in", "replay_exec",  i,            FD_TOPOB_RELIABLE, FD_TOPOB_POLLED     );
 

src/ballet/bmtree/fd_bmtree.h

@@ -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/disco/fd_disco_base.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 the size of a coding shred header + size of a
+   merkle root. */
+
+#define FD_SHRED_REPAIR_MTU (FD_SHRED_CODE_HEADER_SZ + FD_SHRED_MERKLE_ROOT_SZ)
+FD_STATIC_ASSERT( FD_SHRED_REPAIR_MTU == 121 , 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,60 @@ 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.
+
+   type (1) | is_code or data_completes (1) | slot (32) | fec_set_idx (15) | shred_idx or data_cnt or parent_off (15)
+   [63]     | [62]                          | [30, 61]  | [15, 29]         | [0, 14]
+
+   The first bit of the sig is the sig type.  The next 32 bits describe
+   the slot number and 15 bits after that the fec_set_idx, regardless of
+   the sig type.  Note if the bits are saturated caller MUST ignore the
+   value extracted from the sig (ie. UINT_MAX for slot and 2^15 - 1 for
+   fec_set_idx).
+
+   The second bit and last 15 bits vary in interpretation depending on
+   the sig type:
+
+   When type is 0, the sig describes a shred header.  In this case, the
+   second bit describes whether it is a coding shred (is_code) and the
+   last 15 bits either describe a shred_idx if it's a data shred
+   (is_code = 0) or the data_cnt if it's a coding shred (is_code = 1).
+
+   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).  This implies the FEC set is the last one in the entry
+   batch.  The last 15 bits describe the parent slot's offset
+   (parent_off) from the FEC set's slot. */
+
 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( int type, int is_code_or_data_completes, ulong slot, uint fec_set_idx, uint shred_idx_or_data_cnt_or_parent_off ) {
+  ulong type_ul                                = (ulong)type;
+  ulong is_code_or_data_completes_ul           = (ulong)is_code_or_data_completes;
+  ulong slot_ul                                = fd_ulong_min( (ulong)slot, (ulong)UINT_MAX );
+  ulong fec_set_idx_ul                         = fd_ulong_min( (ulong)fec_set_idx, (ulong)FD_SHRED_MAX_PER_SLOT );
+  ulong shred_idx_or_data_cnt_or_parent_off_ul = fd_ulong_min( (ulong)shred_idx_or_data_cnt_or_parent_off, (ulong)FD_SHRED_MAX_PER_SLOT );
+  return type_ul << 63 | is_code_or_data_completes_ul << 62 | slot_ul << 30 | fec_set_idx_ul << 15 | shred_idx_or_data_cnt_or_parent_off_ul;
 }
 
-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      ); }
+#define FD_DISCO_SHRED_REPAIR_SIG_TYPE_HDR (0)
+#define FD_DISCO_SHRED_REPAIR_SIG_TYPE_FEC (1)
+
+/* 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_type          ( ulong sig ) { return       fd_ulong_extract_bit( sig, 63     ); }
+FD_FN_CONST static inline int   fd_disco_shred_repair_sig_is_code       ( ulong sig ) { return       fd_ulong_extract_bit( sig, 62     ); } /* type 0 */
+FD_FN_CONST static inline int   fd_disco_shred_repair_sig_data_completes( ulong sig ) { return       fd_ulong_extract_bit( sig, 62     ); } /* type 1 */
+FD_FN_CONST static inline ulong fd_disco_shred_repair_sig_slot          ( ulong sig ) { return       fd_ulong_extract    ( sig, 30, 61 ); }
+FD_FN_CONST static inline uint  fd_disco_shred_repair_sig_fec_set_idx   ( ulong sig ) { return (uint)fd_ulong_extract    ( sig, 15, 29 ); }
+FD_FN_CONST static inline uint  fd_disco_shred_repair_sig_shred_idx     ( ulong sig ) { return (uint)fd_ulong_extract_lsb( sig, 15     ); } /* type 0, 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     ); } /* type 0, is_code 1 */
+FD_FN_CONST static inline uint  fd_disco_shred_repair_sig_parent_off    ( ulong sig ) { return (uint)fd_ulong_extract_lsb( sig, 15     ); } /* type 1 */
+
 
 FD_FN_PURE static inline ulong
 fd_disco_compact_chunk0( void * wksp ) {