Schedule ordinary ETCP sends without timer latency

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);
 }
 

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)

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();
 }