From c1b0772071944bae10f9010fc4f6cc46e87344cb Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 24 Oct 2026 13:55:48 +0200 Subject: [PATCH] Honor absolute ACK deadlines across wakes and piggyback --- src/transport_layer/etcp.c | 65 ++++++++++++++++++++++++++++------- src/transport_layer/etcp_tx.c | 5 +-- 2 files changed, 55 insertions(+), 15 deletions(-) diff --git a/src/transport_layer/etcp.c b/src/transport_layer/etcp.c index 3faa5f56..bd048197 100644 --- a/src/transport_layer/etcp.c +++ b/src/transport_layer/etcp.c @@ -29,7 +29,6 @@ // Constants from spec (adjusted for completeness) #define MAX_INFLIGHT_BYTES 65536 // Initial window -#define ACK_DELAY_TB 20 // ACK timer delay (2ms in 0.1ms units) #define BURST_DELAY_FACTOR 4 // Delay before burst #define BURST_SIZE 5 // Packets in burst (1 delayed + 4 burst) #define RTT_HISTORY_SIZE 10 // For jitter calc @@ -56,7 +55,7 @@ static void etcp_schedule_send(struct ETCP_CONN* conn, int delay_tb) { if (!conn || conn->close_requested || conn->state == 2 || conn->tx_token) return; if (!conn->initialized || conn->session_required || conn->reinit_pending) return; if (conn->tx_running && !delay_tb) return; - int work = conn->input_send_q->count || conn->input_queue->count || conn->ack_till_pending || conn->ack_request_pending; + int work = conn->input_send_q->count || conn->input_queue->count || conn->ack_till_deadline_us || conn->ack_request_pending; for (struct ETCP_LINK* link = conn->links; !work && link; link = link->next) work = link->burst_active || link->ack_reply_pending || (link->ack_q && link->ack_q->count); if (!work) return; @@ -463,6 +462,7 @@ void etcp_connection_close(struct ETCP_CONN* etcp) { // Cancel active timers if (etcp->ack_resp_timer) { uasync_cancel_timeout(etcp->instance->ua, etcp->ack_resp_timer); etcp->ack_resp_timer = NULL; } + etcp->ack_resp_deadline_us = etcp->ack_till_deadline_us = 0; if (etcp->ack_window_timer) { uasync_cancel_timeout(etcp->instance->ua, etcp->ack_window_timer); etcp->ack_window_timer = NULL; } routing_del_conn(etcp); @@ -522,11 +522,12 @@ void etcp_conn_reset(struct ETCP_CONN* conn) { for (struct ETCP_LINK* link = conn->links; link; link = link->next) etcp_tx_link_reset(link, 0); etcp_cancel_send(conn); if (conn->ack_resp_timer) { uasync_cancel_timeout(conn->instance->ua, conn->ack_resp_timer); conn->ack_resp_timer = NULL; } + conn->ack_resp_deadline_us = 0; if (conn->ack_window_timer) { uasync_cancel_timeout(conn->instance->ua, conn->ack_window_timer); conn->ack_window_timer = NULL; } queue_clear(conn->input_queue); queue_clear(conn->output_queue); queue_clear(conn->input_send_q); queue_clear(conn->input_wait_ack); queue_clear(conn->recv_q); if (conn->normalizer) pn_reset(conn->normalizer); - conn->ack_till_pending = conn->ack_request_pending = 0; + conn->ack_till_deadline_us = conn->ack_request_pending = 0; conn->next_tx_id = 1; conn->last_rx_id = conn->last_delivered_id = conn->rx_ack_till = 0; conn->got_initial_pkt = 0; conn->unacked_bytes = conn->rtt_last = conn->rtt_avg_10 = conn->jitter = conn->last_rtt_cb_time = 0; @@ -962,16 +963,48 @@ void etcp_on_link_down(struct ETCP_CONN* etcp, struct ETCP_LINK* down_link) { static void ack_response_timer_cb(void* arg) { struct ETCP_CONN* conn = arg; - conn->ack_resp_timer = NULL; - etcp_conn_wake_sender(conn); + DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] ACK timer fired due_us=%llu now_us=%llu", conn->log_name, + (unsigned long long)conn->ack_resp_deadline_us, (unsigned long long)get_time_us()); + conn->ack_resp_timer = NULL; conn->ack_resp_deadline_us = 0; + etcp_conn_schedule_ack(conn); } -static void ack_schedule(struct ETCP_CONN* conn) { - if (conn->ack_resp_timer || conn->close_requested) return; - conn->ack_resp_timer = uasync_set_timeout(conn->instance->ua, ACK_DELAY_TB, conn, ack_response_timer_cb, "etcp_ack_resp"); +/* Один таймер обслуживает ближайший абсолютный срок: новый RX не отодвигает старые ACK. */ +void etcp_conn_schedule_ack(struct ETCP_CONN* conn) { + if (!conn) return; + uint64_t now = get_time_us(), deadline = 0; + if (!conn->close_requested && conn->state != 2 && conn->initialized && !conn->session_required && !conn->reinit_pending) { + deadline = conn->ack_till_deadline_us; + for (struct ETCP_LINK* link = conn->links; link; link = link->next) { + if (!link->initialized || !link->link_status) continue; + if (link->ack_reply_pending) { deadline = now; break; } + const struct ACK_PACKET* first = link->ack_q ? (struct ACK_PACKET*)link->ack_q->head : NULL; + if (first && (!deadline || first->received_at_us + ETCP_ACK_DELAY_US < deadline)) + deadline = first->received_at_us + ETCP_ACK_DELAY_US; + } + } + if (conn->ack_resp_timer && conn->ack_resp_deadline_us == deadline && deadline > now) return; + if (conn->ack_resp_timer) { + DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] ACK timer replace old_us=%llu next_us=%llu", conn->log_name, + (unsigned long long)conn->ack_resp_deadline_us, (unsigned long long)deadline); + uasync_cancel_timeout(conn->instance->ua, conn->ack_resp_timer); conn->ack_resp_timer = NULL; + } + conn->ack_resp_deadline_us = 0; + if (!deadline) return; + if (deadline <= now) { etcp_conn_wake_sender(conn); return; } + int delay = (int)((deadline - now + 99) / 100); + conn->ack_resp_timer = uasync_set_timeout(conn->instance->ua, delay, conn, ack_response_timer_cb, "etcp_ack_resp"); if (!conn->ack_resp_timer) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] ACK delay timer allocation failed, requesting immediate send", conn->log_name); + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] ACK timer OOM due_us=%llu delay_tb=%d; forcing urgent ACK", conn->log_name, + (unsigned long long)deadline, delay); + for (struct ETCP_LINK* link = conn->links; link; link = link->next) + if (link->initialized && link->link_status && link->ack_q && link->ack_q->count) link->ack_reply_pending = 1; + if (conn->ack_till_deadline_us) conn->ack_till_deadline_us = now; etcp_conn_wake_sender(conn); + } else { + conn->ack_resp_deadline_us = deadline; + DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] ACK timer arm due_us=%llu now_us=%llu delay_tb=%d", conn->log_name, + (unsigned long long)deadline, (unsigned long long)now, delay); } } @@ -1099,6 +1132,8 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* conn) { if (conn->tx_state) conn->tx_state = ETCP_TX_STATE_LINK_WAIT; return NULL; } + etcp_conn_schedule_ack(conn); + if (conn->close_requested) return NULL; struct ETCP_LINK* link = NULL; struct INFLIGHT_PACKET* packet = send_queue_select(conn, &link); if (!packet) { @@ -1110,13 +1145,16 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* conn) { if (pulled) packet = send_queue_select(conn, &link); } if (!link) { + uint64_t now = get_time_us(); for (struct ETCP_LINK* candidate = conn->links; candidate; candidate = candidate->next) if (candidate->initialized && candidate->link_status && - (candidate->ack_reply_pending || (candidate->ack_q && candidate->ack_q->count))) { + (candidate->ack_reply_pending || (candidate->ack_q && candidate->ack_q->head && + now >= ((struct ACK_PACKET*)candidate->ack_q->head)->received_at_us + ETCP_ACK_DELAY_US))) { link = candidate; packet = NULL; break; } } - if (!link && (conn->ack_till_pending || conn->ack_request_pending)) { + if (!link && (conn->ack_request_pending || + (conn->ack_till_deadline_us && conn->ack_till_deadline_us <= get_time_us()))) { link = etcp_loadbalancer_select_link(conn, 0, NULL); packet = NULL; } if (!link) { @@ -1272,7 +1310,8 @@ void etcp_output_try_assembly(struct ETCP_CONN* etcp) { } if (delivered_count > 0 && !etcp->close_requested && !etcp->reinit_pending) { - etcp->ack_till_pending = 1; ack_schedule(etcp); + if (!etcp->ack_till_deadline_us) etcp->ack_till_deadline_us = get_time_us() + ETCP_ACK_DELAY_US; + etcp_conn_schedule_ack(etcp); DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] delivered %u fragments (%u bytes), till=%u recv_q=%d", etcp->log_name, delivered_count, delivered_bytes, etcp->last_delivered_id, etcp->recv_q->count); } @@ -1475,7 +1514,7 @@ void etcp_conn_input(struct ETCP_DGRAM* pkt) { pkt->link->rx_data_bytes += payload_len; pkt->link->rx_data_time_us = get_time_us(); } etcp->got_initial_pkt = 1; - if (etcp_tx_queue_ack(pkt->link, seq, attempt_id) == 0) ack_schedule(etcp); + if (etcp_tx_queue_ack(pkt->link, seq, attempt_id) == 0) etcp_conn_schedule_ack(etcp); DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] accepted DATA seq=%u attempt=%llu path=%u bytes=%u duplicate=%d", etcp->log_name, seq, (unsigned long long)attempt_no, pkt->link->local_link_id, payload_len, duplicate); if (distance == 1) etcp_output_try_assembly(etcp); diff --git a/src/transport_layer/etcp_tx.c b/src/transport_layer/etcp_tx.c index b6242ad3..0ee6e65d 100644 --- a/src/transport_layer/etcp_tx.c +++ b/src/transport_layer/etcp_tx.c @@ -186,16 +186,17 @@ struct ETCP_TX_ATTEMPT* etcp_tx_prepare(struct ETCP_LINK* link, struct INFLIGHT_ } void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success) { + struct ETCP_CONN* conn = dgram->link->etcp; if (success) { dgram->link->ack_reply_pending = 0; - dgram->link->etcp->ack_till_pending = 0; + conn->ack_till_deadline_us = 0; if (dgram->ack_request) dgram->link->etcp->ack_request_pending = 0; } + etcp_conn_schedule_ack(conn); struct ETCP_TX_ATTEMPT* attempt = dgram->attempt; if (!attempt) return; dgram->attempt = NULL; struct ETCP_LINK* link = dgram->link; - struct ETCP_CONN* conn = link->etcp; struct INFLIGHT_PACKET* packet = logical_find(conn, attempt->seq); if (!success) { attempt_retire(link, attempt, 0, 0); return; } if (!packet || packet->state != INFLIGHT_STATE_WAIT_SEND) {