fix(repair): repair tile not consuming root_out

Author: lidatong

src/app/firedancer-dev/commands/backtest.c

@@ -23,7 +23,7 @@
 #include "../../../disco/metrics/fd_metrics.h"
 #include "../../../util/pod/fd_pod_format.h"
 #include "../../../discof/replay/fd_replay_notif.h"
-#include "../../../discof/repair/fd_reasm.h"
+#include "../../../discof/reasm/fd_reasm.h"
 #include "../../../flamenco/runtime/fd_runtime_public.h" /* FD_RUNTIME_PUBLIC_ACCOUNT_UPDATE_MSG_MTU */
 #include "../main.h"
 

src/app/firedancer/topology.c

@@ -1,7 +1,7 @@
 #include "topology.h"
 
 #include "../../choreo/fd_choreo_base.h"
-#include "../../discof/repair/fd_reasm.h"
+#include "../../discof/reasm/fd_reasm.h"
 #include "../../discof/replay/fd_replay_notif.h"
 #include "../../disco/net/fd_net_tile.h"
 #include "../../disco/quic/fd_tpu.h"
@@ -668,6 +668,7 @@ fd_topo_initialize( config_t * config ) {
 
   FOR(net_tile_cnt)    fd_topob_tile_in(  topo, "repair",  0UL,          "metric_in", "net_repair",    i,            FD_TOPOB_UNRELIABLE, FD_TOPOB_POLLED   ); /* No reliable consumers of networking fragments, may be dropped or overrun */
   /**/                 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", "root_out",      0UL,          FD_TOPOB_RELIABLE,   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", "snap_out",      0UL,          FD_TOPOB_RELIABLE,   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   );

src/discof/backtest/fd_backtest_tile.c

@@ -8,7 +8,7 @@
 #include "../../disco/store/fd_store.h"
 #include "../../disco/topo/fd_topo.h"
 #include "../../discof/fd_discof.h"
-#include "../../discof/repair/fd_reasm.h"
+#include "../../discof/reasm/fd_reasm.h"
 #include "../../discof/replay/fd_replay_notif.h"
 #include "../../flamenco/runtime/fd_rocksdb.h"
 #include "../../util/pod/fd_pod_format.h"

src/discof/reasm/Local.mk

@@ -0,0 +1,7 @@
+ifdef FD_HAS_INT128
+$(call add-objs,fd_reasm,fd_discof)
+ifdef FD_HAS_HOSTED
+$(call make-unit-test,test_reasm,test_reasm,fd_discof fd_flamenco fd_ballet fd_util)
+$(call run-unit-test,test_reasm)
+endif
+endif

src/discof/repair/fd_reasm.c renamed to src/discof/reasm/fd_reasm.c

@@ -337,13 +337,13 @@ fd_reasm_fec_t *
 fd_reasm_publish( fd_reasm_t * reasm, fd_hash_t const * merkle_root ) {
 # if FD_REASM_USE_HANDHOLDING
   if( FD_UNLIKELY( !pool_ele( reasm->pool, reasm->root ) ) ) { FD_LOG_WARNING(( "missing root"                                                     )); return NULL; }
-  if( FD_UNLIKELY( !pool_ele( reasm->pool, reasm->root ) ) ) { FD_LOG_WARNING(( "merkle root %s not found", FD_BASE58_ENC_32_ALLOCA( merkle_root ) )); return NULL; }
+  if( FD_UNLIKELY( !fd_reasm_query( reasm, merkle_root ) ) ) { FD_LOG_WARNING(( "merkle root %s not found", FD_BASE58_ENC_32_ALLOCA( merkle_root ) )); return NULL; }
 # endif
 
   fd_reasm_fec_t *  pool = reasm->pool;
   ulong             null = pool_idx_null( pool );
   fd_reasm_fec_t  * oldr = pool_ele( pool, reasm->root );
-  fd_reasm_fec_t  * newr = pool_ele( pool, reasm->root );
+  fd_reasm_fec_t  * newr = fd_reasm_query( reasm, merkle_root );
 
   /* First, remove the previous root, and push it as the first element
      of the BFS queue. */
@@ -373,3 +373,36 @@ fd_reasm_publish( fd_reasm_t * reasm, fd_hash_t const * merkle_root ) {
   reasm->root  = pool_idx( pool, newr ); /* replace with new root */
   return newr;
 }
+
+#include <stdio.h>
+
+static void
+print( fd_reasm_t const * reasm, fd_reasm_fec_t const * fec, int space, const char * prefix ) {
+  fd_reasm_fec_t * pool = reasm->pool;
+
+  if( fec == NULL ) return;
+
+  if( space > 0 ) printf( "\n" );
+  for( int i = 0; i < space; i++ ) printf( " " );
+  printf( "%s%s", prefix, FD_BASE58_ENC_32_ALLOCA( &fec->key ) );
+
+  fd_reasm_fec_t const * curr = pool_ele_const( pool, fec->child );
+  char new_prefix[1024]; /* FIXME size this correctly */
+  while( curr ) {
+    if( pool_ele_const( pool, curr->sibling ) ) {
+      sprintf( new_prefix, "├── " ); /* branch indicating more siblings follow */
+      print( reasm, curr, space + 4, new_prefix );
+    } else {
+      sprintf( new_prefix, "└── " ); /* end branch */
+      print( reasm, curr, space + 4, new_prefix );
+    }
+    curr = pool_ele_const( pool, curr->sibling );
+  }
+}
+
+void
+fd_reasm_print( fd_reasm_t const * reasm ) {
+  FD_LOG_NOTICE( ( "\n\n[reasm]" ) );
+  print( reasm, pool_ele_const( reasm->pool, reasm->root ), 0, "" );
+  printf( "\n\n" );
+}

src/discof/repair/fd_reasm.h renamed to src/discof/reasm/fd_reasm.h

@@ -251,8 +251,8 @@ fd_reasm_leave( fd_reasm_t * reasm );
 void *
 fd_reasm_delete( void * reasm );
 
-/* fd_reasm_{query,query_const} returns a pointer to the current root of
-   of the reasm.  Returns NULL if there is no root. */
+/* fd_reasm_root returns a pointer to the current root of of the reasm,
+   NULL if there is no root. */
 
 fd_reasm_fec_t *
 fd_reasm_root( fd_reasm_t * reasm );
@@ -262,8 +262,8 @@ fd_reasm_root( fd_reasm_t * reasm );
 ulong
 fd_reasm_slot0( fd_reasm_t * reasm );
 
-/* fd_reasm_{query,query_const} queries the FEC set keyed by merkle.
-   Returns a pointer to the fd_reasm_fec_t if found, NULL otherwise. */
+/* fd_reasm_query returns a pointer to the ele keyed by merkle_root if
+   found, NULL otherwise. */
 
 fd_reasm_fec_t *
 fd_reasm_query( fd_reasm_t * reasm, fd_hash_t const * merkle_root );
@@ -311,6 +311,9 @@ fd_reasm_insert( fd_reasm_t *      reasm,
 fd_reasm_fec_t *
 fd_reasm_publish( fd_reasm_t * reasm, fd_hash_t const * merkle_root );
 
+void
+fd_reasm_print( fd_reasm_t const * reasm );
+
 FD_PROTOTYPES_END
 
 #endif /* HEADER_fd_src_discof_repair_fd_reasm_h */

src/discof/repair/fd_reasm_private.h renamed to src/discof/reasm/fd_reasm_private.h

@@ -32,17 +32,18 @@ test_insert( fd_wksp_t * wksp ) {
 
   /* Receive (0, 64) -> (1, 32) -> (1, 0) -> (2, 0) -> (3, 64) -> (2, 32) -> (3, 32) -> (3, 0) */
 
-  fd_hash_t mrnull[1] = {{{ 0 }}};
-  fd_hash_t mr0_64[1] = {{{ 1 }}};
-  fd_hash_t mr1_00[1] = {{{ 2 }}};
-  fd_hash_t mr1_32[1] = {{{ 3 }}};
-  fd_hash_t mr2_00[1] = {{{ 4 }}};
-  fd_hash_t mr2_32[1] = {{{ 5 }}};
-  fd_hash_t mr3_00[1] = {{{ 6 }}};
-  fd_hash_t mr3_32[1] = {{{ 7 }}};
-  fd_hash_t mr3_64[1] = {{{ 8 }}};
-
-  fd_reasm_fec_t * f0_64 = fd_reasm_insert( reasm, mr0_64, mrnull, 0, 64, 1, 32, 1, 1 );
+  fd_hash_t mr0_64[1] = {{{ 0 }}};
+  fd_hash_t mr1_00[1] = {{{ 1 }}};
+  fd_hash_t mr1_32[1] = {{{ 2 }}};
+  fd_hash_t mr2_00[1] = {{{ 3 }}};
+  fd_hash_t mr2_32[1] = {{{ 4 }}};
+  fd_hash_t mr3_00[1] = {{{ 5 }}};
+  fd_hash_t mr3_32[1] = {{{ 6 }}};
+  fd_hash_t mr3_64[1] = {{{ 7 }}};
+
+  fd_reasm_init( reasm, mr0_64, 0 );
+  fd_reasm_fec_t * f0_64 = fd_reasm_query( reasm, mr0_64 );
+  FD_TEST( f0_64 );
   FD_TEST( frontier_ele_query( frontier, &f0_64->key, NULL, pool ) == f0_64 );
 
   fd_reasm_fec_t * f1_32 = fd_reasm_insert( reasm, mr1_32, mr1_00, 1, 32, 1, 32, 1, 1 );
@@ -135,7 +136,7 @@ test_publish( fd_wksp_t * wksp ){
 
   /* Set the root (snapshot slot). */
 
-  fd_reasm_insert( reasm, mr0, mr0, 0, 0, 0, 0, 1, 1 );
+  fd_reasm_init( reasm, mr0, 0 );
 
   /* Typical startup behavior, turbine orphan FECs added. */
 
@@ -164,9 +165,12 @@ test_publish( fd_wksp_t * wksp ){
 
   /* Publish. */
 
+  fd_reasm_fec_t * oldr = fd_reasm_root( reasm );
+  FD_TEST( oldr );
   fd_reasm_publish( reasm, mr2 );
-  fd_reasm_fec_t * root = pool_ele( reasm->pool, reasm->root );
-  FD_TEST( 0==memcmp( root, mr2, sizeof(fd_hash_t) ) );
+  fd_reasm_fec_t * newr = fd_reasm_root( reasm );
+  FD_TEST( newr );
+  FD_TEST( 0==memcmp( newr, mr2, sizeof(fd_hash_t) ) );
   FD_TEST( ancestry_ele_query( reasm->ancestry, mr2, NULL, reasm->pool ) != NULL );
   FD_TEST( frontier_ele_query( reasm->frontier, mr4, NULL, reasm->pool ) != NULL );
   FD_TEST( !fd_reasm_query( reasm, mr0 ) );

src/discof/repair/test_reasm.c renamed to src/discof/reasm/test_reasm.c

@@ -1,5 +1,3 @@
 ifdef FD_HAS_INT128
-$(call add-objs,fd_reasm,fd_discof)
 $(call add-objs,fd_repair_tile,fd_discof)
-$(call make-unit-test,test_reasm,test_reasm,fd_discof fd_flamenco fd_ballet fd_util)
 endif

src/discof/repair/Local.mk

@@ -17,7 +17,7 @@
 #include "../../util/net/fd_net_headers.h"
 
 #include "../forest/fd_forest.h"
-#include "fd_reasm.h"
+#include "../reasm/fd_reasm.h"
 
 #define IN_KIND_CONTACT (0)
 #define IN_KIND_NET     (1)
@@ -458,6 +458,10 @@ during_frag( fd_repair_tile_ctx_t * ctx,
     dcache_entry = fd_chunk_to_laddr_const( in_ctx->mem, chunk );
     dcache_entry_sz = sz * sizeof(fd_shred_dest_wire_t);
 
+  } else if( FD_UNLIKELY( in_kind==IN_KIND_ROOT ) ) {
+    memcpy( ctx->root_block_id.uc, fd_chunk_to_laddr_const( in_ctx->mem, chunk ), sizeof(fd_hash_t) );
+    return;
+
   } else if( FD_UNLIKELY( in_kind==IN_KIND_STAKE ) ) {
     if( FD_UNLIKELY( chunk<in_ctx->chunk0 || chunk>in_ctx->wmark ) ) {
       FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, in_ctx->chunk0, in_ctx->wmark ));
@@ -513,7 +517,7 @@ static void
 after_frag( fd_repair_tile_ctx_t * ctx,
             ulong                  in_idx,
             ulong                  seq    FD_PARAM_UNUSED,
-            ulong                  sig    FD_PARAM_UNUSED,
+            ulong                  sig,
             ulong                  sz,
             ulong                  tsorig FD_PARAM_UNUSED,
             ulong                  tspub  FD_PARAM_UNUSED,
@@ -672,9 +676,12 @@ after_credit( fd_repair_tile_ctx_t * ctx,
 
   fd_reasm_fec_t * rfec = fd_reasm_next( ctx->reasm );
   if( FD_LIKELY( rfec ) ) {
-    if( FD_LIKELY( ctx->store ) ) {
+
+    if( FD_LIKELY( ctx->store ) ) { /* some topologies don't run with store */
+
       /* Linking only requires a shared lock because the fields that are
           modified are only read on publish which uses exclusive lock. */
+
       long shacq_start, shacq_end, shrel_end;
 
       FD_STORE_SHACQ_TIMED( ctx->store, shacq_start, shacq_end );