From ff74ac8295477ef0baa474c874c7c7514804946e Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 24 Oct 2026 11:31:46 +0200 Subject: [PATCH] Scan past blocked DATA while prioritizing sendable retries --- src/transport_layer/etcp.c | 95 +++++++++++++++++++++++++------------- src/transport_layer/etcp.h | 2 +- 2 files changed, 63 insertions(+), 34 deletions(-) diff --git a/src/transport_layer/etcp.c b/src/transport_layer/etcp.c index 4cfa8376..62f62af5 100644 --- a/src/transport_layer/etcp.c +++ b/src/transport_layer/etcp.c @@ -780,30 +780,20 @@ int etcp_int_send(struct ETCP_CONN* etcp, const void* data, uint16_t len) { -static void input_queue_try_push(struct ETCP_CONN* etcp) {// пробуем протолкнуть при отправке +static int input_queue_try_push(struct ETCP_CONN* etcp) {// подгрузить не более одного нового фрагмента DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); - - // когда очередь отправки пуста - пробуем взять новый пакет на обработку - size_t wait_ack_bytes = queue_total_bytes(etcp->input_wait_ack); - if (wait_ack_bytes <= etcp->optimal_inflight) { - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] resume input queue: inflight_bytes=%d, input_len=%d", etcp->log_name, wait_ack_bytes, etcp->input_queue->total_bytes); - if (etcp->input_queue->count > 0) input_queue_cb(etcp->input_queue, etcp); - else queue_resume_callback(etcp->input_queue);// и только когда больше нечего отправлять - забираем новый пакет - } + if (etcp->unacked_bytes >= etcp->optimal_inflight) return 0; + if (etcp->input_queue->count > 0) { input_queue_cb(etcp->input_queue, etcp); return 1; } + queue_resume_callback(etcp->input_queue); + return 0; } static void input_queue_try_resume(struct ETCP_CONN* etcp) {// при ACK DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); - // Сперва отправим всё из очереди отправки - size_t send_q_bytes = queue_total_bytes(etcp->input_send_q); -// queue_resume_callback(etcp->input_send_q);// вызвать лишний раз resume не страшно. - if (send_q_bytes>0) return; - - // когда очередь отправки пуста - пробуем взять новый пакет на обработку - size_t wait_ack_bytes = queue_total_bytes(etcp->input_wait_ack); - if (wait_ack_bytes <= etcp->optimal_inflight) { - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] resume input queue: inflight_bytes=%d, input_len=%d", etcp->log_name, wait_ack_bytes, etcp->input_queue->total_bytes); + if (etcp->unacked_bytes < etcp->optimal_inflight) { + DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] resume input queue: unacked_bytes=%u input_len=%zu", etcp->log_name, + etcp->unacked_bytes, etcp->input_queue->total_bytes); queue_resume_callback(etcp->input_queue);// и только когда больше нечего отправлять - забираем новый пакет } } @@ -877,6 +867,12 @@ static void input_queue_cb(struct ll_queue* q, void* arg) { etcp->log_name, etcp->next_tx_id, etcp->rx_ack_till, q->count); ack_window_arm(etcp); return; } + if (q->head && (q->head->len > etcp->optimal_inflight || + etcp->unacked_bytes > etcp->optimal_inflight - q->head->len)) { + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] input held by DATA budget: unacked=%u limit=%u next_bytes=%u send_q=%d", + etcp->log_name, etcp->unacked_bytes, etcp->optimal_inflight, q->head->len, etcp->input_send_q->count); + return; + } struct ETCP_FRAGMENT* in_pkt = (struct ETCP_FRAGMENT*)queue_data_get(q); if (!in_pkt) { @@ -1054,28 +1050,58 @@ static void write_le(uint8_t* data, uint64_t value, unsigned size) { for (unsigned i = 0; i < size; i++) data[i] = (uint8_t)(value >> (i * 8)); } +/* Один просмотр без извлечения/резервов: первый готовый повтор, иначе первый готовый новый DATA. + * Сохранённый кандидат живёт весь просмотр: здесь нет callbacks, poll или изменения очереди. */ +static struct INFLIGHT_PACKET* send_queue_select(struct ETCP_CONN* conn, struct ETCP_LINK** chosen) { + struct ETCP_LINK* burst = NULL; + for (struct ETCP_LINK* l = conn->links; l; l = l->next) + if (l->burst_active && l->initialized && l->link_status) { burst = l; break; } + struct INFLIGHT_PACKET* fresh = NULL; + struct ETCP_LINK* fresh_link = NULL; + unsigned examined = 0; + for (struct ll_entry* e = conn->input_send_q->head; e; e = e->next) { + struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)e; + examined++; + if (!packet->last_attempt_id && fresh) continue; + struct ETCP_LINK* link; + if (burst && !packet->last_attempt_id) { + if (conn->active_attempts >= 2 * MAX_INFLIGHT_SIZE || burst->attempts->count >= MAX_INFLIGHT_SIZE || + packet->ll.len > burst->inflight_lim_bytes || burst->inflight_bytes > burst->inflight_lim_bytes - packet->ll.len) { + burst->cwnd_limited = 1; continue; + } + link = burst; + } else link = etcp_loadbalancer_select_link(conn, packet->ll.len, packet->last_attempt_id ? packet : NULL); + if (!link || (burst && link != burst)) continue; + if (packet->last_attempt_id) { + *chosen = link; + DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] send scan examined=%u retry_seq=%u link=%u", conn->log_name, + examined, packet->seq, link->local_link_id); + return packet; + } + fresh = packet; fresh_link = link; + } + *chosen = fresh ? fresh_link : burst; + DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] send scan examined=%u fresh_seq=%u link=%u", conn->log_name, examined, + fresh ? fresh->seq : 0, *chosen ? (*chosen)->local_link_id : 0); + return fresh; +} + struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* conn) { if (!conn || conn->state == 2 || conn->close_requested) return NULL; if (!conn->initialized || conn->session_required || conn->reinit_pending) { if (conn->tx_state) conn->tx_state = ETCP_TX_STATE_LINK_WAIT; return NULL; } - if (!conn->input_send_q->head) input_queue_try_push(conn); - // Callback input_queue мог сменить эпоху при исчерпании seq или OOM; старый проход отправки прекращается. - if (conn->close_requested || conn->state == 2 || !conn->initialized || conn->session_required || conn->reinit_pending) return NULL; - struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)conn->input_send_q->head; struct ETCP_LINK* link = NULL; - for (struct ETCP_LINK* candidate = conn->links; candidate; candidate = candidate->next) - if (candidate->burst_active && candidate->initialized && candidate->link_status) { link = candidate; break; } - if (link && packet) { - if (packet->last_attempt_id && etcp_loadbalancer_select_link(conn, packet->ll.len, packet) != link) packet = NULL; - if (packet && (conn->active_attempts >= 2 * MAX_INFLIGHT_SIZE || link->attempts->count >= MAX_INFLIGHT_SIZE || - packet->ll.len > link->inflight_lim_bytes || - link->inflight_bytes > link->inflight_lim_bytes - packet->ll.len)) { - link->cwnd_limited = 1; packet = NULL; - } + struct INFLIGHT_PACKET* packet = send_queue_select(conn, &link); + if (!packet) { + uint32_t reset = conn->reset_count; + int pulled = input_queue_try_push(conn); + // Подгрузка может сменить эпоху/закрыть conn либо удалить burst-линк: прежние кандидаты больше не используются. + if (conn->close_requested || conn->state == 2 || conn->reset_count != reset || !conn->initialized || + conn->session_required || conn->reinit_pending) return NULL; + if (pulled) packet = send_queue_select(conn, &link); } - if (!link && packet) link = etcp_loadbalancer_select_link(conn, packet->ll.len, packet->last_attempt_id ? packet : NULL); if (!link) { for (struct ETCP_LINK* candidate = conn->links; candidate; candidate = candidate->next) if (candidate->initialized && candidate->link_status && @@ -1086,7 +1112,10 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* conn) { if (!link && (conn->ack_till_pending || conn->ack_request_pending)) { link = etcp_loadbalancer_select_link(conn, 0, NULL); packet = NULL; } - if (!link) { conn->tx_state = packet ? ETCP_TX_STATE_LINK_WAIT : ETCP_TX_STATE_DATA_WAIT; return NULL; } + if (!link) { + conn->tx_state = conn->input_send_q->count || conn->input_queue->count ? ETCP_TX_STATE_LINK_WAIT : ETCP_TX_STATE_DATA_WAIT; + return NULL; + } conn->tx_state = ETCP_TX_STATE_DATA_WAIT; int max_enc = link->mtu - WIRE_OVERHEAD(link->remote_addr.ss_family); if (max_enc > PACKET_DATA_SIZE) max_enc = PACKET_DATA_SIZE; diff --git a/src/transport_layer/etcp.h b/src/transport_layer/etcp.h index 57c3141c..4ff1ea63 100644 --- a/src/transport_layer/etcp.h +++ b/src/transport_layer/etcp.h @@ -127,7 +127,7 @@ struct ETCP_CONN { // Links (channels) - linked list struct ETCP_LINK* links; - struct ETCP_LINK* last_rr_link; // последний линк, выбранный round-robin + struct ETCP_LINK* last_rr_link; // последняя принятая DATA-отправка; пробный выбор/ACK курсор не меняет uint8_t last_assigned_link_id; // последний выданный link_id (для циклического перебора) struct ETCP_PADDING padding; // диапазон и независимый таймер всех линков этого соединения