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

Measure multipath recovery stages separately from RTO waits

Browse Source
This commit is contained in:
evgeny committed 2026-10-24 10:05:13 +02:00
1 parent 1c6a7e7bf9
commit 1d295c7d3a
2 files changed
+94 -7
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/Makefile.am
+2 -1
View File
@@ -1044,7 +1044,8 @@ test_dummynet_channels_LDADD = $(LIBUTUN) $(COMMON_LIBS)
test_dummynet_channels_LDFLAGS = $(test_clock_ldflags)
test_etcp_multipath_SOURCES = test_etcp_multipath.c $(test_clock_sources)
test_etcp_multipath_LDADD = $(LIBUTUN) $(COMMON_LIBS)
test_etcp_multipath_LDFLAGS = $(test_clock_ldflags) -Wl,--wrap=etcp_dump_pkt_sections -Wl,--wrap=bbr_main
test_etcp_multipath_LDFLAGS = $(test_clock_ldflags) -Wl,--wrap=etcp_dump_pkt_sections -Wl,--wrap=bbr_main \
-Wl,--wrap=etcp_tx_complete -Wl,--wrap=queue_data_put_with_index
test_stcp_sigpipe_SOURCES = test_stcp_sigpipe.c
test_stcp_sigpipe_LDADD = $(LIBUTUN) $(COMMON_LIBS)
test_etcp_windows_LDADD = $(LIBUTUN) $(COMMON_LIBS)
tests/test_etcp_multipath.c
+92 -6
View File
@@ -21,9 +21,19 @@
#define WARMUP_MS 3000
#define STALL_MS 3000
#define MAX_GAP_US 100000
#define MAX_RTO_LATE_US 20000 // запас на миллисекундный таймер и планирование тестового процесса
struct profile { unsigned rtt_ms, kbps; };
struct message { uint8_t cmd, from, reserved[2]; uint32_t seq; uint64_t enqueued; uint8_t body[1008]; };
struct trace { uint64_t first_tx, first_rx; uint32_t attempts; uint8_t first_path; };
struct trace {
uint64_t first_tx, first_rx, last_tx, retry_ready;
uint32_t attempts;
uint16_t last_number;
uint8_t first_path, last_path;
};
struct recovery_stats {
uint64_t rto_count, retry_count, rto_late_us, ready_wait_us, transit_extra_us;
uint32_t late_seq, ready_seq, transit_seq;
};
struct endpoint {
struct UTUN_INSTANCE* inst;
struct NODE_CONN_DIRECT* owner;
@@ -45,6 +55,8 @@ static uint32_t fault_seq;
static unsigned fault_trace_count;
static int log_bbr_samples;
static uint64_t sample_logged[2][PATHS];
static struct recovery_stats recovery;
static uint32_t fault_rto_us[PATHS]; // максимум RTO каждого пути с начала отказа, независимо от измеренной паузы
static FILE* csv;
static const char* case_name;
static const char* pubkeys[2] = {
@@ -81,9 +93,8 @@ void __wrap_bbr_main(struct bbr* model, const struct bbr_rate_sample* sample, ui
}
}
/* Наблюдение не заменяет передачу/приём и не меняет состояние протокола. */
void __real_etcp_dump_pkt_sections(struct ETCP_DGRAM*, struct ETCP_LINK*, int);
void __wrap_etcp_dump_pkt_sections(struct ETCP_DGRAM* pkt, struct ETCP_LINK* link, int sending) {
/* Статистика TX снимается только после принятия send; RX — перед разбором уже расшифрованного пакета. */
static void observe_packet(struct ETCP_DGRAM* pkt, struct ETCP_LINK* link, int sending) {
int side = link->etcp->instance == ep[0].inst ? 0 : 1, path = -1;
for (unsigned p = 0; p < path_count; p++)
if (ss_get_port(&link->remote_addr) == dummynet_get_listen_port(channels[p])) path = p;
@@ -103,13 +114,26 @@ void __wrap_etcp_dump_pkt_sections(struct ETCP_DGRAM* pkt, struct ETCP_LINK* lin
CHECK(seq < TRACE_COUNT);
struct trace* trace = &traces[sending ? side : 1-side][seq];
uint64_t now = get_time_us();
uint16_t number = (uint16_t)data[5] | ((uint16_t)data[6] << 8);
if (sending) {
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; }
}
if (trace->attempts++ && measuring) traffic[side][path].retries++;
if (!trace->first_tx) { trace->first_tx = now; trace->first_path = path; }
trace->last_tx = now; trace->last_number = number; trace->last_path = path; trace->retry_ready = 0;
if (measuring && fault_start && !side && etcp_tx_rto_us(link) > fault_rto_us[path])
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) {
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) {
uint64_t elapsed = now - trace->last_tx, base = profiles[path].rtt_ms * 500ULL;
uint64_t extra = elapsed > base ? elapsed - base : 0;
if (extra > recovery.transit_extra_us) { recovery.transit_extra_us = extra; recovery.transit_seq = seq; }
}
}
if (fault_start && (seq == fault_seq || (sending && trace->attempts > 1 && fault_trace_count++ < 32)))
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "FAULT trace dir=%d seq=%u path=%d tx=%d attempts=%u elapsed_us=%llu", side,
@@ -120,9 +144,45 @@ void __wrap_etcp_dump_pkt_sections(struct ETCP_DGRAM* pkt, struct ETCP_LINK* lin
}
CHECK(len <= remaining); offset += len;
}
}
void __real_etcp_dump_pkt_sections(struct ETCP_DGRAM*, struct ETCP_LINK*, int);
void __wrap_etcp_dump_pkt_sections(struct ETCP_DGRAM* pkt, struct ETCP_LINK* link, int sending) {
if (!sending) observe_packet(pkt, link, 0);
__real_etcp_dump_pkt_sections(pkt, link, sending);
}
void __real_etcp_tx_complete(struct ETCP_DGRAM*, int);
void __wrap_etcp_tx_complete(struct ETCP_DGRAM* pkt, int success) {
if (success) observe_packet(pkt, pkt->link, 1);
__real_etcp_tx_complete(pkt, success);
}
/* Возврат логического DATA после RTO: таймер уже сработал, новая физическая попытка ещё не создана. */
int __real_queue_data_put_with_index(struct ll_queue*, struct ll_entry*);
int __wrap_queue_data_put_with_index(struct ll_queue* q, struct ll_entry* entry) {
if (measuring && fault_start && ep[0].conn && q == ep[0].conn->input_send_q) {
struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)entry;
if (packet->last_attempt_id && packet->last_link) {
CHECK(packet->seq < TRACE_COUNT && !packet->active_attempts);
struct trace* trace = &traces[0][packet->seq];
struct ETCP_LINK* link = packet->last_link;
uint64_t now = get_time_us(); trace->retry_ready = now;
CHECK(trace->last_tx && trace->last_number == (uint16_t)packet->last_attempt_id);
if (!link->closing) {
uint32_t rto = etcp_tx_rto_us(link);
CHECK(now - trace->last_tx >= rto);
uint64_t late = now - trace->last_tx - rto; recovery.rto_count++;
if (late > recovery.rto_late_us) { recovery.rto_late_us = late; recovery.late_seq = packet->seq; }
for (unsigned p = 0; p < path_count; p++)
if (ss_get_port(&link->remote_addr) == dummynet_get_listen_port(channels[p]) && rto > fault_rto_us[p])
fault_rto_us[p] = rto;
}
}
}
return __real_queue_data_put_with_index(q, entry);
}
static void receive_message(struct ETCP_CONN* conn, struct ll_entry* entry) {
int side = conn->instance == ep[0].inst ? 0 : 1;
struct endpoint* dest = &ep[side];
@@ -166,6 +226,7 @@ static void setup(const struct profile* settings, unsigned count, unsigned mask)
memset(traces, 0, sizeof(traces)); memset(traffic, 0, sizeof(traffic));
memset(next_message, 0, sizeof(next_message));
memset(sample_logged, 0, sizeof(sample_logged));
memset(&recovery, 0, sizeof(recovery)); memset(fault_rto_us, 0, sizeof(fault_rto_us));
path_count = count; enabled_mask = mask; measuring = 0; fault_start = 0; fault_trace_count = 0;
memcpy(profiles, settings, count * sizeof(*settings));
ua = uasync_create(); CHECK(ua);
@@ -365,6 +426,9 @@ static void stalled(const struct profile* settings, unsigned count, unsigned bro
}
CHECK(pending == broken && fault_seq);
begin_measure(); fault_start = get_time_us();
for (struct ETCP_LINK* link = ep[0].conn->links; link; link = link->next)
for (unsigned p = 0; p < count; p++)
if (ss_get_port(&link->remote_addr) == dummynet_get_listen_port(channels[p])) fault_rto_us[p] = etcp_tx_rto_us(link);
uint32_t epochs[2] = {ep[0].conn->reinit_count, ep[1].conn->reinit_count};
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "FAULT %s mode=%d paths=%x directions=%x seq=%u duration_ms=%u sparse=%d",
case_name, mode, broken, directions, fault_seq, duration_ms, sparse);
@@ -386,6 +450,24 @@ static void stalled(const struct profile* settings, unsigned count, unsigned bro
for (unsigned p = 0; p < count; p++) unique_now += traffic[0][p].rx_unique;
uint64_t fresh_rate = now > steady_start ? (unique_now-steady_unique) * 8000 / (now-steady_start) : 0;
if (fresh_rate < rate) rate = fresh_rate;
uint64_t rto_budget = 0;
for (unsigned p = 0; p < count; p++) if (broken & (1u << p)) rto_budget += fault_rto_us[p];
// Быстрые профили сохраняют 100мс. Для длинных учитываем по одному RTO каждого отказавшего пути.
uint64_t delivery_limit = MAX_GAP_US + (rto_budget > MAX_GAP_US ? rto_budget : 0);
for (struct ll_entry* entry = ep[0].conn->input_send_q->head; entry; entry = entry->next) {
struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)entry;
CHECK(packet->seq < TRACE_COUNT);
uint64_t ready = traces[0][packet->seq].retry_ready;
if (ready && now - ready > recovery.ready_wait_us) {
recovery.ready_wait_us = now - ready; recovery.ready_seq = packet->seq;
}
}
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "FAULT stages %s RTO_count=%llu retry_count=%llu RTO_budget_us=%llu delivery_limit_us=%llu "
"RTO_late_us=%llu late_seq=%u ready_wait_us=%llu ready_seq=%u transit_extra_us=%llu transit_seq=%u", case_name,
(unsigned long long)recovery.rto_count, (unsigned long long)recovery.retry_count,
(unsigned long long)rto_budget, (unsigned long long)delivery_limit,
(unsigned long long)recovery.rto_late_us, recovery.late_seq, (unsigned long long)recovery.ready_wait_us, recovery.ready_seq,
(unsigned long long)recovery.transit_extra_us, recovery.transit_seq);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "FAULT result %s delivered=%llu gap_us=%llu p50=%llu p95=%llu p99=%llu max_latency_us=%llu "
"steady_kbps=%llu healthy_capacity=%llu missing_seq=%u first_rx_us=%llu attempts=%u", case_name,
(unsigned long long)(ep[1].bytes-before), (unsigned long long)ep[1].max_gap,
@@ -395,8 +477,12 @@ static void stalled(const struct profile* settings, unsigned count, unsigned bro
(unsigned long long)(traces[0][fault_seq].first_rx > fault_start ? traces[0][fault_seq].first_rx-fault_start : 0),
traces[0][fault_seq].attempts);
if (duration_ms > 100) expect(ep[1].bytes > before, "delivery during active fault", ep[1].bytes-before, 1);
expect(ep[1].max_gap <= MAX_GAP_US, "maximum delivery pause <=100ms", ep[1].max_gap, MAX_GAP_US);
if (sparse) expect(ep[1].max_latency <= MAX_GAP_US, "bounded-source latency <=100ms", ep[1].max_latency, MAX_GAP_US);
expect(recovery.rto_late_us <= MAX_RTO_LATE_US, "RTO processing overhead <=20ms", recovery.rto_late_us, MAX_RTO_LATE_US);
expect(recovery.ready_wait_us <= MAX_GAP_US, "ready retry waits <=100ms", recovery.ready_wait_us, MAX_GAP_US);
expect(recovery.transit_extra_us <= MAX_GAP_US, "DATA transit beyond base delay <=100ms", recovery.transit_extra_us, MAX_GAP_US);
expect(ep[1].max_gap <= delivery_limit, "delivery pause within profile RTO budget", ep[1].max_gap, delivery_limit);
if (sparse) expect(ep[1].max_latency <= delivery_limit, "bounded-source latency within profile RTO budget",
ep[1].max_latency, delivery_limit);
else if (duration_ms > 1000) expect(rate >= capacity * 85 / 100,
"fault steady goodput >=85% healthy capacity", rate, capacity * 85 / 100);
for (int i = 0; i < 2; i++) CHECK(ep[i].conn->reinit_count == epochs[i] && ep[i].conn->links_up);