fix: remove retry_timer, use waiter-based round-robin on send_q for fairness
- tcp_proxy_server: drain_cb uses etcp_router_on_send_ready instead of retry_timer - socks_proxy: on_read_cb + tx_waiter_cb use etcp_router_on_send_ready instead of waiter_register - Result: FIFO round-robin via ll_queue waiters on send_q, all streams get fair share - No timers — wakeup is event-driven via queue_data_get → check_waiters - TCP backpressure: read_queue callback_suspended prevents over-reading
This commit is contained in:
Evgeny committed
1 parent
5b850ea5da
commit
ba14e31e09
4 files changed
+12
-32
No files matched your search
@@ -331,7 +331,7 @@ static void process_http_request(struct socks_proxy_conn* c) {
|
||||
u_free(pkt);
|
||||
} else {
|
||||
c->tx_buf = pkt; c->tx_len = total;
|
||||
etcp_router_waiter_register(c->inst, c->via_node_id, &c->tx_waiter, tx_waiter_cb, c);
|
||||
etcp_router_on_send_ready(c->inst, c->via_node_id, ETCP_ID_TCP_PROXY, &c->tx_waiter, tx_waiter_cb, c);
|
||||
}
|
||||
|
||||
c->buf_len = 0;
|
||||
@@ -395,7 +395,7 @@ static void on_read_cb(struct ll_queue* q, void* arg) {
|
||||
if (c->tx_buf) { memcpy(c->tx_buf, e->dgram, e->len); c->tx_len = e->len; }
|
||||
else { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "socks_proxy: tx_buf malloc=%u failed sid=%08x — drop", e->len, c->stream_id); }
|
||||
memory_pool_free(c->tc->data_pool, e->dgram); queue_entry_free(e);
|
||||
etcp_router_waiter_register(c->inst, c->via_node_id, &c->tx_waiter, tx_waiter_cb, c);
|
||||
etcp_router_on_send_ready(c->inst, c->via_node_id, ETCP_ID_TCP_PROXY, &c->tx_waiter, tx_waiter_cb, c);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -408,7 +408,7 @@ static void tx_waiter_cb(struct ll_queue* q, void* arg) {
|
||||
u_free(c->tx_buf); c->tx_buf = NULL; c->tx_len = 0;
|
||||
queue_resume_callback(c->tc->read_queue);
|
||||
} else {
|
||||
etcp_router_waiter_register(c->inst, c->via_node_id, &c->tx_waiter, tx_waiter_cb, c);
|
||||
etcp_router_on_send_ready(c->inst, c->via_node_id, ETCP_ID_TCP_PROXY, &c->tx_waiter, tx_waiter_cb, c);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -522,7 +522,7 @@ int socks_proxy_handle_etcp(struct socks_proxy_conn** head, int* count,
|
||||
if (subcmd == TCP_PROXY_SUBCMD_CLOSE) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "socks_proxy: REM_CLOSED sid=%08x", stream_id);
|
||||
c->rem_closed = 1;
|
||||
etcp_router_waiter_cancel(c->inst, c->via_node_id, &c->tx_waiter);
|
||||
etcp_router_cancel_send_ready(c->inst, c->via_node_id, ETCP_ID_TCP_PROXY, &c->tx_waiter);
|
||||
tcp_conn_push_close(c->tc);
|
||||
return 1;
|
||||
}
|
||||
@@ -530,7 +530,7 @@ int socks_proxy_handle_etcp(struct socks_proxy_conn** head, int* count,
|
||||
if (subcmd == TCP_PROXY_SUBCMD_ERROR) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "socks_proxy: ERROR from exit sid=%08x", stream_id);
|
||||
c->rem_closed = 1;
|
||||
etcp_router_waiter_cancel(c->inst, c->via_node_id, &c->tx_waiter);
|
||||
etcp_router_cancel_send_ready(c->inst, c->via_node_id, ETCP_ID_TCP_PROXY, &c->tx_waiter);
|
||||
tcp_conn_push_close(c->tc);
|
||||
return 1;
|
||||
}
|
||||
|
||||
@@ -33,7 +33,6 @@ static void read_queue_drain_cb(struct ll_queue* q, void* arg);
|
||||
static void pause_resume_cb(struct ll_queue* q, void* arg);
|
||||
static void close_retry_cb(void* arg);
|
||||
static void diag_timer_cb(void* arg);
|
||||
static void retry_timer_cb(void* arg);
|
||||
static int send_msg(struct UTUN_INSTANCE* inst, uint64_t dst, uint8_t subcmd,
|
||||
uint32_t sid, const uint8_t* data, size_t len, int force);
|
||||
static void send_close(struct tcp_proxy_server_conn* rc);
|
||||
@@ -218,14 +217,14 @@ static void read_queue_drain_cb(struct ll_queue* q, void* arg) {
|
||||
int ret = send_msg(inst, rc->peer_node_id, TCP_PROXY_SUBCMD_DATA, rc->stream_id, rc->tx_buf, rc->tx_len, 0);
|
||||
if (ret == 0) {
|
||||
u_free(rc->tx_buf); rc->tx_buf = NULL; rc->tx_len = 0;
|
||||
if (rc->retry_timer) { uasync_cancel_timeout(rc->ua, rc->retry_timer); rc->retry_timer = NULL; }
|
||||
if (rc->dst_fin_deferred && !q->head) {
|
||||
rc->dst_fin_deferred = 0;
|
||||
send_fin(rc);
|
||||
return;
|
||||
}
|
||||
} else {
|
||||
if (!rc->retry_timer) rc->retry_timer = uasync_set_timeout(rc->ua, 5000, rc, retry_timer_cb, "tps_retry");
|
||||
etcp_router_on_send_ready(inst, rc->peer_node_id, ETCP_ID_TCP_PROXY_CLIENT,
|
||||
&rc->pause_waiter, pause_resume_cb, rc);
|
||||
return;
|
||||
}
|
||||
}
|
||||
@@ -262,7 +261,8 @@ static void read_queue_drain_cb(struct ll_queue* q, void* arg) {
|
||||
memory_pool_free(rc->tc->data_pool, e->dgram); queue_entry_free(e);
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "SOCK:BACKPRESSURE fd=%d sid=%08x — buffered %u for retry",
|
||||
(int)rc->tc->sock, rc->stream_id, rc->tx_len);
|
||||
if (!rc->retry_timer) rc->retry_timer = uasync_set_timeout(rc->ua, 5000, rc, retry_timer_cb, "tps_retry");
|
||||
etcp_router_on_send_ready(inst, rc->peer_node_id, ETCP_ID_TCP_PROXY_CLIENT,
|
||||
&rc->pause_waiter, pause_resume_cb, rc);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -273,25 +273,6 @@ static void pause_resume_cb(struct ll_queue* q, void* arg) {
|
||||
queue_resume_callback(rc->tc->read_queue);
|
||||
}
|
||||
|
||||
static void retry_timer_cb(void* arg) {
|
||||
struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
|
||||
rc->retry_timer = NULL;
|
||||
if (!rc->tx_buf || !rc->tc || rc->tc->sock == SOCKET_INVALID || rc->cli_closed) return;
|
||||
struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL;
|
||||
int ret = send_msg(inst, rc->peer_node_id, TCP_PROXY_SUBCMD_DATA, rc->stream_id, rc->tx_buf, rc->tx_len, 1);
|
||||
if (ret == 0) {
|
||||
u_free(rc->tx_buf); rc->tx_buf = NULL; rc->tx_len = 0;
|
||||
if (rc->dst_fin_deferred && rc->tc && rc->tc->read_queue && !rc->tc->read_queue->head) {
|
||||
rc->dst_fin_deferred = 0;
|
||||
send_fin(rc);
|
||||
return;
|
||||
}
|
||||
queue_resume_callback(rc->tc->read_queue);
|
||||
} else {
|
||||
rc->retry_timer = uasync_set_timeout(rc->ua, 5000, rc, retry_timer_cb, "tps_retry");
|
||||
}
|
||||
}
|
||||
|
||||
// ====================================================================
|
||||
// Управление жизненным циклом коннекта
|
||||
// ====================================================================
|
||||
@@ -323,8 +304,8 @@ void tcp_proxy_server_conn_free(struct tcp_proxy_server_conn* rc) {
|
||||
struct tcp_proxy_server_conn** prev = &rc->ctx->conns;
|
||||
while (*prev) { if (*prev == rc) { *prev = rc->next; rc->ctx->conn_count--; break; } prev = &(*prev)->next; }
|
||||
}
|
||||
if (rc->ctx && rc->ctx->inst) etcp_router_cancel_send_ready(rc->ctx->inst, rc->peer_node_id, ETCP_ID_TCP_PROXY_CLIENT, &rc->pause_waiter);
|
||||
if (rc->tx_buf) { u_free(rc->tx_buf); rc->tx_buf = NULL; rc->tx_len = 0; }
|
||||
if (rc->retry_timer) { uasync_cancel_timeout(rc->ua, rc->retry_timer); rc->retry_timer = 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->tc) { tcp_conn_destroy(rc->tc); rc->tc = NULL; }
|
||||
|
||||
@@ -36,7 +36,6 @@ struct tcp_proxy_server_conn {
|
||||
void* close_timer; // таймер повтора CLOSE/ERROR
|
||||
int close_backoff; // backoff: 50..5000 tb (5ms..500ms)
|
||||
void* diag_timer; // 1-секундный таймер диагностики
|
||||
void* retry_timer; // таймер повтора tx_buf при backpressure (500ms)
|
||||
uint8_t dst_fin_deferred; // FIN от destination отложен (ждём drain read_queue + tx_buf)
|
||||
uint8_t freed; // 1 = уже освобождён, защита от double-free
|
||||
|
||||
|
||||
Reference in new issue
Block a user