jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Exercise DATA windows, ACKs, retries and sender wakeups

Browse Source
This commit is contained in:
evgeny committed 2026-10-22 22:33:11 +02:00
1 parent 86a0f041c7
commit 75ff7c4ce3
2 files changed
+253
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/Makefile.am
+5
View File
@@ -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)
tests/test_etcp_windows.c
+248
View File
@@ -0,0 +1,248 @@
/* Настоящие очереди/таймеры ETCP; управляемые часы, перехват только передачи и отказа аллокации. */
#include <stdlib.h>
#include <string.h>
#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;
}