From d57a729b92b624eab7227c2ec121b3570b36fcfd Mon Sep 17 00:00:00 2001 From: evgeny Date: Fri, 23 Oct 2026 10:45:39 +0200 Subject: [PATCH] Preserve bounded pacing credit across millisecond timer rounding --- src/transport_layer/etcp_loadbalancer.c | 16 ++++++++++------ tests/test_etcp_multipath.c | 6 +----- tests/test_etcp_windows.c | 15 ++++++++++++++- 3 files changed, 25 insertions(+), 12 deletions(-) diff --git a/src/transport_layer/etcp_loadbalancer.c b/src/transport_layer/etcp_loadbalancer.c index f919aa70..60ad3270 100644 --- a/src/transport_layer/etcp_loadbalancer.c +++ b/src/transport_layer/etcp_loadbalancer.c @@ -33,9 +33,13 @@ void etcp_loadbalancer_set_pacing_rate(struct ETCP_LINK* link, uint32_t rate) { if (!link || !link->etcp) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "pacing: invalid link=%p", (void*)link); return; } uint32_t old_rate = link->bbr_pacing_rate; if (rate == old_rate) return; - uint64_t now = get_time_tb(), debt = 0; + uint64_t now = get_time_tb(), debt = 0, credit = 0, fraction = 0; int was_blocked = link->shaper_timer != NULL; - if (old_rate && rate && link->shaper_load_time_tb >= now) { + if (old_rate && rate && link->shaper_load_time_tb < now) { + credit = now - link->shaper_load_time_tb; + if (credit > SHAPER_BURST_DELAY_TB) credit = SHAPER_BURST_DELAY_TB; + else fraction = link->shaper_sub_nanotime; + } else if (old_rate && rate) { debt = (link->shaper_load_time_tb - now) * SUB_MODULO + link->shaper_sub_nanotime; // debt*old_rate/rate без переполнения: остаток и обе ставки не шире uint32. uint64_t quotient = debt / rate, product = (debt % rate) * old_rate; @@ -48,8 +52,8 @@ void etcp_loadbalancer_set_pacing_rate(struct ETCP_LINK* link, uint32_t rate) { } link->bbr_pacing_rate = rate; link->bandwidth = (uint32_t)(((uint64_t)rate * 8 + 999) / 1000); - link->shaper_load_time_tb = now + debt / SUB_MODULO; - link->shaper_sub_nanotime = debt % SUB_MODULO; + link->shaper_load_time_tb = now + debt / SUB_MODULO - credit; + link->shaper_sub_nanotime = credit ? fraction : debt % SUB_MODULO; DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] pacing link=%u rate_Bps=%u->%u debt_sub=%llu deadline_tb=%llu now_tb=%llu", link->etcp->log_name, link->local_link_id, old_rate, rate, (unsigned long long)debt, (unsigned long long)link->shaper_load_time_tb, (unsigned long long)now); @@ -122,8 +126,8 @@ int etcp_loadbalancer_send(struct ETCP_DGRAM* dgram) { // Don't return here, let the function continue to free dgram at the end } else { uint64_t now_tb = get_time_tb(); - if (link->shaper_load_time_tb < now_tb) { - link->shaper_load_time_tb = now_tb; link->shaper_sub_nanotime = 0; + if (link->shaper_load_time_tb + SHAPER_BURST_DELAY_TB < now_tb) { + link->shaper_load_time_tb = now_tb - SHAPER_BURST_DELAY_TB; link->shaper_sub_nanotime = 0; } // Ставка B/s; SUB_MODULO соответствует 0.1мс, один sub равен 0.1нс. uint64_t tx_sub = ((uint64_t)pkt_size * 10000000000ULL + link->bbr_pacing_rate - 1) / link->bbr_pacing_rate; diff --git a/tests/test_etcp_multipath.c b/tests/test_etcp_multipath.c index 6f0d903a..0d5d00c0 100644 --- a/tests/test_etcp_multipath.c +++ b/tests/test_etcp_multipath.c @@ -231,11 +231,7 @@ static void snapshot(const char* phase) { 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 = conn->input_wait_ack->head; e; e = e->next) { - struct INFLIGHT_PACKET* pkt = (struct INFLIGHT_PACKET*)e; - CHECK(pkt->state == INFLIGHT_STATE_WAIT_ACK); - if (pkt->last_link == l) { occupied += pkt->ll.len; packets++; } - } + 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++) diff --git a/tests/test_etcp_windows.c b/tests/test_etcp_windows.c index c970ab4d..31693cbf 100644 --- a/tests/test_etcp_windows.c +++ b/tests/test_etcp_windows.c @@ -356,6 +356,19 @@ static void pacing_timer_failure(void) { etcp_conn_ref_free(conn); teardown(); } +/* Миллисекундное округление таймеров не должно превращать 1MB/s в 0.5MB/s. */ +static void pacing_continuity(void) { + setup(); b->link_status = 0; a->data_rtt_us = 1000000; + etcp_loadbalancer_set_pacing_rate(a, 1000000); + uint64_t started = get_time_tb(); + for (unsigned i = 0; i < 50; i++) enqueue(1000); + wait_sends(50); + uint64_t elapsed = get_time_tb() - started; + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Pacing continuity: DATA=50000 rate=1000000 elapsed_tb=%llu", (unsigned long long)elapsed); + CHECK(elapsed >= 450 && elapsed <= 650); + ack_all(); teardown(); +} + /* Отправитель не превращает pending transport в UP до первого INIT/подтверждения сессии. */ static void initial_readiness(void) { setup(); conn->state = 0; conn->initialized = conn->links_up = conn->tx_state = 0; @@ -608,7 +621,7 @@ int main(void) { grow_and_session, allocation_and_send_failure, removed_link_and_bbr, pacing_reset_close, initial_readiness, config_limits, timer_failure, immediate_send, ack_without_data_pacing, bbr_delivery_clock, bbr_probe_window, tcp_initial_window, retransmission_timeout, pacing_rate_changes, pacing_timer_failure, retry_rotation, independent_attempt_acks, - receive_admission_and_path_acks, peer_window_queries}; + receive_admission_and_path_acks, peer_window_queries, pacing_continuity}; for (unsigned i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) { cases[i](); CHECK(u_get_allocated_count() == baseline); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS ETCP window scenario %u", i);