Skip to content

rotor: report fec masks & sample rtt off shred tile arrival time #12414

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
emwang-jump merged 1 commit into from
Oct 21, 2026
+100 −47
Merged

rotor: report fec masks & sample rtt off shred tile arrival time #12414

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
8 changes: 4 additions & 4 deletions book/api/metrics-generated.md
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

14 changes: 13 additions & 1 deletion src/choreo/rotor/fd_rotor.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -678,6 +678,9 @@ fd_rotor_fec_complete( fd_rotor_t * rotor,
uint turbine_cnt,
uint repair_cnt,
uint recovered_cnt,
uint data_rcvd,
uint parity_rcvd,
uint repair_rcvd,
long ts ) {
fd_rotor_shred_insert( rotor, last_shred, fec_mr, 0, ts ); /* the last shred's header, its source was given when it arrived */

Expand All @@ -689,11 +692,14 @@ fd_rotor_fec_complete( fd_rotor_t * rotor,
int first_complete = !fec->complete;
fec->cmpl_ts = fd_long_if( first_complete, ts, fec->cmpl_ts );
fec->final_src = fd_int_if( first_complete && is_leader, FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_LEADER, fec->final_src );
fec->data_rcvd = fd_uint_if( first_complete, data_rcvd, fec->data_rcvd );
fec->parity_rcvd = fd_uint_if( first_complete, parity_rcvd, fec->parity_rcvd );
fec->repair_rcvd = fd_uint_if( first_complete, repair_rcvd, fec->repair_rcvd );
fec->mr32 = *fec_mr;
fec->rcvd = UINT_MAX;
fec->complete = 1;
fec->data_complete = !!( last_shred->data.flags & FD_SHRED_DATA_FLAG_DATA_COMPLETE );
fec->is_leader |= is_leader; /* a completion again from the network does not clear our own */
fec->is_leader |= !!is_leader; /* a completion again from the network does not clear our own */

ulong slot = last_shred->slot; /* fd_shred_t is packed, the map query needs an aligned key */
fd_rotor_slot_meta_t * meta = fd_rotor_slot_meta( rotor, slot );
Expand Down Expand Up @@ -788,6 +794,9 @@ fd_rotor_fec_notarized( fd_rotor_t * rotor,
fec->first_ts = 0L;
fec->cmpl_ts = 0L;
fec->is_leader = 0;
fec->data_rcvd = 0U;
fec->parity_rcvd = 0U;
fec->repair_rcvd = 0U;
fd_rotor_fec_map_ele_insert( rotor->fec_map, fec, rotor->fec_pool );
}
fec->notarized = 1;
Expand Down Expand Up @@ -929,6 +938,9 @@ fd_rotor_shred_insert( fd_rotor_t * rotor,
fec->cmpl_ts = 0L;
fec->is_leader = 0;
fec->notarized = 0;
fec->data_rcvd = 0U;
fec->parity_rcvd = 0U;
fec->repair_rcvd = 0U;
fd_rotor_fec_map_ele_insert( rotor->fec_map, fec, rotor->fec_pool );
}
blk->fecs[ fec_idx ] = fd_rotor_fec_pool_idx( rotor->fec_pool, fec );
Expand Down
23 changes: 16 additions & 7 deletions src/choreo/rotor/fd_rotor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,15 +40,19 @@ struct fd_rotor_fec {
uint fec_idx;
ulong slot;
uint rcvd;
int complete;
int data_complete;
uchar complete : 1;
uchar data_complete : 1;
uchar is_leader : 1;
uchar notarized : 1;
ulong next;
ulong prev;
uint data_rcvd; /* shred tile masks at first completion: bit j set if the data shred with in-type index j was received (vs recovered) */
uint parity_rcvd; /* ... if the parity shred with in-type index j was received */
uint repair_rcvd; /* ... if the data shred with in-type index j arrived by repair */

int final_src; /* FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_* of the last data shred received before it completed, 0 if none */
long first_ts; /* when its first shred arrived, 0 if none yet */
long cmpl_ts; /* when it first completed, 0 if not yet */
int is_leader;
int notarized;
ulong next;
ulong prev;
};
typedef struct fd_rotor_fec fd_rotor_fec_t;
FD_STATIC_ASSERT( sizeof(fd_rotor_fec_t)==120UL, fd_rotor_fec );
Comment on lines +49 to -54

@lidatong lidatong Oct 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i think if you just jiggle these fields around, you'll find it fits in exactly 128, it probably is just interleaving padding

@emwang-jump emwang-jump Oct 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

reordered it so theres no interleaved padding, but its 136 because w/out padding the size of the struct is 132, could cut it down other ways

Expand Down Expand Up @@ -338,7 +342,9 @@ fd_rotor_slot_catchup( fd_rotor_t * rotor,
the eager blk's dmr once it has every FEC set, our leader FECs go to
the head of the reasm deque. {turbine,repair,recovered}_cnt
are the FEC set's shred counts from the shred tile, added to the
telemetry of each blk holding the FEC set on its first completion. */
telemetry of each blk holding the FEC set on its first completion.
{data,parity,repair}_rcvd are its received-shred masks from the
shred tile, kept on the FEC set from its first completion. */

fd_rotor_fec_t *
fd_rotor_fec_complete( fd_rotor_t * rotor,
Expand All @@ -348,6 +354,9 @@ fd_rotor_fec_complete( fd_rotor_t * rotor,
uint turbine_cnt,
uint repair_cnt,
uint recovered_cnt,
uint data_rcvd,
uint parity_rcvd,
uint repair_rcvd,
long ts );

/* shred_out SHRED_SIG_FEC_EVICTED
Expand Down
16 changes: 8 additions & 8 deletions src/choreo/rotor/test_fd_rotor.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ complete_as( fd_rotor_t * rotor,
shred->idx = fec_idx*FD_FEC_SHRED_CNT + FD_FEC_SHRED_CNT - 1U;
shred->data.parent_off = (ushort)( slot-parent_slot );
shred->data.flags = last ? FD_SHRED_DATA_FLAG_SLOT_COMPLETE : 0;
return fd_rotor_fec_complete( rotor, shred, mr, is_leader, 0U, 0U, 0U, 0L );
return fd_rotor_fec_complete( rotor, shred, mr, is_leader, 0U, 0U, 0U, 0U, 0U, 0U, 0L );
}

static fd_rotor_fec_t *
Expand Down Expand Up @@ -396,13 +396,13 @@ test_telemetry( void ) {
e1->telemetry.first_req_ts = 6L;

shred->idx = 2U*FD_FEC_SHRED_CNT-1U; shred->data.flags = FD_SHRED_DATA_FLAG_SLOT_COMPLETE;
fd_rotor_fec_t const * f11 = fd_rotor_fec_complete( rotor, shred, &m11, 0, 10U, 2U, 4U, 20L );
FD_TEST( f11 && f11->cmpl_ts==20L );
fd_rotor_fec_t const * f11 = fd_rotor_fec_complete( rotor, shred, &m11, 0, 10U, 2U, 4U, 0x0fffffffU, 0x3U, 0x0000000cU, 20L );
FD_TEST( f11 && f11->cmpl_ts==20L && f11->data_rcvd==0x0fffffffU && f11->parity_rcvd==0x3U && f11->repair_rcvd==0x0000000cU );
FD_TEST( e1->telemetry.last_cmpl_fec_idx==FD_FEC_SHRED_CNT && !e1->telemetry.last_shred_ts ); /* FEC set 0 is missing */
FD_TEST( fd_rotor_fec_complete( rotor, shred, &m11, 0, 99U, 99U, 99U, 25L )->cmpl_ts==20L ); /* again, not counted */
FD_TEST( fd_rotor_fec_complete( rotor, shred, &m11, 0, 99U, 99U, 99U, 99U, 99U, 99U, 25L )->cmpl_ts==20L && f11->data_rcvd==0x0fffffffU ); /* again, not counted */

shred->idx = FD_FEC_SHRED_CNT-1U; shred->data.flags = 0;
FD_TEST( fd_rotor_fec_complete( rotor, shred, &m10, 0, 30U, 1U, 0U, 40L )->cmpl_ts==40L );
FD_TEST( fd_rotor_fec_complete( rotor, shred, &m10, 0, 30U, 1U, 0U, 0U, 0U, 0U, 40L )->cmpl_ts==40L );
FD_TEST( is_eager( rotor, e1 ) && !fd_rotor_blk_map_ele_next_const( e1, NULL, rotor->blk_pool ) ); /* the notar blk was merged */
FD_TEST( e1->telemetry.turbine_shred_cnt==40U && e1->telemetry.repair_shred_cnt==3U && e1->telemetry.recovered_shred_cnt==4U );
FD_TEST( !e1->telemetry.cancelled_reason && !e1->telemetry.cancelled_ts );
Expand Down Expand Up @@ -477,7 +477,7 @@ test_telemetry_final_src( void ) {
FD_TEST( f10->final_src==FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_REPAIR );

shred->idx = FD_FEC_SHRED_CNT-1U;
FD_TEST( fd_rotor_fec_complete( rotor, shred, &m10, 0, 0U, 0U, 0U, 4L )==f10 && f10->final_src==FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_REPAIR );
FD_TEST( fd_rotor_fec_complete( rotor, shred, &m10, 0, 0U, 0U, 0U, 0U, 0U, 0U, 4L )==f10 && f10->final_src==FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_REPAIR );
shred->idx = 4U; fd_rotor_shred_insert( rotor, shred, &m10, FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_TURBINE, 5L );
FD_TEST( f10->final_src==FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_REPAIR ); /* frozen */

Expand All @@ -486,7 +486,7 @@ test_telemetry_final_src( void ) {
FD_TEST( f11 && f11->final_src==FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_TURBINE ); /* a new set takes its first shred's source */

shred->slot = 2UL; shred->idx = FD_FEC_SHRED_CNT-1U; shred->data.flags = FD_SHRED_DATA_FLAG_SLOT_COMPLETE;
FD_TEST( fd_rotor_fec_complete( rotor, shred, &m20, 1, 0U, 0U, 0U, 7L )->final_src==FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_LEADER );
FD_TEST( fd_rotor_fec_complete( rotor, shred, &m20, 1, 0U, 0U, 0U, 0U, 0U, 0U, 7L )->final_src==FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_LEADER );
orphan_cnt( rotor );
}

Expand Down Expand Up @@ -1330,7 +1330,7 @@ test_complete_again( void ) {
shred->idx = FD_FEC_SHRED_CNT - 1U;
shred->data.parent_off = 1;
shred->data.flags = FD_SHRED_DATA_FLAG_SLOT_COMPLETE;
FD_TEST( fd_rotor_fec_complete( rotor, shred, &m10, 1, 0U, 0U, 0U, 0L )->is_leader );
FD_TEST( fd_rotor_fec_complete( rotor, shred, &m10, 1, 0U, 0U, 0U, 0U, 0U, 0U, 0L )->is_leader );
FD_TEST( complete( rotor, 1UL, 0UL, 0U, 1, &m10 )->is_leader );
ulong one = 1UL;
expect( rotor, fd_rotor_blk_map_ele_query( rotor->blk_map, &one, NULL, rotor->blk_pool ), 0U );
Expand Down
8 changes: 4 additions & 4 deletions src/disco/events/schema/events.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4062,11 +4062,11 @@ message MetricsRotorPingRejected {

// Data shreds received as repair responses, by the pending request they matched
message MetricsRotorShredRx {
// First response to a WindowIndexForBlockId request
// Response to a pending WindowIndexForBlockId request
uint64 for_block_id = 1;
// First response to a WindowIndex or HighestWindowIndex request
// Response to a pending WindowIndex or HighestWindowIndex request
uint64 slot_index = 2;
// Matched no pending request, including later answers to an already matched or expired request
// Matched no pending request: a duplicate, or an answer to an expired or evicted request
uint64 unmatched = 3;
}

Expand Down Expand Up @@ -10036,7 +10036,7 @@ message MetricsRotor {
uint64 first_turbine_slot = 29;
// Blocks in the eager, notar and final block treaps, waiting for discovery
uint64 block_pending = 30;
// Pending request attempts awaiting a first response; a hedge is another attempt
// Pending request attempts awaiting a response; a hedge is another attempt
uint64 request_in_flight = 31;
// FEC sets published to replay
uint64 fec_delivered = 32;
Expand Down
8 changes: 4 additions & 4 deletions src/disco/events/schema/metrics_rotor.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,7 @@
"highest_received_slot": { "type": "UInt64", "compression": "T64, ZSTD", "description": "Highest slot a shred was received for, 0 if none" },
"first_turbine_slot": { "type": "UInt64", "compression": "T64, ZSTD", "description": "First turbine slot seen, the catchup target, 0 if none" },
"block_pending": { "type": "UInt64", "compression": "T64, ZSTD", "description": "Blocks in the eager, notar and final block treaps, waiting for discovery" },
"request_in_flight": { "type": "UInt64", "compression": "T64, ZSTD", "description": "Pending request attempts awaiting a first response; a hedge is another attempt" },
"request_in_flight": { "type": "UInt64", "compression": "T64, ZSTD", "description": "Pending request attempts awaiting a response; a hedge is another attempt" },
Comment thread
emwang-jump marked this conversation as resolved.
"fec_delivered": { "type": "UInt64", "compression": "DoubleDelta, ZSTD", "description": "FEC sets published to replay" },
"shred_old_ignored": { "type": "UInt64", "compression": "DoubleDelta, ZSTD", "description": "Shreds received at or below the root, ignored" },
"shred_rx": {
Expand All @@ -116,9 +116,9 @@
"all": ["for_block_id", "slot_index", "unmatched"]
},
"fields": {
"for_block_id": { "type": "UInt64", "compression": "DoubleDelta, ZSTD", "description": "First response to a WindowIndexForBlockId request" },
"slot_index": { "type": "UInt64", "compression": "DoubleDelta, ZSTD", "description": "First response to a WindowIndex or HighestWindowIndex request" },
"unmatched": { "type": "UInt64", "compression": "DoubleDelta, ZSTD", "description": "Matched no pending request, including later answers to an already matched or expired request" }
"for_block_id": { "type": "UInt64", "compression": "DoubleDelta, ZSTD", "description": "Response to a pending WindowIndexForBlockId request" },
"slot_index": { "type": "UInt64", "compression": "DoubleDelta, ZSTD", "description": "Response to a pending WindowIndex or HighestWindowIndex request" },
"unmatched": { "type": "UInt64", "compression": "DoubleDelta, ZSTD", "description": "Matched no pending request: a duplicate, or an answer to an expired or evicted request" }
}
},
"meta_rx": { "type": "UInt64", "compression": "DoubleDelta, ZSTD", "description": "Alpenglow repair metadata responses received" },
Expand Down
2 changes: 1 addition & 1 deletion src/disco/metrics/generated/fd_metrics_rotor.h
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

8 changes: 4 additions & 4 deletions src/disco/metrics/metrics.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1086,9 +1086,9 @@ deliberate exception with lint-allow="<rule id>")
</enum>

<enum name="ShredResponseMatchResult">
<int value="0" name="ForBlockId" label="First response to a WindowIndexForBlockId request" />
<int value="1" name="SlotIndex" label="First response to a WindowIndex or HighestWindowIndex request" />
<int value="2" name="Unmatched" label="Matched no pending request, including later answers to an already matched or expired request" />
<int value="0" name="ForBlockId" label="Response to a pending WindowIndexForBlockId request" />
<int value="1" name="SlotIndex" label="Response to a pending WindowIndex or HighestWindowIndex request" />
<int value="2" name="Unmatched" label="Matched no pending request: a duplicate, or an answer to an expired or evicted request" />
Comment on lines +1089 to +1091
</enum>

<enum name="RepairResponseType">
Expand Down Expand Up @@ -1144,7 +1144,7 @@ deliberate exception with lint-allow="<rule id>")
<gauge name="FirstTurbineSlot" summary="First turbine slot seen, the catchup target, 0 if none" />
<gauge name="CatchupSlot" summary="Highest slot past the window with a verified cert, the staked catchup target, 0 if none" />
<gauge name="BlockPending" summary="Blocks in the eager, notar and final block treaps, waiting for discovery" />
<gauge name="RequestInFlight" summary="Pending request attempts awaiting a first response; a hedge is another attempt" />
<gauge name="RequestInFlight" summary="Pending request attempts awaiting a response; a hedge is another attempt" />
<counter name="FecDelivered" summary="FEC sets published to replay" />
<counter name="ShredOldIgnored" summary="Shreds received at or below the root, ignored" />
<counter name="ShredRx" enum="ShredResponseMatchResult" summary="Data shreds received as repair responses, by the pending request they matched" />
Expand Down
6 changes: 6 additions & 0 deletions src/disco/shred/fd_shred_tile.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1015,10 +1015,16 @@ complete_fec_set( fd_shred_ctx_t * ctx,
complete_msg->turbine_shred_cnt = (ulong)fd_ulong_popcnt( set->turbine_shred_rcvd );
complete_msg->repair_shred_cnt = (ulong)fd_ulong_popcnt( set->repair_shred_rcvd );
complete_msg->reconstructed_shred_cnt = 2U*(uint)FD_FEC_SHRED_CNT-(uint)(fd_uint_popcnt( set->data_shred_rcvd ) + fd_uint_popcnt( set->parity_shred_rcvd ));
complete_msg->data_shred_rcvd = set->data_shred_rcvd;
complete_msg->parity_shred_rcvd = set->parity_shred_rcvd;
complete_msg->repair_data_rcvd = (uint)set->repair_shred_rcvd; /* data shreds are the low FD_FEC_SHRED_CNT bits */
} else {
complete_msg->turbine_shred_cnt = 0U;
complete_msg->repair_shred_cnt = 0U;
complete_msg->reconstructed_shred_cnt = 0U;
complete_msg->data_shred_rcvd = 0U;
complete_msg->parity_shred_rcvd = 0U;
complete_msg->repair_data_rcvd = 0U;
}

fd_stem_publish( stem, ctx->shred_out_idx, sig, ctx->shred_out_chunk, sizeof(fd_fec_complete_t), 0UL, tsorig, fd_frag_meta_ts_comp( fd_tickcount() ) );
Expand Down
7 changes: 7 additions & 0 deletions src/disco/shred/fd_shred_tile.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,13 @@ struct fd_fec_complete {
ulong turbine_shred_cnt;
ulong repair_shred_cnt;
ulong reconstructed_shred_cnt; /* data plus coding */

/* Bit j set if the shred with in-type index j was received rather
than recovered, repair_data_rcvd restricted to data shreds that
arrived by repair. All 0 for LEADER and AGAIN completions. */
uint data_shred_rcvd;
uint parity_shred_rcvd;
uint repair_data_rcvd;
};
typedef struct fd_fec_complete fd_fec_complete_t;

Expand Down
Loading
Loading