From 03490cfc0053b2fed2492016047f098059c874e8 Mon Sep 17 00:00:00 2001 From: Evgeny Date: Fri, 3 Jul 2026 19:27:36 +0300 Subject: [PATCH] refactor: extract etcp_router_recv_cb into 5 functions + fix send_q threshold MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - router_verify_signature — Ed25519 signature verification - router_handle_ack — ACK processing + drain on send_blocked - router_forward_transit — packet forwarding - router_check_peer_restart — session restart + RST detection - router_handle_data_packet — reorder/assembly/dedup - free_entry helper - etcp_router_recv_cb: 198 → 42 lines - ROUTER_MAX_SEND_Q_PACKETS threshold: MAX/2 → MAX-1 (min wait) --- src/etcp_router.c | 364 +++++++++++++++++------------------ src/etcp_router.h | 2 +- src/proxy/tcp_proxy_server.c | 14 +- 3 files changed, 188 insertions(+), 192 deletions(-) diff --git a/src/etcp_router.c b/src/etcp_router.c index 4c8538fb..fd1d5d9e 100644 --- a/src/etcp_router.c +++ b/src/etcp_router.c @@ -28,6 +28,12 @@ static void router_drain_send_q(struct ETCP_ROUTER_CONN* rconn); static void router_send_resume_cb(void* arg); static void router_send_close_to_service(struct ETCP_ROUTER_CONN* rconn); static void router_close_and_notify(struct ETCP_ROUTER_CONN* rconn); +static void free_entry(struct ll_entry* e) { queue_dgram_free(e); queue_entry_free(e); } +static int router_verify_signature(struct UTUN_INSTANCE* inst, struct ll_entry* entry, struct SVC_ROUTE_HDR* hdr, size_t* pl_len); +static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN* rconn, uint32_t seq, uint64_t src_node_id); +static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* entry, struct SVC_ROUTE_HDR* hdr); +static int router_check_peer_restart(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN** prconn, struct SVC_ROUTE_HDR* hdr); +static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, uint32_t seq, const uint8_t* pl, size_t pl_len); // ==================================================================== // Управление ROUTER_CONN @@ -380,6 +386,157 @@ static void router_try_assembly(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN } } +// ==================================================================== +// Извлечённые функции для etcp_router_recv_cb +// ==================================================================== + +static int router_verify_signature(struct UTUN_INSTANCE* inst, struct ll_entry* entry, + struct SVC_ROUTE_HDR* hdr, size_t* pl_len) { + if (!(hdr->flags & ROUTER_FLAG_SIGNED)) return 0; + if (*pl_len < SC_SIGN_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: SIGNED packet too short len=%zu from %016llx", + *pl_len, (unsigned long long)hdr->src_node_id); + free_entry(entry); return -1; + } + size_t actual_payload = *pl_len - SC_SIGN_SIZE; + uint8_t* sig = (uint8_t*)entry->dgram + SVC_ROUTE_HDR_SIZE + actual_payload; + struct ROUTE_BGP* bgp = inst->bgp; + struct NODEINFO_Q* nq = bgp ? nodeinfo_find_by_id(bgp, hdr->src_node_id) : NULL; + if (!nq) { + DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: SIGNED packet from unknown node %016llx — dropping", + (unsigned long long)hdr->src_node_id); + free_entry(entry); return -1; + } + { + uint8_t zero[SC_PUBKEY_SIZE] = {0}; + if (memcmp(nq->node.ed25519_public_key, zero, SC_PUBKEY_SIZE) == 0) { + DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: no Ed25519 pubkey for node %016llx — dropping", + (unsigned long long)hdr->src_node_id); + free_entry(entry); return -1; + } + } + struct sc_stream_sign_state vfy_state; + if (sc_stream_sign_verify_init(&vfy_state, nq->node.ed25519_public_key) != SC_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router: sign_verify_init failed for node %016llx", + (unsigned long long)hdr->src_node_id); + free_entry(entry); return -1; + } + size_t signed_len = SVC_ROUTE_HDR_SIZE + actual_payload; + if (sc_stream_sign_update(&vfy_state, entry->dgram, signed_len) != SC_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router: sign_update failed for node %016llx", + (unsigned long long)hdr->src_node_id); + sc_stream_sign_cleanup(&vfy_state); free_entry(entry); return -1; + } + if (sc_stream_sign_verify(&vfy_state, sig, SC_SIGN_SIZE) != SC_OK) { + DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: signature verification FAILED for node %016llx svc_id=%u — dropping", + (unsigned long long)hdr->src_node_id, hdr->svc_id); + struct ETCP_ROUTER_CONN* sr = router_conn_find(inst, hdr->src_node_id, hdr->svc_id); + if (sr) sr->c_sign_fail++; + free_entry(entry); return -1; + } + *pl_len = actual_payload; + DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router: signature verified for node %016llx svc_id=%u", + (unsigned long long)hdr->src_node_id, hdr->svc_id); + return 0; +} + +static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN* rconn, + uint32_t seq, uint64_t src_node_id) { + (void)src_node_id; + if (!rconn) return; + if ((int32_t)(seq - rconn->tx_acked) >= 0) { + rconn->tx_acked = seq; + DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "SEND_Q_ACKED: svc_id=%u ack=%u inflight=%d send_q=%d from %016llx", + rconn->svc_id, seq, (int32_t)(rconn->tx_seq - rconn->tx_acked), + queue_entry_count(rconn->send_q), (unsigned long long)src_node_id); + rconn->c_ack_recv++; + } else { + DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: stale ACK seq=%u tx_acked=%u from %016llx, ignoring", + seq, rconn->tx_acked, (unsigned long long)src_node_id); + rconn->c_stale_ack++; + } + rconn->last_dgram_ts = get_current_timestamp(); + if (rconn->send_blocked) router_drain_send_q(rconn); +} + +static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* entry, + struct SVC_ROUTE_HDR* hdr) { + struct ETCP_CONN* next = route_bgp_find_conn_for_node(inst->bgp, hdr->dst_node_id); + if (!next) { + DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: no route to %016llx svc_id=%u, dropping", + (unsigned long long)hdr->dst_node_id, hdr->svc_id); + free_entry(entry); return; + } + DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: forwarding svc_id=%u → %016llx via %s", + hdr->svc_id, (unsigned long long)hdr->dst_node_id, next->log_name); + int fwd_err = etcp_send(next, entry); + if (fwd_err != 0) { + free_entry(entry); + DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: transit etcp_send failed ret=%d", fwd_err); + } +} + +static int router_check_peer_restart(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN** prconn, + struct SVC_ROUTE_HDR* hdr) { + struct ETCP_ROUTER_CONN* rconn = *prconn; + uint8_t incoming_sess = (hdr->flags >> ROUTER_SESS_ID_SHIFT) & 0x3; + if (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, + (unsigned long long)hdr->src_node_id); + etcp_router_conn_restart(inst, hdr->src_node_id, hdr->svc_id); + *prconn = etcp_router_conn_get(inst, hdr->src_node_id, hdr->svc_id); + if (!*prconn) return -1; + (*prconn)->peer_sess_id = incoming_sess; + rconn = *prconn; + } + if (hdr->flags & ROUTER_FLAG_RST) { + DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router: RST from %016llx svc_id=%u", + (unsigned long long)hdr->src_node_id, hdr->svc_id); + router_close_and_notify(rconn); + return -1; + } + return 0; +} + +static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, + uint32_t seq, const uint8_t* pl, size_t pl_len) { + rconn->last_dgram_ts = get_current_timestamp(); + + if (!rconn->peer_sync_done) { rconn->peer_sync_done = 1; rconn->rx_seq = seq; } + + { + int32_t d = (int32_t)(seq - rconn->rx_seq); + if (d > ROUTER_MAX_INFLIGHT || d < -ROUTER_MAX_INFLIGHT) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router: seq=%u out of bounds, rx_seq=%u (d=%d), dropping", + seq, rconn->rx_seq, d); + rconn->c_oob_dropped++; return; + } + } + + if (((int32_t)(rconn->rx_seq - seq) > 0) || queue_find_data_by_index(rconn->recv_q, &seq)) { + DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router: dup seq=%u rx_seq=%u, dropping", seq, rconn->rx_seq); + rconn->c_dup_dropped++; return; + } + + struct ll_entry* qe = queue_entry_new(4); + if (!qe) return; + *(uint32_t*)qe->data = seq; + qe->dgram = u_malloc(pl_len); + if (!qe->dgram) { queue_dgram_free(qe); queue_entry_free(qe); return; } + qe->len = pl_len; memcpy(qe->dgram, pl, pl_len); + + DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "router: queued seq=%u rx_seq=%u recv_q=%d from %016llx svc_id=%u", + seq, rconn->rx_seq, queue_entry_count(rconn->recv_q), + (unsigned long long)rconn->remote_node_id, rconn->svc_id); + queue_data_put_with_index(rconn->recv_q, qe); + + if (seq == rconn->rx_seq) router_try_assembly(rconn, conn); + router_schedule_ack(rconn); +} + // ==================================================================== // etcp_router_recv_cb — обработчик ETCP_ID_SVC_ROUTE // ==================================================================== @@ -390,197 +547,36 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) if (!inst || entry->len < SVC_ROUTE_HDR_SIZE) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: invalid packet inst=%p len=%zu min=%u", (void*)inst, entry->len, (unsigned)SVC_ROUTE_HDR_SIZE); - if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } - return; + free_entry(entry); return; } - struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)entry->dgram; - size_t pl_len = entry->len - SVC_ROUTE_HDR_SIZE; - uint8_t* pl = entry->dgram + SVC_ROUTE_HDR_SIZE; + struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)entry->dgram; + size_t pl_len = entry->len - SVC_ROUTE_HDR_SIZE; + uint8_t* pl = entry->dgram + SVC_ROUTE_HDR_SIZE; DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "ETCP_RECV: svc_id=%u seq=%u len=%zu from %016llx", hdr->svc_id, hdr->seq, pl_len, (unsigned long long)hdr->src_node_id); - if (hdr->dst_node_id == inst->node_id) { - // ========== Проверка подписи (если флаг ROUTER_FLAG_SIGNED) ========== - size_t actual_payload = pl_len; - if (hdr->flags & ROUTER_FLAG_SIGNED) { - if (pl_len < SC_SIGN_SIZE) { - DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: SIGNED packet too short len=%zu from %016llx", pl_len, (unsigned long long)hdr->src_node_id); - queue_dgram_free(entry); queue_entry_free(entry); - return; - } - actual_payload = pl_len - SC_SIGN_SIZE; - uint8_t* sig = pl + actual_payload; - struct ROUTE_BGP* bgp = inst->bgp; - struct NODEINFO_Q* nq = bgp ? nodeinfo_find_by_id(bgp, hdr->src_node_id) : NULL; - if (!nq) { - DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: SIGNED packet from unknown node %016llx — dropping", (unsigned long long)hdr->src_node_id); - queue_dgram_free(entry); queue_entry_free(entry); - return; - } - { - uint8_t zero[SC_PUBKEY_SIZE] = {0}; - if (memcmp(nq->node.ed25519_public_key, zero, SC_PUBKEY_SIZE) == 0) { - DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: no Ed25519 pubkey for node %016llx — dropping", (unsigned long long)hdr->src_node_id); - queue_dgram_free(entry); queue_entry_free(entry); - return; - } - } - struct sc_stream_sign_state vfy_state; - if (sc_stream_sign_verify_init(&vfy_state, nq->node.ed25519_public_key) != SC_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router: sign_verify_init failed for node %016llx", (unsigned long long)hdr->src_node_id); - queue_dgram_free(entry); queue_entry_free(entry); - return; - } - size_t signed_len = SVC_ROUTE_HDR_SIZE + actual_payload; - if (sc_stream_sign_update(&vfy_state, entry->dgram, signed_len) != SC_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router: sign_update failed for node %016llx", (unsigned long long)hdr->src_node_id); - sc_stream_sign_cleanup(&vfy_state); queue_dgram_free(entry); queue_entry_free(entry); - return; - } - if (sc_stream_sign_verify(&vfy_state, sig, SC_SIGN_SIZE) != SC_OK) { - DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: signature verification FAILED for node %016llx svc_id=%u — dropping", (unsigned long long)hdr->src_node_id, hdr->svc_id); - { - struct ETCP_ROUTER_CONN* sr = router_conn_find(inst, hdr->src_node_id, hdr->svc_id); - if (sr) sr->c_sign_fail++; - } - queue_dgram_free(entry); queue_entry_free(entry); - return; - } - pl_len = actual_payload; - DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router: signature verified for node %016llx svc_id=%u", (unsigned long long)hdr->src_node_id, hdr->svc_id); - } + if (hdr->dst_node_id != inst->node_id) { router_forward_transit(inst, entry, hdr); return; } - // ========== Мы — целевая нода ========== - struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, hdr->src_node_id, hdr->svc_id); + if (router_verify_signature(inst, entry, hdr, &pl_len) != 0) return; - // CLOSE или RST — pl_len==0 и флаг в seq (закрываем conn) - if (pl_len == 0 && (hdr->seq & (ROUTER_SEQ_CLOSE_FLAG | ROUTER_SEQ_RST_FLAG))) { - if (rconn) router_close_and_notify(rconn); - queue_dgram_free(entry); queue_entry_free(entry); - return; - } + struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, hdr->src_node_id, hdr->svc_id); - if (pl_len == 0) { - if (rconn) { - if ((int32_t)(hdr->seq - rconn->tx_acked) >= 0) { - rconn->tx_acked = hdr->seq; - DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "SEND_Q_ACKED: svc_id=%u ack=%u inflight=%d send_q=%d from %016llx", - rconn->svc_id, hdr->seq, (int32_t)(rconn->tx_seq - rconn->tx_acked), - queue_entry_count(rconn->send_q), - (unsigned long long)hdr->src_node_id); - rconn->c_ack_recv++; - } else { - DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: stale ACK seq=%u tx_acked=%u from %016llx, ignoring", - hdr->seq, rconn->tx_acked, (unsigned long long)hdr->src_node_id); - rconn->c_stale_ack++; - } - rconn->last_dgram_ts = get_current_timestamp(); - if (rconn->send_blocked) router_drain_send_q(rconn); - } - queue_dgram_free(entry); queue_entry_free(entry); - return; - } - - // Data-пакет — авто-создаём conn если ещё нет - if (!rconn) { - rconn = etcp_router_conn_get(inst, hdr->src_node_id, hdr->svc_id); - if (!rconn) { queue_dgram_free(entry); queue_entry_free(entry); return; } - rconn->peer_sess_id = (hdr->flags >> ROUTER_SESS_ID_SHIFT) & 0x3; - } - - // Обнаружение перезапуска 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) && 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, - (unsigned long long)hdr->src_node_id); - etcp_router_conn_restart(inst, hdr->src_node_id, hdr->svc_id); - rconn = etcp_router_conn_get(inst, hdr->src_node_id, hdr->svc_id); - if (!rconn) { queue_dgram_free(entry); queue_entry_free(entry); return; } - rconn->peer_sess_id = incoming_sess; - } - // RST-флаг — peer сигналит нарушение последовательности - if (hdr->flags & ROUTER_FLAG_RST) { - DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, - "router: RST from %016llx svc_id=%u", - (unsigned long long)hdr->src_node_id, hdr->svc_id); - router_close_and_notify(rconn); - queue_dgram_free(entry); queue_entry_free(entry); - return; - } - } - - // Seq-состояние есть — reorder - rconn->last_dgram_ts = get_current_timestamp(); - uint32_t seq = hdr->seq; - - // Синхронизация после авто-создания/рестарта: принимаем первый пришедший seq - if (!rconn->peer_sync_done) { - rconn->peer_sync_done = 1; - rconn->rx_seq = seq; - } - - // Проверка границ (32-bit circular, аналог ETCP MAX_INFLIGHT_SIZE) - { - int32_t d = (int32_t)(seq - rconn->rx_seq); - if (d > ROUTER_MAX_INFLIGHT || d < -ROUTER_MAX_INFLIGHT) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router: seq=%u out of bounds, rx_seq=%u (d=%d), dropping", - seq, rconn->rx_seq, d); - rconn->c_oob_dropped++; - queue_dgram_free(entry); queue_entry_free(entry); - return; - } - } - - // Дубликат (seq уже доставлен или в recv_q) - if (((int32_t)(rconn->rx_seq - seq) > 0) || - queue_find_data_by_index(rconn->recv_q, &seq)) { - DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router: dup seq=%u rx_seq=%u, dropping", seq, rconn->rx_seq); - rconn->c_dup_dropped++; - queue_dgram_free(entry); queue_entry_free(entry); - return; - } - - // Кладём в recv_q - struct ll_entry* qe = queue_entry_new(4); - if (!qe) { queue_dgram_free(entry); queue_entry_free(entry); return; } - *(uint32_t*)qe->data = seq; - qe->dgram = u_malloc(pl_len); - if (!qe->dgram) { queue_dgram_free(qe); queue_entry_free(qe); queue_dgram_free(entry); queue_entry_free(entry); return; } - qe->len = pl_len; - memcpy(qe->dgram, pl, pl_len); - - DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "router: queued seq=%u rx_seq=%u recv_q=%d from %016llx svc_id=%u", - seq, rconn->rx_seq, queue_entry_count(rconn->recv_q), - (unsigned long long)hdr->src_node_id, rconn->svc_id); - queue_data_put_with_index(rconn->recv_q, qe); - - if (seq == rconn->rx_seq) - router_try_assembly(rconn, conn); - - router_schedule_ack(rconn); - queue_dgram_free(entry); queue_entry_free(entry); - } else { - // ========== Транзит ========== - struct ETCP_CONN* next = route_bgp_find_conn_for_node(inst->bgp, hdr->dst_node_id); - if (!next) { - DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: no route to %016llx svc_id=%u, dropping", - (unsigned long long)hdr->dst_node_id, hdr->svc_id); - queue_dgram_free(entry); queue_entry_free(entry); - return; - } - DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: forwarding svc_id=%u → %016llx via %s", - hdr->svc_id, (unsigned long long)hdr->dst_node_id, next->log_name); - int fwd_err = etcp_send(next, entry); - if (fwd_err != 0) { - queue_dgram_free(entry); queue_entry_free(entry); - DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: transit etcp_send failed ret=%d", fwd_err); - } + if (pl_len == 0 && (hdr->seq & (ROUTER_SEQ_CLOSE_FLAG | ROUTER_SEQ_RST_FLAG))) { + if (rconn) router_close_and_notify(rconn); + free_entry(entry); return; } + if (pl_len == 0) { router_handle_ack(inst, rconn, hdr->seq, hdr->src_node_id); free_entry(entry); return; } + + if (!rconn) { + rconn = etcp_router_conn_get(inst, hdr->src_node_id, hdr->svc_id); + if (!rconn) { free_entry(entry); return; } + rconn->peer_sess_id = (hdr->flags >> ROUTER_SESS_ID_SHIFT) & 0x3; + } + if (router_check_peer_restart(inst, &rconn, hdr) != 0) { free_entry(entry); return; } + + router_handle_data_packet(rconn, conn, hdr->seq, pl, pl_len); + free_entry(entry); } // ==================================================================== @@ -758,7 +754,7 @@ void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t remote_node_i } 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); + queue_set_threshold(rconn->send_q, ROUTER_MAX_SEND_Q_PACKETS - 1, 0); rconn->tx_seq = 0; rconn->rx_seq = 0; @@ -838,7 +834,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); + queue_set_threshold(rconn->send_q, ROUTER_MAX_SEND_Q_PACKETS - 1, 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); diff --git a/src/etcp_router.h b/src/etcp_router.h index 4d734591..c61c1946 100644 --- a/src/etcp_router.h +++ b/src/etcp_router.h @@ -75,7 +75,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 4 // порог backpressure на send_q +#define ROUTER_MAX_SEND_Q_PACKETS 64 // порог 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 aedb0e40..efce0116 100644 --- a/src/proxy/tcp_proxy_server.c +++ b/src/proxy/tcp_proxy_server.c @@ -212,7 +212,6 @@ static void read_queue_drain_cb(struct ll_queue* q, void* arg) { struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL; struct ll_entry* e; - // Сначала добиваем отложенный tx_buf — force=1 чтобы обойти backpressure if (rc->tx_buf) { 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) { @@ -248,9 +247,8 @@ 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, e->dgram, e->len, 0); rc->bytes_relayed += e->len; rc->drain_count++; - if ((rc->drain_count & 63) == 0 || rc->drain_count <= 4) - DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "TPS DRAIN #%u sid=%08x len=%u → ret=%d sent=%u recv=%u relayed=%u", - rc->drain_count, rc->stream_id, e->len, ret, rc->bytes_sent, rc->bytes_recv, rc->bytes_relayed); + DEBUG_TRACE(DEBUG_CATEGORY_TRAFFIC, "TPS DRAIN #%u sid=%08x len=%u → ret=%d rq=%d", + rc->drain_count, rc->stream_id, e->len, ret, q->count); if (ret == 0) { memory_pool_free(rc->tc->data_pool, e->dgram); queue_entry_free(e); queue_resume_callback(q); @@ -270,6 +268,9 @@ static void pause_resume_cb(struct ll_queue* q, void* arg) { (void)q; struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg; if (!rc->tc || rc->tc->sock == SOCKET_INVALID || rc->cli_closed) return; + DEBUG_TRACE(DEBUG_CATEGORY_SOCKET, "TPS RESUME sid=%08x tx_buf=%s rq=%d", + rc->stream_id, rc->tx_buf ? "y" : "n", + rc->tc->read_queue ? rc->tc->read_queue->count : 0); queue_resume_callback(rc->tc->read_queue); } @@ -378,9 +379,8 @@ int tcp_proxy_server_handle_data(struct UTUN_INSTANCE* inst, struct ETCP_CONN* c } size_t data_len = entry->len - TCP_PROXY_HDR_SIZE; rc->bytes_sent += (uint32_t)data_len; rc->data_count++; - if ((rc->data_count & 63) == 0 || rc->data_count <= 4) - DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "TPS DATA #%u sid=%08x len=%zu sent=%u recv=%u relayed=%u", - rc->data_count, stream_id, data_len, rc->bytes_sent, rc->bytes_recv, rc->bytes_relayed); + DEBUG_TRACE(DEBUG_CATEGORY_TRAFFIC, "TPS DATA #%u sid=%08x len=%zu", + rc->data_count, stream_id, data_len); if (data_len > 0) { if (data_len > rc->tc->data_pool->object_size) { DEBUG_ERROR(DEBUG_CATEGORY_SOCKET, "tcp_proxy_server: data_len=%zu > pool_sz=%zu, dropping sid=%08x",