jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
b98b87ea12bcf6c8465d85b85ae3efa8dfce08d3
utun2/tests/test_etcp_multipath.c
T

599 lines
37 KiB
C
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/* Настоящие ETCP/NCD/BBR поверх независимых UDP-каналов. Строгие проверки скорости и пауз при DROP/HOLD. */
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include "etcp.h"
#include "etcp_api.h"
#include "etcp_session.h"
#include "node_conn_direct.h"
#include "topo_node.h"
#include "utun_instance.h"
#include "dummynet.h"
#include "test_net.h"
#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)
#define PATHS 3
#define TRACE_COUNT 131072
#define CMD 253
#define SAMPLE_MS 4000
#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, last_tx, retry_ready;
uint32_t attempts, path_history; // последние16 принятых отправок, по2 бита на номер тестового канала
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;
struct ETCP_CONN* conn;
struct ETCP_SOCKET* sockets[PATHS];
unsigned sent, received;
uint64_t bytes, last_delivery, max_gap, max_latency, latency_hist[10001], samples;
uint64_t max_latency_enqueued, max_latency_delivered, max_latency_trigger_tx;
uint32_t max_latency_msg, max_latency_fragment, max_latency_trigger, max_latency_attempts, max_latency_paths;
};
struct path_stats { uint64_t data, retries, rx_unique; };
static struct UASYNC* ua;
static struct endpoint ep[2];
static struct dummynet* channels[PATHS];
static struct profile profiles[PATHS];
static struct trace traces[2][TRACE_COUNT];
static struct path_stats traffic[2][PATHS];
static unsigned path_count, failures, enabled_mask, bidirectional, low_load, measuring;
static uint64_t started, next_message[2], measure_start, fault_start;
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 uint32_t rx_trigger[2]; // DATA, разбор которого сейчас разблокировал выдачу сообщения
static unsigned fault_paths;
static FILE* csv;
static const char* case_name;
static const char* pubkeys[2] = {
"b3193173def895bd0fcea6f86af077c7d77216f10395275f627ac18242ec0f01",
"ce8871f07fa056c636d297115f231b08c29cdf94e0d440fce83a07c34416d36a"
};
static const char* privkeys[2] = {
"704f2e012c8fa8768130cb0f988a997dccb628372bc5ceccacc78dcbfec5916f",
"38240cb82199e504686507f11f6eaa4f740fde6f0c425c495e49a523019a5d68"
};
/* Выборочная диагностика модели: один 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 (!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) {
if (l->bbr != model) continue;
for (unsigned p = 0; p < path_count; p++) {
if (ss_get_port(&l->remote_addr) != dummynet_get_listen_port(channels[p]) || now < sample_logged[side][p] + 100000) continue;
sample_logged[side][p] = now;
const struct dummynet_stats* ds = dummynet_get_stats(channels[p], side);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "BBR %s mask=%x side=%d path=%u mode=%u cycle=%u sample=%u/%uus rtt=%u "
"min_rtt=%u inflight=%u cwnd=%u pace=%u hi=%u lo=%u app=%d limited=%d DATA_MSS=%u "
"bw_q16=%u/%u bw_lo=%u extra=%u/%u budget_us=%u queue_cap=%u queue_max_us=%llu time_ms=%llu measuring=%u", case_name, enabled_mask,
side, p, model->mode, model->cycle_idx, sample->delivered, sample->interval_us, sample->rtt_us,
model->min_rtt_us, inflight, *cwnd, *pacing, model->inflight_hi, model->inflight_lo,
sample->is_app_limited, limited, mss, model->bw_hi[0], model->bw_hi[1], model->bw_lo,
model->extra_acked[0], model->extra_acked[1], model->queue_budget_us, model->queue_cwnd_cap,
(unsigned long long)ds->max_queue_delay_us, (unsigned long long)((now-started)/1000), measuring);
}
}
}
/* Статистика 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;
if (path >= 0) for (unsigned offset = 0; offset < pkt->data_len;) {
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_REQUEST: len = 1; break;
case ETCP_SECTION_TIMESTAMP: len = 5; break;
case ETCP_SECTION_MEAS_TS: len = 9; break;
case ETCP_SECTION_MEAS_RESP: len = 13; break;
case ETCP_SECTION_FILLER: CHECK(remaining >= 3); len = 3 + data[1] + (data[2] << 8); break;
case ETCP_SECTION_PAYLOAD: {
CHECK(remaining >= ETCP_PAYLOAD_HEADER_SIZE);
uint32_t seq = (uint32_t)data[1] | ((uint32_t)data[2] << 8) | ((uint32_t)data[3] << 16) | ((uint32_t)data[4] << 24);
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) rx_trigger[side] = seq;
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++;
trace->path_history = (trace->path_history << 2) | (unsigned)path;
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,
seq, path, sending, trace->attempts, (unsigned long long)(now - fault_start));
len = remaining; break;
}
default: len = remaining; break;
}
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);
}
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 (!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++) {
if (ss_get_port(&retry->last_link->remote_addr) == dummynet_get_listen_port(channels[p])) previous = p;
if (ss_get_port(&selected->remote_addr) == dummynet_get_listen_port(channels[p])) chosen = p;
}
if (previous < 0 || chosen < 0 || !(fault_paths & (1u << previous)) || !(fault_paths & (1u << chosen))) return selected;
for (struct ETCP_LINK* healthy = conn->links; healthy; healthy = healthy->next) for (unsigned p = 0; p < path_count; p++) {
if ((fault_paths & (1u << p)) || ss_get_port(&healthy->remote_addr) != dummynet_get_listen_port(channels[p])) continue;
int full = bytes > healthy->inflight_lim_bytes || healthy->inflight_bytes > healthy->inflight_lim_bytes - bytes;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "FAULT retry-choice %s seq=%u attempt=%u previous=%d selected=%d elapsed_us=%llu "
"healthy=%u initialized=%u UP=%u pacing_blocked=%d window_full=%d inflight=%u cwnd=%u total_attempts=%u", case_name,
retry->seq, retry->attempt_count, previous, chosen, (unsigned long long)(get_time_us() - fault_start),
p, healthy->initialized, healthy->link_status, healthy->shaper_timer != NULL, full,
healthy->inflight_bytes, healthy->inflight_lim_bytes, conn->active_attempts);
}
return selected;
}
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];
CHECK(entry->len == sizeof(struct message));
struct message msg; memcpy(&msg, entry->dgram, sizeof(msg));
CHECK(msg.cmd == CMD && msg.from == 1-side && msg.seq == dest->received);
for (unsigned i = 0; i < sizeof(msg.body); i++) CHECK(msg.body[i] == (uint8_t)(msg.seq * 3 + msg.from + i));
dest->received++; dest->bytes += sizeof(msg);
uint64_t now = get_time_us();
if (measuring) {
uint64_t gap = now - dest->last_delivery, latency = now - msg.enqueued;
if (gap > dest->max_gap) dest->max_gap = gap;
if (latency > dest->max_latency) {
dest->max_latency = latency; dest->max_latency_msg = msg.seq;
dest->max_latency_fragment = conn->last_delivered_id + 1; dest->max_latency_trigger = rx_trigger[side];
dest->max_latency_enqueued = msg.enqueued; dest->max_latency_delivered = now;
struct trace* trigger = &traces[1-side][rx_trigger[side]];
dest->max_latency_trigger_tx = trigger->first_tx; dest->max_latency_attempts = trigger->attempts;
dest->max_latency_paths = trigger->path_history;
}
dest->latency_hist[latency / 1000 < 10000 ? latency / 1000 : 10000]++; dest->samples++;
}
dest->last_delivery = now;
queue_dgram_free(entry); queue_entry_free(entry);
}
static void create_endpoint(unsigned side) {
char config[2048], listeners[768] = "";
for (unsigned p = 0; p < path_count; p++) {
char text[160]; snprintf(text, sizeof(text), "[server:p%u]\naddr=127.0.0.1:0\ntype=public\n", p);
strcat(listeners, text);
}
snprintf(config, sizeof(config), "[global]\nmy_node_name=multipath-%u\nmy_public_key=%s\nmy_private_key=%s\n"
"keepalive_interval=100\nkeepalive_timeout=2000\nkeepalive_adaptive=0\nauto_sockets=no\n"
"[allowed_keys]\nallow_all=1\n%s",
side, pubkeys[side], privkeys[side], listeners);
ep[side].inst = utun_instance_create_from_str(ua, config); CHECK(ep[side].inst);
CHECK(utun_core_start(ep[side].inst) == 0);
CHECK(etcp_bind(ep[side].inst, CMD, receive_message) == 0);
for (unsigned p = 0; p < path_count; p++) {
char name[16]; snprintf(name, sizeof(name), "p%u", p);
ep[side].sockets[p] = test_etcp_bound_socket(ep[side].inst, name);
}
}
static void setup(const struct profile* settings, unsigned count, unsigned mask) {
memset(ep, 0, sizeof(ep)); memset(channels, 0, sizeof(channels));
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));
memset(rx_trigger, 0, sizeof(rx_trigger));
fault_paths = 0;
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);
create_endpoint(0); create_endpoint(1);
for (unsigned p = 0; p < count; p++) {
if (!(mask & (1u << p))) continue;
channels[p] = dummynet_create(ua, "127.0.0.1", 0); CHECK(channels[p]);
for (int dir = 0; dir < 2; dir++) CHECK(dummynet_set_direction(channels[p], dir, settings[p].rtt_ms / 2, 0,
settings[p].kbps, 512, 0, "127.0.0.1", ss_get_port(&ep[1-dir].sockets[p]->local_addr)) == 0);
struct TOPO_ADDR4 addr = { .protocol = TOPO_PROTO_UDP, .port = dummynet_get_listen_port(channels[p]) };
uint32_t ip = htonl(INADDR_LOOPBACK); memcpy(addr.addr, &ip, sizeof(ip));
struct TOPO_NODE peer = { .node_id = ep[1].inst->node_id, .node_name = "multipath-peer", .v4_addrs = &addr };
memcpy(peer.public_key, ep[1].inst->my_keys.public_key, SC_PUBKEY_SIZE);
struct NODE_CONN_DIRECT* extra = NULL;
CHECK(node_conn_direct_open_node(ep[0].inst, peer.node_id, NULL, NULL, &extra, &peer, ep[0].sockets[p]) >= 0);
if (!ep[0].owner) ep[0].owner = extra;
else node_conn_direct_close(extra);
}
uint64_t end = get_time_us() + 10000000;
int ready = 0;
while (get_time_us() < end) {
test_clock_poll(ua, 10); ready = 1;
for (int i = 0; i < 2; i++) {
ep[i].conn = instance_find_conn(ep[i].inst, ep[1-i].inst->node_id);
struct ETCP_CONN* c = ep[i].conn;
if (!c || !c->initialized || !c->links_up || c->session_required || c->reinit_pending) { ready = 0; continue; }
for (unsigned p = 0; p < count; p++) if (mask & (1u << p)) {
int found = 0;
for (struct ETCP_LINK* l = c->links; l; l = l->next)
if (ss_get_port(&l->remote_addr) == dummynet_get_listen_port(channels[p]) && l->initialized && l->link_status) found++;
if (found != 1) ready = 0;
}
}
if (ready) break;
}
CHECK(ready);
struct TOPO_NODE peer = { .node_id = ep[0].inst->node_id, .node_name = "held-peer" };
memcpy(peer.public_key, ep[0].inst->my_keys.public_key, SC_PUBKEY_SIZE);
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;
}
/* Ограниченная очередь источника: latency не скрывается за неограниченным накоплением тестовых сообщений. */
static void produce(unsigned side) {
struct endpoint* source = &ep[side];
uint64_t now = get_time_us();
if (low_load && now < next_message[side]) return;
for (unsigned budget = 0; budget < 64 && source->conn->send_input_q->count < 4; budget++) {
if (source->sent - ep[1-side].received >= 2048) break;
struct ll_entry* entry = ll_alloc_lldgram(sizeof(struct message)); CHECK(entry);
struct message msg = { .cmd = CMD, .from = side, .seq = source->sent, .enqueued = now };
for (unsigned i = 0; i < sizeof(msg.body); i++) msg.body[i] = msg.seq * 3 + side + i;
memcpy(entry->dgram, &msg, sizeof(msg)); entry->len = sizeof(msg);
CHECK(etcp_send(source->conn, entry) == 0); source->sent++;
if (low_load) { next_message[side] = now + 10000; break; }
}
}
static void snapshot(const char* phase) {
uint64_t now = get_time_us();
for (int side = 0; side < 2; side++) {
struct ETCP_CONN* conn = ep[side].conn;
for (struct ETCP_LINK* l = conn->links; l; l = l->next) {
uint64_t occupied = 0; unsigned packets = 0;
for (struct ll_entry* e = l->attempts->head; e; e = e->next) { occupied += e->len; packets++; }
CHECK(l->inflight_bytes == occupied && l->inflight_packets == packets);
int p = -1;
for (unsigned j = 0; j < path_count; j++)
if (ss_get_port(&l->remote_addr) == dummynet_get_listen_port(channels[j])) p = j;
if (p < 0) continue;
const struct dummynet_stats* ds = dummynet_get_stats(channels[p], side);
if (csv) fprintf(csv, "%s,%x,%s,%llu,%d,%d,%u,%u,%u,%u,%u,%u,%d,%d,%llu,%llu,%llu,%u,%llu,%llu,%d,%u,%llu\n",
case_name, enabled_mask, phase, (unsigned long long)((now-started)/1000), side, p, l->rtt_last, l->bandwidth,
l->inflight_bytes, l->inflight_lim_bytes, conn->rtt_avg_10, conn->jitter, conn->input_send_q->count,
conn->recv_q->count, (unsigned long long)ds->sent_bytes, (unsigned long long)traffic[side][p].rx_unique,
(unsigned long long)traffic[side][p].retries, ds->queue_size,
(unsigned long long)ep[1-side].bytes, (unsigned long long)ep[1-side].max_gap,
ep[1-side].conn->recv_q->count, ep[1-side].conn->last_delivered_id+1, (unsigned long long)ds->dropped);
}
CHECK(conn->unacked_bytes == conn->input_send_q->total_bytes + conn->input_wait_ack->total_bytes);
}
}
static void run_for(unsigned ms, int sending, const char* phase) {
uint64_t end = get_time_us() + ms * 1000ULL, next_snapshot = get_time_us();
while (get_time_us() < end) {
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 (!sending && ep[0].sent == ep[1].received && ep[1].sent == ep[0].received &&
!ep[0].conn->unacked_bytes && !ep[1].conn->unacked_bytes) break;
}
}
static void begin_measure(void) {
measure_start = get_time_us(); measuring = 1;
memset(traffic, 0, sizeof(traffic));
for (int i = 0; i < 2; i++) {
ep[i].max_gap = ep[i].max_latency = ep[i].samples = 0; ep[i].last_delivery = measure_start;
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);
}
static void expect(int passed, const char* requirement, uint64_t actual, uint64_t limit) {
if (passed) return;
failures++;
DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "FAIL %s: %s actual=%llu limit=%llu", case_name, requirement,
(unsigned long long)actual, (unsigned long long)limit);
}
static uint64_t percentile(const struct endpoint* endpoint, unsigned percent) {
uint64_t sum = 0, target = (endpoint->samples * percent + 99) / 100;
if (!target) return 0;
for (unsigned ms = 0; ms <= 10000; ms++) {
sum += endpoint->latency_hist[ms];
if (sum >= target) return ms;
}
return 10000;
}
static void teardown(void) {
measuring = 0;
// Уничтожаем и активные 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);
uasync_drain_immediate(ua);
for (int i = 0; i < 2; i++) { utun_instance_destroy(ep[i].inst); ep[i].inst = NULL; }
uasync_poll(ua, 0); CHECK(ua->timer_alloc_count == ua->timer_free_count); uasync_destroy(ua, 0);
}
/* Полезные байты приложения — отдельный критерий; FILLER/повторы не могут улучшить результат. */
static uint64_t measure_rate(const struct profile* settings, unsigned count, unsigned mask) {
setup(settings, count, mask); low_load = 0;
run_for(WARMUP_MS, 1, "warmup"); begin_measure();
uint64_t before[2] = {ep[0].bytes, ep[1].bytes};
run_for(SAMPLE_MS, 1, "steady");
uint64_t elapsed = get_time_us() - measure_start, rate = 0;
for (unsigned side = 0; side <= bidirectional; side++) {
uint64_t useful = (ep[1-side].bytes-before[1-side]) * 8000 / elapsed, capacity = 0, unique_total = 0;
for (unsigned p = 0; p < count; p++) if (mask & (1u << p)) {
uint64_t wire = dummynet_get_stats(channels[p], side)->sent_bytes * 8000 / elapsed;
uint64_t unique = traffic[side][p].rx_unique * 8000 / elapsed;
uint64_t queue_us = dummynet_get_stats(channels[p], side)->max_queue_delay_us;
unique_total += traffic[side][p].rx_unique;
capacity += settings[p].kbps;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "RATE %s dir=%u path=%u RTT=%u BWlim=%u wire=%llu unique=%llu kbps retries=%llu queue_us=%llu",
case_name, side, p, settings[p].rtt_ms, settings[p].kbps, (unsigned long long)wire,
(unsigned long long)unique, (unsigned long long)traffic[side][p].retries, (unsigned long long)queue_us);
expect(wire >= settings[p].kbps * 90 / 100, "channel wire utilization >=90%", wire, settings[p].kbps * 90 / 100);
expect(unique >= settings[p].kbps * 80 / 100, "channel unique DATA >=80%", unique, settings[p].kbps * 80 / 100);
expect(queue_us <= (settings[p].rtt_ms + 20) * 1000ULL, "queue delay <=base RTT+20ms",
queue_us, (settings[p].rtt_ms + 20) * 1000ULL);
}
uint64_t fresh = unique_total * 8000 / elapsed;
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "GOODPUT %s dir=%u application=%llu fresh_DATA=%llu capacity=%llu kbps", case_name, side,
(unsigned long long)useful, (unsigned long long)fresh, (unsigned long long)capacity);
// Нельзя пройти за счёт выдачи фрагментов, накопленных до начала измерения.
if (fresh < useful) useful = fresh;
if (!side) rate = useful;
expect(useful >= capacity * 85 / 100, "application goodput >=85% capacity", useful, capacity * 85 / 100);
}
measuring = 0;
run_for(10000, 0, "drain");
for (int i = 0; i < 2; i++) expect(ep[i].sent == ep[1-i].received, "complete ordered delivery after drain",
ep[1-i].received, ep[i].sent);
teardown(); return rate;
}
/* Не ждём восстановления отказавшего пути: проверяем паузу и скорость пока DROP/HOLD ещё включён. */
static void stalled(const struct profile* settings, unsigned count, unsigned broken, unsigned directions,
enum dummynet_mode mode, int sparse, unsigned duration_ms) {
low_load = sparse;
setup(settings, count, (1u << count)-1);
run_for(WARMUP_MS, 1, "warmup");
// Выбираем момент, когда на каждом останавливаемом пути есть действительно неподтверждённый пакет.
uint64_t deadline = get_time_us() + 2000000;
unsigned pending = 0;
fault_seq = 0;
while (get_time_us() < deadline) {
pending = 0; fault_seq = 0;
for (struct ll_entry* entry = ep[0].conn->input_wait_ack->head; entry; entry = entry->next) {
struct INFLIGHT_PACKET* pkt = (struct INFLIGHT_PACKET*)entry;
if (!pkt->last_link || ((directions & 1) && traces[0][pkt->seq].first_rx)) continue;
for (unsigned p = 0; p < count; p++) if ((broken & (1u << p)) &&
ss_get_port(&pkt->last_link->remote_addr) == dummynet_get_listen_port(channels[p])) {
// Пакет должен ещё находиться в propagation delay, а не уже лежать в сокете получателя.
if ((directions & 1) && get_time_us() - traces[0][pkt->seq].first_tx >= settings[p].rtt_ms * 500ULL) continue;
pending |= 1u << p;
if (!fault_seq || pkt->seq < fault_seq) fault_seq = pkt->seq;
}
}
if (pending == broken) break;
run_for(1, 1, "arm-fault");
}
CHECK(pending == broken && fault_seq);
begin_measure(); fault_start = get_time_us(); fault_paths = broken;
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);
for (unsigned p = 0; p < count; p++) if (broken & (1u << p))
for (unsigned dir = 0; dir < 2; dir++) if (directions & (1u << dir))
CHECK(dummynet_set_mode(channels[p], dir, mode) == 0);
uint64_t before = ep[1].bytes;
run_for(duration_ms > 1000 ? 1000 : duration_ms, 1, "fault-transition");
uint64_t steady_bytes = ep[1].bytes, steady_start = get_time_us();
uint64_t steady_unique = 0;
for (unsigned p = 0; p < count; p++) steady_unique += traffic[0][p].rx_unique;
if (duration_ms > 1000) run_for(duration_ms - 1000, 1, "fault-steady");
uint64_t now = get_time_us(), tail_gap = now - ep[1].last_delivery;
if (tail_gap > ep[1].max_gap) ep[1].max_gap = tail_gap;
uint64_t capacity = 0;
for (unsigned p = 0; p < count; p++) if (!(broken & (1u << p))) capacity += settings[p].kbps;
uint64_t rate = now > steady_start ? (ep[1].bytes-steady_bytes) * 8000 / (now-steady_start) : 0;
uint64_t unique_now = 0;
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,
(unsigned long long)percentile(&ep[1], 50), (unsigned long long)percentile(&ep[1], 95),
(unsigned long long)percentile(&ep[1], 99), (unsigned long long)ep[1].max_latency,
(unsigned long long)rate, (unsigned long long)capacity, fault_seq,
(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);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "FAULT worst %s message=%u fragment=%u trigger=%u trigger_attempts=%u paths_2bit=%08x "
"enqueued_us=%lld delivered_us=%lld trigger_first_tx_us=%lld latency_us=%llu", case_name,
ep[1].max_latency_msg, ep[1].max_latency_fragment, ep[1].max_latency_trigger,
ep[1].max_latency_attempts, ep[1].max_latency_paths,
(long long)((int64_t)ep[1].max_latency_enqueued - (int64_t)fault_start),
(long long)((int64_t)ep[1].max_latency_delivered - (int64_t)fault_start),
(long long)((int64_t)ep[1].max_latency_trigger_tx - (int64_t)fault_start), (unsigned long long)ep[1].max_latency);
if (duration_ms > 100) expect(ep[1].bytes > before, "delivery during active fault", ep[1].bytes-before, 1);
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);
measuring = duration_ms <= 100;
for (unsigned p = 0; p < count; p++) if (broken & (1u << p))
for (unsigned dir = 0; dir < 2; dir++) CHECK(dummynet_set_mode(channels[p], dir, DUMMYNET_RUN) == 0);
run_for(1000, 1, "recovery");
if (duration_ms <= 100) {
expect(ep[1].samples > 0, "delivery after short HOLD", ep[1].samples, 1);
expect(ep[1].max_gap <= MAX_GAP_US, "short HOLD including resume: pause <=100ms", ep[1].max_gap, MAX_GAP_US);
expect(ep[1].max_latency <= MAX_GAP_US, "short HOLD including resume: latency <=100ms", ep[1].max_latency, MAX_GAP_US);
}
measuring = 0;
run_for(2000, 1, "recovery"); run_for(10000, 0, "drain");
for (int i = 0; i < 2; i++) expect(ep[i].sent == ep[1-i].received, "complete delivery after recovery",
ep[1-i].received, ep[i].sent);
teardown();
}
int main(int argc, char** argv) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); test_clock_init(10);
log_bbr_samples = getenv("UTUN_BBR_SAMPLES") != NULL;
if (getenv("UTUN_TEST_DEBUG")) debug_set_category_level(DEBUG_CATEGORY_ETCP, DEBUG_LEVEL_DEBUG);
CHECK(socket_platform_init() == 0);
const char* output = getenv("UTUN_MULTIPATH_CSV");
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"); }
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}}
};
const char* names[] = {"equal", "bandwidth", "rtt", "fast-distant", "slow-distant", "three", "bidirectional"};
size_t baseline = u_get_allocated_count();
unsigned selected = 0;
for (unsigned n = 0; n < sizeof(names)/sizeof(names[0]); n++) {
if (argc > 1 && strcmp(argv[1], names[n])) continue;
selected++;
bidirectional = n == 6;
case_name = names[n]; unsigned count = n == 5 ? 3 : 2;
uint64_t solo_sum = 0;
for (unsigned p = 0; p < count; p++) solo_sum += measure_rate(matrix[n], count, 1u << p);
uint64_t aggregate = measure_rate(matrix[n], count, (1u << count)-1);
expect(aggregate >= solo_sum * 90 / 100, "aggregation >=90% sum of solo rates", aggregate, solo_sum * 90 / 100);
CHECK(u_get_allocated_count() == baseline);
}
struct { const char* name; unsigned count, broken, directions, sparse, ms; enum dummynet_mode mode; } faults[] = {
{"drop",2,2,3,1,3000,DUMMYNET_DROP}, {"hold",2,2,3,1,3000,DUMMYNET_HOLD},
{"drop-forward",2,2,1,1,3000,DUMMYNET_DROP}, {"drop-reverse",2,2,2,1,3000,DUMMYNET_DROP},
{"short-hold",2,2,3,1,30,DUMMYNET_HOLD}, {"slow-drop",2,2,3,1,3000,DUMMYNET_DROP},
{"two-stuck",3,6,3,1,3000,DUMMYNET_DROP}, {"load-drop",2,2,3,0,5000,DUMMYNET_DROP},
{"load-hold",2,2,3,0,5000,DUMMYNET_HOLD}
};
for (unsigned n = 0; n < sizeof(faults)/sizeof(faults[0]); n++) {
if (argc > 1 && strcmp(argv[1], faults[n].name)) continue;
selected++;
case_name = faults[n].name;
bidirectional = 0;
struct profile settings[] = {{20,8000},{20,8000},{20,2000}};
if (!strcmp(case_name, "slow-drop")) settings[1].rtt_ms = 200;
if (faults[n].count == 3) { settings[1].rtt_ms = 80; settings[2].rtt_ms = 200; }
stalled(settings, faults[n].count, faults[n].broken, faults[n].directions, faults[n].mode, faults[n].sparse, faults[n].ms);
CHECK(u_get_allocated_count() == baseline);
}
CHECK(selected && argc <= 2);
if (csv) CHECK(fclose(csv) == 0);
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Multipath completed: failures=%u", failures);
test_clock_finish(); socket_platform_cleanup(); return failures ? 1 : 0;
}
Reference in new issue View Git Blame Copy Permalink