diff --git a/src/transport_layer/etcp.c b/src/transport_layer/etcp.c index 0499953c..97c29cc9 100644 --- a/src/transport_layer/etcp.c +++ b/src/transport_layer/etcp.c @@ -56,14 +56,16 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* etcp); /* Очереди и сетевые события только планируют проход: синхронные callbacks не входят в отправитель повторно. */ static void etcp_schedule_send(struct ETCP_CONN* conn, int delay_tb) { - if (!conn || conn->close_requested || conn->state == 2 || conn->tx_timer) return; + if (!conn || conn->close_requested || conn->state == 2 || conn->tx_token) return; if (!conn->initialized || conn->session_required || conn->reinit_pending) return; - if (conn->tx_running && delay_tb == 1) return; + if (conn->tx_running && !delay_tb) return; int work = conn->input_send_q->count || conn->input_queue->count || conn->ack_q->count; for (struct ETCP_LINK* link = conn->links; !work && link; link = link->next) work = link->burst_active; if (!work) return; - conn->tx_timer = uasync_set_timeout(conn->instance->ua, delay_tb, conn, etcp_send_timer_cb, "etcp_tx"); - if (!conn->tx_timer) { + conn->tx_delayed = delay_tb != 0; + conn->tx_token = delay_tb ? uasync_set_timeout(conn->instance->ua, delay_tb, conn, etcp_send_timer_cb, "etcp_tx") : + uasync_call_soon(conn->instance->ua, conn, etcp_send_timer_cb); + if (!conn->tx_token) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot schedule sender delay=%d", conn->log_name, delay_tb); // Вызывающий может ещё собирать линк/менять очередь: ресурсы освобождаются после выхода из его стека. conn->close_requested = 1; @@ -75,7 +77,15 @@ static void etcp_schedule_send(struct ETCP_CONN* conn, int delay_tb) { } } -void etcp_conn_wake_sender(struct ETCP_CONN* conn) { etcp_schedule_send(conn, 1); } +void etcp_conn_wake_sender(struct ETCP_CONN* conn) { etcp_schedule_send(conn, 0); } + +/* Обычный запуск и OOM retry имеют разные операции отмены, но одновременно существует только один token. */ +static void etcp_cancel_send(struct ETCP_CONN* conn) { + if (!conn->tx_token) return; + if (conn->tx_delayed) uasync_cancel_timeout(conn->instance->ua, conn->tx_token); + else uasync_call_soon_cancel(conn->instance->ua, conn->tx_token); + conn->tx_token = NULL; +} /* Учитывается только последняя отправленная попытка, ещё не признанная потерянной по RTO. */ static void etcp_inflight_release(struct INFLIGHT_PACKET* pkt) { @@ -518,7 +528,7 @@ void etcp_connection_close(struct ETCP_CONN* etcp) { callring_put(CR_CCLOSE_IN, (uintptr_t)etcp, 0); etcp->close_requested = 1; - if (etcp->tx_timer) { uasync_cancel_timeout(etcp->instance->ua, etcp->tx_timer); etcp->tx_timer = NULL; } + etcp_cancel_send(etcp); etcp_down_timer_cancel(etcp); etcp_padding_stop(&etcp->padding); if (etcp->callbacks_running) { @@ -609,7 +619,7 @@ void etcp_conn_reset(struct ETCP_CONN* etcp) { } // Reset IDs DEBUG_WARN(DEBUG_CATEGORY_ETCP, "Resetting ETCP instance [%s]", etcp->log_name); - if (etcp->tx_timer) { uasync_cancel_timeout(etcp->instance->ua, etcp->tx_timer); etcp->tx_timer = NULL; } + etcp_cancel_send(etcp); etcp->next_tx_id = 1; etcp->last_rx_id = 0; etcp->last_delivered_id = 0; @@ -1234,7 +1244,7 @@ static void etcp_conn_process_send_queue(struct ETCP_CONN* etcp) {// вызыв /* Один запуск обслуживает накопленные события; заблокированный DATA ждёт ACK/RTO/shaper, а не следующий poll. */ static void etcp_send_timer_cb(void* arg) { struct ETCP_CONN* conn = arg; - conn->tx_timer = NULL; + conn->tx_token = NULL; etcp_conn_process_send_queue(conn); } diff --git a/src/transport_layer/etcp.h b/src/transport_layer/etcp.h index 3daaf683..a2a5826b 100644 --- a/src/transport_layer/etcp.h +++ b/src/transport_layer/etcp.h @@ -150,7 +150,8 @@ struct ETCP_CONN { struct ll_queue* recv_q; // очередь на сборку фрагментированных пакетов(rx_pool -> struct ETCP_FRAGMENT) + index [+0 SEQ] void (*link_ready_for_send_fn)(struct ETCP_CONN*);// функцию которую должен вызвать драйвер линка при готовности линка принимать данные - void* tx_timer; // единственный отложенный запуск отправителя, включая retry после OOM + void* tx_token; // единственный call_soon либо retry-таймер после OOM + uint8_t tx_delayed; // 1 — tx_token отменяется как timeout; 0 — как call_soon uint8_t tx_running; // объединяет пробуждения внутри прохода, запрещает рекурсивную отправку struct ll_queue* output_queue; // Assembled outgoing packets (storage: ETCP_FRAGMENT / rx_pool) diff --git a/tests/test_etcp_windows.c b/tests/test_etcp_windows.c index cacfe6af..cf8f2086 100644 --- a/tests/test_etcp_windows.c +++ b/tests/test_etcp_windows.c @@ -18,6 +18,12 @@ 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; +void* __real_uasync_call_soon(struct UASYNC*, void*, timeout_callback_t); +void* __wrap_uasync_call_soon(struct UASYNC* loop, void* arg, timeout_callback_t callback) { + if (fail_tx_timer && arg == conn) { fail_tx_timer--; return NULL; } + return __real_uasync_call_soon(loop, arg, callback); +} + 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_tx_timer && !strcmp(name, "etcp_tx")) { fail_tx_timer--; return NULL; } @@ -214,7 +220,7 @@ static void grow_and_session(void) { static void allocation_and_send_failure(void) { setup(); b->link_status = 0; enqueue(100); fail_alloc = 1; test_clock_run_for(ua, 20); - CHECK(!sends && conn->input_send_q->count == 1 && !a->inflight_bytes && conn->tx_timer); + 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); wait_sends(2); @@ -250,13 +256,13 @@ static void pacing_reset_close(void) { a->shaper_timer = uasync_set_timeout(ua, 20, a, shaper_ready, "test_shaper"); CHECK(a->shaper_timer); enqueue(100); CHECK(!request()); wait_sends(1); ack_all(); - enqueue(100); CHECK(conn->tx_timer); + enqueue(100); CHECK(conn->tx_token); etcp_conn_reinit_id(conn, "queued sender reset", conn->reset_id); - CHECK(!conn->tx_timer && !conn->unacked_bytes && conn->reinit_pending); + CHECK(!conn->tx_token && !conn->unacked_bytes && conn->reinit_pending); etcp_conn_ready(conn); - enqueue(100); CHECK(conn->tx_timer); + enqueue(100); CHECK(conn->tx_token); CHECK(etcp_conn_ref_take(conn) == 0); - etcp_connection_close(conn); CHECK(conn->state == 2 && !conn->tx_timer); + etcp_connection_close(conn); CHECK(conn->state == 2 && !conn->tx_token); etcp_conn_ref_free(conn); teardown(); } @@ -285,11 +291,11 @@ static void config_limits(void) { /* Отказ TX-таймера закрывает транспорт после выхода из вызывающего стека, без UAF очереди/линков. */ static void timer_failure(void) { - setup(); CHECK(!conn->tx_timer && etcp_conn_ref_take(conn) == 0); + setup(); CHECK(!conn->tx_token && etcp_conn_ref_take(conn) == 0); fail_tx_timer = 1; enqueue(100); CHECK(conn->close_requested && conn->close_token && conn->state != 2 && conn->links); uasync_drain_immediate(ua); - CHECK(conn->state == 2 && !conn->tx_timer && !conn->links); + CHECK(conn->state == 2 && !conn->tx_token && !conn->links); etcp_conn_ref_free(conn); teardown(); }