backtest: aggressive rocksdb tower publishing

Author: ibhatt-jumptrading

src/discof/backtest/fd_backtest_tile.c

@@ -102,6 +102,13 @@ typedef struct {
   fd_store_t *           store;
   fd_tower_t *           tower;
   fd_shred_t const *     curr;
+
+  /* is_ready is used to determine if the backtest tile should send
+     more fec sets to the replay tile. is_ready==1 if more fec sets
+     should be sent; it gets set to 0 while waiting for an end of slot
+     notification from the replay tile. When a slot notification is
+     received, is_ready is set to 1 again. */
+  int is_ready;
 } ctx_t;
 
 FD_FN_CONST static inline ulong
@@ -181,9 +188,18 @@ notify_tower_root( ctx_t *             ctx,
                    ulong               tsorig,
                    ulong               tspub ) {
   ulong replayed_slot = ctx->replay_notification.slot_exec.slot;
-  ulong root          = fd_tower_vote( ctx->tower, replayed_slot );
-  if( FD_LIKELY( root != FD_SLOT_NULL ) ) {
-    fd_stem_publish( stem, ctx->tower_replay_out_idx, root, 0UL, 0UL, 0UL, tsorig, tspub );
+  if( ctx->ingest_mode == FD_BACKTEST_ROCKSDB_INGEST ) {
+    /* We want to publish the previous last_replayed_slot, when we have
+       finished replaying the current slot. Then we can update the
+       last_replayed_slot to the newly executed slot. */
+    fd_stem_publish( stem, ctx->tower_replay_out_idx, ctx->replay_notification.slot_exec.slot, 0UL, 0UL, 0UL, tsorig, tspub );
+    ctx->is_ready = 1;
+
+  } else if( ctx->ingest_mode == FD_BACKTEST_SHREDCAP_INGEST ) {
+    ulong root = fd_tower_vote( ctx->tower, replayed_slot );
+    if( FD_LIKELY( root != FD_SLOT_NULL ) ) {
+      fd_stem_publish( stem, ctx->tower_replay_out_idx, root, 0UL, 0UL, 0UL, tsorig, tspub );
+    }
   }
 }
 
@@ -307,6 +323,10 @@ unprivileged_init( fd_topo_t *      topo,
   ctx->slot_cnt    = 0UL;
 
   ctx->curr = NULL;
+
+  ctx->is_ready = 1;
+
+  FD_LOG_NOTICE(( "Finished unprivileged init" ));
 }
 
 fd_hash_t
@@ -340,6 +360,14 @@ after_credit_rocksdb( ctx_t *             ctx,
     ctx->rocksdb_iter = rocksdb_create_iterator_cf(ctx->rocksdb.db, ctx->rocksdb.ro, ctx->rocksdb.cf_handles[FD_ROCKSDB_CFIDX_DATA_SHRED]);
   }
 
+  /* If the slot we just sent to the replay tile has not finished
+     replaying, then we will block until it's done replaying and a
+     notification is received from the replay tile. */
+
+  if( ctx->is_ready==0 ) {
+    return;
+  }
+
   ulong sz;
   fd_shred_t const * prev = NULL;
   fd_shred_t const * curr = ctx->curr ? ctx->curr : rocksdb_get_shred( ctx, &sz );
@@ -380,6 +408,11 @@ after_credit_rocksdb( ctx_t *             ctx,
   ctx->replay_out_chunk = fd_dcache_compact_next( ctx->replay_out_chunk, sizeof(fd_fec_out_t), ctx->replay_out_chunk0, ctx->replay_out_wmark );
   ctx->curr = curr;
   *charge_busy = 1;
+
+  if( out.slot_complete ) {
+    ctx->is_ready = 0;
+  }
+
   return; /* yield otherwise it will overrun */
 }
 

src/discof/replay/fd_replay_tile.c

@@ -1023,11 +1023,8 @@ after_frag( fd_replay_tile_ctx_t *   ctx,
   if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_TOWER ) ) {
     ulong root = sig;
 
-    if( FD_LIKELY( root <= fd_fseq_query( ctx->published_wmark ) ) ) return;
-
-    ulong const slot = fd_bank_slot_get( ctx->slot_ctx->bank );
-    if( FD_UNLIKELY( slot==root ) ) {
-      FD_LOG_CRIT(( "invariant violation: root %lu is the same as the current slot %lu", root, slot ));
+    if( FD_LIKELY( root <= fd_fseq_query( ctx->published_wmark ) ) ) {
+      return;
     }
 
     ctx->root = root;
@@ -1373,9 +1370,6 @@ handle_new_slot( fd_replay_tile_ctx_t * ctx,
   }
 
   fd_funk_txn_t * parent_txn = fd_funk_txn_query( &parent_xid, txn_map );
-  if( FD_UNLIKELY( !parent_txn && parent_slot!=ctx->snapshot_slot ) ) {
-    FD_LOG_CRIT(( "parent_txn is NULL for slot %lu", parent_slot ));
-  }
 
   fd_funk_txn_t * funk_txn = fd_funk_txn_prepare( ctx->funk, parent_txn, &xid, 1 );
   if( FD_UNLIKELY( !funk_txn ) ) {

src/flamenco/runtime/fd_bank.c

@@ -517,7 +517,7 @@ fd_banks_clone_from_parent( fd_banks_t * banks,
   fd_bank_t * parent_bank = fd_banks_map_ele_query( bank_map, &parent_slot, NULL, bank_pool );
 
   if( FD_UNLIKELY( !parent_bank ) ) {
-    FD_LOG_WARNING(( "Failed to get bank" ));
+    FD_LOG_WARNING(( "Failed to get bank for parent slot %lu", parent_slot ));
     fd_rwlock_unwrite( &banks->rwlock );
     return NULL;
   }

src/flamenco/runtime/tests/run_ledger_backtest.sh

@@ -176,8 +176,8 @@ echo "
 [runtime]
     heap_size_gib = 50
     [runtime.limits]
-        max_total_banks = 46
-        max_fork_width = 32
+        max_total_banks = 4
+        max_fork_width = 4
 [development]
     sandbox = false
     no_agave = true