From d2703b21451ccb8997082346ee0aae3a3860c21d Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 24 Oct 2026 15:23:23 +0200 Subject: [PATCH] Use receiver delivery clock for ETCP bandwidth samples --- src/transport_layer/etcp_tx.c | 7 +++--- src/transport_layer/etcp_tx.h | 6 +++-- tests/test_etcp_windows.c | 41 ++++++++++++++++++++++++++++++----- 3 files changed, 44 insertions(+), 10 deletions(-) diff --git a/src/transport_layer/etcp_tx.c b/src/transport_layer/etcp_tx.c index 0ee6e65d..e4ef17f9 100644 --- a/src/transport_layer/etcp_tx.c +++ b/src/transport_layer/etcp_tx.c @@ -308,7 +308,7 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint uint64_t numerator = (uint64_t)rs.delivered * rx_us; uint64_t bound = numerator / rx_bytes + !!(numerator % rx_bytes); if (bound > UINT32_MAX) rs.interval_us = 0; - else if (bound > rs.interval_us) rs.interval_us = (uint32_t)bound; + else rs.interval_us = (uint32_t)(bound > send_elapsed ? bound : send_elapsed); } } if (interval > UINT32_MAX || (interval && !rs.interval_us)) @@ -323,9 +323,10 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint etcp_link_update_inflight_lim(link, cwnd); etcp_loadbalancer_set_pacing_rate(link, pacing); } DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] ACK seq=%u attempt=%u path=%u RTT_us=%u delay_us=%u sample=%u/%u " - "local_interval_us=%llu RX=%llu/%llu flight=%u cwnd=%u pace=%u RTO_us=%u", + "clock=%s send_us=%llu ACK_us=%llu RX=%llu/%llu flight=%u cwnd=%u pace=%u RTO_us=%u", conn->log_name, seq, (unsigned)(uint16_t)attempt_id, link->local_link_id, - rtt, receiver_delay_us, rs.delivered, rs.interval_us, (unsigned long long)interval, + rtt, receiver_delay_us, rs.delivered, rs.interval_us, rx_us ? "RX" : "ACK", + (unsigned long long)send_elapsed, (unsigned long long)ack_elapsed, (unsigned long long)rx_bytes, (unsigned long long)rx_us, link->inflight_bytes, link->inflight_lim_bytes, link->bbr_pacing_rate, etcp_tx_rto_us(link)); link->acked_bytes += sample.ll.len; link->acked_packets++; diff --git a/src/transport_layer/etcp_tx.h b/src/transport_layer/etcp_tx.h index 079ebb6e..ad1689d3 100644 --- a/src/transport_layer/etcp_tx.h +++ b/src/transport_layer/etcp_tx.h @@ -61,8 +61,10 @@ void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success); * Неизвестный старый ID не создаёт измерение и не вычитает окно. UINT32_MAX delay — RTT недостоверен. * До застревания действующий ACK сбрасывает серию RTO; после — только ACK отправки после установки флага сбрасывает серию/флаг. * peer_rx_bytes/time_us — согласованная точка приёма пути из ACK; 0/0 — неизвестна. - * Достоверная пара ограничивает BW темпом приёма, сохраняя границы send/ACK; RTT и доставка от неё не зависят. - * interval=max(send_us,ack_us,ceil(delivered*delta_rx_us/delta_rx_bytes)); дельты от снимка при отправке. + * При достоверных положительных дельтах interval=max(send_us,ceil(delivered*delta_rx_us/delta_rx_bytes)); + * дельты от снимка при отправке. Темп приёма заменяет ACK clock, исключая и сжатие, и растяжение обратной очередью. + * send-интервал сохраняет защиту от сжатия RX clock. Без приёмной базы/прогресса используется max(send_us,ack_us). + * RTT, min_rtt, логическая доставка и счётчики подтверждённых байтов от выбора часов BW не зависят. * Используется неубывающая точка пути; старый/чужой attempt не меняет её, перестановка ACK не откатывает. * Непредставимый интервал не создаёт оценку BW. Пустой flight заново начинает приёмную базу, исключая простой. */ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint32_t receiver_delay_us, diff --git a/tests/test_etcp_windows.c b/tests/test_etcp_windows.c index d7115687..98233564 100644 --- a/tests/test_etcp_windows.c +++ b/tests/test_etcp_windows.c @@ -1323,11 +1323,43 @@ static void receive_admission_and_path_acks(void) { teardown(); } -/* Реальный ACK handler: RX clock независим, граница ограничивает BW и округляется безопасно. */ +/* Обратная очередь растягивает ACK, но не меняет темп DATA; send-интервал защищает от сжатия RX clock. */ +static void receiver_delivery_clock(void) { + const struct { uint32_t ack_us, send_us; uint64_t rx_bytes, rx_us; unsigned anchor; uint32_t expected; } cases[] = { + {20000,2000,1000,5000,1,5000}, {100000,2000,1000,5000,1,5000}, + {100000,8000,1000,5000,1,8000}, {100000,2000,5000,1000,1,2000}, + {20000,2000,0,0,1,20000}, {20000,2000,1000,0,1,20000}, + {20000,2000,1000,5000,0,20000} + }; + for (unsigned i = 0; i < sizeof(cases)/sizeof(cases[0]); i++) { + setup(); b->link_status = 0; a->data_rtt_us = 1000000; + ack_now_us = get_time_us(); ack_clock_active = 1; + enable_models(); fill(a,1,1); + a->peer_rx_data_bytes = 10000; a->peer_rx_data_time_us = 1000000; + struct INFLIGHT_PACKET* packet = enqueue(1000); + struct ETCP_DGRAM* pkt = request(); CHECK(pkt && pkt->link == a); memory_pool_free(inst.pkt_pool,pkt); + struct ETCP_TX_ATTEMPT* attempt = last_attempt(packet); + attempt->delivered_time_us = ack_now_us-cases[i].ack_us; + attempt->sent_at_us = ack_now_us-5000; attempt->first_tx_us = attempt->sent_at_us-cases[i].send_us; + if (!cases[i].anchor) attempt->peer_rx_bytes_at_send = attempt->peer_rx_time_at_send = 0; + uint64_t bytes = cases[i].rx_bytes ? 10000+cases[i].rx_bytes : 0; + uint64_t time = cases[i].rx_bytes ? 1000000+cases[i].rx_us : 0; + etcp_tx_ack(a,packet->seq,packet->last_attempt_id,0,bytes,time); + CHECK(last_sample.delivered == 1000 && last_sample.acked_sacked == 1000); + CHECK(last_sample.interval_us == cases[i].expected && last_sample.rtt_us == 5000); + CHECK(a->bbr->min_rtt_us == 5000 && !b->delivered_bytes); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Delivery clock: case=%u ACK_us=%u send_us=%u RX=%llu/%llu anchor=%u sample=%u/%u", + i,cases[i].ack_us,cases[i].send_us,(unsigned long long)cases[i].rx_bytes, + (unsigned long long)cases[i].rx_us,cases[i].anchor,last_sample.delivered,last_sample.interval_us); + teardown(); + } +} + +/* Реальный ACK handler: RX clock независим, арифметика округляется вверх без переполнения. */ static void receiver_bandwidth_bounds(void) { const struct { uint32_t delivered, local_us; uint64_t rx_bytes, rx_us; uint32_t expected; } cases[] = { {170502,150732,170502,186359,186359}, {1000,10000,3000,40001,13334}, - {1000,20000,5000,10000,0}, {UINT32_MAX,10000,UINT32_MAX,UINT32_MAX,UINT32_MAX}, + {1000,20000,5000,10000,2000}, {UINT32_MAX,10000,UINT32_MAX,UINT32_MAX,UINT32_MAX}, {UINT32_MAX,10000,1,UINT32_MAX,0}, {1000,10000,1,(uint64_t)UINT32_MAX+1,0} }; for (unsigned i = 0; i < sizeof(cases)/sizeof(cases[0]); i++) { @@ -1349,8 +1381,7 @@ static void receiver_bandwidth_bounds(void) { etcp_tx_ack(a,packet->seq,packet->last_attempt_id,0, (UINT64_C(1)<<60)+cases[i].rx_bytes,(UINT64_C(1)<<61)+cases[i].rx_us); CHECK(last_sample.delivered == cases[i].delivered && last_sample.acked_sacked == 1000); - if (i == 2) CHECK(last_sample.interval_us >= cases[i].local_us); - else CHECK(last_sample.interval_us == cases[i].expected); + CHECK(last_sample.interval_us == cases[i].expected); if (last_sample.interval_us) CHECK((uint64_t)last_sample.delivered*cases[i].rx_us <= (uint64_t)last_sample.interval_us*cases[i].rx_bytes); CHECK(a->bbr->min_rtt_us == last_sample.rtt_us && a->data_rtt_us == last_sample.rtt_us); @@ -1729,7 +1760,7 @@ int main(void) { ack_urgent_and_oom, ack_partial_frame_and_lifecycle, ack_sender_failure, ack_link_lifecycle, ack_due_other_link, receive_admission_and_path_acks, peer_window_queries, pacing_continuity, pacing_data_units, attempt_memory_limits, normalizer_rx_oom, normalizer_tx_oom, normalizer_callback_abort, reset_without_allocation, restart_allocation_failures, - receiver_bandwidth_bounds, receiver_point_lifecycle, receiver_point_wire, receiver_point_truncated, + receiver_delivery_clock, receiver_bandwidth_bounds, receiver_point_lifecycle, receiver_point_wire, receiver_point_truncated, compact_attempt_identity, compact_attempt_exhaustion, compact_sequence_exhaustion, pacing_first_ready, pacing_late_wakeup, pacing_rearm_failure, pacing_ready_lifecycle}; for (unsigned i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) {