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

Register NCD transport events atomically and reuse ETCP close lifecycle

Browse Source
This commit is contained in:
evgeny committed 2026-10-24 20:15:30 +02:00
1 parent 28c951ed9e
commit 4695fe4371
1 file changed
+34 -72
Download Patch File Download Diff File Expand all files Collapse all files
src/transport_layer/node_conn_direct.c
+34 -72
View File
@@ -42,7 +42,7 @@ struct ncd_entry {
struct ETCP_CONN* conn;
struct UASYNC* ua;
struct UTUN_INSTANCE* inst;
unsigned dispatch_depth;
unsigned active_calls; /* удержание entry внутри операций и callback-рассылок */
uint64_t event_generation;
uint8_t closed;
uint8_t node_ref;
@@ -72,14 +72,10 @@ struct NODE_CONN_DIRECT {
static void ncd_event_dispatch(struct ncd_entry* entry, enum ncd_event event);
static void ncd_connect_timeout_cb(void* arg);
static void ncd_deliver_up_cb(void* arg);
static void ncd_init_cb(struct ETCP_CONN* conn, int event, void* arg);
static void ncd_up_cb(struct ETCP_CONN* conn, int event, void* arg);
static void ncd_down_cb(struct ETCP_CONN* conn, int event, void* arg);
static void ncd_delete_cb(struct ETCP_CONN* conn, int event, void* arg);
static void ncd_conn_event(struct ETCP_CONN* conn, int event, void* arg);
static void ncd_finish(struct ncd_entry* entry, int close_conn);
static void ncd_collect(struct ncd_entry* entry);
static void ncd_cancel_deliver_up(struct NODE_CONN_DIRECT* h, struct UASYNC* ua);
static void ncd_deferred_close_conn(void* arg);
/* ═══════════ реестр ═══════════ */
/*
@@ -105,7 +101,13 @@ int node_conn_direct_handle_count(const struct ETCP_CONN* conn) {
static int ncd_registry_add(struct UTUN_INSTANCE* inst, struct ncd_entry* entry) {
entry->inst = inst;
if (etcp_conn_ref_take(entry->conn) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] register closing conn"); return -1; }
etcp_conn_add_cbk(entry->conn, ncd_delete_cb, entry, ETCP_CBK_EVENT_DELETE);
int events = ETCP_CBK_EVENT_INIT | ETCP_CBK_EVENT_UP | ETCP_CBK_EVENT_DOWN | ETCP_CBK_EVENT_DELETE;
etcp_conn_add_cbk(entry->conn, ncd_conn_event, entry, events);
struct etcp_cbk_entry* subscription = entry->conn->cbks;
if (!subscription || subscription->fn != ncd_conn_event || subscription->arg != entry || subscription->event_mask != events) {
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot subscribe to transport node=%016llx", (unsigned long long)entry->node_id);
etcp_conn_ref_free(entry->conn); return -1;
}
entry->next = (struct ncd_entry*)inst->ncd_registry; inst->ncd_registry = entry;
return 0;
}
@@ -475,8 +477,8 @@ static void ncd_event_dispatch(struct ncd_entry* entry, enum ncd_event event) {
case NCD_EVENT_CLOSED: entry->up = 0; break;
}
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] event=%d node=%016llx handles=%d depth=%u", event,
(unsigned long long)entry->node_id, entry->handle_count, entry->dispatch_depth);
entry->dispatch_depth++;
(unsigned long long)entry->node_id, entry->handle_count, entry->active_calls);
entry->active_calls++;
uint64_t generation = ++entry->event_generation;
// Закрытые элементы не освобождаются до выхода из внешней рассылки. Новые добавляются в голову.
for (struct NODE_CONN_DIRECT* h = entry->handles; h; h = h->next) {
@@ -487,7 +489,7 @@ static void ncd_event_dispatch(struct ncd_entry* entry, enum ncd_event event) {
if (event != NCD_EVENT_UP) ncd_cancel_deliver_up(h, entry->ua);
if (!h->closed && h->cb && (!entry->closed || event == NCD_EVENT_CLOSED)) h->cb(h, event, h->cb_arg);
}
entry->dispatch_depth--;
entry->active_calls--;
ncd_collect(entry);
}
@@ -502,31 +504,22 @@ static void ncd_event_dispatch(struct ncd_entry* entry, enum ncd_event event) {
* получает UP только когда conn реально готов отправлять данные.
*/
static void ncd_init_cb(struct ETCP_CONN* conn, int event, void* arg) { (void)event;
struct ncd_entry* entry = (struct ncd_entry*)arg;
if (!entry || conn->fin_wait) return;
if (conn->links_up > 0) ncd_event_dispatch(entry, NCD_EVENT_UP);
}
static void ncd_up_cb(struct ETCP_CONN* conn, int event, void* arg) { (void)event;
struct ncd_entry* entry = (struct ncd_entry*)arg;
if (!entry || conn->fin_wait) return;
ncd_event_dispatch(entry, NCD_EVENT_UP);
}
// Каждый отложенный запрос удерживает ETCP до завершения callback, в том числе при повторном close.
static void ncd_deferred_close_conn(void* arg) {
struct ETCP_CONN* conn = arg;
etcp_connection_close(conn);
etcp_conn_ref_free(conn);
}
static void ncd_schedule_close(struct ETCP_CONN* conn) {
if (!conn || conn->close_requested || conn->state == 2) return;
if (etcp_conn_ref_take(conn) != 0) return;
conn->close_requested = 1;
uasync_call_soon(conn->instance->ua, conn, ncd_deferred_close_conn);
static void ncd_conn_event(struct ETCP_CONN* conn, int event, void* arg) {
struct ncd_entry* entry = arg;
if (event == ETCP_CBK_EVENT_DELETE) { ncd_finish(entry, 0); return; }
if (event == ETCP_CBK_EVENT_DOWN) {
if (conn->fin_wait && !entry->handle_count) {
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] DOWN confirms FIN_WAIT node=%016llx", (unsigned long long)entry->node_id);
ncd_finish(entry, 1); return;
}
ncd_event_dispatch(entry, NCD_EVENT_DOWN); return;
}
if (!conn->fin_wait && !conn->close_requested && conn->state == 1 && conn->links_up)
ncd_event_dispatch(entry, NCD_EVENT_UP);
}
static void ncd_collect(struct ncd_entry* entry) {
if (entry->dispatch_depth) return;
if (entry->active_calls) return;
struct NODE_CONN_DIRECT** pp = &entry->handles;
while (*pp) {
struct NODE_CONN_DIRECT* h = *pp;
@@ -541,6 +534,7 @@ static void ncd_collect(struct ncd_entry* entry) {
// Сначала исключение из реестра и отсоединение транспорта, затем уведомление владельцев.
static void ncd_finish(struct ncd_entry* entry, int close_conn) {
if (entry->closed) return;
entry->active_calls++;
entry->closed = 1;
ncd_registry_remove(entry->inst, entry);
if (entry->node_ref) {
@@ -550,19 +544,16 @@ static void ncd_finish(struct ncd_entry* entry, int close_conn) {
if (entry->fin_wait_timer) { uasync_cancel_timeout(entry->ua, entry->fin_wait_timer); entry->fin_wait_timer = NULL; }
struct ETCP_CONN* conn = entry->conn; entry->conn = NULL;
if (conn) {
conn->fin_wait = 0; conn->fin_wait_clear_cb = NULL; conn->fin_wait_clear_arg = NULL;
etcp_conn_remove_cbk(conn, ncd_init_cb, entry);
etcp_conn_remove_cbk(conn, ncd_up_cb, entry);
etcp_conn_remove_cbk(conn, ncd_down_cb, entry);
etcp_conn_remove_cbk(conn, ncd_delete_cb, entry);
if (close_conn) ncd_schedule_close(conn);
if (conn->fin_wait_clear_arg == entry) {
conn->fin_wait = 0; conn->fin_wait_clear_cb = NULL; conn->fin_wait_clear_arg = NULL;
}
etcp_conn_remove_cbk(conn, ncd_conn_event, entry);
if (close_conn) etcp_connection_close(conn); /* ETCP сам откладывает close из своих callbacks. */
etcp_conn_ref_free(conn);
}
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] detached node=%016llx handles=%d", (unsigned long long)entry->node_id, entry->handle_count);
ncd_event_dispatch(entry, NCD_EVENT_CLOSED); // collect может освободить entry
}
static void ncd_delete_cb(struct ETCP_CONN* conn, int event, void* arg) {
(void)conn; (void)event; ncd_finish(arg, 0);
ncd_event_dispatch(entry, NCD_EVENT_CLOSED);
entry->active_calls--; ncd_collect(entry);
}
/*
@@ -570,19 +561,6 @@ static void ncd_delete_cb(struct ETCP_CONN* conn, int event, void* arg) {
* Если conn в fin_wait и handle'ов нет — peer подтвердил закрытие,
* делаем немедленную очистку. Иначе транслируем NCD_EVENT_DOWN потребителям.
*/
static void ncd_down_cb(struct ETCP_CONN* conn, int event, void* arg) { (void)event;
struct ncd_entry* entry = (struct ncd_entry*)arg;
if (!entry) return;
if (conn->fin_wait && entry->handle_count <= 0) {
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] DOWN during fin_wait, cleaning up node=0x%016llx", (unsigned long long)entry->node_id);
conn->fin_wait = 0; conn->fin_wait_clear_cb = NULL; conn->fin_wait_clear_arg = NULL;
if (entry->fin_wait_timer) { uasync_cancel_timeout(entry->ua, entry->fin_wait_timer); entry->fin_wait_timer = NULL; }
ncd_finish(entry, 1);
return;
}
ncd_event_dispatch(entry, NCD_EVENT_DOWN);
}
/* ═══════════ таймер подключения ═══════════ */
/*
* Таймер первого подключения истёк — соединение не установилось за etcp_connect_timeout_tb.
@@ -694,7 +672,7 @@ static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e)
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] recv CLOSE from 0x%016llx", (unsigned long long)sender_id);
if (!entry) {
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] no entry for CLOSE sender 0x%016llx, deferred close conn", (unsigned long long)sender_id);
ncd_schedule_close(conn);
etcp_connection_close(conn);
break;
}
if (entry->handle_count > 0) {
@@ -859,10 +837,6 @@ int node_conn_direct_open(struct UTUN_INSTANCE* inst, uint64_t node_id,
entry->handle_count = 1;
*out_handle = h;
etcp_conn_add_cbk(conn, ncd_init_cb, entry, ETCP_CBK_EVENT_INIT);
etcp_conn_add_cbk(conn, ncd_up_cb, entry, ETCP_CBK_EVENT_UP);
etcp_conn_add_cbk(conn, ncd_down_cb, entry, ETCP_CBK_EVENT_DOWN);
if (conn->state == 1 && entry->up) {
ncd_schedule_deliver_up(inst->ua, h);
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] open REUSED new-entry (ready) node=0x%016llx conn=%p", (unsigned long long)node_id, (void*)conn);
@@ -929,10 +903,6 @@ int node_conn_direct_open(struct UTUN_INSTANCE* inst, uint64_t node_id,
entry->handle_count = 1;
*out_handle = h;
etcp_conn_add_cbk(conn, ncd_init_cb, entry, ETCP_CBK_EVENT_INIT);
etcp_conn_add_cbk(conn, ncd_up_cb, entry, ETCP_CBK_EVENT_UP);
etcp_conn_add_cbk(conn, ncd_down_cb, entry, ETCP_CBK_EVENT_DOWN);
int link_count = ncd_create_links(entry, ni, specific_sock, NULL);
if (link_count == 0)
@@ -1011,10 +981,6 @@ static int ncd_open_node(struct UTUN_INSTANCE* inst, uint64_t node_id,
entry->handle_count = 1;
*out_handle = h;
etcp_conn_add_cbk(conn, ncd_init_cb, entry, ETCP_CBK_EVENT_INIT);
etcp_conn_add_cbk(conn, ncd_up_cb, entry, ETCP_CBK_EVENT_UP);
etcp_conn_add_cbk(conn, ncd_down_cb, entry, ETCP_CBK_EVENT_DOWN);
if (conn->state == 1 && entry->up) {
ncd_create_links(entry, ni, specific_sock, NULL);
ncd_schedule_deliver_up(inst->ua, h);
@@ -1068,10 +1034,6 @@ static int ncd_open_node(struct UTUN_INSTANCE* inst, uint64_t node_id,
entry->handle_count = 1;
*out_handle = h;
etcp_conn_add_cbk(conn, ncd_init_cb, entry, ETCP_CBK_EVENT_INIT);
etcp_conn_add_cbk(conn, ncd_up_cb, entry, ETCP_CBK_EVENT_UP);
etcp_conn_add_cbk(conn, ncd_down_cb, entry, ETCP_CBK_EVENT_DOWN);
int link_count = ncd_create_links(entry, ni, specific_sock, NULL);
if (link_count == 0)