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

feat: 1-byte flags in SVC_ROUTE_HDR — START, RST, 2-bit sess_id

Browse Source 
SVC_ROUTE_HDR: +uint8_t flags (23 bytes total).
  bit7=START (0x80) — first data packet of new session
  bit6=RST (0x40) — sequence violation detected
  bit5-4=sess_id (0-3) — cyclically incremented on restart

router_send_one_flags: sets START+sess_id on first data packet.
router_send_ack: sets sess_id only (no START for ACK).
etcp_router_recv_cb: detects restart via sess_id change or START flag.
  Old seq=0 heuristic replaced with explicit flags.
etcp_router_conn_get: init sess_id=0, peer_sess_id=0, start_sent=0.
etcp_router_conn_restart: increment sess_id before closing rconn.
This commit is contained in:
Evgeny committed 2026-06-03 21:20:05 +03:00
1 parent ca7f6c0d01
commit d0029178e7
2 files changed
+49 -9
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/etcp_router.c
+36 -7
View File
@@ -64,6 +64,14 @@ static int router_send_one_flags(struct ETCP_ROUTER_CONN* rconn, const uint8_t*
hdr->src_node_id = inst->node_id;
hdr->seq = seq;
hdr->svc_id = rconn->svc_id;
{
uint8_t f = (rconn->sess_id << ROUTER_SESS_ID_SHIFT);
if (!rconn->start_sent && pl_len > 0 && !(seq_flags & (ROUTER_SEQ_CLOSE_FLAG | ROUTER_SEQ_RST_FLAG))) {
f |= ROUTER_FLAG_START;
rconn->start_sent = 1;
}
hdr->flags = f;
}
if (pl_len > 0) memcpy(dgram + SVC_ROUTE_HDR_SIZE, payload, pl_len);
struct ll_entry* entry = queue_entry_new(0);
@@ -222,6 +230,7 @@ static void router_send_ack(struct ETCP_ROUTER_CONN* rconn) {
hdr->src_node_id = inst->node_id;
hdr->seq = rconn->rx_seq;
hdr->svc_id = rconn->svc_id;
hdr->flags = (rconn->sess_id << ROUTER_SESS_ID_SHIFT);
struct ll_entry* entry = queue_entry_new(0);
if (!entry) { u_free(hdr); return; }
@@ -368,13 +377,27 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
if (!rconn) { queue_dgram_free(entry); queue_entry_free(entry); return; }
}
// Обнаружение перезапуска peer'a: seq=0 при rx_seq >= MAX_INFLIGHT
if (rconn->rx_seq >= (uint32_t)ROUTER_MAX_INFLIGHT && hdr->seq == 0 && pl_len > 0) {
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router: peer restart detected svc_id=%u from %016llx — resetting",
hdr->svc_id, (unsigned long long)hdr->src_node_id);
etcp_router_conn_restart(inst, hdr->src_node_id, hdr->svc_id);
rconn = etcp_router_conn_get(inst, hdr->src_node_id, hdr->svc_id);
if (!rconn) { queue_dgram_free(entry); queue_entry_free(entry); return; }
// Обнаружение перезапуска peer'а: новый sess_id или START-флаг
{
uint8_t incoming_sess = (hdr->flags >> ROUTER_SESS_ID_SHIFT) & 0x3;
if (pl_len > 0 && (incoming_sess != rconn->peer_sess_id || (hdr->flags & ROUTER_FLAG_START))) {
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE,
"router: peer restart sess_id=%u→%u svc_id=%u from %016llx",
rconn->peer_sess_id, incoming_sess, hdr->svc_id,
(unsigned long long)hdr->src_node_id);
etcp_router_conn_restart(inst, hdr->src_node_id, hdr->svc_id);
rconn = etcp_router_conn_get(inst, hdr->src_node_id, hdr->svc_id);
if (!rconn) { queue_dgram_free(entry); queue_entry_free(entry); return; }
rconn->peer_sess_id = incoming_sess;
}
// RST-флаг — peer сигналит нарушение последовательности
if (hdr->flags & ROUTER_FLAG_RST) {
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE,
"router: RST from %016llx svc_id=%u",
(unsigned long long)hdr->src_node_id, hdr->svc_id);
router_close_and_notify(rconn);
rconn = NULL;
}
}
// Seq-состояние есть — reorder
@@ -589,6 +612,9 @@ void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t remote_node_i
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_restart: svc_id=%u remote=%016llx — notifying service",
svc_id, (unsigned long long)remote_node_id);
rconn->sess_id = (rconn->sess_id + 1) & 0x3;
rconn->start_sent = 0;
etcp_recv_fn cb = inst->router_bindings.callbacks[svc_id];
if (cb) {
struct ll_entry* e = queue_entry_new(0);
@@ -633,6 +659,9 @@ struct ETCP_ROUTER_CONN* etcp_router_conn_get(struct UTUN_INSTANCE* inst,
rconn->idle_ack_timer = NULL;
rconn->send_blocked = 0;
rconn->send_resume_timer = NULL;
rconn->sess_id = 0;
rconn->peer_sess_id = 0;
rconn->start_sent = 0;
rconn->recv_q = queue_new(inst->ua, ROUTER_RECVQ_HASH_SIZE, 0, 4, "router_recv_q");
if (!rconn->recv_q) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router_conn_get: queue_new(recv_q) failed"); queue_entry_free(&rconn->ll); return NULL; }
rconn->send_q = queue_new(inst->ua, 0, 0, 0, "router_send_q");
src/etcp_router.h
+13 -2
View File
@@ -3,7 +3,8 @@
// Упрощённый TCP поверх ETCP: восстановление порядка, дедупликация, без переповторов
//
// Формат SVC_ROUTE пакета:
// [cmd:1] [dst_node_id:8] [src_node_id:8] [seq:4] [svc_id:1] [payload...]
// [cmd:1] [dst_node_id:8] [src_node_id:8] [seq:4] [svc_id:1] [flags:1] [payload...]
// flags: bit7=START, bit6=RST, bit5-4=sess_id(0-3)
// ACK-пакет: тот же заголовок, seq=rx_seq, payload_len=0
#ifndef ETCP_ROUTER_H
#define ETCP_ROUTER_H
@@ -18,11 +19,18 @@ struct SVC_ROUTE_HDR {
uint64_t src_node_id;
uint32_t seq; // data: tx_seq; ACK: rx_seq (ожидаемый seq)
uint8_t svc_id; // идентификатор сервиса
uint8_t flags; // bit7=START, bit6=RST, bit5-4=sess_id
};
#pragma pack(pop)
#define SVC_ROUTE_HDR_SIZE sizeof(struct SVC_ROUTE_HDR) // 22
#define SVC_ROUTE_HDR_SIZE sizeof(struct SVC_ROUTE_HDR) // 23
#define SVC_ROUTE_MAX_BINDINGS 256
// Биты в flags
#define ROUTER_FLAG_START 0x80
#define ROUTER_FLAG_RST 0x40
#define ROUTER_SESS_ID_SHIFT 4
#define ROUTER_SESS_ID_MASK 0x30
// Состояние одного логического подключения (remote_node_id + svc_id)
struct ETCP_ROUTER_CONN {
struct ll_entry ll; // data[0..7]=remote_node_id, data[8]=svc_id — хеш-индекс
@@ -45,6 +53,9 @@ struct ETCP_ROUTER_CONN {
struct ll_queue* send_q; // очередь ожидающих отправки (inflight полон)
void* send_resume_timer; // таймер возобновления отправки
uint8_t send_blocked; // 1 = inflight полон, ждём ack/таймер
uint8_t sess_id; // наш session id (0-3)
uint8_t peer_sess_id; // последний sess_id от peer'а
uint8_t start_sent; // 0 = нужно отправить START в первом data
};
#define ROUTER_CONN_HASH_SIZE 256