combining block info wmark and shred idx wmark

Author: emwang-jump

src/discof/replay/fd_replay_tile.c

@@ -1,4 +1,3 @@
-#include <climits>
 #define _GNU_SOURCE
 #include "../../disco/tiles.h"
 #include "generated/fd_replay_tile_seccomp.h"
@@ -439,6 +438,15 @@ slice_poll( fd_replay_tile_ctx_t * ctx,
             fd_replay_slice_t    * slice_deque,
             ulong slot ) {
   uint consumed_idx, slices_added;
+
+  fd_replay_idxs_t * shred_idxs = fd_replay_idxs_map_query( ctx->replay->idxs_map, slot, NULL );
+  uint turbine_shred_wmark;
+  if( !shred_idxs ) {
+    turbine_shred_wmark = UINT_MAX;
+  } else {
+    turbine_shred_wmark = (uint)shred_idxs->wmark;
+  }
+
   for(;;) { /* speculative query */
     fd_block_map_query_t query[1] = { 0 };
     int err = fd_block_map_query_try( ctx->blockstore->block_map, &slot, NULL, query, 0 );
@@ -447,7 +455,7 @@ slice_poll( fd_replay_tile_ctx_t * ctx,
     if( FD_UNLIKELY( err == FD_MAP_ERR_KEY   ) ) return 0;
     if( FD_UNLIKELY( err == FD_MAP_ERR_AGAIN ) ) continue;
 
-    consumed_idx = block_info->consumed_idx;
+    consumed_idx = fd_uint_min( block_info->consumed_idx, turbine_shred_wmark );
     slices_added = 0;
 
     if( FD_UNLIKELY( block_info->buffered_idx == UINT_MAX ) ) return 1;
@@ -513,6 +521,10 @@ before_frag( fd_replay_tile_ctx_t * ctx,
   (void)ctx;
   (void)seq;
 
+  if( in_idx == STORE_IN_IDX ) {
+    FD_LOG_WARNING((" RECEIVE STORE MESSAGE: slot %lu", fd_disco_replay_old_sig_slot( sig )));
+  }
+
   if( in_idx == SHRED_IN_IDX ) {
     // FD_LOG_NOTICE(( "shred in idx: %lu, seq: %lu, sig: %lu", in_idx, seq, sig ));
 
@@ -589,8 +601,6 @@ during_frag( fd_replay_tile_ctx_t * ctx,
              ulong                  chunk,
              ulong                  sz,
              ulong                  ctl FD_PARAM_UNUSED ) {
-  return;
-
   ctx->skip_frag = 0;
 
   if( in_idx == STORE_IN_IDX ) {
@@ -604,6 +614,7 @@ during_frag( fd_replay_tile_ctx_t * ctx,
        Microblock as a list of fd_txn_p_t (sz * sizeof(fd_txn_p_t)) */
 
     ctx->curr_slot = fd_disco_replay_old_sig_slot( sig );
+    FD_LOG_WARNING(("store sent slot %lu", ctx->curr_slot));
     /* slot changes */
     if( FD_UNLIKELY( ctx->curr_slot < fd_fseq_query( ctx->published_wmark ) ) ) {
       FD_LOG_WARNING(( "store sent slot %lu before our root.", ctx->curr_slot ));
@@ -1921,7 +1932,7 @@ after_frag( fd_replay_tile_ctx_t * ctx,
   (void)sig;
   (void)sz;
   (void)seq;
-  return;
+  //return;
 
   if( FD_LIKELY( in_idx == SHRED_IN_IDX ) ) {
     /* We are here if any of the slot / shred idx numbers have been overrun */
@@ -2475,9 +2486,9 @@ after_credit( fd_replay_tile_ctx_t * ctx,
     }
   }
 
-  // exec_slices( ctx, stem, ctx->curr_slot );
+  exec_slices( ctx, stem, ctx->curr_slot );
 
-  return;
+  //return;
 
   ulong curr_slot   = ctx->curr_slot;
   ulong parent_slot = ctx->parent_slot;
@@ -2740,7 +2751,7 @@ after_credit( fd_replay_tile_ctx_t * ctx,
 
 static void
 during_housekeeping( void * _ctx ) {
-  return;
+  //return;
 
   fd_replay_tile_ctx_t * ctx = (fd_replay_tile_ctx_t *)_ctx;