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

Honor absolute ACK deadlines across wakes and piggyback

Browse Source
This commit is contained in:
evgeny committed 2026-10-24 14:43:03 +02:00
1 parent 159237a70c
commit c1b0772071
2 files changed
+55 -15
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/transport_layer/etcp.c
+52 -13
View File
@@ -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);
src/transport_layer/etcp_tx.c
+3 -2
View File
@@ -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) {