From fc21c34edc68e4600c27d75770bdd88ecd3f26a0 Mon Sep 17 00:00:00 2001 From: Emily Wang Date: Wed, 21 Oct 2026 18:50:53 +0000 Subject: [PATCH] rotor: report fec masks & sample rtt off shred tile arrival time --- book/api/metrics-generated.md | 8 ++-- src/choreo/rotor/fd_rotor.c | 14 +++++- src/choreo/rotor/fd_rotor.h | 23 ++++++--- src/choreo/rotor/test_fd_rotor.c | 16 +++---- src/disco/events/schema/events.proto | 8 ++-- src/disco/events/schema/metrics_rotor.json | 8 ++-- .../metrics/generated/fd_metrics_rotor.h | 2 +- src/disco/metrics/metrics.xml | 8 ++-- src/disco/shred/fd_shred_tile.c | 6 +++ src/disco/shred/fd_shred_tile.h | 7 +++ src/discof/rotor/fd_rotor_tile.c | 47 +++++++++++++------ 11 files changed, 100 insertions(+), 47 deletions(-) diff --git a/book/api/metrics-generated.md b/book/api/metrics-generated.md index 93be2347403..b253396f5c2 100644 --- a/book/api/metrics-generated.md +++ b/book/api/metrics-generated.md @@ -2200,12 +2200,12 @@ | rotor_​first_​turbine_​slot | gauge | First turbine slot seen, the catchup target, 0 if none | | rotor_​catchup_​slot | gauge | Highest slot past the window with a verified cert, the staked catchup target, 0 if none | | rotor_​block_​pending | gauge | Blocks in the eager, notar and final block treaps, waiting for discovery | -| rotor_​request_​in_​flight | gauge | Pending request attempts awaiting a first response; a hedge is another attempt | +| rotor_​request_​in_​flight | gauge | Pending request attempts awaiting a response; a hedge is another attempt | | rotor_​fec_​delivered | counter | FEC sets published to replay | | rotor_​shred_​old_​ignored | counter | Shreds received at or below the root, ignored | -| rotor_​shred_​rx
{shred_​response_​match_​result="for_​block_​id"} | counter | Data shreds received as repair responses, by the pending request they matched (First response to a WindowIndexForBlockId request) | -| rotor_​shred_​rx
{shred_​response_​match_​result="slot_​index"} | counter | Data shreds received as repair responses, by the pending request they matched (First response to a WindowIndex or HighestWindowIndex request) | -| rotor_​shred_​rx
{shred_​response_​match_​result="unmatched"} | counter | Data shreds received as repair responses, by the pending request they matched (Matched no pending request, including later answers to an already matched or expired request) | +| rotor_​shred_​rx
{shred_​response_​match_​result="for_​block_​id"} | counter | Data shreds received as repair responses, by the pending request they matched (Response to a pending WindowIndexForBlockId request) | +| rotor_​shred_​rx
{shred_​response_​match_​result="slot_​index"} | counter | Data shreds received as repair responses, by the pending request they matched (Response to a pending WindowIndex or HighestWindowIndex request) | +| rotor_​shred_​rx
{shred_​response_​match_​result="unmatched"} | counter | Data shreds received as repair responses, by the pending request they matched (Matched no pending request: a duplicate, or an answer to an expired or evicted request) | | rotor_​meta_​rx | counter | Alpenglow repair metadata responses received | | rotor_​meta_​malformed_​rejected | counter | Metadata responses that failed to decode | | rotor_​meta_​unsolicited_​rejected | counter | Metadata responses matching no outstanding request | diff --git a/src/choreo/rotor/fd_rotor.c b/src/choreo/rotor/fd_rotor.c index 35546a073ad..76b408510c0 100644 --- a/src/choreo/rotor/fd_rotor.c +++ b/src/choreo/rotor/fd_rotor.c @@ -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 */ @@ -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 ); @@ -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; @@ -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 ); diff --git a/src/choreo/rotor/fd_rotor.h b/src/choreo/rotor/fd_rotor.h index 6e3a0060b4a..50543ff5d00 100644 --- a/src/choreo/rotor/fd_rotor.h +++ b/src/choreo/rotor/fd_rotor.h @@ -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 ); @@ -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, @@ -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 diff --git a/src/choreo/rotor/test_fd_rotor.c b/src/choreo/rotor/test_fd_rotor.c index 872e77c6f2a..be180f9ea36 100644 --- a/src/choreo/rotor/test_fd_rotor.c +++ b/src/choreo/rotor/test_fd_rotor.c @@ -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 * @@ -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 ); @@ -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 */ @@ -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 ); } @@ -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 ); diff --git a/src/disco/events/schema/events.proto b/src/disco/events/schema/events.proto index 0beca63c70a..ebed8ba9148 100644 --- a/src/disco/events/schema/events.proto +++ b/src/disco/events/schema/events.proto @@ -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; } @@ -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; diff --git a/src/disco/events/schema/metrics_rotor.json b/src/disco/events/schema/metrics_rotor.json index 6851f6cdd67..0ac57b971a9 100644 --- a/src/disco/events/schema/metrics_rotor.json +++ b/src/disco/events/schema/metrics_rotor.json @@ -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" }, "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": { @@ -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" }, diff --git a/src/disco/metrics/generated/fd_metrics_rotor.h b/src/disco/metrics/generated/fd_metrics_rotor.h index 8d9bdd66535..9dc7f573392 100644 --- a/src/disco/metrics/generated/fd_metrics_rotor.h +++ b/src/disco/metrics/generated/fd_metrics_rotor.h @@ -96,7 +96,7 @@ enum { #define FD_METRICS_GAUGE_ROTOR_REQUEST_IN_FLIGHT_NAME "rotor_request_in_flight" #define FD_METRICS_GAUGE_ROTOR_REQUEST_IN_FLIGHT_TYPE (FD_METRICS_TYPE_GAUGE) -#define FD_METRICS_GAUGE_ROTOR_REQUEST_IN_FLIGHT_DESC "Pending request attempts awaiting a first response; a hedge is another attempt" +#define FD_METRICS_GAUGE_ROTOR_REQUEST_IN_FLIGHT_DESC "Pending request attempts awaiting a response; a hedge is another attempt" #define FD_METRICS_GAUGE_ROTOR_REQUEST_IN_FLIGHT_CVT (FD_METRICS_CONVERTER_NONE) #define FD_METRICS_COUNTER_ROTOR_FEC_DELIVERED_NAME "rotor_fec_delivered" diff --git a/src/disco/metrics/metrics.xml b/src/disco/metrics/metrics.xml index e483b20f340..3b78ec0c33f 100644 --- a/src/disco/metrics/metrics.xml +++ b/src/disco/metrics/metrics.xml @@ -1086,9 +1086,9 @@ deliberate exception with lint-allow="") - - - + + + @@ -1144,7 +1144,7 @@ deliberate exception with lint-allow="") - + diff --git a/src/disco/shred/fd_shred_tile.c b/src/disco/shred/fd_shred_tile.c index 6b0f0ac9158..b21c80d50ae 100644 --- a/src/disco/shred/fd_shred_tile.c +++ b/src/disco/shred/fd_shred_tile.c @@ -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() ) ); diff --git a/src/disco/shred/fd_shred_tile.h b/src/disco/shred/fd_shred_tile.h index dac90f39bb1..babb4c7c669 100644 --- a/src/disco/shred/fd_shred_tile.h +++ b/src/disco/shred/fd_shred_tile.h @@ -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; diff --git a/src/discof/rotor/fd_rotor_tile.c b/src/discof/rotor/fd_rotor_tile.c index ab609668d09..5323c00f82c 100644 --- a/src/discof/rotor/fd_rotor_tile.c +++ b/src/discof/rotor/fd_rotor_tile.c @@ -76,6 +76,8 @@ struct pending { uint fec_idx; /* FEC index */ uint fec_set_cnt; /* the blk's when sent */ uint tag; /* FD_ROTOR_SERDE_TAG_* of the request */ + uint mask; /* bit i set while shred i of the FEC set is unanswered, bit 0 for a HIGHEST or ORPHAN; the pending outlives its last answer */ + uint answered; /* shreds matched so far */ }; typedef struct pending pending_t; @@ -452,6 +454,9 @@ report_block_received( void * ctx_, if( FD_UNLIKELY( !fec ) ) continue; /* an incomplete blk can miss FEC sets */ memcpy( f->merkle_root, fec->mr32.uc, sizeof(fd_mr32_t) ); f->index = (uint)k*FD_FEC_SHRED_CNT; + f->data_shreds_received_mask = fec->data_rcvd; + f->parity_shreds_received_mask = fec->parity_rcvd; + f->repair_shreds_received_mask = fec->repair_rcvd; f->first_shred_received_time = (ulong)fec->first_ts; f->completed_time = (ulong)fec->cmpl_ts; f->final_shred_source = fd_int_if( fec->is_leader, FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_LEADER, fec->final_src ); @@ -495,7 +500,7 @@ handle_shred( fd_rotor_tile_t * ctx, if( FD_UNLIKELY( shred->idx>=ctx->rotor->fec_blk_max*FD_FEC_SHRED_CNT ) ) return; if( FD_UNLIKELY( !shred->data.parent_off || shred->data.parent_off>shred->slot-ctx->rotor->root ) ) return; /* parent is the slot itself, or below the root */ - fd_rotor_fec_t const * fec = fd_rotor_fec_complete( ctx->rotor, shred, &complete->merkle_root, sig==SHRED_SIG_FEC_COMPLETE_LEADER, (uint)complete->turbine_shred_cnt, (uint)complete->repair_shred_cnt, (uint)complete->reconstructed_shred_cnt, rx_ts ); + fd_rotor_fec_t const * fec = fd_rotor_fec_complete( ctx->rotor, shred, &complete->merkle_root, sig==SHRED_SIG_FEC_COMPLETE_LEADER, (uint)complete->turbine_shred_cnt, (uint)complete->repair_shred_cnt, (uint)complete->reconstructed_shred_cnt, complete->data_shred_rcvd, complete->parity_shred_rcvd, complete->repair_data_rcvd, rx_ts ); if( FD_UNLIKELY( sig!=SHRED_SIG_FEC_COMPLETE || !fec || !fec->first_ts ) ) break; /* a first completion from the network */ fd_rotor_strat_turbine_done( ctx->strat, fd_multi_epoch_leaders_get_leader_for_slot( ctx->mleaders, shred->slot ), rx_ts-fec->first_ts ); @@ -519,17 +524,25 @@ handle_shred( fd_rotor_tile_t * ctx, if( FD_UNLIKELY( shred->idx>=ctx->rotor->fec_blk_max*FD_FEC_SHRED_CNT ) ) return; /* A repair shred answers the request whose nonce it carries: the - first one ends that request's pick. A HighestWindowIndex answer - that does not complete the slot arrives as BAD_REPAIR. */ + first one ends that request's pick, the last one ends the + request. A HighestWindowIndex answer that does not complete the + slot arrives as BAD_REPAIR. */ int normal = fd_rnonce_ss_normal_repair( msg->rnonce ); int reply = fd_shred_sig_src( sig )==SHRED_SIG_SRC_REPAIR || fd_shred_sig_src( sig )==SHRED_SIG_SRC_BAD_REPAIR; ulong key = (ulong)(uint)slot<<32 | fd_uint_if( normal, shred->idx/FD_FEC_SHRED_CNT, (uint)ctx->rotor->fec_blk_max ); + uint bit = fd_uint_if( normal, 1U<<( shred->idx%FD_FEC_SHRED_CNT ), 1U ); pending_t * pending = reply ? pending_map_ele_query( ctx->pending_map, &key, NULL, ctx->pending_pool ) : NULL; for( ; pending; pending = (pending_t *)pending_map_ele_next_const( pending, NULL, ctx->pending_pool ) ) { + if( FD_UNLIKELY( !( pending->mask & bit ) ) ) continue; /* already answered, or not asked of this request */ if( FD_LIKELY( fd_rnonce_ss_compute( ctx->rnonce_ss, normal, slot, shred->idx, pending->ts )!=msg->rnonce ) ) continue; - fd_rotor_strat_request_done( ctx->strat, &pending->peer, now-pending->ts, pending->tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID ); - fd_histf_sample( ctx->metrics->response_latency, (ulong)( now-pending->ts ) ); + if( FD_LIKELY( !pending->answered ) ) { /* rtt to the shred tile's network rx, not to this tile's turn at the frag */ + long rtt = fd_long_max( rx_ts-pending->ts, 1L ); + fd_rotor_strat_request_done( ctx->strat, &pending->peer, rtt, pending->tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID ); + fd_histf_sample( ctx->metrics->response_latency, (ulong)rtt ); + } + pending->mask &= ~bit; + pending->answered++; ctx->metrics->shred_rx[ fd_ulong_if( pending->tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID, FD_METRICS_ENUM_SHRED_RESPONSE_MATCH_RESULT_V_FOR_BLOCK_ID_IDX, FD_METRICS_ENUM_SHRED_RESPONSE_MATCH_RESULT_V_SLOT_INDEX_IDX ) ]++; fd_rotor_blk_t * blk = fd_rotor_blk_map_ele_query( ctx->rotor->blk_map, &pending->slot, NULL, ctx->rotor->blk_pool ); while( blk && memcmp( &blk->dmr, &pending->dmr, sizeof(fd_mr32_t) ) ) blk = (fd_rotor_blk_t *)fd_rotor_blk_map_ele_next_const( blk, NULL, ctx->rotor->blk_pool ); /* not by pending->blk, it may be freed or reused, and a notar blk lives on as eager after dedup */ @@ -537,9 +550,11 @@ handle_shred( fd_rotor_tile_t * ctx, blk->telemetry.shred_res_cnt++; blk->telemetry.last_shred_res_ts = rx_ts; } - pending_map_ele_remove_fast( ctx->pending_map, pending, ctx->pending_pool ); - pending_dlist_ele_remove ( ctx->pending_dlist, pending, ctx->pending_pool ); - pending_pool_ele_release ( ctx->pending_pool, pending ); + if( FD_LIKELY( !pending->mask ) ) { /* every shred answered */ + pending_map_ele_remove_fast( ctx->pending_map, pending, ctx->pending_pool ); + pending_dlist_ele_remove ( ctx->pending_dlist, pending, ctx->pending_pool ); + pending_pool_ele_release ( ctx->pending_pool, pending ); + } break; } ctx->metrics->shred_rx[ FD_METRICS_ENUM_SHRED_RESPONSE_MATCH_RESULT_V_UNMATCHED_IDX ] += (ulong)( reply && !pending ); @@ -1015,14 +1030,16 @@ after_credit( fd_rotor_tile_t * ctx, /* No rtt sample: silence also means a deduped reply, or a highest or orphan request no peer can answer yet. Only a run of shred - and meta requests left unanswered bans the peer. */ + and meta requests left unanswered bans the peer. A request + answered in part settled its pick on the first shred. */ int answerable = pending->slot>ctx->rotor->root && pending->tag!=FD_ROTOR_SERDE_TAG_HIGHEST_WINDOW_INDEX && pending->tag!=FD_ROTOR_SERDE_TAG_ORPHAN; int byid = pending->tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID; - ctx->metrics->byid_expired += (ulong)( answerable && byid ); - if( FD_LIKELY( answerable ) ) ctx->metrics->peer_byid_banned += (ulong)( fd_rotor_strat_request_expired( ctx->strat, &pending->peer, now+BAN_TTL, byid )==1 ); - else fd_rotor_strat_request_failed ( ctx->strat, &pending->peer, 0 ); - ctx->metrics->req_expired++; + ctx->metrics->byid_expired += (ulong)( answerable && byid && !pending->answered ); + if( FD_UNLIKELY( pending->answered ) ) {} + else if( FD_LIKELY ( answerable ) ) ctx->metrics->peer_byid_banned += (ulong)( fd_rotor_strat_request_expired( ctx->strat, &pending->peer, now+BAN_TTL, byid )==1 ); + else fd_rotor_strat_request_failed ( ctx->strat, &pending->peer, 0 ); + ctx->metrics->req_expired+= (ulong)!pending->answered; pending_dlist_ele_pop_head ( ctx->pending_dlist, ctx->pending_pool ); pending_map_ele_remove_fast( ctx->pending_map, pending, ctx->pending_pool ); pending_pool_ele_release ( ctx->pending_pool, pending ); @@ -1113,7 +1130,7 @@ after_credit( fd_rotor_tile_t * ctx, if( FD_UNLIKELY( !pending_pool_free( ctx->pending_pool ) ) ) { /* evict the oldest, a reply to it no longer matches */ pending_t * oldest = pending_dlist_ele_pop_head( ctx->pending_dlist, ctx->pending_pool ); - fd_rotor_strat_request_failed( ctx->strat, &oldest->peer, 0 ); + if( FD_LIKELY( !oldest->answered ) ) fd_rotor_strat_request_failed( ctx->strat, &oldest->peer, 0 ); pending_map_ele_remove_fast( ctx->pending_map, oldest, ctx->pending_pool ); pending_pool_ele_release ( ctx->pending_pool, oldest ); } @@ -1127,6 +1144,8 @@ after_credit( fd_rotor_tile_t * ctx, pending->fec_idx = req.fec_idx; pending->fec_set_cnt = blk->cmpl_fec_cnt; pending->tag = tag; + pending->mask = mask; + pending->answered = 0U; pending_map_ele_insert ( ctx->pending_map, pending, ctx->pending_pool ); pending_dlist_ele_push_tail( ctx->pending_dlist, pending, ctx->pending_pool );