From abe3d4bc89abc3e0cf700ce51f29781a961f5b37 Mon Sep 17 00:00:00 2001 From: evgeny Date: Fri, 23 Oct 2026 10:14:46 +0200 Subject: [PATCH] Rotate retransmissions per fragment and commit only accepted sends --- src/transport_layer/etcp_connections.c | 6 +++++- src/transport_layer/etcp_loadbalancer.c | 22 ++++++++++++++-------- src/transport_layer/etcp_loadbalancer.h | 6 ++++-- 3 files changed, 23 insertions(+), 11 deletions(-) diff --git a/src/transport_layer/etcp_connections.c b/src/transport_layer/etcp_connections.c index 6f304d3b..8c0900b9 100644 --- a/src/transport_layer/etcp_connections.c +++ b/src/transport_layer/etcp_connections.c @@ -1101,7 +1101,7 @@ struct ETCP_LINK* etcp_link_new(struct ETCP_CONN* etcp, struct ETCP_SOCKET* conn link->delivered_bytes = 0; link->acked_bytes = 0; link->acked_packets = 0; - link->last_ack_time_tb = get_time_tb(); + link->last_ack_time_us = get_time_us(); link->bbr_pacing_rate = 0; link->bbr_loss_since_ack = 0; link->bbr = u_calloc(1, sizeof(struct bbr)); @@ -1110,11 +1110,13 @@ struct ETCP_LINK* etcp_link_new(struct ETCP_CONN* etcp, struct ETCP_SOCKET* conn return NULL; } bbr_init(link->bbr); + if (etcp_tx_link_init(link) < 0) { u_free(link->bbr); u_free(link); return NULL; } // Выделяем свободный local_link_id int free_id = etcp_find_free_local_link_id(etcp); if (free_id <= 0) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "no free local_link_id available"); + etcp_tx_link_destroy(link); u_free(link->bbr); u_free(link); return NULL; } @@ -1132,6 +1134,7 @@ struct ETCP_LINK* etcp_link_new(struct ETCP_CONN* etcp, struct ETCP_SOCKET* conn if (!is_tcp && insert_link_queue(conn, link) < 0) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "Can not insert link to socket"); + etcp_tx_link_destroy(link); u_free(link->bbr); u_free(link); return NULL; } @@ -1202,6 +1205,7 @@ void etcp_link_close(struct ETCP_LINK* link) { uasync_cancel_timeout(link->etcp->instance->ua, link->burst_resp_timer); link->burst_resp_timer = NULL; } + etcp_tx_link_destroy(link); if (!link->conn) { struct ETCP_LINK **pp = &link->etcp->links; diff --git a/src/transport_layer/etcp_loadbalancer.c b/src/transport_layer/etcp_loadbalancer.c index cf6fce6f..f919aa70 100644 --- a/src/transport_layer/etcp_loadbalancer.c +++ b/src/transport_layer/etcp_loadbalancer.c @@ -58,11 +58,11 @@ void etcp_loadbalancer_set_pacing_rate(struct ETCP_LINK* link, uint32_t rate) { } // Минимум inflight среди вмещающих целый DATA; при равенстве начинаем после прошлого выбранного линка. -struct ETCP_LINK* etcp_loadbalancer_select_link(struct ETCP_CONN* etcp, uint32_t data_len) { +struct ETCP_LINK* etcp_loadbalancer_select_link(struct ETCP_CONN* etcp, uint32_t data_len, const struct INFLIGHT_PACKET* retry) { 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; } + if (l == (retry ? retry->last_link : etcp->last_rr_link)) { start = l->next ? l->next : etcp->links; break; } struct ETCP_LINK* best = NULL; struct ETCP_LINK* link = start; do { @@ -70,7 +70,10 @@ struct ETCP_LINK* etcp_loadbalancer_select_link(struct ETCP_CONN* etcp, uint32_t 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 ((!data_len || loadbalancer_link_can_send(link)) && (!best || link->inflight_bytes < best->inflight_bytes)) best = link; + else if ((!data_len || loadbalancer_link_can_send(link)) && (retry || !best || link->inflight_bytes < best->inflight_bytes)) { + best = link; + if (retry) break; // За один круг прежний путь встретится только после всех альтернатив. + } } link = link->next ? link->next : etcp->links; } while (link != start); @@ -83,16 +86,16 @@ struct ETCP_LINK* etcp_loadbalancer_select_link(struct ETCP_CONN* etcp, uint32_t } // New: Send dgram (select link, encrypt/send, update shaper) -void etcp_loadbalancer_send(struct ETCP_DGRAM* dgram) { +int etcp_loadbalancer_send(struct ETCP_DGRAM* dgram) { if (!dgram) { DEBUG_WARN(DEBUG_CATEGORY_ETCP, "called with NULL dgram"); - return; + return -1; } struct ETCP_CONN* etcp = dgram->link ? dgram->link->etcp : NULL; if (!etcp) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "no ETCP_CONN associated with dgram (link=%p)", (void*)dgram->link); - return; + return -1; } struct ETCP_LINK* link = dgram->link; @@ -102,15 +105,17 @@ void etcp_loadbalancer_send(struct ETCP_DGRAM* dgram) { // Encrypt and send (from etcp_connections.c) int send_result = etcp_encrypt_send(dgram); + etcp_tx_complete(dgram, send_result >= 0); if (send_result < 0) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] encrypt/send failed (%d)", etcp->log_name, send_result); - // Don't return here, let the function continue to free dgram at the end + memory_pool_free(etcp->instance->pkt_pool, dgram); + return send_result; } // Update shaper after successful send if (!dgram->payload_len) { 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) { + } else if (dgram->burst_packet) { // burst bypasses shaper — не обновляем нагрузку } else if (link->bbr_pacing_rate == 0) { DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] unlimited bandwidth, skipping shaper update", etcp->log_name); @@ -135,6 +140,7 @@ void etcp_loadbalancer_send(struct ETCP_DGRAM* dgram) { // (unsigned long long)link->shaper_load_time_tb, (unsigned long long)link->shaper_sub_nanotime, link->shaper_timer==NULL?1:0); memory_pool_free(etcp->instance->pkt_pool, dgram); + return send_result; } int loadbalancer_link_can_send(struct ETCP_LINK* link) { diff --git a/src/transport_layer/etcp_loadbalancer.h b/src/transport_layer/etcp_loadbalancer.h index 7b4b59e6..f690cd1d 100644 --- a/src/transport_layer/etcp_loadbalancer.h +++ b/src/transport_layer/etcp_loadbalancer.h @@ -25,10 +25,12 @@ void etcp_loadbalancer_set_pacing_rate(struct ETCP_LINK* link, uint32_t bytes_pe * data_len > 0 требует места для целого DATA и готовности shaper; 0 выбирает ACK без DATA-cwnd/pacing. * ACK уже ограничен входящим потоком/ACK_DELAY; исходящая оценка DATA не описывает полосу обратных подтверждений. * Burst обходит только pacing, индивидуальное окно DATA сохраняется. */ -struct ETCP_LINK* etcp_loadbalancer_select_link(struct ETCP_CONN* etcp, uint32_t data_len); +/* retry != NULL: обход после last_link этого фрагмента; первая готовая альтернатива, прежний путь — последний. + * Обычные DATA/ACK передают NULL. Очереди, таймеры и проверка готовности общие для обоих режимов. */ +struct ETCP_LINK* etcp_loadbalancer_select_link(struct ETCP_CONN* etcp, uint32_t data_len, const struct INFLIGHT_PACKET* retry); // Передать dgram с установленным link: после отправки (включая ошибку сокета) возвращает dgram в pkt_pool. -void etcp_loadbalancer_send(struct ETCP_DGRAM* dgram); +int etcp_loadbalancer_send(struct ETCP_DGRAM* dgram); // >=0 принят транспортом; <0 ошибка, резерв попытки снят // Готовность физического линка/его shaper; для ACK/изменения окна использовать etcp_conn_wake_sender. void loadbalancer_link_ready(struct ETCP_LINK* link);