Browse Source

Expose per-service payload backlog and acknowledged delivery

master
evgeny 4 days ago
parent
commit
f87c6a93bc
  1. 25
      src/routing_layer/etcp_router.c
  2. 16
      src/routing_layer/etcp_router.h

25
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,

16
src/routing_layer/etcp_router.h

@ -34,6 +34,7 @@ extern "C" {
#include <stdint.h>
#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)

Loading…
Cancel
Save