diff --git a/src/transport_layer/etcp_tx.c b/src/transport_layer/etcp_tx.c index 519dc968..1bf5e154 100644 --- a/src/transport_layer/etcp_tx.c +++ b/src/transport_layer/etcp_tx.c @@ -24,7 +24,8 @@ uint32_t etcp_tx_rto_us(const struct ETCP_LINK* link) { return (uint32_t)rto; } -static void retransmit_arm(struct ETCP_LINK* link) { +/* Применить текущий RTO к старейшей попытке, сохраняя уже назначенную более раннюю проверку. */ +void etcp_tx_update_rto(struct ETCP_LINK* link) { if (!link->attempts || link->etcp->close_requested || link->etcp->state == 2) return; struct ETCP_TX_ATTEMPT* first = (struct ETCP_TX_ATTEMPT*)link->attempts->head; if (!first || !first->sent_at_us) { @@ -107,7 +108,7 @@ static void retransmit_cb(void* arg) { (unsigned long long)(now - attempt->sent_at_us), rto, link->data_rtt_us, link->rtt_last * 100, link->data_jitter_us); attempt_retire(link, attempt, 1, 1); } - retransmit_arm(link); + etcp_tx_update_rto(link); } int etcp_tx_link_init(struct ETCP_LINK* link) { @@ -237,7 +238,7 @@ void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot commit logical transmission", conn->log_name); conn->callbacks_running++; etcp_connection_close(conn); conn->callbacks_running--; return; } - retransmit_arm(link); + etcp_tx_update_rto(link); } void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint32_t receiver_delay_us, @@ -332,7 +333,7 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint link->bbr_pacing_rate, etcp_tx_rto_us(link)); link->acked_bytes += sample.ll.len; link->acked_packets++; link->bbr_loss_since_ack = 0; link->cwnd_limited = 0; - retransmit_arm(link); + etcp_tx_update_rto(link); } else DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] old/foreign ACK seq=%u attempt=%u path=%u: logical only", conn->log_name, seq, (unsigned)(uint16_t)attempt_id, link->local_link_id); etcp_ack_recv(conn, seq); diff --git a/src/transport_layer/etcp_tx.h b/src/transport_layer/etcp_tx.h index 65f62f39..7374a89e 100644 --- a/src/transport_layer/etcp_tx.h +++ b/src/transport_layer/etcp_tx.h @@ -47,6 +47,10 @@ void etcp_tx_link_destroy(struct ETCP_LINK* link); * Timestamp-RTT задаёт нижнюю границу, чтобы рост задержки не удалял все попытки до их ACK и не замораживал DATA-измерения. * Это граница RTO; RTT/min_rtt и BW модели BBR по-прежнему обновляет только ACK действующей DATA-попытки. */ uint32_t etcp_tx_rto_us(const struct ETCP_LINK* link); +/* Пересчитать таймер старейшей попытки после изменения RTT; без попыток/при закрытии действий нет. + * Более раннюю проверку можно сохранить, сокращённый срок назначается сразу. При OOM закрывает conn с ERROR. + * Timestamp handler вызывает после обновления метрик; отправка/ACK/RTO вызывают после изменения attempts. */ +void etcp_tx_update_rto(struct ETCP_LINK* link); /* Резервировать окно для packet; NULL — отказ с логом. OOM/бюджет сохраняют packet в очереди, * исчерпание номера меняет эпоху и удаляет прежний поток, включая packet. * Физические записи ограничены MAX_INFLIGHT_SIZE на линк и 2*MAX_INFLIGHT_SIZE на соединение. diff --git a/tests/test_etcp_windows.c b/tests/test_etcp_windows.c index 33feedfe..491cd513 100644 --- a/tests/test_etcp_windows.c +++ b/tests/test_etcp_windows.c @@ -20,6 +20,7 @@ static struct ETCP_CONN* conn; static struct ETCP_LINK *a, *b; static unsigned sends, fail_alloc, fail_send, fail_tx_timer, bbr_calls, saw_cwnd_limited, saw_app_limited; static unsigned fail_shaper_timer; +static unsigned fail_rto_timer; static unsigned pacer_clock_active, capture_shaper, shaper_arm_count, pacer_ready_count, pacer_ready_action; static uint64_t pacer_now_tb; static int captured_shaper_delay; @@ -60,6 +61,7 @@ void* __wrap_uasync_call_soon(struct UASYNC* loop, void* arg, timeout_callback_t void* __real_uasync_set_timeout(struct UASYNC*, int, void*, timeout_callback_t, const char*); void* __wrap_uasync_set_timeout(struct UASYNC* loop, int delay, void* arg, timeout_callback_t callback, const char* name) { + if (fail_rto_timer && !strcmp(name,"etcp_attempt_rto")) { fail_rto_timer--; return NULL; } if (capture_ack_timer && !strcmp(name, "etcp_ack_resp")) { captured_ack_callback = callback; captured_ack_arg = arg; } if (fail_ack_timer && !strcmp(name, "etcp_ack_resp")) { fail_ack_timer--; return NULL; } if (capture_shaper && name && !strcmp(name, "etcp_shaper")) { @@ -156,6 +158,7 @@ static void setup(void) { memset(&inst, 0, sizeof(inst)); memset(&cfg, 0, sizeof(cfg)); sends = fail_alloc = fail_send = fail_tx_timer = bbr_calls = saw_cwnd_limited = saw_app_limited = 0; fail_shaper_timer = 0; + fail_rto_timer = 0; pacer_clock_active = capture_shaper = shaper_arm_count = pacer_ready_count = pacer_ready_action = 0; captured_shaper_delay = 0; captured_shaper_callback = NULL; captured_shaper_arg = NULL; fail_rx_data = fail_rx_fragment = fail_ack_entry = 0; @@ -767,6 +770,32 @@ static void rto_path_rtt_growth(void) { teardown(); } +/* Timestamp сокращает срок без новых DATA/ACK; OOM пересчёта безопасно закрывает conn внутри RX. */ +static void rto_timestamp_rearm(void) { + for (unsigned oom = 0; oom < 2; oom++) { + setup(); b->link_status = 0; a->data_rtt_us = 20000; a->rtt_last = 1000; + ack_now_us = get_time_us(); ack_clock_active = 1; + fill(a,1,1000); + uint64_t old_deadline = a->retrans_deadline_us; + CHECK(old_deadline == ack_now_us+123000); + struct ETCP_DGRAM* pkt = memory_pool_alloc(inst.pkt_pool); CHECK(pkt); + pkt->link = a; pkt->data_len = 5; memset(pkt->data,0,5); pkt->data[0] = ETCP_SECTION_TIMESTAMP; + uint16_t ret = get_current_timestamp()-200; + pkt->data[1] = ret; pkt->data[2] = ret>>8; + fail_rto_timer = oom; + etcp_conn_input(pkt); + if (oom) CHECK(conn->close_requested && !conn->callbacks_running && !fail_rto_timer); + else { + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Timestamp RTO rearm: RTO_us=%u old_deadline=%llu new_deadline=%llu expected=%llu", + etcp_tx_rto_us(a),(unsigned long long)old_deadline,(unsigned long long)a->retrans_deadline_us, + (unsigned long long)(ack_now_us+27000)); + CHECK(a->rtt_last == 200 && a->retrans_deadline_us == ack_now_us+27000); + CHECK(!a->acked_packets && !bbr_calls && a->inflight_packets == 1); + } + teardown(); + } +} + static void receive_till(struct ETCP_LINK* link, uint32_t till, int query) { struct ETCP_DGRAM* pkt = memory_pool_alloc(inst.pkt_pool); CHECK(pkt); pkt->link = link; pkt->data_len = ETCP_ACK_BASE_SIZE + query; @@ -1772,7 +1801,8 @@ int main(void) { void (*cases[])(void) = {selection, boundaries_ack_burst, late_ack_and_migration, lost_ack_and_shrink, 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, - rto_path_rtt_growth, retransmission_timeout, pacing_rate_changes, pacing_timer_failure, retry_rotation, independent_attempt_acks, + rto_path_rtt_growth, rto_timestamp_rearm, retransmission_timeout, pacing_rate_changes, pacing_timer_failure, + retry_rotation, independent_attempt_acks, retry_waits_for_alternative, retry_alternative_lifecycle, send_queue_scan, send_queue_input_admission, retry_full_round_history, retry_history_slot_reuse, fast_stall_progress, stalled_retry_selection, input_admission_callbacks,