From 6e350760d1554c8c7893114de63ed9c2e8ec7fbe Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 23:29:29 +0200 Subject: [PATCH] Check active channel stalls, delivery deadlines and delayed packet recovery --- tests/test_etcp_multipath.c | 98 +++++++++++++++++++++++++++++++++++-- 1 file changed, 93 insertions(+), 5 deletions(-) diff --git a/tests/test_etcp_multipath.c b/tests/test_etcp_multipath.c index 02c9cea3..5f207952 100644 --- a/tests/test_etcp_multipath.c +++ b/tests/test_etcp_multipath.c @@ -41,6 +41,7 @@ static struct trace traces[2][TRACE_COUNT]; static struct path_stats traffic[2][PATHS]; static unsigned path_count, failures, enabled_mask, bidirectional, low_load, measuring; static uint64_t started, next_message[2], measure_start, fault_start; +static uint32_t fault_seq; static FILE* csv; static const char* case_name; static const char* pubkeys[2] = { @@ -81,8 +82,8 @@ void __wrap_etcp_dump_pkt_sections(struct ETCP_DGRAM* pkt, struct ETCP_LINK* lin trace->first_rx = now; if (measuring) traffic[1-side][path].rx_unique += remaining - 5; } - if (fault_start && seq < 100 && trace->attempts > 1) - DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "trace dir=%d seq=%u path=%d tx=%d attempts=%u elapsed_us=%llu", side, + if (fault_start && seq == fault_seq) + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "FAULT trace dir=%d seq=%u path=%d tx=%d attempts=%u elapsed_us=%llu", side, seq, path, sending, trace->attempts, (unsigned long long)(now - fault_start)); len = remaining; break; } @@ -119,7 +120,8 @@ static void create_endpoint(unsigned side) { strcat(listeners, text); } snprintf(config, sizeof(config), "[global]\nmy_node_name=multipath-%u\nmy_public_key=%s\nmy_private_key=%s\n" - "keepalive_interval=100\nkeepalive_timeout=2000\nkeepalive_adaptive=0\nauto_sockets=no\n%s", + "keepalive_interval=100\nkeepalive_timeout=2000\nkeepalive_adaptive=0\nauto_sockets=no\n" + "[allowed_keys]\nallow_all=1\n%s", side, pubkeys[side], privkeys[side], listeners); ep[side].inst = utun_instance_create_from_str(ua, config); CHECK(ep[side].inst); CHECK(utun_core_start(ep[side].inst) == 0); @@ -161,8 +163,6 @@ static void setup(const struct profile* settings, unsigned count, unsigned mask) struct ETCP_CONN* c = ep[i].conn; if (!c || !c->initialized || !c->links_up || c->session_required || c->reinit_pending) { ready = 0; continue; } for (unsigned p = 0; p < count; p++) if (mask & (1u << p)) { - struct ETCP_LINK* link = etcp_link_find_by_addr(ep[i].sockets[p], &c->links->remote_addr, 0); - (void)link; int found = 0; for (struct ETCP_LINK* l = c->links; l; l = l->next) if (ss_get_port(&l->remote_addr) == dummynet_get_listen_port(channels[p]) && l->initialized && l->link_status) found++; @@ -242,6 +242,16 @@ static void expect(int passed, const char* requirement, uint64_t actual, uint64_ (unsigned long long)actual, (unsigned long long)limit); } +static uint64_t percentile(const struct endpoint* endpoint, unsigned percent) { + uint64_t sum = 0, target = (endpoint->samples * percent + 99) / 100; + if (!target) return 0; + for (unsigned ms = 0; ms <= 10000; ms++) { + sum += endpoint->latency_hist[ms]; + if (sum >= target) return ms; + } + return 10000; +} + static void teardown(void) { measuring = 0; // Уничтожаем и активные delay/HOLD пакеты: эмуляторы не должны пережить свои endpoints. @@ -281,6 +291,69 @@ static uint64_t measure_rate(const struct profile* settings, unsigned count, uns teardown(); return rate; } +/* Не ждём восстановления отказавшего пути: проверяем паузу и скорость пока DROP/HOLD ещё включён. */ +static void stalled(const struct profile* settings, unsigned count, unsigned broken, unsigned directions, + enum dummynet_mode mode, int sparse, unsigned duration_ms) { + low_load = sparse; + setup(settings, count, (1u << count)-1); + run_for(WARMUP_MS, 1, "warmup"); + // Выбираем момент, когда на каждом останавливаемом пути есть действительно неподтверждённый пакет. + uint64_t deadline = get_time_us() + 2000000; + unsigned pending = 0; + fault_seq = 0; + while (get_time_us() < deadline) { + pending = 0; + for (struct ll_entry* entry = ep[0].conn->input_wait_ack->head; entry; entry = entry->next) { + struct INFLIGHT_PACKET* pkt = (struct INFLIGHT_PACKET*)entry; + if (!pkt->last_link || ((directions & 1) && traces[0][pkt->seq].first_rx)) continue; + for (unsigned p = 0; p < count; p++) if ((broken & (1u << p)) && + ss_get_port(&pkt->last_link->remote_addr) == dummynet_get_listen_port(channels[p])) { + pending |= 1u << p; + if (!fault_seq || pkt->seq < fault_seq) fault_seq = pkt->seq; + } + } + if (pending == broken) break; + run_for(1, 1, "arm-fault"); + } + CHECK(pending == broken && fault_seq); + begin_measure(); fault_start = get_time_us(); + 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); + for (unsigned p = 0; p < count; p++) if (broken & (1u << p)) + for (unsigned dir = 0; dir < 2; dir++) if (directions & (1u << dir)) + CHECK(dummynet_set_mode(channels[p], dir, mode) == 0); + uint64_t before = ep[1].bytes; + run_for(duration_ms > 1000 ? 1000 : duration_ms, 1, "fault-transition"); + uint64_t steady_bytes = ep[1].bytes, steady_start = get_time_us(); + if (duration_ms > 1000) run_for(duration_ms - 1000, 1, "fault-steady"); + uint64_t now = get_time_us(), tail_gap = now - ep[1].last_delivery; + if (tail_gap > ep[1].max_gap) ep[1].max_gap = tail_gap; + uint64_t capacity = 0; + for (unsigned p = 0; p < count; p++) if (!(broken & (1u << p))) capacity += settings[p].kbps; + uint64_t rate = now > steady_start ? (ep[1].bytes-steady_bytes) * 8000 / (now-steady_start) : 0; + 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, + (unsigned long long)percentile(&ep[1], 50), (unsigned long long)percentile(&ep[1], 95), + (unsigned long long)percentile(&ep[1], 99), (unsigned long long)ep[1].max_latency, + (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); + 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); + 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); + measuring = 0; + for (unsigned p = 0; p < count; p++) if (broken & (1u << p)) + for (unsigned dir = 0; dir < 2; dir++) CHECK(dummynet_set_mode(channels[p], dir, DUMMYNET_RUN) == 0); + run_for(3000, 1, "recovery"); run_for(5000, 0, "drain"); + for (int i = 0; i < 2; i++) CHECK(ep[i].sent == ep[1-i].received); + teardown(); +} + int main(int argc, char** argv) { debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); test_clock_init(10); if (getenv("UTUN_TEST_DEBUG")) debug_set_category_level(DEBUG_CATEGORY_ETCP, DEBUG_LEVEL_DEBUG); @@ -303,6 +376,21 @@ int main(int argc, char** argv) { expect(aggregate >= solo_sum * 90 / 100, "aggregation >=90% sum of solo rates", aggregate, solo_sum * 90 / 100); CHECK(u_get_allocated_count() == baseline); } + struct { const char* name; unsigned count, broken, directions, sparse, ms; enum dummynet_mode mode; } faults[] = { + {"drop",2,2,3,1,3000,DUMMYNET_DROP}, {"hold",2,2,3,1,3000,DUMMYNET_HOLD}, + {"drop-forward",2,2,1,1,3000,DUMMYNET_DROP}, {"drop-reverse",2,2,2,1,3000,DUMMYNET_DROP}, + {"short-hold",2,2,3,1,30,DUMMYNET_HOLD}, {"slow-drop",2,2,3,1,3000,DUMMYNET_DROP}, + {"two-stuck",3,6,3,1,3000,DUMMYNET_DROP}, {"load-drop",2,2,3,0,5000,DUMMYNET_DROP}, + {"load-hold",2,2,3,0,5000,DUMMYNET_HOLD} + }; + for (unsigned n = 0; n < sizeof(faults)/sizeof(faults[0]); n++) { + if (argc > 1 && strcmp(argv[1], faults[n].name)) continue; + case_name = faults[n].name; + struct profile settings[] = {{20,8000},{20,8000},{20,2000}}; + if (!strcmp(case_name, "slow-drop")) settings[1].rtt_ms = 200; + stalled(settings, faults[n].count, faults[n].broken, faults[n].directions, faults[n].mode, faults[n].sparse, faults[n].ms); + CHECK(u_get_allocated_count() == baseline); + } if (csv) CHECK(fclose(csv) == 0); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Multipath completed: failures=%u", failures); test_clock_finish(); socket_platform_cleanup(); return failures ? 1 : 0;