diff --git a/tests/test_etcp_multipath.c b/tests/test_etcp_multipath.c index 4c2ee4f4..f19531b6 100644 --- a/tests/test_etcp_multipath.c +++ b/tests/test_etcp_multipath.c @@ -41,6 +41,8 @@ struct endpoint { 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; }; struct path_stats { uint64_t data, retries, rx_unique; }; static struct UASYNC* ua; @@ -57,6 +59,7 @@ 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 FILE* csv; static const char* case_name; static const char* pubkeys[2] = { @@ -115,6 +118,7 @@ static void observe_packet(struct ETCP_DGRAM* pkt, struct ETCP_LINK* link, int s 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++; @@ -195,7 +199,13 @@ static void receive_message(struct ETCP_CONN* conn, struct ll_entry* entry) { 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; + 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->latency_hist[latency / 1000 < 10000 ? latency / 1000 : 10000]++; dest->samples++; } dest->last_delivery = now; @@ -227,6 +237,7 @@ static void setup(const struct profile* settings, unsigned count, unsigned mask) 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)); 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); @@ -476,6 +487,12 @@ static void stalled(const struct profile* settings, unsigned count, unsigned bro (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 " + "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, + (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);