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

refactor: remove consumer_ack, isolate proxy from router internals 

etcp_router:
  - removed consumer_ack field, consumer_ack function, consumer_ack logic
  - ACK always auto-sent via router_schedule_ack (10ms throttle)
  - conn_restart: local-only reset, no remote CLOSE, unified close notification (len=9)
  - START detection: only when rx_seq>0 (first conn doesn't trigger)
  - new API: etcp_router_on_send_ready / cancel_send_ready for send_q backpressure

tcp_proxy_server:
  - removed write_on_get_cb, queue_set_on_get, rconn->consumer_ack
  - removed etcp_router_conn_get, direct send_q access
  - uses etcp_router_on_send_ready for backpressure

tcp_proxy_client:
  - removed consumer_ack call
  - unified conn=NULL handler (len=9 = close/restart with node_id)

protocol: close/restart notification now len=9 (svc_id + node_id), no RESTART subcmd
etcp-inflight-fix
Evgeny 4 months ago
parent
d784b2bfbe
commit
e0378ebf1d
6 changed files with 91 additions and 101 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. 72
      src/etcp_router.c
  2. 10
      src/etcp_router.h
  3. 45
      src/proxy/tcp_proxy_client.c
  4. 57
      src/proxy/tcp_proxy_server.c
  5. 2
      src/proxy/tcp_proxy_server.h
  6. 6
      tests/test_etcp_router_unit.c

72
src/etcp_router.c
Unescape View File

@ -119,10 +119,11 @@ static void router_send_close_to_service(struct ETCP_ROUTER_CONN* rconn) {
if (!cb) return;
struct ll_entry* e = queue_entry_new(0);
if (!e) return;
e->dgram = u_malloc(1);
e->dgram = u_malloc(9);
if (!e->dgram) { queue_entry_free(e); return; }
e->dgram[0] = rconn->svc_id;
e->len = 1;
memcpy(e->dgram + 1, &rconn->remote_node_id, 8);
e->len = 9;
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_close_svc: svc_id=%u remote=%016llx",
rconn->svc_id, (unsigned long long)rconn->remote_node_id);
cb(NULL, e);
@ -380,7 +381,8 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
// Обнаружение перезапуска 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))) {
if (pl_len > 0 && (incoming_sess != rconn->peer_sess_id ||
((hdr->flags & ROUTER_FLAG_START) && rconn->rx_seq > 0))) {
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,
@ -441,7 +443,7 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
if (seq == rconn->rx_seq)
router_try_assembly(rconn, conn);
if (!rconn->consumer_ack) router_schedule_ack(rconn);
router_schedule_ack(rconn);
queue_dgram_free(entry); queue_entry_free(entry);
} else {
// ========== Транзит ==========
@ -597,40 +599,65 @@ void etcp_router_waiter_cancel(struct UTUN_INSTANCE* inst, uint64_t peer_node_id
queue_waiter_cancel(conn->normalizer->input, h);
}
void etcp_router_consumer_ack(struct UTUN_INSTANCE* inst, uint64_t remote_node_id, uint8_t svc_id) {
struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, remote_node_id, svc_id);
if (!rconn) return;
rconn->consumer_ack = 1;
DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "CONSUMER_ACK: svc_id=%u rx_seq=%u → %016llx",
svc_id, rconn->rx_seq, (unsigned long long)remote_node_id);
router_ack_do_send(rconn);
}
void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t remote_node_id, uint8_t svc_id) {
struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, remote_node_id, svc_id);
if (!rconn) return;
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_restart: svc_id=%u remote=%016llx — notifying service",
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "router_restart: svc_id=%u remote=%016llx — resetting local state",
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);
if (e) {
e->dgram = u_malloc(10);
e->dgram = u_malloc(9);
if (e->dgram) {
e->dgram[0] = svc_id;
e->dgram[1] = 0xFE; // RESTART subcmd
memcpy(e->dgram + 2, &remote_node_id, 8);
e->len = 10;
memcpy(e->dgram + 1, &remote_node_id, 8);
e->len = 9;
cb(NULL, e);
} else { queue_entry_free(e); }
}
}
router_close_and_notify(rconn);
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); }
queue_free(rconn->send_q);
}
if (rconn->recv_q) {
while ((f = queue_data_get(rconn->recv_q)) != NULL) { queue_dgram_free(f); queue_entry_free(f); }
queue_free(rconn->recv_q);
}
rconn->send_q = queue_new(inst->ua, 0, 0, 0, "router_send_q");
rconn->recv_q = queue_new(inst->ua, ROUTER_RECVQ_HASH_SIZE, 0, 4, "router_recv_q");
queue_set_threshold(rconn->send_q, ROUTER_MAX_SEND_Q_PACKETS / 2, 0);
rconn->tx_seq = 0;
rconn->rx_seq = 0;
rconn->tx_acked = 0;
rconn->last_sent_ack_seq = 0;
rconn->sess_id = (rconn->sess_id + 1) & 0x3;
rconn->peer_sess_id = 0;
rconn->start_sent = 0;
rconn->send_blocked = 0;
}
void etcp_router_on_send_ready(struct UTUN_INSTANCE* inst, uint64_t node_id, uint8_t svc_id,
struct queue_waiter_handle* h,
queue_threshold_callback_fn callback, void* arg) {
if (!inst || !h) return;
struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, node_id, svc_id);
if (!rconn || !rconn->send_q) return;
queue_waiter_wait(rconn->send_q, h, callback, arg);
}
void etcp_router_cancel_send_ready(struct UTUN_INSTANCE* inst, uint64_t node_id, uint8_t svc_id,
struct queue_waiter_handle* h) {
if (!inst || !h) return;
struct ETCP_ROUTER_CONN* rconn = router_conn_find(inst, node_id, svc_id);
if (!rconn || !rconn->send_q) return;
queue_waiter_cancel(rconn->send_q, h);
}
// ====================================================================
@ -653,7 +680,6 @@ struct ETCP_ROUTER_CONN* etcp_router_conn_get(struct UTUN_INSTANCE* inst,
rconn->rx_seq = 0;
rconn->tx_acked = 0;
rconn->last_sent_ack_seq = 0;
rconn->consumer_ack = 0;
rconn->last_ack_sent_tb = 0;
rconn->inst = inst;
rconn->ack_timer = NULL;

10
src/etcp_router.h
Unescape View File

@ -42,7 +42,6 @@ struct ETCP_ROUTER_CONN {
uint32_t rx_seq; // ожидаемый seq для сборки (next expected)
uint32_t tx_acked; // сколько наших пакетов подтвердил remote (для inflight)
uint32_t last_sent_ack_seq; // последний отправленный ACK (= rx_seq на момент отправки)
uint8_t consumer_ack; // 1 = ACK отправляется только при потреблении, не при сборке
uint64_t last_ack_sent_tb; // время отправки последнего ACK (timebase 0.1ms)
struct UTUN_INSTANCE* inst;
@ -113,8 +112,13 @@ void etcp_router_waiter_register(struct UTUN_INSTANCE* inst, uint64_t peer_node_
void etcp_router_waiter_cancel(struct UTUN_INSTANCE* inst, uint64_t peer_node_id,
struct queue_waiter_handle* h);
// Уведомить router о потреблении данных потребителем (для consumer-driven ACK)
void etcp_router_consumer_ack(struct UTUN_INSTANCE* inst, uint64_t remote_node_id, uint8_t svc_id);
// Backpressure: зарегистрировать/отменить waiter на send_q очереди
// h — handle из структуры сервиса, callback вызывается когда send_q.count <= threshold (128)
void etcp_router_on_send_ready(struct UTUN_INSTANCE* inst, uint64_t node_id, uint8_t svc_id,
struct queue_waiter_handle* h,
queue_threshold_callback_fn callback, void* arg);
void etcp_router_cancel_send_ready(struct UTUN_INSTANCE* inst, uint64_t node_id, uint8_t svc_id,
struct queue_waiter_handle* h);
// Сбросить состояние роутера для конкретного peer+svc (перезапуск удалённой стороны).
// Очищает send_q/recv_q, уведомляет сервис через cb(NULL, entry) с remote_node_id.

45
src/proxy/tcp_proxy_client.c
Unescape View File

@ -186,7 +186,6 @@ static void tcp_proxy_client_feed_from_transport(struct tcp_proxy_client_conn *p
}
queue_resume_callback(pc->to_lwip);
if (sent_any) {
etcp_router_consumer_ack(pc->proxy->inst, pc->proxy->via_node_id, ETCP_ID_TCP_PROXY);
uint32_t unsent = 0; struct tcp_seg* s;
for (s = pc->pcb->unsent; s; s = s->next) unsent++;
DEBUG_DEBUG(DEBUG_CATEGORY_TRAFFIC, "PROXY FEED exit->client sid=%08x fed=%u q=%u->%u snd_wnd=%u cwnd=%u unsent=%u",
@ -467,33 +466,35 @@ static void tcp_proxy_client_handle_error(struct tcp_proxy_client* p, uint32_t s
// Единый обработчик etcp_router (диспетчеризация в tcp_proxy_client или tcp_proxy_server)
// ====================================================================
void tcp_proxy_client_etcp_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (!entry || !entry->dgram || entry->len < TCP_PROXY_HDR_SIZE) {
if (entry) { DEBUG_WARN(DEBUG_CATEGORY_SOCKET, "TCP proxy client: bad entry len=%u", entry->len); queue_dgram_free(entry); queue_entry_free(entry); }
return;
}
uint8_t subcmd = entry->dgram[1];
uint32_t stream_id; memcpy(&stream_id, entry->dgram + 2, 4);
struct UTUN_INSTANCE* inst = conn ? conn->instance : NULL;
struct tcp_proxy_client* proxy = inst ? inst->tcp_proxy_client : NULL;
if (subcmd == TCP_PROXY_SUBCMD_RESTART) {
uint64_t peer_id = 0;
if (entry->len >= 10) memcpy(&peer_id, entry->dgram + 2, 8);
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "PROXY RESTART from %016llx — clearing all conns for peer", (unsigned long long)peer_id);
if (proxy) {
struct tcp_proxy_client_conn *pc, *next;
for (pc = proxy->conns; pc; pc = next) { next = pc->next; tcp_proxy_client_conn_free(pc); }
proxy->conns = NULL; proxy->conn_count = 0;
}
if (inst && inst->tcp_proxy_server.enabled) {
struct tcp_proxy_server_conn *rc, *next;
for (rc = inst->tcp_proxy_server.conns; rc; rc = next) {
next = rc->next;
if (rc->peer_node_id == peer_id) tcp_proxy_server_conn_free(rc);
if (!entry || !entry->dgram || entry->len < TCP_PROXY_HDR_SIZE) {
// conn==NULL + len==9: close/restart уведомление от роутера
if (entry && entry->len == 9 && !conn && entry->dgram[0] == ETCP_ID_TCP_PROXY) {
uint64_t peer_id;
memcpy(&peer_id, entry->dgram + 1, 8);
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "PROXY CLOSE_ALL from %016llx — clearing all conns for peer",
(unsigned long long)peer_id);
if (proxy) {
struct tcp_proxy_client_conn *pc, *next;
for (pc = proxy->conns; pc; pc = next) { next = pc->next; tcp_proxy_client_conn_free(pc); }
proxy->conns = NULL; proxy->conn_count = 0;
inst = proxy->inst;
}
if (inst && inst->tcp_proxy_server.enabled) {
struct tcp_proxy_server_conn *rc, *next;
for (rc = inst->tcp_proxy_server.conns; rc; rc = next) {
next = rc->next;
if (rc->peer_node_id == peer_id) tcp_proxy_server_conn_free(rc);
}
}
}
queue_dgram_free(entry); queue_entry_free(entry); return;
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
return;
}
uint8_t subcmd = entry->dgram[1];
uint32_t stream_id; memcpy(&stream_id, entry->dgram + 2, 4);
if (subcmd == TCP_PROXY_SUBCMD_CONNECT) {
uint64_t src_node_id = conn ? conn->peer_node_id : (inst ? inst->node_id : 0);

57
src/proxy/tcp_proxy_server.c
Unescape View File

@ -100,23 +100,9 @@ static void send_error(struct tcp_proxy_server_conn* rc) {
}
// ====================================================================
// Коллбэки tcp_io + очередь
// Коллбэки tcp_io
// ====================================================================
static void write_on_get_cb(struct ll_queue* q, void* arg) {
struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL;
if (!inst) return;
rc->ack_batch++;
if (rc->ack_batch >= 8) {
rc->ack_batch = 0;
DEBUG_DEBUG(DEBUG_CATEGORY_SOCKET, "SOCK:ACK fd=%d sid=%08x write_q=%d",
rc->tc ? (int)rc->tc->sock : -1, rc->stream_id,
queue_entry_count(q));
etcp_router_consumer_ack(inst, rc->peer_node_id, ETCP_ID_TCP_PROXY);
}
}
static void on_fin_cb(struct tcp_conn* tc, void* arg) {
struct tcp_proxy_server_conn* rc = (struct tcp_proxy_server_conn*)arg;
if (write_pending(tc))
@ -147,15 +133,6 @@ static void diag_timer_cb(void* arg) {
int ent_f = tc->entry_pool->free_count;
int dat_f = tc->data_pool->free_count;
int sq = 0;
{
struct UTUN_INSTANCE* inst = rc->ctx ? rc->ctx->inst : NULL;
if (inst) {
struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, rc->peer_node_id, ETCP_ID_TCP_PROXY);
if (rconn && rconn->send_q) sq = queue_entry_count(rconn->send_q);
}
}
int rcv_buf = 0, snd_buf = 0;
socklen_t optlen = sizeof(int);
getsockopt(tc->sock, SOL_SOCKET, SO_RCVBUF, &rcv_buf, &optlen);
@ -163,12 +140,11 @@ static void diag_timer_cb(void* arg) {
DEBUG_INFO(DEBUG_CATEGORY_SOCKET,
"SOCK:DIAG fd=%d sid=%08x conn=%d fin=%d "
"rq=%d(%zub) wq=%d(%zub) send_q=%d wbuf=%s "
"rq=%d(%zub) wq=%d(%zub) wbuf=%s "
"ent_f=%d dat_f=%d tcp_rb=%zu tcp_sb=%zu",
(int)tc->sock, rc->stream_id, tc->connected, tc->fin,
tc->read_queue->count, queue_total_bytes(tc->read_queue),
tc->write_queue->count, queue_total_bytes(tc->write_queue),
sq,
tc->write_buf ? "y" : "n",
ent_f, dat_f,
(size_t)rcv_buf, (size_t)snd_buf);
@ -193,13 +169,10 @@ static void read_queue_drain_cb(struct ll_queue* q, void* arg) {
if (ret == 0) {
queue_resume_callback(q);
} else {
struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, rc->peer_node_id, ETCP_ID_TCP_PROXY);
if (rconn && rconn->send_q) {
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:BACKPRESSURE fd=%d sid=%08x send_q=%d — pausing drain",
(int)rc->tc->sock, rc->stream_id, queue_entry_count(rconn->send_q));
queue_waiter_wait(rconn->send_q, &rc->pause_waiter, pause_resume_cb, rc);
} else
etcp_router_waiter_register(inst, rc->peer_node_id, &rc->pause_waiter, pause_resume_cb, rc);
DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "SOCK:BACKPRESSURE fd=%d sid=%08x — pausing drain",
(int)rc->tc->sock, rc->stream_id);
etcp_router_on_send_ready(inst, rc->peer_node_id, ETCP_ID_TCP_PROXY,
&rc->pause_waiter, pause_resume_cb, rc);
}
}
@ -230,15 +203,6 @@ void tcp_proxy_server_conn_free(struct tcp_proxy_server_conn* rc) {
struct tcp_proxy_server_conn** prev = &rc->ctx->conns;
while (*prev) { if (*prev == rc) { *prev = rc->next; rc->ctx->conn_count--; break; } prev = &(*prev)->next; }
}
if (rc->tc) { tcp_conn_destroy(rc->tc); rc->tc = NULL; }
if (rc->close_timer) { uasync_cancel_timeout(rc->ua, rc->close_timer); rc->close_timer = NULL; }
if (rc->diag_timer) { uasync_cancel_timeout(rc->ua, rc->diag_timer); rc->diag_timer = NULL; }
if (rc->ctx && rc->ctx->inst) {
struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(rc->ctx->inst, rc->peer_node_id, ETCP_ID_TCP_PROXY);
if (rconn && rconn->send_q) queue_waiter_cancel(rconn->send_q, &rc->pause_waiter);
else etcp_router_waiter_cancel(rc->ctx->inst, rc->peer_node_id, &rc->pause_waiter);
}
if (rc->tc) queue_set_on_get(rc->tc->write_queue, NULL, NULL);
u_free(rc);
}
@ -291,15 +255,6 @@ int tcp_proxy_server_handle_connect(struct UTUN_INSTANCE* inst, struct ll_entry*
rc->next = ctx->conns; ctx->conns = rc;
rc->diag_timer = uasync_set_timeout(rc->ua, 10000, rc, diag_timer_cb, "tps_diag");
{
struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, src_node_id, ETCP_ID_TCP_PROXY);
if (rconn) {
rconn->consumer_ack = 1;
etcp_router_consumer_ack(inst, src_node_id, ETCP_ID_TCP_PROXY);
}
}
queue_set_on_get(rc->tc->write_queue, write_on_get_cb, rc);
queue_dgram_free(entry); queue_entry_free(entry);
return 0;
}

2
src/proxy/tcp_proxy_server.h
Unescape View File

@ -16,7 +16,6 @@ struct ETCP_CONN;
#define TCP_PROXY_SUBCMD_DATA 0x03
#define TCP_PROXY_SUBCMD_CLOSE 0x04
#define TCP_PROXY_SUBCMD_ERROR 0x05
#define TCP_PROXY_SUBCMD_RESTART 0xFE
#define TCP_PROXY_HDR_SIZE 6 // svc_id(1)+subcmd(1)+stream_id(4)
#define TCP_PROXY_CONNECT_HDR_SIZE 12 // HDR_SIZE + dest_ip(4)+dest_port(2)
@ -36,7 +35,6 @@ struct tcp_proxy_server_conn {
void* close_timer; // таймер повтора CLOSE/ERROR
int close_backoff; // backoff: 50..5000 tb (5ms..500ms)
void* diag_timer; // 1-секундный таймер диагностики
uint8_t ack_batch; // счётчик для throttled consumer_ack (каждые 8 записей)
struct queue_waiter_handle pause_waiter;
struct UASYNC* ua;

6
tests/test_etcp_router_unit.c
Unescape View File

@ -40,6 +40,12 @@ static void test_handler(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
return;
}
// close/restart: len==9 (svc_id + node_id), conn==NULL
if (entry->len == 9 && !conn) {
g_notify_count++;
queue_dgram_free(entry); queue_entry_free(entry);
return;
}
if (entry->dgram[1] != rx.marker) { rx.errors++; } else { rx.delivered++; rx.last_seq = rx.expected_seq; rx.expected_seq++; }
queue_entry_free(entry); queue_dgram_free(entry);
}

Write Preview
Loading…
Cancel
Save