From 1ab73151df4be0307455d0687b8d64834965639b Mon Sep 17 00:00:00 2001 From: jherrera-jump Date: Fri, 23 Oct 2026 19:52:01 +0000 Subject: [PATCH] replay: track and publish scheduler slot states --- src/discof/replay/fd_replay_tile.c | 12 ++- src/discof/replay/fd_replay_tile.h | 9 ++ src/discof/replay/fd_sched.c | 149 +++++++++++++++++++++++++-- src/discof/replay/fd_sched.h | 43 ++++++++ src/discof/replay/test_replay_tile.c | 2 + src/discof/replay/test_sched.c | 114 ++++++++++++-------- 6 files changed, 276 insertions(+), 53 deletions(-) diff --git a/src/discof/replay/fd_replay_tile.c b/src/discof/replay/fd_replay_tile.c index b5bbbb4e438..b16f273e932 100644 --- a/src/discof/replay/fd_replay_tile.c +++ b/src/discof/replay/fd_replay_tile.c @@ -588,7 +588,7 @@ publish_replay_out( fd_replay_tile_t * ctx, ulong sig, ulong sz ) { ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() ); - if( FD_UNLIKELY( sig!=REPLAY_SIG_TXN_EXECUTED ) ) { + if( FD_UNLIKELY( sig!=REPLAY_SIG_TXN_EXECUTED && sig!=REPLAY_SIG_SLOT_STATES ) ) { fd_replay_out_link_t * out = sig==REPLAY_SIG_MISSING_FEC ? ctx->rotor_out : ctx->slot_out; if( FD_LIKELY( out->idx!=ULONG_MAX ) ) { /* backtest has no rotor */ fd_memcpy( fd_chunk_to_laddr( out->mem, out->chunk ), fd_chunk_to_laddr_const( ctx->replay_out->mem, ctx->replay_out->chunk ), sz ); @@ -3356,6 +3356,11 @@ after_credit( fd_replay_tile_t * ctx, fd_stem_context_t * stem, int * opt_poll_in, int * charge_busy ) { + /* Publish slot state intervals before the halt checks. */ + fd_replay_slot_states_t * states = fd_chunk_to_laddr( ctx->replay_out->mem, ctx->replay_out->chunk ); + states->cnt = fd_sched_slot_states_drain( ctx->sched, states->states ); + if( FD_UNLIKELY( states->cnt ) ) publish_replay_out( ctx, stem, REPLAY_SIG_SLOT_STATES, sizeof(fd_replay_slot_states_t) ); + if( FD_UNLIKELY( ctx->halt_replay && !ctx->is_leader ) ) return; if( FD_UNLIKELY( !ctx->is_booted || !ctx->wfs_complete ) ) return; @@ -4810,8 +4815,9 @@ during_housekeeping( fd_replay_tile_t * ctx ) { completed, next leader, reset, became leader). after_credit and returnable_frag never both publish in one iteration: every after_credit path that publishes returns nonzero, which skips the - input poll. */ -#define STEM_BURST (4UL) + input poll. The exception is REPLAY_SIG_SLOT_STATES, which doesn't + skip the input poll. */ +#define STEM_BURST (5UL) /* fd_tempo_lazy_default( 16384 ) where 16384 is the minimum out-link depth (i.e. cr_max) but excludes replay_epoch and replay_slot, which diff --git a/src/discof/replay/fd_replay_tile.h b/src/discof/replay/fd_replay_tile.h index 4ec977e4bdb..91a82806950 100644 --- a/src/discof/replay/fd_replay_tile.h +++ b/src/discof/replay/fd_replay_tile.h @@ -100,6 +100,7 @@ #include "../../disco/tiles.h" #include "../../choreo/votor/ag_cert.h" #include "../../flamenco/alpenglow/fd_block_marker.h" +#include "fd_sched.h" #define REPLAY_SIG_SLOT_COMPLETED (0) #define REPLAY_SIG_SLOT_DEAD (1) @@ -116,6 +117,7 @@ #define REPLAY_SIG_LEADER_FOOTER (12) #define REPLAY_SIG_MISSING_FEC (13) #define REPLAY_SIG_NEXT_LEADER (14) +#define REPLAY_SIG_SLOT_STATES (15) /* replay_out mcache seq[i] slots */ #define REPLAY_SYNC_SEQ (0UL) /* mcache->seq[0]: recently published seq no */ @@ -314,6 +316,12 @@ struct fd_replay_next_leader { }; typedef struct fd_replay_next_leader fd_replay_next_leader_t; +struct fd_replay_slot_states { + ulong cnt; + fd_sched_slot_state_t states[ FD_SCHED_SLOT_STATE_MAX ]; +}; +typedef struct fd_replay_slot_states fd_replay_slot_states_t; + union fd_replay_message { fd_replay_slot_completed_t slot_completed; fd_replay_slot_dead_t slot_dead; @@ -326,6 +334,7 @@ union fd_replay_message { fd_replay_drop_bank_ref_t drop_bank_ref; fd_replay_leader_footer_t leader_footer; fd_replay_next_leader_t next_leader; + fd_replay_slot_states_t slot_states; }; typedef union fd_replay_message fd_replay_message_t; diff --git a/src/discof/replay/fd_sched.c b/src/discof/replay/fd_sched.c index 8debd7bceb3..fefd9cc5b3e 100644 --- a/src/discof/replay/fd_sched.c +++ b/src/discof/replay/fd_sched.c @@ -188,6 +188,14 @@ struct fd_sched_block { uint staged:1; /* Set if the block is in a dispatcher staging lane; a staged block is tracked by the dispatcher. */ ulong staging_lane; /* Ignored if staged==0. */ + + /* Slot state tracking */ + long state_start_tick; /* Start of the committed state's interval. */ + long state_cand_tick; /* Start of the candidate state. */ + uchar state; /* Committed state, NONE until the first commit. */ + uchar state_cand; /* Candidate state, committed once held for FD_SCHED_SLOT_STATE_DEBOUNCE_NS. */ + uchar state_open; + uchar state_conflict; /* Last next_ready found queued txns but none ready. */ ulong luf_depth; /* Depth of longest unstaged fork starting from this node; only stageable unstaged descendants are counted. */ int dead_reason; /* One of FD_SCHED_DEAD_REASON_*; records the first reason the block @@ -324,6 +332,12 @@ struct fd_sched { fd_sched_block_t * block_pool; /* Just a flat array. */ ulong block_pool_popcnt; ulong * ref_q; + + long state_debounce_ticks; + ulong state_bank_idx; /* Block whose slot state is open, ULONG_MAX if none. */ + ulong state_cnt; + ulong state_drop_cnt; + fd_sched_slot_state_t state_buf[ FD_SCHED_SLOT_STATE_MAX ]; }; typedef struct fd_sched fd_sched_t; @@ -521,6 +535,90 @@ block_is_prunable( fd_sched_block_t * block ) { static inline ulong block_to_idx( fd_sched_t * sched, fd_sched_block_t * block ) { return (ulong)(block-sched->block_pool); } +static void +slot_state_push( fd_sched_t * sched, fd_sched_block_t * block, long end_tick ) { + if( FD_UNLIKELY( sched->state_cnt>=FD_SCHED_SLOT_STATE_MAX ) ) { sched->state_drop_cnt++; return; } + sched->state_buf[ sched->state_cnt++ ] = (fd_sched_slot_state_t){ + .slot = block->slot, + .start_tick = block->state_start_tick, + .end_tick = end_tick, + .state = block->state + }; +} + +static void +slot_state_maybe_commit( fd_sched_t * sched, fd_sched_block_t * block, long now ) { + if( block->state_cand==block->state || now-block->state_cand_tickstate_debounce_ticks ) return; + /* The first committed state starts at the window start. */ + if( FD_LIKELY( block->state!=FD_SCHED_SLOT_STATE_NONE ) ) { + slot_state_push( sched, block, block->state_cand_tick ); + block->state_start_tick = block->state_cand_tick; + } + block->state = block->state_cand; +} + +static void +slot_state_eval( fd_sched_t * sched, fd_sched_block_t * block ) { + if( FD_LIKELY( !block->state_open ) ) return; + + ulong queued_cnt = block->txn_parsed_cnt-block->txn_exec_in_flight_cnt-block->txn_exec_done_cnt; + ulong free_tiles = sched->txn_exec_ready_bitset[ 0 ] & sched->sigverify_ready_bitset[ 0 ] & sched->poh_ready_bitset[ 0 ] & sched->cert_ready_bitset[ 0 ]; + uchar next; + if( block->block_end_signaled ) next = FD_SCHED_SLOT_STATE_FINISH; + else if( queued_cnt && !free_tiles ) next = FD_SCHED_SLOT_STATE_BLOCKED_TILES; + else if( queued_cnt && block->state_conflict ) next = FD_SCHED_SLOT_STATE_BLOCKED_CONFLICTS; + else if( queued_cnt ) next = block->state_cand; /* ready txn and free tile, dispatch is imminent */ + else if( !block->fec_eos ) next = FD_SCHED_SLOT_STATE_BLOCKED_NETWORK; + else next = FD_SCHED_SLOT_STATE_FINISH; + + long now = fd_tickcount(); + slot_state_maybe_commit( sched, block, now ); + if( next!=block->state_cand ) { + block->state_cand = next; + block->state_cand_tick = now; + } +} + +static void +slot_state_close( fd_sched_t * sched, fd_sched_block_t * block ); + +static void +slot_state_open( fd_sched_t * sched, fd_sched_block_t * block ) { + ulong bank_idx = block_to_idx( sched, block ); + if( FD_UNLIKELY( sched->state_bank_idx!=ULONG_MAX && sched->state_bank_idx!=bank_idx ) ) { + slot_state_close( sched, block_pool_ele( sched, sched->state_bank_idx ) ); + } + sched->state_bank_idx = bank_idx; + block->state_open = 1; + block->state_conflict = 0; + block->state = FD_SCHED_SLOT_STATE_NONE; + block->state_cand = FD_SCHED_SLOT_STATE_NONE; + block->state_start_tick = block->state_cand_tick = fd_tickcount(); +} + +static void +slot_state_close( fd_sched_t * sched, fd_sched_block_t * block ) { + if( FD_LIKELY( !block->state_open ) ) return; + block->state_open = 0; + sched->state_bank_idx = ULONG_MAX; + long now = fd_tickcount(); + slot_state_maybe_commit( sched, block, now ); + if( FD_UNLIKELY( block->state==FD_SCHED_SLOT_STATE_NONE ) ) block->state = block->state_cand; /* window shorter than the debounce */ + if( FD_UNLIKELY( block->state==FD_SCHED_SLOT_STATE_NONE ) ) return; + slot_state_push( sched, block, now ); +} + +/* active_bank_set changes the active block. Slot states are tracked + for at most one block at a time. */ + +static void +active_bank_set( fd_sched_t * sched, ulong bank_idx ) { + sched->active_bank_idx = bank_idx; + if( FD_UNLIKELY( bank_idx==ULONG_MAX ) ) return; + fd_sched_block_t * block = block_pool_ele( sched, bank_idx ); + if( FD_LIKELY( block->block_start_signaled && !block->block_end_done && !block->dying && !block->state_open ) ) slot_state_open( sched, block ); +} + __attribute__((format(printf,2,3))) static void fd_sched_printf( fd_sched_t * sched, @@ -802,6 +900,10 @@ fd_sched_new( void * mem, sched->txn_in_flight_last_tick = fd_tickcount(); /* boot counts as idle */ sched->next_ready_last_tick = LONG_MAX; sched->next_ready_last_bank_idx = ULONG_MAX; + sched->state_debounce_ticks = (long)( (double)FD_SCHED_SLOT_STATE_DEBOUNCE_NS*fd_tempo_tick_per_ns( NULL ) ); + sched->state_bank_idx = ULONG_MAX; + sched->state_cnt = 0UL; + sched->state_drop_cnt = 0UL; sched->canary = FD_SCHED_MAGIC; sched->depth = depth; @@ -1253,14 +1355,16 @@ fd_sched_fec_ingest( fd_sched_t * sched, FD_TEST( mixin_res==1||mixin_res==2 ); } + slot_state_eval( sched, block ); + /* Check if we need to set the active block. */ check_or_set_active_block( sched ); return 1; } -ulong -fd_sched_task_next_ready( fd_sched_t * sched, fd_sched_task_t * out ) { +static ulong +task_next_ready( fd_sched_t * sched, fd_sched_task_t * out ) { FD_TEST( sched->canary==FD_SCHED_MAGIC ); FD_TEST( ref_q_empty( sched->ref_q ) ); @@ -1303,6 +1407,7 @@ fd_sched_task_next_ready( fd_sched_t * sched, fd_sched_task_t * out ) { out->block_start->parent_bank_idx = block->parent_idx; out->block_start->slot = block->slot; block->block_start_signaled = 1; + slot_state_open( sched, block ); sched->next_ready_last_tick = fd_tickcount(); sched->next_ready_last_bank_idx = bank_idx; return 1UL; @@ -1323,6 +1428,7 @@ fd_sched_task_next_ready( fd_sched_t * sched, fd_sched_task_t * out ) { tile, e.g. the one that just completed said transaction, readily available to continue executing down the critical path. */ out->txn_exec->txn_idx = fd_rdisp_get_next_ready( sched->rdisp, bank_idx ); + block->state_conflict = !out->txn_exec->txn_idx; if( FD_UNLIKELY( out->txn_exec->txn_idx==0UL ) ) { /* There are transactions queued but none ready for execution. This implies that there must be in-flight transactions on whose @@ -1510,6 +1616,14 @@ fd_sched_task_next_ready( fd_sched_t * sched, fd_sched_task_t * out ) { return 0UL; } +ulong +fd_sched_task_next_ready( fd_sched_t * sched, fd_sched_task_t * out ) { + ulong bank_idx = sched->active_bank_idx; + ulong ret = task_next_ready( sched, out ); + if( FD_LIKELY( bank_idx!=ULONG_MAX ) ) slot_state_eval( sched, block_pool_ele( sched, bank_idx ) ); + return ret; +} + int fd_sched_task_done( fd_sched_t * sched, ulong task_type, ulong txn_idx, ulong exec_idx, void * data ) { FD_TEST( sched->canary==FD_SCHED_MAGIC ); @@ -1573,6 +1687,7 @@ fd_sched_task_done( fd_sched_t * sched, ulong task_type, ulong txn_idx, ulong ex transition for empty blocks or slow blocks. */ FD_TEST( !block->block_end_done ); block->block_end_done = 1; + if( FD_LIKELY( block_is_done( block ) ) ) slot_state_close( sched, block ); /* else on the last cert task done */ sched->print_buf_sz = 0UL; print_block_metrics( sched, block ); FD_LOG_DEBUG(( "block %lu:%lu replayed fully: %s", block->slot, bank_idx, sched->print_buf )); @@ -1665,6 +1780,7 @@ fd_sched_task_done( fd_sched_t * sched, ulong task_type, ulong txn_idx, ulong ex block->cert_in_flight_cnt--; FD_TEST( !fd_ulong_extract_bit( sched->cert_ready_bitset[ 0 ], exec_tile_idx ) ); sched->cert_ready_bitset[ 0 ] = fd_ulong_set_bit( sched->cert_ready_bitset[ 0 ], exec_tile_idx ); + if( FD_UNLIKELY( block_is_done( block ) ) ) slot_state_close( sched, block ); break; } } @@ -1680,6 +1796,8 @@ fd_sched_task_done( fd_sched_t * sched, ulong task_type, ulong txn_idx, ulong ex return 0; } + slot_state_eval( sched, block ); + if( FD_UNLIKELY( !block->dying && sched->active_bank_idx!=bank_idx ) ) { /* Block is not dead. So we should be actively replaying it. */ fd_sched_block_t * active_block = block_pool_ele( sched, sched->active_bank_idx ); @@ -1905,6 +2023,18 @@ fd_sched_root_notify( fd_sched_t * sched, ulong root_idx ) { } } +ulong +fd_sched_slot_states_drain( fd_sched_t * sched, fd_sched_slot_state_t * out ) { + ulong cnt = sched->state_cnt; + if( FD_UNLIKELY( sched->state_drop_cnt ) ) { + FD_LOG_WARNING(( "dropped %lu slot state intervals", sched->state_drop_cnt )); + sched->state_drop_cnt = 0UL; + } + fd_memcpy( out, sched->state_buf, cnt*sizeof(fd_sched_slot_state_t) ); + sched->state_cnt = 0UL; + return cnt; +} + ulong fd_sched_pruned_block_next( fd_sched_t * sched ) { if( !ref_q_empty( sched->ref_q ) ) { @@ -2145,6 +2275,7 @@ add_block( fd_sched_t * sched, block->block_start_done = 0; block->block_end_done = 0; block->staged = 0; + block->state_open = 0; block->luf_depth = 0UL; @@ -3052,7 +3183,7 @@ try_activate_block( fd_sched_t * sched ) { FD_LOG_DEBUG(( "activating block %lu:%lu: lane %d waking up on slot boundary", head_block->slot, head_idx, lane_idx )); } } - sched->active_bank_idx = head_idx; + active_bank_set( sched, head_idx ); return; } } @@ -3153,7 +3284,7 @@ try_activate_block( fd_sched_t * sched ) { unstaged fork. */ fd_sched_block_t * head_block = block_pool_ele( sched, head_bank_idx ); FD_LOG_DEBUG(( "activating block %lu:%lu: unstaged promotion to lane %d", head_block->slot, head_bank_idx, lane_idx )); - sched->active_bank_idx = head_bank_idx; + active_bank_set( sched, head_bank_idx ); return; } /* No unstaged blocks to promote. So we're done. Yay. */ @@ -3220,6 +3351,7 @@ subtree_mark_and_maybe_prune_rdisp( fd_sched_t * sched, fd_sched_block_t * block /* Setting the flag is non-optional and can happen more than once. */ block->dying = 1; + slot_state_close( sched, block ); /* Removal from dispatcher should only happen once. */ if( block->in_rdisp ) { @@ -3241,7 +3373,7 @@ subtree_mark_and_maybe_prune_rdisp( fd_sched_t * sched, fd_sched_block_t * block FD_TEST( block_pool_ele( sched, sched->active_bank_idx )==block ); FD_LOG_DEBUG(( "reset active_bank_idx %lu: abandon", sched->active_bank_idx )); sched->last_active_bank_idx = sched->active_bank_idx; - sched->active_bank_idx = ULONG_MAX; + active_bank_set( sched, ULONG_MAX ); sched->metrics->deactivate_abandoned_cnt++; } @@ -3367,6 +3499,7 @@ subtree_prune( fd_sched_t * sched, ulong bank_idx, ulong except_idx ) { FD_TEST( !head->refcnt ); FD_TEST( head->in_sched ); head->in_sched = 0; + slot_state_close( sched, head ); ulong child_idx = head->child_idx; while( child_idx!=ULONG_MAX ) { @@ -3445,7 +3578,7 @@ maybe_switch_block( fd_sched_t * sched, ulong bank_idx ) { sched->metrics->block_removed_cnt++; FD_LOG_DEBUG(( "reset active_bank_idx %lu: remove", sched->active_bank_idx )); sched->last_active_bank_idx = sched->active_bank_idx; - sched->active_bank_idx = ULONG_MAX; + active_bank_set( sched, ULONG_MAX ); /* See if there is a child block down the same staging lane. This is a policy decision to minimize fork churn. We could in theory @@ -3469,7 +3602,7 @@ maybe_switch_block( fd_sched_t * sched, ulong bank_idx ) { block to it, and have the child inherit the head status of the lane. This is the common case. */ FD_LOG_DEBUG(( "activating block %lu:%lu: child inheritance on lane %lu", child->slot, child_idx, child->staging_lane )); - sched->active_bank_idx = child_idx; + active_bank_set( sched, child_idx ); sched->staged_head_bank_idx[ block->staging_lane ] = child_idx; if( FD_UNLIKELY( !fd_ulong_extract_bit( sched->staged_bitset, (int)block->staging_lane ) ) ) { FD_LOG_CRIT(( "invariant violation: staged_bitset 0x%lx bit %lu is not set, slot %lu, parent slot %lu, child slot %lu, parent slot %lu", @@ -3498,7 +3631,7 @@ maybe_switch_block( fd_sched_t * sched, ulong bank_idx ) { happen when the network path is congested, or when the leader simply went down. Reset the active block. */ sched->last_active_bank_idx = sched->active_bank_idx; - sched->active_bank_idx = ULONG_MAX; + active_bank_set( sched, ULONG_MAX ); sched->metrics->deactivate_no_txn_cnt++; try_activate_block( sched ); } diff --git a/src/discof/replay/fd_sched.h b/src/discof/replay/fd_sched.h index 2a33b55982a..0580fce9176 100644 --- a/src/discof/replay/fd_sched.h +++ b/src/discof/replay/fd_sched.h @@ -152,6 +152,43 @@ typedef struct fd_sched_txn_info fd_sched_txn_info_t; #define FD_SCHED_TT_MARK_DEAD (7UL) /* (i) Mark the block dead. */ #define FD_SCHED_TT_CERT_VERIFY (8UL) /* (e) Footer cert verify. */ +/* Slot-level replay states. States are tracked for one block at a + time, the one the scheduler is working on. A block's interval opens + at its first dispatch (BLOCK_START). It closes when another block's + interval opens, or when end of block processing completes or the + block dies. A block deactivated while waiting on the network stays + tracked until another block takes over. + + BLOCKED_TILES no exec tile is free + BLOCKED_CONFLICTS txns queued, but waiting on in-flight conflicting txns + BLOCKED_NETWORK no txns queued, waiting for more to arrive + FINISH end of the slot received, txns or PoH still in flight + + States are only reevaluated on FEC ingest, task_next_ready and + task_done. A new state is committed only once it has held for at + least FD_SCHED_SLOT_STATE_DEBOUNCE_NS. */ + +#define FD_SCHED_SLOT_STATE_NONE (0) +#define FD_SCHED_SLOT_STATE_BLOCKED_TILES (1) +#define FD_SCHED_SLOT_STATE_BLOCKED_CONFLICTS (2) +#define FD_SCHED_SLOT_STATE_BLOCKED_NETWORK (3) +#define FD_SCHED_SLOT_STATE_FINISH (4) + +#define FD_SCHED_SLOT_STATE_DEBOUNCE_NS (10000L) + +/* Max state transitions between after_credit drains. The real bound + is ~3, but a tight bound needs an exhaustive proof tightly coupled + to the sched state machine. Filled buffer is handled gracefully. */ +#define FD_SCHED_SLOT_STATE_MAX (32UL) + +struct fd_sched_slot_state { + ulong slot; + long start_tick; + long end_tick; + ulong state; /* FD_SCHED_SLOT_STATE_* */ +}; +typedef struct fd_sched_slot_state fd_sched_slot_state_t; + struct fd_sched_block_start { ulong bank_idx; /* Same as in fd_sched_fec_t. */ ulong parent_bank_idx; /* Same as in fd_sched_fec_t. */ @@ -500,6 +537,12 @@ fd_sched_root_notify( fd_sched_t * sched, ulong root_idx ); ulong fd_sched_pruned_block_next( fd_sched_t * sched ); +/* fd_sched_slot_states_drain copies the slot state intervals closed + since the last drain into out (room for FD_SCHED_SLOT_STATE_MAX) and + returns how many were copied. Intervals are in tickcount units. */ +ulong +fd_sched_slot_states_drain( fd_sched_t * sched, fd_sched_slot_state_t * out ); + void fd_sched_set_poh_params( fd_sched_t * sched, ulong bank_idx, ulong tick_height, ulong max_tick_height, fd_hash_t const * start_poh ); diff --git a/src/discof/replay/test_replay_tile.c b/src/discof/replay/test_replay_tile.c index a3947ea9095..5d9c822c386 100644 --- a/src/discof/replay/test_replay_tile.c +++ b/src/discof/replay/test_replay_tile.c @@ -112,6 +112,7 @@ void mock_sched_abandon_fn ( fd_sched_t * s FD_PARAM_UNUSED, ulong i, int i void mock_sched_cancel_fn ( fd_sched_t * s FD_PARAM_UNUSED, ulong i ) { mock_sched_cancel_idx = i; } int mock_sched_is_discarded_fn( fd_sched_t * s FD_PARAM_UNUSED, ulong i FD_PARAM_UNUSED ) { return 0; } ulong mock_sched_pruned_fn ( fd_sched_t * s FD_PARAM_UNUSED ) { return ULONG_MAX; } +ulong mock_sched_states_fn ( fd_sched_t * s FD_PARAM_UNUSED, fd_sched_slot_state_t * o FD_PARAM_UNUSED ) { return 0UL; } void mock_sched_metrics_fn ( fd_sched_t * s FD_PARAM_UNUSED, fd_clock_tile_t const * c FD_PARAM_UNUSED ) {} void mock_sched_poh_fn ( fd_sched_t * s FD_PARAM_UNUSED, ulong a FD_PARAM_UNUSED, ulong b FD_PARAM_UNUSED, ulong c FD_PARAM_UNUSED, fd_hash_t const * d FD_PARAM_UNUSED ) {} ulong mock_sched_task_next_fn ( fd_sched_t * s FD_PARAM_UNUSED, fd_sched_task_t * t FD_PARAM_UNUSED ) { return 0UL; } @@ -155,6 +156,7 @@ mock_sched_get_txn_info_fn( fd_sched_t * s FD_PARAM_UNUSED, #define fd_sched_cancel mock_sched_cancel_fn #define fd_sched_block_is_discarded mock_sched_is_discarded_fn #define fd_sched_pruned_block_next mock_sched_pruned_fn +#define fd_sched_slot_states_drain mock_sched_states_fn #define fd_sched_metrics_write mock_sched_metrics_fn #define fd_sched_set_poh_params mock_sched_poh_fn #define fd_sched_task_next_ready mock_sched_task_next_fn diff --git a/src/discof/replay/test_sched.c b/src/discof/replay/test_sched.c index e298e5e194c..4b25d245bc3 100644 --- a/src/discof/replay/test_sched.c +++ b/src/discof/replay/test_sched.c @@ -17,10 +17,10 @@ static void test_sched_footprint( void ) { /* Retain the savings from compact shred lengths and 4992-byte transactions under the default scheduler sizing. */ - FD_TEST( fd_sched_footprint( 65536UL, 2048UL, FD_SHRED_BLK_MAX, FD_MAX_TXN_PER_SLOT )==1692836096UL ); + FD_TEST( fd_sched_footprint( 65536UL, 2048UL, FD_SHRED_BLK_MAX, FD_MAX_TXN_PER_SLOT )==1692837120UL ); /* Only the shred length array scales with the shred limit. */ - FD_TEST( fd_sched_footprint( 65536UL, 2048UL, 4UL*FD_SHRED_BLK_MAX, FD_MAX_TXN_PER_SLOT )==1692836096UL+2048UL*3UL*FD_SHRED_BLK_MAX*sizeof(ushort) ); - FD_TEST( fd_sched_footprint( 65536UL, 2048UL, FD_SHRED_BLK_MAX, 5UL*FD_MAX_TXN_PER_SLOT )==1692836096UL ); + FD_TEST( fd_sched_footprint( 65536UL, 2048UL, 4UL*FD_SHRED_BLK_MAX, FD_MAX_TXN_PER_SLOT )==1692837120UL+2048UL*3UL*FD_SHRED_BLK_MAX*sizeof(ushort) ); + FD_TEST( fd_sched_footprint( 65536UL, 2048UL, FD_SHRED_BLK_MAX, 5UL*FD_MAX_TXN_PER_SLOT )==1692837120UL ); FD_TEST( !fd_sched_footprint( 65536UL, 2048UL, 0UL, FD_MAX_TXN_PER_SLOT ) ); FD_TEST( !fd_sched_footprint( 65536UL, 2048UL, FD_SHRED_BLK_MAX, 0UL ) ); } @@ -169,6 +169,48 @@ ingest_ag_component( fd_sched_t * sched, return fd_sched_fec_ingest( sched, fec ); } +static void +run_interleaved_fec_tasks( fd_sched_t * sched, uint const * signature_tag, ulong * txn_exec_cnt ) { + for( ulong step=0UL; step<100UL; step++ ) { + while( fd_sched_pruned_block_next( sched )!=ULONG_MAX ) {} + + fd_sched_task_t task[ 1 ]; + if( !fd_sched_task_next_ready( sched, task ) ) break; + switch( task->task_type ) { + case FD_SCHED_TT_BLOCK_START: + FD_TEST( !fd_sched_task_done( sched, FD_SCHED_TT_BLOCK_START, ULONG_MAX, ULONG_MAX, NULL ) ); + break; + case FD_SCHED_TT_BLOCK_END: + FD_TEST( !fd_sched_task_done( sched, FD_SCHED_TT_BLOCK_END, ULONG_MAX, ULONG_MAX, NULL ) ); + break; + case FD_SCHED_TT_TXN_EXEC: { + fd_txn_p_t * txn = fd_sched_get_txn( sched, task->txn_exec->txn_idx ); + FD_TEST( task->txn_exec->bank_idx==2UL || task->txn_exec->bank_idx==3UL ); + FD_TEST( FD_LOAD( uint, txn->payload+1UL )==signature_tag[ task->txn_exec->bank_idx-2UL ] ); + FD_TEST( txn->start_shred_idx==0U ); + FD_TEST( txn->end_shred_idx==fd_ushort_if( task->txn_exec->bank_idx==2UL, 3U, 2U ) ); + (*txn_exec_cnt)++; + FD_TEST( !fd_sched_task_done( sched, FD_SCHED_TT_TXN_EXEC, task->txn_exec->txn_idx, task->txn_exec->exec_idx, NULL ) ); + break; + } + case FD_SCHED_TT_TXN_SIGVERIFY: + FD_TEST( !fd_sched_task_done( sched, FD_SCHED_TT_TXN_SIGVERIFY, task->txn_sigverify->txn_idx, task->txn_sigverify->exec_idx, NULL ) ); + break; + case FD_SCHED_TT_POH_HASH: { + fd_execrp_poh_hash_done_msg_t msg[ 1 ]; + msg->cnt = task->poh_hash->cnt; + for( ulong i=0UL; ipoh_hash->cnt; i++ ) { + repeat_hash( msg->hash+i, task->poh_hash->hash+i, task->poh_hash->hashcnt ); + } + FD_TEST( !fd_sched_task_done( sched, FD_SCHED_TT_POH_HASH, ULONG_MAX, task->poh_hash->exec_idx, msg ) ); + break; + } + default: + FD_LOG_ERR(( "unexpected task type %lu in interleaved FEC test", task->task_type )); + } + } +} + static void run_interleaved_fec_residual_case( void ) { ulong footprint = fd_sched_footprint( FD_SCHED_MIN_DEPTH, 4UL, FD_SHRED_BLK_MAX, FD_MAX_TXN_PER_SLOT ); @@ -241,6 +283,13 @@ run_interleaved_fec_residual_case( void ) { FD_TEST( fd_sched_fec_ingest( sched, fec ) ); } + /* Only the first half of each txn has arrived, so both blocks wait + on the network after BLOCK_START. */ + ulong txn_exec_cnt = 0UL; + run_interleaved_fec_tasks( sched, signature_tag, &txn_exec_cnt ); + long wait_until = fd_log_wallclock()+1000000L; + while( fd_log_wallclock()task_type ) { - case FD_SCHED_TT_BLOCK_START: - FD_TEST( !fd_sched_task_done( sched, FD_SCHED_TT_BLOCK_START, ULONG_MAX, ULONG_MAX, NULL ) ); - break; - case FD_SCHED_TT_BLOCK_END: - FD_TEST( !fd_sched_task_done( sched, FD_SCHED_TT_BLOCK_END, ULONG_MAX, ULONG_MAX, NULL ) ); - break; - case FD_SCHED_TT_TXN_EXEC: { - fd_txn_p_t * txn = fd_sched_get_txn( sched, task->txn_exec->txn_idx ); - FD_TEST( task->txn_exec->bank_idx==2UL || task->txn_exec->bank_idx==3UL ); - FD_TEST( FD_LOAD( uint, txn->payload+1UL )==signature_tag[ task->txn_exec->bank_idx-2UL ] ); - FD_TEST( txn->start_shred_idx==0U ); - FD_TEST( txn->end_shred_idx==fd_ushort_if( task->txn_exec->bank_idx==2UL, 3U, 2U ) ); - txn_exec_cnt++; - FD_TEST( !fd_sched_task_done( sched, FD_SCHED_TT_TXN_EXEC, task->txn_exec->txn_idx, task->txn_exec->exec_idx, NULL ) ); - break; - } - case FD_SCHED_TT_TXN_SIGVERIFY: - FD_TEST( !fd_sched_task_done( sched, FD_SCHED_TT_TXN_SIGVERIFY, task->txn_sigverify->txn_idx, task->txn_sigverify->exec_idx, NULL ) ); - break; - case FD_SCHED_TT_POH_HASH: { - fd_execrp_poh_hash_done_msg_t msg[ 1 ]; - msg->cnt = task->poh_hash->cnt; - for( ulong i=0UL; ipoh_hash->cnt; i++ ) { - repeat_hash( msg->hash+i, task->poh_hash->hash+i, task->poh_hash->hashcnt ); - } - FD_TEST( !fd_sched_task_done( sched, FD_SCHED_TT_POH_HASH, ULONG_MAX, task->poh_hash->exec_idx, msg ) ); - break; - } - default: - FD_LOG_ERR(( "unexpected task type %lu in interleaved FEC test", task->task_type )); - } - } + run_interleaved_fec_tasks( sched, signature_tag, &txn_exec_cnt ); while( fd_sched_pruned_block_next( sched )!=ULONG_MAX ) {} FD_TEST( txn_exec_cnt==2UL ); FD_TEST( fd_sched_is_drained( sched ) ); + /* Both blocks wait on the network, but only one block is tracked at + a time: block 2's interval closes when block 3 starts, and block 3 + covers the ~1ms wait. Intervals are emitted in order and must not + overlap, and the network wait is counted once. */ + fd_sched_slot_state_t states[ FD_SCHED_SLOT_STATE_MAX ]; + ulong state_cnt = fd_sched_slot_states_drain( sched, states ); + long network_ticks = 0L; + long min_wait = (long)( 500000.0*fd_tempo_tick_per_ns( NULL ) ); + FD_TEST( state_cnt ); + for( ulong i=0UL; i=states[ i ].start_tick ); + if( i ) FD_TEST( states[ i ].start_tick>=states[ i-1UL ].end_tick ); + if( states[ i ].state==FD_SCHED_SLOT_STATE_BLOCKED_NETWORK ) network_ticks += states[ i ].end_tick-states[ i ].start_tick; + } + FD_TEST( network_ticks>=min_wait ); + FD_TEST( network_ticks<=states[ state_cnt-1UL ].end_tick-states[ 0 ].start_tick ); + FD_TEST( !fd_sched_slot_states_drain( sched, states ) ); + fd_sched_delete( fd_sched_leave( sched ) ); free( mem ); }