diff --git a/src/routing_layer/etcp_router.c b/src/routing_layer/etcp_router.c index 267049de..faeca5d4 100644 --- a/src/routing_layer/etcp_router.c +++ b/src/routing_layer/etcp_router.c @@ -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"); } diff --git a/tests/test_etcp_router_unit.c b/tests/test_etcp_router_unit.c index 58d5dbdf..e00420e9 100644 --- a/tests/test_etcp_router_unit.c +++ b/tests/test_etcp_router_unit.c @@ -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();