diff --git a/src/etcp_router.c b/src/etcp_router.c index cfe12c28..c541d4a4 100644 --- a/src/etcp_router.c +++ b/src/etcp_router.c @@ -119,10 +119,11 @@ static void router_send_close_to_service(struct ETCP_ROUTER_CONN* rconn) { if (!cb) return; struct ll_entry* e = queue_entry_new(0); if (!e) return; - e->dgram = u_malloc(1); + e->dgram = u_malloc(9); if (!e->dgram) { queue_entry_free(e); return; } e->dgram[0] = rconn->svc_id; - e->len = 1; + memcpy(e->dgram + 1, &rconn->remote_node_id, 8); + e->len = 9; DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_close_svc: svc_id=%u remote=%016llx", rconn->svc_id, (unsigned long long)rconn->remote_node_id); cb(NULL, e); @@ -380,7 +381,8 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) // Обнаружение перезапуска peer'а: новый sess_id или START-флаг { uint8_t incoming_sess = (hdr->flags >> ROUTER_SESS_ID_SHIFT) & 0x3; - if (pl_len > 0 && (incoming_sess != rconn->peer_sess_id || (hdr->flags & ROUTER_FLAG_START))) { + if (pl_len > 0 && (incoming_sess != rconn->peer_sess_id || + ((hdr->flags & ROUTER_FLAG_START) && rconn->rx_seq > 0))) { DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router: peer restart sess_id=%u→%u svc_id=%u from %016llx", rconn->peer_sess_id, incoming_sess, hdr->svc_id, @@ -441,7 +443,7 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) if (seq == rconn->rx_seq) router_try_assembly(rconn, conn); - if (!rconn->consumer_ack) router_schedule_ack(rconn); + router_schedule_ack(rconn); queue_dgram_free(entry); queue_entry_free(entry); } else { // ========== Транзит ========== @@ -597,40 +599,65 @@ void etcp_router_waiter_cancel(struct UTUN_INSTANCE* inst, uint64_t peer_node_id queue_waiter_cancel(conn->normalizer->input, h); } -void etcp_router_consumer_ack(struct UTUN_INSTANCE* inst, uint64_t remote_node_id, uint8_t svc_id) { - struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, remote_node_id, svc_id); - if (!rconn) return; - rconn->consumer_ack = 1; - DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "CONSUMER_ACK: svc_id=%u rx_seq=%u → %016llx", - svc_id, rconn->rx_seq, (unsigned long long)remote_node_id); - router_ack_do_send(rconn); -} - void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t remote_node_id, uint8_t svc_id) { struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, remote_node_id, svc_id); if (!rconn) return; - DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_restart: svc_id=%u remote=%016llx — notifying service", + DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_restart: svc_id=%u remote=%016llx — resetting local state", svc_id, (unsigned long long)remote_node_id); - rconn->sess_id = (rconn->sess_id + 1) & 0x3; - rconn->start_sent = 0; - etcp_recv_fn cb = inst->router_bindings.callbacks[svc_id]; if (cb) { struct ll_entry* e = queue_entry_new(0); if (e) { - e->dgram = u_malloc(10); + e->dgram = u_malloc(9); if (e->dgram) { e->dgram[0] = svc_id; - e->dgram[1] = 0xFE; // RESTART subcmd - memcpy(e->dgram + 2, &remote_node_id, 8); - e->len = 10; + memcpy(e->dgram + 1, &remote_node_id, 8); + e->len = 9; cb(NULL, e); } else { queue_entry_free(e); } } } - router_close_and_notify(rconn); + + struct ll_entry* f; + if (rconn->send_q) { + while ((f = queue_data_get(rconn->send_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); } + queue_free(rconn->send_q); + } + if (rconn->recv_q) { + while ((f = queue_data_get(rconn->recv_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); } + queue_free(rconn->recv_q); + } + rconn->send_q = queue_new(inst->ua, 0, 0, 0, "router_send_q"); + rconn->recv_q = queue_new(inst->ua, ROUTER_RECVQ_HASH_SIZE, 0, 4, "router_recv_q"); + queue_set_threshold(rconn->send_q, ROUTER_MAX_SEND_Q_PACKETS / 2, 0); + + rconn->tx_seq = 0; + rconn->rx_seq = 0; + rconn->tx_acked = 0; + rconn->last_sent_ack_seq = 0; + rconn->sess_id = (rconn->sess_id + 1) & 0x3; + rconn->peer_sess_id = 0; + rconn->start_sent = 0; + rconn->send_blocked = 0; +} + +void etcp_router_on_send_ready(struct UTUN_INSTANCE* inst, uint64_t node_id, uint8_t svc_id, + struct queue_waiter_handle* h, + queue_threshold_callback_fn callback, void* arg) { + if (!inst || !h) return; + struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, node_id, svc_id); + if (!rconn || !rconn->send_q) return; + queue_waiter_wait(rconn->send_q, h, callback, arg); +} + +void etcp_router_cancel_send_ready(struct UTUN_INSTANCE* inst, uint64_t node_id, uint8_t svc_id, + struct queue_waiter_handle* h) { + if (!inst || !h) return; + struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, node_id, svc_id); + if (!rconn || !rconn->send_q) return; + queue_waiter_cancel(rconn->send_q, h); } // ==================================================================== @@ -653,7 +680,6 @@ struct ETCP_ROUTER_CONN* etcp_router_conn_get(struct UTUN_INSTANCE* inst, rconn->rx_seq = 0; rconn->tx_acked = 0; rconn->last_sent_ack_seq = 0; - rconn->consumer_ack = 0; rconn->last_ack_sent_tb = 0; rconn->inst = inst; rconn->ack_timer = NULL; diff --git a/src/etcp_router.h b/src/etcp_router.h index 5399ed1e..41763c6c 100644 --- a/src/etcp_router.h +++ b/src/etcp_router.h @@ -42,7 +42,6 @@ struct ETCP_ROUTER_CONN { uint32_t rx_seq; // ожидаемый seq для сборки (next expected) uint32_t tx_acked; // сколько наших пакетов подтвердил remote (для inflight) uint32_t last_sent_ack_seq; // последний отправленный ACK (= rx_seq на момент отправки) - uint8_t consumer_ack; // 1 = ACK отправляется только при потреблении, не при сборке uint64_t last_ack_sent_tb; // время отправки последнего ACK (timebase 0.1ms) struct UTUN_INSTANCE* inst; @@ -113,8 +112,13 @@ void etcp_router_waiter_register(struct UTUN_INSTANCE* inst, uint64_t peer_node_ void etcp_router_waiter_cancel(struct UTUN_INSTANCE* inst, uint64_t peer_node_id, struct queue_waiter_handle* h); -// Уведомить router о потреблении данных потребителем (для consumer-driven ACK) -void etcp_router_consumer_ack(struct UTUN_INSTANCE* inst, uint64_t remote_node_id, uint8_t svc_id); +// Backpressure: зарегистрировать/отменить waiter на send_q очереди +// h — handle из структуры сервиса, callback вызывается когда send_q.count <= threshold (128) +void etcp_router_on_send_ready(struct UTUN_INSTANCE* inst, uint64_t node_id, uint8_t svc_id, + struct queue_waiter_handle* h, + queue_threshold_callback_fn callback, void* arg); +void etcp_router_cancel_send_ready(struct UTUN_INSTANCE* inst, uint64_t node_id, uint8_t svc_id, + struct queue_waiter_handle* h); // Сбросить состояние роутера для конкретного peer+svc (перезапуск удалённой стороны). // Очищает send_q/recv_q, уведомляет сервис через cb(NULL, entry) с remote_node_id. diff --git a/src/proxy/tcp_proxy_client.c b/src/proxy/tcp_proxy_client.c index e6149e86..d49f5904 100644 --- a/src/proxy/tcp_proxy_client.c +++ b/src/proxy/tcp_proxy_client.c @@ -186,7 +186,6 @@ static void tcp_proxy_client_feed_from_transport(struct tcp_proxy_client_conn *p } queue_resume_callback(pc->to_lwip); if (sent_any) { - etcp_router_consumer_ack(pc->proxy->inst, pc->proxy->via_node_id, ETCP_ID_TCP_PROXY); uint32_t unsent = 0; struct tcp_seg* s; for (s = pc->pcb->unsent; s; s = s->next) unsent++; DEBUG_DEBUG(DEBUG_CATEGORY_TRAFFIC, "PROXY FEED exit->client sid=%08x fed=%u q=%u->%u snd_wnd=%u cwnd=%u unsent=%u", @@ -467,33 +466,35 @@ static void tcp_proxy_client_handle_error(struct tcp_proxy_client* p, uint32_t s // Единый обработчик etcp_router (диспетчеризация в tcp_proxy_client или tcp_proxy_server) // ==================================================================== void tcp_proxy_client_etcp_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) { - if (!entry || !entry->dgram || entry->len < TCP_PROXY_HDR_SIZE) { - if (entry) { DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "TCP proxy client: bad entry len=%u", entry->len); queue_dgram_free(entry); queue_entry_free(entry); } - return; - } - uint8_t subcmd = entry->dgram[1]; - uint32_t stream_id; memcpy(&stream_id, entry->dgram + 2, 4); struct UTUN_INSTANCE* inst = conn ? conn->instance : NULL; struct tcp_proxy_client* proxy = inst ? inst->tcp_proxy_client : NULL; - if (subcmd == TCP_PROXY_SUBCMD_RESTART) { - uint64_t peer_id = 0; - if (entry->len >= 10) memcpy(&peer_id, entry->dgram + 2, 8); - DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "PROXY RESTART from %016llx — clearing all conns for peer", (unsigned long long)peer_id); - if (proxy) { - struct tcp_proxy_client_conn *pc, *next; - for (pc = proxy->conns; pc; pc = next) { next = pc->next; tcp_proxy_client_conn_free(pc); } - proxy->conns = NULL; proxy->conn_count = 0; - } - if (inst && inst->tcp_proxy_server.enabled) { - struct tcp_proxy_server_conn *rc, *next; - for (rc = inst->tcp_proxy_server.conns; rc; rc = next) { - next = rc->next; - if (rc->peer_node_id == peer_id) tcp_proxy_server_conn_free(rc); + if (!entry || !entry->dgram || entry->len < TCP_PROXY_HDR_SIZE) { + // conn==NULL + len==9: close/restart уведомление от роутера + if (entry && entry->len == 9 && !conn && entry->dgram[0] == ETCP_ID_TCP_PROXY) { + uint64_t peer_id; + memcpy(&peer_id, entry->dgram + 1, 8); + DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "PROXY CLOSE_ALL from %016llx — clearing all conns for peer", + (unsigned long long)peer_id); + if (proxy) { + struct tcp_proxy_client_conn *pc, *next; + for (pc = proxy->conns; pc; pc = next) { next = pc->next; tcp_proxy_client_conn_free(pc); } + proxy->conns = NULL; proxy->conn_count = 0; + inst = proxy->inst; + } + if (inst && inst->tcp_proxy_server.enabled) { + struct tcp_proxy_server_conn *rc, *next; + for (rc = inst->tcp_proxy_server.conns; rc; rc = next) { + next = rc->next; + if (rc->peer_node_id == peer_id) tcp_proxy_server_conn_free(rc); + } } } - queue_dgram_free(entry); queue_entry_free(entry); return; + if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } + return; } + uint8_t subcmd = entry->dgram[1]; + uint32_t stream_id; memcpy(&stream_id, entry->dgram + 2, 4); if (subcmd == TCP_PROXY_SUBCMD_CONNECT) { uint64_t src_node_id = conn ? conn->peer_node_id : (inst ? inst->node_id : 0); diff --git a/src/proxy/tcp_proxy_server.c b/src/proxy/tcp_proxy_server.c index 7ba90b62..24978ead 100644 --- a/src/proxy/tcp_proxy_server.c +++ b/src/proxy/tcp_proxy_server.c @@ -100,23 +100,9 @@ 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) { - struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg; - struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL; - if (!inst) return; - rc->ack_batch++; - if (rc->ack_batch >= 8) { - rc->ack_batch = 0; - DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "SOCK:ACK fd=%d sid=%08x write_q=%d", - rc->tc ? (int)rc->tc->sock : -1, rc->stream_id, - queue_entry_count(q)); - 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)) @@ -147,15 +133,6 @@ static void diag_timer_cb(void* arg) { int ent_f = tc->entry_pool->free_count; int dat_f = tc->data_pool->free_count; - int sq = 0; - { - struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL; - if (inst) { - struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, rc->peer_node_id, ETCP_ID_TCP_PROXY); - if (rconn && rconn->send_q) sq = queue_entry_count(rconn->send_q); - } - } - int rcv_buf = 0, snd_buf = 0; socklen_t optlen = sizeof(int); getsockopt(tc->sock, SOL_SOCKET, SO_RCVBUF, &rcv_buf, &optlen); @@ -163,12 +140,11 @@ static void diag_timer_cb(void* arg) { DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:DIAG fd=%d sid=%08x conn=%d fin=%d " - "rq=%d(%zub) wq=%d(%zub) send_q=%d wbuf=%s " + "rq=%d(%zub) wq=%d(%zub) wbuf=%s " "ent_f=%d dat_f=%d tcp_rb=%zu tcp_sb=%zu", (int)tc->sock, rc->stream_id, tc->connected, tc->fin, tc->read_queue->count, queue_total_bytes(tc->read_queue), tc->write_queue->count, queue_total_bytes(tc->write_queue), - sq, tc->write_buf ? "y" : "n", ent_f, dat_f, (size_t)rcv_buf, (size_t)snd_buf); @@ -193,13 +169,10 @@ static void read_queue_drain_cb(struct ll_queue* q, void* arg) { if (ret == 0) { queue_resume_callback(q); } else { - struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, rc->peer_node_id, ETCP_ID_TCP_PROXY); - if (rconn && rconn->send_q) { - DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:BACKPRESSURE fd=%d sid=%08x send_q=%d — pausing drain", - (int)rc->tc->sock, rc->stream_id, queue_entry_count(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); + DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:BACKPRESSURE fd=%d sid=%08x — pausing drain", + (int)rc->tc->sock, rc->stream_id); + etcp_router_on_send_ready(inst, rc->peer_node_id, ETCP_ID_TCP_PROXY, + &rc->pause_waiter, pause_resume_cb, rc); } } @@ -230,15 +203,6 @@ 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->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) { - 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); } @@ -291,15 +255,6 @@ 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; - etcp_router_consumer_ack(inst, src_node_id, ETCP_ID_TCP_PROXY); - } - } - queue_set_on_get(rc->tc->write_queue, write_on_get_cb, rc); - queue_dgram_free(entry); queue_entry_free(entry); return 0; } diff --git a/src/proxy/tcp_proxy_server.h b/src/proxy/tcp_proxy_server.h index c5efd5dd..64f2520b 100644 --- a/src/proxy/tcp_proxy_server.h +++ b/src/proxy/tcp_proxy_server.h @@ -16,7 +16,6 @@ struct ETCP_CONN; #define TCP_PROXY_SUBCMD_DATA 0x03 #define TCP_PROXY_SUBCMD_CLOSE 0x04 #define TCP_PROXY_SUBCMD_ERROR 0x05 -#define TCP_PROXY_SUBCMD_RESTART 0xFE #define TCP_PROXY_HDR_SIZE 6 // svc_id(1)+subcmd(1)+stream_id(4) #define TCP_PROXY_CONNECT_HDR_SIZE 12 // HDR_SIZE + dest_ip(4)+dest_port(2) @@ -36,7 +35,6 @@ struct tcp_proxy_server_conn { void* close_timer; // таймер повтора CLOSE/ERROR int close_backoff; // backoff: 50..5000 tb (5ms..500ms) void* diag_timer; // 1-секундный таймер диагностики - uint8_t ack_batch; // счётчик для throttled consumer_ack (каждые 8 записей) struct queue_waiter_handle pause_waiter; struct UASYNC* ua; diff --git a/tests/test_etcp_router_unit.c b/tests/test_etcp_router_unit.c index 60fb1547..27bf78b5 100644 --- a/tests/test_etcp_router_unit.c +++ b/tests/test_etcp_router_unit.c @@ -40,6 +40,12 @@ static void test_handler(struct ETCP_CONN* conn, struct ll_entry* entry) { if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } return; } + // close/restart: len==9 (svc_id + node_id), conn==NULL + if (entry->len == 9 && !conn) { + g_notify_count++; + queue_dgram_free(entry); queue_entry_free(entry); + return; + } if (entry->dgram[1] != rx.marker) { rx.errors++; } else { rx.delivered++; rx.last_seq = rx.expected_seq; rx.expected_seq++; } queue_entry_free(entry); queue_dgram_free(entry); }