diff --git a/src/transport_layer/node_conn_direct.c b/src/transport_layer/node_conn_direct.c index 216ee6a4..a7d2fa20 100644 --- a/src/transport_layer/node_conn_direct.c +++ b/src/transport_layer/node_conn_direct.c @@ -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)