jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Enforce per-link DATA windows without blocking ACKs

Browse Source
This commit is contained in:
evgeny committed 2026-10-22 22:29:31 +02:00
1 parent 50d5486a9f
commit c1f07ec6ae
3 files changed
+131 -177
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/transport_layer/etcp.c
+104 -76
View File
@@ -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);
}
src/transport_layer/etcp_loadbalancer.c
+21 -95
View File
@@ -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
src/transport_layer/etcp_loadbalancer.h
+6 -6
View File
@@ -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);