From c1f07ec6ae81dd9f7fb53caa21692a96b47bd676 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 22:29:31 +0200 Subject: [PATCH] Enforce per-link DATA windows without blocking ACKs --- src/transport_layer/etcp.c | 180 ++++++++++++++---------- src/transport_layer/etcp_loadbalancer.c | 116 +++------------ src/transport_layer/etcp_loadbalancer.h | 12 +- 3 files changed, 131 insertions(+), 177 deletions(-) diff --git a/src/transport_layer/etcp.c b/src/transport_layer/etcp.c index 647131ae..26dde4cd 100644 --- a/src/transport_layer/etcp.c +++ b/src/transport_layer/etcp.c @@ -48,10 +48,37 @@ static void input_send_q_cb(struct ll_queue* q, void* arg); static void wait_ack_cb(struct ll_queue* q, void* arg); static void send_ack_req_cb(struct ll_queue* q, void* arg); static void etcp_conn_process_send_queue(struct ETCP_CONN* etcp); +static void etcp_send_timer_cb(void* arg); static void etcp_connection_free_deferred(void* arg); void etcp_fire_conn_status(struct ETCP_CONN* etcp, int status); struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* etcp); +/* Очереди и сетевые события только планируют проход: синхронные callbacks не входят в отправитель повторно. */ +static void etcp_schedule_send(struct ETCP_CONN* conn, int delay_tb) { + if (!conn || conn->close_requested || conn->state == 2 || conn->tx_timer) return; + if (conn->tx_running && delay_tb == 1) return; + conn->tx_timer = uasync_set_timeout(conn->instance->ua, delay_tb, conn, etcp_send_timer_cb, "etcp_tx"); + if (!conn->tx_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot schedule sender delay=%d", conn->log_name, delay_tb); + etcp_connection_close(conn); + } +} + +void etcp_conn_wake_sender(struct ETCP_CONN* conn) { etcp_schedule_send(conn, 1); } + +/* Учитывается только последняя отправленная попытка, ещё не признанная потерянной по RTO. */ +static void etcp_inflight_release(struct INFLIGHT_PACKET* pkt) { + struct ETCP_LINK* link = pkt->last_link; + if (!link || pkt->state != INFLIGHT_STATE_WAIT_ACK) return; + if (link->inflight_bytes < pkt->ll.len || !link->inflight_packets) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] inflight accounting mismatch link=%u seq=%u bytes=%u packets=%u len=%u", + link->etcp->log_name, link->local_link_id, pkt->seq, link->inflight_bytes, link->inflight_packets, pkt->ll.len); + return; + } + link->inflight_bytes -= pkt->ll.len; + link->inflight_packets--; +} + static void clear_queue(struct ll_queue* q) { if (!q) return; struct ll_entry* pkt; @@ -475,6 +502,7 @@ void etcp_connection_close(struct ETCP_CONN* etcp) { callring_put(CR_CCLOSE_IN, (uintptr_t)etcp, 0); etcp->close_requested = 1; + if (etcp->tx_timer) { uasync_cancel_timeout(etcp->instance->ua, etcp->tx_timer); etcp->tx_timer = NULL; } etcp_down_timer_cancel(etcp); etcp_padding_stop(&etcp->padding); if (etcp->callbacks_running) { @@ -565,6 +593,7 @@ void etcp_conn_reset(struct ETCP_CONN* etcp) { } // Reset IDs DEBUG_WARN(DEBUG_CATEGORY_ETCP, "Resetting ETCP instance [%s]", etcp->log_name); + if (etcp->tx_timer) { uasync_cancel_timeout(etcp->instance->ua, etcp->tx_timer); etcp->tx_timer = NULL; } etcp->next_tx_id = 1; etcp->last_rx_id = 0; etcp->last_delivered_id = 0; @@ -620,6 +649,7 @@ void etcp_conn_reset(struct ETCP_CONN* etcp) { while (l) { l->inflight_bytes = 0; l->inflight_packets = 0; + l->cwnd_limited = 0; l->delivered_bytes = 0; l->acked_bytes = 0; l->acked_packets = 0; @@ -755,6 +785,7 @@ void etcp_conn_ready(struct ETCP_CONN* conn) { etcp_cbk_fire(conn, ETCP_CBK_EVENT_REINIT); etcp_fire_conn_status(conn, ETCP_CONN_STATUS_REINIT); } + etcp_conn_wake_sender(conn); } // Move from pending to indexed connections, fire ready callbacks @@ -922,7 +953,7 @@ void etcp_conn_on_inflight_lim_changed(struct ETCP_CONN* etcp) { uint32_t sum = 0; for (struct ETCP_LINK* tl = etcp->links; tl; tl = tl->next) sum += tl->inflight_lim_bytes; etcp->optimal_inflight = sum; -// input_queue_try_resume(etcp); + etcp_conn_wake_sender(etcp); } void etcp_stats(struct ETCP_CONN* etcp) { @@ -1054,6 +1085,13 @@ static void ack_timeout_check(struct ETCP_CONN* etcp) { // Remove from wait_ack pkt=(struct INFLIGHT_PACKET*)queue_data_get(etcp->input_wait_ack); if (!pkt) break; + etcp_inflight_release(pkt); + if (pkt->last_link && pkt->last_link->bbr) { + pkt->last_link->bbr_loss_since_ack += pkt->ll.len; + bbr_note_loss(pkt->last_link->bbr); + } + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] RTO released window seq=%u len=%u last_link=%u", + etcp->log_name, pkt->seq, pkt->ll.len, pkt->last_link ? pkt->last_link->local_link_id : 0); // Increment counters pkt->retrans_req_count++; // Optional, if used for retrans request logic pkt->last_timestamp = now; @@ -1061,6 +1099,7 @@ static void ack_timeout_check(struct ETCP_CONN* etcp) { // Change state and add to send_q for retransmission pkt->state = INFLIGHT_STATE_WAIT_SEND; queue_data_put_with_index(etcp->input_send_q, (struct ll_entry*)pkt); + etcp_conn_wake_sender(etcp); // Update stats etcp->retransmissions_count++; @@ -1076,6 +1115,7 @@ static void ack_timeout_check(struct ETCP_CONN* etcp) { current = etcp->input_wait_ack->head; } queue_resume_callback(etcp->input_wait_ack); + etcp_conn_wake_sender(etcp); } static void ack_timeout_cb(void* arg) {// сработал таймер переотправки @@ -1092,14 +1132,9 @@ static void wait_ack_cb(struct ll_queue* q, void* arg) {// добавили па } static void input_send_q_cb(struct ll_queue* q, void* arg) {// etcp->input_send_q data ready - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); - struct ETCP_CONN* etcp=(struct ETCP_CONN*)arg; - etcp_conn_process_send_queue(etcp); - if (etcp->tx_state==ETCP_TX_STATE_DATA_WAIT) { - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "resume input_send_q"); - queue_resume_callback(etcp->input_send_q); - } else DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "no resume - link busy"); - } + (void)q; + etcp_conn_wake_sender(arg); +} /* static void send_ack_req_cb(struct ll_queue* q, void* arg) {// etcp->ack_q data ready DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); @@ -1137,7 +1172,7 @@ static void ack_response_timer_cb(void* arg) {// проверяем неотпр struct ETCP_CONN* etcp = (struct ETCP_CONN*)arg; etcp->ack_resp_timer=NULL; if (etcp->ack_q->count==0) return;// нечего отправлять - etcp_conn_process_send_queue(etcp);// проталкиваем (она же должна отправлять только ack если больше ничего нет) + etcp_conn_wake_sender(etcp); // если ack все еще заняты - обновляем таймаут if (etcp->ack_q->count) etcp->ack_resp_timer = uasync_set_timeout(etcp->instance->ua, ACK_DELAY_TB, etcp, ack_response_timer_cb, "etcp_ack_resp"); // else etcp->ack_resp_timer=NULL; @@ -1153,46 +1188,37 @@ static void etcp_link_ready_callback(struct ETCP_CONN* etcp) { return;// not initialized } - if (etcp->links_up==0) { + int up = 0; + for (struct ETCP_LINK* link = etcp->links; link; link = link->next) + if (link->initialized && link->link_status) { up = 1; break; } + if (etcp->links_up==0 && up) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] etcp_link_ready_callback: links_up 0→1, calling etcp_on_up (initialized=%d tx_state=%d)", etcp->log_name, etcp->initialized, etcp->tx_state); etcp->links_up=1; etcp_on_up(etcp); } - if (etcp->tx_state!=ETCP_TX_STATE_LINK_WAIT) return; - etcp->tx_state = ETCP_TX_STATE_DATA_WAIT; - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "resume input_send_q+ack_q; link_wait->data_wait"); - queue_resume_callback(etcp->input_send_q); -// queue_resume_callback(etcp->ack_q); - if (etcp->ack_q->count && etcp->ack_resp_timer == NULL) { - ack_response_timer_cb(etcp); - } + etcp_conn_wake_sender(etcp); } // Process packets in send queue and transmit them static void etcp_conn_process_send_queue(struct ETCP_CONN* etcp) {// вызываем когда есть элемент в send_q или надо отправить ack - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); + if (etcp->tx_running || etcp->close_requested || etcp->state == 2) return; + etcp->tx_running = 1; + etcp->callbacks_running++; struct ETCP_DGRAM* dgram; - if (etcp->tx_state!=ETCP_TX_STATE_DATA_WAIT) { - struct strbuf sb = strbuf_new(); - struct ETCP_LINK* link = etcp->links; - while (link) { - strbuf_addf(&sb, "L%d%d%s ", link->recv_keepalive, link->remote_keepalive, (link->shaper_timer)?"wait":"rdy"); - link = link->next; - } - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] TX state: %d (link not ready, skip send) %s", etcp->log_name, etcp->tx_state, strbuf_str(&sb)); - strbuf_free(&sb); - return; - } - dgram = etcp_request_pkt(etcp); - while(dgram) { + while (!etcp->close_requested && (dgram = etcp_request_pkt(etcp))) { etcp_loadbalancer_send(dgram); - dgram = etcp_request_pkt(etcp); - } - if (etcp->tx_state==ETCP_TX_STATE_DATA_WAIT) { - queue_resume_callback(etcp->input_send_q); } + etcp->callbacks_running--; + etcp->tx_running = 0; +} + +/* Один запуск обслуживает накопленные события; заблокированный DATA ждёт ACK/RTO/shaper, а не следующий poll. */ +static void etcp_send_timer_cb(void* arg) { + struct ETCP_CONN* conn = arg; + conn->tx_timer = NULL; + etcp_conn_process_send_queue(conn); } // Подготовить и отправить кодограмму @@ -1201,32 +1227,45 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* etcp) { DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); if (!etcp) return NULL; if (etcp->state == 2) { DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] request_pkt on deleted conn", etcp->log_name); return NULL; } + if (etcp->close_requested || !etcp->initialized || etcp->session_required || etcp->reinit_pending) { + etcp->tx_state = ETCP_TX_STATE_LINK_WAIT; + return NULL; + } + if (!etcp->input_send_q->head) input_queue_try_push(etcp); + struct INFLIGHT_PACKET* inf_pkt = (struct INFLIGHT_PACKET*)etcp->input_send_q->head; struct ETCP_LINK* link = NULL; -// если есть активный burst — используем этот линк напрямую + // Измерительная серия не ждёт DATA-cwnd; PAYLOAD добавляется только при наличии окна. for (struct ETCP_LINK* l = etcp->links; l; l = l->next) { - if (l->burst_active) { link = l; break; } + if (l->burst_active && l->initialized && l->link_status) { link = l; break; } + } + if (link && inf_pkt && (inf_pkt->ll.len > link->inflight_lim_bytes || + link->inflight_bytes > link->inflight_lim_bytes - inf_pkt->ll.len)) { + link->cwnd_limited = 1; + inf_pkt = NULL; + } + if (!link && inf_pkt) link = etcp_loadbalancer_select_link(etcp, inf_pkt->ll.len); + if (!link && etcp->ack_q->count) { + inf_pkt = NULL; + link = etcp_loadbalancer_select_link(etcp, 0); } - if (!link) link = etcp_loadbalancer_select_link(etcp); if (!link) { + if (!inf_pkt) { etcp->tx_state = ETCP_TX_STATE_DATA_WAIT; return NULL; } + if (etcp->tx_state != ETCP_TX_STATE_LINK_WAIT) + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] DATA waits: seq=%u len=%u (link/window/shaper)", + etcp->log_name, inf_pkt->seq, inf_pkt->ll.len); etcp->tx_state=ETCP_TX_STATE_LINK_WAIT; etcp->cnt_link_wait++; - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] no link available", etcp->log_name); - return NULL;// если линков нет - ждём появления свободного + return NULL; } etcp->tx_state=ETCP_TX_STATE_DATA_WAIT; - - size_t send_q_size = queue_entry_count(etcp->input_send_q); - - if (send_q_size == 0) {// сгребаем из input_queue -// DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "input_send_q empty, check if avail input_queue -> inflight"); - input_queue_try_push(etcp); + struct ETCP_DGRAM* dgram = memory_pool_alloc(etcp->instance->pkt_pool); + if (!dgram) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot allocate TX packet; retry in 10ms", etcp->log_name); + etcp_schedule_send(etcp, 100); + return NULL; } - - - // First, check if there's a packet in input_send_q (retrans or new) -// DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "getting packet from input_send_q"); - struct INFLIGHT_PACKET* inf_pkt = (struct INFLIGHT_PACKET*)queue_data_get(etcp->input_send_q); if (inf_pkt) { + inf_pkt = (struct INFLIGHT_PACKET*)queue_data_get(etcp->input_send_q); uint64_t now=get_time_tb(); DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] send_q->wait_ack seq=%d TS=%llu", etcp->log_name, inf_pkt->seq, now); @@ -1244,15 +1283,9 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* etcp) { // Fill history (send_count is the index before this attempt) inf_pkt->send_hist[inf_pkt->send_count % 8] = link_num; - // Always subtract from previous last_link (if any) – this handles retransmission + // RTO уже освободил старое окно; здесь учитывается только новая попытка. if (inf_pkt->last_link) { inf_pkt->last_link->total_retransmissions++; - if (inf_pkt->last_link->bbr) { - inf_pkt->last_link->bbr_loss_since_ack += inf_pkt->ll.len; - bbr_note_loss(inf_pkt->last_link->bbr); - } - inf_pkt->last_link->inflight_bytes -= inf_pkt->ll.len; - inf_pkt->last_link->inflight_packets--; } // Always add to the CURRENT link (first send or retransmission) @@ -1262,7 +1295,9 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* etcp) { // BBR: сохраняем снэпшот на момент отправки inf_pkt->delivered_at_send = link->delivered_bytes; inf_pkt->inflight_at_send = link->inflight_bytes; - inf_pkt->is_app_limited = (etcp->input_queue->count == 0) ? 1 : 0; + struct PKTNORM* pn = etcp->normalizer; + inf_pkt->is_app_limited = !etcp->input_queue->count && !etcp->input_send_q->count && + (!pn || (!pn->input->count && !pn->pending && !pn->data_ptr)); // Update last_link for future ACK/retrans inf_pkt->last_link = link; @@ -1275,17 +1310,6 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* etcp) { } size_t ack_q_size = queue_entry_count(etcp->ack_q); - if (!inf_pkt && ack_q_size == 0 && !link->burst_active) { - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] no data/ack to send", etcp->log_name); - return NULL; - } - - struct ETCP_DGRAM* dgram = memory_pool_alloc(etcp->instance->pkt_pool); - if (!dgram) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] failed to allocate ETCP_DGRAM", etcp->log_name); - return NULL; - } - dgram->link = link; dgram->noencrypt_len=0; dgram->timestamp = get_current_timestamp(); @@ -1548,11 +1572,13 @@ void etcp_ack_recv(struct ETCP_CONN* etcp, uint32_t seq, uint16_t ts, uint16_t d } etcp_tx_record_ack(&etcp->instance->tx_history, get_time_us(), acked_pkt->ll.len); - // === subtract inflight from the LAST link the packet was sent on === + int cwnd_limited = acked_pkt->last_link && (acked_pkt->last_link->cwnd_limited || + acked_pkt->last_link->inflight_bytes >= acked_pkt->last_link->inflight_lim_bytes); + etcp_inflight_release(acked_pkt); + if (acked_pkt->state == INFLIGHT_STATE_WAIT_SEND) + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] late ACK seq=%u before retry, no window debit", etcp->log_name, seq); if (acked_pkt->last_link && acked_pkt->last_link->bbr) { struct ETCP_LINK* link = acked_pkt->last_link; - link->inflight_bytes -= acked_pkt->ll.len; - link->inflight_packets--; // BBR: собираем rate_sample и вызываем bbr_main uint64_t now_tb = get_time_tb(); @@ -1580,7 +1606,8 @@ void etcp_ack_recv(struct ETCP_CONN* etcp, uint32_t seq, uint16_t ts, uint16_t d uint32_t new_cwnd = old_cwnd; uint32_t new_pacing = old_pacing; bbr_main(link->bbr, &rs, &new_cwnd, &new_pacing, link->mtu, link->inflight_bytes, - (link->inflight_bytes >= link->inflight_lim_bytes)); + cwnd_limited); + link->cwnd_limited = 0; etcp_link_update_inflight_lim(link, new_cwnd); link->bbr_pacing_rate = new_pacing; link->bandwidth = (uint32_t)((uint64_t)new_pacing * 8 / 1000); @@ -1614,6 +1641,7 @@ void etcp_ack_recv(struct ETCP_CONN* etcp, uint32_t seq, uint16_t ts, uint16_t d // Try to resume sending more packets if window space opened up input_queue_try_resume(etcp); + etcp_conn_wake_sender(etcp); // DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] completed for seq=%u", etcp->log_name, seq); } diff --git a/src/transport_layer/etcp_loadbalancer.c b/src/transport_layer/etcp_loadbalancer.c index 27ca8a95..85ef5245 100644 --- a/src/transport_layer/etcp_loadbalancer.c +++ b/src/transport_layer/etcp_loadbalancer.c @@ -1,12 +1,4 @@ -// etcp_loadbalancer.c - Load Balancer Implementation (updated to match etcp_loadbalancer.h) -// Changes: -// - Replaced etcp_loadbalancer_update_after_send with etcp_loadbalancer_send, which now handles link selection, encryption/send, and shaper update. -// - Added loadbalancer_link_ready to notify ETCP_CONN when a link becomes ready (assumes ETCP_CONN has a resume_fn callback; implement in etcp.h/c if needed). -// - Updated constants to match .h (TIMEBASE_NS, SUB_MODULO, etc.). -// - Refined shaper logic for better precision and added normalization. -// - Removed duplicate update_after_send definitions; consolidated into send. -// - Added debug traces for new functions. -// - Assumed ETCP_CONN has a 'resume_fn' field: void (*resume_fn)(struct ETCP_CONN*); // Add to etcp.h struct ETCP_CONN. +// Выбор линка по индивидуальному окну DATA и pacing; ACK не занимает окно DATA. #include "etcp_loadbalancer.h" #include "../lib/debug_config.h" @@ -24,85 +16,29 @@ // Forward declarations static void shaper_timer_cb(void* arg); // Shaper wait callback -// Select link for transmission -// Algorithm: min inflight_bytes, round-robin among ties -struct ETCP_LINK* etcp_loadbalancer_select_link(struct ETCP_CONN* etcp) { - if (!etcp || !etcp->links) { - DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] invalid parameters (etcp=%p, links=%p)", - etcp ? etcp->log_name : "????→????", etcp, etcp ? etcp->links : NULL); - return NULL; - } - - // First pass: find minimum inflight_bytes among available links - uint32_t min_inflight = UINT32_MAX; - int available_count = 0; - - struct ETCP_LINK* link = etcp->links; - while (link) { - if (!link->initialized || link->link_status != 1) { - link = link->next; - continue; - } - if (loadbalancer_link_can_send(link) == 0) { - link = link->next; - continue; - } - - available_count++; - if (link->inflight_bytes < min_inflight) { - min_inflight = link->inflight_bytes; - } - link = link->next; - } - - if (available_count == 0) { - if (etcp->links) - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] no suitable link found: tmr:%s", - etcp->log_name, etcp->links->shaper_timer?"wait":"rdy"); - return NULL; - } - - // Count how many links have the minimum inflight - int min_count = 0; - link = etcp->links; - while (link) { - if (link->initialized && link->link_status == 1 && - loadbalancer_link_can_send(link) && link->inflight_bytes == min_inflight) { - min_count++; - } - link = link->next; - } - - if (min_count == 1) { - // Only one link with minimum inflight - return it directly - link = etcp->links; - while (link) { - if (link->initialized && link->link_status == 1 && - loadbalancer_link_can_send(link) && link->inflight_bytes == min_inflight) { - return link; - } - link = link->next; - } - } - - // Round-robin among links with minimum inflight - // Start from the link after last_rr_link (or from the beginning if last_rr_link is NULL) - struct ETCP_LINK* start = etcp->last_rr_link ? etcp->last_rr_link->next : etcp->links; - if (!start) start = etcp->links; - - link = start; +// Минимум inflight среди вмещающих целый DATA; при равенстве начинаем после прошлого выбранного линка. +struct ETCP_LINK* etcp_loadbalancer_select_link(struct ETCP_CONN* etcp, uint32_t data_len) { + if (!etcp || !etcp->links) return NULL; + struct ETCP_LINK* start = etcp->links; + for (struct ETCP_LINK* l = etcp->links; l; l = l->next) + if (l == etcp->last_rr_link) { start = l->next ? l->next : etcp->links; break; } + struct ETCP_LINK* best = NULL; + struct ETCP_LINK* link = start; do { - if (link->initialized && link->link_status == 1 && - loadbalancer_link_can_send(link) && link->inflight_bytes == min_inflight) { - etcp->last_rr_link = link; - return link; + if (link->initialized && link->link_status) { + int full = data_len && (data_len > link->inflight_lim_bytes || + link->inflight_bytes > link->inflight_lim_bytes - data_len); + if (full) link->cwnd_limited = 1; + else if (loadbalancer_link_can_send(link) && (!best || link->inflight_bytes < best->inflight_bytes)) best = link; } - link = link->next; - if (!link) link = etcp->links; + link = link->next ? link->next : etcp->links; } while (link != start); - - // Should not reach here, but return NULL just in case - return NULL; + if (best) { + etcp->last_rr_link = best; + DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] select link=%u DATA=%u inflight=%u/%u", etcp->log_name, + best->local_link_id, data_len, best->inflight_bytes, best->inflight_lim_bytes); + } + return best; } // New: Send dgram (select link, encrypt/send, update shaper) @@ -118,16 +54,6 @@ void etcp_loadbalancer_send(struct ETCP_DGRAM* dgram) { return; } - // Select link if not already set - if (!dgram->link) { - dgram->link = etcp_loadbalancer_select_link(etcp); - if (!dgram->link) { - DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] no link available, dropping dgram", etcp->log_name); - memory_pool_free(etcp->instance->pkt_pool, dgram); - return; - } - } - struct ETCP_LINK* link = dgram->link; size_t pkt_size = dgram->data_len + sizeof(uint16_t); // Include timestamp diff --git a/src/transport_layer/etcp_loadbalancer.h b/src/transport_layer/etcp_loadbalancer.h index c9cd364c..c093908e 100644 --- a/src/transport_layer/etcp_loadbalancer.h +++ b/src/transport_layer/etcp_loadbalancer.h @@ -1,6 +1,7 @@ /* etcp_loadbalancer — выбор доступного линка одного ETCP_CONN с учётом cwnd, inflight и pacing/shaper. * select_link возвращает заимствованный готовый линк или NULL. send шифрует/отправляет пакет и учитывает shaper, - * link_ready пробуждает packet_request_fn после снятия ограничений; все вызовы — uasync-поток соединения. */ + * link_ready вызывает link_ready_for_send_fn при готовности физического линка/shaper; + * окно пробуждает etcp_conn_wake_sender без изменения UP/DOWN. Все вызовы — uasync-поток соединения. */ #ifndef ETCP_LOADBALANCER_H #define ETCP_LOADBALANCER_H @@ -16,11 +17,10 @@ extern "C" { #define DELTA_TIME_NS 10000ULL // Delta for inactivity #define SHAPER_BURST_DELAY_TB 10ULL // 10 = 1ms burst allowance -// при формировании пакета - выбрать линк для передачи -// должен либо возвращать линк который готов забрать пакет, либо возвратить NULL если все линки заняты. -struct ETCP_LINK* etcp_loadbalancer_select_link(struct ETCP_CONN* etcp); - -// Когда линк снова ready (timer или link_ready) - loadbalancer должен вызвать ETCP_CONN->packet_request_fn(etcp) +/* Заимствованный рабочий линк с минимальным inflight, round-robin при равенстве; NULL — нет подходящего. + * data_len > 0 требует места для целого DATA, 0 выбирает линк для ACK без проверки cwnd. + * Оба режима учитывают shaper; burst обходит только pacing, индивидуальное окно DATA сохраняется. */ +struct ETCP_LINK* etcp_loadbalancer_select_link(struct ETCP_CONN* etcp, uint32_t data_len); // Передать dgram с установленным link: после отправки (включая ошибку сокета) возвращает dgram в pkt_pool. void etcp_loadbalancer_send(struct ETCP_DGRAM* dgram);