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

Make CM cleanup allocation-free and reject incomplete startup

Browse Source
This commit is contained in:
evgeny committed 2026-10-24 16:09:39 +02:00
1 parent 007f7c8ae8
commit 80451c43be
3 files changed
+66 -20
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/routing_layer/conn_mgr_core.c
+55 -19
View File
@@ -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 <stdlib.h>
#include <string.h>
#ifdef _WIN32
#include <winsock2.h>
#include <ws2tcpip.h>
#else
#include <arpa/inet.h>
#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) {
src/routing_layer/conn_mgr_monitor.c
+4
View File
@@ -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");
src/routing_layer/conn_mgr_priv.h
+7 -1
View File
@@ -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);