diff --git a/src/routing_layer/etcp_router.c b/src/routing_layer/etcp_router.c index 016e1e6f..e2c8e81d 100644 --- a/src/routing_layer/etcp_router.c +++ b/src/routing_layer/etcp_router.c @@ -484,6 +484,7 @@ static void router_send_queue_free(void* arg) { queue_free(arg); } // Отменить таймеры/waiter и освободить все очереди rconn (общий код для close/restart). static void router_conn_free_queues(struct ETCP_ROUTER_CONN* rconn) { struct UTUN_INSTANCE* inst = rconn->inst; + if (rconn->send_request) etcp_router_cancel_request(rconn->send_request); if (rconn->receive_pump) { uasync_call_soon_cancel(inst->ua, rconn->receive_pump); rconn->receive_pump = NULL; } if (rconn->handshake_timer) { uasync_cancel_timeout(inst->ua, rconn->handshake_timer); rconn->handshake_timer = NULL; } if (rconn->ack_timer) { uasync_cancel_timeout(inst->ua, rconn->ack_timer); rconn->ack_timer = NULL; } @@ -594,16 +595,16 @@ static void router_set_no_route(struct ETCP_ROUTER_CONN* rconn) { // Попытаться продвинуть отправку: зарегистрировать waiter на send_input_q, если есть что слать. // Инвариант: send_q непуст ∧ !closed ∧ !no_route ∧ inflight свободен ⇒ waiter зарегистрирован. static void router_send_kick(struct ETCP_ROUTER_CONN* rconn) { - if (rconn->closed || rconn->no_route) return; + if (rconn->closed || rconn->no_route || rconn->send_producing) return; if (!rconn->peer_reset_id) { - if (queue_entry_count(rconn->send_q)) { + if (queue_entry_count(rconn->send_q) || rconn->send_request) { router_handshake_start(rconn); } return; } int retry = rconn->retrans_seq != rconn->retrans_end; if (rconn->peer_receive_blocked && !retry) return; - if (!retry && queue_entry_count(rconn->send_q) == 0) return; + if (!retry && queue_entry_count(rconn->send_q) == 0 && !rconn->send_request) return; if (!retry && queue_entry_count(rconn->inflight_q) >= (int)router_effective_max_inflight(rconn)) { rconn->send_blocked = 1; return; @@ -643,6 +644,23 @@ static void router_send_drain_cb(struct ll_queue* q, void* arg) { rconn->send_blocked = 1; return; } + if (!rconn->send_q->head && rconn->send_request) { + struct router_send_request *h = rconn->send_request; + void (*ready)(struct ETCP_ROUTER_CONN*, void*) = h->ready; + void *context = h->arg; + uint64_t local = rconn->reset_id, peer = rconn->peer_reset_id; + DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router producer ready svc=%u waited_us=%llu inflight=%d/%u", + rconn->svc_id, (unsigned long long)(get_time_us() - h->since_us), + queue_entry_count(rconn->inflight_q), router_effective_max_inflight(rconn)); + rconn->send_request = NULL; memset(h, 0, sizeof(*h)); + rconn->send_producing = 1; + ready(rconn, context); + rconn->send_producing = 0; + if (rconn->closed) return; + if (rconn->reset_id != local || rconn->peer_reset_id != peer || router_send_conn(rconn) != conn) { + router_send_kick(rconn); return; + } + } struct ll_entry* pending = rconn->send_q->head; if (!pending) { rconn->send_blocked = 0; router_send_watchdog_disarm(rconn); return; } struct ll_entry* e = queue_entry_new(0); @@ -697,17 +715,17 @@ static void router_send_watchdog_cb(void* arg) { struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; rconn->watchdog_timer = NULL; if (rconn->closed) return; - if (queue_entry_count(rconn->send_q) == 0) return; + if (queue_entry_count(rconn->send_q) == 0 && !rconn->send_request) return; // Маршрут мог смениться, пока waiter остаётся на занятой очереди прежнего next hop. if (rconn->peer_reset_id && !rconn->no_route) router_send_kick(rconn); - if (queue_entry_count(rconn->send_q) > 0 && !rconn->watchdog_timer) + if ((queue_entry_count(rconn->send_q) > 0 || rconn->send_request) && !rconn->watchdog_timer) rconn->watchdog_timer = uasync_set_timeout(rconn->inst->ua, ROUTER_SEND_WATCHDOG_TB, rconn, router_send_watchdog_cb, "router_send_watchdog"); } // Взвести watchdog пока send_q непуст (страховка от пропущенного re-arm). static void router_send_watchdog_arm(struct ETCP_ROUTER_CONN* rconn) { - if (rconn->watchdog_timer || queue_entry_count(rconn->send_q) == 0) return; + if (rconn->watchdog_timer || (queue_entry_count(rconn->send_q) == 0 && !rconn->send_request)) return; rconn->watchdog_timer = uasync_set_timeout(rconn->inst->ua, ROUTER_SEND_WATCHDOG_TB, rconn, router_send_watchdog_cb, "router_send_watchdog"); } @@ -910,7 +928,7 @@ static void router_update_inflight_limit(struct ETCP_ROUTER_CONN* rconn) { if (new_limit < old && inflight >= new_limit) { if (!rconn->send_blocked) rconn->send_blocked = 1; } - if (new_limit > old && (rconn->send_blocked || queue_entry_count(rconn->send_q) > 0)) { + if (new_limit > old && (rconn->send_blocked || queue_entry_count(rconn->send_q) > 0 || rconn->send_request)) { rconn->send_blocked = 0; router_send_kick(rconn); } @@ -1680,6 +1698,31 @@ int etcp_router_tx_stats(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t pe return 0; } +void etcp_router_cancel_request(struct router_send_request* h) { + if (!h || !h->conn) return; + struct ETCP_ROUTER_CONN *rconn = h->conn; + DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router producer cancelled svc=%u waited_us=%llu", + rconn->svc_id, (unsigned long long)(get_time_us() - h->since_us)); + rconn->send_request = NULL; memset(h, 0, sizeof(*h)); + if (!rconn->send_q->head && rconn->retrans_seq == rconn->retrans_end) router_cancel_send_waiter(rconn); + if (!rconn->send_q->head) router_send_watchdog_disarm(rconn); +} + +int etcp_router_request_send(struct ETCP_ROUTER_CONN* rconn, struct router_send_request* h, + void (*ready)(struct ETCP_ROUTER_CONN*, void*), void* arg) { + if (!rconn || rconn->closed || !h || !ready) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router producer registration: missing/closed channel, handle or callback"); return -1; + } + if ((h->conn && h->conn != rconn) || (rconn->send_request && rconn->send_request != h) + || (h->conn && (h->ready != ready || h->arg != arg))) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router producer registration conflicts svc=%u", rconn->svc_id); return -1; + } + if (h->conn) return 0; + *h = (struct router_send_request){rconn, ready, arg, get_time_us()}; rconn->send_request = h; + router_send_kick(rconn); router_send_watchdog_arm(rconn); + return 0; +} + // Backpressure: зарегистрировать waiter на send_q для (group_id, node_id, svc_id). void etcp_router_on_send_ready(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id, uint8_t svc_id, struct queue_waiter_handle* h, diff --git a/src/routing_layer/etcp_router.h b/src/routing_layer/etcp_router.h index 14d3835e..c1d911fd 100644 --- a/src/routing_layer/etcp_router.h +++ b/src/routing_layer/etcp_router.h @@ -87,6 +87,16 @@ struct SVC_ROUTE_HDR { #define TRANSIT_QUEUE_HASH 256 +/* Одноразовый запрос слота отправки. Обнулить перед использованием; живёт до callback/отмены. + * Только uasync. Router снимает связь до callback; close/reset автоматически отменяет запрос. */ +struct ETCP_ROUTER_CONN; +struct router_send_request { + struct ETCP_ROUTER_CONN *conn; /* Владелец регистрации, NULL после снятия. */ + void (*ready)(struct ETCP_ROUTER_CONN*, void*); /* Формирует не более одной кодограммы через conn_send. */ + void *arg; /* Заимствованный контекст сервиса. */ + uint64_t since_us; /* Начало ожидания слота, без сброса при повторной регистрации. */ +}; + // Очередь транзитных пакетов для пары (group_id, src_node_id, dst_node_id) // Создаётся когда не получается отправить пакет в ETCP напрямую, // удаляется когда очередь становится пустой. @@ -140,6 +150,8 @@ struct ETCP_ROUTER_CONN { void* idle_ack_timer; // idle таймер (дослать последний ack) struct ll_queue* send_q; // очередь ожидающих отправки (inflight полон) + struct router_send_request *send_request; /* Ленивый producer: payload ещё принадлежит сервису. */ + uint8_t send_producing; /* Запрет рекурсивного kick внутри ready callback. */ struct queue_waiter_handle send_waiter; // waiter на send_input_q (по 1 пакету, round-robin) struct ll_queue* send_waiter_q; // очередь регистрации, не результат повторного поиска маршрута uint32_t retrans_seq; // следующий seq текущего прохода ретрансмитов @@ -322,6 +334,17 @@ void etcp_router_conn_close(struct ETCP_ROUTER_CONN* rconn); // В minRTT probe лимит равен 4; reset сессии возвращает default=256. void etcp_router_set_max_inflight(struct ETCP_ROUTER_CONN* rconn, uint16_t new_max); +/* Запросить готовность handshake/маршрута/окна/ETCP при пустой send_q. 0/-1 с логом. + * Callback всегда отложен, разделяет round-robin с обычными очередями, может отменить/перезапустить канал. + * Один запрос на канал; повтор того же запроса не меняет место в очереди. В callback можно поставить + * одну кодограмму через conn_send и зарегистрировать следующий запрос. Вызов conn_send в этом callback + * только принимает копию: не вызывает сервисы синхронно. Нельзя сохранять готовность после возврата. + * h и arg должны жить до callback/отмены; reset/close обнуляет h->conn без вызова ready. */ +int etcp_router_request_send(struct ETCP_ROUTER_CONN* rconn, struct router_send_request* h, + void (*ready)(struct ETCP_ROUTER_CONN*, void*), void* arg); +/* Отменить исходную регистрацию; обнулённый h допустим, callback после возврата не придёт. */ +void etcp_router_cancel_request(struct router_send_request* h); + // Дождаться порога send_q. h должен быть обнулён и жить до callback/отмены. // Callback может выполниться внутри вызова. Повторная регистрация отменяет прежнюю. // Если канала нет, ожидание не регистрируется. Перед освобождением arg отменить waiter.