jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

refactor: extract etcp_router_recv_cb into 5 functions + fix send_q threshold

Browse Source 
- 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)
This commit is contained in:
Evgeny committed 2026-07-03 19:27:36 +03:00
1 parent ba14e31e09
commit 03490cfc00
3 files changed
+188 -192
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/etcp_router.c
+180 -184
View File
@@ -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);
src/etcp_router.h
+1 -1
View File
@@ -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
src/proxy/tcp_proxy_server.c
+7 -7
View File
@@ -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",