From bbf39c091df7b00ec50835c2f63f97ecbf522650 Mon Sep 17 00:00:00 2001 From: Evgeny Date: Wed, 3 Jun 2026 18:51:12 +0300 Subject: [PATCH] feat: send_q backpressure + ll_queue on_get callback + consumer_ack flow control MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit etcp_router: ROUTER_MAX_SEND_Q_PACKETS=256, router_enqueue_send returns error when send_q full, etcp_route_send propagates -1 on backpressure. ll_queue: queue_set_on_get — deferred callback after each queue_data_get. Replaces manual consumer_ack polling with queue-driven notification. tcp_proxy_server: consumer_ack=1, write_on_get_cb triggers ACK on each write_queue dequeue. Full backpressure chain: write_queue full → no dequeue → no ACK → sender stops → all queues pause write_queue drains → dequeue → on_get → ACK → sender resumes --- lib/ll_queue.c | 17 +++++++++++++++++ lib/ll_queue.h | 11 +++++++++++ src/etcp_router.c | 20 +++++++++++++------- src/etcp_router.h | 1 + src/proxy/tcp_proxy_server.c | 30 +++++++++++++++++++++++++++--- 5 files changed, 69 insertions(+), 10 deletions(-) diff --git a/lib/ll_queue.c b/lib/ll_queue.c index f362432f..5cfb9bb4 100644 --- a/lib/ll_queue.c +++ b/lib/ll_queue.c @@ -47,6 +47,7 @@ static inline void queue_check_thread(struct ll_queue* q) { static void queue_resume_timeout_cb(void* arg); static void check_waiters(struct ll_queue* q); static void empty_trampoline(void* arg); +static void on_get_trampoline(void* arg); static void empty_trampoline(void* arg) { struct ll_queue* q = (struct ll_queue*)arg; @@ -58,6 +59,11 @@ static void empty_trampoline(void* arg) { if (q->count == 0 && cb) cb(q, cb_arg); } +static void on_get_trampoline(void* arg) { + struct ll_queue* q = (struct ll_queue*)arg; + if (q->on_get) q->on_get(q, q->on_get_arg); +} + // ==================== Управление очередью ==================== struct ll_queue* queue_new(struct UASYNC* ua, size_t hash_size, uint16_t index_offset, uint16_t index_size, char* name) { @@ -151,6 +157,8 @@ void queue_free(struct ll_queue* q) { if (old_eid) uasync_call_soon_cancel(q->ua, old_eid); q->empty_callback = NULL; q->empty_callback_arg = NULL; + q->on_get = NULL; + q->on_get_arg = NULL; u_free(q); } @@ -552,6 +560,9 @@ struct ll_entry* queue_data_get(struct ll_queue* q) { if (q->count == 0 && q->empty_callback && !q->empty_call_soon_id) q->empty_call_soon_id = uasync_call_soon(q->ua, q, (timeout_callback_t)empty_trampoline); + if (q->on_get) + uasync_call_soon(q->ua, q, (timeout_callback_t)on_get_trampoline); + #ifdef QUEUE_DEBUG queue_check_consistency(q);// !!!! for debug #endif @@ -644,6 +655,12 @@ void queue_set_empty_callback(struct ll_queue* q, queue_callback_fn cbk_fn, void q->empty_call_soon_id = uasync_call_soon(q->ua, q, (timeout_callback_t)empty_trampoline); } +void queue_set_on_get(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg) { + if (!q) return; + q->on_get = cbk_fn; + q->on_get_arg = arg; +} + int queue_waiter_wait(struct ll_queue* q, struct queue_waiter_handle* h, queue_threshold_callback_fn callback, void* arg) { if (!q || !h) return -1; diff --git a/lib/ll_queue.h b/lib/ll_queue.h index 23f06cd0..8ba2eb35 100644 --- a/lib/ll_queue.h +++ b/lib/ll_queue.h @@ -148,6 +148,9 @@ struct ll_queue { void* empty_callback_arg; void* empty_call_soon_id; // handle от uasync_call_soon + queue_callback_fn on_get; // коллбэк после каждого queue_data_get (deferred) + void* on_get_arg; + struct ll_entry** hash_table; // Хеш-таблица для поиска по id (если hash_size > 0) size_t hash_size; // Размер хеш-таблицы uint16_t index_offset; // Смещение индекса в data[] (для всех entry очереди) @@ -243,6 +246,14 @@ void queue_set_waiter_defer(struct ll_queue* q, int enable); */ void queue_set_empty_callback(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg); +/** + * @brief Устанавливает коллбэк вызываемый после каждого queue_data_get (deferred через call_soon). + * @param q очередь + * @param cbk_fn коллбэк (NULL для отмены) + * @param arg пользовательский аргумент + */ +void queue_set_on_get(struct ll_queue* q, queue_callback_fn cbk_fn, void* arg); + /** * @brief Регистрирует ожидание освобождения очереди до общего порога. * @param q очередь diff --git a/src/etcp_router.c b/src/etcp_router.c index 4e5426ff..6c6c7c44 100644 --- a/src/etcp_router.c +++ b/src/etcp_router.c @@ -171,11 +171,16 @@ static void router_send_resume_cb(void* arg) { router_drain_send_q(rconn); } -static void router_enqueue_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len) { +static int router_enqueue_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t pl_len) { + if (queue_entry_count(rconn->send_q) >= ROUTER_MAX_SEND_Q_PACKETS) { + DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_send_q: FULL svc_id=%u count=%d — backpressure", + rconn->svc_id, queue_entry_count(rconn->send_q)); + return -1; + } struct ll_entry* qe = queue_entry_new(0); - if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_enqueue_send: queue_entry_new failed"); return; } + if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_enqueue_send: queue_entry_new failed"); return -1; } qe->dgram = u_malloc(pl_len); - if (!qe->dgram) { queue_entry_free(qe); return; } + if (!qe->dgram) { queue_entry_free(qe); return -1; } qe->len = pl_len; if (pl_len > 0) memcpy(qe->dgram, payload, pl_len); DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router_send_q: queued svc_id=%u inflight=%d send_q=%d", @@ -185,6 +190,7 @@ static void router_enqueue_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* p if (!rconn->send_resume_timer) rconn->send_resume_timer = uasync_set_timeout(rconn->inst->ua, ROUTER_SEND_RESUME_TB, rconn, router_send_resume_cb, "router_send_resume"); + return 0; } // ==================================================================== @@ -513,9 +519,9 @@ int etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t dst_node_id, struct ll_ if (!rconn) { queue_dgram_free(entry); queue_entry_free(entry); return -1; } if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= ROUTER_MAX_INFLIGHT) { - router_enqueue_send(rconn, entry->dgram + 1, payload_len); + int eq_ret = router_enqueue_send(rconn, entry->dgram + 1, payload_len); queue_dgram_free(entry); queue_entry_free(entry); - return 0; + return eq_ret; } int ret = router_send_one(rconn, entry->dgram + 1, payload_len); @@ -585,6 +591,7 @@ struct ETCP_ROUTER_CONN* etcp_router_conn_get(struct UTUN_INSTANCE* inst, if (!rconn->recv_q) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: queue_new(recv_q) failed"); queue_entry_free(&rconn->ll); return NULL; } rconn->send_q = queue_new(inst->ua, 0, 0, 0, "router_send_q"); if (!rconn->send_q) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: queue_new(send_q) failed"); queue_free(rconn->recv_q); queue_entry_free(&rconn->ll); return NULL; } + queue_set_threshold(rconn->send_q, ROUTER_MAX_SEND_Q_PACKETS / 2, 0); DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_conn: new conn remote=%016llx svc_id=%u", (unsigned long long)remote_node_id, svc_id); queue_data_put_with_index(inst->router_conns, &rconn->ll); @@ -599,8 +606,7 @@ int etcp_router_conn_send(struct ETCP_ROUTER_CONN* rconn, return -1; } if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= ROUTER_MAX_INFLIGHT) { - router_enqueue_send(rconn, data, len); - return 0; + return router_enqueue_send(rconn, data, len); } return router_send_one(rconn, data, len); } diff --git a/src/etcp_router.h b/src/etcp_router.h index 7645a1f3..1047c2e1 100644 --- a/src/etcp_router.h +++ b/src/etcp_router.h @@ -53,6 +53,7 @@ struct ETCP_ROUTER_CONN { #define ROUTER_ACK_INTERVAL_TB 100 // интервал ACK: 10ms в timebase (0.1ms) #define ROUTER_ACK_IDLE_TB 5000 // idle таймаут: 500ms #define ROUTER_SEND_RESUME_TB 500 // retry интервал send_q: 50ms +#define ROUTER_MAX_SEND_Q_PACKETS 256 // порог backpressure на send_q // Флаги в старших битах seq (взаимоисключающие) #define ROUTER_SEQ_CLOSE_FLAG 0x80000000 // нормальное закрытие conn diff --git a/src/proxy/tcp_proxy_server.c b/src/proxy/tcp_proxy_server.c index ed1f9483..90a9ab2b 100644 --- a/src/proxy/tcp_proxy_server.c +++ b/src/proxy/tcp_proxy_server.c @@ -100,9 +100,16 @@ static void send_error(struct tcp_proxy_server_conn* rc) { } // ==================================================================== -// Коллбэки tcp_io +// Коллбэки tcp_io + очередь // ==================================================================== +static void write_on_get_cb(struct ll_queue* q, void* arg) { + (void)q; + struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg; + struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL; + if (inst) etcp_router_consumer_ack(inst, rc->peer_node_id, ETCP_ID_TCP_PROXY); +} + static void on_fin_cb(struct tcp_conn* tc, void* arg) { struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg; if (write_pending(tc)) @@ -112,6 +119,7 @@ static void on_fin_cb(struct tcp_conn* tc, void* arg) { } static void on_flushed_cb(struct tcp_conn* tc, void* arg) { + (void)tc; struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg; if (rc->cli_closed) { tcp_proxy_server_conn_free(rc); return; } send_close(rc); @@ -168,7 +176,11 @@ static void read_queue_drain_cb(struct ll_queue* q, void* arg) { if (ret == 0) { queue_resume_callback(q); } else { - etcp_router_waiter_register(inst, rc->peer_node_id, &rc->pause_waiter, pause_resume_cb, rc); + struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, rc->peer_node_id, ETCP_ID_TCP_PROXY); + if (rconn && rconn->send_q) + queue_waiter_wait(rconn->send_q, &rc->pause_waiter, pause_resume_cb, rc); + else + etcp_router_waiter_register(inst, rc->peer_node_id, &rc->pause_waiter, pause_resume_cb, rc); } } @@ -202,7 +214,12 @@ void tcp_proxy_server_conn_free(struct tcp_proxy_server_conn* rc) { if (rc->tc) { tcp_conn_destroy(rc->tc); rc->tc = NULL; } if (rc->close_timer) { uasync_cancel_timeout(rc->ua, rc->close_timer); rc->close_timer = NULL; } if (rc->diag_timer) { uasync_cancel_timeout(rc->ua, rc->diag_timer); rc->diag_timer = NULL; } - if (rc->ctx && rc->ctx->inst) etcp_router_waiter_cancel(rc->ctx->inst, rc->peer_node_id, &rc->pause_waiter); + if (rc->ctx && rc->ctx->inst) { + struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(rc->ctx->inst, rc->peer_node_id, ETCP_ID_TCP_PROXY); + if (rconn && rconn->send_q) queue_waiter_cancel(rconn->send_q, &rc->pause_waiter); + else etcp_router_waiter_cancel(rc->ctx->inst, rc->peer_node_id, &rc->pause_waiter); + } + if (rc->tc) queue_set_on_get(rc->tc->write_queue, NULL, NULL); u_free(rc); } @@ -254,6 +271,13 @@ int tcp_proxy_server_handle_connect(struct UTUN_INSTANCE* inst, struct ll_entry* rc->next = ctx->conns; ctx->conns = rc; rc->diag_timer = uasync_set_timeout(rc->ua, 10000, rc, diag_timer_cb, "tps_diag"); + + { + struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, src_node_id, ETCP_ID_TCP_PROXY); + if (rconn) rconn->consumer_ack = 1; + } + queue_set_on_get(rc->tc->write_queue, write_on_get_cb, rc); + queue_dgram_free(entry); queue_entry_free(entry); return 0; }