From 1d295c7d3aa7f9079faa09fe59f467c87da39519 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 24 Oct 2026 10:05:13 +0200 Subject: [PATCH] Measure multipath recovery stages separately from RTO waits --- tests/Makefile.am | 3 +- tests/test_etcp_multipath.c | 98 ++++++++++++++++++++++++++++++++++--- 2 files changed, 94 insertions(+), 7 deletions(-) diff --git a/tests/Makefile.am b/tests/Makefile.am index 81191429..5524a223 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -1044,7 +1044,8 @@ test_dummynet_channels_LDADD = $(LIBUTUN) $(COMMON_LIBS) test_dummynet_channels_LDFLAGS = $(test_clock_ldflags) test_etcp_multipath_SOURCES = test_etcp_multipath.c $(test_clock_sources) test_etcp_multipath_LDADD = $(LIBUTUN) $(COMMON_LIBS) -test_etcp_multipath_LDFLAGS = $(test_clock_ldflags) -Wl,--wrap=etcp_dump_pkt_sections -Wl,--wrap=bbr_main +test_etcp_multipath_LDFLAGS = $(test_clock_ldflags) -Wl,--wrap=etcp_dump_pkt_sections -Wl,--wrap=bbr_main \ + -Wl,--wrap=etcp_tx_complete -Wl,--wrap=queue_data_put_with_index test_stcp_sigpipe_SOURCES = test_stcp_sigpipe.c test_stcp_sigpipe_LDADD = $(LIBUTUN) $(COMMON_LIBS) test_etcp_windows_LDADD = $(LIBUTUN) $(COMMON_LIBS) diff --git a/tests/test_etcp_multipath.c b/tests/test_etcp_multipath.c index 0c450373..4c2ee4f4 100644 --- a/tests/test_etcp_multipath.c +++ b/tests/test_etcp_multipath.c @@ -21,9 +21,19 @@ #define WARMUP_MS 3000 #define STALL_MS 3000 #define MAX_GAP_US 100000 +#define MAX_RTO_LATE_US 20000 // запас на миллисекундный таймер и планирование тестового процесса struct profile { unsigned rtt_ms, kbps; }; struct message { uint8_t cmd, from, reserved[2]; uint32_t seq; uint64_t enqueued; uint8_t body[1008]; }; -struct trace { uint64_t first_tx, first_rx; uint32_t attempts; uint8_t first_path; }; +struct trace { + uint64_t first_tx, first_rx, last_tx, retry_ready; + uint32_t attempts; + uint16_t last_number; + uint8_t first_path, last_path; +}; +struct recovery_stats { + uint64_t rto_count, retry_count, rto_late_us, ready_wait_us, transit_extra_us; + uint32_t late_seq, ready_seq, transit_seq; +}; struct endpoint { struct UTUN_INSTANCE* inst; struct NODE_CONN_DIRECT* owner; @@ -45,6 +55,8 @@ static uint32_t fault_seq; static unsigned fault_trace_count; 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 FILE* csv; static const char* case_name; static const char* pubkeys[2] = { @@ -81,9 +93,8 @@ void __wrap_bbr_main(struct bbr* model, const struct bbr_rate_sample* sample, ui } } -/* Наблюдение не заменяет передачу/приём и не меняет состояние протокола. */ -void __real_etcp_dump_pkt_sections(struct ETCP_DGRAM*, struct ETCP_LINK*, int); -void __wrap_etcp_dump_pkt_sections(struct ETCP_DGRAM* pkt, struct ETCP_LINK* link, int sending) { +/* Статистика TX снимается только после принятия send; RX — перед разбором уже расшифрованного пакета. */ +static void observe_packet(struct ETCP_DGRAM* pkt, struct ETCP_LINK* link, int sending) { int side = link->etcp->instance == ep[0].inst ? 0 : 1, path = -1; for (unsigned p = 0; p < path_count; p++) if (ss_get_port(&link->remote_addr) == dummynet_get_listen_port(channels[p])) path = p; @@ -103,13 +114,26 @@ void __wrap_etcp_dump_pkt_sections(struct ETCP_DGRAM* pkt, struct ETCP_LINK* lin CHECK(seq < TRACE_COUNT); 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) { + if (measuring && fault_start && !side && trace->retry_ready) { + uint64_t wait = now - trace->retry_ready; recovery.retry_count++; + if (wait > recovery.ready_wait_us) { recovery.ready_wait_us = wait; recovery.ready_seq = seq; } + } if (trace->attempts++ && measuring) traffic[side][path].retries++; if (!trace->first_tx) { trace->first_tx = now; trace->first_path = path; } + trace->last_tx = now; trace->last_number = number; trace->last_path = path; trace->retry_ready = 0; + if (measuring && fault_start && !side && etcp_tx_rto_us(link) > fault_rto_us[path]) + fault_rto_us[path] = etcp_tx_rto_us(link); if (measuring) traffic[side][path].data += remaining - ETCP_PAYLOAD_HEADER_SIZE; } else if (!trace->first_rx) { trace->first_rx = now; if (measuring) traffic[1-side][path].rx_unique += remaining - ETCP_PAYLOAD_HEADER_SIZE; + if (measuring && fault_start && side && trace->last_number == number && trace->last_path == path) { + uint64_t elapsed = now - trace->last_tx, base = profiles[path].rtt_ms * 500ULL; + uint64_t extra = elapsed > base ? elapsed - base : 0; + if (extra > recovery.transit_extra_us) { recovery.transit_extra_us = extra; recovery.transit_seq = seq; } + } } if (fault_start && (seq == fault_seq || (sending && trace->attempts > 1 && fault_trace_count++ < 32))) DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "FAULT trace dir=%d seq=%u path=%d tx=%d attempts=%u elapsed_us=%llu", side, @@ -120,9 +144,45 @@ void __wrap_etcp_dump_pkt_sections(struct ETCP_DGRAM* pkt, struct ETCP_LINK* lin } CHECK(len <= remaining); offset += len; } +} + +void __real_etcp_dump_pkt_sections(struct ETCP_DGRAM*, struct ETCP_LINK*, int); +void __wrap_etcp_dump_pkt_sections(struct ETCP_DGRAM* pkt, struct ETCP_LINK* link, int sending) { + if (!sending) observe_packet(pkt, link, 0); __real_etcp_dump_pkt_sections(pkt, link, sending); } +void __real_etcp_tx_complete(struct ETCP_DGRAM*, int); +void __wrap_etcp_tx_complete(struct ETCP_DGRAM* pkt, int success) { + if (success) observe_packet(pkt, pkt->link, 1); + __real_etcp_tx_complete(pkt, success); +} + +/* Возврат логического DATA после RTO: таймер уже сработал, новая физическая попытка ещё не создана. */ +int __real_queue_data_put_with_index(struct ll_queue*, struct ll_entry*); +int __wrap_queue_data_put_with_index(struct ll_queue* q, struct ll_entry* entry) { + if (measuring && fault_start && ep[0].conn && q == ep[0].conn->input_send_q) { + struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)entry; + if (packet->last_attempt_id && packet->last_link) { + CHECK(packet->seq < TRACE_COUNT && !packet->active_attempts); + struct trace* trace = &traces[0][packet->seq]; + struct ETCP_LINK* link = packet->last_link; + uint64_t now = get_time_us(); trace->retry_ready = now; + CHECK(trace->last_tx && trace->last_number == (uint16_t)packet->last_attempt_id); + if (!link->closing) { + uint32_t rto = etcp_tx_rto_us(link); + CHECK(now - trace->last_tx >= rto); + uint64_t late = now - trace->last_tx - rto; recovery.rto_count++; + if (late > recovery.rto_late_us) { recovery.rto_late_us = late; recovery.late_seq = packet->seq; } + for (unsigned p = 0; p < path_count; p++) + if (ss_get_port(&link->remote_addr) == dummynet_get_listen_port(channels[p]) && rto > fault_rto_us[p]) + fault_rto_us[p] = rto; + } + } + } + return __real_queue_data_put_with_index(q, entry); +} + static void receive_message(struct ETCP_CONN* conn, struct ll_entry* entry) { int side = conn->instance == ep[0].inst ? 0 : 1; struct endpoint* dest = &ep[side]; @@ -166,6 +226,7 @@ static void setup(const struct profile* settings, unsigned count, unsigned mask) memset(traces, 0, sizeof(traces)); memset(traffic, 0, sizeof(traffic)); 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)); 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); @@ -365,6 +426,9 @@ static void stalled(const struct profile* settings, unsigned count, unsigned bro } CHECK(pending == broken && fault_seq); begin_measure(); fault_start = get_time_us(); + for (struct ETCP_LINK* link = ep[0].conn->links; link; link = link->next) + for (unsigned p = 0; p < count; p++) + if (ss_get_port(&link->remote_addr) == dummynet_get_listen_port(channels[p])) fault_rto_us[p] = etcp_tx_rto_us(link); uint32_t epochs[2] = {ep[0].conn->reinit_count, ep[1].conn->reinit_count}; DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "FAULT %s mode=%d paths=%x directions=%x seq=%u duration_ms=%u sparse=%d", case_name, mode, broken, directions, fault_seq, duration_ms, sparse); @@ -386,6 +450,24 @@ static void stalled(const struct profile* settings, unsigned count, unsigned bro for (unsigned p = 0; p < count; p++) unique_now += traffic[0][p].rx_unique; uint64_t fresh_rate = now > steady_start ? (unique_now-steady_unique) * 8000 / (now-steady_start) : 0; if (fresh_rate < rate) rate = fresh_rate; + uint64_t rto_budget = 0; + for (unsigned p = 0; p < count; p++) if (broken & (1u << p)) rto_budget += fault_rto_us[p]; + // Быстрые профили сохраняют 100мс. Для длинных учитываем по одному RTO каждого отказавшего пути. + uint64_t delivery_limit = MAX_GAP_US + (rto_budget > MAX_GAP_US ? rto_budget : 0); + for (struct ll_entry* entry = ep[0].conn->input_send_q->head; entry; entry = entry->next) { + struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)entry; + CHECK(packet->seq < TRACE_COUNT); + uint64_t ready = traces[0][packet->seq].retry_ready; + if (ready && now - ready > recovery.ready_wait_us) { + recovery.ready_wait_us = now - ready; recovery.ready_seq = packet->seq; + } + } + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "FAULT stages %s RTO_count=%llu retry_count=%llu RTO_budget_us=%llu delivery_limit_us=%llu " + "RTO_late_us=%llu late_seq=%u ready_wait_us=%llu ready_seq=%u transit_extra_us=%llu transit_seq=%u", case_name, + (unsigned long long)recovery.rto_count, (unsigned long long)recovery.retry_count, + (unsigned long long)rto_budget, (unsigned long long)delivery_limit, + (unsigned long long)recovery.rto_late_us, recovery.late_seq, (unsigned long long)recovery.ready_wait_us, recovery.ready_seq, + (unsigned long long)recovery.transit_extra_us, recovery.transit_seq); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "FAULT result %s delivered=%llu gap_us=%llu p50=%llu p95=%llu p99=%llu max_latency_us=%llu " "steady_kbps=%llu healthy_capacity=%llu missing_seq=%u first_rx_us=%llu attempts=%u", case_name, (unsigned long long)(ep[1].bytes-before), (unsigned long long)ep[1].max_gap, @@ -395,8 +477,12 @@ static void stalled(const struct profile* settings, unsigned count, unsigned bro (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); if (duration_ms > 100) expect(ep[1].bytes > before, "delivery during active fault", ep[1].bytes-before, 1); - expect(ep[1].max_gap <= MAX_GAP_US, "maximum delivery pause <=100ms", ep[1].max_gap, MAX_GAP_US); - if (sparse) expect(ep[1].max_latency <= MAX_GAP_US, "bounded-source latency <=100ms", ep[1].max_latency, MAX_GAP_US); + 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); + expect(recovery.transit_extra_us <= MAX_GAP_US, "DATA transit beyond base delay <=100ms", recovery.transit_extra_us, MAX_GAP_US); + expect(ep[1].max_gap <= delivery_limit, "delivery pause within profile RTO budget", ep[1].max_gap, delivery_limit); + if (sparse) expect(ep[1].max_latency <= delivery_limit, "bounded-source latency within profile RTO budget", + ep[1].max_latency, delivery_limit); else if (duration_ms > 1000) expect(rate >= capacity * 85 / 100, "fault steady goodput >=85% healthy capacity", rate, capacity * 85 / 100); for (int i = 0; i < 2; i++) CHECK(ep[i].conn->reinit_count == epochs[i] && ep[i].conn->links_up);