From 93bf846fc4768f634c30e7de2fe16dfa455bf7ab Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 24 Oct 2026 11:46:42 +0200 Subject: [PATCH] Preserve retry priority during bounded burst measurements --- src/transport_layer/etcp.c | 7 +++++-- tests/test_etcp_windows.c | 17 +++++++++++++++-- 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/src/transport_layer/etcp.c b/src/transport_layer/etcp.c index d075dd43..e4325cd7 100644 --- a/src/transport_layer/etcp.c +++ b/src/transport_layer/etcp.c @@ -1062,7 +1062,7 @@ static struct INFLIGHT_PACKET* send_queue_select(struct ETCP_CONN* conn, struct 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; + unsigned examined = 0, retry_elsewhere = 0; for (struct ll_entry* e = conn->input_send_q->head; e; e = e->next) { struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)e; examined++; @@ -1075,7 +1075,8 @@ static struct INFLIGHT_PACKET* send_queue_select(struct ETCP_CONN* conn, struct } 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 (!link) continue; + if (burst && link != burst) { if (packet->last_attempt_id) retry_elsewhere = 1; 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, @@ -1084,6 +1085,8 @@ static struct INFLIGHT_PACKET* send_queue_select(struct ETCP_CONN* conn, struct } fresh = packet; fresh_link = link; } + // Короткая burst-серия идёт FILLER, пока готовый повтор другого пути ждёт её завершения; новые DATA его не обгоняют. + if (retry_elsewhere) fresh = NULL; *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); diff --git a/tests/test_etcp_windows.c b/tests/test_etcp_windows.c index 0be95fb5..993df7f8 100644 --- a/tests/test_etcp_windows.c +++ b/tests/test_etcp_windows.c @@ -871,6 +871,16 @@ static void send_queue_scan(void) { pkt = request(); CHECK(pkt && fresh->state == INFLIGHT_STATE_WAIT_ACK); memory_pool_free(inst.pkt_pool, pkt); ack_all(); teardown(); + setup(); fill(a, 1, 100); + conn->initialized = 0; test_clock_run_for(ua, 120); conn->initialized = 1; + fresh = enqueue(80); a->burst_active = 1; a->burst_count = 1; + pkt = request(); CHECK(pkt && pkt->link == a && pkt->burst_packet && !pkt->payload_len); + memory_pool_free(inst.pkt_pool, pkt); CHECK(fresh->state == INFLIGHT_STATE_WAIT_SEND); + pkt = request(); CHECK(pkt && pkt->link == b && fresh->state == INFLIGHT_STATE_WAIT_SEND); + memory_pool_free(inst.pkt_pool, pkt); + pkt = request(); CHECK(pkt && fresh->state == INFLIGHT_STATE_WAIT_ACK); + memory_pool_free(inst.pkt_pool, pkt); ack_all(); teardown(); + setup(); fill(a, 256, 100); conn->initialized = 0; test_clock_run_for(ua, 120); conn->initialized = 1; b->inflight_lim_bytes = 1; fresh = enqueue(80); @@ -1039,9 +1049,12 @@ static void stalled_retry_selection(void) { setup(); fill(a, 3, 100); fill(b, 3, 100); conn->initialized = 0; test_clock_run_for(ua, 120); conn->initialized = 1; CHECK(a->tx_stalled && b->tx_stalled); - pkt = request(); CHECK(pkt && pkt->payload_len && pkt->link == b); + packet = (struct INFLIGHT_PACKET*)conn->input_send_q->head; CHECK(packet); + struct ETCP_LINK* fallback = packet->last_link == a ? b : a; + struct ETCP_LINK* other = fallback == a ? b : a; + pkt = request(); CHECK(pkt && pkt->payload_len && pkt->link == fallback); memory_pool_free(inst.pkt_pool, pkt); ack_all(); - CHECK(!b->tx_stalled); etcp_tx_link_reset(a, 0); CHECK(!a->tx_stalled && !a->data_rto_streak); + CHECK(!fallback->tx_stalled); etcp_tx_link_reset(other, 0); CHECK(!other->tx_stalled && !other->data_rto_streak); teardown(); }