From 75ff7c4ce3b39ebc7408f078132a43a9ccf707aa Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 22:33:11 +0200 Subject: [PATCH] Exercise DATA windows, ACKs, retries and sender wakeups --- tests/Makefile.am | 5 + tests/test_etcp_windows.c | 248 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 253 insertions(+) create mode 100644 tests/test_etcp_windows.c diff --git a/tests/Makefile.am b/tests/Makefile.am index 9dd8bd77..ebaf812c 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -99,6 +99,7 @@ check_PROGRAMS = \ if OS_LINUX check_PROGRAMS += \ test_ntp_regressions \ + test_etcp_windows \ test_firewall \ test_mem \ test_config_updater \ @@ -1009,6 +1010,10 @@ test_etcp_session_SOURCES = test_etcp_session.c $(test_clock_sources) test_etcp_session_LDADD = $(LIBUTUN) $(COMMON_LIBS) test_etcp_session_LDFLAGS = $(test_clock_ldflags) +test_etcp_windows_SOURCES = test_etcp_windows.c $(test_clock_sources) +test_etcp_windows_LDADD = $(LIBUTUN) $(COMMON_LIBS) +test_etcp_windows_LDFLAGS = $(test_clock_ldflags) -Wl,--wrap=etcp_encrypt_send -Wl,--wrap=memory_pool_alloc -Wl,--wrap=bbr_main + test_etcp_router_tcp_SOURCES = test_etcp_router_tcp.c test_etcp_router_tcp_LDADD = $(LIBUTUN) $(COMMON_LIBS) diff --git a/tests/test_etcp_windows.c b/tests/test_etcp_windows.c new file mode 100644 index 00000000..c82c7b26 --- /dev/null +++ b/tests/test_etcp_windows.c @@ -0,0 +1,248 @@ +/* Настоящие очереди/таймеры ETCP; управляемые часы, перехват только передачи и отказа аллокации. */ +#include +#include +#include "etcp.h" +#include "etcp_api.h" +#include "etcp_bbr.h" +#include "etcp_loadbalancer.h" +#include "config_parser.h" +#include "utun_instance.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) +static struct UASYNC* ua; +static struct UTUN_INSTANCE inst; +static struct utun_config cfg; +static struct ETCP_CONN* conn; +static struct ETCP_LINK *a, *b; +static unsigned sends, fail_alloc, fail_send, bbr_calls, saw_cwnd_limited, saw_app_limited; + +int __wrap_etcp_encrypt_send(struct ETCP_DGRAM* pkt) { + CHECK(conn->tx_running && pkt->link && pkt->link->initialized && pkt->link->link_status); + sends++; + if (fail_send) { fail_send--; return -1; } + return pkt->data_len; +} + +void* __real_memory_pool_alloc(struct memory_pool* pool); +void* __wrap_memory_pool_alloc(struct memory_pool* pool) { + if (pool == inst.pkt_pool && fail_alloc) { fail_alloc--; return NULL; } + return __real_memory_pool_alloc(pool); +} + +void __real_bbr_main(struct bbr*, const struct bbr_rate_sample*, uint32_t*, uint32_t*, uint32_t, uint32_t, int); +void __wrap_bbr_main(struct bbr* model, const struct bbr_rate_sample* sample, uint32_t* cwnd, + uint32_t* pacing, uint32_t mss, uint32_t inflight, int limited) { + bbr_calls++; saw_cwnd_limited |= !!limited; saw_app_limited |= !!sample->is_app_limited; + __real_bbr_main(model, sample, cwnd, pacing, mss, inflight, limited); +} + +/* Счётчики сверяются с содержимым очередей, а не с ожидаемыми результатами самого алгоритма. */ +static void check_accounting(void) { + for (struct ETCP_LINK* l = conn->links; l; l = l->next) { + uint32_t bytes = 0, packets = 0; + for (struct ll_entry* e = conn->input_wait_ack->head; e; e = e->next) { + struct INFLIGHT_PACKET* p = (struct INFLIGHT_PACKET*)e; + CHECK(p->state == INFLIGHT_STATE_WAIT_ACK); + if (p->last_link == l) { bytes += p->ll.len; packets++; } + } + CHECK(l->inflight_bytes == bytes && l->inflight_packets == packets); + } + for (struct ll_entry* e = conn->input_send_q->head; e; e = e->next) + CHECK(((struct INFLIGHT_PACKET*)e)->state == INFLIGHT_STATE_WAIT_SEND); + CHECK(conn->unacked_bytes == conn->input_send_q->total_bytes + conn->input_wait_ack->total_bytes); +} + +static void setup(void) { + memset(&inst, 0, sizeof(inst)); memset(&cfg, 0, sizeof(cfg)); + sends = fail_alloc = fail_send = bbr_calls = saw_cwnd_limited = saw_app_limited = 0; + ua = uasync_create(); CHECK(ua); + inst.ua = ua; inst.node_id = 1; inst.config = &cfg; cfg.global.bbr_max_cwnd = INFLIGHT_LIM_MAX; + inst.connections = queue_new(ua, 16, 0, 8, "windows"); + inst.pkt_pool = memory_pool_init(sizeof(struct ETCP_DGRAM) + PACKET_DATA_SIZE, "window_packets"); + inst.data_pool = memory_pool_init(PACKET_DATA_SIZE, "window_data"); + inst.ack_pool = memory_pool_init(sizeof(struct ACK_PACKET), "window_acks"); + CHECK(inst.connections && inst.pkt_pool && inst.data_pool && inst.ack_pool); + conn = etcp_connection_create(&inst, "windows"); CHECK(conn); + conn->peer_node_id = 2; conn->links_up = 1; + etcp_conn_ready(conn); + a = u_calloc(1, sizeof(*a)); b = u_calloc(1, sizeof(*b)); CHECK(a && b); + a->next = b; conn->links = a; + unsigned id = 0; + for (struct ETCP_LINK* l = a; l; l = l->next) { + l->etcp = conn; l->local_link_id = ++id; l->initialized = l->link_status = 1; + l->mtu = 1280; l->remote_addr.ss_family = AF_INET; l->inflight_lim_bytes = 65536; + } + etcp_conn_on_inflight_lim_changed(conn); +} + +static void teardown(void) { + check_accounting(); + etcp_connection_close(conn); + uasync_poll(ua, 0); + queue_free(inst.connections); + memory_pool_destroy(inst.pkt_pool); memory_pool_destroy(inst.data_pool); memory_pool_destroy(inst.ack_pool); + uasync_poll(ua, 0); + CHECK(ua->timer_alloc_count == ua->timer_free_count); + uasync_destroy(ua, 0); +} + +/* Вход отправителя уже после normalizer: фрагмент остаётся в реальной indexed send_q. */ +static struct INFLIGHT_PACKET* enqueue(unsigned len) { + struct INFLIGHT_PACKET* p = (struct INFLIGHT_PACKET*)queue_entry_new_from_pool(conn->inflight_pool); CHECK(p); + p->seq = conn->next_tx_id++; p->state = INFLIGHT_STATE_WAIT_SEND; + p->ll.len = len; p->ll.dgram = memory_pool_alloc(inst.data_pool); p->ll.dgram_pool = inst.data_pool; + CHECK(p->ll.dgram); memset(p->ll.dgram, 0x5a, len); + conn->unacked_bytes += len; + CHECK(queue_data_put_with_index(conn->input_send_q, &p->ll) == 0); + return p; +} + +static struct ETCP_DGRAM* request(void) { + struct ETCP_DGRAM* pkt = etcp_request_pkt(conn); + check_accounting(); + return pkt; +} + +static void fill(struct ETCP_LINK* target, unsigned count, unsigned len) { + int up_a = a->link_status, up_b = b->link_status; + a->link_status = target == a; b->link_status = target == b; + for (unsigned i = 0; i < count; i++) { + enqueue(len); + struct ETCP_DGRAM* pkt = request(); CHECK(pkt && pkt->link == target); + memory_pool_free(inst.pkt_pool, pkt); + } + a->link_status = up_a; b->link_status = up_b; +} + +static void ack(uint32_t seq) { etcp_ack_recv(conn, seq, 0, 0); check_accounting(); } +static void ack_all(void) { + while (conn->input_wait_ack->head) ack(((struct INFLIGHT_PACKET*)conn->input_wait_ack->head)->seq); +} + +/* В полном пути свободное окно другого линка не разрешает DATA через заполненный. */ +static void selection(void) { + setup(); fill(a, 8, 1024); fill(b, 16, 1024); a->inflight_lim_bytes = 8192; + enqueue(512); + struct ETCP_DGRAM* pkt = request(); CHECK(pkt && pkt->link == b); + CHECK(a->inflight_bytes == 8192 && b->inflight_bytes == 16896); + memory_pool_free(inst.pkt_pool, pkt); + teardown(); +} + +static void boundaries_ack_burst(void) { + setup(); b->link_status = 0; a->inflight_lim_bytes = 8192; + fill(a, 7, 1024); enqueue(1024); + struct ETCP_DGRAM* pkt = request(); CHECK(pkt && a->inflight_bytes == 8192); + memory_pool_free(inst.pkt_pool, pkt); + struct INFLIGHT_PACKET* waiting = enqueue(1); CHECK(!request() && a->cwnd_limited); + struct ACK_PACKET* confirmation = (struct ACK_PACKET*)queue_entry_new_from_pool(inst.ack_pool); CHECK(confirmation); + confirmation->seq = 777; + CHECK(queue_data_put_with_index(conn->ack_q, &confirmation->ll) == 0); + pkt = request(); CHECK(pkt && pkt->data[0] == ETCP_SECTION_ACK && pkt->data[1] == 1); + CHECK(waiting == (struct INFLIGHT_PACKET*)conn->input_send_q->head && a->inflight_bytes == 8192); + memory_pool_free(inst.pkt_pool, pkt); + a->burst_active = 1; a->burst_count = 2; + for (int i = 0; i < 2; i++) { + pkt = request(); CHECK(pkt && pkt->data[8] == ETCP_SECTION_MEAS_TS && (pkt->data[11] & MEAS_FLAG_IS_FILLER)); + CHECK(a->inflight_bytes == 8192 && conn->input_send_q->head == &waiting->ll); + memory_pool_free(inst.pkt_pool, pkt); + } + CHECK(!a->burst_active); + teardown(); +} + +/* RTO освобождает окно; поздний ACK удаляет ожидающий retry, затем остальные повторы переходят A -> B. */ +static void late_ack_and_migration(void) { + setup(); fill(a, 8, 1024); enqueue(600); + a->link_status = b->link_status = 0; + test_clock_run_for(ua, 70); + check_accounting(); CHECK(!conn->input_wait_ack->count && !a->inflight_bytes); + ack(1); ack(1); CHECK(!a->inflight_bytes); + b->link_status = 1; etcp_link_update_inflight_lim(b, 8192); + test_clock_run_for(ua, 10); + CHECK(!conn->input_send_q->count && b->inflight_bytes == 7768 && sends == 8); + for (struct ll_entry* e = conn->input_wait_ack->head; e; e = e->next) + CHECK(((struct INFLIGHT_PACKET*)e)->last_link == b); + ack_all(); teardown(); +} + +static void lost_ack_and_shrink(void) { + setup(); b->link_status = 0; a->inflight_lim_bytes = 16384; fill(a, 16, 1024); + etcp_link_update_inflight_lim(a, 8192); + enqueue(1024); CHECK(!request() && a->inflight_bytes == 16384); + test_clock_run_for(ua, 70); + check_accounting(); CHECK(sends == 8 && a->inflight_bytes == 8192 && conn->input_send_q->count == 9); + ack_all(); test_clock_run_for(ua, 10); + CHECK(sends == 16 && a->inflight_bytes == 8192 && conn->input_send_q->count == 1); + ack_all(); test_clock_run_for(ua, 10); ack_all(); + CHECK(!conn->unacked_bytes && !conn->input_send_q->count); + teardown(); +} + +static void grow_and_session(void) { + setup(); b->link_status = 0; a->inflight_lim_bytes = 8192; fill(a, 8, 1024); enqueue(100); + CHECK(!request()); + uint64_t allocated = ua->timer_alloc_count; + test_clock_run_for(ua, 20); CHECK(!sends && ua->timer_alloc_count - allocated < 4); + etcp_link_update_inflight_lim(a, 16384); test_clock_run_for(ua, 10); + CHECK(sends == 1 && a->inflight_bytes == 8292); ack_all(); + conn->initialized = 0; conn->session_required = conn->reinit_pending = 1; + enqueue(100); CHECK(!request()); test_clock_run_for(ua, 10); CHECK(sends == 1); + conn->session_required = 0; etcp_conn_ready(conn); test_clock_run_for(ua, 10); + CHECK(sends == 2 && conn->input_wait_ack->count == 1); + ack_all(); teardown(); +} + +static void allocation_and_send_failure(void) { + setup(); b->link_status = 0; enqueue(100); fail_alloc = 1; + test_clock_run_for(ua, 10); + CHECK(!sends && conn->input_send_q->count == 1 && !a->inflight_bytes && conn->tx_timer); + fail_send = 1; test_clock_run_for(ua, 110); + CHECK(sends == 1 && conn->input_wait_ack->count == 1); + test_clock_run_for(ua, 70); CHECK(sends == 2); + ack_all(); teardown(); +} + +static void removed_link_and_bbr(void) { + setup(); fill(a, 1, 1024); + etcp_link_close(a); a = NULL; + ack(1); CHECK(!conn->unacked_bytes); + b->bbr = u_calloc(1, sizeof(*b->bbr)); CHECK(b->bbr); bbr_init(b->bbr); + b->inflight_lim_bytes = 8192; + for (int i = 0; i < 8; i++) { + enqueue(1024); struct ETCP_DGRAM* pkt = request(); CHECK(pkt); memory_pool_free(inst.pkt_pool, pkt); + } + enqueue(100); CHECK(!request() && b->cwnd_limited); + // Первый ACK относится к отправке при наличии ожидающих данных: она не app_limited. + struct INFLIGHT_PACKET* first = (struct INFLIGHT_PACKET*)conn->input_wait_ack->head; + first->is_app_limited = 0; + ack(first->seq); CHECK(bbr_calls == 1 && saw_cwnd_limited && !saw_app_limited); + teardown(); +} + +static void config_limits(void) { + const char* values[] = {"0", "8191", "-1", "2147483648", "999999999999999999999", "abc", "8192x", "8192", "65536"}; + for (unsigned i = 0; i < sizeof(values) / sizeof(values[0]); i++) { + char text[128]; snprintf(text, sizeof(text), "[global]\nbbr_max_cwnd=%s\n", values[i]); + struct utun_config* parsed = parse_config_from_buf(text, strlen(text), "window_config"); + CHECK((parsed != NULL) == (i >= 7)); + free_config(parsed); + } +} + +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); + if (getenv("UTUN_TEST_DEBUG")) debug_set_category_level(DEBUG_CATEGORY_ETCP, DEBUG_LEVEL_DEBUG); + test_clock_init(10); + size_t baseline = u_get_allocated_count(); + void (*cases[])(void) = {selection, boundaries_ack_burst, late_ack_and_migration, lost_ack_and_shrink, + grow_and_session, allocation_and_send_failure, removed_link_and_bbr, config_limits}; + for (unsigned i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) { + cases[i](); CHECK(u_get_allocated_count() == baseline); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS ETCP window scenario %u", i); + } + test_clock_finish(); + return 0; +}