diff --git a/src/transport_layer/etcp_loadbalancer.c b/src/transport_layer/etcp_loadbalancer.c index 160f9beb..cf6fce6f 100644 --- a/src/transport_layer/etcp_loadbalancer.c +++ b/src/transport_layer/etcp_loadbalancer.c @@ -6,16 +6,57 @@ #include #include // For UINT64_MAX #include // For timespec -#include // For rounding +#include #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 diff --git a/src/transport_layer/etcp_loadbalancer.h b/src/transport_layer/etcp_loadbalancer.h index 80243262..7b4b59e6 100644 --- a/src/transport_layer/etcp_loadbalancer.h +++ b/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 не описывает полосу обратных подтверждений. diff --git a/tests/test_etcp_windows.c b/tests/test_etcp_windows.c index 53f3c771..dfa7aa6f 100644 --- a/tests/test_etcp_windows.c +++ b/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);