From 4bf47ca35729a75b3a01650e3556dc21e764c136 Mon Sep 17 00:00:00 2001 From: evgeny Date: Fri, 23 Oct 2026 16:39:12 +0200 Subject: [PATCH] Exercise receiver bandwidth bounds and ACK progress framing --- tests/test_etcp_multipath.c | 2 +- tests/test_etcp_session.c | 2 +- tests/test_etcp_windows.c | 178 ++++++++++++++++++++++++++++++++++-- 3 files changed, 172 insertions(+), 10 deletions(-) diff --git a/tests/test_etcp_multipath.c b/tests/test_etcp_multipath.c index ad483bc5..0c450373 100644 --- a/tests/test_etcp_multipath.c +++ b/tests/test_etcp_multipath.c @@ -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; diff --git a/tests/test_etcp_session.c b/tests/test_etcp_session.c index 19cb9cf2..f0d51b1a 100644 --- a/tests/test_etcp_session.c +++ b/tests/test_etcp_session.c @@ -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); } diff --git a/tests/test_etcp_windows.c b/tests/test_etcp_windows.c index 2b281cd9..02623e92 100644 --- a/tests/test_etcp_windows.c +++ b/tests/test_etcp_windows.c @@ -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);