From 80451c43be576571daa52ee3f019eea32360275c Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 24 Oct 2026 16:09:39 +0200 Subject: [PATCH] Make CM cleanup allocation-free and reject incomplete startup --- src/routing_layer/conn_mgr_core.c | 74 +++++++++++++++++++++------- src/routing_layer/conn_mgr_monitor.c | 4 ++ src/routing_layer/conn_mgr_priv.h | 8 ++- 3 files changed, 66 insertions(+), 20 deletions(-) diff --git a/src/routing_layer/conn_mgr_core.c b/src/routing_layer/conn_mgr_core.c index 8034cab3..4909a8fd 100644 --- a/src/routing_layer/conn_mgr_core.c +++ b/src/routing_layer/conn_mgr_core.c @@ -1,15 +1,10 @@ -/* conn_mgr_core.c — init/destroy, open/close/handles, DIRECT, REVERSE, invite */ +/* conn_mgr_core.c — владение handles, жизненный цикл и DIRECT/REVERSE-пробы. */ #include "conn_mgr_priv.h" #include #include -#ifdef _WIN32 -#include -#include -#else -#include -#endif #include "../lib/platform_compat.h" +#include "../lib/socket_compat.h" #include "../lib/debug_config.h" #include "../lib/mem.h" #include "../lib/u_async.h" @@ -35,6 +30,7 @@ void conn_mgr_router_recv_handler(struct ETCP_CONN* conn, struct ll_entry* entry); static void cm_attempt_tick(void* arg); static void cm_start_attempt(struct CONN_MGR_ENTRY* entry, int recovery); +static void cm_deliver_up_cb(void* arg); static const char* cm_path_name(uint8_t type) { static const char* names[] = { "NONE", "DIRECT", "REVERSE", "INDIRECT" }; @@ -49,7 +45,16 @@ static struct CONN_MGR_HANDLE* cm_handle_new(struct CONN_MGR_ENTRY* entry, uint6 if (!h) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: handle allocation failed"); return NULL; } h->node_id = entry->node_id; h->group_id = group_id; h->cb = cb; h->cb_arg = cb_arg; h->ua = entry->mgr->instance->ua; + h->mgr = entry->mgr; + if (entry->state == CONN_MGR_STATE_CONNECTED) { + h->deliver_up_token = uasync_call_soon(h->ua, h, cm_deliver_up_cb); + if (!h->deliver_up_token) { + DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: initial UP scheduling failed peer=%016llx", (unsigned long long)entry->node_id); + u_free(h); return NULL; + } + } h->entry = entry; h->next = entry->handles; entry->handles = h; + h->owner_next = h->mgr->handles; h->mgr->handles = h; return h; } @@ -60,7 +65,7 @@ static void cm_deliver_up_cb(void* arg) { h->cb(h, h->node_id, h->group_id, CONN_EVENT_UP, h->cb_arg); } -/* Закрытые handles освобождаются через call_soon, поэтому next переживает callbacks. +/* Закрытые handles освобождаются фоновым GC, поэтому next переживает callbacks. * Рассылает событие ВСЕМ handle'ам entry. Одно соединение — много слушателей: * каждый conn_mgr_open создал свой handle, все получают UP/DOWN/TIMEOUT. * Все поля handle'а читаются до вызова коллбэка — коллбэк может закрыть @@ -149,8 +154,20 @@ struct CONN_MGR_ENTRY* cm_ensure_entry(struct CONN_MGR* mgr, uint64_t node_id) { return new_e; } -static void cm_entry_free_cb(void* arg) { queue_entry_free(arg); } -static void cm_handle_free_cb(void* arg) { u_free(arg); } +/* Объекты сняты с рабочих списков, но активный CM-стек ещё может читать state/next. + * В начале фонового callback и при destroy этих стеков уже нет. Аллокации не нужны. */ +void cm_collect(struct CONN_MGR* mgr) { + size_t handles = 0, entries = 0; + while (mgr->retired_handles) { + struct CONN_MGR_HANDLE* h = mgr->retired_handles; + mgr->retired_handles = h->owner_next; u_free(h); handles++; + } + while (mgr->retired_entries) { + struct CONN_MGR_ENTRY* entry = mgr->retired_entries; + mgr->retired_entries = entry->retired_next; queue_entry_free(&entry->ll); entries++; + } + if (handles || entries) DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "conn_mgr: collected closed handles=%zu entries=%zu", handles, entries); +} void cm_entry_cleanup(struct CONN_MGR_ENTRY* entry) { if (!entry) return; @@ -171,7 +188,7 @@ void cm_entry_cleanup(struct CONN_MGR_ENTRY* entry) { cm_cleanup_db_node(entry); DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "conn_mgr: cleanup group=%016llx peer=%016llx", (unsigned long long)mgr->group->group_id, (unsigned long long)entry->node_id); - uasync_call_soon(mgr->instance->ua, entry, cm_entry_free_cb); + entry->retired_next = mgr->retired_entries; mgr->retired_entries = entry; } void cm_clear_nodeinfo(struct CONN_MGR* mgr, uint64_t node_id) { @@ -496,6 +513,10 @@ struct CONN_MGR* conn_mgr_init(struct TOPO_GROUP* group) { if (!mgr->entries) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "entries alloc failed"); u_free(mgr); return NULL; } mgr->bg_ping_timer = uasync_set_timeout(mgr->instance->ua, CONN_MGR_BG_PING_INTERVAL_TB, mgr, cm_bg_ping_timer_cb, "conn_mgr_bg_ping"); mgr->candidate_ping_timer = uasync_set_timeout(mgr->instance->ua, CONN_MGR_CANDIDATE_PING_TB, mgr, cm_candidate_ping_timer_cb, "conn_mgr_cand_ping"); + if (!mgr->bg_ping_timer || !mgr->candidate_ping_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: monitor timers unavailable group=%016llx", (unsigned long long)group->group_id); + conn_mgr_destroy(mgr); return NULL; + } mgr->bg_ping_cycle_start_tb = get_time_tb(); mgr->initialized = 1; DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: initialized for group %016llx", (unsigned long long)group->group_id); return mgr; @@ -514,6 +535,11 @@ void conn_mgr_destroy(struct CONN_MGR* mgr) { if (mgr->cand_tcp_pings[i]) stcp_client_destroy(mgr->cand_tcp_pings[i]); while (mgr->entries->head) cm_entry_cleanup((struct CONN_MGR_ENTRY*)mgr->entries->head); queue_free(mgr->entries); + while (mgr->handles) { + struct CONN_MGR_HANDLE* h = mgr->handles; + mgr->handles = h->owner_next; h->mgr = NULL; h->owner_next = NULL; + } + cm_collect(mgr); DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: destroyed group=%016llx", (unsigned long long)mgr->group->group_id); u_free(mgr); } @@ -539,7 +565,11 @@ int cm_open(struct CONN_MGR* mgr, uint64_t node_id, if (!target && mgr->instance->topo_sqlite_db && mgr->instance->topo_groups) { struct TOPO_NODE* ni = topo_node_sqlite_node_load(mgr->instance->topo_sqlite_db, mgr->instance->topo_groups, node_id); if (ni) { - ni = topo_node_registry_store(mgr->instance->topo_groups, ni); + struct TOPO_NODE* stored = topo_node_registry_store(mgr->instance->topo_groups, ni); + if (!stored) { + DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: cannot register DB node=%016llx", (unsigned long long)node_id); + topo_node_destroy(mgr->instance->topo_groups, ni); return -1; + } struct ll_entry* qe = queue_entry_new(sizeof(struct TOPO_GROUP_NODE) - sizeof(struct ll_entry)); if (qe) { target = (struct TOPO_GROUP_NODE*)qe; @@ -547,7 +577,10 @@ int cm_open(struct CONN_MGR* mgr, uint64_t node_id, target->node_id = node_id; target->conn_mgr_type = CONN_TYPE_NONE; queue_data_put_with_index(group->nodes, &target->ll); loaded_from_db = 1; DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: node 0x%016llx loaded from DB", (unsigned long long)node_id); - } else DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: failed to alloc TOPO_GROUP_NODE for DB node 0x%016llx", (unsigned long long)node_id); + } else { + DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: failed to alloc TOPO_GROUP_NODE for DB node 0x%016llx", (unsigned long long)node_id); + topo_node_registry_unref(mgr->instance->topo_groups, node_id); return -1; + } } } if (!target) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "node 0x%016llx skipped — not in routing, no active connections", (unsigned long long)node_id); return -1; } @@ -555,28 +588,26 @@ int cm_open(struct CONN_MGR* mgr, uint64_t node_id, struct CONN_MGR_ENTRY* entry = cm_find_entry(mgr, node_id); if (entry && entry->handles) { + entry->db_loaded |= (uint8_t)loaded_from_db; struct CONN_MGR_HANDLE* h = cm_handle_new(entry, gid, cb, cb_arg); if (!h) return -1; if (out_handle) *out_handle = h; - if (entry->state == CONN_MGR_STATE_CONNECTED) - h->deliver_up_token = uasync_call_soon(mgr->instance->ua, h, cm_deliver_up_cb); DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "conn_mgr: joined demand peer=%016llx state=%u retry=%llu", (unsigned long long)node_id, entry->state, (unsigned long long)entry->retry_tb); return 0; } entry = cm_ensure_entry(mgr, node_id); - if (!entry) return -1; + if (!entry) { if (loaded_from_db) topo_nodeq_remove_node(group, target); return -1; } + entry->db_loaded |= (uint8_t)loaded_from_db; int already_up = entry->state == CONN_MGR_STATE_CONNECTED; if (!already_up) entry->state = CONN_MGR_STATE_CONNECTING; struct CONN_MGR_HANDLE* h = cm_handle_new(entry, gid, cb, cb_arg); if (!h) { cm_entry_cleanup(entry); return -1; } if (out_handle) *out_handle = h; // Первый владелец входящего пути начинает свои пробы, сохраняя рабочую доставку. if (entry->timer) { uasync_cancel_timeout(mgr->instance->ua, entry->timer); entry->timer = NULL; } - if (already_up) h->deliver_up_token = uasync_call_soon(mgr->instance->ua, h, cm_deliver_up_cb); entry->last_traffic_tb = get_time_tb(); entry->timeout_notified = 0; entry->retry_delay_ms = CONN_MGR_RETRY_MIN_MS; - entry->db_loaded |= (uint8_t)loaded_from_db; cm_start_attempt(entry, 0); if (entry->state == CONN_MGR_STATE_DISCONNECTED) return 0; cm_arm_timer(entry); @@ -597,7 +628,12 @@ void conn_mgr_close(struct CONN_MGR_HANDLE* h) { if (!entry->handles) cm_entry_cleanup(entry); } // Не посылаем DISCONNECT: удалённые владельцы и групповой транспорт живут независимо. - uasync_call_soon(h->ua, h, cm_handle_free_cb); + struct CONN_MGR* mgr = h->mgr; + if (!mgr) { u_free(h); return; } /* менеджер уже разрушен вне callbacks */ + struct CONN_MGR_HANDLE** owner = &mgr->handles; + while (*owner && *owner != h) owner = &(*owner)->owner_next; + if (*owner) *owner = h->owner_next; + h->owner_next = mgr->retired_handles; mgr->retired_handles = h; } struct ETCP_CONN* conn_mgr_get_conn(struct CONN_MGR_HANDLE* h) { diff --git a/src/routing_layer/conn_mgr_monitor.c b/src/routing_layer/conn_mgr_monitor.c index 28043e25..8560abd8 100644 --- a/src/routing_layer/conn_mgr_monitor.c +++ b/src/routing_layer/conn_mgr_monitor.c @@ -111,6 +111,8 @@ static void cm_bg_ping_to_node(struct CONN_MGR* mgr, struct TOPO_NODE* ni) { * и INDIRECT-фаза не может выбрать посредника. */ void cm_bg_ping_timer_cb(void* arg) { struct CONN_MGR* mgr = (struct CONN_MGR*)arg; + mgr->bg_ping_timer = NULL; mgr->bg_ping_wait = NULL; + cm_collect(mgr); uint64_t nat_delay = nat_detection_probe_delay(mgr->instance->nat_det); if (nat_delay) { mgr->bg_ping_timer = uasync_set_timeout(mgr->instance->ua, nat_delay, mgr, cm_bg_ping_timer_cb, "cm_bg_nat_defer"); @@ -165,6 +167,8 @@ static int cm_has_active_conn(struct TOPO_GROUP* g, uint64_t nid) { * Вызывается раз в ~2с, самоперезапускается. */ void cm_candidate_ping_timer_cb(void* arg) { struct CONN_MGR* mgr = (struct CONN_MGR*)arg; uint64_t now = get_time_tb(); + mgr->candidate_ping_timer = NULL; mgr->candidate_ping_wait = NULL; + cm_collect(mgr); uint64_t nat_delay = nat_detection_probe_delay(mgr->instance->nat_det); if (nat_delay) { mgr->candidate_ping_timer = uasync_set_timeout(mgr->instance->ua, nat_delay, mgr, cm_candidate_ping_timer_cb, "cm_candidate_nat_defer"); diff --git a/src/routing_layer/conn_mgr_priv.h b/src/routing_layer/conn_mgr_priv.h index 66343118..63b0a2a7 100644 --- a/src/routing_layer/conn_mgr_priv.h +++ b/src/routing_layer/conn_mgr_priv.h @@ -92,8 +92,10 @@ struct CM_DISCONNECT { uint8_t cmd, subcmd; uint64_t node_id; }; struct CONN_MGR_HANDLE { uint64_t node_id; uint64_t group_id; conn_mgr_cb_t cb; void* cb_arg; struct CONN_MGR_ENTRY* entry; struct CONN_MGR_HANDLE* next; + struct CONN_MGR* mgr; /* владелец памяти до destroy; NULL у отсоединённого после destroy handle */ + struct CONN_MGR_HANDLE* owner_next; /* список всех handles mgr либо список закрытых до безопасного GC */ struct UASYNC* ua; - uint8_t closed; /* close отменяет callbacks; память освобождается вне текущего стека */ + uint8_t closed; /* close отменяет callbacks; менеджер освобождает память вне текущего стека */ void* deliver_up_token; /* токен отложенного cm_deliver_up_cb (uasync_call_soon), NULL если не постился/уже сработал */ }; @@ -115,11 +117,14 @@ struct CONN_MGR_ENTRY { uint32_t reverse_request_id, exchange_request_id, peer_exchange_request_id, peer_reverse_request_id; struct CM_INTERM_SEL selection; /* Исходящий выбор до подтверждения пира. */ struct CONN_MGR* mgr; + struct CONN_MGR_ENTRY* retired_next; /* уже снятая с entries запись до безопасного GC */ }; struct CONN_MGR { struct UTUN_INSTANCE* instance; struct TOPO_GROUP* group; struct ll_queue* entries; + struct CONN_MGR_HANDLE *handles, *retired_handles; + struct CONN_MGR_ENTRY* retired_entries; /* сборка в фоновых callbacks/destroy не требует выделений памяти */ void *bg_ping_timer, *candidate_ping_timer; void *bg_ping_wait, *candidate_ping_wait; /* standby_wait handle (Android) */ size_t bg_ping_cursor; uint64_t bg_ping_cycle_start_tb; @@ -138,6 +143,7 @@ int cm_open(struct CONN_MGR* mgr, uint64_t node_id, struct CONN_MGR_ENTRY* cm_find_entry(struct CONN_MGR* mgr, uint64_t node_id); struct CONN_MGR_ENTRY* cm_ensure_entry(struct CONN_MGR* mgr, uint64_t node_id); void cm_entry_cleanup(struct CONN_MGR_ENTRY* entry); +void cm_collect(struct CONN_MGR* mgr); /* только вне вложенных CM-вызовов: начало фонового callback либо destroy */ void cm_cleanup_db_node(struct CONN_MGR_ENTRY* entry); void cm_deliver_event(struct CONN_MGR_ENTRY* entry, enum conn_mgr_event event); void cm_update_nodeinfo(struct CONN_MGR_ENTRY* entry);