diff --git a/src/routing_layer/etcp_router.c b/src/routing_layer/etcp_router.c index 2c101dc6..69b091af 100644 --- a/src/routing_layer/etcp_router.c +++ b/src/routing_layer/etcp_router.c @@ -564,6 +564,7 @@ static int router_conn_reset(struct ETCP_ROUTER_CONN* rconn) { rconn->c_ack_sent = 0; rconn->c_ack_recv = 0; rconn->c_retrans_done = 0; rconn->c_dup_dropped = 0; rconn->c_oob_dropped = 0; rconn->c_stale_ack = 0; rconn->c_sign_fail = 0; rconn->last_ack_changed_tb = 0; + route_tx_meter_reset(&rconn->delivery, get_time_us()); return 0; } @@ -666,7 +667,7 @@ static void router_send_drain_cb(struct ll_queue* q, void* arg) { free_entry(e); u_free(inf->dgram); u_free(inf); return; } memcpy(e->dgram, inf->dgram, len); e->len = len; - memcpy(&inf->seq, pending->data, 4); inf->ll.size = 4; + memcpy(&inf->seq, pending->data, 4); inf->ll.size = 4; inf->ll.len = pending->len; inf->dgram_len = len; inf->last_sent_tb = get_time_tb(); inf->send_count = 1; uint32_t seq = inf->seq; queue_data_put_with_index(rconn->inflight_q, &inf->ll); @@ -1210,6 +1211,7 @@ static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CON while (rconn->inflight_q->head) { struct ROUTER_INFLIGHT* inf = (struct ROUTER_INFLIGHT*)rconn->inflight_q->head; if ((int32_t)(inf->seq - seq) >= 0) break; + route_tx_meter_ack(&rconn->delivery, get_time_us(), inf->ll.len); queue_remove_data(rconn->inflight_q, &inf->ll); u_free(inf->dgram); u_free(inf); } @@ -1548,6 +1550,27 @@ void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t group_id, uin if (!rconn->closed) router_handshake_start(rconn); } +/* Снимок логической доставки: физические копии inflight второй раз не учитываются. */ +int etcp_router_tx_stats(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t peer, uint8_t svc, + uint64_t now_us, struct ETCP_ROUTER_TX_STATS *out) { + if (!inst || !out) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router TX snapshot invalid arguments"); return -1; } + struct ETCP_ROUTER_CONN *rconn = router_conn_find(inst, group, peer, svc); + if (!rconn || rconn->closed) { memset(out, 0, sizeof(*out)); return 1; } + struct ETCP_ROUTER_TX_STATS result = { + .local_epoch = rconn->reset_id, .peer_epoch = rconn->peer_reset_id, + .queued_bytes = rconn->send_q->total_bytes, .inflight_bytes = rconn->inflight_q->total_bytes, + .queued_packets = rconn->send_q->count, .inflight_packets = rconn->inflight_q->count, + .acked_bytes = rconn->delivery.bytes, .acked_packets = rconn->delivery.packets, + .rtt_us = (uint32_t)rconn->rtt * 100, + .baseline_rtt_us = rconn->minrtt_window_count ? (uint32_t)rconn->minrtt * 100 : 0, + .ready = rconn->peer_reset_id != 0, .no_route = rconn->no_route, + .receive_busy = rconn->peer_receive_blocked, .probing = rconn->minrtt_probe + }; + if (route_tx_meter_window(&rconn->delivery, now_us, &result.window) < 0) return -1; + *out = result; + return 0; +} + // Backpressure: зарегистрировать waiter на send_q для (group_id, node_id, svc_id). void etcp_router_on_send_ready(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id, uint8_t svc_id, struct queue_waiter_handle* h, diff --git a/src/routing_layer/etcp_router.h b/src/routing_layer/etcp_router.h index 21671fdd..299b4111 100644 --- a/src/routing_layer/etcp_router.h +++ b/src/routing_layer/etcp_router.h @@ -34,6 +34,7 @@ extern "C" { #include #include "etcp_api.h" +#include "route_tx_stats.h" // ==================================================================== // 1. Протокол SVC_ROUTE (wire-формат и формат доставки в сервис) @@ -179,8 +180,23 @@ struct ETCP_ROUTER_CONN { uint32_t c_oob_dropped; // дропнуто out-of-bounds seq uint32_t c_stale_ack; // устаревших ACK uint32_t c_sign_fail; // signature verification failures + struct route_tx_meter delivery; // первые подтверждения исходного payload, без wire overhead }; +/* Заимствований нет. queued/inflight — исходный payload, а не размер криптограммы. + * ACK подтверждает доставку callback сервиса; декодирование/показ сервис отслеживает отдельно. */ +struct ETCP_ROUTER_TX_STATS { + uint64_t local_epoch, peer_epoch, queued_bytes, inflight_bytes, acked_bytes, acked_packets; + uint32_t queued_packets, inflight_packets; + struct route_tx_window window; + uint32_t rtt_us, baseline_rtt_us; + int ready, no_route, receive_busy, probing; +}; + +/* Только uasync. 0 — снимок, 1 — канал отсутствует/закрыт, -1 — неверные аргументы. Канал не создаёт. */ +int etcp_router_tx_stats(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t peer, uint8_t svc, + uint64_t now_us, struct ETCP_ROUTER_TX_STATS *out); + #define ROUTER_CONN_HASH_SIZE 256 #define ROUTER_RECVQ_HASH_SIZE 1024 #define ROUTER_MAX_INFLIGHT 256 // макс пакетов в полёте (для контроля inflight)