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

Exercise receiver bandwidth bounds and ACK progress framing

Browse Source
This commit is contained in:
evgeny committed 2026-10-23 16:39:12 +02:00
1 parent ea4a92125b
commit 4bf47ca357
3 files changed
+172 -10
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/test_etcp_multipath.c
+1 -1
View File
@@ -91,7 +91,7 @@ void __wrap_etcp_dump_pkt_sections(struct ETCP_DGRAM* pkt, struct ETCP_LINK* lin
const uint8_t* data = pkt->data + offset;
unsigned remaining = pkt->data_len - offset, len;
switch (data[0]) {
case ETCP_SECTION_ACK: CHECK(remaining >= ETCP_ACK_BASE_SIZE); len = ETCP_ACK_BASE_SIZE + ETCP_ACK_ENTRY_SIZE * data[1]; break;
case ETCP_SECTION_ACK: CHECK(remaining >= ETCP_ACK_BASE_SIZE); len = etcp_ack_section_size(data[1]); break;
case ETCP_SECTION_ACK_REQUEST: len = 1; break;
case ETCP_SECTION_TIMESTAMP: len = 5; break;
case ETCP_SECTION_MEAS_TS: len = 9; break;
tests/test_etcp_session.c
+1 -1
View File
@@ -359,7 +359,7 @@ static void receive_window_case(int wrap) {
CHECK(c->input_queue->count == 1 && c->next_tx_id == next);
for (int cumulative = 0; cumulative < 2; cumulative++) {
struct ETCP_DGRAM* p = memory_pool_alloc(ep[0].inst.pkt_pool); CHECK(p);
p->link = &ep[0].links[0]; p->data_len = ETCP_ACK_BASE_SIZE + (cumulative ? 0 : ETCP_ACK_ENTRY_SIZE);
p->link = &ep[0].links[0]; p->data_len = etcp_ack_section_size(!cumulative);
memset(p->data, 0, p->data_len); p->data[0] = ETCP_SECTION_ACK; p->data[1] = !cumulative;
uint32_t ack = till + cumulative, selective = next - 1;
for (int i = 0; i < 4; i++) { p->data[2+i] = ack >> (8*i); p->data[8+i] = selective >> (8*i); }
tests/test_etcp_windows.c
+170 -8
View File
@@ -200,7 +200,7 @@ static void fill(struct ETCP_LINK* target, unsigned count, unsigned len) {
static void ack(uint32_t seq) {
struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)queue_find_data_by_index(conn->input_wait_ack, &seq);
if (!packet) packet = (struct INFLIGHT_PACKET*)queue_find_data_by_index(conn->input_send_q, &seq);
if (packet && packet->last_link) etcp_tx_ack(packet->last_link, seq, packet->last_attempt_id, 0);
if (packet && packet->last_link) etcp_tx_ack(packet->last_link, seq, packet->last_attempt_id, 0, 0, 0);
else etcp_ack_recv(conn, seq);
check_accounting();
}
@@ -459,7 +459,7 @@ static void attempt_memory_limits(void) {
CHECK(!etcp_loadbalancer_select_link(conn, 1, NULL));
CHECK(!etcp_tx_prepare(a, pending));
struct ETCP_TX_ATTEMPT old = *(struct ETCP_TX_ATTEMPT*)a->attempts->head;
etcp_tx_ack(a, old.seq, old.attempt_id, UINT32_MAX);
etcp_tx_ack(a, old.seq, old.attempt_id, UINT32_MAX, 0, 0);
CHECK(etcp_loadbalancer_select_link(conn, 1, NULL) == a);
CHECK(conn->active_attempts == 2 * MAX_INFLIGHT_SIZE - 1);
for (unsigned i = 0; i < MAX_INFLIGHT_SIZE; i++) CHECK(etcp_tx_queue_ack(a, i+1, i+1) == 0);
@@ -647,10 +647,10 @@ static void independent_attempt_acks(void) {
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);
etcp_tx_ack(a, packet->seq, old_id, 0, 0, 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);
etcp_tx_ack(b, 1, new_id, 0, 0, 0); etcp_tx_ack(b, 1, new_id, 0, 0, 0);
CHECK(!b->inflight_bytes && b->bbr->delivered == 100 && conn->ack_packets_count == 1);
check_accounting(); teardown();
@@ -658,7 +658,7 @@ static void independent_attempt_acks(void) {
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);
etcp_tx_ack(a, 1, old_id, 0, 0, 0); CHECK(!a->inflight_bytes && a->bbr->delivered == 100);
teardown();
setup(); enable_models(); fill(a, 1, 100);
@@ -666,10 +666,10 @@ static void independent_attempt_acks(void) {
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);
etcp_tx_ack(a, 1, old_id, 0, 0, 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);
etcp_tx_ack(b, 1, new_id, 0, 0, 0); CHECK(b->bbr->delivered == 100 && conn->ack_packets_count == 1);
check_accounting(); teardown();
}
@@ -677,17 +677,178 @@ 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);
CHECK(!a->rx_data_bytes && !a->rx_data_time_us);
fail_rx_fragment = 1; receive_fragment(a, 2, 2);
CHECK(!fail_rx_fragment && !conn->recv_q->count && !a->ack_q->count);
CHECK(!a->rx_data_bytes && !a->rx_data_time_us);
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);
CHECK(!a->rx_data_bytes && !a->rx_data_time_us);
receive_fragment(b, 2, 4);
CHECK(conn->recv_q->count == 1 && b->ack_q->count == 1 && !a->ack_q->count);
CHECK(b->rx_data_bytes == 1 && b->rx_data_time_us && !a->rx_data_bytes);
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);
CHECK(a->rx_data_bytes == 2 && b->rx_data_bytes == 2);
teardown();
}
/* Реальный ACK handler: RX clock независим, граница ограничивает BW и округляется безопасно. */
static void receiver_bandwidth_bounds(void) {
const struct { uint32_t delivered, local_us; uint64_t rx_bytes, rx_us; uint32_t expected; } cases[] = {
{170502,150732,170502,186359,186359}, {1000,10000,3000,40001,13334},
{1000,20000,5000,10000,0}, {UINT32_MAX,10000,UINT32_MAX,UINT32_MAX,UINT32_MAX},
{UINT32_MAX,10000,1,UINT32_MAX,0}, {1000,10000,1,(uint64_t)UINT32_MAX+1,0}
};
for (unsigned i = 0; i < sizeof(cases)/sizeof(cases[0]); i++) {
setup(); b->link_status = 0; a->data_rtt_us = 1000000;
a->bbr = u_calloc(1, sizeof(*a->bbr)); CHECK(a->bbr); bbr_init(a->bbr);
fill(a,1,1); // Первая неподтверждённая попытка удерживает flight, пока фиксируется база следующей.
a->peer_rx_data_bytes = UINT64_C(1)<<60;
a->peer_rx_data_time_us = UINT64_C(1)<<61;
struct INFLIGHT_PACKET* packet = enqueue(1000);
struct ETCP_DGRAM* pkt = request(); CHECK(pkt && pkt->link == a); memory_pool_free(inst.pkt_pool,pkt);
struct ETCP_TX_ATTEMPT* attempt = last_attempt(packet);
attempt->delivered_at_send = 10000;
a->delivered_bytes = 10000 + (uint64_t)cases[i].delivered - 1000;
a->bbr->delivered = (uint32_t)a->delivered_bytes;
uint64_t now = get_time_us();
attempt->delivered_time_us = now-cases[i].local_us;
attempt->sent_at_us = attempt->first_tx_us = now-5000;
a->data_rtt_us = 0;
etcp_tx_ack(a,packet->seq,packet->last_attempt_id,0,
(UINT64_C(1)<<60)+cases[i].rx_bytes,(UINT64_C(1)<<61)+cases[i].rx_us);
CHECK(last_sample.delivered == cases[i].delivered && last_sample.acked_sacked == 1000);
if (i == 2) CHECK(last_sample.interval_us >= cases[i].local_us);
else CHECK(last_sample.interval_us == cases[i].expected);
if (last_sample.interval_us) CHECK((uint64_t)last_sample.delivered*cases[i].rx_us <=
(uint64_t)last_sample.interval_us*cases[i].rx_bytes);
CHECK(a->bbr->min_rtt_us == last_sample.rtt_us && a->data_rtt_us == last_sample.rtt_us);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Receiver rate bound: case=%u sample=%u/%u RX=%llu/%llu RTT_us=%u",
i,last_sample.delivered,last_sample.interval_us,(unsigned long long)cases[i].rx_bytes,
(unsigned long long)cases[i].rx_us,last_sample.rtt_us);
check_accounting(); teardown();
}
}
/* Переставленные, повторные и чужие ACK не откатывают точку пути; после простоя база начинается заново. */
static void receiver_point_lifecycle(void) {
setup(); b->link_status = 0; a->data_rtt_us = 1000000; enable_models();
fill(a,1,1);
a->peer_rx_data_bytes = 10000; a->peer_rx_data_time_us = 1000000;
struct INFLIGHT_PACKET* packets[3]; uint32_t seq[3]; uint64_t ids[3];
for (unsigned i = 0; i < 3; i++) {
packets[i] = enqueue(1000); seq[i] = packets[i]->seq;
struct ETCP_DGRAM* pkt = request(); CHECK(pkt); memory_pool_free(inst.pkt_pool,pkt);
ids[i] = packets[i]->last_attempt_id;
last_attempt(packets[i])->delivered_time_us = get_time_us()-10000;
}
const unsigned order[] = {2,0,1};
for (unsigned j = 0; j < 3; j++) {
unsigned i = order[j];
etcp_tx_ack(a,seq[i],ids[i],0,11000+1000*i,1020000+20000*i);
CHECK(a->peer_rx_data_bytes == 13000 && a->peer_rx_data_time_us == 1060000);
CHECK(!b->peer_rx_data_bytes && last_sample.interval_us >= (j+1)*20000);
}
unsigned calls = bbr_calls;
etcp_tx_ack(a,seq[2],ids[2],0,UINT64_MAX,UINT64_MAX);
etcp_tx_ack(a,999,UINT64_MAX,0,UINT64_MAX,UINT64_MAX);
struct INFLIGHT_PACKET* anchor = (struct INFLIGHT_PACKET*)conn->input_wait_ack->head;
CHECK(anchor && anchor->seq == 1);
etcp_tx_ack(a,999,anchor->last_attempt_id,0,UINT64_MAX,UINT64_MAX);
CHECK(bbr_calls == calls && a->peer_rx_data_bytes == 13000 && a->peer_rx_data_time_us == 1060000);
const uint64_t invalid[][2] = {{14000,0},{0,1070000},{14000,1050000},{12000,1070000},{13000,1070000}};
for (unsigned i = 0; i < sizeof(invalid)/sizeof(invalid[0]); i++) {
struct INFLIGHT_PACKET* packet = enqueue(1000);
struct ETCP_DGRAM* pkt = request(); CHECK(pkt); memory_pool_free(inst.pkt_pool,pkt);
last_attempt(packet)->delivered_time_us = get_time_us()-10000;
etcp_tx_ack(a,packet->seq,packet->last_attempt_id,0,invalid[i][0],invalid[i][1]);
CHECK(a->peer_rx_data_bytes == 13000 && a->peer_rx_data_time_us == 1060000);
CHECK(last_sample.interval_us >= 10000);
}
ack_all(); CHECK(!a->inflight_packets);
struct INFLIGHT_PACKET* first = enqueue(1000);
struct ETCP_DGRAM* pkt = request(); CHECK(pkt); memory_pool_free(inst.pkt_pool,pkt);
CHECK(!a->peer_rx_data_time_us && !last_attempt(first)->peer_rx_time_at_send);
struct INFLIGHT_PACKET* second = enqueue(1000);
pkt = request(); CHECK(pkt); memory_pool_free(inst.pkt_pool,pkt);
last_attempt(first)->delivered_time_us = get_time_us()-10000;
uint64_t after_idle = UINT64_C(3600)*1000000;
etcp_tx_ack(a,first->seq,first->last_attempt_id,0,14000,after_idle);
CHECK(last_sample.interval_us >= 10000 && last_sample.interval_us < 1000000);
struct INFLIGHT_PACKET* third = enqueue(1000);
pkt = request(); CHECK(pkt); memory_pool_free(inst.pkt_pool,pkt);
CHECK(last_attempt(third)->peer_rx_time_at_send == after_idle);
last_attempt(third)->delivered_time_us = get_time_us()-10000;
etcp_tx_ack(a,third->seq,third->last_attempt_id,0,15000,after_idle+20000);
CHECK(last_sample.interval_us == 20000);
CHECK(second->active_attempts == 1);
a->rx_data_bytes = 55; a->rx_data_time_us = 99;
etcp_tx_link_reset(a,1);
CHECK(!a->rx_data_bytes && !a->rx_data_time_us && !a->peer_rx_data_bytes && !a->peer_rx_data_time_us);
check_accounting(); teardown();
}
/* Точка приёма занимает место только с положительными ACK; проверяем обе стороны границы MTU. */
static void receiver_point_wire(void) {
setup(); b->link_status = 0; conn->got_initial_pkt = 1;
receive_fragment(a,2,101); receive_fragment(a,2,101);
CHECK(a->rx_data_bytes == 2 && a->ack_q->count == 1 && conn->recv_q->count == 1);
a->rx_data_bytes = UINT64_C(0xfedcba9876543210); a->rx_data_time_us = UINT64_C(0x0011223344556677);
const uint8_t expected[] = {0x10,0x32,0x54,0x76,0x98,0xba,0xdc,0xfe,0x77,0x66,0x55,0x44,0x33,0x22,0x11,0x00};
struct ETCP_DGRAM* pkt = request(); CHECK(pkt && pkt->data[1] == 1 && !pkt->payload_len);
CHECK(pkt->data_len == etcp_ack_section_size(1) && !memcmp(pkt->data+ETCP_ACK_BASE_SIZE+ETCP_ACK_ENTRY_SIZE,expected,16));
memory_pool_free(inst.pkt_pool,pkt);
receive_till(a,0,1); pkt = request(); CHECK(pkt && !pkt->data[1] && pkt->data_len == ETCP_ACK_BASE_SIZE);
memory_pool_free(inst.pkt_pool,pkt);
a->rx_data_bytes = 2; a->rx_data_time_us = get_time_us();
unsigned max_body = a->mtu-WIRE_OVERHEAD(AF_INET);
for (unsigned space = ETCP_ACK_ENTRY_SIZE+ETCP_ACK_RX_POINT_SIZE-1; space <= ETCP_ACK_ENTRY_SIZE+ETCP_ACK_RX_POINT_SIZE; space++) {
receive_fragment(a,3+space,200+space);
struct INFLIGHT_PACKET* packet = enqueue(max_body-ETCP_ACK_BASE_SIZE-ETCP_PAYLOAD_HEADER_SIZE-space);
uint32_t seq = packet->seq;
pkt = request(); CHECK(pkt && pkt->payload_len && pkt->data_len <= max_body);
if (space == ETCP_ACK_ENTRY_SIZE+ETCP_ACK_RX_POINT_SIZE) CHECK(pkt->data[1] == 1 && pkt->data_len == max_body);
else CHECK(!pkt->data[1] && a->ack_q->count == 1);
memory_pool_free(inst.pkt_pool,pkt);
if (a->ack_q->count) {
pkt = request(); CHECK(pkt && pkt->data[1] == 1 && !pkt->payload_len);
memory_pool_free(inst.pkt_pool,pkt);
}
ack(seq);
}
for (unsigned i = 100; i < 200; i++) receive_fragment(a,i,1000+i);
unsigned count = (max_body-ETCP_ACK_BASE_SIZE-ETCP_ACK_RX_POINT_SIZE)/ETCP_ACK_ENTRY_SIZE;
pkt = request(); CHECK(pkt && pkt->data[1] == count && pkt->data_len <= max_body && a->ack_q->count == 100-(int)count);
memory_pool_free(inst.pkt_pool,pkt);
pkt = request(); CHECK(pkt && pkt->data[1] == 100-count && !a->ack_q->count); memory_pool_free(inst.pkt_pool,pkt);
teardown();
}
/* Усечённый feedback не подтверждает даже корректный entry, не трогает BW и не читает хвост буфера. */
static void receiver_point_truncated(void) {
setup(); b->link_status = 0; enable_models();
struct INFLIGHT_PACKET* packet = enqueue(1000);
struct ETCP_DGRAM* sent = request(); CHECK(sent); memory_pool_free(inst.pkt_pool,sent);
uint32_t seq = packet->seq; uint64_t id = packet->last_attempt_id;
unsigned calls = bbr_calls, size = etcp_ack_section_size(1);
for (unsigned len = ETCP_ACK_BASE_SIZE; len <= size; len++) {
struct ETCP_DGRAM* pkt = memory_pool_alloc(inst.pkt_pool); CHECK(pkt);
pkt->link = a; pkt->data_len = len; memset(pkt->data,0,size); pkt->data[0] = ETCP_SECTION_ACK; pkt->data[1] = 1;
for (unsigned i = 0; i < 4; i++) pkt->data[ETCP_ACK_BASE_SIZE+i] = seq >> (8*i);
for (unsigned i = 0; i < 8; i++) pkt->data[ETCP_ACK_BASE_SIZE+4+i] = id >> (8*i);
pkt->data[ETCP_ACK_BASE_SIZE+ETCP_ACK_ENTRY_SIZE] = 1;
pkt->data[ETCP_ACK_BASE_SIZE+ETCP_ACK_ENTRY_SIZE+8] = 1;
etcp_conn_input(pkt);
if (len < size) CHECK(a->inflight_bytes == 1000 && conn->unacked_bytes == 1000 && bbr_calls == calls);
else CHECK(!a->inflight_bytes && !conn->unacked_bytes && bbr_calls == calls+1);
}
teardown();
setup(); conn->got_initial_pkt = 1; a->rx_data_bytes = UINT64_MAX;
receive_fragment(a,2,1); CHECK(conn->close_requested && !conn->callbacks_running && !a->ack_q->count);
teardown();
}
@@ -850,7 +1011,8 @@ int main(void) {
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, retry_rotation, independent_attempt_acks,
receive_admission_and_path_acks, peer_window_queries, pacing_continuity, pacing_data_units, attempt_memory_limits,
normalizer_rx_oom, normalizer_tx_oom, normalizer_callback_abort, reset_without_allocation, restart_allocation_failures};
normalizer_rx_oom, normalizer_tx_oom, normalizer_callback_abort, reset_without_allocation, restart_allocation_failures,
receiver_bandwidth_bounds, receiver_point_lifecycle, receiver_point_wire, receiver_point_truncated};
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);