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

etcp_router: корректный handshake/restart без таймаут-петли

Browse Source 
- START на первом пакете сеанса (seq=0), гейт отправки до первого ACK
- дедуп рестарта по payload init-пакета (ретрансмит START не триггерит рестарт)
- RST переинициализирует только отправку (без CLOSE_ALL), дедуп повторных RST
This commit is contained in:
evgeny committed 2026-09-06 00:33:45 +03:00
1 parent 232ad2de6b
commit 53bce51201
3 files changed
+107 -20
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/routing_layer/etcp_router.c
+75 -17
View File
@@ -34,6 +34,7 @@ static void router_try_assembly(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN
static void router_deliver(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, const uint8_t* wire, size_t wire_len);
static void router_send_ctrl(struct ETCP_ROUTER_CONN* rconn, uint8_t flag_bits);
static void router_send_rst(struct ETCP_ROUTER_CONN* rconn);
static void router_restart_send(struct ETCP_ROUTER_CONN* rconn);
static struct ETCP_CONN* router_send_conn(struct ETCP_ROUTER_CONN* rconn);
static void router_send_kick(struct ETCP_ROUTER_CONN* rconn);
static void router_send_drain_cb(struct ll_queue* q, void* arg);
@@ -49,7 +50,7 @@ static void router_conn_reset(struct ETCP_ROUTER_CONN* rconn);
static void free_entry(struct ll_entry* e) { queue_dgram_free(e); queue_entry_free(e); }
static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN* rconn, uint32_t seq, uint64_t src_node_id, uint16_t ack_ts);
static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* entry, struct SVC_ROUTE_HDR* hdr);
static int router_check_peer_restart(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN** prconn, struct SVC_ROUTE_HDR* hdr);
static int router_check_peer_restart(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN** prconn, struct SVC_ROUTE_HDR* hdr, size_t wire_len);
static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, uint32_t seq, const uint8_t* wire, size_t wire_len);
static void router_retransmit_one(struct ETCP_ROUTER_CONN* rconn, struct ROUTER_INFLIGHT* inf);
static void router_retrans_schedule(struct ETCP_ROUTER_CONN* rconn);
@@ -279,6 +280,33 @@ static void router_send_rst(struct ETCP_ROUTER_CONN* rconn) {
}
}
// Рестарт только отправки (по RST от пира): сброс tx-состояния, очистка send_q/inflight_q.
// Приёмная сторона (rx) и сервис не трогаются. Повторные RST игнорируются через send_restart_pending.
static void router_restart_send(struct ETCP_ROUTER_CONN* rconn) {
struct UTUN_INSTANCE* inst = rconn->inst;
if (rconn->retrans_timer) { uasync_cancel_timeout(inst->ua, rconn->retrans_timer); rconn->retrans_timer = NULL; }
if (rconn->watchdog_timer) { uasync_cancel_timeout(inst->ua, rconn->watchdog_timer); rconn->watchdog_timer = NULL; }
{
struct ETCP_CONN* c = router_send_conn(rconn);
if (c && c->send_input_q) queue_waiter_cancel(c->send_input_q, &rconn->send_waiter);
memset(&rconn->send_waiter, 0, sizeof(rconn->send_waiter));
}
struct ll_entry* f;
if (rconn->send_q) {
while ((f = queue_data_get(rconn->send_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); }
}
if (rconn->inflight_q) {
while ((f = queue_data_get(rconn->inflight_q)) != NULL) {
struct ROUTER_INFLIGHT* inf = (struct ROUTER_INFLIGHT*)f;
if (inf->dgram) u_free(inf->dgram);
u_free(inf);
}
}
rconn->tx_seq = 0; rconn->tx_acked = 0; rconn->start_sent = 0;
rconn->send_blocked = 0; rconn->no_ack_count = 0; rconn->last_ack_changed_tb = 0;
rconn->send_restart_pending = 1;
}
static struct ETCP_CONN* loopback_conn(struct UTUN_INSTANCE* inst) {
static struct ETCP_CONN c;
memset(&c, 0, sizeof(c));
@@ -345,6 +373,7 @@ static void router_conn_free_queues(struct ETCP_ROUTER_CONN* rconn) {
queue_resume_callback(rconn->incoming_q);
queue_free(rconn->incoming_q); rconn->incoming_q = NULL;
}
if (rconn->last_start_dgram) { u_free(rconn->last_start_dgram); rconn->last_start_dgram = NULL; rconn->last_start_dgram_len = 0; }
}
// Пересоздать очереди и сбросить всё состояние rconn (кроме identity: group/remote/svc/ll.data/inst/sess_id/last_dgram_ts).
@@ -373,7 +402,7 @@ static void router_conn_reset(struct ETCP_ROUTER_CONN* rconn) {
rconn->tx_seq = 0; rconn->rx_seq = 0; rconn->tx_acked = 0;
rconn->last_sent_ack_seq = 0; rconn->last_ack_sent_tb = 0;
rconn->send_blocked = 0; rconn->start_sent = 0; rconn->peer_sync_done = 0;
rconn->send_blocked = 0; rconn->start_sent = 0; rconn->peer_sync_done = 0; rconn->send_restart_pending = 0;
rconn->no_route = 0; rconn->no_route_timer = NULL; rconn->no_ack_count = 0; rconn->closed = 0;
rconn->c_pkts_sent = 0; rconn->c_pkts_send_err = 0; rconn->c_pkts_rcvd = 0;
rconn->c_ack_sent = 0; rconn->c_ack_recv = 0; rconn->c_retrans_done = 0;
@@ -410,6 +439,7 @@ static void router_set_no_route(struct ETCP_ROUTER_CONN* rconn) {
static void router_send_kick(struct ETCP_ROUTER_CONN* rconn) {
if (rconn->closed || rconn->no_route) return;
if (queue_entry_count(rconn->send_q) == 0) return;
if (!rconn->start_sent && queue_entry_count(rconn->inflight_q) > 0) { rconn->send_blocked = 1; return; }
if (queue_entry_count(rconn->inflight_q) >= (int)router_effective_max_inflight(rconn)) {
rconn->send_blocked = 1;
return;
@@ -468,6 +498,7 @@ static void router_send_drain_cb(struct ll_queue* q, void* arg) {
if (!rconn->retrans_timer) router_retrans_schedule(rconn);
}
router_track_inflight_state(rconn);
if (!rconn->start_sent) { rconn->send_blocked = 1; return; }
}
if (queue_entry_count(rconn->send_q) == 0) { rconn->send_blocked = 0; router_send_watchdog_disarm(rconn); return; }
@@ -487,7 +518,7 @@ static void router_send_watchdog_cb(void* arg) {
rconn->watchdog_timer = NULL;
if (rconn->closed) return;
if (queue_entry_count(rconn->send_q) == 0) return;
if (!rconn->no_route
if (!rconn->no_route && !rconn->send_blocked
&& queue_entry_count(rconn->inflight_q) < (int)router_effective_max_inflight(rconn)
&& !rconn->send_waiter.internal && !rconn->send_waiter.call_soon_id) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router watchdog: send_q stalled svc_id=%u send_q=%d inflight=%d blocked=%d — force kick",
@@ -940,7 +971,7 @@ static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CON
if ((int32_t)(seq - rconn->tx_acked) >= 0) {
uint32_t old_acked = rconn->tx_acked;
rconn->tx_acked = seq;
if (rconn->tx_acked > 0) rconn->start_sent = 1;// первый ACK подтвердил seq 0 — сессия больше не «новая»
if (rconn->tx_acked > 0) { rconn->start_sent = 1; rconn->send_restart_pending = 0; }// первый ACK подтвердил seq 0 — сессия больше не «новая»
uint32_t clean_seq = old_acked;
while ((int32_t)(clean_seq - rconn->tx_acked) < 0) {
struct ROUTER_INFLIGHT* inf = (struct ROUTER_INFLIGHT*)queue_find_data_by_index(rconn->inflight_q, &clean_seq);
@@ -1022,17 +1053,25 @@ static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry*
}
static int router_check_peer_restart(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN** prconn,
struct SVC_ROUTE_HDR* hdr) {
struct SVC_ROUTE_HDR* hdr, size_t wire_len) {
struct ETCP_ROUTER_CONN* rconn = *prconn;
if ((hdr->flags & ROUTER_FLAG_START) && rconn->rx_seq > 0) {
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE,
"router: peer restart svc_id=%u from %016llx",
hdr->svc_id, (unsigned long long)hdr->src_node_id);
etcp_router_conn_restart(inst, hdr->group_id, hdr->src_node_id, hdr->svc_id);
*prconn = etcp_router_conn_get(inst, hdr->group_id, hdr->src_node_id, hdr->svc_id);
if (!*prconn) return -1;
rconn = *prconn;
if (!(hdr->flags & ROUTER_FLAG_START)) return 0; // не START — не рестарт
if (!rconn->peer_sync_done) return 0; // чистый: принимаем как первый пакет сеанса
// dirty + START: дедуп по payload. Ретрансмит уже принятого init-пакета побайтно идентичен —
// это НЕ новый сеанс, рестарт не делаем (пере-ACK сделает dup-ветка в router_handle_data_packet).
if (rconn->last_start_dgram && rconn->last_start_dgram_len == wire_len
&& memcmp(rconn->last_start_dgram, hdr, wire_len) == 0) {
return 0;
}
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE,
"router: peer restart svc_id=%u from %016llx",
hdr->svc_id, (unsigned long long)hdr->src_node_id);
etcp_router_conn_restart(inst, hdr->group_id, hdr->src_node_id, hdr->svc_id);
*prconn = etcp_router_conn_get(inst, hdr->group_id, hdr->src_node_id, hdr->svc_id);
if (!*prconn) return -1;
rconn = *prconn;
return 0;
}
@@ -1041,7 +1080,17 @@ static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETC
rconn->recv_conn = conn;
rconn->last_dgram_ts = get_current_timestamp();
if (!rconn->peer_sync_done) { rconn->peer_sync_done = 1; rconn->rx_seq = seq; }
if (!rconn->peer_sync_done) {
rconn->peer_sync_done = 1;
rconn->rx_seq = seq;
// сохранить init-пакет (START) для дедупа ретрансмитов
if (((const struct SVC_ROUTE_HDR*)wire)->flags & ROUTER_FLAG_START) {
if (rconn->last_start_dgram) u_free(rconn->last_start_dgram);
rconn->last_start_dgram = u_malloc(wire_len);
if (rconn->last_start_dgram) { memcpy(rconn->last_start_dgram, wire, wire_len); rconn->last_start_dgram_len = wire_len; }
else rconn->last_start_dgram_len = 0;
}
}
{// SEQ out of bounds
int32_t d = (int32_t)(seq - rconn->rx_seq);
@@ -1100,8 +1149,17 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, hdr->group_id, hdr->src_node_id, hdr->svc_id);
if (pl_len == 0 && (hdr->flags & ROUTER_FLAG_RST)) {
// Пир отклонил нашу «старую» сессию — сброс локального состояния (пере-шлём START).
if (rconn) etcp_router_conn_restart(inst, hdr->group_id, hdr->src_node_id, hdr->svc_id);
// Пир просит переинициализировать нашу отправку (его приём «чистый»).
if (rconn) {
if (rconn->send_restart_pending) {
DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router: RST (дубль) от %016llx svc_id=%u — игнор",
(unsigned long long)hdr->src_node_id, hdr->svc_id);
} else {
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router: RST от %016llx svc_id=%u — рестарт отправки",
(unsigned long long)hdr->src_node_id, hdr->svc_id);
router_restart_send(rconn);
}
}
free_entry(entry); return;
}
if (pl_len == 0 && (hdr->flags & ROUTER_FLAG_CLOSE)) {
@@ -1114,7 +1172,7 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
rconn = etcp_router_conn_get(inst, hdr->group_id, hdr->src_node_id, hdr->svc_id);
if (!rconn) { free_entry(entry); return; }
}
if (router_check_peer_restart(inst, &rconn, hdr) != 0) { free_entry(entry); return; }
if (router_check_peer_restart(inst, &rconn, hdr, entry->len) != 0) { free_entry(entry); return; }
// Сервер после рестарта «ждёт START»: пришёл не-START пакет от старой сессии → RST.
if (!rconn->peer_sync_done && !(hdr->flags & ROUTER_FLAG_START)) {
src/routing_layer/etcp_router.h
+3
View File
@@ -121,6 +121,9 @@ struct ETCP_ROUTER_CONN {
uint32_t sess_id; // локальный счётчик переустановок (0 при создании, ++ при рестарте; на wire не пишется)
uint8_t start_sent; // 0 = нужно отправить START в первом data
uint8_t peer_sync_done; // 0 = rx_seq ещё не синхронизирован с удалённым seq (авто-создание/рестарт)
uint8_t* last_start_dgram; // копия wire-пакета последнего принятого START (дедуп ретрансмитов)
size_t last_start_dgram_len;
uint8_t send_restart_pending; // 1 = RST принят и отправка переинициализирована (игнор повторных RST до первого ACK)
// Ретрансмиты: inflight очередь
struct ll_queue* inflight_q; // хеш по seq (4 байта), непрерывный блок без SACK