From 076016f34a94ed35a64c6ba4e024085188410d65 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 23:34:20 +0200 Subject: [PATCH] Expose multipath queue accounting and add a dedicated long-running check target --- tests/Makefile.am | 12 +++++++++++- tests/test_etcp_multipath.c | 27 +++++++++++++++++++++------ 2 files changed, 32 insertions(+), 7 deletions(-) diff --git a/tests/Makefile.am b/tests/Makefile.am index 59ffddf3..1634cd9f 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -101,7 +101,6 @@ check_PROGRAMS += \ test_ntp_regressions \ test_etcp_windows \ test_dummynet_channels \ - test_etcp_multipath \ test_stcp_sigpipe \ test_firewall \ test_mem \ @@ -213,6 +212,9 @@ endif # Долгие тесты: запускаются только вручную noinst_PROGRAMS = +if OS_LINUX +noinst_PROGRAMS += test_etcp_multipath +endif EXTRA_PROGRAMS = test_merkle_sync_stress CLEANFILES = test_merkle_sync_stress$(EXEEXT) @@ -999,6 +1001,14 @@ check-burst: check-load: @cd $(srcdir)/tcp_proxy_full && bash run_load_test.sh +# Полная RTT/BWlim/fault-матрица дольше 90s общего runner; отдельный запуск не скрывает её результат. +if OS_LINUX +.PHONY: check-multipath +check-multipath: test_dummynet_channels test_etcp_multipath + ./test_dummynet_channels + ./test_etcp_multipath +endif + test_etcp_lifecycle_SOURCES = test_etcp_lifecycle.c test_etcp_lifecycle_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) diff --git a/tests/test_etcp_multipath.c b/tests/test_etcp_multipath.c index d6807d56..504565f4 100644 --- a/tests/test_etcp_multipath.c +++ b/tests/test_etcp_multipath.c @@ -42,6 +42,7 @@ 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 unsigned fault_trace_count; static FILE* csv; static const char* case_name; static const char* pubkeys[2] = { @@ -82,7 +83,7 @@ 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 == fault_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, seq, path, sending, trace->attempts, (unsigned long long)(now - fault_start)); len = remaining; break; @@ -136,7 +137,7 @@ static void setup(const struct profile* settings, unsigned count, unsigned mask) memset(ep, 0, sizeof(ep)); memset(channels, 0, sizeof(channels)); memset(traces, 0, sizeof(traces)); memset(traffic, 0, sizeof(traffic)); memset(next_message, 0, sizeof(next_message)); - path_count = count; enabled_mask = mask; measuring = 0; fault_start = 0; + 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); create_endpoint(0); create_endpoint(1); @@ -200,18 +201,27 @@ static void snapshot(const char* phase) { for (int side = 0; side < 2; side++) { struct ETCP_CONN* conn = ep[side].conn; for (struct ETCP_LINK* l = conn->links; l; l = l->next) { + uint64_t occupied = 0; unsigned packets = 0; + for (struct ll_entry* e = conn->input_wait_ack->head; e; e = e->next) { + struct INFLIGHT_PACKET* pkt = (struct INFLIGHT_PACKET*)e; + CHECK(pkt->state == INFLIGHT_STATE_WAIT_ACK); + if (pkt->last_link == l) { occupied += pkt->ll.len; packets++; } + } + CHECK(l->inflight_bytes == occupied && l->inflight_packets == packets); int p = -1; for (unsigned j = 0; j < path_count; j++) if (ss_get_port(&l->remote_addr) == dummynet_get_listen_port(channels[j])) p = j; if (p < 0) continue; const struct dummynet_stats* ds = dummynet_get_stats(channels[p], side); - if (csv) fprintf(csv, "%s,%s,%llu,%d,%d,%u,%u,%u,%u,%u,%u,%d,%d,%llu,%llu,%llu,%u,%llu,%llu\n", - case_name, phase, (unsigned long long)((now-started)/1000), side, p, l->rtt_last, l->bandwidth, + if (csv) fprintf(csv, "%s,%x,%s,%llu,%d,%d,%u,%u,%u,%u,%u,%u,%d,%d,%llu,%llu,%llu,%u,%llu,%llu,%d,%u,%llu\n", + case_name, enabled_mask, phase, (unsigned long long)((now-started)/1000), side, p, l->rtt_last, l->bandwidth, l->inflight_bytes, l->inflight_lim_bytes, conn->rtt_avg_10, conn->jitter, conn->input_send_q->count, conn->recv_q->count, (unsigned long long)ds->sent_bytes, (unsigned long long)traffic[side][p].rx_unique, (unsigned long long)traffic[side][p].retries, ds->queue_size, - (unsigned long long)ep[1-side].bytes, (unsigned long long)ep[1-side].max_gap); + (unsigned long long)ep[1-side].bytes, (unsigned long long)ep[1-side].max_gap, + ep[1-side].conn->recv_q->count, ep[1-side].conn->last_delivered_id+1, (unsigned long long)ds->dropped); } + CHECK(conn->unacked_bytes == conn->input_send_q->total_bytes + conn->input_wait_ack->total_bytes); } } @@ -369,15 +379,18 @@ int main(int argc, char** argv) { CHECK(socket_platform_init() == 0); const char* output = getenv("UTUN_MULTIPATH_CSV"); if (output) { csv = fopen(output, "w"); CHECK(csv); fprintf(csv, - "case,phase,time_ms,direction,path,rtt_tb,pacing_kbps,inflight,cwnd,conn_rtt_tb,jitter,send_q,recv_q,wire_bytes,unique_bytes,retries,channel_q,app_bytes,max_gap_us\n"); } + "case,mask,phase,time_ms,direction,path,rtt_tb,pacing_kbps,inflight,cwnd,conn_rtt_tb,jitter,send_q,local_recv_q,wire_bytes," + "unique_bytes,retries,channel_q,app_bytes,max_gap_us,dest_recv_q,needed_seq,queue_drops\n"); } struct profile matrix[][3] = { {{20,8000},{20,8000}}, {{20,2000},{20,8000}}, {{20,8000},{200,8000}}, {{20,2000},{200,8000}}, {{20,8000},{200,2000}}, {{20,2000},{80,8000},{200,2000}}, {{20,2000},{80,8000}} }; const char* names[] = {"equal", "bandwidth", "rtt", "fast-distant", "slow-distant", "three", "bidirectional"}; size_t baseline = u_get_allocated_count(); + unsigned selected = 0; for (unsigned n = 0; n < sizeof(names)/sizeof(names[0]); n++) { if (argc > 1 && strcmp(argv[1], names[n])) continue; + selected++; bidirectional = n == 6; case_name = names[n]; unsigned count = n == 5 ? 3 : 2; uint64_t solo_sum = 0; @@ -395,6 +408,7 @@ int main(int argc, char** argv) { }; for (unsigned n = 0; n < sizeof(faults)/sizeof(faults[0]); n++) { if (argc > 1 && strcmp(argv[1], faults[n].name)) continue; + selected++; case_name = faults[n].name; bidirectional = 0; struct profile settings[] = {{20,8000},{20,8000},{20,2000}}; @@ -402,6 +416,7 @@ int main(int argc, char** argv) { 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); } + CHECK(selected && argc <= 2); 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;