Keep pacing in bytes per second and reprice outstanding shaper debt

src/transport_layer/etcp_loadbalancer.c

@@ -6,16 +6,57 @@
 #include <stdlib.h>
 #include <stdint.h>  // For UINT64_MAX
 #include <time.h>    // For timespec
-#include <math.h>    // For rounding
+#include <limits.h>
 #include "../lib/mem.h"
 
 // Enable comprehensive debug output for loadbalancer module
 #define DEBUG_CATEGORY_LOADBALANCER 1
-#define ALLOWED_DELTA 5 // 0.5ms allowed gap for time correction
 
 // Forward declarations
 static void shaper_timer_cb(void* arg);  // Shaper wait callback
 
+static void shaper_arm(struct ETCP_LINK* link) {
+    uint64_t now = get_time_tb();
+    if (link->shaper_timer || !link->bbr_pacing_rate || link->shaper_load_time_tb < now + SHAPER_BURST_DELAY_TB) return;
+    uint64_t delay = link->shaper_load_time_tb - now + !!link->shaper_sub_nanotime;
+    if (delay > INT_MAX) delay = INT_MAX;
+    link->shaper_timer = uasync_set_timeout(link->etcp->instance->ua, (int)delay, link, shaper_timer_cb, "etcp_shaper");
+    if (!link->shaper_timer) {
+        struct ETCP_CONN* conn = link->etcp;
+        DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot schedule pacing link=%u rate_Bps=%u delay_tb=%llu",
+            conn->log_name, link->local_link_id, link->bbr_pacing_rate, (unsigned long long)delay);
+        conn->callbacks_running++; etcp_connection_close(conn); conn->callbacks_running--;
+    }
+}
+
+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;
+    int was_blocked = link->shaper_timer != NULL;
+    if (old_rate && rate && link->shaper_load_time_tb >= now) {
+        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;
+        uint64_t tail = product / rate + !!(product % rate);
+        if (quotient > (UINT64_MAX - tail) / old_rate) debt = UINT64_MAX;
+        else debt = quotient * old_rate + tail;
+    }
+    if (link->shaper_timer) {
+        uasync_cancel_timeout(link->etcp->instance->ua, link->shaper_timer); link->shaper_timer = NULL;
+    }
+    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;
+    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);
+    shaper_arm(link);
+    if (was_blocked && !link->shaper_timer) etcp_conn_wake_sender(link->etcp);
+}
+
 // Минимум inflight среди вмещающих целый DATA; при равенстве начинаем после прошлого выбранного линка.
 struct ETCP_LINK* etcp_loadbalancer_select_link(struct ETCP_CONN* etcp, uint32_t data_len) {
     if (!etcp || !etcp->links) return NULL;
@@ -71,16 +112,16 @@ void etcp_loadbalancer_send(struct ETCP_DGRAM* dgram) {
         DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] ACK/control bypasses DATA pacer link=%u", etcp->log_name, link->local_link_id);
     } else if (link->burst_active) {
         // burst bypasses shaper — не обновляем нагрузку
-    } else if (link->bandwidth == 0) {
+    } else if (link->bbr_pacing_rate == 0) {
         DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] unlimited bandwidth, skipping shaper update", etcp->log_name);
         // Don't return here, let the function continue to free dgram at the end
     } else {
-        // Time to transmit (ns per byte = 8e6 / bw for Kbits/sec)
-        double byte_time_ns = 8000000.0 / (double)link->bandwidth;
-        uint64_t tx_time_ns = (uint64_t)(pkt_size * byte_time_ns + 0.5);  // Round nearest
-        
-        // To sub_nanotime (*10 for 0.1ns)
-        uint64_t tx_sub = tx_time_ns * 10;
+        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;
+        }
+        // Ставка 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;
         
         // Add to load
         link->shaper_sub_nanotime += tx_sub;
@@ -88,17 +129,7 @@ void etcp_loadbalancer_send(struct ETCP_DGRAM* dgram) {
         link->shaper_sub_nanotime -= carry * SUB_MODULO;
         link->shaper_load_time_tb += carry;
         
-        // Check if exceeds burst -> schedule timer
-        uint64_t now_tb=get_time_tb();
-        if (link->shaper_load_time_tb >= now_tb + SHAPER_BURST_DELAY_TB) {
-            uint64_t wait_tb = link->shaper_load_time_tb - now_tb;
-            DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] scheduling shaper timer (wait_tb=%llu) BW=%d pkt_size=%d", etcp->log_name, (unsigned long long)wait_tb, link->bandwidth, pkt_size);
-            link->shaper_timer = uasync_set_timeout(link->etcp->instance->ua, wait_tb, link, shaper_timer_cb, "shaper");
-            if (link->shaper_timer == NULL) DEBUG_ERROR(DEBUG_CATEGORY_TIMERS, "Filed to allocate timer");
-        }
-        
-        // Inactivity correction
-        if (link->shaper_load_time_tb < now_tb-ALLOWED_DELTA) link->shaper_load_time_tb = now_tb-ALLOWED_DELTA;
+        shaper_arm(link);
     }
 //    DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] updated load_tb=%llu, sub=%llu, state=%u", etcp->log_name,
 //                (unsigned long long)link->shaper_load_time_tb, (unsigned long long)link->shaper_sub_nanotime, link->shaper_timer==NULL?1:0);
@@ -159,6 +190,8 @@ static void shaper_timer_cb(void* arg) {
     if (!link) return;
     
     link->shaper_timer = NULL;
+    shaper_arm(link);
+    if (link->shaper_timer || link->etcp->close_requested) return;
     DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "link=%p now ready", link);
     
     // Notify loadbalancer that link is ready

src/transport_layer/etcp_loadbalancer.h

@@ -17,6 +17,10 @@ extern "C" {
 #define DELTA_TIME_NS 10000ULL // Delta for inactivity
 #define SHAPER_BURST_DELAY_TB 10ULL  // 10 = 1ms burst allowance
 
+/* Применить pacing в B/s, пересчитать оставшийся долг/таймер по новой ставке и разбудить готовый отправитель.
+ * 0 явно отключает pacing; положительная скорость не округляется до Kbit/s. При OOM таймера закрытие отложено. */
+void etcp_loadbalancer_set_pacing_rate(struct ETCP_LINK* link, uint32_t bytes_per_second);
+
 /* Заимствованный рабочий линк с минимальным inflight, round-robin при равенстве; NULL — нет подходящего.
  * data_len > 0 требует места для целого DATA и готовности shaper; 0 выбирает ACK без DATA-cwnd/pacing.
  * ACK уже ограничен входящим потоком/ACK_DELAY; исходящая оценка DATA не описывает полосу обратных подтверждений.

tests/test_etcp_windows.c

@@ -17,6 +17,7 @@ static struct utun_config cfg;
 static struct ETCP_CONN* conn;
 static struct ETCP_LINK *a, *b;
 static unsigned sends, fail_alloc, fail_send, fail_tx_timer, bbr_calls, saw_cwnd_limited, saw_app_limited;
+static unsigned fail_shaper_timer;
 static struct bbr_rate_sample last_sample;
 static uint32_t last_mss;
 
@@ -28,6 +29,7 @@ void* __wrap_uasync_call_soon(struct UASYNC* loop, void* arg, timeout_callback_t
 
 void* __real_uasync_set_timeout(struct UASYNC*, int, void*, timeout_callback_t, const char*);
 void* __wrap_uasync_set_timeout(struct UASYNC* loop, int delay, void* arg, timeout_callback_t callback, const char* name) {
+    if (fail_shaper_timer && !strcmp(name, "etcp_shaper")) { fail_shaper_timer--; return NULL; }
     if (fail_tx_timer && !strcmp(name, "etcp_tx")) { fail_tx_timer--; return NULL; }
     return __real_uasync_set_timeout(loop, delay, arg, callback, name);
 }
@@ -73,6 +75,7 @@ static void check_accounting(void) {
 static void setup(void) {
     memset(&inst, 0, sizeof(inst)); memset(&cfg, 0, sizeof(cfg));
     sends = fail_alloc = fail_send = fail_tx_timer = bbr_calls = saw_cwnd_limited = saw_app_limited = 0;
+    fail_shaper_timer = 0;
     ua = uasync_create(); CHECK(ua);
     inst.ua = ua; inst.node_id = 1; inst.config = &cfg; cfg.global.bbr_max_cwnd = INFLIGHT_LIM_MAX;
     inst.connections = queue_new(ua, 16, 0, 8, "windows");
@@ -296,6 +299,37 @@ static void pacing_reset_close(void) {
     teardown();
 }
 
+/* Смена ставки пересчитывает долг, а не оставляет таймер от прежней скорости. */
+static void pacing_rate_changes(void) {
+    setup(); b->link_status = 0; conn->rtt_avg_10 = 10000;
+    etcp_loadbalancer_set_pacing_rate(a, 100);
+    enqueue(1000); wait_sends(1); enqueue(500);
+    uint64_t debt = a->shaper_load_time_tb - get_time_tb();
+    DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "Pacing below 1Kbit/s: rate=%u debt_tb=%llu", a->bbr_pacing_rate, (unsigned long long)debt);
+    CHECK(a->shaper_timer && a->bbr_pacing_rate == 100 && debt >= 99000 && debt <= 103000);
+    etcp_loadbalancer_set_pacing_rate(a, 1000000);
+    CHECK(a->shaper_load_time_tb <= get_time_tb() + 20);
+    wait_sends(2); ack_all(); teardown();
+
+    setup(); b->link_status = 0; conn->rtt_avg_10 = 10000;
+    etcp_loadbalancer_set_pacing_rate(a, 1000000);
+    enqueue(1000); wait_sends(1); enqueue(500);
+    etcp_loadbalancer_set_pacing_rate(a, 100);
+    CHECK(a->shaper_timer && a->shaper_load_time_tb > get_time_tb() + 1000);
+    test_clock_run_for(ua, 200); CHECK(sends == 1 && conn->input_send_q->count == 1);
+    etcp_loadbalancer_set_pacing_rate(a, 0);
+    CHECK(!a->shaper_timer && !a->shaper_sub_nanotime);
+    wait_sends(2); ack_all(); teardown();
+}
+
+static void pacing_timer_failure(void) {
+    setup(); CHECK(etcp_conn_ref_take(conn) == 0); b->link_status = 0;
+    etcp_loadbalancer_set_pacing_rate(a, 100); fail_shaper_timer = 1;
+    enqueue(100); uasync_drain_immediate(ua);
+    CHECK(!fail_shaper_timer && conn->state == 2 && !conn->links);
+    etcp_conn_ref_free(conn); teardown();
+}
+
 /* Отправитель не превращает pending transport в UP до первого INIT/подтверждения сессии. */
 static void initial_readiness(void) {
     setup(); conn->state = 0; conn->initialized = conn->links_up = conn->tx_state = 0;
@@ -338,7 +372,7 @@ static void immediate_send(void) {
 
 /* Подтверждения входящего потока не должны ждать pacing собственного исходящего DATA. */
 static void ack_without_data_pacing(void) {
-    setup(); b->link_status = 0; a->bandwidth = 1;
+    setup(); b->link_status = 0; etcp_loadbalancer_set_pacing_rate(a, 125);
     a->shaper_timer = uasync_set_timeout(ua, 1000, a, shaper_ready, "test_shaper"); CHECK(a->shaper_timer);
     void* timer = a->shaper_timer;
     uint64_t load = a->shaper_load_time_tb, fraction = a->shaper_sub_nanotime;
@@ -418,7 +452,7 @@ int main(void) {
     void (*cases[])(void) = {selection, boundaries_ack_burst, late_ack_and_migration, lost_ack_and_shrink,
         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};
+        retransmission_timeout, pacing_rate_changes, pacing_timer_failure};
     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);