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 );