Identify the fragment behind maximum multipath message latency

This commit is contained in:
evgeny committed 2026-10-24 10:18:11 +02:00
1 parent 1d295c7d3a
commit 516ed04897
1 file changed
+18 -1
+18 -1
View File
@@ -41,6 +41,8 @@ struct endpoint {
struct ETCP_SOCKET* sockets[PATHS];
unsigned sent, received;
uint64_t bytes, last_delivery, max_gap, max_latency, latency_hist[10001], samples;
uint64_t max_latency_enqueued, max_latency_delivered, max_latency_trigger_tx;
uint32_t max_latency_msg, max_latency_fragment, max_latency_trigger, max_latency_attempts;
};
struct path_stats { uint64_t data, retries, rx_unique; };
static struct UASYNC* ua;
@@ -57,6 +59,7 @@ static int log_bbr_samples;
static uint64_t sample_logged[2][PATHS];
static struct recovery_stats recovery;
static uint32_t fault_rto_us[PATHS]; // максимум RTO каждого пути с начала отказа, независимо от измеренной паузы
static uint32_t rx_trigger[2]; // DATA, разбор которого сейчас разблокировал выдачу сообщения
static FILE* csv;
static const char* case_name;
static const char* pubkeys[2] = {
@@ -115,6 +118,7 @@ static void observe_packet(struct ETCP_DGRAM* pkt, struct ETCP_LINK* link, int s
struct trace* trace = &traces[sending ? side : 1-side][seq];
uint64_t now = get_time_us();
uint16_t number = (uint16_t)data[5] | ((uint16_t)data[6] << 8);
if (!sending) rx_trigger[side] = seq;
if (sending) {
if (measuring && fault_start && !side && trace->retry_ready) {
uint64_t wait = now - trace->retry_ready; recovery.retry_count++;
@@ -195,7 +199,13 @@ static void receive_message(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (measuring) {
uint64_t gap = now - dest->last_delivery, latency = now - msg.enqueued;
if (gap > dest->max_gap) dest->max_gap = gap;
if (latency > dest->max_latency) dest->max_latency = latency;
if (latency > dest->max_latency) {
dest->max_latency = latency; dest->max_latency_msg = msg.seq;
dest->max_latency_fragment = conn->last_delivered_id + 1; dest->max_latency_trigger = rx_trigger[side];
dest->max_latency_enqueued = msg.enqueued; dest->max_latency_delivered = now;
struct trace* trigger = &traces[1-side][rx_trigger[side]];
dest->max_latency_trigger_tx = trigger->first_tx; dest->max_latency_attempts = trigger->attempts;
}
dest->latency_hist[latency / 1000 < 10000 ? latency / 1000 : 10000]++; dest->samples++;
}
dest->last_delivery = now;
@@ -227,6 +237,7 @@ static void setup(const struct profile* settings, unsigned count, unsigned mask)
memset(next_message, 0, sizeof(next_message));
memset(sample_logged, 0, sizeof(sample_logged));
memset(&recovery, 0, sizeof(recovery)); memset(fault_rto_us, 0, sizeof(fault_rto_us));
memset(rx_trigger, 0, sizeof(rx_trigger));
path_count = count; enabled_mask = mask; measuring = 0; fault_start = 0; fault_trace_count = 0;
memcpy(profiles, settings, count * sizeof(*settings));
ua = uasync_create(); CHECK(ua);
@@ -476,6 +487,12 @@ static void stalled(const struct profile* settings, unsigned count, unsigned bro
(unsigned long long)rate, (unsigned long long)capacity, fault_seq,
(unsigned long long)(traces[0][fault_seq].first_rx > fault_start ? traces[0][fault_seq].first_rx-fault_start : 0),
traces[0][fault_seq].attempts);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "FAULT worst %s message=%u fragment=%u trigger=%u trigger_attempts=%u "
"enqueued_us=%lld delivered_us=%lld trigger_first_tx_us=%lld latency_us=%llu", case_name,
ep[1].max_latency_msg, ep[1].max_latency_fragment, ep[1].max_latency_trigger, ep[1].max_latency_attempts,
(long long)((int64_t)ep[1].max_latency_enqueued - (int64_t)fault_start),
(long long)((int64_t)ep[1].max_latency_delivered - (int64_t)fault_start),
(long long)((int64_t)ep[1].max_latency_trigger_tx - (int64_t)fault_start), (unsigned long long)ep[1].max_latency);
if (duration_ms > 100) expect(ep[1].bytes > before, "delivery during active fault", ep[1].bytes-before, 1);
expect(recovery.rto_late_us <= MAX_RTO_LATE_US, "RTO processing overhead <=20ms", recovery.rto_late_us, MAX_RTO_LATE_US);
expect(recovery.ready_wait_us <= MAX_GAP_US, "ready retry waits <=100ms", recovery.ready_wait_us, MAX_GAP_US);