Browse Source

Add instance lifetime ACK windows and exclusive ETCP send queue snapshots

stream-multiview
evgeny 1 week ago
parent
commit
ac89601fc5
  1. 52
      src/transport_layer/etcp_tx_stats.c
  2. 21
      src/transport_layer/etcp_tx_stats.h

52
src/transport_layer/etcp_tx_stats.c

@ -0,0 +1,52 @@
#include "etcp_tx_stats.h"
#include "etcp.h"
#include "pkt_normalizer.h"
#include "../utun_instance.h"
#include "../../lib/debug_config.h"
void etcp_tx_record_ack(struct etcp_tx_history *h, uint64_t now, uint64_t bytes) {
if (!h) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "TX ACK history missing"); return; }
uint64_t tick = now / 100000;
struct etcp_ack_bucket *b = &h->buckets[tick % 11];
if (b->tick_plus_one != tick + 1) { b->tick_plus_one = tick + 1; b->bytes = 0; }
if (bytes > UINT64_MAX - b->bytes || bytes > UINT64_MAX - h->total) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "TX ACK counter overflow bytes=%llu", (unsigned long long)bytes); return;
}
b->bytes += bytes; h->total += bytes;
}
uint64_t etcp_tx_acked_window(const struct etcp_tx_history *h, uint64_t now) {
if (!h) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "TX ACK window missing"); return 0; }
uint64_t end = now / 100000, start = end > 10 ? end - 10 : 0, bytes = 0;
for (unsigned i = 0; i < 11; i++) {
const struct etcp_ack_bucket *b = &h->buckets[i];
if (b->tick_plus_one && b->tick_plus_one - 1 >= start && b->tick_plus_one - 1 < end) bytes += b->bytes;
}
return bytes;
}
/* input_send_q уже входит в unacked_bytes. В normalizer считаем только ещё не переданный остаток. */
uint64_t etcp_tx_queued(const struct ETCP_CONN *c) {
if (!c) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "TX queue snapshot missing connection"); return 0; }
uint64_t bytes = c->send_input_q ? c->send_input_q->total_bytes : 0;
if (c->input_queue) bytes += c->input_queue->total_bytes;
if (c->normalizer) {
const struct PKTNORM *p = c->normalizer;
bytes += p->data_ptr;
if (p->pending && p->pending->len >= p->pending_in_ptr) bytes += p->pending->len - p->pending_in_ptr;
}
return bytes;
}
int etcp_tx_get_snapshot(struct UTUN_INSTANCE *inst, uint64_t now, struct etcp_tx_snapshot *out) {
if (!inst || !out) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "TX snapshot invalid arguments"); return -1; }
struct etcp_tx_snapshot s = {.acked_1s = etcp_tx_acked_window(&inst->tx_history, now),
.window_end_us = now / 100000 * 100000};
/* Каждый ETCP_CONN, включая TCP-only/pending, зарегистрирован здесь ровно один раз при create. */
for (struct ll_entry *e = inst->connections ? inst->connections->head : NULL; e; e = e->next) {
struct ETCP_CONN *c = ((struct conn_queue_entry *)e->data)->conn;
if (!c || c->close_requested) continue;
s.connections++; s.queued_bytes += etcp_tx_queued(c); s.inflight_bytes += c->unacked_bytes;
}
*out = s; return 0;
}

21
src/transport_layer/etcp_tx_stats.h

@ -0,0 +1,21 @@
/* Секундное окно ACK всего ядра: десять завершённых интервалов и один текущий, шаг 100ms.
* Владелец UTUN_INSTANCE; все операции в uasync. Reset/close отдельных соединений историю не меняют. */
#ifndef ETCP_TX_STATS_H
#define ETCP_TX_STATS_H
#include <stdint.h>
#ifdef __cplusplus
extern "C" {
#endif
struct UTUN_INSTANCE;
struct ETCP_CONN;
struct etcp_ack_bucket { uint64_t tick_plus_one, bytes; };
struct etcp_tx_history { struct etcp_ack_bucket buckets[11]; uint64_t total; };
struct etcp_tx_snapshot { uint64_t queued_bytes, inflight_bytes, acked_1s, window_end_us; unsigned connections; };
void etcp_tx_record_ack(struct etcp_tx_history *history, uint64_t now_us, uint64_t bytes);
uint64_t etcp_tx_acked_window(const struct etcp_tx_history *history, uint64_t now_us);
uint64_t etcp_tx_queued(const struct ETCP_CONN *conn);
int etcp_tx_get_snapshot(struct UTUN_INSTANCE *inst, uint64_t now_us, struct etcp_tx_snapshot *snapshot);
#ifdef __cplusplus
}
#endif
#endif
Loading…
Cancel
Save