diff --git a/src/transport_layer/etcp_tx.h b/src/transport_layer/etcp_tx.h index 11e8f4d8..d99c93c9 100644 --- a/src/transport_layer/etcp_tx.h +++ b/src/transport_layer/etcp_tx.h @@ -35,7 +35,9 @@ void etcp_tx_link_reset(struct ETCP_LINK* link, int retry); void etcp_tx_link_destroy(struct ETCP_LINK* link); /* Расчётный RTO в микросекундах: 2*(собственный RTT+jitter), 10мс..1с; до DATA ACK использует KA RTT/1с. */ uint32_t etcp_tx_rto_us(const struct ETCP_LINK* link); -/* Резервировать окно для packet; NULL — ошибка с логом, packet остаётся в очереди отправки. */ +/* Резервировать окно для packet; NULL — ошибка с логом, packet остаётся в очереди отправки. + * Физические записи ограничены MAX_INFLIGHT_SIZE на линк и 2*MAX_INFLIGHT_SIZE на соединение. + * Исчерпание бюджета блокирует выбор DATA до ACK/RTO/закрытия; ACK-очередь линка ограничена MAX_INFLIGHT_SIZE. */ struct ETCP_TX_ATTEMPT* etcp_tx_prepare(struct ETCP_LINK* link, struct INFLIGHT_PACKET* packet); /* Завершить синхронный send: success=1 принят транспортом, 0 — снять резерв без сетевой потери. * Вызвать ровно один раз до освобождения dgram и до следующего poll; dgram->attempt обнуляется. */ diff --git a/tests/test_etcp_windows.c b/tests/test_etcp_windows.c index 996740ae..1f5ab311 100644 --- a/tests/test_etcp_windows.c +++ b/tests/test_etcp_windows.c @@ -65,11 +65,14 @@ void __wrap_bbr_main(struct bbr* model, const struct bbr_rate_sample* sample, ui /* Счётчики сверяются с содержимым очередей, а не с ожидаемыми результатами самого алгоритма. */ static void check_accounting(void) { + uint32_t attempts = 0; for (struct ETCP_LINK* l = conn->links; l; l = l->next) { uint32_t bytes = 0, packets = 0; 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); + attempts += packets; } + CHECK(conn->active_attempts == attempts); 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; @@ -387,6 +390,36 @@ static void pacing_data_units(void) { teardown(); } +/* Физические копии уже доставленных seq также занимают бюджет, пока не придёт собственный ACK. */ +static void attempt_memory_limits(void) { + setup(); + a->inflight_lim_bytes = b->inflight_lim_bytes = MAX_INFLIGHT_SIZE * 4; + for (unsigned path = 0; path < 2; path++) { + struct ETCP_LINK* link = path ? b : a; + for (unsigned i = 0; i < MAX_INFLIGHT_SIZE; i++) { + struct INFLIGHT_PACKET* packet = enqueue(1); + struct ETCP_TX_ATTEMPT* attempt = etcp_tx_prepare(link, packet); CHECK(attempt); + struct ETCP_DGRAM pkt = { .link = link, .attempt = attempt }; + etcp_tx_complete(&pkt, 1); etcp_ack_recv(conn, packet->seq); + } + CHECK(link->attempts->count == MAX_INFLIGHT_SIZE); + } + CHECK(conn->active_attempts == 2 * MAX_INFLIGHT_SIZE); + struct INFLIGHT_PACKET* pending = enqueue(1); + 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); + 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); + CHECK(etcp_tx_queue_ack(a, MAX_INFLIGHT_SIZE+1, MAX_INFLIGHT_SIZE+1) < 0); + CHECK(a->ack_q->count == MAX_INFLIGHT_SIZE); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Attempt limits: physical=%u path=%d ACK=%d ready_seq=%u", + conn->active_attempts, a->attempts->count, a->ack_q->count, pending->seq); + teardown(); +} + /* Отправитель не превращает pending transport в UP до первого INIT/подтверждения сессии. */ static void initial_readiness(void) { setup(); conn->state = 0; conn->initialized = conn->links_up = conn->tx_state = 0; @@ -639,7 +672,7 @@ int main(void) { 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, retry_rotation, independent_attempt_acks, - receive_admission_and_path_acks, peer_window_queries, pacing_continuity, pacing_data_units}; + receive_admission_and_path_acks, peer_window_queries, pacing_continuity, pacing_data_units, attempt_memory_limits}; 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);