|
|
|
|
@ -29,6 +29,11 @@ in ---> {Q} -> [src, etcp] --> ... --> [dst,etcp] -> {asm_q} -> {buf q} ---> out
|
|
|
|
|
#include "../lib/u_async.h" |
|
|
|
|
#include <string.h> |
|
|
|
|
|
|
|
|
|
/* Текущая диагностика файлового сервиса; остальные сервисы не засоряют debug. */ |
|
|
|
|
#define MF_ROUTE_DEBUG(svc, fmt, ...) do { \ |
|
|
|
|
if ((svc) == ETCP_RT_ID_MEMBER_FILES) DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files route: " fmt, ##__VA_ARGS__); \
|
|
|
|
|
} while (0) |
|
|
|
|
|
|
|
|
|
// ====================================================================
|
|
|
|
|
// Внутренние forward declarations
|
|
|
|
|
// ====================================================================
|
|
|
|
|
@ -356,7 +361,11 @@ static int router_build_packet(struct ETCP_ROUTER_CONN* rconn, struct ll_entry*
|
|
|
|
|
static void router_control(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, uint8_t flags, |
|
|
|
|
uint64_t peer_id, uint64_t challenge) { |
|
|
|
|
if (!conn) conn = router_send_conn(rconn); |
|
|
|
|
if (!conn) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router control: no route svc=%u", rconn->svc_id); return; } |
|
|
|
|
if (!conn) { |
|
|
|
|
MF_ROUTE_DEBUG(rconn->svc_id, "control waiting for route group=%llx peer=%llx flags=%02x", |
|
|
|
|
(unsigned long long)rconn->group_id, (unsigned long long)rconn->remote_node_id, flags); |
|
|
|
|
DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router control: no route svc=%u", rconn->svc_id); return; |
|
|
|
|
} |
|
|
|
|
struct ll_entry* e = ll_alloc_lldgram(SVC_ROUTE_HDR_SIZE); |
|
|
|
|
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router control: allocation failed"); return; } |
|
|
|
|
struct SVC_ROUTE_HDR* h = (struct SVC_ROUTE_HDR*)e->dgram; |
|
|
|
|
@ -365,6 +374,9 @@ static void router_control(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* con
|
|
|
|
|
h->src_node_id = rconn->inst->node_id; h->dst_node_id = rconn->remote_node_id; h->svc_id = rconn->svc_id; |
|
|
|
|
h->flags = flags; h->reset_id = rconn->reset_id; h->peer_reset_id = peer_id; h->challenge = challenge; |
|
|
|
|
e->len = SVC_ROUTE_HDR_SIZE; |
|
|
|
|
MF_ROUTE_DEBUG(rconn->svc_id, "control tx group=%llx peer=%llx via=%llx flags=%02x local=%llx remote=%llx", |
|
|
|
|
(unsigned long long)rconn->group_id, (unsigned long long)rconn->remote_node_id, (unsigned long long)conn->peer_node_id, |
|
|
|
|
flags, (unsigned long long)h->reset_id, (unsigned long long)peer_id); |
|
|
|
|
DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, |
|
|
|
|
"router control: tx group=%016llx peer=%016llx svc=%u flags=%02x local=%016llx remote=%016llx challenge=%016llx", |
|
|
|
|
(unsigned long long)rconn->group_id, (unsigned long long)rconn->remote_node_id, rconn->svc_id, flags, |
|
|
|
|
@ -425,6 +437,9 @@ static void router_confirm_peer(struct ETCP_ROUTER_CONN* rconn, const struct SVC
|
|
|
|
|
if (rconn->handshake_timer) { uasync_cancel_timeout(rconn->inst->ua, rconn->handshake_timer); rconn->handshake_timer = NULL; } |
|
|
|
|
rconn->peer_reset_id = peer_id; |
|
|
|
|
rconn->peer_sync_done = 1; |
|
|
|
|
MF_ROUTE_DEBUG(rconn->svc_id, "session ready group=%llx peer=%llx local=%llx remote=%llx previous=%llx", |
|
|
|
|
(unsigned long long)rconn->group_id, (unsigned long long)rconn->remote_node_id, (unsigned long long)rconn->reset_id, |
|
|
|
|
(unsigned long long)peer_id, (unsigned long long)previous); |
|
|
|
|
DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, |
|
|
|
|
"router session: group=%016llx peer=%016llx svc=%u local=%016llx remote=%016llx previous=%016llx", |
|
|
|
|
(unsigned long long)rconn->group_id, (unsigned long long)rconn->remote_node_id, rconn->svc_id, |
|
|
|
|
@ -581,11 +596,20 @@ static void router_set_no_route(struct ETCP_ROUTER_CONN* rconn) {
|
|
|
|
|
// Инвариант: send_q непуст ∧ !closed ∧ !no_route ∧ inflight свободен ⇒ waiter зарегистрирован.
|
|
|
|
|
static void router_send_kick(struct ETCP_ROUTER_CONN* rconn) { |
|
|
|
|
if (rconn->closed || rconn->no_route) return; |
|
|
|
|
if (!rconn->peer_reset_id) { if (queue_entry_count(rconn->send_q)) router_handshake_start(rconn); return; } |
|
|
|
|
if (!rconn->peer_reset_id) { |
|
|
|
|
if (queue_entry_count(rconn->send_q)) { |
|
|
|
|
MF_ROUTE_DEBUG(rconn->svc_id, "send waiting for session peer=%llx queued=%d", |
|
|
|
|
(unsigned long long)rconn->remote_node_id, queue_entry_count(rconn->send_q)); |
|
|
|
|
router_handshake_start(rconn); |
|
|
|
|
} |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
int retry = rconn->retrans_seq != rconn->retrans_end; |
|
|
|
|
if (!retry && queue_entry_count(rconn->send_q) == 0) return; |
|
|
|
|
if (!retry && queue_entry_count(rconn->inflight_q) >= (int)router_effective_max_inflight(rconn)) { |
|
|
|
|
rconn->send_blocked = 1; |
|
|
|
|
MF_ROUTE_DEBUG(rconn->svc_id, "send blocked inflight=%d limit=%u pending=%d", |
|
|
|
|
queue_entry_count(rconn->inflight_q), router_effective_max_inflight(rconn), queue_entry_count(rconn->send_q)); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
struct ETCP_CONN* conn = router_send_conn(rconn); |
|
|
|
|
@ -652,6 +676,9 @@ static void router_send_drain_cb(struct ll_queue* q, void* arg) {
|
|
|
|
|
free_entry(queue_data_get(rconn->send_q)); |
|
|
|
|
if (rconn->closed || rconn->reset_id != local_id || rconn->peer_reset_id != peer_id) return; |
|
|
|
|
rconn->c_pkts_sent++; |
|
|
|
|
MF_ROUTE_DEBUG(rconn->svc_id, "data tx peer=%llx via=%llx seq=%u bytes=%zu acked=%u inflight=%d pending=%d", |
|
|
|
|
(unsigned long long)rconn->remote_node_id, (unsigned long long)conn->peer_node_id, seq, len, rconn->tx_acked, |
|
|
|
|
queue_entry_count(rconn->inflight_q), queue_entry_count(rconn->send_q)); |
|
|
|
|
rconn->last_dgram_ts = get_current_timestamp(); |
|
|
|
|
if (!rconn->retrans_timer) router_retrans_schedule(rconn); |
|
|
|
|
router_track_inflight_state(rconn); |
|
|
|
|
@ -961,6 +988,7 @@ static void router_send_ack(struct ETCP_ROUTER_CONN* rconn) {
|
|
|
|
|
rconn->c_ack_sent++; |
|
|
|
|
rconn->last_sent_ack_seq = rconn->rx_seq; |
|
|
|
|
rconn->last_ack_sent_tb = get_time_tb(); |
|
|
|
|
MF_ROUTE_DEBUG(rconn->svc_id, "ACK tx peer=%llx next_seq=%u", (unsigned long long)rconn->remote_node_id, rconn->rx_seq); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Запланировать отправку ACK: снять idle-таймер, и либо отправить сразу (интервал истёк),
|
|
|
|
|
@ -1065,6 +1093,8 @@ static int router_deliver(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn
|
|
|
|
|
rconn->svc_id, payload_len, rx_flags, (unsigned long long)rconn->remote_node_id, |
|
|
|
|
(unsigned long long)inst->node_id); |
|
|
|
|
rconn->c_pkts_rcvd++; |
|
|
|
|
MF_ROUTE_DEBUG(rconn->svc_id, "deliver group=%llx peer=%llx seq=%u payload=%zu verified_flags=%02x handler=%p", |
|
|
|
|
(unsigned long long)rconn->group_id, (unsigned long long)rconn->remote_node_id, hdr->seq, payload_len, rx_flags, (void*)cb); |
|
|
|
|
cb(conn, svc_entry); |
|
|
|
|
return 0; |
|
|
|
|
} |
|
|
|
|
@ -1273,8 +1303,11 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
|
|
|
|
|
free_entry(entry); return; |
|
|
|
|
} |
|
|
|
|
struct SVC_ROUTE_HDR* h = (struct SVC_ROUTE_HDR*)entry->dgram; |
|
|
|
|
MF_ROUTE_DEBUG(h->svc_id, "wire rx group=%llx src=%llx dst=%llx via=%llx seq=%u flags=%02x bytes=%u", |
|
|
|
|
(unsigned long long)h->group_id, (unsigned long long)h->src_node_id, (unsigned long long)h->dst_node_id, |
|
|
|
|
(unsigned long long)conn->peer_node_id, h->seq, h->flags, entry->len); |
|
|
|
|
size_t body = router_path_check(conn, entry); |
|
|
|
|
if (!body) { free_entry(entry); return; } |
|
|
|
|
if (!body) { MF_ROUTE_DEBUG(h->svc_id, "wire rejected by path validation"); free_entry(entry); return; } |
|
|
|
|
if (h->dst_node_id != inst->node_id) { |
|
|
|
|
if (router_path_append(entry, inst->node_id, entry->dgram[entry->len - 1]) < 0) { |
|
|
|
|
inst->router_path_errors++; free_entry(entry); return; |
|
|
|
|
@ -1286,6 +1319,8 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
|
|
|
|
|
if (h->flags & (ROUTER_FLAG_ENCRYPTED | ROUTER_FLAG_SIGNED)) { |
|
|
|
|
uint8_t* decoded = NULL; size_t len = 0; uint8_t flags = 0; |
|
|
|
|
if (route_crypto_decode(inst, entry->dgram, entry->len, &decoded, &len, &flags) != 0) { |
|
|
|
|
MF_ROUTE_DEBUG(h->svc_id, "wire rejected by crypto src=%llx seq=%u flags=%02x", |
|
|
|
|
(unsigned long long)h->src_node_id, h->seq, h->flags); |
|
|
|
|
if (rconn) rconn->c_sign_fail++; |
|
|
|
|
free_entry(entry); return; |
|
|
|
|
} |
|
|
|
|
@ -1316,12 +1351,17 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
|
|
|
|
|
} else if (control == (ROUTER_FLAG_START | ROUTER_FLAG_RST)) { |
|
|
|
|
router_confirm_peer(rconn, h); |
|
|
|
|
} else if (h->peer_reset_id != rconn->reset_id || h->reset_id != rconn->peer_reset_id) { |
|
|
|
|
MF_ROUTE_DEBUG(h->svc_id, "epoch rejected peer=%llx seq=%u received=%llx/%llx expected=%llx/%llx", |
|
|
|
|
(unsigned long long)h->src_node_id, h->seq, (unsigned long long)h->reset_id, (unsigned long long)h->peer_reset_id, |
|
|
|
|
(unsigned long long)rconn->peer_reset_id, (unsigned long long)rconn->reset_id); |
|
|
|
|
DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "router epoch mismatch: peer=%016llx svc=%u seq=%u flags=%02x", |
|
|
|
|
(unsigned long long)h->src_node_id, h->svc_id, h->seq, h->flags); |
|
|
|
|
if (payload_len) router_challenge_peer(rconn, conn, h->reset_id); |
|
|
|
|
} else if (control == ROUTER_FLAG_CLOSE) { |
|
|
|
|
router_close_and_notify(rconn); |
|
|
|
|
} else if (!payload_len) { |
|
|
|
|
MF_ROUTE_DEBUG(h->svc_id, "ACK rx peer=%llx next_seq=%u old_acked=%u sent=%u inflight=%d", |
|
|
|
|
(unsigned long long)h->src_node_id, h->seq, rconn->tx_acked, rconn->tx_sent, queue_entry_count(rconn->inflight_q)); |
|
|
|
|
if (!h->challenge) router_handle_ack(inst, rconn, h->seq, h->src_node_id, h->timestamp); |
|
|
|
|
} else { |
|
|
|
|
rconn->last_recv_pkt_ts = h->timestamp; |
|
|
|
|
|