Check active channel stalls, delivery deadlines and delayed packet recovery
This commit is contained in:
evgeny committed
1 parent
672f04b833
commit
6e350760d1
1 file changed
+93
-5
@@ -41,6 +41,7 @@ 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 FILE* csv;
|
||||
static const char* case_name;
|
||||
static const char* pubkeys[2] = {
|
||||
@@ -81,8 +82,8 @@ void __wrap_etcp_dump_pkt_sections(struct ETCP_DGRAM* pkt, struct ETCP_LINK* lin
|
||||
trace->first_rx = now;
|
||||
if (measuring) traffic[1-side][path].rx_unique += remaining - 5;
|
||||
}
|
||||
if (fault_start && seq < 100 && trace->attempts > 1)
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "trace dir=%d seq=%u path=%d tx=%d attempts=%u elapsed_us=%llu", side,
|
||||
if (fault_start && seq == fault_seq)
|
||||
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;
|
||||
}
|
||||
@@ -119,7 +120,8 @@ static void create_endpoint(unsigned side) {
|
||||
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%s",
|
||||
"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);
|
||||
@@ -161,8 +163,6 @@ static void setup(const struct profile* settings, unsigned count, unsigned mask)
|
||||
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)) {
|
||||
struct ETCP_LINK* link = etcp_link_find_by_addr(ep[i].sockets[p], &c->links->remote_addr, 0);
|
||||
(void)link;
|
||||
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++;
|
||||
@@ -242,6 +242,16 @@ static void expect(int passed, const char* requirement, uint64_t actual, uint64_
|
||||
(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.
|
||||
@@ -281,6 +291,69 @@ static uint64_t measure_rate(const struct profile* settings, unsigned count, uns
|
||||
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;
|
||||
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])) {
|
||||
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();
|
||||
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();
|
||||
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;
|
||||
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);
|
||||
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);
|
||||
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 = 0;
|
||||
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(3000, 1, "recovery"); run_for(5000, 0, "drain");
|
||||
for (int i = 0; i < 2; i++) CHECK(ep[i].sent == ep[1-i].received);
|
||||
teardown();
|
||||
}
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); test_clock_init(10);
|
||||
if (getenv("UTUN_TEST_DEBUG")) debug_set_category_level(DEBUG_CATEGORY_ETCP, DEBUG_LEVEL_DEBUG);
|
||||
@@ -303,6 +376,21 @@ int main(int argc, char** argv) {
|
||||
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;
|
||||
case_name = faults[n].name;
|
||||
struct profile settings[] = {{20,8000},{20,8000},{20,2000}};
|
||||
if (!strcmp(case_name, "slow-drop")) settings[1].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);
|
||||
}
|
||||
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
Block a user