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

Capture buffered multipath queue and pacing diagnostics

Browse Source
This commit is contained in:
evgeny committed 2026-10-24 14:43:03 +02:00
1 parent a9bbb470c2
commit 12b0ab437e
1 file changed
+263 -3
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_etcp_multipath.c
+263 -3
View File
@@ -2,9 +2,11 @@
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <errno.h>
#include "etcp.h"
#include "etcp_api.h"
#include "etcp_session.h"
#include "etcp_loadbalancer.h"
#include "node_conn_direct.h"
#include "topo_node.h"
#include "utun_instance.h"
@@ -13,7 +15,13 @@
#include "test_clock.h"
#include "../lib/mem.h"
#define CHECK(x) do { if (!(x)) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "FAIL %s:%d: %s", __func__, __LINE__, #x); exit(1); } } while (0)
static void diagnostic_flush(void);
static int diagnostic_flushing;
#define CHECK(x) do { if (!(x)) { \
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "FAIL %s:%d: %s", __func__, __LINE__, #x); \
if (!diagnostic_flushing) diagnostic_flush(); \
exit(1); \
} } while (0)
#define PATHS 3
#define TRACE_COUNT 131072
#define CMD 253
@@ -78,11 +86,181 @@ static const char* privkeys[2] = {
"38240cb82199e504686507f11f6eaa4f740fde6f0c425c495e49a523019a5d68"
};
/* Диагностика включается UTUN_MULTIPATH_DIAG=path.csv. Срезы и счётчики пишутся в память,
* файл/лог — только после drain каждого стенда. Пики наблюдаются на TX/RX/ACK/выборе пути.
* Интервал по умолчанию 100мс; UTUN_MULTIPATH_DIAG_MS=10..1000 позволяет уточнить короткий участок. */
struct diagnostic_interval {
uint64_t tx_data, fresh_rx, data_packets, retries, ack_only_packets, ack_only_wire, piggy_packets, piggy_bytes;
uint64_t ack_entries, ack_delay_sum, ack_delay_max, ack_only_entries, ack_only_delay_sum, early_ack_entries;
uint64_t control_packets, samples, limited, app_limited;
uint64_t rtt_min, rtt_max, rtt_sum, rtt_samples, searches, window_blocked, pacer_blocked, observations;
uint32_t flight_min, flight_max, cwnd_min, cwnd_max, send_max, ack_max;
int64_t debt_min_us, debt_max_us;
};
struct diagnostic_sample {
const char* phase;
uint64_t time_us, interval_us, wire_bytes, channel_received, channel_lost, channel_dropped, app_bytes, ack_age_us, attempt_age_us;
uint64_t pacer_ready_us;
uint64_t queue_bytes[5];
unsigned queue_count[5], side, path, fresh, retries, pending_bytes, normalizer_bytes, normalizer_waiter, normalizer_flush;
unsigned unacked, budget, flight, attempts, cwnd, pace, mode, cycle, min_rtt, rtt, jitter, rto, bw, queue_cap;
unsigned ack_count, ack_timer, ack_till, urgent_ack, pacer_timer, burst, stalled, rto_streak;
unsigned channel_pending, channel_peak;
uint64_t channel_delay_max_us;
int64_t debt_us;
struct diagnostic_interval stats;
};
static FILE* diagnostic_file;
static struct diagnostic_sample diagnostic_rows[16384]; // хватает и для трёх путей/10мс с полным drain
static struct diagnostic_interval diagnostic_stats[2][PATHS];
static struct dummynet_stats diagnostic_previous[2][PATHS];
static unsigned diagnostic_count, diagnostic_active, diagnostic_period_us = 100000;
static uint64_t diagnostic_last_us;
/* Состояние pacer в микросекундах: отрицательное значение — накопленный кредит, положительное — долг. */
static int64_t diagnostic_debt(const struct ETCP_LINK* link) {
return ((int64_t)link->shaper_load_time_tb - (int64_t)get_time_tb()) * 100 + (int64_t)(link->shaper_sub_nanotime / 10000);
}
static void diagnostic_observe(struct diagnostic_interval* d, struct ETCP_LINK* link) {
uint32_t flight = link->inflight_bytes, cwnd = link->inflight_lim_bytes;
int64_t debt = diagnostic_debt(link);
if (!d->observations || flight < d->flight_min) d->flight_min = flight;
if (!d->observations || cwnd < d->cwnd_min) d->cwnd_min = cwnd;
if (!d->observations || debt < d->debt_min_us) d->debt_min_us = debt;
if (!d->observations || debt > d->debt_max_us) d->debt_max_us = debt;
if (flight > d->flight_max) d->flight_max = flight;
if (cwnd > d->cwnd_max) d->cwnd_max = cwnd;
if ((unsigned)link->etcp->input_send_q->count > d->send_max) d->send_max = link->etcp->input_send_q->count;
if ((unsigned)link->ack_q->count > d->ack_max) d->ack_max = link->ack_q->count;
d->observations++;
}
static struct diagnostic_interval* diagnostic_for(struct ETCP_LINK* link) {
if (!diagnostic_active) return NULL;
for (unsigned p = 0; p < path_count; p++)
if (ss_get_port(&link->remote_addr) == dummynet_get_listen_port(channels[p])) {
struct diagnostic_interval* d = &diagnostic_stats[link->etcp->instance == ep[1].inst][p];
diagnostic_observe(d, link); return d;
}
return NULL;
}
/* Общие очереди соединения повторяются в строках его линков; интервальные счётчики относятся к одному пути. */
static void diagnostic_snapshot(const char* phase) {
if (!diagnostic_active) return;
uint64_t now = get_time_us();
if (now == diagnostic_last_us) return;
for (unsigned side = 0; side < 2; side++) for (struct ETCP_LINK* l = ep[side].conn->links; l; l = l->next) {
struct diagnostic_interval* d = diagnostic_for(l);
if (!d) continue;
unsigned p = (unsigned)(d - diagnostic_stats[side]);
CHECK(diagnostic_count < sizeof(diagnostic_rows) / sizeof(diagnostic_rows[0]));
struct diagnostic_sample* s = &diagnostic_rows[diagnostic_count++];
memset(s, 0, sizeof(*s));
struct ETCP_CONN* c = l->etcp;
struct PKTNORM* pn = c->normalizer;
const struct ll_queue* queues[] = {c->send_input_q, c->input_queue, c->input_send_q, c->input_wait_ack, c->recv_q};
s->phase = phase; s->time_us = now - started; s->interval_us = now - diagnostic_last_us; s->side = side; s->path = p;
for (unsigned q = 0; q < 5; q++) { s->queue_count[q] = queues[q]->count; s->queue_bytes[q] = queues[q]->total_bytes; }
for (struct ll_entry* e = c->input_send_q->head; e; e = e->next)
if (((struct INFLIGHT_PACKET*)e)->last_attempt_id) s->retries++; else s->fresh++;
s->normalizer_bytes = pn->data_ptr; s->normalizer_waiter = !!pn->input_waiter_handle.internal;
s->normalizer_flush = !!pn->flush_timer; s->pending_bytes = pn->pending ? pn->pending->len - pn->pending_in_ptr : 0;
s->unacked = c->unacked_bytes; s->budget = c->optimal_inflight;
s->flight = l->inflight_bytes; s->attempts = l->inflight_packets; s->cwnd = l->inflight_lim_bytes; s->pace = l->bbr_pacing_rate;
s->mode = l->bbr->mode; s->cycle = l->bbr->cycle_idx; s->min_rtt = l->bbr->min_rtt_us;
s->rtt = l->data_rtt_us; s->jitter = l->data_jitter_us; s->rto = etcp_tx_rto_us(l);
s->bw = l->bbr->bw_hi[0] > l->bbr->bw_hi[1] ? l->bbr->bw_hi[0] : l->bbr->bw_hi[1];
if (s->bw > l->bbr->bw_lo) s->bw = l->bbr->bw_lo;
s->queue_cap = l->bbr->queue_cwnd_cap; s->ack_count = l->ack_q->count;
s->ack_timer = !!c->ack_resp_timer; s->ack_till = c->ack_till_pending; s->urgent_ack = l->ack_reply_pending;
if (l->ack_q->head) s->ack_age_us = now - ((struct ACK_PACKET*)l->ack_q->head)->received_at_us;
if (l->attempts->head && ((struct ETCP_TX_ATTEMPT*)l->attempts->head)->sent_at_us)
s->attempt_age_us = now - ((struct ETCP_TX_ATTEMPT*)l->attempts->head)->sent_at_us;
s->debt_us = diagnostic_debt(l); s->pacer_timer = !!l->shaper_timer; s->burst = l->burst_active;
uint64_t tb = get_time_tb();
s->pacer_ready_us = l->shaper_load_time_tb >= tb + SHAPER_BURST_DELAY_TB ?
(l->shaper_load_time_tb - tb - SHAPER_BURST_DELAY_TB + 1) * 100 : 0;
s->stalled = l->tx_stalled; s->rto_streak = l->data_rto_streak; s->app_bytes = ep[1-side].bytes;
const struct dummynet_stats* ds = dummynet_get_stats(channels[p], side);
struct dummynet_stats* previous = &diagnostic_previous[side][p];
CHECK(ds->sent_bytes >= previous->sent_bytes && ds->recv_bytes >= previous->recv_bytes);
s->wire_bytes = ds->sent_bytes - previous->sent_bytes; s->channel_received = ds->recv_bytes - previous->recv_bytes;
s->channel_lost = ds->lost - previous->lost; s->channel_dropped = ds->dropped - previous->dropped;
s->channel_pending = ds->queue_size; s->channel_peak = ds->queue_max; s->channel_delay_max_us = ds->max_queue_delay_us;
*previous = *ds; s->stats = *d; memset(d, 0, sizeof(*d));
}
diagnostic_last_us = now;
}
/* Вывод после замера: dt_us обязателен для скоростей, особенно в коротком последнем срезе фазы. */
static void diagnostic_flush(void) {
if (!diagnostic_file) return;
diagnostic_flushing = 1;
for (unsigned n = 0; n < diagnostic_count; n++) {
const struct diagnostic_sample* s = &diagnostic_rows[n];
const struct diagnostic_interval* d = &s->stats;
CHECK(fprintf(diagnostic_file, "%s,%x,%s,%llu,%llu,%u,%u", case_name, enabled_mask, s->phase,
(unsigned long long)s->time_us, (unsigned long long)s->interval_us, s->side, s->path) > 0);
for (unsigned q = 0; q < 5; q++)
CHECK(fprintf(diagnostic_file, ",%u,%llu", s->queue_count[q], (unsigned long long)s->queue_bytes[q]) > 0);
CHECK(fprintf(diagnostic_file, ",%u,%u,%u,%u,%u,%u,%u,%u,%u,%u,%u,%u,%u,%u,%u,%u,%u,%u,%u,%u,%u,%u",
s->fresh, s->retries, s->normalizer_bytes, s->pending_bytes, s->normalizer_waiter, s->normalizer_flush,
s->unacked, s->budget, s->flight, s->attempts, s->cwnd, s->pace, s->mode, s->cycle, s->min_rtt,
s->rtt, s->jitter, s->rto, s->bw, s->queue_cap, s->ack_count, s->ack_timer) > 0);
CHECK(fprintf(diagnostic_file, ",%u,%u,%llu,%llu,%lld,%u,%llu,%u,%u,%u,%u,%u,%llu,%llu,%llu,%llu,%llu,%llu",
s->ack_till, s->urgent_ack, (unsigned long long)s->ack_age_us, (unsigned long long)s->attempt_age_us,
(long long)s->debt_us, s->pacer_timer, (unsigned long long)s->pacer_ready_us, s->burst, s->stalled, s->rto_streak,
s->channel_pending, s->channel_peak, (unsigned long long)s->channel_delay_max_us,
(unsigned long long)s->wire_bytes, (unsigned long long)s->channel_received, (unsigned long long)s->channel_lost,
(unsigned long long)s->channel_dropped, (unsigned long long)s->app_bytes) > 0);
CHECK(fprintf(diagnostic_file, ",%llu,%llu,%llu,%llu,%llu,%llu,%llu,%llu,%llu,%llu,%llu,%llu,%llu,%llu,%llu,%llu,%llu,%llu",
(unsigned long long)d->tx_data, (unsigned long long)d->fresh_rx, (unsigned long long)d->data_packets,
(unsigned long long)d->retries, (unsigned long long)d->ack_only_packets, (unsigned long long)d->ack_only_wire,
(unsigned long long)d->piggy_packets, (unsigned long long)d->piggy_bytes, (unsigned long long)d->ack_entries,
(unsigned long long)d->ack_delay_sum, (unsigned long long)d->ack_delay_max, (unsigned long long)d->early_ack_entries,
(unsigned long long)d->control_packets, (unsigned long long)d->samples, (unsigned long long)d->limited,
(unsigned long long)d->app_limited, (unsigned long long)d->rtt_min, (unsigned long long)d->rtt_max) > 0);
CHECK(fprintf(diagnostic_file, ",%llu,%llu,%llu,%llu,%llu,%llu,%llu,%llu,%u,%u,%u,%u,%u,%u,%lld,%lld\n",
(unsigned long long)d->ack_only_entries, (unsigned long long)d->ack_only_delay_sum,
(unsigned long long)d->rtt_sum, (unsigned long long)d->rtt_samples,
(unsigned long long)d->searches, (unsigned long long)d->window_blocked,
(unsigned long long)d->pacer_blocked, (unsigned long long)d->observations, d->flight_min, d->flight_max,
d->cwnd_min, d->cwnd_max, d->send_max, d->ack_max, (long long)d->debt_min_us, (long long)d->debt_max_us) > 0);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "MPATH %s mask=%x %s t_ms=%llu dt_us=%llu side=%u path=%u "
"Q_app=%u Q_fragment=%u Q_send=%u(new=%u retry=%u) Q_ACK=%u age_us=%llu flight=%u/%u "
"pace_Bps=%u debt_us=%lld ready_us=%llu timer=%u window_wait=%llu pacer_wait=%llu "
"channel_pending=%u mode=%u/%u DATA_B=%llu ACKonly=%llu early_ACK=%llu",
case_name, enabled_mask, s->phase, (unsigned long long)(s->time_us/1000), (unsigned long long)s->interval_us,
s->side, s->path, s->queue_count[0], s->queue_count[1], s->queue_count[2], s->fresh, s->retries,
s->ack_count, (unsigned long long)s->ack_age_us, s->flight, s->cwnd, s->pace, (long long)s->debt_us,
(unsigned long long)s->pacer_ready_us, s->pacer_timer,
(unsigned long long)d->window_blocked, (unsigned long long)d->pacer_blocked,
s->channel_pending, s->mode, s->cycle, (unsigned long long)d->tx_data,
(unsigned long long)d->ack_only_packets, (unsigned long long)d->early_ack_entries);
}
CHECK(fflush(diagnostic_file) == 0);
diagnostic_count = 0; diagnostic_flushing = 0;
}
/* Выборочная диагностика модели: один sample на линк/100мс, без packet-level спама всего ETCP. */
void __real_bbr_main(struct bbr*, const struct bbr_rate_sample*, uint32_t*, uint32_t*, uint32_t, uint32_t, int);
void __wrap_bbr_main(struct bbr* model, const struct bbr_rate_sample* sample, uint32_t* cwnd, uint32_t* pacing,
uint32_t mss, uint32_t inflight, int limited) {
__real_bbr_main(model, sample, cwnd, pacing, mss, inflight, limited);
if (diagnostic_active) for (int side = 0; side < 2; side++)
for (struct ETCP_LINK* l = ep[side].conn->links; l; l = l->next) {
if (l->bbr != model) continue;
struct diagnostic_interval* d = diagnostic_for(l);
if (!d) continue;
d->samples++; d->limited += !!limited; d->app_limited += !!sample->is_app_limited;
if (sample->rtt_us) {
if (!d->rtt_min || sample->rtt_us < d->rtt_min) d->rtt_min = sample->rtt_us;
if (sample->rtt_us > d->rtt_max) d->rtt_max = sample->rtt_us;
d->rtt_sum += sample->rtt_us; d->rtt_samples++;
}
}
if (!log_bbr_samples) return;
uint64_t now = get_time_us();
for (int side = 0; side < 2; side++) for (struct ETCP_LINK* l = ep[side].conn ? ep[side].conn->links : NULL; l; l = l->next) {
@@ -112,7 +290,30 @@ static void observe_packet(struct ETCP_DGRAM* pkt, struct ETCP_LINK* link, int s
const uint8_t* data = pkt->data + offset;
unsigned remaining = pkt->data_len - offset, len;
switch (data[0]) {
case ETCP_SECTION_ACK: CHECK(remaining >= ETCP_ACK_BASE_SIZE); len = etcp_ack_section_size(data[1]); break;
case ETCP_SECTION_ACK: {
CHECK(remaining >= ETCP_ACK_BASE_SIZE); len = etcp_ack_section_size(data[1]); CHECK(len <= remaining);
struct diagnostic_interval* d = sending ? diagnostic_for(link) : NULL;
if (d) {
int ack_only = !pkt->payload_len && len == pkt->data_len;
if (pkt->payload_len && data[1]) { d->piggy_packets++; d->piggy_bytes += len; }
else if (ack_only) {
d->ack_only_packets++;
d->ack_only_wire += len + ETCP_ENCRYPTED_HDR_SIZE + ETCP_SESSION_HEADER_SIZE +
SC_NONCE_SIZE + SC_CRC32_SIZE + SC_TAG_SIZE;
} else if (!pkt->payload_len) d->control_packets++;
for (unsigned n = 0; n < data[1]; n++) {
const uint8_t* a = data + ETCP_ACK_BASE_SIZE + n * ETCP_ACK_ENTRY_SIZE;
uint32_t delay = (uint32_t)a[6] | ((uint32_t)a[7] << 8) | ((uint32_t)a[8] << 16) | ((uint32_t)a[9] << 24);
d->ack_entries++; d->ack_delay_sum += delay;
if (delay > d->ack_delay_max) d->ack_delay_max = delay;
if (ack_only) {
d->ack_only_entries++; d->ack_only_delay_sum += delay;
if (!link->ack_reply_pending && delay < 2000) d->early_ack_entries++;
}
}
}
break;
}
case ETCP_SECTION_ACK_REQUEST: len = 1; break;
case ETCP_SECTION_TIMESTAMP: len = 5; break;
case ETCP_SECTION_MEAS_TS: len = 9; break;
@@ -127,6 +328,11 @@ static void observe_packet(struct ETCP_DGRAM* pkt, struct ETCP_LINK* link, int s
uint16_t number = (uint16_t)data[5] | ((uint16_t)data[6] << 8);
if (!sending) rx_trigger[side] = seq;
if (sending) {
struct diagnostic_interval* d = diagnostic_for(link);
if (d) {
d->tx_data += remaining - ETCP_PAYLOAD_HEADER_SIZE; d->data_packets++;
if (trace->attempts) d->retries++;
}
if (measuring && fault_start && !side && trace->retry_ready) {
uint64_t wait = now - trace->retry_ready; recovery.retry_count++;
if (wait > recovery.ready_wait_us) { recovery.ready_wait_us = wait; recovery.ready_seq = seq; }
@@ -139,6 +345,7 @@ static void observe_packet(struct ETCP_DGRAM* pkt, struct ETCP_LINK* link, int s
fault_rto_us[path] = etcp_tx_rto_us(link);
if (measuring) traffic[side][path].data += remaining - ETCP_PAYLOAD_HEADER_SIZE;
} else if (!trace->first_rx) {
if (diagnostic_active) diagnostic_stats[1-side][path].fresh_rx += remaining - ETCP_PAYLOAD_HEADER_SIZE;
trace->first_rx = now;
if (measuring) traffic[1-side][path].rx_unique += remaining - ETCP_PAYLOAD_HEADER_SIZE;
if (measuring && fault_start && side && trace->last_number == number && trace->last_path == path) {
@@ -198,6 +405,13 @@ int __wrap_queue_data_put_with_index(struct ll_queue* q, struct ll_entry* entry)
struct ETCP_LINK* __real_etcp_loadbalancer_select_link(struct ETCP_CONN*, uint32_t, const struct INFLIGHT_PACKET*);
struct ETCP_LINK* __wrap_etcp_loadbalancer_select_link(struct ETCP_CONN* conn, uint32_t bytes, const struct INFLIGHT_PACKET* retry) {
struct ETCP_LINK* selected = __real_etcp_loadbalancer_select_link(conn, bytes, retry);
if (bytes && diagnostic_active) for (struct ETCP_LINK* l = conn->links; l; l = l->next) {
struct diagnostic_interval* d = diagnostic_for(l);
if (!d) continue;
d->searches++;
d->window_blocked += bytes > l->inflight_lim_bytes || l->inflight_bytes > l->inflight_lim_bytes - bytes;
d->pacer_blocked += l->shaper_timer != NULL;
}
if (!measuring || !fault_start || conn != ep[0].conn || !retry || !selected || !retry->last_link) return selected;
int previous = -1, chosen = -1;
for (unsigned p = 0; p < path_count; p++) {
@@ -312,6 +526,10 @@ static void setup(const struct profile* settings, unsigned count, unsigned mask)
CHECK(node_conn_direct_open_node(ep[1].inst, peer.node_id, NULL, NULL, &ep[1].owner, &peer, NULL) >= 0);
started = get_time_us();
for (int i = 0; i < 2; i++) ep[i].last_delivery = started;
memset(diagnostic_stats, 0, sizeof(diagnostic_stats)); memset(diagnostic_previous, 0, sizeof(diagnostic_previous));
diagnostic_count = 0; diagnostic_last_us = started; diagnostic_active = diagnostic_file != NULL;
if (diagnostic_active) for (unsigned side = 0; side < 2; side++) for (unsigned p = 0; p < path_count; p++)
if (channels[p]) diagnostic_previous[side][p] = *dummynet_get_stats(channels[p], side);
}
/* Ограниченная очередь источника: latency не скрывается за неограниченным накоплением тестовых сообщений. */
@@ -331,6 +549,7 @@ static void produce(unsigned side) {
}
static void snapshot(const char* phase) {
diagnostic_snapshot(phase);
uint64_t now = get_time_us();
for (int side = 0; side < 2; side++) {
struct ETCP_CONN* conn = ep[side].conn;
@@ -361,10 +580,13 @@ static void run_for(unsigned ms, int sending, const char* phase) {
if (sending) { produce(0); if (bidirectional) produce(1); }
test_clock_poll(ua, 10);
for (int i = 0; i < 2; i++) CHECK(ep[i].conn->state != 2 && !ep[i].conn->reinit_pending);
if (get_time_us() >= next_snapshot) { snapshot(phase); next_snapshot = get_time_us() + 100000; }
if (get_time_us() >= next_snapshot) {
snapshot(phase); next_snapshot = get_time_us() + (diagnostic_file ? diagnostic_period_us : 100000);
}
if (!sending && ep[0].sent == ep[1].received && ep[1].sent == ep[0].received &&
!ep[0].conn->unacked_bytes && !ep[1].conn->unacked_bytes) break;
}
diagnostic_snapshot(phase);
}
static void begin_measure(void) {
@@ -375,6 +597,7 @@ static void begin_measure(void) {
memset(ep[i].latency_hist, 0, sizeof(ep[i].latency_hist));
}
for (unsigned p = 0; p < path_count; p++) if (channels[p]) dummynet_reset_stats(channels[p], -1);
memset(diagnostic_previous, 0, sizeof(diagnostic_previous));
}
static void expect(int passed, const char* requirement, uint64_t actual, uint64_t limit) {
@@ -396,6 +619,7 @@ static uint64_t percentile(const struct endpoint* endpoint, unsigned percent) {
static void teardown(void) {
measuring = 0;
diagnostic_active = 0; diagnostic_flush();
// Уничтожаем и активные delay/HOLD пакеты: эмуляторы не должны пережить свои endpoints.
for (unsigned p = 0; p < path_count; p++) { dummynet_destroy(channels[p]); channels[p] = NULL; }
for (int i = 0; i < 2; i++) node_conn_direct_close(ep[i].owner);
@@ -570,6 +794,38 @@ int main(int argc, char** argv) {
if (output) { csv = fopen(output, "w"); CHECK(csv); fprintf(csv,
"case,mask,phase,time_ms,direction,path,rtt_tb,pacing_kbps,inflight,cwnd,conn_rtt_tb,jitter,send_q,local_recv_q,wire_bytes,"
"unique_bytes,retries,channel_q,app_bytes,max_gap_us,dest_recv_q,needed_seq,queue_drops\n"); }
const char* detail = getenv("UTUN_MULTIPATH_DIAG");
if (detail) {
const char* period = getenv("UTUN_MULTIPATH_DIAG_MS");
if (period) {
char* end = NULL; errno = 0;
unsigned long ms = strtoul(period, &end, 10);
if (errno || end == period || *end || ms < 10 || ms > 1000) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "Invalid UTUN_MULTIPATH_DIAG_MS='%s': expected 10..1000", period); return 1;
}
diagnostic_period_us = ms * 1000;
}
diagnostic_file = fopen(detail, "w");
if (!diagnostic_file) {
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "Cannot open multipath diagnostic file '%s': errno=%d", detail, errno); return 1;
}
CHECK(fprintf(diagnostic_file,
"case,mask,phase,time_us,dt_us,side,path,app_count,app_bytes,fragment_count,fragment_bytes,send_count,send_bytes,"
"wait_ack_count,wait_ack_bytes,recv_count,recv_bytes,send_fresh,send_retries,normalizer_bytes,pending_bytes,"
"normalizer_waiter,normalizer_flush,unacked_bytes,admission_budget,flight_bytes,attempts,cwnd_bytes,pacing_Bps,"
"bbr_mode,bbr_cycle,min_rtt_us,srtt_us,jitter_us,rto_us,bw_q16,queue_cap_bytes,ack_count,ack_timer,"
"till_pending,urgent_reply,ack_oldest_us,attempt_oldest_us,pacer_debt_us,pacer_timer,pacer_ready_us,burst,stalled,"
"rto_streak,channel_pending,channel_peak_since_reset,channel_delay_max_since_reset_us,wire_sent_bytes,"
"channel_received_bytes,channel_lost,channel_dropped,app_delivered_bytes,data_tx_bytes,fresh_rx_bytes,"
"data_packets,retries,ack_only_packets,ack_only_wire_bytes,ack_piggy_packets,ack_piggy_bytes,ack_entries,"
"ack_delay_sum_us,ack_delay_max_us,early_ordinary_ack_entries,control_packets,bbr_samples,cwnd_limited_samples,"
"app_limited_samples,rtt_min_us,rtt_max_us,ack_only_entries,ack_only_delay_sum_us,rtt_sum_us,rtt_samples,"
"selection_checks,selection_window_blocked,selection_pacer_blocked,observed_states,flight_min,flight_max,"
"cwnd_min,cwnd_max,send_peak_observed,ack_peak_observed,debt_min_us,debt_max_us\n") > 0);
debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_INFO);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Multipath diagnostics: period_us=%u file=%s; snapshots buffered until drain",
diagnostic_period_us, detail);
}
struct profile matrix[][3] = {
{{20,8000},{20,8000}}, {{20,2000},{20,8000}}, {{20,8000},{200,8000}},
{{20,2000},{200,8000}}, {{20,8000},{200,2000}}, {{20,2000},{80,8000},{200,2000}}, {{20,2000},{80,8000}}
@@ -608,6 +864,10 @@ int main(int argc, char** argv) {
}
CHECK(selected && argc <= 2);
if (csv) CHECK(fclose(csv) == 0);
if (diagnostic_file) {
FILE* completed = diagnostic_file; diagnostic_file = NULL;
CHECK(fclose(completed) == 0);
}
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Multipath completed: failures=%u", failures);
test_clock_finish(); socket_platform_cleanup(); return failures ? 1 : 0;
}