diff --git a/tests/test_etcp_windows.c b/tests/test_etcp_windows.c index dfa7aa6f..50eae40b 100644 --- a/tests/test_etcp_windows.c +++ b/tests/test_etcp_windows.c @@ -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); diff --git a/tests/test_stream_tx_stats.c b/tests/test_stream_tx_stats.c index a3e9cc1d..b374ba19 100644 --- a/tests/test_stream_tx_stats.c +++ b/tests/test_stream_tx_stats.c @@ -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);