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

Exercise retry rotation, late attempt ACKs and receiver admission faults

Browse Source
This commit is contained in:
evgeny committed 2026-10-23 10:38:23 +02:00
1 parent a4a9ae7b0a
commit 5e436975c6
1 file changed
+134 -1
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_etcp_windows.c
+134 -1
View File
@@ -18,6 +18,9 @@ static struct ETCP_CONN* conn;
static struct ETCP_LINK *a, *b;
static unsigned sends, fail_alloc, fail_send, fail_tx_timer, bbr_calls, saw_cwnd_limited, saw_app_limited;
static unsigned fail_shaper_timer;
static unsigned fail_rx_data, fail_rx_fragment, fail_ack_entry;
static struct ETCP_LINK* last_sent_link;
static unsigned last_sent_request, last_sent_payload;
static struct bbr_rate_sample last_sample;
static uint32_t last_mss;
@@ -37,12 +40,16 @@ void* __wrap_uasync_set_timeout(struct UASYNC* loop, int delay, void* arg, timeo
int __wrap_etcp_encrypt_send(struct ETCP_DGRAM* pkt) {
CHECK(conn->tx_running && pkt->link && pkt->link->initialized && pkt->link->link_status);
sends++;
last_sent_link = pkt->link; last_sent_request = pkt->ack_request; last_sent_payload = pkt->payload_len;
if (fail_send) { fail_send--; return -1; }
return pkt->data_len;
}
void* __real_memory_pool_alloc_impl(struct memory_pool* pool, const char* location);
void* __wrap_memory_pool_alloc_impl(struct memory_pool* pool, const char* location) {
if (pool == inst.data_pool && fail_rx_data) { fail_rx_data--; return NULL; }
if (conn && pool == conn->io_pool && fail_rx_fragment) { fail_rx_fragment--; return NULL; }
if (pool == inst.ack_pool && fail_ack_entry) { fail_ack_entry--; return NULL; }
if (pool == inst.pkt_pool && fail_alloc) { fail_alloc--; return NULL; }
return __real_memory_pool_alloc_impl(pool, location);
}
@@ -80,6 +87,8 @@ static void setup(void) {
memset(&inst, 0, sizeof(inst)); memset(&cfg, 0, sizeof(cfg));
sends = fail_alloc = fail_send = fail_tx_timer = bbr_calls = saw_cwnd_limited = saw_app_limited = 0;
fail_shaper_timer = 0;
fail_rx_data = fail_rx_fragment = fail_ack_entry = 0;
last_sent_link = NULL; last_sent_request = last_sent_payload = 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");
@@ -462,6 +471,129 @@ static void tcp_initial_window(void) {
teardown();
}
static void enable_models(void) {
for (struct ETCP_LINK* link = conn->links; link; link = link->next) {
link->bbr = u_calloc(1, sizeof(*link->bbr)); CHECK(link->bbr);
bbr_init(link->bbr); link->bbr->link_id = link->local_link_id;
}
}
static void receive_till(struct ETCP_LINK* link, uint32_t till, int query) {
struct ETCP_DGRAM* pkt = memory_pool_alloc(inst.pkt_pool); CHECK(pkt);
pkt->link = link; pkt->data_len = ETCP_ACK_BASE_SIZE + query;
memset(pkt->data, 0, pkt->data_len); pkt->data[0] = ETCP_SECTION_ACK;
for (unsigned i = 0; i < 4; i++) pkt->data[2 + i] = till >> (8 * i);
if (query) pkt->data[ETCP_ACK_BASE_SIZE] = ETCP_SECTION_ACK_REQUEST;
etcp_conn_input(pkt); check_accounting();
}
static void receive_fragment(struct ETCP_LINK* link, uint32_t seq, uint64_t id) {
struct ETCP_DGRAM* pkt = memory_pool_alloc(inst.pkt_pool); CHECK(pkt);
pkt->link = link; pkt->data_len = ETCP_PAYLOAD_HEADER_SIZE + 1;
memset(pkt->data, 0, pkt->data_len); pkt->data[0] = ETCP_SECTION_PAYLOAD;
for (unsigned i = 0; i < 4; i++) pkt->data[1 + i] = seq >> (8 * i);
for (unsigned i = 0; i < 8; i++) pkt->data[5 + i] = id >> (8 * i);
pkt->data[ETCP_PAYLOAD_HEADER_SIZE] = 0x5a;
etcp_conn_input(pkt);
}
/* Повтор обходит ещё UP-канал, даже если у него меньше inflight; общий RR не определяет курсор пакета. */
static void retry_rotation(void) {
setup(); fill(a, 1, 100); fill(b, 1, 1000); b->data_rtt_us = 1000000;
conn->initialized = 0; test_clock_run_for(ua, 120); conn->initialized = 1;
CHECK(a->link_status && b->link_status && !a->inflight_bytes && b->inflight_bytes == 1000);
conn->last_rr_link = b;
struct ETCP_DGRAM* pkt = request(); CHECK(pkt && pkt->link == b);
memory_pool_free(inst.pkt_pool, pkt); ack_all(); teardown();
setup();
struct ETCP_LINK* c = u_calloc(1, sizeof(*c)); CHECK(c);
c->etcp = conn; c->local_link_id = 3; c->initialized = c->link_status = 1;
c->mtu = 1280; c->remote_addr.ss_family = AF_INET; c->rtt_last = 1; c->inflight_lim_bytes = 65536;
CHECK(etcp_tx_link_init(c) == 0); b->next = c;
fill(a, 1, 100);
struct ETCP_LINK* expected[] = {b, c, a};
for (unsigned i = 0; i < 3; i++) {
conn->initialized = 0; test_clock_run_for(ua, 120); conn->initialized = 1;
conn->last_rr_link = expected[i]; // посторонний выбор/ACK меняет общий RR, но не порядок повторов
pkt = request(); CHECK(pkt && pkt->link == expected[i]);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Retry rotation: step=%u selected=%u", i, pkt->link->local_link_id);
memory_pool_free(inst.pkt_pool, pkt);
}
ack_all(); teardown();
}
/* Поздняя копия и общий till не освобождают окно/не обучают модель другого канала. */
static void independent_attempt_acks(void) {
setup(); enable_models(); fill(a, 1, 100);
struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)conn->input_wait_ack->head;
uint64_t old_id = packet->last_attempt_id;
conn->initialized = 0; test_clock_run_for(ua, 120); conn->initialized = 1;
struct ETCP_DGRAM* pkt = request(); CHECK(pkt && pkt->link == b);
memory_pool_free(inst.pkt_pool, pkt);
packet = (struct INFLIGHT_PACKET*)conn->input_wait_ack->head; uint64_t new_id = packet->last_attempt_id;
CHECK(old_id != new_id && !a->inflight_bytes && b->inflight_bytes == 100);
etcp_tx_ack(a, packet->seq, old_id, 0);
CHECK(!conn->unacked_bytes && conn->ack_packets_count == 1 && b->inflight_bytes == 100);
CHECK(!a->bbr->delivered && !b->bbr->delivered);
etcp_tx_ack(b, 1, new_id, 0); etcp_tx_ack(b, 1, new_id, 0);
CHECK(!b->inflight_bytes && b->bbr->delivered == 100 && conn->ack_packets_count == 1);
check_accounting(); teardown();
setup(); enable_models(); fill(a, 1, 100);
packet = (struct INFLIGHT_PACKET*)conn->input_wait_ack->head; old_id = packet->last_attempt_id;
receive_till(b, 1, 0);
CHECK(!conn->unacked_bytes && a->inflight_bytes == 100 && !a->bbr->delivered && !b->bbr->delivered);
etcp_tx_ack(a, 1, old_id, 0); CHECK(!a->inflight_bytes && a->bbr->delivered == 100);
teardown();
setup(); enable_models(); fill(a, 1, 100);
packet = (struct INFLIGHT_PACKET*)queue_data_get(conn->input_wait_ack); old_id = packet->last_attempt_id;
packet->state = INFLIGHT_STATE_WAIT_SEND; CHECK(queue_data_put_with_index(conn->input_send_q, &packet->ll) == 0);
pkt = request(); CHECK(pkt && pkt->link == b && packet->active_attempts == 2);
memory_pool_free(inst.pkt_pool, pkt); new_id = packet->last_attempt_id;
etcp_tx_ack(a, 1, old_id, 0);
CHECK(!conn->unacked_bytes && !a->inflight_bytes && b->inflight_bytes == 100);
CHECK(a->bbr->delivered == 100 && !b->bbr->delivered);
etcp_tx_ack(b, 1, new_id, 0); CHECK(b->bbr->delivered == 100 && conn->ack_packets_count == 1);
check_accounting(); teardown();
}
static void receive_admission_and_path_acks(void) {
setup(); conn->got_initial_pkt = 1;
fail_rx_data = 1; receive_fragment(a, 2, 1);
CHECK(!fail_rx_data && !conn->recv_q->count && !a->ack_q->count && !conn->last_delivered_id);
fail_rx_fragment = 1; receive_fragment(a, 2, 2);
CHECK(!fail_rx_fragment && !conn->recv_q->count && !a->ack_q->count);
queue_set_size_limit(conn->recv_q, 0); receive_fragment(a, 2, 3);
CHECK(!conn->recv_q->count && !a->ack_q->count); queue_set_size_limit(conn->recv_q, -1);
receive_fragment(b, 2, 4);
CHECK(conn->recv_q->count == 1 && b->ack_q->count == 1 && !a->ack_q->count);
receive_fragment(a, 2, 5);
CHECK(conn->recv_q->count == 1 && a->ack_q->count == 1 && b->ack_q->count == 1);
fail_ack_entry = 1; receive_fragment(a, 3, 6);
CHECK(!fail_ack_entry && conn->recv_q->count == 2 && a->ack_q->count == 1);
receive_fragment(b, 3, 7); CHECK(conn->recv_q->count == 2 && b->ack_q->count == 2);
teardown();
}
static void peer_window_queries(void) {
setup(); conn->last_delivered_id = 9; conn->last_rr_link = b;
receive_till(b, 0, 1);
struct ETCP_DGRAM* pkt = request(); CHECK(pkt && pkt->link == b && !pkt->payload_len && pkt->data[2] == 9);
memory_pool_free(inst.pkt_pool, pkt); CHECK(!b->ack_reply_pending && !request()); teardown();
setup(); conn->next_tx_id = MAX_INFLIGHT_SIZE + 1;
struct ETCP_FRAGMENT* fragment = (struct ETCP_FRAGMENT*)queue_entry_new_from_pool(conn->io_pool); CHECK(fragment);
fragment->ll.len = 1; fragment->ll.dgram = memory_pool_alloc(inst.data_pool); CHECK(fragment->ll.dgram);
fragment->ll.dgram_pool = inst.data_pool;
CHECK(queue_data_put(conn->input_queue, &fragment->ll) == 0 && conn->ack_window_timer);
wait_sends(1); CHECK(last_sent_request && !last_sent_payload && last_sent_link);
receive_till(last_sent_link, MAX_INFLIGHT_SIZE, 0);
CHECK(!conn->ack_window_timer && !conn->input_queue->count && conn->input_send_q->count == 1);
teardown();
}
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);
@@ -470,7 +602,8 @@ int main(void) {
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, pacing_reset_close, initial_readiness,
config_limits, timer_failure, immediate_send, ack_without_data_pacing, bbr_delivery_clock, bbr_probe_window, tcp_initial_window,
retransmission_timeout, pacing_rate_changes, pacing_timer_failure};
retransmission_timeout, pacing_rate_changes, pacing_timer_failure, retry_rotation, independent_attempt_acks,
receive_admission_and_path_acks, peer_window_queries};
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);