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

Initialize inbound graceful close and honor remote transport ownership

master
evgeny 14 hours ago
parent
3cfdb9e6ec
commit
f090d697c4
3 changed files with 55 additions and 26 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 5
      src/transport_layer/etcp_connections.c
  2. 69
      src/transport_layer/node_conn_direct.c
  3. 7
      src/transport_layer/node_conn_direct.h

5
src/transport_layer/etcp_connections.c
Unescape View File

@ -1184,6 +1184,9 @@ static void etcp_inflight_nullify_link(struct ll_queue* q, struct ETCP_LINK* dea
void etcp_link_close(struct ETCP_LINK* link) {
DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "");
if (!link) return;
if (link->rx_timeout_timer) {
uasync_cancel_timeout(link->etcp->instance->ua, link->rx_timeout_timer); link->rx_timeout_timer = NULL;
}
callring_put(CR_LINK_CLOSE_IN, (uintptr_t)link, (uintptr_t)(link->etcp));
if (link->etcp && link->etcp->last_rr_link == link) link->etcp->last_rr_link = NULL;
@ -2607,6 +2610,7 @@ int etcp_packet_decrypted(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt,
link->last_recv_timestamp=pkt->timestamp;
link->last_recv_updated=1;
link->total_decrypted += pkt->data_len;
if (pkt->data[0] != ETCP_KEEPALIVE) etcp_keepalive_watch_rx(link);
uint8_t pkt_code = pkt->data[0];
@ -2902,6 +2906,7 @@ int init_connections(struct UTUN_INSTANCE* instance) {
}
}
node_conn_direct_init(instance);
etcp_keepalive_register(instance);
// Return 1 if there was a partial socket initialization error

69
src/transport_layer/node_conn_direct.c
Unescape View File

@ -26,6 +26,7 @@
#include "../lib/socket_compat.h"
#include "../lib/platform_compat.h"
#include <string.h>
#include <limits.h>
#define DEBUG_CATEGORY_NCD DEBUG_CATEGORY_ETCP
@ -649,11 +650,12 @@ static void ncd_fin_wait_timeout_cb(void* arg) {
*
* KEEP_ALIVE (пир просит не закрывать conn — у него ещё есть handle'ы):
* - Если мы в fin_wait — отменяем его, conn продолжает работать.
* - Если у нас handle'ов нет — перезапускаем fin_wait (ждём ещё 5 сек).
* - Отсутствие наших handle'ов не отменяет потребность пира: FIN_WAIT снимается полностью.
*/
static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e) {
if (!conn || !e || e->len < NCD_CONTROL_MSG_SIZE) {
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] malformed control conn=%p size=%u", (void*)conn, e ? e->len : 0);
if (e) { queue_dgram_free(e); queue_entry_free(e); }
return;
}
@ -676,13 +678,16 @@ static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e)
if (entry->handle_count > 0) {
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] refusing CLOSE, still have %d handles for 0x%016llx",
entry->handle_count, (unsigned long long)sender_id);
{ struct ncd_control_msg resp; resp.cmd = ETCP_RT_ID_NCD_CONTROL; resp.subcmd = NCD_SUBCMD_KEEP_ALIVE;
resp.node_id = conn->instance->node_id;
struct ll_entry* qe = queue_entry_new(0);
if (qe) { qe->dgram = u_malloc(sizeof(resp));
if (qe->dgram) { memcpy(qe->dgram, &resp, sizeof(resp)); qe->len = sizeof(resp);
if (etcp_send(conn, qe) != 0) { queue_dgram_free(qe); queue_entry_free(qe); } }
else queue_entry_free(qe); }
{ struct ncd_control_msg resp = { ETCP_RT_ID_NCD_CONTROL, NCD_SUBCMD_KEEP_ALIVE, conn->instance->node_id };
struct ll_entry* qe = ll_alloc_lldgram(sizeof(resp));
if (!qe) DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE allocation failed");
else {
memcpy(qe->dgram, &resp, sizeof(resp)); qe->len = sizeof(resp);
if (etcp_send(conn, qe) != 0) {
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE send failed node=%016llx", (unsigned long long)sender_id);
queue_dgram_free(qe); queue_entry_free(qe);
}
}
}
} else {
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] deferred close entry for CLOSE from 0x%016llx", (unsigned long long)sender_id);
@ -694,15 +699,12 @@ static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e)
if (entry && entry->conn->fin_wait) {
entry->conn->fin_wait = 0; entry->conn->fin_wait_clear_cb = NULL; entry->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; }
if (entry->handle_count <= 0) {
entry->conn->fin_wait = 1;
entry->fin_wait_timer = uasync_set_timeout(entry->ua, NCD_FIN_WAIT_TIMEOUT_TB, entry, ncd_fin_wait_timeout_cb, "ncd_fin_wait");
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE but no handles, restart fin_wait");
} else {
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE: fin_wait cleared, conn stays alive");
}
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] KEEP_ALIVE: fin_wait cleared, conn stays alive handles=%d", entry->handle_count);
}
break;
default:
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] unknown control subcmd=%u node=%016llx", msg->subcmd, (unsigned long long)sender_id);
break;
}
queue_dgram_free(e); queue_entry_free(e);
}
@ -710,15 +712,19 @@ static void ncd_recv_control_handler(struct ETCP_CONN* conn, struct ll_entry* e)
/* ═══════════ инициализация глобального обработчика ═══════════ */
/*
* Однократная привязка обработчика NCD control-сообщений к ETCP (etcp_bind).
* Вызывается при первом node_conn_direct_open, повторные вызовы безвредны
* Вызывается при запуске сети и node_conn_direct_open, повторные вызовы безвредны
* (защита через inst->ncd_control_bound).
*/
static void ncd_init_control_binding(struct UTUN_INSTANCE* inst) {
if (inst->ncd_control_bound) return;
etcp_bind(inst, ETCP_RT_ID_NCD_CONTROL, ncd_recv_control_handler);
int node_conn_direct_init(struct UTUN_INSTANCE* inst) {
if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] init without instance"); return -1; }
if (inst->ncd_control_bound) return 0;
if (etcp_bind(inst, ETCP_RT_ID_NCD_CONTROL, ncd_recv_control_handler) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot bind CLOSE/KEEP_ALIVE handler"); return -1;
}
inst->ncd_control_bound = 1;
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] bound ETCP_RT_ID_NCD_CONTROL handler");
return 0;
}
/* ═══════════ оживление «зависшего» (timed_out) entry ═══════════ */
@ -776,7 +782,7 @@ int node_conn_direct_open(struct UTUN_INSTANCE* inst, uint64_t node_id,
struct ETCP_SOCKET* specific_sock) {
if (!inst || !out_handle) return NCD_ERR;
*out_handle = NULL;
ncd_init_control_binding(inst);
if (node_conn_direct_init(inst) < 0) return NCD_ERR;
/* 1. Ищем в реестре — conn уже есть */
struct ncd_entry* entry = ncd_registry_find(inst, node_id);
@ -938,7 +944,7 @@ static int ncd_open_node(struct UTUN_INSTANCE* inst, uint64_t node_id,
struct ETCP_SOCKET* specific_sock) {
if (!inst || !out_handle || !ni) return NCD_ERR;
*out_handle = NULL;
ncd_init_control_binding(inst);
if (node_conn_direct_init(inst) < 0) return NCD_ERR;
/* 1. Ищем в реестре — conn уже есть */
{ struct ncd_entry* entry = ncd_registry_find(inst, node_id);
@ -1134,9 +1140,14 @@ static void ncd_handle_close(struct NODE_CONN_DIRECT* h, int force) {
(unsigned long long)entry->node_id, entry->handle_count, force);
if (entry->handle_count || entry->closed) { ncd_collect(entry); return; }
struct ETCP_CONN* conn = entry->conn;
if (force || !conn || conn->close_requested || conn->state == 2 || entry->timed_out || !conn->links_up) {
if (force || !conn || conn->close_requested || conn->state == 2 || entry->timed_out) {
ncd_finish(entry, 1); return;
}
if (!conn->links_up) {
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] last handle released while DOWN, retaining until ETCP timeout node=%016llx",
(unsigned long long)entry->node_id);
ncd_finish(entry, 0); return;
}
if (entry->connect_timer) { uasync_cancel_timeout(entry->ua, entry->connect_timer); entry->connect_timer = NULL; }
conn->fin_wait = 1; conn->fin_wait_clear_cb = ncd_fin_wait_cancelled; conn->fin_wait_clear_arg = entry;
struct ncd_control_msg msg = { ETCP_RT_ID_NCD_CONTROL, NCD_SUBCMD_CLOSE, entry->inst->node_id };
@ -1148,8 +1159,18 @@ static void ncd_handle_close(struct NODE_CONN_DIRECT* h, int force) {
DEBUG_WARN(DEBUG_CATEGORY_NCD, "[ncd] CLOSE send failed"); queue_dgram_free(e); queue_entry_free(e);
}
}
entry->fin_wait_timer = uasync_set_timeout(entry->ua, NCD_FIN_WAIT_TIMEOUT_TB, entry, ncd_fin_wait_timeout_cb, "ncd_fin_wait");
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] fin_wait node=%016llx", (unsigned long long)entry->node_id);
uint64_t now = get_time_tb(), wait_tb = NCD_FIN_WAIT_TIMEOUT_TB;
for (struct ETCP_LINK* link = conn->links; link; link = link->next) {
if (!link->link_status || (!link->ka_my_sleeping && !link->ka_peer_sleeping)) continue;
if (link->rx_deadline_tb > now && link->rx_deadline_tb - now > wait_tb) wait_tb = link->rx_deadline_tb - now;
}
int delay = wait_tb > INT_MAX ? INT_MAX : (int)wait_tb;
entry->fin_wait_timer = uasync_set_timeout(entry->ua, delay, entry, ncd_fin_wait_timeout_cb, "ncd_fin_wait");
if (!entry->fin_wait_timer) {
DEBUG_ERROR(DEBUG_CATEGORY_NCD, "[ncd] cannot schedule FIN_WAIT node=%016llx", (unsigned long long)entry->node_id);
ncd_finish(entry, 1); return;
}
DEBUG_INFO(DEBUG_CATEGORY_NCD, "[ncd] fin_wait node=%016llx timeout_ms=%d", (unsigned long long)entry->node_id, delay / 10);
ncd_collect(entry);
}
void node_conn_direct_close(struct NODE_CONN_DIRECT* h) { ncd_handle_close(h, 0); }

7
src/transport_layer/node_conn_direct.h
Unescape View File

@ -13,8 +13,8 @@
*
* Модуль управляет ETCP-соединениями через непрозрачные handle'ы (NODE_CONN_DIRECT).
* Каждый владелец открывает свой handle; несколько handle разделяют ETCP_CONN.
* close снимает только своё владение. Последний handle закрывает неготовый транспорт
* сразу; для UP начинает CLOSE/KEEP_ALIVE (до 5 s), пир может сохранить соединение.
* close снимает только своё владение. Последний handle работающего транспорта начинает
* CLOSE/KEEP_ALIVE (5 s либо срок пробуждения пира); DOWN удаляется по таймауту ETCP.
*
* Кто использует:
* - conn_mgr (Connection Manager) — временные handle'ы попыток установления прямого транспорта
@ -36,6 +36,9 @@ struct ETCP_CONN;
struct TOPO_NODE;
struct ETCP_SOCKET;
/* Привязать CLOSE/KEEP_ALIVE при запуске ядра, в том числе без исходящих handles. */
int node_conn_direct_init(struct UTUN_INSTANCE* inst);
/* ─── Return codes ─── */
#define NCD_NEW 0 /* создано новое соединение */
#define NCD_REUSED 1 /* переиспользовано существующее */

Write Preview
Loading…
Cancel
Save