From 672f04b83336eea9d3f3454d795a6daff179eef5 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 23:23:05 +0200 Subject: [PATCH] Exercise real ETCP multipath against independent bandwidth and RTT profiles --- tests/Makefile.am | 4 + tests/test_etcp_multipath.c | 309 ++++++++++++++++++++++++++++++++++++ 2 files changed, 313 insertions(+) create mode 100644 tests/test_etcp_multipath.c diff --git a/tests/Makefile.am b/tests/Makefile.am index b56018e8..59ffddf3 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -101,6 +101,7 @@ check_PROGRAMS += \ test_ntp_regressions \ test_etcp_windows \ test_dummynet_channels \ + test_etcp_multipath \ test_stcp_sigpipe \ test_firewall \ test_mem \ @@ -1016,6 +1017,9 @@ test_etcp_windows_SOURCES = test_etcp_windows.c $(test_clock_sources) test_dummynet_channels_SOURCES = test_dummynet_channels.c $(test_clock_sources) 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 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 new file mode 100644 index 00000000..02c9cea3 --- /dev/null +++ b/tests/test_etcp_multipath.c @@ -0,0 +1,309 @@ +/* Настоящие ETCP/NCD/BBR поверх независимых UDP-каналов. Строгие проверки скорости и пауз при DROP/HOLD. */ +#include +#include +#include +#include "etcp.h" +#include "etcp_api.h" +#include "etcp_session.h" +#include "node_conn_direct.h" +#include "topo_node.h" +#include "utun_instance.h" +#include "dummynet.h" +#include "test_net.h" +#include "test_clock.h" +#include "../lib/mem.h" + +#define CHECK(x) do { if (!(x)) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "FAIL %s:%d: %s", __func__, __LINE__, #x); exit(1); } } while (0) +#define PATHS 3 +#define TRACE_COUNT 131072 +#define CMD 253 +#define SAMPLE_MS 4000 +#define WARMUP_MS 3000 +#define STALL_MS 3000 +#define MAX_GAP_US 100000 +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 endpoint { + struct UTUN_INSTANCE* inst; + struct NODE_CONN_DIRECT* owner; + struct ETCP_CONN* conn; + struct ETCP_SOCKET* sockets[PATHS]; + unsigned sent, received; + uint64_t bytes, last_delivery, max_gap, max_latency, latency_hist[10001], samples; +}; +struct path_stats { uint64_t data, retries, rx_unique; }; +static struct UASYNC* ua; +static struct endpoint ep[2]; +static struct dummynet* channels[PATHS]; +static struct profile profiles[PATHS]; +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 FILE* csv; +static const char* case_name; +static const char* pubkeys[2] = { + "b3193173def895bd0fcea6f86af077c7d77216f10395275f627ac18242ec0f01", + "ce8871f07fa056c636d297115f231b08c29cdf94e0d440fce83a07c34416d36a" +}; +static const char* privkeys[2] = { + "704f2e012c8fa8768130cb0f988a997dccb628372bc5ceccacc78dcbfec5916f", + "38240cb82199e504686507f11f6eaa4f740fde6f0c425c495e49a523019a5d68" +}; + +/* Наблюдение не заменяет передачу/приём и не меняет состояние протокола. */ +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) { + 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; + if (path >= 0) for (unsigned offset = 0; offset < pkt->data_len;) { + const uint8_t* data = pkt->data + offset; + unsigned remaining = pkt->data_len - offset, len; + switch (data[0]) { + case ETCP_SECTION_ACK: CHECK(remaining >= 8); len = 8 + 8 * data[1]; break; + case ETCP_SECTION_TIMESTAMP: len = 5; break; + case ETCP_SECTION_MEAS_TS: len = 9; break; + case ETCP_SECTION_MEAS_RESP: len = 13; break; + case ETCP_SECTION_FILLER: CHECK(remaining >= 3); len = 3 + data[1] + (data[2] << 8); break; + case ETCP_SECTION_PAYLOAD: { + CHECK(remaining >= 5); + uint32_t seq = (uint32_t)data[1] | ((uint32_t)data[2] << 8) | ((uint32_t)data[3] << 16) | ((uint32_t)data[4] << 24); + CHECK(seq < TRACE_COUNT); + struct trace* trace = &traces[sending ? side : 1-side][seq]; + uint64_t now = get_time_us(); + if (sending) { + if (trace->attempts++ && measuring) traffic[side][path].retries++; + if (!trace->first_tx) { trace->first_tx = now; trace->first_path = path; } + if (measuring) traffic[side][path].data += remaining - 5; + } else if (!trace->first_rx) { + 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, + seq, path, sending, trace->attempts, (unsigned long long)(now - fault_start)); + len = remaining; break; + } + default: len = remaining; break; + } + CHECK(len <= remaining); offset += len; + } + __real_etcp_dump_pkt_sections(pkt, link, sending); +} + +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]; + CHECK(entry->len == sizeof(struct message)); + struct message msg; memcpy(&msg, entry->dgram, sizeof(msg)); + CHECK(msg.cmd == CMD && msg.from == 1-side && msg.seq == dest->received); + for (unsigned i = 0; i < sizeof(msg.body); i++) CHECK(msg.body[i] == (uint8_t)(msg.seq * 3 + msg.from + i)); + dest->received++; dest->bytes += sizeof(msg); + uint64_t now = get_time_us(); + 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; + dest->latency_hist[latency / 1000 < 10000 ? latency / 1000 : 10000]++; dest->samples++; + } + dest->last_delivery = now; + queue_dgram_free(entry); queue_entry_free(entry); +} + +static void create_endpoint(unsigned side) { + char config[2048], listeners[768] = ""; + for (unsigned p = 0; p < path_count; p++) { + char text[160]; snprintf(text, sizeof(text), "[server:p%u]\naddr=127.0.0.1:0\ntype=public\n", p); + 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", + 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); + CHECK(etcp_bind(ep[side].inst, CMD, receive_message) == 0); + for (unsigned p = 0; p < path_count; p++) { + char name[16]; snprintf(name, sizeof(name), "p%u", p); + ep[side].sockets[p] = test_etcp_bound_socket(ep[side].inst, name); + } +} + +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; + memcpy(profiles, settings, count * sizeof(*settings)); + ua = uasync_create(); CHECK(ua); + create_endpoint(0); create_endpoint(1); + for (unsigned p = 0; p < count; p++) { + if (!(mask & (1u << p))) continue; + channels[p] = dummynet_create(ua, "127.0.0.1", 0); CHECK(channels[p]); + for (int dir = 0; dir < 2; dir++) CHECK(dummynet_set_direction(channels[p], dir, settings[p].rtt_ms / 2, 0, + settings[p].kbps, 512, 0, "127.0.0.1", ss_get_port(&ep[1-dir].sockets[p]->local_addr)) == 0); + struct TOPO_ADDR4 addr = { .protocol = TOPO_PROTO_UDP, .port = dummynet_get_listen_port(channels[p]) }; + uint32_t ip = htonl(INADDR_LOOPBACK); memcpy(addr.addr, &ip, sizeof(ip)); + struct TOPO_NODE peer = { .node_id = ep[1].inst->node_id, .node_name = "multipath-peer", .v4_addrs = &addr }; + memcpy(peer.public_key, ep[1].inst->my_keys.public_key, SC_PUBKEY_SIZE); + struct NODE_CONN_DIRECT* extra = NULL; + CHECK(node_conn_direct_open_node(ep[0].inst, peer.node_id, NULL, NULL, &extra, &peer, ep[0].sockets[p]) >= 0); + if (!ep[0].owner) ep[0].owner = extra; + else node_conn_direct_close(extra); + } + uint64_t end = get_time_us() + 10000000; + int ready = 0; + while (get_time_us() < end) { + test_clock_poll(ua, 10); ready = 1; + for (int i = 0; i < 2; i++) { + ep[i].conn = instance_find_conn(ep[i].inst, ep[1-i].inst->node_id); + 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++; + if (found != 1) ready = 0; + } + } + if (ready) break; + } + CHECK(ready); + struct TOPO_NODE peer = { .node_id = ep[0].inst->node_id, .node_name = "held-peer" }; + memcpy(peer.public_key, ep[0].inst->my_keys.public_key, SC_PUBKEY_SIZE); + CHECK(node_conn_direct_open_node(ep[1].inst, peer.node_id, NULL, NULL, &ep[1].owner, &peer, NULL) >= 0); + started = get_time_us(); + for (int i = 0; i < 2; i++) ep[i].last_delivery = started; +} + +/* Ограниченная очередь источника: latency не скрывается за неограниченным накоплением тестовых сообщений. */ +static void produce(unsigned side) { + struct endpoint* source = &ep[side]; + uint64_t now = get_time_us(); + if (low_load && now < next_message[side]) return; + for (unsigned budget = 0; budget < 64 && source->conn->send_input_q->count < 4; budget++) { + if (source->sent - ep[1-side].received >= 2048) break; + struct ll_entry* entry = ll_alloc_lldgram(sizeof(struct message)); CHECK(entry); + struct message msg = { .cmd = CMD, .from = side, .seq = source->sent, .enqueued = now }; + for (unsigned i = 0; i < sizeof(msg.body); i++) msg.body[i] = msg.seq * 3 + side + i; + memcpy(entry->dgram, &msg, sizeof(msg)); entry->len = sizeof(msg); + CHECK(etcp_send(source->conn, entry) == 0); source->sent++; + if (low_load) { next_message[side] = now + 10000; break; } + } +} + +static void snapshot(const char* phase) { + uint64_t now = get_time_us(); + 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) { + 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, + 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); + } + } +} + +static void run_for(unsigned ms, int sending, const char* phase) { + uint64_t end = get_time_us() + ms * 1000ULL, next_snapshot = get_time_us(); + while (get_time_us() < end) { + if (sending) { produce(0); if (bidirectional) produce(1); } + test_clock_poll(ua, 10); + for (int i = 0; i < 2; i++) CHECK(ep[i].conn->state != 2 && !ep[i].conn->reinit_pending); + if (get_time_us() >= next_snapshot) { snapshot(phase); next_snapshot = get_time_us() + 100000; } + } +} + +static void begin_measure(void) { + measure_start = get_time_us(); measuring = 1; + memset(traffic, 0, sizeof(traffic)); + for (int i = 0; i < 2; i++) { + ep[i].max_gap = ep[i].max_latency = ep[i].samples = 0; ep[i].last_delivery = measure_start; + memset(ep[i].latency_hist, 0, sizeof(ep[i].latency_hist)); + } + for (unsigned p = 0; p < path_count; p++) if (channels[p]) dummynet_reset_stats(channels[p], -1); +} + +static void expect(int passed, const char* requirement, uint64_t actual, uint64_t limit) { + if (passed) return; + failures++; + DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "FAIL %s: %s actual=%llu limit=%llu", case_name, requirement, + (unsigned long long)actual, (unsigned long long)limit); +} + +static void teardown(void) { + measuring = 0; + // Уничтожаем и активные delay/HOLD пакеты: эмуляторы не должны пережить свои endpoints. + for (unsigned p = 0; p < path_count; p++) { dummynet_destroy(channels[p]); channels[p] = NULL; } + for (int i = 0; i < 2; i++) node_conn_direct_close(ep[i].owner); + uasync_drain_immediate(ua); + for (int i = 0; i < 2; i++) { utun_instance_destroy(ep[i].inst); ep[i].inst = NULL; } + uasync_poll(ua, 0); CHECK(ua->timer_alloc_count == ua->timer_free_count); uasync_destroy(ua, 0); +} + +/* Полезные байты приложения — отдельный критерий; FILLER/повторы не могут улучшить результат. */ +static uint64_t measure_rate(const struct profile* settings, unsigned count, unsigned mask) { + setup(settings, count, mask); low_load = 0; + run_for(WARMUP_MS, 1, "warmup"); begin_measure(); + uint64_t before[2] = {ep[0].bytes, ep[1].bytes}; + run_for(SAMPLE_MS, 1, "steady"); + uint64_t elapsed = get_time_us() - measure_start, rate = (ep[1].bytes-before[1]) * 8000 / elapsed; + for (unsigned side = 0; side <= bidirectional; side++) { + uint64_t useful = (ep[1-side].bytes-before[1-side]) * 8000 / elapsed, capacity = 0; + for (unsigned p = 0; p < count; p++) if (mask & (1u << p)) { + uint64_t wire = dummynet_get_stats(channels[p], side)->sent_bytes * 8000 / elapsed; + uint64_t unique = traffic[side][p].rx_unique * 8000 / elapsed; + capacity += settings[p].kbps; + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "RATE %s dir=%u path=%u RTT=%u BWlim=%u wire=%llu unique=%llu kbps retries=%llu", + case_name, side, p, settings[p].rtt_ms, settings[p].kbps, (unsigned long long)wire, + (unsigned long long)unique, (unsigned long long)traffic[side][p].retries); + expect(wire >= settings[p].kbps * 90 / 100, "channel wire utilization >=90%", wire, settings[p].kbps * 90 / 100); + expect(unique >= settings[p].kbps * 80 / 100, "channel unique DATA >=80%", unique, settings[p].kbps * 80 / 100); + } + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "GOODPUT %s dir=%u useful=%llu capacity=%llu kbps", case_name, side, + (unsigned long long)useful, (unsigned long long)capacity); + expect(useful >= capacity * 85 / 100, "application goodput >=85% capacity", useful, capacity * 85 / 100); + } + measuring = 0; + run_for(5000, 0, "drain"); + for (int i = 0; i < 2; i++) CHECK(ep[i].sent == ep[1-i].received); + teardown(); return rate; +} + +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); + 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"); } + 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}} + }; + const char* names[] = {"equal", "bandwidth", "rtt", "fast-distant", "slow-distant", "three"}; + size_t baseline = u_get_allocated_count(); + for (unsigned n = 0; n < sizeof(names)/sizeof(names[0]); n++) { + if (argc > 1 && strcmp(argv[1], names[n])) continue; + case_name = names[n]; unsigned count = n == 5 ? 3 : 2; + uint64_t solo_sum = 0; + for (unsigned p = 0; p < count; p++) solo_sum += measure_rate(matrix[n], count, 1u << p); + uint64_t aggregate = measure_rate(matrix[n], count, (1u << count)-1); + expect(aggregate >= solo_sum * 90 / 100, "aggregation >=90% sum of solo rates", aggregate, solo_sum * 90 / 100); + 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; +}