etcp_router: транзитные очереди per (src,dst) с backpressure + унификация etcp_send

- TRANSIT_QUEUE: отдельная очередь на каждую пару src+dst узлов
- ll_queue с хеш-индексом для быстрого поиска transit-очередей
- backpressure через waiter на send_input_q (threshold=0, один пакет за вызов, round-robin)
- Очередь создаётся при первом пакете который не получается отправить напрямую, удаляется при опустошении
- Унифицирован etcp_send: единый путь queue_data_put(send_input_q) для UDP (normalizer->input) и STCP (tx_queue)
- Убран STCP-ветвления из router_forward_transit/drain_cb
- Исправлено перекрытие ll.data и tq->q в TRANSIT_QUEUE (добавлены явные поля src_node_id/dst_node_id)
- Фикс лика dgram в tx_queue_cb/client_tx_queue_cb (добавлен queue_dgram_free)
- transit_queues живут внутри ETCP_CONN, инициализируются лениво, очищаются в etcp_connection_close (до pn_deinit) и stcp_link_close
This commit is contained in:
Evgeny committed 2026-07-05 12:57:41 +03:00
1 parent 55fa5c7759
commit 4914348557
13 files changed
+379 -26

No files matched your search

+15
View File
@@ -1196,6 +1196,7 @@ static void send_metrics(struct control_server* server, struct control_client* c
/* Fill router congestion metrics — aggregate over all router_conns */
memset(&rsp->router, 0, sizeof(rsp->router));
if (instance->router_conns) {
uint32_t rtt_cnt = 0;
struct ll_entry* re = instance->router_conns->head;
while (re) {
struct ETCP_ROUTER_CONN* rc = (struct ETCP_ROUTER_CONN*)re;
@@ -1211,8 +1212,22 @@ static void send_metrics(struct control_server* server, struct control_client* c
rsp->router.oob_dropped += rc->c_oob_dropped;
rsp->router.stale_ack += rc->c_stale_ack;
rsp->router.sign_fail += rc->c_sign_fail;
if (rc->rtt > 0) {
if (rc->rtt > rsp->router.rtt_last) rsp->router.rtt_last = rc->rtt;
rsp->router.rtt_avg += rc->rtt;
rtt_cnt++;
}
{
uint16_t j = (uint16_t)(rc->rtt_jitter / 32768);
if (j > rsp->router.jitter) rsp->router.jitter = j;
}
if (rc->minrtt_window_count > 0) {
if (rsp->router.minrtt == 0 || rc->minrtt < rsp->router.minrtt)
rsp->router.minrtt = rc->minrtt;
}
re = re->next;
}
if (rtt_cnt > 0) rsp->router.rtt_avg /= rtt_cnt;
rsp->router.total_conns = (uint32_t)queue_entry_count(instance->router_conns);
}