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

Cancel pending transit queues when destroying the router

Browse Source
This commit is contained in:
evgeny committed 2026-10-24 15:06:48 +02:00
1 parent d8089f8d12
commit a258b77240
2 files changed
+51 -1
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/routing_layer/etcp_router.c
+9 -1
View File
@@ -239,18 +239,22 @@ static void transit_queue_drain_cb(struct ll_queue* q, void* arg) {
// Удалить все транзитные очереди соединения (вызывается при закрытии ETCP_CONN).
void etcp_router_transit_queues_destroy(struct ETCP_CONN* conn) {
if (!conn || !conn->transit_queues) return;
int pairs = 0, dropped = 0;
struct ll_entry* entry;
while ((entry = queue_data_get(conn->transit_queues)) != NULL) {
struct TRANSIT_QUEUE* tq = (struct TRANSIT_QUEUE*)entry;
if (tq->route_timer) uasync_cancel_timeout(conn->instance->ua, tq->route_timer);
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);
while ((e = queue_data_get(tq->q)) != NULL) { free_entry(e); dropped++; }
queue_free(tq->q);
queue_entry_free(&tq->ll);
pairs++;
}
queue_free(conn->transit_queues);
conn->transit_queues = NULL;
if (pairs) DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "transit queues released: peer=%016llx pairs=%d discarded=%d",
(unsigned long long)conn->peer_node_id, pairs, dropped);
}
// Сбросить recv_conn у всех ROUTER_CONN, ссылающихся на уничтожаемый ETCP_CONN.
@@ -1438,6 +1442,10 @@ void etcp_router_destroy(struct UTUN_INSTANCE* inst) {
queue_free(inst->router_conns);
inst->router_conns = NULL;
}
for (struct ll_entry* entry = inst->connections ? inst->connections->head : NULL; entry; entry = entry->next) {
struct ETCP_CONN* conn = ((struct conn_queue_entry*)entry->data)->conn;
if (conn) etcp_router_transit_queues_destroy(conn);
}
memset(&inst->router_bindings, 0, sizeof(inst->router_bindings));
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "etcp_router destroyed");
}
tests/test_etcp_router_unit.c
+42
View File
@@ -1580,6 +1580,47 @@ static int test_loopback_length_and_errors(void) {
PASS(); return 0;
}
/* Destroy должен отменить и транзит, даже если физический ETCP остаётся живым. */
static int test_destroy_pending_transit(void) {
TEST("router destroy cancels transit waiters and route timers on live transports");
struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb;
SETUP();
struct TOPO_GROUPS groups = {.instance = &inst}; inst.topo_groups = &groups;
groups.group_list = queue_new(ua, 16, 0, 8, "transit_groups"); TESTASSERT(groups.group_list);
struct TOPO_GROUP* group = (void*)queue_entry_new(sizeof(*group) - sizeof(struct ll_entry)); TESTASSERT(group);
group->instance = &inst; group->group_id = TOPO_GROUP_UTUN;
group->nodes = queue_new(ua, 16, offsetof(struct TOPO_GROUP_NODE, node_id) - sizeof(struct ll_entry), 8, "transit_nodes");
TESTASSERT(group->nodes); queue_data_put_with_index(groups.group_list, &group->ll);
struct ETCP_CONN next = {.instance = &inst, .peer_node_id = TEST_REMOTE_NODE2, .links_up = 1};
next.send_input_q = queue_new(ua, 0, 0, 0, "transit_transport"); TESTASSERT(next.send_input_q);
queue_set_waiter_defer(next.send_input_q, 1);
struct ll_entry* blocker = queue_entry_new(0); TESTASSERT(blocker); queue_data_put(next.send_input_q, blocker);
inst.connections = queue_new(ua, 16, 0, 8, "transit_connections"); TESTASSERT(inst.connections);
struct ll_entry* ce = queue_entry_new(sizeof(struct conn_queue_entry)); TESTASSERT(ce);
((struct conn_queue_entry*)ce->data)->peer_node_id = next.peer_node_id;
((struct conn_queue_entry*)ce->data)->conn = &next; queue_data_put_with_index(inst.connections, ce);
struct TOPO_GROUP_NODE* node = (void*)queue_entry_new(sizeof(*node) - sizeof(struct ll_entry)); TESTASSERT(node);
node->node_id = next.peer_node_id; node->paths = queue_new(ua, 0, 0, 0, "transit_paths"); TESTASSERT(node->paths);
struct TOPO_NODEPATH* path = (void*)queue_entry_new(sizeof(*path) - sizeof(struct ll_entry) + sizeof(uint64_t)); TESTASSERT(path);
path->conn = &next; path->hop_count = 1;
memcpy((uint8_t*)path + sizeof(*path), &next.peer_node_id, sizeof(uint64_t));
queue_data_put(node->paths, &path->ll); queue_data_put_with_index(group->nodes, &node->ll);
struct SVC_ROUTE_HDR packet = {.cmd = ETCP_ID_SVC_ROUTE, .group_id = TOPO_GROUP_UTUN,
.src_node_id = TEST_REMOTE_NODE, .dst_node_id = next.peer_node_id, .svc_id = TEST_SVC_ID,
.reset_id = 1, .flags = ROUTER_FLAG_START};
struct ll_entry* e = ll_alloc_lldgram(sizeof(packet)); TESTASSERT(e);
memcpy(e->dgram, &packet, sizeof(packet)); e->len = sizeof(packet); recv_cb(&fake_conn, e);
TESTASSERT(next.transit_queues && next.transit_queues->count == 1 && next.send_input_q->waiter_head);
etcp_router_destroy(&inst);
int cancelled = !next.transit_queues && !next.send_input_q->waiter_head;
etcp_router_transit_queues_destroy(&next);
queue_entry_free(queue_data_get(next.send_input_q)); queue_free(next.send_input_q);
queue_entry_free(queue_data_get(inst.connections)); queue_free(inst.connections); inst.connections = NULL;
topo_nodeq_clear_paths(node); queue_entry_free(queue_data_get(group->nodes)); queue_free(group->nodes);
queue_entry_free(queue_data_get(groups.group_list)); queue_free(groups.group_list); inst.topo_groups = NULL;
TEARDOWN(); TESTASSERT(cancelled); PASS(); return 0;
}
int main(void) {
size_t allocated_before = u_get_allocated_count();
// Run all tests
@@ -1587,6 +1628,7 @@ int main(void) {
test_reopen_from_close_callback();
test_nested_loopback();
test_loopback_length_and_errors();
test_destroy_pending_transit();
test_init_destroy();
test_bind_unbind();
test_conn_get_create();