Skip to content

replay: track and publish scheduler slot states #12513

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jherrera-jump merged 1 commit into from
Oct 23, 2026
Merged

replay: track and publish scheduler slot states #12513

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions src/discof/replay/fd_replay_tile.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 );
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -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
Expand Down
9 changes: 9 additions & 0 deletions src/discof/replay/fd_replay_tile.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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 */
Expand Down Expand Up @@ -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;
Expand All @@ -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;
Expand Down
149 changes: 141 additions & 8 deletions src/discof/replay/fd_sched.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -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_tick<sched->state_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,
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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 ) );

Expand Down Expand Up @@ -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;
Expand All @@ -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
Expand Down Expand Up @@ -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 );
Expand Down Expand Up @@ -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 ));
Expand Down Expand Up @@ -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;
}
}
Expand All @@ -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 );
Expand Down Expand Up @@ -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 ) ) {
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -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;
}
}
Expand Down Expand Up @@ -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. */
Expand Down Expand Up @@ -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 ) {
Expand All @@ -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++;
}

Expand Down Expand Up @@ -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 ) {
Expand Down Expand Up @@ -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
Expand All @@ -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",
Expand Down Expand Up @@ -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 );
}
Expand Down
Loading
Loading