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

Verify logical delivery and physical window counters independently

Browse Source
This commit is contained in:
evgeny committed 2026-10-23 10:27:38 +02:00
1 parent 147648ffd0
commit 432017cc22
2 files changed
+39 -21
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/test_etcp_windows.c
+37 -19
View File
@@ -60,15 +60,19 @@ void __wrap_bbr_main(struct bbr* model, const struct bbr_rate_sample* sample, ui
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++; }
}
for (struct ll_entry* e = l->attempts->head; e; e = e->next) { bytes += e->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);
struct ll_queue* logical_queues[] = {conn->input_send_q, conn->input_wait_ack};
for (unsigned i = 0; i < 2; i++) for (struct ll_entry* e = logical_queues[i]->head; e; e = e->next) {
struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)e;
unsigned active = 0;
for (struct ETCP_LINK* link = conn->links; link; link = link->next)
for (struct ll_entry* entry = link->attempts->head; entry; entry = entry->next)
active += ((struct ETCP_TX_ATTEMPT*)entry)->seq == packet->seq;
CHECK(packet->active_attempts == active);
CHECK(packet->state == (i ? INFLIGHT_STATE_WAIT_ACK : INFLIGHT_STATE_WAIT_SEND));
}
CHECK(conn->unacked_bytes == conn->input_send_q->total_bytes + conn->input_wait_ack->total_bytes);
}
@@ -92,6 +96,7 @@ static void setup(void) {
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;
l->rtt_last = 1; CHECK(etcp_tx_link_init(l) == 0);
}
etcp_conn_on_inflight_lim_changed(conn);
}
@@ -121,6 +126,7 @@ static struct INFLIGHT_PACKET* enqueue(unsigned len) {
static struct ETCP_DGRAM* request(void) {
struct ETCP_DGRAM* pkt = etcp_request_pkt(conn);
if (pkt) etcp_tx_complete(pkt, 1); // ручной стенд принимает подготовленный пакет без сетевого send
check_accounting();
return pkt;
}
@@ -136,7 +142,16 @@ static void fill(struct ETCP_LINK* target, unsigned count, unsigned len) {
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(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);
else etcp_ack_recv(conn, seq);
check_accounting();
}
static struct ETCP_TX_ATTEMPT* last_attempt(struct INFLIGHT_PACKET* packet) {
return (struct ETCP_TX_ATTEMPT*)queue_find_data_by_index(packet->last_link->attempts, &packet->last_attempt_id);
}
static void ack_all(void) {
while (conn->input_wait_ack->head) ack(((struct INFLIGHT_PACKET*)conn->input_wait_ack->head)->seq);
}
@@ -168,7 +183,8 @@ static void boundaries_ack_burst(void) {
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);
confirmation->attempt_id = 777;
CHECK(queue_data_put_with_index(a->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);
@@ -229,7 +245,7 @@ static void allocation_and_send_failure(void) {
test_clock_run_for(ua, 20);
CHECK(!sends && conn->input_send_q->count == 1 && !a->inflight_bytes && conn->tx_token && conn->tx_delayed);
fail_send = 1; wait_sends(1);
CHECK(sends == 1 && conn->input_wait_ack->count == 1);
CHECK(sends == 1 && conn->input_send_q->count == 1 && !a->inflight_bytes);
wait_sends(2);
ack_all(); teardown();
}
@@ -247,7 +263,7 @@ static void removed_link_and_bbr(void) {
enqueue(1025); CHECK(!request() && b->cwnd_limited && b->inflight_bytes < b->inflight_lim_bytes);
// Первый ACK относится к отправке при наличии ожидающих данных: она не app_limited.
struct INFLIGHT_PACKET* first = (struct INFLIGHT_PACKET*)conn->input_wait_ack->head;
CHECK(!first->is_app_limited);
CHECK(!last_attempt(first)->is_app_limited);
ack(first->seq); CHECK(bbr_calls == 1 && saw_cwnd_limited && !saw_app_limited);
teardown();
}
@@ -261,12 +277,13 @@ static void shaper_ready(void* arg) {
/* Реальный RTO: нижняя/верхняя границы и влияние RTT/jitter, без подмены обработчика таймера. */
static void retransmission_timeout(void) {
const struct { uint16_t rtt, jitter; uint32_t expected_tb; } cases[] = {
{0, 0, 100}, {20, 10, 100}, {200, 0, 400}, {200, 75, 550},
{0, 0, 10000}, {20, 10, 100}, {200, 0, 400}, {200, 75, 550},
{2000, 100, 4200}, {6000, 1000, 10000}, {UINT16_MAX, UINT16_MAX, 10000}
};
for (unsigned i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) {
setup(); b->link_status = 0;
conn->rtt_avg_10 = cases[i].rtt; conn->jitter = cases[i].jitter;
a->rtt_last = 0; a->data_rtt_us = cases[i].rtt * 100; a->data_jitter_us = cases[i].jitter * 100;
fill(a, 1, 100);
struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)conn->input_wait_ack->head;
uint64_t sent = packet->last_timestamp, deadline = sent + cases[i].expected_tb + 1000;
@@ -301,7 +318,7 @@ static void pacing_reset_close(void) {
/* Смена ставки пересчитывает долг, а не оставляет таймер от прежней скорости. */
static void pacing_rate_changes(void) {
setup(); b->link_status = 0; conn->rtt_avg_10 = 10000;
setup(); b->link_status = 0; a->data_rtt_us = 1000000;
etcp_loadbalancer_set_pacing_rate(a, 100);
enqueue(1000); wait_sends(1); enqueue(500);
uint64_t debt = a->shaper_load_time_tb - get_time_tb();
@@ -311,7 +328,7 @@ static void pacing_rate_changes(void) {
CHECK(a->shaper_load_time_tb <= get_time_tb() + 20);
wait_sends(2); ack_all(); teardown();
setup(); b->link_status = 0; conn->rtt_avg_10 = 10000;
setup(); b->link_status = 0; a->data_rtt_us = 1000000;
etcp_loadbalancer_set_pacing_rate(a, 1000000);
enqueue(1000); wait_sends(1); enqueue(500);
etcp_loadbalancer_set_pacing_rate(a, 100);
@@ -378,9 +395,10 @@ static void ack_without_data_pacing(void) {
uint64_t load = a->shaper_load_time_tb, fraction = a->shaper_sub_nanotime;
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);
confirmation->attempt_id = 777;
CHECK(queue_data_put_with_index(a->ack_q, &confirmation->ll) == 0);
enqueue(100); uasync_drain_immediate(ua);
CHECK(sends == 1 && conn->input_send_q->count == 1 && !conn->ack_q->count);
CHECK(sends == 1 && conn->input_send_q->count == 1 && !a->ack_q->count);
CHECK(a->shaper_timer == timer && a->shaper_load_time_tb == load && a->shaper_sub_nanotime == fraction);
uasync_cancel_timeout(ua, timer); a->shaper_timer = NULL;
etcp_conn_wake_sender(conn); wait_sends(2); ack_all(); teardown();
@@ -388,7 +406,7 @@ static void ack_without_data_pacing(void) {
/* ACK одной пачки приходят в один tick: скорость определяется доставкой всей группы, а не промежутком между ACK. */
static void bbr_delivery_clock(void) {
setup(); b->link_status = 0; conn->rtt_avg_10 = 200;
setup(); b->link_status = 0; a->data_rtt_us = 20000;
a->bbr = u_calloc(1, sizeof(*a->bbr)); CHECK(a->bbr); bbr_init(a->bbr);
fill(a, 8, 1000);
uint64_t sent_tb = get_time_tb();
@@ -407,7 +425,7 @@ static void bbr_delivery_clock(void) {
/* Модель и транспорт используют один DATA MSS и одно окно, в том числе ниже 8KiB. */
static void bbr_probe_window(void) {
setup(); b->link_status = 0; conn->rtt_avg_10 = 10000;
setup(); b->link_status = 0; a->data_rtt_us = 1000000;
etcp_update_mtu(conn);
uint32_t minimum = BBR_CWND_MIN_TARGET * conn->normalizer->frag_size;
a->rtt_last = 100;
@@ -423,7 +441,7 @@ static void bbr_probe_window(void) {
struct INFLIGHT_PACKET* probe = enqueue(1000); CHECK(!request());
ack(2); ack(3); ack(4); CHECK(a->bbr->probe_rtt_done_stamp_tb);
ack(5); // оставшиеся 3000 байт и новый фрагмент целиком помещаются в 4388
struct ETCP_DGRAM* pkt = request(); CHECK(pkt && probe->is_app_limited);
struct ETCP_DGRAM* pkt = request(); CHECK(pkt && last_attempt(probe)->is_app_limited);
memory_pool_free(inst.pkt_pool, pkt);
test_clock_run_for(ua, 10); a->bbr->now_tb += BBR_PROBE_RTT_MS * 10;
ack(probe->seq); CHECK(a->bbr->mode == BBR_PROBE_BW);
tests/test_stream_tx_stats.c
+2 -2
View File
@@ -38,9 +38,9 @@ int main(void) {
a->normalizer->data = memory_pool_alloc(inst->data_pool); CHECK(a->normalizer->data); a->normalizer->data_ptr = 10;
struct etcp_tx_snapshot s; uint64_t now = get_time_us();
CHECK(etcp_tx_get_snapshot(inst, now, &s) == 0 && s.connections == 2 && s.inflight_bytes == 200 && s.queued_bytes == 10);
etcp_ack_recv(a, 1, 0, 0); etcp_ack_recv(a, 1, 0, 0);
etcp_ack_recv(a, 1); etcp_ack_recv(a, 1);
CHECK(inst->tx_history.total == 80 && !a->unacked_bytes);
etcp_ack_recv(b, 2, 0, 0); CHECK(inst->tx_history.total == 200);
etcp_ack_recv(b, 2); CHECK(inst->tx_history.total == 200);
etcp_connection_close(a); etcp_connection_close(b);
CHECK(etcp_tx_get_snapshot(inst, get_time_us() + 100000, &s) == 0 && !s.connections && !s.inflight_bytes && s.acked_1s == 200);
utun_instance_destroy(inst); uasync_destroy(ua, 1); CHECK(u_get_allocated_count() == before);