Use receiver delivery clock for ETCP bandwidth samples
This commit is contained in:
evgeny committed
1 parent
9b72a0dadf
commit
d2703b2145
3 files changed
+44
-10
No files matched your search
@@ -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++;
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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++) {
|
||||
|
||||
Reference in new issue
Block a user