You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1490 lines
82 KiB
1490 lines
82 KiB
// etcp_router.c — Сервисный слой маршрутизации поверх ETCP |
|
// Упрощённый TCP: восстановление порядка (recv_q), дедупликация, ретрансмиты (inflight_q) |
|
// ACK: периодическая отправка rx_seq (не чаще 10ms, ROUTER_ACK_INTERVAL_TB), idle-таймер для последнего seq |
|
// Inflight контроль через send_q: при переполнении — очередь + retry-таймер, без дропов |
|
// Подпись/шифрование — в route_crypto.c (encode при отправке, decode до изменения состояния приёма) |
|
// |
|
// Поток сервисной кодограммы: |
|
// in -> {send_q} -> [src: etcp_send] -> ... -> [dst: recv_q] -> {incoming_q} -> out |
|
// ack возвращается по обратному пути; (pause) — backpressure при заполнении очередей. |
|
|
|
|
|
/* |
|
/----- ack ack <-- id |
|
|(pause) | |
|
in ---> {Q} -> [src, etcp] --> ... --> [dst,etcp] -> {asm_q} -> {buf q} ---> out |
|
|
|
*/ |
|
|
|
#include "etcp_router.h" |
|
#include "route_crypto.h" |
|
#include "pkt_normalizer.h" |
|
#include "etcp.h" |
|
#include "utun_instance.h" |
|
#include "topo_group.h" |
|
#include "../lib/debug_config.h" |
|
#include "../lib/mem.h" |
|
#include "../lib/ll_queue.h" |
|
#include "../lib/u_async.h" |
|
#include <string.h> |
|
|
|
// ==================================================================== |
|
// Внутренние forward declarations |
|
// ==================================================================== |
|
static void router_ack_timer_cb(void* arg); |
|
static void router_idle_ack_timer_cb(void* arg); |
|
static void router_send_ack(struct ETCP_ROUTER_CONN* rconn); |
|
static void router_schedule_ack(struct ETCP_ROUTER_CONN* rconn); |
|
static void router_try_assembly(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn); |
|
static int router_deliver(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, const uint8_t* wire, size_t wire_len); |
|
static void router_send_ctrl(struct ETCP_ROUTER_CONN* rconn, uint8_t flag_bits); |
|
static void router_handshake_timer_cb(void* arg); |
|
static void router_handshake_start(struct ETCP_ROUTER_CONN* rconn); |
|
static void router_cancel_send_waiter(struct ETCP_ROUTER_CONN* rconn); |
|
static struct ETCP_CONN* router_send_conn(struct ETCP_ROUTER_CONN* rconn); |
|
static void router_send_kick(struct ETCP_ROUTER_CONN* rconn); |
|
static void router_send_drain_cb(struct ll_queue* q, void* arg); |
|
static void router_send_watchdog_cb(void* arg); |
|
static void router_send_watchdog_arm(struct ETCP_ROUTER_CONN* rconn); |
|
static void router_send_watchdog_disarm(struct ETCP_ROUTER_CONN* rconn); |
|
static void router_no_route_retry_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 router_close_finalize(void* arg); |
|
static void router_conn_free_queues(struct ETCP_ROUTER_CONN* rconn); |
|
static int router_conn_reset(struct ETCP_ROUTER_CONN* rconn); |
|
static void free_entry(struct ll_entry* e) { queue_dgram_free(e); queue_entry_free(e); } |
|
static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN* rconn, uint32_t seq, uint64_t src_node_id, uint16_t ack_ts); |
|
static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* entry, 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* wire, size_t wire_len); |
|
static int router_retransmit_one(struct ETCP_ROUTER_CONN* rconn, struct ROUTER_INFLIGHT* inf); |
|
static void router_retrans_schedule(struct ETCP_ROUTER_CONN* rconn); |
|
static void router_retrans_timer_cb(void* arg); |
|
static void router_incoming_q_cb(struct ll_queue* q, void* arg); |
|
static void router_track_inflight_state(struct ETCP_ROUTER_CONN* rconn); |
|
static uint32_t router_effective_max_inflight(struct ETCP_ROUTER_CONN* rconn); |
|
static void router_update_inflight_limit(struct ETCP_ROUTER_CONN* rconn); |
|
static void router_update_minrtt(struct ETCP_ROUTER_CONN* rconn); |
|
|
|
// Очередь-владелец waiter фиксируется при регистрации, не вычисляется по текущему маршруту. |
|
static void router_cancel_send_waiter(struct ETCP_ROUTER_CONN* rconn) { |
|
if (rconn->send_waiter_q) queue_waiter_cancel(rconn->send_waiter_q, &rconn->send_waiter); |
|
rconn->send_waiter_q = NULL; |
|
} |
|
|
|
// ==================================================================== |
|
// Транзитные очереди — per (group_id, src_node_id, dst_node_id) pair, backpressure через waiter на send_input_q |
|
// ==================================================================== |
|
|
|
// Найти транзитную очередь для пары (group_id, src, dst) на conn (NULL если не создана). |
|
static struct TRANSIT_QUEUE* transit_queue_find(struct ETCP_CONN* conn, |
|
uint64_t group_id, |
|
uint64_t src_node_id, uint64_t dst_node_id) { |
|
if (!conn->transit_queues) return NULL; |
|
uint8_t key[24]; |
|
memcpy(key, &group_id, 8); |
|
memcpy(key + 8, &src_node_id, 8); |
|
memcpy(key + 16, &dst_node_id, 8); |
|
return (struct TRANSIT_QUEUE*)queue_find_data_by_index(conn->transit_queues, key); |
|
} |
|
|
|
// Найти транзитную очередь, при отсутствии — создать (реестр транзитных очередей живёт на conn). |
|
static struct TRANSIT_QUEUE* transit_queue_get_or_create(struct ETCP_CONN* conn, |
|
uint64_t group_id, |
|
uint64_t src_node_id, uint64_t dst_node_id) { |
|
if (!conn->transit_queues) { |
|
struct UASYNC* ua = conn->instance ? conn->instance->ua : NULL; |
|
if (!ua) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "transit_queue_get_or_create: no ua"); return NULL; } |
|
conn->transit_queues = queue_new(ua, TRANSIT_QUEUE_HASH, 0, 24, "transit_q_reg"); |
|
if (!conn->transit_queues) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "transit_queue_get_or_create: queue_new failed"); return NULL; } |
|
} |
|
struct TRANSIT_QUEUE* tq = transit_queue_find(conn, group_id, src_node_id, dst_node_id); |
|
if (tq) return tq; |
|
size_t data_size = sizeof(struct TRANSIT_QUEUE) - sizeof(struct ll_entry); |
|
tq = (struct TRANSIT_QUEUE*)queue_entry_new(data_size); |
|
if (!tq) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "transit_queue_get_or_create: queue_entry_new failed"); |
|
return NULL; |
|
} |
|
memset(&tq->waiter, 0, sizeof(tq->waiter)); |
|
tq->group_id = group_id; |
|
tq->src_node_id = src_node_id; |
|
tq->dst_node_id = dst_node_id; |
|
tq->conn = conn; |
|
tq->q = queue_new(conn->instance->ua, 0, 0, 0, "transit_pair_q"); |
|
if (!tq->q) { queue_entry_free(&tq->ll); return NULL; } |
|
memcpy(tq->ll.data, &group_id, 8); |
|
memcpy(tq->ll.data + 8, &src_node_id, 8); |
|
memcpy(tq->ll.data + 16, &dst_node_id, 8); |
|
queue_data_put_with_index(conn->transit_queues, &tq->ll); |
|
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "transit_q: new group=%016llx src=%016llx dst=%016llx", |
|
(unsigned long long)group_id, (unsigned long long)src_node_id, (unsigned long long)dst_node_id); |
|
return tq; |
|
} |
|
|
|
// Уничтожить транзитную очередь: снять waiter, дропнуть оставшиеся пакеты, удалить из реестра conn. |
|
static void transit_queue_destroy(struct ETCP_CONN* conn, struct TRANSIT_QUEUE* tq) { |
|
if (conn->send_input_q) queue_waiter_cancel(conn->send_input_q, &tq->waiter); |
|
int dropped = 0; |
|
struct ll_entry* e; |
|
while ((e = queue_data_get(tq->q)) != NULL) { free_entry(e); dropped++; } |
|
if (dropped > 0) |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "transit_q: destroyed with %d pending pkts src=%016llx dst=%016llx", |
|
dropped, (unsigned long long)tq->src_node_id, (unsigned long long)tq->dst_node_id); |
|
queue_free(tq->q); |
|
queue_remove_data(conn->transit_queues, &tq->ll); |
|
queue_entry_free(&tq->ll); |
|
} |
|
|
|
// Waiter-коллбэк транзитной очереди: send_input_q освободился → шлём один пакет, |
|
// при непустой очереди снова встаём в waiter, иначе уничтожаем очередь. |
|
static void transit_queue_drain_cb(struct ll_queue* q, void* arg) { |
|
struct TRANSIT_QUEUE* tq = (struct TRANSIT_QUEUE*)arg; |
|
struct ETCP_CONN* conn = tq->conn; |
|
struct ll_entry* e = queue_data_get(tq->q); |
|
if (!e) { transit_queue_destroy(conn, tq); return; } |
|
int ret = etcp_send(conn, e); |
|
if (ret != 0) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "transit_drain: etcp_send failed ret=%d src=%016llx dst=%016llx", |
|
ret, (unsigned long long)tq->src_node_id, (unsigned long long)tq->dst_node_id); |
|
free_entry(e); |
|
} |
|
if (queue_entry_count(tq->q) > 0) |
|
queue_waiter_wait(conn->send_input_q, &tq->waiter, transit_queue_drain_cb, tq); |
|
else |
|
transit_queue_destroy(conn, tq); |
|
} |
|
|
|
// Удалить все транзитные очереди соединения (вызывается при закрытии ETCP_CONN). |
|
void etcp_router_transit_queues_destroy(struct ETCP_CONN* conn) { |
|
if (!conn || !conn->transit_queues) return; |
|
struct ll_entry* entry; |
|
while ((entry = queue_data_get(conn->transit_queues)) != NULL) { |
|
struct TRANSIT_QUEUE* tq = (struct TRANSIT_QUEUE*)entry; |
|
if (conn->send_input_q) queue_waiter_cancel(conn->send_input_q, &tq->waiter); |
|
struct ll_entry* e; |
|
while ((e = queue_data_get(tq->q)) != NULL) free_entry(e); |
|
queue_free(tq->q); |
|
queue_entry_free(&tq->ll); |
|
} |
|
queue_free(conn->transit_queues); |
|
conn->transit_queues = NULL; |
|
} |
|
|
|
// Сбросить recv_conn у всех ROUTER_CONN, ссылающихся на уничтожаемый ETCP_CONN. |
|
// recv_conn — заимствованный указатель; без очистки он висит после u_free(conn) (UAF). |
|
void etcp_router_conn_destroyed(struct ETCP_CONN* conn) { |
|
if (!conn || !conn->instance || !conn->instance->router_conns) return; |
|
struct ll_queue* q = conn->instance->router_conns; |
|
for (uint32_t slot = 0; slot < q->hash_size; slot++) { |
|
struct ll_entry* entry = q->hash_table[slot]; |
|
while (entry) { |
|
struct ll_entry* next = entry->hash_next; |
|
struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)entry; |
|
if (rconn->send_waiter_q == conn->send_input_q) router_cancel_send_waiter(rconn); |
|
if (rconn->recv_conn == conn) { |
|
DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, |
|
"router: clear recv_conn remote=%016llx svc_id=%u (conn destroyed)", |
|
(unsigned long long)rconn->remote_node_id, rconn->svc_id); |
|
rconn->recv_conn = NULL; |
|
} |
|
entry = next; |
|
} |
|
} |
|
} |
|
|
|
// ==================================================================== |
|
// Управление ROUTER_CONN |
|
// ==================================================================== |
|
|
|
// Найти состояние seq-подключения по (group_id, remote_node_id, svc_id). NULL если не создано. |
|
static struct ETCP_ROUTER_CONN* router_conn_find(struct UTUN_INSTANCE* inst, |
|
uint64_t group_id, |
|
uint64_t remote_node_id, uint8_t svc_id) { |
|
if (!inst || !inst->router_conns) return NULL; |
|
uint8_t key[17]; |
|
memcpy(key, &group_id, 8); |
|
memcpy(key + 8, &remote_node_id, 8); |
|
key[16] = svc_id; |
|
return (struct ETCP_ROUTER_CONN*)queue_find_data_by_index(inst->router_conns, key); |
|
} |
|
|
|
// ==================================================================== |
|
// Отправка: build+encode при дренировании send_q, дренирование через waiter на send_input_q |
|
// ==================================================================== |
|
|
|
// Найти ETCP_CONN для отправки (учитывая indirect-посредников). NULL = нет маршрута. |
|
static struct ETCP_CONN* router_route_conn(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t remote_node_id) { |
|
struct TOPO_GROUP* group = topo_groups_find(inst->topo_groups, group_id); |
|
if (group) { |
|
struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(group, remote_node_id); |
|
if (nq && nq->conn_mgr_type == CONN_TYPE_INDIRECT && nq->conn_mgr_intermediariy_count > 0) { |
|
for (uint8_t i = 0; i < nq->conn_mgr_intermediariy_count; i++) { |
|
struct ETCP_CONN* c = topo_group_find_conn_for_node(group, nq->conn_mgr_intermediaries[i]); |
|
if (c) return c; |
|
} |
|
} |
|
return topo_group_find_conn_for_node(group, remote_node_id); |
|
} |
|
/* group_id=0 (или группа не найдена) — глобальная маршрутизация: прямое соединение по node_id */ |
|
return instance_find_conn(inst, remote_node_id); |
|
} |
|
|
|
// Тонкая обёртка над router_route_conn для rconn (next_hop для отправки). |
|
static struct ETCP_CONN* router_send_conn(struct ETCP_ROUTER_CONN* rconn) { |
|
return router_route_conn(rconn->inst, rconn->group_id, rconn->remote_node_id); |
|
} |
|
|
|
// Есть ли физический маршрут до узла (прямой/indirect/глобальный): 1 — есть, 0 — нет. |
|
// Локальная доставка (dst == self) всегда возвращает 1. |
|
int etcp_router_has_route(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id) { |
|
if (!inst) return 0; |
|
if (dst_node_id == inst->node_id) return 1; |
|
return router_route_conn(inst, group_id, dst_node_id) != NULL; |
|
} |
|
|
|
// Собрать и закодировать (подпись/шифрование) пакет в начале отправки. |
|
// Возвращает финальный wire-пакет [SVC_ROUTE_HDR][payload][sig?] (u_malloc). |
|
static int router_build_packet(struct ETCP_ROUTER_CONN* rconn, struct ll_entry* pending, |
|
uint8_t** out, size_t* out_len) { |
|
size_t len = SVC_ROUTE_HDR_SIZE + pending->len; |
|
uint8_t* base = u_calloc(1, len); |
|
if (!base) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router build: allocation failed"); return -1; } |
|
struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)base; |
|
hdr->cmd = ETCP_RT_ID_SVC_ROUTE; |
|
hdr->group_id = rconn->group_id; |
|
hdr->dst_node_id = rconn->remote_node_id; |
|
hdr->src_node_id = rconn->inst->node_id; |
|
memcpy(&hdr->seq, pending->data, 4); |
|
hdr->svc_id = rconn->svc_id; |
|
hdr->reset_id = rconn->reset_id; |
|
hdr->peer_reset_id = rconn->peer_reset_id; |
|
hdr->timestamp = get_current_timestamp(); |
|
memcpy(base + SVC_ROUTE_HDR_SIZE, pending->dgram, pending->len); |
|
int ret = route_crypto_encode(rconn->inst, rconn->remote_node_id, base, len, pending->data[4], out, out_len); |
|
u_free(base); |
|
return ret; |
|
} |
|
|
|
// Управляющие пакеты используют ту же пару идентификаторов, что DATA. |
|
static void router_control(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, uint8_t flags, |
|
uint64_t peer_id, uint64_t challenge) { |
|
if (!conn) conn = router_send_conn(rconn); |
|
if (!conn) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router control: no route svc=%u", rconn->svc_id); return; } |
|
struct ll_entry* e = ll_alloc_lldgram(SVC_ROUTE_HDR_SIZE); |
|
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router control: allocation failed"); return; } |
|
struct SVC_ROUTE_HDR* h = (struct SVC_ROUTE_HDR*)e->dgram; |
|
memset(h, 0, sizeof(*h)); |
|
h->cmd = ETCP_RT_ID_SVC_ROUTE; h->group_id = rconn->group_id; |
|
h->src_node_id = rconn->inst->node_id; h->dst_node_id = rconn->remote_node_id; h->svc_id = rconn->svc_id; |
|
h->flags = flags; h->reset_id = rconn->reset_id; h->peer_reset_id = peer_id; h->challenge = challenge; |
|
e->len = SVC_ROUTE_HDR_SIZE; |
|
DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, |
|
"router control: tx group=%016llx peer=%016llx svc=%u flags=%02x local=%016llx remote=%016llx challenge=%016llx", |
|
(unsigned long long)rconn->group_id, (unsigned long long)rconn->remote_node_id, rconn->svc_id, flags, |
|
(unsigned long long)h->reset_id, (unsigned long long)peer_id, (unsigned long long)challenge); |
|
if (etcp_send(conn, e) != 0) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router control: send failed"); free_entry(e); } |
|
} |
|
|
|
static void router_send_ctrl(struct ETCP_ROUTER_CONN* rconn, uint8_t flags) { |
|
if (rconn->peer_reset_id) router_control(rconn, NULL, flags, rconn->peer_reset_id, 0); |
|
} |
|
|
|
static int router_random_id(uint64_t* id) { |
|
if (random_bytes((uint8_t*)id, sizeof(*id)) != 0 || !*id) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router: failed to generate nonzero session nonce"); |
|
return -1; |
|
} |
|
return 0; |
|
} |
|
|
|
static void router_handshake_start(struct ETCP_ROUTER_CONN* rconn) { |
|
if (rconn->closed || rconn->peer_reset_id || rconn->handshake_timer) return; |
|
router_control(rconn, NULL, ROUTER_FLAG_START, 0, 0); |
|
rconn->handshake_timer = uasync_set_timeout(rconn->inst->ua, ROUTER_RETRANS_TIMEOUT_TB, |
|
rconn, router_handshake_timer_cb, "router_handshake"); |
|
} |
|
|
|
static void router_handshake_timer_cb(void* arg) { |
|
struct ETCP_ROUTER_CONN* rconn = arg; |
|
rconn->handshake_timer = NULL; |
|
if (rconn->closed || rconn->peer_reset_id) return; |
|
router_handshake_start(rconn); |
|
} |
|
|
|
// Незнакомый идентификатор не сбрасывает рабочую сессию: сначала проверяем живость challenge/echo. |
|
static void router_challenge_peer(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, uint64_t peer_id) { |
|
uint64_t now = get_time_tb(); |
|
if (rconn->pending_peer_id && now - rconn->challenge_sent_tb < ROUTER_RETRANS_TIMEOUT_TB) return; |
|
if (router_random_id(&rconn->pending_challenge) != 0) return; |
|
rconn->pending_peer_id = peer_id; |
|
rconn->challenge_sent_tb = now; |
|
router_control(rconn, conn, ROUTER_FLAG_RST, peer_id, rconn->pending_challenge); |
|
} |
|
|
|
// Только ответ на наш свежий challenge разрешает сменить пару эпох. Свой id сохраняем: |
|
// иначе два конца бесконечно перезапускали бы друг друга. |
|
static void router_confirm_peer(struct ETCP_ROUTER_CONN* rconn, const struct SVC_ROUTE_HDR* h) { |
|
if (h->peer_reset_id != rconn->reset_id || !h->challenge || h->challenge != rconn->pending_challenge |
|
|| h->reset_id != rconn->pending_peer_id) { |
|
DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router confirm: stale response peer=%016llx svc=%u", |
|
(unsigned long long)rconn->remote_node_id, rconn->svc_id); |
|
return; |
|
} |
|
uint64_t previous = rconn->peer_reset_id; |
|
uint64_t peer_id = h->reset_id; |
|
rconn->pending_peer_id = 0; rconn->pending_challenge = 0; |
|
if (previous == peer_id) return; |
|
if (previous && router_conn_reset(rconn) != 0) { router_close_and_notify(rconn); return; } |
|
if (rconn->handshake_timer) { uasync_cancel_timeout(rconn->inst->ua, rconn->handshake_timer); rconn->handshake_timer = NULL; } |
|
rconn->peer_reset_id = peer_id; |
|
rconn->peer_sync_done = 1; |
|
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, |
|
"router session: group=%016llx peer=%016llx svc=%u local=%016llx remote=%016llx previous=%016llx", |
|
(unsigned long long)rconn->group_id, (unsigned long long)rconn->remote_node_id, rconn->svc_id, |
|
(unsigned long long)rconn->reset_id, (unsigned long long)peer_id, (unsigned long long)previous); |
|
if (previous) router_send_close_to_service(rconn); |
|
if (!rconn->closed) router_send_kick(rconn); |
|
} |
|
|
|
// Синтетический loopback-conn (static) для доставки сервисного коллбэка при локальных событиях |
|
// (CLOSE, restart) без реального сетевого соединения. |
|
static struct ETCP_CONN* loopback_conn(struct UTUN_INSTANCE* inst) { |
|
static struct ETCP_CONN c; |
|
memset(&c, 0, sizeof(c)); |
|
c.instance = inst; c.peer_node_id = inst->node_id; |
|
return &c; |
|
} |
|
|
|
// Доставить локальному сервису событие закрытия conn (пустая кодограмма ROUTER_SVC_HDR_SIZE). |
|
static void router_send_close_to_service(struct ETCP_ROUTER_CONN* rconn) { |
|
struct UTUN_INSTANCE* inst = rconn->inst; |
|
etcp_recv_fn cb = inst->router_bindings.callbacks[rconn->svc_id]; |
|
if (!cb) return; |
|
struct ll_entry* e = queue_entry_new(0); |
|
if (!e) return; |
|
e->dgram = u_malloc(ROUTER_SVC_HDR_SIZE); |
|
if (!e->dgram) { queue_entry_free(e); return; } |
|
e->dgram[0] = rconn->svc_id; |
|
memcpy(e->dgram + ROUTER_SVC_SRC_OFF, &rconn->remote_node_id, 8); |
|
memcpy(e->dgram + ROUTER_SVC_DST_OFF, &inst->node_id, 8); |
|
e->dgram[ROUTER_SVC_FLAGS_OFF] = 0; |
|
memcpy(e->dgram + ROUTER_SVC_GROUP_OFF, &rconn->group_id, 8); |
|
e->len = ROUTER_SVC_HDR_SIZE; |
|
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_close_svc: svc_id=%u remote=%016llx", |
|
rconn->svc_id, (unsigned long long)rconn->remote_node_id); |
|
cb(loopback_conn(rconn->inst), e); |
|
} |
|
|
|
// Финальный шаг закрытия: освобождает ll entry (вызывается через uasync_call_soon, уже вне очередей). |
|
static void router_close_finalize(void* arg) { |
|
struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; |
|
queue_entry_free(&rconn->ll); |
|
} |
|
|
|
static void router_send_queue_free(void* arg) { queue_free(arg); } |
|
|
|
// Отменить таймеры/waiter и освободить все очереди rconn (общий код для close/restart). |
|
static void router_conn_free_queues(struct ETCP_ROUTER_CONN* rconn) { |
|
struct UTUN_INSTANCE* inst = rconn->inst; |
|
if (rconn->handshake_timer) { uasync_cancel_timeout(inst->ua, rconn->handshake_timer); rconn->handshake_timer = NULL; } |
|
if (rconn->ack_timer) { uasync_cancel_timeout(inst->ua, rconn->ack_timer); rconn->ack_timer = NULL; } |
|
if (rconn->idle_ack_timer) { uasync_cancel_timeout(inst->ua, rconn->idle_ack_timer); rconn->idle_ack_timer = NULL; } |
|
if (rconn->watchdog_timer) { uasync_cancel_timeout(inst->ua, rconn->watchdog_timer); rconn->watchdog_timer = NULL; } |
|
if (rconn->no_route_timer) { uasync_cancel_timeout(inst->ua, rconn->no_route_timer); rconn->no_route_timer = NULL; } |
|
if (rconn->retrans_timer) { uasync_cancel_timeout(inst->ua, rconn->retrans_timer); rconn->retrans_timer = NULL; } |
|
router_cancel_send_waiter(rconn); |
|
struct ll_entry* f; |
|
if (rconn->send_q) { |
|
// Освобождение очереди не должно запускать производителей новых данных. |
|
queue_set_threshold(rconn->send_q, -1, 0); |
|
while ((f = queue_data_get(rconn->send_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); } |
|
// Producer callback может закрыть/перезапустить rconn внутри queue_data_get(send_q). |
|
uasync_call_soon(inst->ua, rconn->send_q, router_send_queue_free); rconn->send_q = NULL; |
|
} |
|
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->recv_q = NULL; |
|
} |
|
if (rconn->inflight_q) { |
|
while ((f = queue_data_get(rconn->inflight_q)) != NULL) { |
|
struct ROUTER_INFLIGHT* inf = (struct ROUTER_INFLIGHT*)f; |
|
if (inf->dgram) u_free(inf->dgram); |
|
u_free(inf); |
|
} |
|
queue_free(rconn->inflight_q); rconn->inflight_q = NULL; |
|
} |
|
if (rconn->incoming_q) { |
|
while ((f = queue_data_get(rconn->incoming_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); } |
|
queue_resume_callback(rconn->incoming_q); |
|
queue_free(rconn->incoming_q); rconn->incoming_q = NULL; |
|
} |
|
} |
|
|
|
// Пересоздать очереди и сбросить всё состояние rconn (кроме identity: group/remote/svc/ll.data/inst/reset_id/last_dgram_ts). |
|
static int router_conn_reset(struct ETCP_ROUTER_CONN* rconn) { |
|
router_conn_free_queues(rconn); |
|
|
|
rconn->send_q = queue_new(rconn->inst->ua, 0, 0, 0, "router_send_q"); |
|
rconn->recv_q = queue_new(rconn->inst->ua, ROUTER_RECVQ_HASH_SIZE, 0, 4, "router_recv_q"); |
|
rconn->inflight_q = queue_new(rconn->inst->ua, ROUTER_INFLIGHT_HASH_SIZE, 0, 4, "router_inflight_q"); |
|
rconn->incoming_q = queue_new(rconn->inst->ua, 0, 0, 0, "router_incoming_q"); |
|
if (!rconn->send_q || !rconn->recv_q || !rconn->inflight_q || !rconn->incoming_q) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router reset: queue allocation failed"); |
|
router_conn_free_queues(rconn); return -1; |
|
} |
|
queue_set_threshold(rconn->send_q, ROUTER_MAX_SEND_Q_PACKETS - 1, 0); |
|
queue_set_callback(rconn->incoming_q, router_incoming_q_cb, rconn); |
|
queue_set_threshold(rconn->incoming_q, 0, 0); |
|
rconn->incoming_data_ready = 0; |
|
rconn->recv_conn = NULL; |
|
|
|
rconn->rtt = 0; rconn->rtt_jitter = 0; |
|
rconn->max_inflight = ROUTER_MAX_INFLIGHT; |
|
rconn->inflight_limit = ROUTER_MAX_INFLIGHT; |
|
rconn->minrtt = MINRTT_DEFAULT_TB; |
|
memset(rconn->minrtt_window, 0, sizeof(rconn->minrtt_window)); |
|
memset(rconn->minrtt_window_tb, 0, sizeof(rconn->minrtt_window_tb)); |
|
rconn->minrtt_window_idx = 0; rconn->minrtt_window_count = 0; rconn->minrtt_probe = 0; |
|
rconn->last_inflight_loaded_tb = 0; rconn->last_inflight_unloaded_tb = 0; rconn->inflight_was_loaded = 0; |
|
rconn->last_recv_pkt_ts = 0; rconn->last_recv_pkt_local_tb = 0; rconn->last_recv_updated = 0; |
|
|
|
rconn->tx_seq = 0; rconn->tx_sent = 0; rconn->rx_seq = 0; rconn->tx_acked = 0; |
|
rconn->retrans_seq = 0; rconn->retrans_end = 0; |
|
rconn->last_sent_ack_seq = 0; rconn->last_ack_sent_tb = 0; |
|
rconn->send_blocked = 0; rconn->start_sent = 0; rconn->peer_sync_done = 0; |
|
rconn->peer_reset_id = 0; |
|
rconn->no_route = 0; rconn->no_route_timer = NULL; rconn->no_ack_count = 0; rconn->closed = 0; |
|
rconn->c_pkts_sent = 0; rconn->c_pkts_send_err = 0; rconn->c_pkts_rcvd = 0; |
|
rconn->c_ack_sent = 0; rconn->c_ack_recv = 0; rconn->c_retrans_done = 0; |
|
rconn->c_dup_dropped = 0; rconn->c_oob_dropped = 0; rconn->c_stale_ack = 0; rconn->c_sign_fail = 0; |
|
rconn->last_ack_changed_tb = 0; |
|
return 0; |
|
} |
|
|
|
// Закрыть conn: уведомить локальный сервис, отправить CLOSE удалённой стороне, очистить |
|
// очереди/таймеры, удалить из реестра router_conns и отложить освобождение ll через call_soon. |
|
static void router_close_and_notify(struct ETCP_ROUTER_CONN* rconn) { |
|
if (!rconn || rconn->closed) return; |
|
struct UTUN_INSTANCE* inst = rconn->inst; |
|
rconn->closed = 1; |
|
router_send_close_to_service(rconn); |
|
// Отправляем CLOSE удалённой стороне + снимаем waiter отправки |
|
if (router_send_conn(rconn)) router_send_ctrl(rconn, ROUTER_FLAG_CLOSE); |
|
router_conn_free_queues(rconn); |
|
if (inst->router_conns) |
|
queue_remove_data(inst->router_conns, &rconn->ll); |
|
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_conn: closed group=%016llx remote=%016llx svc_id=%u", |
|
(unsigned long long)rconn->group_id, (unsigned long long)rconn->remote_node_id, rconn->svc_id); |
|
uasync_call_soon(inst->ua, rconn, router_close_finalize); |
|
} |
|
|
|
// Пометить отсутствие маршрута и взвести таймер повторных проверок (идемпотентно). |
|
static void router_set_no_route(struct ETCP_ROUTER_CONN* rconn) { |
|
if (rconn->no_route) return; |
|
rconn->no_route = 1; |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: no route to %016llx svc_id=%u group=%016llx — send_q=%d queued, will retry", |
|
(unsigned long long)rconn->remote_node_id, rconn->svc_id, |
|
(unsigned long long)rconn->group_id, queue_entry_count(rconn->send_q)); |
|
if (!rconn->no_route_timer) |
|
rconn->no_route_timer = uasync_set_timeout(rconn->inst->ua, |
|
ROUTER_NO_ROUTE_RETRY_TB, rconn, router_no_route_retry_cb, "router_no_route"); |
|
} |
|
|
|
// Попытаться продвинуть отправку: зарегистрировать waiter на send_input_q, если есть что слать. |
|
// Инвариант: send_q непуст ∧ !closed ∧ !no_route ∧ inflight свободен ⇒ waiter зарегистрирован. |
|
static void router_send_kick(struct ETCP_ROUTER_CONN* rconn) { |
|
if (rconn->closed || rconn->no_route) return; |
|
if (!rconn->peer_reset_id) { if (queue_entry_count(rconn->send_q)) router_handshake_start(rconn); return; } |
|
int retry = rconn->retrans_seq != rconn->retrans_end; |
|
if (!retry && queue_entry_count(rconn->send_q) == 0) return; |
|
if (!retry && queue_entry_count(rconn->inflight_q) >= (int)router_effective_max_inflight(rconn)) { |
|
rconn->send_blocked = 1; |
|
return; |
|
} |
|
struct ETCP_CONN* conn = router_send_conn(rconn); |
|
if (!conn || !conn->send_input_q || conn->state == 2) { router_cancel_send_waiter(rconn); router_set_no_route(rconn); return; } |
|
if (rconn->send_waiter_q != conn->send_input_q) router_cancel_send_waiter(rconn); |
|
if (rconn->send_waiter.internal || rconn->send_waiter.call_soon_id) return; |
|
rconn->send_waiter_q = conn->send_input_q; |
|
if (queue_waiter_wait(conn->send_input_q, &rconn->send_waiter, router_send_drain_cb, rconn) < 0) |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router send: waiter registration failed svc=%u", rconn->svc_id); |
|
} |
|
|
|
// Waiter-коллбэк: send_input_q опустел → кодируем и шлём один пакет, при необходимости |
|
// снова встаём в хвост (round-robin). Успешно отправленный пакет кладём в inflight_q для ретрансмита. |
|
static void router_send_drain_cb(struct ll_queue* q, void* arg) { |
|
struct ETCP_ROUTER_CONN* rconn = arg; |
|
rconn->send_waiter_q = NULL; |
|
if (rconn->closed || rconn->no_route) return; |
|
struct ETCP_CONN* conn = router_send_conn(rconn); |
|
if (!conn || conn->send_input_q != q) { router_send_kick(rconn); return; } |
|
while (rconn->retrans_seq != rconn->retrans_end) { |
|
uint32_t seq = rconn->retrans_seq; |
|
struct ROUTER_INFLIGHT* inf = (struct ROUTER_INFLIGHT*)queue_find_data_by_index(rconn->inflight_q, &seq); |
|
if (!inf) { rconn->retrans_seq++; continue; } |
|
if (router_retransmit_one(rconn, inf) != 0) return; |
|
rconn->retrans_seq++; |
|
router_send_kick(rconn); |
|
return; |
|
} |
|
if (queue_entry_count(rconn->inflight_q) >= (int)router_effective_max_inflight(rconn)) { |
|
rconn->send_blocked = 1; |
|
return; |
|
} |
|
struct ll_entry* pending = rconn->send_q->head; |
|
if (!pending) { rconn->send_blocked = 0; router_send_watchdog_disarm(rconn); return; } |
|
struct ll_entry* e = queue_entry_new(0); |
|
struct ROUTER_INFLIGHT* inf = u_calloc(1, sizeof(*inf)); |
|
if (!e || !inf) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router send: allocation failed; packet retained"); |
|
if (e) free_entry(e); |
|
u_free(inf); |
|
return; |
|
} |
|
size_t len; |
|
if (router_build_packet(rconn, pending, &inf->dgram, &len) != 0) { free_entry(e); u_free(inf); return; } |
|
e->dgram = u_malloc(len); |
|
if (!e->dgram) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router send: wire copy failed; packet retained"); |
|
free_entry(e); u_free(inf->dgram); u_free(inf); return; |
|
} |
|
memcpy(e->dgram, inf->dgram, len); e->len = len; |
|
memcpy(&inf->seq, pending->data, 4); inf->ll.size = 4; |
|
inf->dgram_len = len; inf->last_sent_tb = get_time_tb(); inf->send_count = 1; |
|
uint32_t seq = inf->seq; |
|
queue_data_put_with_index(rconn->inflight_q, &inf->ll); |
|
rconn->tx_sent = seq + 1; |
|
if (etcp_send(conn, e) != 0) { |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router send failed svc=%u seq=%u; packet retained", rconn->svc_id, seq); |
|
rconn->tx_sent = seq; |
|
queue_remove_data(rconn->inflight_q, &inf->ll); |
|
free_entry(e); u_free(inf->dgram); u_free(inf); rconn->c_pkts_send_err++; |
|
return; |
|
} |
|
uint64_t local_id = rconn->reset_id, peer_id = rconn->peer_reset_id; |
|
free_entry(queue_data_get(rconn->send_q)); |
|
if (rconn->closed || rconn->reset_id != local_id || rconn->peer_reset_id != peer_id) return; |
|
rconn->c_pkts_sent++; |
|
rconn->last_dgram_ts = get_current_timestamp(); |
|
if (!rconn->retrans_timer) router_retrans_schedule(rconn); |
|
router_track_inflight_state(rconn); |
|
DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "router sent: peer=%016llx svc=%u seq=%u acked=%u inflight=%d pending=%d", |
|
(unsigned long long)rconn->remote_node_id, rconn->svc_id, seq, rconn->tx_acked, |
|
queue_entry_count(rconn->inflight_q), queue_entry_count(rconn->send_q)); |
|
router_send_kick(rconn); |
|
} |
|
|
|
// DEBUG-watchdog: медленная (500мс) проверка инварианта drain. Если send_q непуст, |
|
// канал свободен, но waiter не зарегистрирован — инвариант нарушен: логируем и форсируем kick. |
|
static void router_send_watchdog_cb(void* arg) { |
|
struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; |
|
rconn->watchdog_timer = NULL; |
|
if (rconn->closed) return; |
|
if (queue_entry_count(rconn->send_q) == 0) return; |
|
// Маршрут мог смениться, пока waiter остаётся на занятой очереди прежнего next hop. |
|
if (rconn->peer_reset_id && !rconn->no_route) router_send_kick(rconn); |
|
if (queue_entry_count(rconn->send_q) > 0 && !rconn->watchdog_timer) |
|
rconn->watchdog_timer = uasync_set_timeout(rconn->inst->ua, |
|
ROUTER_SEND_WATCHDOG_TB, rconn, router_send_watchdog_cb, "router_send_watchdog"); |
|
} |
|
|
|
// Взвести watchdog пока send_q непуст (страховка от пропущенного re-arm). |
|
static void router_send_watchdog_arm(struct ETCP_ROUTER_CONN* rconn) { |
|
if (rconn->watchdog_timer || queue_entry_count(rconn->send_q) == 0) return; |
|
rconn->watchdog_timer = uasync_set_timeout(rconn->inst->ua, |
|
ROUTER_SEND_WATCHDOG_TB, rconn, router_send_watchdog_cb, "router_send_watchdog"); |
|
} |
|
|
|
// Снять watchdog когда send_q опустел. |
|
static void router_send_watchdog_disarm(struct ETCP_ROUTER_CONN* rconn) { |
|
if (rconn->watchdog_timer) { uasync_cancel_timeout(rconn->inst->ua, rconn->watchdog_timer); rconn->watchdog_timer = NULL; } |
|
} |
|
|
|
// Таймер повторной проверки маршрута (20ms): при появлении — снять no_route и возобновить отправку. |
|
static void router_no_route_retry_cb(void* arg) { |
|
struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; |
|
if (rconn->closed) return; |
|
rconn->no_route_timer = NULL; |
|
if (router_send_conn(rconn)) { |
|
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router: route appeared for %016llx svc_id=%u, resuming send_q=%d", |
|
(unsigned long long)rconn->remote_node_id, rconn->svc_id, queue_entry_count(rconn->send_q)); |
|
rconn->no_route = 0; |
|
router_send_kick(rconn); |
|
return; |
|
} |
|
rconn->no_route_timer = uasync_set_timeout(rconn->inst->ua, |
|
ROUTER_NO_ROUTE_RETRY_TB, rconn, router_no_route_retry_cb, "router_no_route"); |
|
} |
|
|
|
// ==================================================================== |
|
// Ретрансмиты |
|
// ==================================================================== |
|
|
|
// Отправить повторно один inflight-пакет (финальный wire-пакет уже закодирован — шлём копию как есть). |
|
static int router_retransmit_one(struct ETCP_ROUTER_CONN* rconn, struct ROUTER_INFLIGHT* inf) { |
|
struct ETCP_CONN* conn = router_send_conn(rconn); |
|
if (!conn) { |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router_retransmit: no route to %016llx svc_id=%u", |
|
(unsigned long long)rconn->remote_node_id, rconn->svc_id); |
|
rconn->c_pkts_send_err++; return -1; |
|
} |
|
// Финальный пакет уже закодирован (подпись/шифрование) — шлём как есть. |
|
struct ll_entry* entry = queue_entry_new(0); |
|
if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router retransmit: allocation failed"); rconn->c_pkts_send_err++; return -1; } |
|
entry->dgram = u_malloc(inf->dgram_len); |
|
if (!entry->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router retransmit: payload allocation failed"); queue_entry_free(entry); rconn->c_pkts_send_err++; return -1; } |
|
memcpy(entry->dgram, inf->dgram, inf->dgram_len); |
|
entry->len = inf->dgram_len; |
|
|
|
rconn->last_dgram_ts = get_current_timestamp(); |
|
DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "RETRANS: svc_id=%u seq=%u attempt=%d → %016llx", |
|
rconn->svc_id, inf->seq, inf->send_count + 1, (unsigned long long)rconn->remote_node_id); |
|
int ret = etcp_send(conn, entry); |
|
if (ret != 0) { |
|
queue_dgram_free(entry); queue_entry_free(entry); |
|
rconn->c_pkts_send_err++; |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router retransmit: send failed seq=%u", inf->seq); |
|
} else { |
|
inf->last_sent_tb = get_time_tb(); |
|
inf->send_count++; |
|
rconn->c_retrans_done++; |
|
rconn->c_pkts_sent++; |
|
} |
|
return ret; |
|
} |
|
|
|
// Взвести таймер ретрансмита на остаток до ROUTER_RETRANS_TIMEOUT_TB от last_ack_changed_tb (идемпотентно). |
|
static void router_retrans_schedule(struct ETCP_ROUTER_CONN* rconn) { |
|
if (rconn->retrans_timer) return; |
|
uint64_t now = get_time_tb(); |
|
if (rconn->last_ack_changed_tb == 0) rconn->last_ack_changed_tb = now; |
|
int64_t remaining = (int64_t)ROUTER_RETRANS_TIMEOUT_TB - (int64_t)(now - rconn->last_ack_changed_tb); |
|
if (remaining <= 0) remaining = 1; |
|
rconn->retrans_timer = uasync_set_timeout(rconn->inst->ua, |
|
(uint32_t)remaining, rconn, router_retrans_timer_cb, "router_retrans"); |
|
} |
|
|
|
// Таймер ретрансмита: при застое ACK (ROUTER_RETRANS_TIMEOUT_TB) переотправляет все inflight-пакеты. |
|
// После ROUTER_NO_ACK_MAX_RETRANS циклов без прогресса — закрывает conn. |
|
static void router_retrans_timer_cb(void* arg) { |
|
struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; |
|
if (rconn->closed) return; |
|
rconn->retrans_timer = NULL; |
|
uint64_t now = get_time_tb(); |
|
uint64_t elapsed = now - rconn->last_ack_changed_tb; |
|
|
|
if (elapsed >= ROUTER_RETRANS_TIMEOUT_TB) { |
|
rconn->no_ack_count++; |
|
if (rconn->no_ack_count >= ROUTER_NO_ACK_MAX_RETRANS) { |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router: no ACK progress for %d retrans cycles, closing svc_id=%u remote=%016llx", |
|
rconn->no_ack_count, rconn->svc_id, (unsigned long long)rconn->remote_node_id); |
|
router_close_and_notify(rconn); |
|
return; |
|
} |
|
if (rconn->retrans_seq == rconn->retrans_end) { |
|
rconn->retrans_seq = rconn->tx_acked; |
|
rconn->retrans_end = rconn->tx_sent; |
|
DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router retransmit: peer=%016llx svc=%u range=[%u,%u) cycle=%u", |
|
(unsigned long long)rconn->remote_node_id, rconn->svc_id, |
|
rconn->retrans_seq, rconn->retrans_end, rconn->no_ack_count); |
|
} |
|
router_send_kick(rconn); |
|
rconn->last_ack_changed_tb = now; |
|
} |
|
|
|
if (queue_entry_count(rconn->inflight_q) > 0) { |
|
int64_t remaining = (int64_t)ROUTER_RETRANS_TIMEOUT_TB - (int64_t)(now - rconn->last_ack_changed_tb); |
|
if (remaining <= 0) remaining = 1; |
|
rconn->retrans_timer = uasync_set_timeout(rconn->inst->ua, |
|
(uint32_t)remaining, rconn, router_retrans_timer_cb, "router_retrans"); |
|
} |
|
} |
|
|
|
// Собрать+закодировать пакет в начале отправки и положить в send_q. |
|
static int router_enqueue_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* payload, size_t len, int force, int mode) { |
|
size_t overhead = SVC_ROUTE_HDR_SIZE + ((mode & ROUTE_CRYPTO_SIGN) ? SC_SIGN_SIZE : 0) |
|
+ ((mode & ROUTE_CRYPTO_ENCRYPT) ? SC_NONCE_SIZE + SC_CRC32_SIZE + SC_TAG_SIZE : 0); |
|
if (rconn->closed || !payload || !len || len > PKTNORM_MAX_DGRAM_SIZE - overhead || (mode & ~(ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT))) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router send: invalid payload len=%zu mode=%x closed=%u", len, mode, rconn->closed); |
|
return -1; |
|
} |
|
if (!force && queue_entry_count(rconn->send_q) >= ROUTER_MAX_SEND_Q_PACKETS) return -1; |
|
struct ll_entry* e = queue_entry_new(5); |
|
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router enqueue: allocation failed"); return -1; } |
|
e->dgram = u_malloc(len); |
|
if (!e->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router enqueue: payload allocation failed"); queue_entry_free(e); return -1; } |
|
memcpy(e->data, &rconn->tx_seq, 4); e->data[4] = mode; |
|
memcpy(e->dgram, payload, len); e->len = len; |
|
rconn->tx_seq++; |
|
queue_data_put(rconn->send_q, e); |
|
router_send_kick(rconn); |
|
router_send_watchdog_arm(rconn); |
|
return 0; |
|
} |
|
|
|
// ==================================================================== |
|
// minRTT helpers |
|
// ==================================================================== |
|
|
|
// Эффективный лимит inflight для текущего режима (probe или обычный). |
|
static uint32_t router_effective_max_inflight(struct ETCP_ROUTER_CONN* rconn) { |
|
return rconn->inflight_limit; |
|
} |
|
|
|
// Пересчитать inflight_limit (=MINRTT_PROBE_MAX_INFLIGHT в probe-режиме, иначе max_inflight) и |
|
// скорректировать send_blocked / возобновить отправку при росте лимита. |
|
static void router_update_inflight_limit(struct ETCP_ROUTER_CONN* rconn) { |
|
uint32_t new_limit = rconn->minrtt_probe ? MINRTT_PROBE_MAX_INFLIGHT : rconn->max_inflight; |
|
if (new_limit == rconn->inflight_limit) return; |
|
uint32_t old = rconn->inflight_limit; |
|
rconn->inflight_limit = new_limit; |
|
uint32_t inflight = (uint32_t)queue_entry_count(rconn->inflight_q); |
|
|
|
if (new_limit < old && inflight >= new_limit) { |
|
if (!rconn->send_blocked) rconn->send_blocked = 1; |
|
} |
|
if (new_limit > old && (rconn->send_blocked || queue_entry_count(rconn->send_q) > 0)) { |
|
rconn->send_blocked = 0; |
|
router_send_kick(rconn); |
|
} |
|
} |
|
|
|
// Установить рабочий max_inflight (не ниже MINRTT_PROBE_MAX_INFLIGHT), пересчитать лимит и состояние канала. |
|
void etcp_router_set_max_inflight(struct ETCP_ROUTER_CONN* rconn, uint16_t new_max) { |
|
if (!rconn) return; |
|
if (new_max < MINRTT_PROBE_MAX_INFLIGHT) new_max = MINRTT_PROBE_MAX_INFLIGHT; |
|
rconn->max_inflight = new_max; |
|
if (!rconn->minrtt_probe) router_update_inflight_limit(rconn); |
|
router_track_inflight_state(rconn); |
|
} |
|
|
|
// Отслеживать состояние загрузки канала: фиксирует моменты перехода loaded/unloaded |
|
// (inflight >= max_inflight/2) для последующей оценки minRTT. |
|
static void router_track_inflight_state(struct ETCP_ROUTER_CONN* rconn) { |
|
uint32_t inflight = (uint32_t)queue_entry_count(rconn->inflight_q); |
|
uint32_t threshold = rconn->max_inflight / 2; |
|
uint8_t is_loaded = (inflight >= threshold); |
|
if (is_loaded && !rconn->inflight_was_loaded) { |
|
rconn->last_inflight_loaded_tb = get_time_tb(); |
|
rconn->inflight_was_loaded = 1; |
|
} else if (!is_loaded && rconn->inflight_was_loaded) { |
|
rconn->last_inflight_unloaded_tb = get_time_tb(); |
|
rconn->inflight_was_loaded = 0; |
|
} |
|
} |
|
|
|
// Обновить minRTT: в probe-режиме (старые замеры > MINRTT_PROBE_TIMEOUT_TB) ограничить inflight, |
|
// при разгруженном канале достаточно долго — добавить свежий замер rtt в окно и пересчитать среднее. |
|
static void router_update_minrtt(struct ETCP_ROUTER_CONN* rconn) { |
|
uint64_t now = get_time_tb(); |
|
uint16_t w_idx = rconn->minrtt_window_idx; |
|
uint8_t w_cnt = rconn->minrtt_window_count; |
|
|
|
// Проверка probe: свежесть самого старого замера в окне |
|
uint8_t prev_probe = rconn->minrtt_probe; |
|
if (w_cnt > 0) { |
|
uint8_t oldest = (w_idx - w_cnt + MINRTT_WINDOW_SIZE) % MINRTT_WINDOW_SIZE; |
|
uint64_t oldest_age = now - rconn->minrtt_window_tb[oldest]; |
|
if (w_cnt >= MINRTT_WINDOW_SIZE && oldest_age < MINRTT_PROBE_TIMEOUT_TB) { |
|
rconn->minrtt_probe = 0; |
|
} else if (oldest_age >= MINRTT_PROBE_TIMEOUT_TB) { |
|
rconn->minrtt_probe = 1; |
|
} |
|
} |
|
if (prev_probe != rconn->minrtt_probe) router_update_inflight_limit(rconn); |
|
|
|
// Обновить minRTT, если канал разгружен достаточно долго |
|
if (!rconn->inflight_was_loaded) { |
|
uint64_t unloaded_dur = now - rconn->last_inflight_unloaded_tb; |
|
uint64_t min_dur = (uint64_t)rconn->rtt * 2; |
|
if (min_dur < MINRTT_UNLOADED_MIN_TB) min_dur = MINRTT_UNLOADED_MIN_TB; |
|
if (unloaded_dur >= min_dur) { |
|
rconn->minrtt_window[w_idx] = rconn->rtt; |
|
rconn->minrtt_window_tb[w_idx] = now; |
|
rconn->minrtt_window_idx = (w_idx + 1) % MINRTT_WINDOW_SIZE; |
|
if (w_cnt < MINRTT_WINDOW_SIZE) w_cnt++; |
|
rconn->minrtt_window_count = w_cnt; |
|
uint32_t sum = 0, cnt = 0; |
|
for (int i = 0; i < MINRTT_WINDOW_SIZE; i++) { |
|
if (rconn->minrtt_window[i] > 0) { sum += rconn->minrtt_window[i]; cnt++; } |
|
} |
|
rconn->minrtt = cnt > 0 ? (uint16_t)(sum / cnt) : MINRTT_DEFAULT_TB; |
|
DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "MINRTT_UPD: svc_id=%u minrtt=%u (rtt=%u cnt=%u probe=%d unloaded=%lu)", |
|
rconn->svc_id, rconn->minrtt, rconn->rtt, cnt, rconn->minrtt_probe, (unsigned long)unloaded_dur); |
|
} |
|
} |
|
} |
|
|
|
// ==================================================================== |
|
// ACK timers |
|
// ==================================================================== |
|
|
|
// Тонкая обёртка для единообразного вызова отправки ACK из таймеров. |
|
static void router_ack_do_send(struct ETCP_ROUTER_CONN* rconn) { |
|
router_send_ack(rconn); |
|
} |
|
|
|
// Отправить ACK с текущим rx_seq (header-only пакет, seq=rx_seq, payload пуст). |
|
// timestamp — эхо последнего data-пакета пира с поправкой на локальное время (для RTT). |
|
static void router_send_ack(struct ETCP_ROUTER_CONN* rconn) { |
|
struct UTUN_INSTANCE* inst = rconn->inst; |
|
struct ETCP_CONN* conn = rconn->recv_conn; |
|
if (!conn) conn = topo_group_find_conn_for_node(topo_groups_find(inst->topo_groups, rconn->group_id), rconn->remote_node_id); |
|
if (!conn) { |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router_ack: no route to %016llx svc_id=%u", |
|
(unsigned long long)rconn->remote_node_id, rconn->svc_id); |
|
rconn->last_ack_sent_tb = get_time_tb(); |
|
return; |
|
} |
|
struct SVC_ROUTE_HDR* hdr = (struct SVC_ROUTE_HDR*)u_malloc(SVC_ROUTE_HDR_SIZE); |
|
if (!hdr) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_ack: u_malloc failed"); return; } |
|
hdr->cmd = ETCP_RT_ID_SVC_ROUTE; |
|
hdr->group_id = rconn->group_id; |
|
hdr->dst_node_id = rconn->remote_node_id; |
|
hdr->src_node_id = inst->node_id; |
|
hdr->seq = rconn->rx_seq; |
|
hdr->svc_id = rconn->svc_id; |
|
hdr->flags = 0; |
|
hdr->reset_id = rconn->reset_id; |
|
hdr->peer_reset_id = rconn->peer_reset_id; |
|
hdr->challenge = 0; |
|
if (rconn->last_recv_updated) { |
|
uint64_t now = get_time_tb(); |
|
hdr->timestamp = rconn->last_recv_pkt_ts + (uint32_t)(now - rconn->last_recv_pkt_local_tb); |
|
rconn->last_recv_updated = 0; |
|
} else { |
|
hdr->timestamp = 0; |
|
} |
|
|
|
struct ll_entry* entry = queue_entry_new(0); |
|
if (!entry) { u_free(hdr); return; } |
|
entry->dgram = (uint8_t*)hdr; |
|
entry->len = SVC_ROUTE_HDR_SIZE; |
|
|
|
DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "ACK_SEND: svc_id=%u rx_seq=%u → %016llx", |
|
rconn->svc_id, rconn->rx_seq, (unsigned long long)rconn->remote_node_id); |
|
int ret = etcp_send(conn, entry); |
|
if (ret != 0) { |
|
queue_dgram_free(entry); queue_entry_free(entry); |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_ack: etcp_send failed ret=%d svc_id=%u rx_seq=%u", ret, rconn->svc_id, rconn->rx_seq); |
|
return; |
|
} |
|
rconn->c_ack_sent++; |
|
rconn->last_sent_ack_seq = rconn->rx_seq; |
|
rconn->last_ack_sent_tb = get_time_tb(); |
|
} |
|
|
|
// Запланировать отправку ACK: снять idle-таймер, и либо отправить сразу (интервал истёк), |
|
// либо взвести ack_timer на остаток ROUTER_ACK_INTERVAL_TB. |
|
static void router_schedule_ack(struct ETCP_ROUTER_CONN* rconn) { |
|
if (rconn->idle_ack_timer) { |
|
uasync_cancel_timeout(rconn->inst->ua, rconn->idle_ack_timer); |
|
rconn->idle_ack_timer = NULL; |
|
} |
|
if (rconn->rx_seq == rconn->last_sent_ack_seq) return; |
|
uint64_t now = get_time_tb(); |
|
uint64_t elapsed = now - rconn->last_ack_sent_tb; |
|
if (elapsed >= ROUTER_ACK_INTERVAL_TB) { |
|
router_ack_do_send(rconn); |
|
return; |
|
} |
|
if (!rconn->ack_timer) |
|
rconn->ack_timer = uasync_set_timeout(rconn->inst->ua, |
|
ROUTER_ACK_INTERVAL_TB - (uint32_t)elapsed, rconn, router_ack_timer_cb, "router_ack"); |
|
} |
|
|
|
// Периодический ACK-таймер: отправить ACK при неподтверждённом rx_seq, иначе взвести idle-таймер. |
|
static void router_ack_timer_cb(void* arg) { |
|
struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; |
|
if (rconn->closed) return; |
|
rconn->ack_timer = NULL; |
|
if (rconn->rx_seq != rconn->last_sent_ack_seq) |
|
router_ack_do_send(rconn); |
|
if (rconn->rx_seq != rconn->last_sent_ack_seq) { |
|
if (!rconn->ack_timer) |
|
rconn->ack_timer = uasync_set_timeout(rconn->inst->ua, |
|
ROUTER_ACK_INTERVAL_TB, rconn, router_ack_timer_cb, "router_ack"); |
|
} else if (!rconn->idle_ack_timer) { |
|
rconn->idle_ack_timer = uasync_set_timeout(rconn->inst->ua, |
|
ROUTER_ACK_IDLE_TB, rconn, router_idle_ack_timer_cb, "router_idle_ack"); |
|
} |
|
} |
|
|
|
// Idle-таймер: дослать последний ACK после паузы в приёме (чтобы пир не висел на ретрансмитах). |
|
static void router_idle_ack_timer_cb(void* arg) { |
|
struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; |
|
if (rconn->closed) return; |
|
rconn->idle_ack_timer = NULL; |
|
if (rconn->rx_seq != rconn->last_sent_ack_seq) |
|
router_ack_do_send(rconn); |
|
} |
|
|
|
// ==================================================================== |
|
// Reorder / assembly (аналог etcp_output_try_assembly) |
|
// ==================================================================== |
|
|
|
// Доставка сервисной кодограммы в едином формате [svc_id][src][dst][rx_flags][group_id][payload]. |
|
// Для loopback (dst == self) src и dst оба = self, rx_flags = 0. |
|
static void router_deliver_loopback(struct UTUN_INSTANCE* inst, uint64_t group_id, uint8_t svc_id, struct ll_entry* entry) { |
|
etcp_recv_fn cb = inst->router_bindings.callbacks[svc_id]; |
|
if (!cb) { queue_dgram_free(entry); queue_entry_free(entry); return; } |
|
size_t payload_len = entry->len - 1; |
|
struct ll_entry* e = queue_entry_new(0); |
|
if (!e) { queue_dgram_free(entry); queue_entry_free(entry); return; } |
|
e->dgram = u_malloc(ROUTER_SVC_HDR_SIZE + payload_len); |
|
if (!e->dgram) { queue_entry_free(e); queue_dgram_free(entry); queue_entry_free(entry); return; } |
|
e->dgram[0] = svc_id; |
|
memcpy(e->dgram + ROUTER_SVC_SRC_OFF, &inst->node_id, 8); |
|
memcpy(e->dgram + ROUTER_SVC_DST_OFF, &inst->node_id, 8); |
|
e->dgram[ROUTER_SVC_FLAGS_OFF] = 0; |
|
memcpy(e->dgram + ROUTER_SVC_GROUP_OFF, &group_id, 8); |
|
if (payload_len > 0) memcpy(e->dgram + ROUTER_SVC_PAYLOAD_OFF, entry->dgram + 1, payload_len); |
|
e->len = ROUTER_SVC_HDR_SIZE + payload_len; |
|
queue_dgram_free(entry); queue_entry_free(entry); |
|
cb(loopback_conn(inst), e); |
|
} |
|
|
|
// Доставка: декодирует (подпись/шифрование) перед вызовом сервисного коллбэка. |
|
// wire = [SVC_ROUTE_HDR][payload][sig?] — финальный пакет, лежавший в recv_q. |
|
static int router_deliver(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, |
|
const uint8_t* wire, size_t wire_len) { |
|
struct UTUN_INSTANCE* inst = rconn->inst; |
|
etcp_recv_fn cb = inst->router_bindings.callbacks[rconn->svc_id]; |
|
if (!cb) { |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router_deliver: no handler for svc_id=%u", rconn->svc_id); |
|
return -1; |
|
} |
|
const struct SVC_ROUTE_HDR* hdr = (const struct SVC_ROUTE_HDR*)wire; |
|
const uint8_t* payload = wire + SVC_ROUTE_HDR_SIZE; |
|
size_t payload_len = wire_len - SVC_ROUTE_HDR_SIZE; |
|
uint8_t rx_flags = hdr->flags & (ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED); |
|
size_t entry_len = ROUTER_SVC_HDR_SIZE + payload_len; |
|
struct ll_entry* svc_entry = queue_entry_new(0); |
|
if (!svc_entry) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_deliver: queue_entry_new failed"); return -1; } |
|
svc_entry->len = entry_len; |
|
svc_entry->dgram = u_malloc(entry_len); |
|
if (!svc_entry->dgram) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router deliver: payload allocation failed"); queue_entry_free(svc_entry); return -1; } |
|
svc_entry->dgram[0] = rconn->svc_id; |
|
memcpy(svc_entry->dgram + ROUTER_SVC_SRC_OFF, &rconn->remote_node_id, 8); |
|
memcpy(svc_entry->dgram + ROUTER_SVC_DST_OFF, &inst->node_id, 8); |
|
svc_entry->dgram[ROUTER_SVC_FLAGS_OFF] = rx_flags; |
|
memcpy(svc_entry->dgram + ROUTER_SVC_GROUP_OFF, &rconn->group_id, 8); |
|
if (payload_len > 0) memcpy(svc_entry->dgram + ROUTER_SVC_PAYLOAD_OFF, payload, payload_len); |
|
|
|
|
|
DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "router_deliver: svc_id=%u len=%zu flags=%02x src=%016llx dst=%016llx", |
|
rconn->svc_id, payload_len, rx_flags, (unsigned long long)rconn->remote_node_id, |
|
(unsigned long long)inst->node_id); |
|
rconn->c_pkts_rcvd++; |
|
cb(conn, svc_entry); |
|
return 0; |
|
} |
|
|
|
// Собрать доставку из recv_q: последовательно доставлять пакеты по возрастанию rx_seq, пока есть следующий. |
|
static void router_try_assembly(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn) { |
|
uint32_t next = rconn->rx_seq; |
|
while (1) { |
|
struct ll_entry* entry = queue_find_data_by_index(rconn->recv_q, &next); |
|
if (!entry) break; |
|
queue_remove_data(rconn->recv_q, entry); |
|
uint64_t peer_id = rconn->peer_reset_id; |
|
int ret = router_deliver(rconn, conn ? conn : loopback_conn(rconn->inst), entry->dgram, entry->len); |
|
if (ret != 0 || rconn->closed || rconn->peer_reset_id != peer_id) { free_entry(entry); return; } |
|
rconn->rx_seq = next + 1; |
|
next++; |
|
queue_dgram_free(entry); |
|
queue_entry_free(entry); |
|
} |
|
} |
|
|
|
// ==================================================================== |
|
// Очередь приёма входящего трафика (между сетью и recv_q) |
|
// ==================================================================== |
|
|
|
// Коллбэк incoming_q: переместить пакет в recv_q (с дедупликацией по seq), при совпадении с |
|
// rx_seq — собрать доставку, затем запланировать ACK. |
|
static void router_incoming_q_cb(struct ll_queue* q, void* arg) { |
|
struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; |
|
if (rconn->closed) { queue_resume_callback(q); return; } |
|
struct ll_entry* e = queue_data_get(q); |
|
if (!e) { queue_resume_callback(q); return; } |
|
|
|
uint32_t seq = *(uint32_t*)e->data; |
|
const uint8_t* wire = e->dgram; |
|
size_t wire_len = e->len; |
|
|
|
if ((int32_t)(seq - rconn->rx_seq) < 0 || queue_find_data_by_index(rconn->recv_q, &seq)) { |
|
DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "router_incoming_q: dup in recv_q seq=%u, dropping", seq); |
|
queue_dgram_free(e); queue_entry_free(e); queue_resume_callback(q); return; |
|
} |
|
|
|
struct ll_entry* rq = queue_entry_new(4); |
|
if (!rq) { queue_dgram_free(e); queue_entry_free(e); queue_resume_callback(q); return; } |
|
*(uint32_t*)rq->data = seq; |
|
rq->dgram = e->dgram; rq->len = e->len; e->dgram = NULL; |
|
queue_entry_free(e); |
|
|
|
queue_data_put_with_index(rconn->recv_q, rq); |
|
|
|
if (seq == rconn->rx_seq) router_try_assembly(rconn, rconn->recv_conn); |
|
if (!rconn->closed && rconn->incoming_q == q) { router_schedule_ack(rconn); queue_resume_callback(q); } |
|
} |
|
|
|
// ==================================================================== |
|
// Функции приёма etcp_router_recv_cb |
|
// ==================================================================== |
|
|
|
// Обработать ACK: обновить RTT/jitter, снять подтверждённые inflight-пакеты, при необходимости |
|
// возобновить отправку (снять send_blocked). |
|
static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN* rconn, |
|
uint32_t seq, uint64_t src_node_id, uint16_t ack_ts) { |
|
(void)src_node_id; |
|
if (!rconn) return; |
|
if ((int32_t)(seq - rconn->tx_sent) > 0) { |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router ACK outside sent range: peer=%016llx svc=%u ack=%u sent=%u acked=%u", |
|
(unsigned long long)src_node_id, rconn->svc_id, seq, rconn->tx_sent, rconn->tx_acked); |
|
rconn->c_stale_ack++; |
|
return; |
|
} |
|
if ((int32_t)(seq - rconn->tx_acked) <= 0) { rconn->c_stale_ack++; return; } |
|
if (ack_ts != 0) { |
|
uint16_t new_rtt = get_current_timestamp() - ack_ts; |
|
int d_rtt = (int)new_rtt - (int)rconn->rtt; |
|
if (d_rtt < 0) d_rtt = -d_rtt; |
|
rconn->rtt_jitter += (int32_t)(((int64_t)d_rtt * 65536 - rconn->rtt_jitter) / 32); |
|
rconn->rtt = new_rtt; |
|
} |
|
rconn->tx_acked = seq; |
|
rconn->start_sent = 1; |
|
while (rconn->inflight_q->head) { |
|
struct ROUTER_INFLIGHT* inf = (struct ROUTER_INFLIGHT*)rconn->inflight_q->head; |
|
if ((int32_t)(inf->seq - seq) >= 0) break; |
|
queue_remove_data(rconn->inflight_q, &inf->ll); |
|
u_free(inf->dgram); u_free(inf); |
|
} |
|
rconn->last_ack_changed_tb = get_time_tb(); |
|
rconn->no_ack_count = 0; |
|
router_track_inflight_state(rconn); |
|
router_update_minrtt(rconn); |
|
if (queue_entry_count(rconn->inflight_q) == 0 && rconn->retrans_timer) { |
|
uasync_cancel_timeout(rconn->inst->ua, rconn->retrans_timer); |
|
rconn->retrans_timer = NULL; |
|
} |
|
DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "SEND_Q_ACKED: svc_id=%u ack=%u inflight=%d send_q=%d inflight_q=%d from %016llx", |
|
rconn->svc_id, seq, queue_entry_count(rconn->inflight_q), |
|
queue_entry_count(rconn->send_q), queue_entry_count(rconn->inflight_q), (unsigned long long)src_node_id); |
|
rconn->c_ack_recv++; |
|
rconn->last_dgram_ts = get_current_timestamp(); |
|
if (rconn->send_blocked) { rconn->send_blocked = 0; router_send_kick(rconn); } |
|
} |
|
|
|
// Ретранслировать транзитный пакет к dst_node_id: отправить напрямую, при занятости send_input_q — |
|
// поставить в транзитную очередь с backpressure. |
|
static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* entry, |
|
struct SVC_ROUTE_HDR* hdr) { |
|
struct TOPO_GROUP* group = topo_groups_find(inst->topo_groups, hdr->group_id); |
|
struct ETCP_CONN* next = topo_group_find_conn_for_node(group, hdr->dst_node_id); |
|
if (!next) { |
|
int node_count = group && group->nodes ? queue_entry_count(group->nodes) : -1; |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: no route to %016llx svc_id=%u, dropping (group=%016llx nodes=%d)", |
|
(unsigned long long)hdr->dst_node_id, hdr->svc_id, |
|
(unsigned long long)hdr->group_id, node_count); |
|
if (group && group->nodes && node_count > 0 && node_count < 10) { |
|
char list[512]; int pos = 0; |
|
struct ll_entry* e = group->nodes->head; |
|
while (e) { |
|
struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)e; |
|
pos += snprintf(list + pos, sizeof(list) - pos, "%s%016llx(up=%02x)", |
|
pos ? "," : "", (unsigned long long)nq->node_id, nq->conn_up); |
|
e = e->next; |
|
} |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: group nodes: %s", list); |
|
} |
|
free_entry(entry); return; |
|
} |
|
struct TRANSIT_QUEUE* tq = transit_queue_find(next, hdr->group_id, hdr->src_node_id, hdr->dst_node_id); |
|
|
|
if (!tq && next->send_input_q && queue_entry_count(next->send_input_q) == 0) { |
|
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 ret = etcp_send(next, entry); |
|
if (ret != 0) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: forward etcp_send failed ret=%d svc_id=%u → %016llx", |
|
ret, hdr->svc_id, (unsigned long long)hdr->dst_node_id); |
|
free_entry(entry); |
|
} |
|
return; |
|
} |
|
|
|
if (!tq) { |
|
tq = transit_queue_get_or_create(next, hdr->group_id, hdr->src_node_id, hdr->dst_node_id); |
|
if (!tq) { free_entry(entry); return; } |
|
} |
|
int was_empty = (queue_entry_count(tq->q) == 0); |
|
queue_data_put(tq->q, entry); |
|
DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: queued transit svc_id=%u src=%016llx dst=%016llx q=%d", |
|
hdr->svc_id, (unsigned long long)hdr->src_node_id, |
|
(unsigned long long)hdr->dst_node_id, queue_entry_count(tq->q)); |
|
if (was_empty && next->send_input_q) |
|
queue_waiter_wait(next->send_input_q, &tq->waiter, transit_queue_drain_cb, tq); |
|
} |
|
|
|
// Принять data-пакет: зафиксировать recv_conn, синхронизировать rx_seq при первом пакете сеанса, |
|
// отсеять out-of-bounds/дубликаты и положить пакет в incoming_q. |
|
static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, |
|
uint32_t seq, const uint8_t* wire, size_t wire_len) { |
|
rconn->recv_conn = conn; |
|
rconn->last_dgram_ts = get_current_timestamp(); |
|
|
|
{// SEQ за пределами окна |
|
int32_t d = (int32_t)(seq - rconn->rx_seq); |
|
if (d > (int32_t)rconn->max_inflight || d < -(int32_t)rconn->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)) {// дубликат или seq <= rx_seq |
|
// Сценарий потерянного ACK: пир ретранслит, значит не получил наш ACK — дослать его. |
|
// router_schedule_ack здесь не годится: после первого ACK rx_seq==last_sent_ack_seq и он выходит рано. |
|
uint64_t now = get_time_tb(); |
|
if (now - rconn->last_ack_sent_tb >= ROUTER_ACK_INTERVAL_TB) { |
|
DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "router: dup seq=%u rx_seq=%u, resending ACK", seq, rconn->rx_seq); |
|
router_send_ack(rconn); |
|
} |
|
DEBUG_TRACE(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(wire_len); |
|
if (!qe->dgram) { queue_dgram_free(qe); queue_entry_free(qe); return; } |
|
qe->len = wire_len; memcpy(qe->dgram, wire, wire_len); |
|
|
|
DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "router: incoming_q seq=%u rx_seq=%u from %016llx svc_id=%u", |
|
seq, rconn->rx_seq, |
|
(unsigned long long)rconn->remote_node_id, rconn->svc_id); |
|
queue_data_put(rconn->incoming_q, qe); |
|
} |
|
|
|
// ==================================================================== |
|
// etcp_router_recv_cb — обработчик ETCP_RT_ID_SVC_ROUTE |
|
// ==================================================================== |
|
|
|
// Входная точка приёма SVC_ROUTE-кодограмм: различает транзит, RST/CLOSE/ACK (пустой payload) |
|
// и data-пакеты, направляя их в router_handle_data_packet. |
|
static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) { |
|
if (!entry) return; |
|
struct UTUN_INSTANCE* inst = conn ? conn->instance : NULL; |
|
if (!inst || !entry->dgram || entry->len < SVC_ROUTE_HDR_SIZE) { |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router recv: invalid packet len=%u", entry->len); |
|
free_entry(entry); return; |
|
} |
|
struct SVC_ROUTE_HDR* h = (struct SVC_ROUTE_HDR*)entry->dgram; |
|
if (h->dst_node_id != inst->node_id) { router_forward_transit(inst, entry, h); return; } |
|
struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, h->group_id, h->src_node_id, h->svc_id); |
|
if (h->flags & (ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED)) { |
|
uint8_t* decoded = NULL; size_t len = 0; uint8_t flags = 0; |
|
if (route_crypto_decode(inst, entry->dgram, entry->len, &decoded, &len, &flags) != 0) { |
|
if (rconn) rconn->c_sign_fail++; |
|
free_entry(entry); return; |
|
} |
|
queue_dgram_free(entry); entry->dgram = decoded; entry->len = len; |
|
h = (struct SVC_ROUTE_HDR*)decoded; h->flags |= flags; |
|
} |
|
uint8_t control = h->flags & ~(ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED); |
|
size_t payload_len = entry->len - SVC_ROUTE_HDR_SIZE; |
|
if (!h->reset_id || (payload_len && (control || h->challenge)) |
|
|| (!payload_len && control != 0 && control != ROUTER_FLAG_START && control != ROUTER_FLAG_RST |
|
&& control != (ROUTER_FLAG_START | ROUTER_FLAG_RST) && control != ROUTER_FLAG_CLOSE)) { |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router recv: malformed flags=%02x seq=%u len=%zu epoch=%016llx", |
|
h->flags, h->seq, payload_len, (unsigned long long)h->reset_id); |
|
free_entry(entry); return; |
|
} |
|
if (!rconn) { |
|
if (!payload_len && control != ROUTER_FLAG_START) { free_entry(entry); return; } |
|
rconn = etcp_router_conn_get(inst, h->group_id, h->src_node_id, h->svc_id); |
|
if (!rconn) { free_entry(entry); return; } |
|
} |
|
if (control == ROUTER_FLAG_START) { |
|
router_challenge_peer(rconn, conn, h->reset_id); |
|
} else if (control == ROUTER_FLAG_RST) { |
|
if (h->peer_reset_id == rconn->reset_id && h->challenge) { |
|
router_control(rconn, conn, ROUTER_FLAG_START | ROUTER_FLAG_RST, h->reset_id, h->challenge); |
|
if (h->reset_id != rconn->peer_reset_id) router_challenge_peer(rconn, conn, h->reset_id); |
|
} |
|
} else if (control == (ROUTER_FLAG_START | ROUTER_FLAG_RST)) { |
|
router_confirm_peer(rconn, h); |
|
} else if (h->peer_reset_id != rconn->reset_id || h->reset_id != rconn->peer_reset_id) { |
|
DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router epoch mismatch: peer=%016llx svc=%u seq=%u flags=%02x", |
|
(unsigned long long)h->src_node_id, h->svc_id, h->seq, h->flags); |
|
if (payload_len) router_challenge_peer(rconn, conn, h->reset_id); |
|
} else if (control == ROUTER_FLAG_CLOSE) { |
|
router_close_and_notify(rconn); |
|
} else if (!payload_len) { |
|
if (!h->challenge) router_handle_ack(inst, rconn, h->seq, h->src_node_id, h->timestamp); |
|
} else { |
|
rconn->last_recv_pkt_ts = h->timestamp; |
|
rconn->last_recv_pkt_local_tb = get_time_tb(); rconn->last_recv_updated = 1; |
|
router_handle_data_packet(rconn, conn, h->seq, entry->dgram, entry->len); |
|
} |
|
free_entry(entry); |
|
} |
|
|
|
// ==================================================================== |
|
// Публичное API |
|
// ==================================================================== |
|
|
|
// Инициализация роутера: создать реестр router_conns, очистить bindings и зарегистрировать |
|
// обработчик ETCP_ID_SVC_ROUTE. |
|
int etcp_router_init(struct UTUN_INSTANCE* inst) { |
|
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "NULL instance"); return -1; } |
|
memset(&inst->router_bindings, 0, sizeof(inst->router_bindings)); |
|
inst->router_conns = queue_new(inst->ua, |
|
ROUTER_CONN_HASH_SIZE, 0, 17, "router_conns"); |
|
if (!inst->router_conns) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "queue_new(router_conns) failed"); |
|
return -1; |
|
} |
|
int ret = etcp_bind(inst, ETCP_ID_SVC_ROUTE, etcp_router_recv_cb); |
|
if (ret != 0) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "etcp_bind failed, ret=%d", ret); |
|
queue_free(inst->router_conns); inst->router_conns = NULL; |
|
return -1; |
|
} |
|
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "etcp_router initialized for node %016llx", (unsigned long long)inst->node_id); |
|
return 0; |
|
} |
|
|
|
// Деинициализация: снять обработчик SVC_ROUTE, закрыть все conn и освободить реестр. |
|
void etcp_router_destroy(struct UTUN_INSTANCE* inst) { |
|
if (!inst) return; |
|
etcp_unbind(inst, ETCP_ID_SVC_ROUTE); |
|
if (inst->router_conns) { |
|
struct ll_entry* entry; |
|
while ((entry = inst->router_conns->head) != NULL) { |
|
struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)entry; |
|
router_close_and_notify(rconn); |
|
} |
|
queue_free(inst->router_conns); |
|
inst->router_conns = NULL; |
|
} |
|
memset(&inst->router_bindings, 0, sizeof(inst->router_bindings)); |
|
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "etcp_router destroyed"); |
|
} |
|
|
|
// Зарегистрировать обработчик сервиса. callback=NULL — снять обработчик (эквивалент unbind без ошибки). |
|
int etcp_router_bind(struct UTUN_INSTANCE* inst, uint8_t svc_id, etcp_recv_fn callback) { |
|
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "NULL instance"); return -1; } |
|
if (!callback) { |
|
DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "unbind svc_id=%u", svc_id); |
|
inst->router_bindings.callbacks[svc_id] = NULL; |
|
return 0; |
|
} |
|
if (inst->router_bindings.callbacks[svc_id]) |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "overwriting svc_id=%u", svc_id); |
|
inst->router_bindings.callbacks[svc_id] = callback; |
|
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "svc_id=%u → cb=%p", svc_id, (void*)callback); |
|
return 0; |
|
} |
|
|
|
// Снять обработчик сервиса. Возвращает -1 если обработчик не был зарегистрирован. |
|
int etcp_router_unbind(struct UTUN_INSTANCE* inst, uint8_t svc_id) { |
|
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "NULL instance"); return -1; } |
|
if (!inst->router_bindings.callbacks[svc_id]) { |
|
DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "svc_id=%u not bound", svc_id); |
|
return -1; |
|
} |
|
inst->router_bindings.callbacks[svc_id] = NULL; |
|
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "svc_id=%u", svc_id); |
|
return 0; |
|
} |
|
|
|
// Отправить сервисный пакет. entry->dgram[0] = svc_id, остальное — payload. Берёт на себя |
|
// освобождение entry в любом исходе. dst==self — loopback-доставка. |
|
int etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, struct ll_entry* entry, int force, int mode) { |
|
if (!inst || !entry) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "NULL inst=%p entry=%p", (void*)inst, (void*)entry); |
|
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } return -1; |
|
} |
|
if (!entry->dgram || entry->len < 1) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "empty entry: dgram=%p len=%u dst=0x%016llx force=%d", |
|
(void*)entry->dgram, entry->len, (unsigned long long)dst_node_id, force); queue_dgram_free(entry); queue_entry_free(entry); return -1; |
|
} |
|
uint8_t svc_id = entry->dgram[0]; |
|
size_t payload_len = entry->len - 1; |
|
|
|
if (dst_node_id == inst->node_id) { |
|
DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "loopback svc_id=%u len=%zu", svc_id, payload_len); |
|
router_deliver_loopback(inst, group_id, svc_id, entry); |
|
return 0; |
|
} |
|
|
|
struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, group_id, dst_node_id, svc_id); |
|
if (!rconn) { queue_dgram_free(entry); queue_entry_free(entry); return -1; } |
|
|
|
int ret = router_enqueue_send(rconn, entry->dgram + 1, payload_len, force, mode); |
|
queue_dgram_free(entry); queue_entry_free(entry); |
|
return ret; |
|
} |
|
|
|
// Рестарт локального состояния для peer+svc в группе (перезапуск удалённой стороны): уведомить |
|
// сервис пустой кодограммой и сбросить conn. |
|
void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t remote_node_id, uint8_t svc_id) { |
|
struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, group_id, remote_node_id, svc_id); |
|
if (!rconn) return; |
|
|
|
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_restart: svc_id=%u remote=%016llx reset_id=%016llx — resetting local state", |
|
svc_id, (unsigned long long)remote_node_id, (unsigned long long)rconn->reset_id); |
|
|
|
if (rconn->closed) return; |
|
uint64_t new_id; |
|
if (router_random_id(&new_id) != 0) { router_close_and_notify(rconn); return; } |
|
if (router_conn_reset(rconn) != 0) { router_close_and_notify(rconn); return; } |
|
rconn->reset_id = new_id; |
|
rconn->pending_peer_id = 0; rconn->pending_challenge = 0; |
|
router_send_close_to_service(rconn); |
|
if (!rconn->closed) router_handshake_start(rconn); |
|
} |
|
|
|
// Backpressure: зарегистрировать waiter на send_q для (group_id, node_id, svc_id). |
|
void etcp_router_on_send_ready(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id, uint8_t svc_id, |
|
struct queue_waiter_handle* h, |
|
queue_threshold_callback_fn callback, void* arg) { |
|
if (!inst || !h) return; |
|
etcp_router_cancel_send_ready(inst, group_id, node_id, svc_id, h); |
|
struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, group_id, node_id, svc_id); |
|
if (!rconn || !rconn->send_q) return; |
|
h->defer_q = rconn->send_q; /* Keep the registration queue across route/session replacement. */ |
|
queue_waiter_wait(rconn->send_q, h, callback, arg); |
|
} |
|
|
|
// Backpressure: отменить waiter на send_q для (group_id, node_id, svc_id). |
|
void etcp_router_cancel_send_ready(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id, uint8_t svc_id, |
|
struct queue_waiter_handle* h) { |
|
(void)inst; (void)group_id; (void)node_id; (void)svc_id; |
|
if (h && (h->internal || h->call_soon_id)) queue_waiter_cancel(h->defer_q, h); |
|
} |
|
|
|
// Backpressure: есть ли место в send_q (без регистрации waiter). 1 — есть, 0 — нет. |
|
int etcp_router_send_q_has_room(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id, uint8_t svc_id) { |
|
if (!inst) return 0; |
|
struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, group_id, node_id, svc_id); |
|
if (!rconn || !rconn->send_q) return 0; |
|
return rconn->send_q->count <= rconn->send_q->threshold_max_packets |
|
&& (rconn->send_q->threshold_max_bytes == 0 || rconn->send_q->total_bytes <= rconn->send_q->threshold_max_bytes); |
|
} |
|
|
|
// ==================================================================== |
|
// API seq-подключений |
|
// ==================================================================== |
|
|
|
// Найти/создать состояние seq-подключения по (group_id, remote_node_id, svc_id): выделяет conn, |
|
// инициализирует случайную эпоху reset_id и очереди. |
|
struct ETCP_ROUTER_CONN* etcp_router_conn_get(struct UTUN_INSTANCE* inst, |
|
uint64_t group_id, |
|
uint64_t remote_node_id, uint8_t svc_id) { |
|
if (!inst || !inst->router_conns) return NULL; |
|
struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, group_id, remote_node_id, svc_id); |
|
if (rconn) return rconn; |
|
|
|
size_t data_size = sizeof(struct ETCP_ROUTER_CONN) - sizeof(struct ll_entry); |
|
rconn = (struct ETCP_ROUTER_CONN*)queue_entry_new(data_size); |
|
if (!rconn) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: queue_entry_new failed"); return NULL; } |
|
rconn->group_id = group_id; |
|
rconn->remote_node_id = remote_node_id; |
|
rconn->svc_id = svc_id; |
|
rconn->inst = inst; |
|
rconn->last_dgram_ts = get_current_timestamp(); |
|
if (router_random_id(&rconn->reset_id) != 0) { queue_entry_free(&rconn->ll); return NULL; } |
|
if (router_conn_reset(rconn) != 0) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: queue_new failed"); |
|
router_conn_free_queues(rconn); |
|
queue_entry_free(&rconn->ll); |
|
return NULL; |
|
} |
|
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_conn: new conn group=%016llx remote=%016llx svc_id=%u", |
|
(unsigned long long)group_id, (unsigned long long)remote_node_id, svc_id); |
|
queue_data_put_with_index(inst->router_conns, &rconn->ll); |
|
return rconn; |
|
} |
|
|
|
// Отправить данные (payload без svc_id) с авто-seq и контролем inflight. mode — биты |
|
// ROUTE_CRYPTO_SIGN / ROUTE_CRYPTO_ENCRYPT (0 = обычный пакет). |
|
int etcp_router_conn_send(struct ETCP_ROUTER_CONN* rconn, |
|
const uint8_t* data, size_t len, int mode) { |
|
if (!rconn || !data || len == 0) { |
|
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_send: invalid args rconn=%p data=%p len=%zu", |
|
(void*)rconn, (const void*)data, len); |
|
return -1; |
|
} |
|
return router_enqueue_send(rconn, data, len, 0, mode); |
|
} |
|
|
|
// Закрыть seq-подключение (нормальное закрытие с уведомлением сервиса и пира). |
|
void etcp_router_conn_close(struct ETCP_ROUTER_CONN* rconn) { |
|
router_close_and_notify(rconn); |
|
} |
|
|
|
// Сбросить ретрансмиты для всех router_conn к узлу (без коллбэков и без CLOSE — облегчённо, |
|
// для etcp_conn_reinit, чтобы избежать гонки с очисткой ETCP-очередей). |
|
void etcp_router_pause_retrans_for_node(struct UTUN_INSTANCE* inst, uint64_t remote_node_id) { |
|
if (!inst || !inst->router_conns) return; |
|
for (struct ll_entry* e = inst->router_conns->head; e; e = e->next) { |
|
struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)e; |
|
if (rconn->remote_node_id != remote_node_id || rconn->closed) continue; |
|
router_cancel_send_waiter(rconn); |
|
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router transport reinit: peer=%016llx svc=%u retained=%d queued=%d acked=%u sent=%u", |
|
(unsigned long long)remote_node_id, rconn->svc_id, queue_entry_count(rconn->inflight_q), |
|
queue_entry_count(rconn->send_q), rconn->tx_acked, rconn->tx_sent); |
|
if (queue_entry_count(rconn->inflight_q) && !rconn->retrans_timer) router_retrans_schedule(rconn); |
|
router_send_watchdog_arm(rconn); |
|
} |
|
} |
|
|
|
// Закрыть все router_conn для (group_id, remote_node_id) (peer умер). |
|
void etcp_router_conn_close_all_for_node(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t remote_node_id) { |
|
if (!inst || !inst->router_conns) return; |
|
// Итерируем по hash-цепочкам, чтобы избежать бесконечного цикла из-за повторного помещения в FIFO-итерации |
|
struct ll_entry* to_close[256]; |
|
int count = 0; |
|
struct ll_queue* q = inst->router_conns; |
|
for (uint32_t slot = 0; slot < q->hash_size && count < 256; slot++) { |
|
struct ll_entry* entry = q->hash_table[slot]; |
|
while (entry) { |
|
struct ll_entry* next = entry->hash_next; |
|
struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)entry; |
|
if (rconn->group_id == group_id && rconn->remote_node_id == remote_node_id) |
|
to_close[count++] = entry; |
|
entry = next; |
|
} |
|
} |
|
for (int i = 0; i < count; i++) |
|
router_close_and_notify((struct ETCP_ROUTER_CONN*)to_close[i]); |
|
} |
|
|
|
/* Service stop: discard every logical channel in a group before its routes disappear. |
|
* Call after service producers have cancelled their waiters. */ |
|
void etcp_router_close_group(struct UTUN_INSTANCE* inst, uint64_t group_id) { |
|
if (!inst) return; |
|
/* Notifications may close other channels too; do not retain list cursors across them. */ |
|
while (inst->router_conns) { |
|
struct ll_entry* entry = inst->router_conns->head; |
|
while (entry && ((struct ETCP_ROUTER_CONN*)entry)->group_id != group_id) entry = entry->next; |
|
if (!entry) break; |
|
router_close_and_notify((struct ETCP_ROUTER_CONN*)entry); |
|
} |
|
for (struct ll_entry* e = inst->connections ? inst->connections->head : NULL; e; e = e->next) { |
|
struct ETCP_CONN* conn = ((struct conn_queue_entry*)e->data)->conn; |
|
if (!conn) continue; |
|
struct ll_entry* entry = conn->transit_queues ? conn->transit_queues->head : NULL; |
|
while (entry) { |
|
struct TRANSIT_QUEUE* transit = (struct TRANSIT_QUEUE*)entry; entry = entry->next; |
|
if (transit->group_id == group_id) transit_queue_destroy(conn, transit); |
|
} |
|
} |
|
}
|
|
|