shred: notify replay tile of completed shreds

Author: emwang-jump

src/app/fdctl/run/tiles/fd_replay.c

@@ -59,6 +59,7 @@
 #define PACK_IN_IDX    (1UL)
 #define GOSSIP_IN_IDX  (2UL)
 #define BATCH_IN_IDX   (3UL)
+#define SHRED_IN_IDX   (4UL)
 
 #define STAKE_OUT_IDX  (0UL)
 #define NOTIF_OUT_IDX  (1UL)
@@ -120,6 +121,11 @@ struct fd_replay_tile_ctx {
   ulong       batch_in_chunk0;
   ulong       batch_in_wmark;
 
+  // Shred tile input
+  fd_wksp_t * shred_in_mem;
+  ulong       shred_in_chunk0;
+  ulong       shred_in_wmark;
+
   // Notification output defs
   fd_frag_meta_t * notif_out_mcache;
   ulong *          notif_out_sync;
@@ -443,6 +449,40 @@ publish_stake_weights( fd_replay_tile_ctx_t * ctx,
   }
 }
 
+static int
+before_frag( fd_replay_tile_ctx_t * ctx,
+             ulong                  in_idx,
+             ulong                  seq,
+             ulong                  sig ) {
+  (void)ctx;
+  (void)seq;
+
+  if( in_idx == SHRED_IN_IDX ){
+    /* NOTE: If we filter on every fec set thats sent to us from the shred
+    tile, there is going to be contention if we query the block map every
+    shred in order to filter them. */
+
+    ulong slot = fd_disco_shred_sig_slot( sig );
+    uchar block_flags = 0;
+    int err = FD_MAP_ERR_AGAIN;
+    while( err == FD_MAP_ERR_AGAIN ){
+      fd_block_map_query_t quer[1] = { 0 };
+      err = fd_block_map_query_try( ctx->blockstore->block_map, &slot, NULL, quer, 0 );
+      fd_block_meta_t * block_map_entry = fd_block_map_query_ele( quer );
+      if( FD_UNLIKELY( err == FD_MAP_ERR_AGAIN ) ) continue;
+      if( FD_UNLIKELY( err == FD_MAP_ERR_KEY )) break;
+      block_flags = block_map_entry->flags;
+      err = fd_block_map_query_test( quer );
+    }
+
+    if( FD_UNLIKELY( fd_uchar_extract_bit( block_flags, FD_BLOCK_FLAG_PROCESSED ) ) ) {
+      FD_LOG_INFO(( "before_frag filtering: block already processed - slot: %lu", slot ));
+      return -1; /* skip this frag */
+    }
+  }
+  return 0;
+}
+
 static void
 during_frag( fd_replay_tile_ctx_t * ctx,
              ulong                  in_idx,
@@ -481,6 +521,17 @@ during_frag( fd_replay_tile_ctx_t * ctx,
     fd_memcpy( dst_poh, src, sz * sizeof(fd_txn_p_t) );
 
     FD_LOG_INFO(( "other microblock - slot: %lu, parent_slot: %lu, txn_cnt: %lu", ctx->curr_slot, ctx->parent_slot, sz ));
+  } else if ( in_idx == SHRED_IN_IDX) {
+    if( FD_UNLIKELY( sz != 0 ) ) {
+      FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, ctx->shred_in_chunk0, ctx->shred_in_wmark ));
+    }
+    ulong slot = fd_disco_shred_sig_slot( sig );
+    ulong parent_off = fd_disco_shred_sig_parent_off( sig );
+    ulong flags = fd_disco_shred_sig_flags( sig );
+
+    FD_LOG_INFO(( "Notified of complete FEC set from the shred tile: slot: %lu, flags: %lu, parent_slot: %lu", slot,  flags, slot - parent_off ));
+    ctx->skip_frag = 1;
+    return;
   } else if( in_idx == PACK_IN_IDX ) {
     if( FD_UNLIKELY( chunk<ctx->pack_in_chunk0 || chunk>ctx->pack_in_wmark || sz>USHORT_MAX ) ) {
       FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, ctx->pack_in_chunk0, ctx->pack_in_wmark ));
@@ -557,7 +608,7 @@ during_frag( fd_replay_tile_ctx_t * ctx,
   }
 
   if( FD_UNLIKELY( fd_uchar_extract_bit( block_flags, FD_BLOCK_FLAG_PROCESSED ) ) ) {
-    FD_LOG_WARNING(( "block already processed - slot: %lu", ctx->curr_slot ));
+    FD_LOG_WARNING(( "block already processed - slot: %lu, from %lu", ctx->curr_slot, in_idx ));
     ctx->skip_frag = 1;
   }
   if( FD_UNLIKELY( fd_uchar_extract_bit( block_flags, FD_BLOCK_FLAG_DEADBLOCK ) ) ) {
@@ -3001,6 +3052,11 @@ unprivileged_init( fd_topo_t *      topo,
   ctx->batch_in_chunk0           = fd_dcache_compact_chunk0( ctx->batch_in_mem, batch_in_link->dcache );
   ctx->batch_in_wmark            = fd_dcache_compact_wmark( ctx->batch_in_mem, batch_in_link->dcache, batch_in_link->mtu );
 
+  fd_topo_link_t * shred_in_link = &topo->links[ tile->in_link_id[ SHRED_IN_IDX ] ];
+  ctx->shred_in_mem              = topo->workspaces[ topo->objs[ shred_in_link->dcache_obj_id ].wksp_id ].wksp;
+  ctx->shred_in_chunk0           = fd_dcache_compact_chunk0( ctx->shred_in_mem, shred_in_link->dcache );
+  ctx->shred_in_wmark            = fd_dcache_compact_wmark( ctx->shred_in_mem, shred_in_link->dcache, shred_in_link->mtu );
+
   fd_topo_link_t * notif_out = &topo->links[ tile->out_link_id[ NOTIF_OUT_IDX ] ];
   ctx->notif_out_mcache      = notif_out->mcache;
   ctx->notif_out_sync        = fd_mcache_seq_laddr( ctx->notif_out_mcache );
@@ -3130,6 +3186,7 @@ metrics_write( fd_replay_tile_ctx_t * ctx ) {
 
 #define STEM_CALLBACK_DURING_HOUSEKEEPING during_housekeeping
 #define STEM_CALLBACK_AFTER_CREDIT        after_credit
+#define STEM_CALLBACK_BEFORE_FRAG         before_frag
 #define STEM_CALLBACK_DURING_FRAG         during_frag
 #define STEM_CALLBACK_AFTER_FRAG          after_frag
 #define STEM_CALLBACK_METRICS_WRITE       metrics_write

src/app/fdctl/run/tiles/fd_shred.c

@@ -91,6 +91,7 @@
 #define STORE_OUT_IDX   0
 #define NET_OUT_IDX     1
 #define SIGN_OUT_IDX    2
+#define MULTI_OUT_IDX   3
 
 #define MAX_SLOTS_PER_EPOCH 432000UL
 
@@ -180,6 +181,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;
+
   struct {
     fd_histf_t contact_info_cnt[ 1 ];
     fd_histf_t batch_sz[ 1 ];
@@ -523,6 +529,7 @@ send_shred( fd_shred_ctx_t *      ctx,
   ctx->net_out_chunk = fd_dcache_compact_next( ctx->net_out_chunk, pkt_sz, ctx->net_out_chunk0, ctx->net_out_wmark );
 }
 
+
 static void
 after_frag( fd_shred_ctx_t *    ctx,
             ulong               in_idx,
@@ -554,11 +561,12 @@ 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) ];
 
-  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;
+  uchar * shred_buffer    = ctx->shred_buffer;
+  ulong   shred_buffer_sz = ctx->shred_buffer_sz;
 
-    fd_shred_t const * shred = fd_shred_parse( shred_buffer, shred_buffer_sz );
+  fd_shred_t const * shred = fd_shred_parse( shred_buffer, shred_buffer_sz );
+
+  if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_NET ) ) {
     if( FD_UNLIKELY( !shred       ) ) { ctx->metrics->shred_processing_result[ 1 ]++; return; }
 
     fd_epoch_leaders_t const * lsched = fd_stake_ci_get_lsched_for_slot( ctx->stake_ci, shred->slot );
@@ -633,14 +641,19 @@ after_frag( fd_shred_ctx_t *    ctx,
   /* Send to the blockstore, skipping any empty shred34_t s. */
   ulong new_sig = ctx->in_kind[ in_idx ]!=IN_KIND_NET; /* sig==0 means the store tile will do extra checks */
   ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
-  fd_stem_publish( stem, 0UL, new_sig, fd_laddr_to_chunk( ctx->store_out_mem, s34+0UL ), sz0, 0UL, ctx->tsorig, tspub );
+  fd_stem_publish( stem, STORE_OUT_IDX, new_sig, fd_laddr_to_chunk( ctx->store_out_mem, s34+0UL ), sz0, 0UL, ctx->tsorig, tspub );
   if( FD_UNLIKELY( s34[ 1 ].shred_cnt ) )
-    fd_stem_publish( stem, 0UL, new_sig, fd_laddr_to_chunk( ctx->store_out_mem, s34+1UL ), sz1, 0UL, ctx->tsorig, tspub );
-  fd_stem_publish( stem, 0UL, new_sig, fd_laddr_to_chunk( ctx->store_out_mem, s34+2UL), sz2, 0UL, ctx->tsorig, tspub );
+    fd_stem_publish( stem, STORE_OUT_IDX, new_sig, fd_laddr_to_chunk( ctx->store_out_mem, s34+1UL ), sz1, 0UL, ctx->tsorig, tspub );
+  fd_stem_publish( stem, STORE_OUT_IDX, new_sig, fd_laddr_to_chunk( ctx->store_out_mem, s34+2UL), sz2, 0UL, ctx->tsorig, tspub );
   if( FD_UNLIKELY( s34[ 3 ].shred_cnt ) )
-    fd_stem_publish( stem, 0UL, new_sig, fd_laddr_to_chunk( ctx->store_out_mem, s34+3UL ), sz3, 0UL, ctx->tsorig, tspub );
-
-
+    fd_stem_publish( stem, STORE_OUT_IDX, new_sig, fd_laddr_to_chunk( ctx->store_out_mem, s34+3UL ), sz3, 0UL, ctx->tsorig, tspub );
+
+  /* Send to replay tile */
+#if FD_HAS_NO_AGAVE
+  fd_shred_t const * last_data_shred = (fd_shred_t const *)set->data_shreds[ set->data_shred_cnt-1 ];
+  ulong replay_sig = fd_disco_shred_sig( shred->slot, last_data_shred->data.parent_off, last_data_shred->data.flags, 1 );
+  fd_stem_publish( stem, MULTI_OUT_IDX, replay_sig, ctx->replay_out_chunk, 0, 0UL, ctx->tsorig, tspub );
+#endif
   /* Compute all the destinations for all the new shreds */
 
   fd_shred_t const * new_shreds[ FD_REEDSOL_DATA_SHREDS_MAX+FD_REEDSOL_PARITY_SHREDS_MAX ];
@@ -697,11 +710,18 @@ unprivileged_init( fd_topo_t *      topo,
                    fd_topo_tile_t * tile ) {
   void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
 
-  if( FD_UNLIKELY( tile->out_cnt!=3UL ||
+  #if FD_HAS_NO_AGAVE
+    int check_multi_out_link = strcmp( topo->links[ tile->out_link_id[ MULTI_OUT_IDX ] ].name,  "shred_multi"  );
+  #else
+    int check_multi_out_link = 0;
+  #endif
+
+  if( FD_UNLIKELY( tile->out_cnt<3UL ||
                    (strcmp( topo->links[ tile->out_link_id[ STORE_OUT_IDX ] ].name, "shred_store" ) &&
                       strcmp( topo->links[ tile->out_link_id[ STORE_OUT_IDX ] ].name, "shred_storei" ) )  ||
                    strcmp( topo->links[ tile->out_link_id[ NET_OUT_IDX ] ].name,   "shred_net"   )  ||
-                   strcmp( topo->links[ tile->out_link_id[ SIGN_OUT_IDX ] ].name,  "shred_sign"  ) ) )
+                   strcmp( topo->links[ tile->out_link_id[ SIGN_OUT_IDX ] ].name,  "shred_sign"  )  ||
+                    check_multi_out_link ) )
     FD_LOG_ERR(( "shred tile has none or unexpected output links %lu %s %s",
                  tile->out_cnt, topo->links[ tile->out_link_id[ 0 ] ].name, topo->links[ tile->out_link_id[ 1 ] ].name ));
 
@@ -859,6 +879,19 @@ 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_HAS_NO_AGAVE
+    fd_topo_link_t * replay_out = &topo->links[ tile->out_link_id[ MULTI_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;
+#else
+    ctx->replay_out_mem    = NULL;
+    ctx->replay_out_chunk0 = 0;
+    ctx->replay_out_wmark  = 0;
+    ctx->replay_out_chunk  = 0;
+#endif
+
   ctx->poh_in_expect_seq = 0UL;
 
   ctx->shredder_fec_set_idx = 0UL;

src/app/fdctl/run/tiles/fd_store_int.c

@@ -451,7 +451,7 @@ fd_store_tile_slot_prepare( fd_store_tile_ctx_t * ctx,
     ulong repair_req_sz = repair_req_cnt * sizeof(fd_repair_request_t);
     FD_TEST( repair_req_sz<=USHORT_MAX );
     fd_mcache_publish( ctx->repair_req_out_mcache, ctx->repair_req_out_depth, ctx->repair_req_out_seq, repair_req_sig, ctx->repair_req_out_chunk,
-      repair_req_sz, 0UL, tsorig, tspub );
+                       repair_req_sz, 0UL, tsorig, tspub );
     ctx->repair_req_out_seq   = fd_seq_inc( ctx->repair_req_out_seq, 1UL );
     ctx->repair_req_out_chunk = fd_dcache_compact_next( ctx->repair_req_out_chunk, repair_req_sz, ctx->repair_req_out_chunk0, ctx->repair_req_out_wmark );
   }

src/app/fdctl/run/topos/fd_firedancer.c

@@ -97,6 +97,7 @@ fd_topo_initialize( config_t * config ) {
 
   fd_topob_wksp( topo, "shred_sign"   );
   fd_topob_wksp( topo, "sign_shred"   );
+  fd_topob_wksp( topo, "shred_multi"  );
 
   fd_topob_wksp( topo, "gossip_sign"  );
   fd_topob_wksp( topo, "sign_gossip"  );
@@ -175,6 +176,7 @@ 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 );
 
+  /**/                 fd_topob_link( topo, "shred_multi",  "shred_multi",  128UL,                                    sizeof(ulong),                 1UL );
   /**/                 fd_topob_link( topo, "gossip_sign",  "gossip_sign",  128UL,                                    2048UL,                        1UL );
   /**/                 fd_topob_link( topo, "sign_gossip",  "sign_gossip",  128UL,                                    64UL,                          1UL );
   /**/                 fd_topob_link( topo, "gossip_repla", "gossip_repla", 128UL,                                    4UL + 128UL + 8192UL,          1UL );
@@ -408,6 +410,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_multi",    i                                                   );
 
   FOR(net_tile_cnt)    fd_topob_tile_out( topo, "net",      i,                         "net_gossip",   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 */
@@ -439,6 +442,7 @@ fd_topo_initialize( config_t * config ) {
   /**/                 fd_topob_tile_out( topo, "replay",  0UL,                       "replay_voter",  0UL                                                  );
   /**/                 fd_topob_tile_out( topo, "replay",  0UL,                       "replay_gossi",  0UL                                                  );
   /**/                 fd_topob_tile_out( topo, "replay",  0UL,                       "replay_store",  0UL                                                  );
+  /**/                 fd_topob_tile_in(  topo, "replay",  0UL,          "metric_in", "shred_multi",   0UL,          FD_TOPOB_RELIABLE,   FD_TOPOB_POLLED   );
   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(exec_tile_cnt)   fd_topob_tile_in(  topo, "exec",    i,             "metric_in", "replay_exec",  i,            FD_TOPOB_RELIABLE, FD_TOPOB_POLLED     );

src/disco/fd_disco_base.h

@@ -121,6 +121,23 @@ fd_disco_replay_sig( ulong slot,
 FD_FN_CONST static inline ulong fd_disco_replay_sig_flags( ulong sig ) { return (sig & 0xFFUL); }
 FD_FN_CONST static inline ulong fd_disco_replay_sig_slot( ulong sig ) { return (sig >> 8); }
 
+FD_FN_CONST static inline ulong
+fd_disco_shred_sig( ulong  slot,
+                    ushort parent_off,
+                    uchar  shred_flags,
+                    int    is_turbine ) {
+  /* lower 32 bits of the slot
+     full 16 bits of parent offset
+     full 8 bits of data shred flags
+     8 bits of extra flags */
+  return (slot << 32) | ((ulong)parent_off << 16) | ((ulong) shred_flags << 8) | ( is_turbine ? 1UL : 0UL );
+}
+
+FD_FN_CONST static inline ulong fd_disco_shred_sig_slot( ulong sig ) { return (sig >> 32); }
+FD_FN_CONST static inline ushort fd_disco_shred_sig_parent_off( ulong sig ) { return (ushort)(sig >> 16); }
+FD_FN_CONST static inline uchar  fd_disco_shred_sig_flags( ulong sig ) { return (uchar)(sig >> 8); }
+FD_FN_CONST static inline int    fd_disco_shred_sig_is_turbine( ulong sig ) { return (int)(sig & 1UL); }
+
 FD_FN_PURE static inline ulong
 fd_disco_compact_chunk0( void * wksp ) {
   return (((struct fd_wksp_private *)wksp)->gaddr_lo) >> FD_CHUNK_LG_SZ;