diff --git a/src/transport_layer/etcp.c b/src/transport_layer/etcp.c index eda9b6e3..99cad98a 100644 --- a/src/transport_layer/etcp.c +++ b/src/transport_layer/etcp.c @@ -29,10 +29,6 @@ // Constants from spec (adjusted for completeness) #define MAX_INFLIGHT_BYTES 65536 // Initial window -#define RETRANS_K1 32 // RTT multiplier for retrans timeout -#define RETRANS_K2 32 // Jitter multiplier -#define RETRANS_MIN_TB 100 // Минимальный RTO 10мс, в единицах 0.1мс -#define RETRANS_MAX_TB 10000 // Максимальный RTO 1с, в единицах 0.1мс #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) @@ -47,11 +43,11 @@ static void input_queue_cb(struct ll_queue* q, void* arg); static void etcp_link_ready_callback(struct ETCP_CONN* etcp); static void input_send_q_cb(struct ll_queue* q, void* arg); -static void wait_ack_cb(struct ll_queue* q, void* arg); static void etcp_conn_process_send_queue(struct ETCP_CONN* etcp); static void etcp_send_timer_cb(void* arg); static void etcp_connection_free_deferred(void* arg); static void etcp_connection_close_deferred(void* arg); +static void ack_window_arm(struct ETCP_CONN* conn); void etcp_fire_conn_status(struct ETCP_CONN* etcp, int status); struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* etcp); @@ -60,8 +56,9 @@ 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_q->count; - for (struct ETCP_LINK* link = conn->links; !work && link; link = link->next) work = link->burst_active; + int work = conn->input_send_q->count || conn->input_queue->count || conn->ack_till_pending || conn->ack_request_pending; + for (struct ETCP_LINK* link = conn->links; !work && link; link = link->next) + work = link->burst_active || (link->ack_q && link->ack_q->count); if (!work) return; conn->tx_delayed = delay_tb != 0; conn->tx_token = delay_tb ? uasync_set_timeout(conn->instance->ua, delay_tb, conn, etcp_send_timer_cb, "etcp_tx") : @@ -88,18 +85,6 @@ static void etcp_cancel_send(struct ETCP_CONN* conn) { conn->tx_token = NULL; } -/* Учитывается только последняя отправленная попытка, ещё не признанная потерянной по RTO. */ -static void etcp_inflight_release(struct INFLIGHT_PACKET* pkt) { - struct ETCP_LINK* link = pkt->last_link; - if (!link || pkt->state != INFLIGHT_STATE_WAIT_ACK) return; - if (link->inflight_bytes < pkt->ll.len || !link->inflight_packets) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] inflight accounting mismatch link=%u seq=%u bytes=%u packets=%u len=%u", - link->etcp->log_name, link->local_link_id, pkt->seq, link->inflight_bytes, link->inflight_packets, pkt->ll.len); - abort(); // Повреждённый учёт нельзя продолжать с обёрнутыми/неверными счётчиками. - } - link->inflight_bytes -= pkt->ll.len; - link->inflight_packets--; -} static void clear_queue(struct ll_queue* q) { if (!q) return; @@ -309,8 +294,9 @@ struct ETCP_CONN* etcp_connection_create(struct UTUN_INSTANCE* instance, char* n etcp->input_send_q = queue_new(instance->ua, INFLIGHT_INITIAL_HASH_SIZE, 0, 4, "input_send_q"); // Hash for send_q etcp->input_wait_ack = queue_new(instance->ua, INFLIGHT_INITIAL_HASH_SIZE, 0, 4, "input_wait_ack"); // Hash for wait_ack etcp->recv_q = queue_new(instance->ua, INFLIGHT_INITIAL_HASH_SIZE, 0, 4, "recv_q"); // Hash for recv_q - etcp->ack_q = queue_new(instance->ua, INFLIGHT_INITIAL_HASH_SIZE, 0, 4, "ack_q"); etcp->inflight_pool = memory_pool_init(sizeof(struct INFLIGHT_PACKET), "inflight_pool"); + etcp->attempt_pool = memory_pool_init(sizeof(struct ETCP_TX_ATTEMPT), "attempt_pool"); + etcp->next_attempt_id = 1; etcp->io_pool = memory_pool_init(sizeof(struct ETCP_FRAGMENT), "io_pool"); etcp->max_inflight = (uint32_t)instance->config->global.bbr_max_cwnd; if (instance->config->global.bbr_max_cwnd < INFLIGHT_LIM_MIN) { @@ -331,7 +317,7 @@ struct ETCP_CONN* etcp_connection_create(struct UTUN_INSTANCE* instance, char* n snprintf(etcp->log_name, sizeof(etcp->log_name), "%04X->???? [%s]", (uint16_t)instance->node_id, etcp->name); if (!etcp->input_queue || !etcp->output_queue || !etcp->transit_queues || - !etcp->input_send_q || !etcp->recv_q || !etcp->ack_q || + !etcp->input_send_q || !etcp->recv_q || !etcp->attempt_pool || !etcp->input_wait_ack || !etcp->inflight_pool || !etcp->io_pool) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "error - closing - input:%p output:%p send:%p wait:%x pool:%p", etcp->input_queue, etcp->output_queue, etcp->input_send_q, etcp->input_wait_ack, etcp->inflight_pool); @@ -359,8 +345,6 @@ struct ETCP_CONN* etcp_connection_create(struct UTUN_INSTANCE* instance, char* n // Set input queue callback queue_set_callback(etcp->input_queue, input_queue_cb, etcp); queue_set_callback(etcp->input_send_q, input_send_q_cb, etcp); - queue_set_callback(etcp->input_wait_ack, wait_ack_cb, etcp); -// queue_set_callback(etcp->ack_q, send_ack_req_cb, etcp);// отправляем через таймаут etcp->link_ready_for_send_fn = etcp_link_ready_callback; @@ -488,12 +472,12 @@ static void etcp_connection_free_resources(struct ETCP_CONN* etcp) { drain_and_free_queue(&etcp->input_send_q); drain_and_free_queue(&etcp->input_wait_ack); drain_and_free_queue(&etcp->recv_q); - drain_and_free_queue(&etcp->ack_q); // Free callback chain { struct etcp_cbk_entry* cbe = etcp->cbks; while (cbe) { struct etcp_cbk_entry* n = cbe->next; u_free(cbe); cbe = n; } etcp->cbks = NULL; } if (etcp->inflight_pool) { memory_pool_destroy(etcp->inflight_pool); etcp->inflight_pool = NULL; } + if (etcp->attempt_pool) { memory_pool_destroy(etcp->attempt_pool); etcp->attempt_pool = NULL; } if (etcp->io_pool) { memory_pool_destroy(etcp->io_pool); etcp->io_pool = NULL; } u_free(etcp->name); @@ -564,8 +548,8 @@ void etcp_connection_close(struct ETCP_CONN* etcp) { DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTION, "[%s] close: state=2 conn=%p peer=%016llx (DELETE fired)", etcp->log_name, (void*)etcp, (unsigned long long)etcp->peer_node_id); // Cancel active timers - if (etcp->retrans_timer) { uasync_cancel_timeout(etcp->instance->ua, etcp->retrans_timer); etcp->retrans_timer = NULL; } if (etcp->ack_resp_timer) { uasync_cancel_timeout(etcp->instance->ua, etcp->ack_resp_timer); etcp->ack_resp_timer = NULL; } + if (etcp->ack_window_timer) { uasync_cancel_timeout(etcp->instance->ua, etcp->ack_window_timer); etcp->ack_window_timer = NULL; } routing_del_conn(etcp); etcp_router_transit_queues_destroy(etcp); @@ -621,6 +605,10 @@ void etcp_conn_reset(struct ETCP_CONN* etcp) { // Reset IDs DEBUG_WARN(DEBUG_CATEGORY_ETCP, "Resetting ETCP instance [%s]", etcp->log_name); etcp_cancel_send(etcp); + for (struct ETCP_LINK* link = etcp->links; link; link = link->next) etcp_tx_link_reset(link, 0); + etcp_cancel_send(etcp); + etcp->ack_till_pending = etcp->ack_request_pending = 0; + if (etcp->ack_window_timer) { uasync_cancel_timeout(etcp->instance->ua, etcp->ack_window_timer); etcp->ack_window_timer = NULL; } etcp->next_tx_id = 1; etcp->last_rx_id = 0; etcp->last_delivered_id = 0; @@ -658,14 +646,12 @@ void etcp_conn_reset(struct ETCP_CONN* etcp) { clear_queue(etcp->input_send_q); clear_queue(etcp->input_wait_ack); clear_queue(etcp->recv_q); - clear_queue(etcp->ack_q); // clear_queue leaves callback_suspended=1, resume to prevent deadlock queue_resume_callback(etcp->input_queue); queue_resume_callback(etcp->input_send_q); queue_resume_callback(etcp->input_wait_ack); queue_resume_callback(etcp->output_queue); - queue_resume_callback(etcp->ack_q); if (etcp->normalizer && etcp->normalizer->input) { clear_queue(etcp->normalizer->input); queue_resume_callback(etcp->normalizer->input); @@ -680,16 +666,12 @@ void etcp_conn_reset(struct ETCP_CONN* etcp) { l->delivered_bytes = 0; l->acked_bytes = 0; l->acked_packets = 0; - l->last_ack_time_tb = get_time_tb(); - l->first_tx_time_tb = 0; + l->last_ack_time_us = get_time_us(); + l->first_tx_time_us = 0; l = l->next; } // Cancel active timers to prevent memory leaks - if (etcp->retrans_timer) { - uasync_cancel_timeout(etcp->instance->ua, etcp->retrans_timer); - etcp->retrans_timer = NULL; - } if (etcp->ack_resp_timer) { uasync_cancel_timeout(etcp->instance->ua, etcp->ack_resp_timer); etcp->ack_resp_timer = NULL; @@ -707,7 +689,6 @@ void etcp_conn_reset(struct ETCP_CONN* etcp) { queue_resume_callback(etcp->input_send_q); queue_resume_callback(etcp->input_wait_ack); queue_resume_callback(etcp->recv_q); - queue_resume_callback(etcp->ack_q); etcp->tx_state=ETCP_TX_STATE_DATA_WAIT; @@ -997,8 +978,9 @@ void etcp_stats(struct ETCP_CONN* etcp) { queue_entry_count(etcp->input_send_q), queue_total_bytes(etcp->input_send_q)); DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] input_wait_ack: %zu pkts, %zu bytes", etcp->log_name, queue_entry_count(etcp->input_wait_ack), queue_total_bytes(etcp->input_wait_ack)); - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] ack_q: %zu pkts", etcp->log_name, - queue_entry_count(etcp->ack_q)); + for (struct ETCP_LINK* link = etcp->links; link; link = link->next) + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] path=%u attempts=%d ACK_queued=%d", etcp->log_name, + link->local_link_id, link->attempts ? link->attempts->count : 0, link->ack_q ? link->ack_q->count : 0); DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] recv_q: %zu pkts", etcp->log_name, queue_entry_count(etcp->recv_q)); DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] output_queue: %zu pkts, %zu bytes", etcp->log_name, @@ -1036,7 +1018,7 @@ static void input_queue_cb(struct ll_queue* q, void* arg) { if ((uint32_t)(etcp->next_tx_id - etcp->rx_ack_till) > MAX_INFLIGHT_SIZE) { DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] peer receive window full: next=%u cumulative_ack=%u queued=%d", etcp->log_name, etcp->next_tx_id, etcp->rx_ack_till, q->count); - return; + ack_window_arm(etcp); return; } struct ETCP_FRAGMENT* in_pkt = (struct ETCP_FRAGMENT*)queue_data_get(q); @@ -1090,77 +1072,6 @@ static void input_queue_cb(struct ll_queue* q, void* arg) { DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] nextloop, input_queue size=%d ", etcp->log_name, q->count); } -static void ack_timeout_cb(void* arg); -static void ack_timeout_check(struct ETCP_CONN* etcp) { - uint64_t now = get_time_tb(); - int32_t raw_timeout = (etcp->rtt_avg_10 * RETRANS_K1 + etcp->jitter * RETRANS_K2) / 16; - int32_t timeout = raw_timeout; - if (timeout < RETRANS_MIN_TB) timeout = RETRANS_MIN_TB; - if (timeout > RETRANS_MAX_TB) timeout = RETRANS_MAX_TB; - - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] RTO: raw_tb=%d timeout_tb=%d bounds_tb=%d..%d RTT_tb=%u jitter_tb=%u now_tb=%llu", - etcp->log_name, raw_timeout, timeout, RETRANS_MIN_TB, RETRANS_MAX_TB, - etcp->rtt_avg_10, etcp->jitter, (unsigned long long)now); - - - struct ll_entry* current; - current = etcp->input_wait_ack->head; - while (current) { - struct INFLIGHT_PACKET* pkt = (struct INFLIGHT_PACKET*)current; - - int64_t elapsed = now - pkt->last_timestamp; - if (elapsed > timeout) { - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] timeout for seq=%u, elapsed=%lld, now=%llu, timeout=%llu, send_count=%u. Moving: wait_ack -> send_q", - etcp->log_name, pkt->seq, (unsigned long long)elapsed, (unsigned long long)now, (unsigned long long)timeout, pkt->send_count); - - // Remove from wait_ack - pkt=(struct INFLIGHT_PACKET*)queue_data_get(etcp->input_wait_ack); - if (!pkt) break; - etcp_inflight_release(pkt); - if (pkt->last_link && pkt->last_link->bbr) { - pkt->last_link->bbr_loss_since_ack += pkt->ll.len; - bbr_note_loss(pkt->last_link->bbr); - } - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] RTO released window seq=%u len=%u last_link=%u", - etcp->log_name, pkt->seq, pkt->ll.len, pkt->last_link ? pkt->last_link->local_link_id : 0); - // Increment counters - pkt->retrans_req_count++; // Optional, if used for retrans request logic - pkt->last_timestamp = now; - - // Change state and add to send_q for retransmission - pkt->state = INFLIGHT_STATE_WAIT_SEND; - queue_data_put_with_index(etcp->input_send_q, (struct ll_entry*)pkt); - etcp_conn_wake_sender(etcp); - - // Update stats - etcp->retransmissions_count++; - } - else {// не надо до конца сканировать - они уже сортированы по таймстемпу т.к. очередь fifo, а timestamp = время добавления в очередь = время отправки - // shedule timer - int64_t next_timeout=timeout - elapsed; - if (next_timeout<0) next_timeout=0; - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] retransmission timer set for %llu units", etcp->log_name, next_timeout); - etcp->retrans_timer = uasync_set_timeout(etcp->instance->ua, next_timeout+10, etcp, ack_timeout_cb, "etcp_retrans"); - return; - } - current = etcp->input_wait_ack->head; - } - queue_resume_callback(etcp->input_wait_ack); - etcp_conn_wake_sender(etcp); -} - -static void ack_timeout_cb(void* arg) {// сработал таймер переотправки - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); - struct ETCP_CONN* etcp = (struct ETCP_CONN*)arg; - etcp->retrans_timer=NULL; - ack_timeout_check(etcp);// он установит новый таймер или выгребет всё и установит ожидание на очередь -} - -static void wait_ack_cb(struct ll_queue* q, void* arg) {// добавили пакет в ожидание ACK - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); - struct ETCP_CONN* etcp = (struct ETCP_CONN*)arg; - if (!etcp->retrans_timer) ack_timeout_check(etcp);// если таймер ретрансмиссий взведен - дожидаемся таймера. -} static void input_send_q_cb(struct ll_queue* q, void* arg) {// etcp->input_send_q data ready (void)q; @@ -1188,17 +1099,45 @@ void etcp_on_link_down(struct ETCP_CONN* etcp, struct ETCP_LINK* down_link) { } } -static void ack_response_timer_cb(void* arg) {// проверяем неотправленные ack response и отправляем если надо. - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); - struct ETCP_CONN* etcp = (struct ETCP_CONN*)arg; - etcp->ack_resp_timer=NULL; - if (etcp->ack_q->count==0) return;// нечего отправлять - etcp_conn_wake_sender(etcp); -// если ack все еще заняты - обновляем таймаут - if (etcp->ack_q->count) etcp->ack_resp_timer = uasync_set_timeout(etcp->instance->ua, ACK_DELAY_TB, etcp, ack_response_timer_cb, "etcp_ack_resp"); -// else etcp->ack_resp_timer=NULL; +static void ack_response_timer_cb(void* arg) { + struct ETCP_CONN* conn = arg; + conn->ack_resp_timer = NULL; + etcp_conn_wake_sender(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"); + if (!conn->ack_resp_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] ACK delay timer allocation failed, requesting immediate send", conn->log_name); + etcp_conn_wake_sender(conn); + } +} + +static void ack_window_cb(void* arg) { + struct ETCP_CONN* conn = arg; + conn->ack_window_timer = NULL; + if (conn->close_requested || !conn->input_queue->count || + (uint32_t)(conn->next_tx_id - conn->rx_ack_till) <= MAX_INFLIGHT_SIZE) return; + conn->ack_request_pending = 1; + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] requesting peer window: next_seq=%u till=%u", + conn->log_name, conn->next_tx_id, conn->rx_ack_till); + etcp_conn_wake_sender(conn); + ack_window_arm(conn); +} + +static void ack_window_arm(struct ETCP_CONN* conn) { + if (conn->ack_window_timer || conn->close_requested) return; + uint32_t delay = 1000000; + for (struct ETCP_LINK* link = conn->links; link; link = link->next) + if (link->initialized && link->link_status && etcp_tx_rto_us(link) < delay) delay = etcp_tx_rto_us(link); + conn->ack_window_timer = uasync_set_timeout(conn->instance->ua, (int)((delay + 99) / 100), conn, + ack_window_cb, "etcp_peer_window"); + if (!conn->ack_window_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot schedule peer-window query", conn->log_name); + conn->callbacks_running++; etcp_connection_close(conn); conn->callbacks_running--; + } +} static void etcp_link_ready_callback(struct ETCP_CONN* etcp) { DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); if (!etcp) return; @@ -1229,7 +1168,7 @@ static void etcp_conn_process_send_queue(struct ETCP_CONN* etcp) {// вызыв etcp->callbacks_running++; struct ETCP_DGRAM* dgram; while (!etcp->close_requested && (dgram = etcp_request_pkt(etcp))) { - etcp_loadbalancer_send(dgram); + if (etcp_loadbalancer_send(dgram) < 0) { etcp_schedule_send(etcp, 100); break; } } etcp->callbacks_running--; etcp->tx_running = 0; @@ -1244,270 +1183,129 @@ static void etcp_send_timer_cb(void* arg) { // Подготовить и отправить кодограмму // вызывается линком когда освобождается или очередью если появляются данные на передачу -struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* etcp) { - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); - if (!etcp) return NULL; - if (etcp->state == 2) { DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] request_pkt on deleted conn", etcp->log_name); return NULL; } - if (etcp->close_requested || !etcp->initialized || etcp->session_required || etcp->reinit_pending) { - if (etcp->tx_state) etcp->tx_state = ETCP_TX_STATE_LINK_WAIT; +static uint64_t read_le(const uint8_t* data, unsigned size) { + uint64_t value = 0; + for (unsigned i = 0; i < size; i++) value |= (uint64_t)data[i] << (i * 8); + return value; +} + +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)); +} + +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 (!etcp->input_send_q->head) input_queue_try_push(etcp); - struct INFLIGHT_PACKET* inf_pkt = (struct INFLIGHT_PACKET*)etcp->input_send_q->head; + if (!conn->input_send_q->head) input_queue_try_push(conn); + struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)conn->input_send_q->head; struct ETCP_LINK* link = NULL; - // Измерительная серия не ждёт DATA-cwnd; PAYLOAD добавляется только при наличии окна. - for (struct ETCP_LINK* l = etcp->links; l; l = l->next) { - if (l->burst_active && l->initialized && l->link_status) { link = l; break; } - } - if (link && inf_pkt && (inf_pkt->ll.len > link->inflight_lim_bytes || - link->inflight_bytes > link->inflight_lim_bytes - inf_pkt->ll.len)) { - link->cwnd_limited = 1; - inf_pkt = NULL; - } - if (!link && inf_pkt) link = etcp_loadbalancer_select_link(etcp, inf_pkt->ll.len); - if (!link && etcp->ack_q->count) { - inf_pkt = NULL; - link = etcp_loadbalancer_select_link(etcp, 0); + 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 && (packet->ll.len > link->inflight_lim_bytes || + link->inflight_bytes > link->inflight_lim_bytes - packet->ll.len)) { + link->cwnd_limited = 1; packet = NULL; + } } + if (!link && packet) link = etcp_loadbalancer_select_link(conn, packet->ll.len, packet->last_attempt_id ? packet : NULL); if (!link) { - if (!inf_pkt) { etcp->tx_state = ETCP_TX_STATE_DATA_WAIT; return NULL; } - if (etcp->tx_state != ETCP_TX_STATE_LINK_WAIT) - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] DATA waits: seq=%u len=%u (link/window/shaper)", - etcp->log_name, inf_pkt->seq, inf_pkt->ll.len); - etcp->tx_state=ETCP_TX_STATE_LINK_WAIT; - etcp->cnt_link_wait++; - return NULL; + for (struct ETCP_LINK* candidate = conn->links; candidate; candidate = candidate->next) + if (candidate->initialized && candidate->link_status && candidate->ack_q && candidate->ack_q->count) { + link = candidate; packet = NULL; break; + } } - etcp->tx_state=ETCP_TX_STATE_DATA_WAIT; - struct ETCP_DGRAM* dgram = memory_pool_alloc(etcp->instance->pkt_pool); - if (!dgram) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot allocate TX packet; retry in 10ms", etcp->log_name); - etcp_schedule_send(etcp, 100); - return NULL; + if (!link && (conn->ack_till_pending || conn->ack_request_pending)) { + link = etcp_loadbalancer_select_link(conn, 0, NULL); packet = NULL; } - if (inf_pkt) { - inf_pkt = (struct INFLIGHT_PACKET*)queue_data_get(etcp->input_send_q); - uint64_t now=get_time_tb(); - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] send_q->wait_ack seq=%d TS=%llu", etcp->log_name, inf_pkt->seq, now); - -// === NEW: per-link inflight + send_hist logic === - // Compute 1-based link number in the linked list (as requested) - uint8_t link_num = 0; - struct ETCP_LINK* tmp = etcp->links; - while (tmp) { - link_num++; - if (tmp == link) break; - tmp = tmp->next; - } - if (link_num == 0) link_num = 255; // safety (should never happen) - - // Fill history (send_count is the index before this attempt) - inf_pkt->send_hist[inf_pkt->send_count % 8] = link_num; - - // RTO уже освободил старое окно; здесь учитывается только новая попытка. - if (inf_pkt->last_link) { - inf_pkt->last_link->total_retransmissions++; - } - - // Новый flight не включает время простоя в оценку скорости. - if (!link->inflight_packets) { link->first_tx_time_tb = now; link->last_ack_time_tb = now; } - // Always add to the CURRENT link (first send or retransmission) - link->inflight_bytes += inf_pkt->ll.len; - link->inflight_packets++; - - // BBR: сохраняем снэпшот на момент отправки - inf_pkt->delivered_at_send = link->delivered_bytes; - inf_pkt->delivered_time_at_send_tb = link->last_ack_time_tb; - inf_pkt->first_tx_at_send_tb = link->first_tx_time_tb; - inf_pkt->inflight_at_send = link->inflight_bytes; - struct PKTNORM* pn = etcp->normalizer; - inf_pkt->is_app_limited = !etcp->input_queue->count && !etcp->input_send_q->count && - (!pn || (!pn->input->count && !pn->pending && !pn->data_ptr)); - if (link->bbr && link->bbr->mode == BBR_PROBE_RTT) inf_pkt->is_app_limited = 1; - - // Update last_link for future ACK/retrans - inf_pkt->last_link = link; - - inf_pkt->last_timestamp=now; - inf_pkt->send_count++; - inf_pkt->state=INFLIGHT_STATE_WAIT_ACK; - - queue_data_put_with_index(etcp->input_wait_ack, &inf_pkt->ll);// move dgram to wait_ack queue - } - size_t ack_q_size = queue_entry_count(etcp->ack_q); - - dgram->link = link; - dgram->noencrypt_len=0; - dgram->payload_len = inf_pkt ? inf_pkt->ll.len : 0; - dgram->timestamp = get_current_timestamp(); - -// формат ack: [01] [elements count] [4 байта last_delivered_id] [2 байта rx_dup_count] и <[4 байта seq][2 байта recv_ts][2 байта txrx delay ts]> x count - dgram->data[0]=1;// ack - int ptr=2; - - dgram->data[ptr++]=etcp->last_delivered_id; - dgram->data[ptr++]=etcp->last_delivered_id>>8; - dgram->data[ptr++]=etcp->last_delivered_id>>16; - dgram->data[ptr++]=etcp->last_delivered_id>>24; - - dgram->data[ptr++]=etcp->rx_dup_count; - dgram->data[ptr++]=etcp->rx_dup_count>>8; - - int data_len=0; - if (inf_pkt) data_len=inf_pkt->ll.len; - int remain_len = link->mtu - - (link->remote_addr.ss_family == AF_INET6 ? IP_UDP_OVERHEAD_V6 : IP_UDP_OVERHEAD_V4) - - SC_NONCE_SIZE - SC_TAG_SIZE - SC_CRC32_SIZE - - (ETCP_ENCRYPTED_HDR_SIZE + ETCP_SESSION_HEADER_SIZE + ETCP_ACK_BASE_SIZE) - - (1 + sizeof(inf_pkt->seq)) /*payload hdr*/ - - data_len; -// DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO, "remain_len= %d pl=%d", remain_len, data_len); - -// добавим опциональные заголовки - struct ACK_PACKET* ack_pkt; - while (remain_len>=8) { - ack_pkt = (struct ACK_PACKET*)queue_data_get(etcp->ack_q); - if (!ack_pkt) break; - remain_len-=8; - // seq 4 байта - dgram->data[ptr++]=ack_pkt->seq; - dgram->data[ptr++]=ack_pkt->seq>>8; - dgram->data[ptr++]=ack_pkt->seq>>16; - dgram->data[ptr++]=ack_pkt->seq>>24; - - // ts приема 2 байта - dgram->data[ptr++]=ack_pkt->recv_timestamp; - dgram->data[ptr++]=ack_pkt->recv_timestamp>>8; - - // время задержки 2 байта между recv и ack - uint16_t dly=get_current_timestamp()-ack_pkt->recv_timestamp; - dgram->data[ptr++]=dly; - dgram->data[ptr++]=dly>>8; - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] RX Send ACK seq=%d need=%d", etcp->log_name, ack_pkt->seq, etcp->last_delivered_id+1); - queue_entry_free((struct ll_entry*)ack_pkt); - - if (inf_pkt && inf_pkt->ll.len+ptr>=etcp->mtu-10) break;// pkt len (надо просчитать точнее включая все заголовки) - if (ptr>500) break; - } - - dgram->data[1]=(ptr - 8) / 8; - + if (!link) { conn->tx_state = packet ? 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; - if (max_enc < 0) max_enc = 0; - uint8_t burst_flags = 0; int pkt_sz_pos = 0; - -// === MEAS_RESP piggyback (если есть готовый ответ и не активен burst) === - if (link->burst_resp_pending && !link->burst_active && max_enc - ptr >= MEAS_RESP_SECTION_SIZE) { - dgram->data[ptr++] = ETCP_SECTION_MEAS_RESP; - dgram->data[ptr++] = link->burst_recv_id & 0xFF; dgram->data[ptr++] = (link->burst_recv_id >> 8) & 0xFF; - dgram->data[ptr++] = link->burst_resp_valid; - uint32_t v = link->burst_resp_gap_avg; - dgram->data[ptr++] = v & 0xFF; dgram->data[ptr++] = (v >> 8) & 0xFF; dgram->data[ptr++] = (v >> 16) & 0xFF; dgram->data[ptr++] = (v >> 24) & 0xFF; - v = link->burst_resp_gap_min; - dgram->data[ptr++] = v & 0xFF; dgram->data[ptr++] = (v >> 8) & 0xFF; dgram->data[ptr++] = (v >> 16) & 0xFF; dgram->data[ptr++] = (v >> 24) & 0xFF; - dgram->data[ptr++] = link->burst_resp_pkt_count; - link->burst_resp_pending = 0; - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] TX burst resp id=%u valid=%u gap_min=%u", etcp->log_name, link->burst_recv_id, link->burst_resp_valid, link->burst_resp_gap_min); + unsigned payload_size = packet ? ETCP_PAYLOAD_HEADER_SIZE + packet->ll.len : 0; + if (max_enc < (int)(ETCP_ACK_BASE_SIZE + payload_size + conn->ack_request_pending)) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] DATA does not fit negotiated MTU path=%u max=%d payload=%u", + conn->log_name, link->local_link_id, max_enc, payload_size); + return NULL; } - -// === MEAS_TS section (burst) === - if (link->burst_active && max_enc - ptr >= MEAS_TS_SECTION_SIZE) { - burst_flags = 0; - if (!inf_pkt) burst_flags |= MEAS_FLAG_IS_FILLER; - if (link->burst_seq == 0) burst_flags |= MEAS_FLAG_IS_FIRST; - if (link->burst_seq + 1 >= link->burst_count) burst_flags |= MEAS_FLAG_IS_LAST; - dgram->data[ptr++] = ETCP_SECTION_MEAS_TS; - dgram->data[ptr++] = link->burst_id & 0xFF; dgram->data[ptr++] = (link->burst_id >> 8) & 0xFF; - dgram->data[ptr++] = burst_flags; - dgram->data[ptr++] = link->burst_seq; - uint16_t ts_us = (uint16_t)get_time_us(); - dgram->data[ptr++] = ts_us & 0xFF; dgram->data[ptr++] = (ts_us >> 8) & 0xFF; - pkt_sz_pos = ptr; - dgram->data[ptr++] = 0; dgram->data[ptr++] = 0; + struct ETCP_DGRAM* dgram = memory_pool_alloc(conn->instance->pkt_pool); + if (!dgram) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot allocate TX packet; retry in 10ms", conn->log_name); + etcp_schedule_send(conn, 100); return NULL; } - - - if (link->last_recv_updated && remain_len>=5) {// если есть данные - добавим channel_timestamp - uint64_t now=get_time_tb(); - uint64_t dt=now - link->last_recv_local_time; - link->last_recv_updated=0; - if (dt<1000000) { - dgram->data[ptr++]=ETCP_SECTION_TIMESTAMP; - DEBUG_DEBUG(DEBUG_CATEGORY_KEEPALIVE, "[%s] KA-TX: ts_section dt=%llu tcp=%d", - link->etcp->log_name, (unsigned long long)dt, dgram->link->is_tcp); - - uint16_t t=link->last_recv_timestamp + dt; - dgram->data[ptr++]=t; - dgram->data[ptr++]=t>>8; - - t=link->last_recv_local_time - link->last_recv_timestamp; - dgram->data[ptr++]=t; - dgram->data[ptr++]=t>>8; - remain_len-=5; + dgram->link = link; dgram->attempt = NULL; dgram->noencrypt_len = 0; + dgram->payload_len = packet ? packet->ll.len : 0; + dgram->timestamp = get_current_timestamp(); dgram->burst_packet = link->burst_active; + dgram->ack_request = conn->ack_request_pending; + if (packet) { + dgram->attempt = etcp_tx_prepare(link, packet); + if (!dgram->attempt) { + memory_pool_free(conn->instance->pkt_pool, dgram); etcp_schedule_send(conn, 100); return NULL; } } - -// DEBUG_ERROR(DEBUG_CATEGORY_CRYPTO, "remain_len(2)= %d", remain_len); - - if (inf_pkt) { -// === FILLER section before PAYLOAD (burst padding) === - if (link->burst_active) { - int payload_overhead = 1 + 4 + data_len; // type + seq + data - int buf_avail = PACKET_DATA_SIZE - ptr - payload_overhead; - int filler_data = max_enc - ptr - FILLER_HDR_SIZE - payload_overhead; - if (filler_data > buf_avail) filler_data = buf_avail; - if (filler_data > 0) { - dgram->data[ptr++] = ETCP_SECTION_FILLER; - dgram->data[ptr++] = filler_data & 0xFF; - dgram->data[ptr++] = (filler_data >> 8) & 0xFF; - memset(&dgram->data[ptr], 0, filler_data); ptr += filler_data; - } + uint8_t* data = dgram->data; + data[0] = ETCP_SECTION_ACK; data[1] = 0; + write_le(data + 2, conn->last_delivered_id, 4); write_le(data + 6, conn->rx_dup_count, 2); + int ptr = ETCP_ACK_BASE_SIZE, remaining = max_enc - ptr - payload_size - dgram->ack_request; + while (remaining >= ETCP_ACK_ENTRY_SIZE && data[1] != UINT8_MAX) { + struct ACK_PACKET* ack = (struct ACK_PACKET*)queue_data_get(link->ack_q); + if (!ack) break; + write_le(data + ptr, ack->seq, 4); write_le(data + ptr + 4, ack->attempt_id, 8); + uint64_t delay = get_time_us() - ack->received_at_us; + write_le(data + ptr + 12, delay >= UINT32_MAX ? UINT32_MAX : delay, 4); + ptr += ETCP_ACK_ENTRY_SIZE; remaining -= ETCP_ACK_ENTRY_SIZE; data[1]++; + queue_entry_free(&ack->ll); + } + if (dgram->ack_request) data[ptr++] = ETCP_SECTION_ACK_REQUEST; + if (link->burst_resp_pending && !link->burst_active && remaining >= MEAS_RESP_SECTION_SIZE) { + data[ptr++] = ETCP_SECTION_MEAS_RESP; + write_le(data + ptr, link->burst_recv_id, 2); ptr += 2; + data[ptr++] = link->burst_resp_valid; + write_le(data + ptr, link->burst_resp_gap_avg, 4); ptr += 4; + write_le(data + ptr, link->burst_resp_gap_min, 4); ptr += 4; + data[ptr++] = link->burst_resp_pkt_count; + remaining -= MEAS_RESP_SECTION_SIZE; link->burst_resp_pending = 0; + } + int pkt_size_offset = 0; + if (dgram->burst_packet && remaining >= MEAS_TS_SECTION_SIZE) { + uint8_t flags = packet ? 0 : MEAS_FLAG_IS_FILLER; + if (!link->burst_seq) flags |= MEAS_FLAG_IS_FIRST; + if (link->burst_seq + 1 >= link->burst_count) flags |= MEAS_FLAG_IS_LAST; + data[ptr++] = ETCP_SECTION_MEAS_TS; + write_le(data + ptr, link->burst_id, 2); ptr += 2; + data[ptr++] = flags; data[ptr++] = link->burst_seq; + write_le(data + ptr, get_time_us(), 2); ptr += 2; + pkt_size_offset = ptr; ptr += 2; remaining -= MEAS_TS_SECTION_SIZE; + } + if (link->last_recv_updated && remaining >= 5) { + uint64_t dt = get_time_tb() - link->last_recv_local_time; + link->last_recv_updated = 0; + if (dt < 1000000) { + data[ptr++] = ETCP_SECTION_TIMESTAMP; + write_le(data + ptr, link->last_recv_timestamp + dt, 2); ptr += 2; + write_le(data + ptr, link->last_recv_local_time - link->last_recv_timestamp, 2); ptr += 2; + remaining -= 5; } - // фрейм data (0) обязательно в конец - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] TX DATA: seq=%u len=%u retry=%d; Qlen: in=%d snd=%d wait=%d", etcp->log_name, inf_pkt->seq, inf_pkt->ll.len, inf_pkt->send_count, etcp->input_queue->count, etcp->input_send_q->count, etcp->input_wait_ack->count); - - dgram->data[ptr++]=0;// payload - dgram->data[ptr++]=inf_pkt->seq; - dgram->data[ptr++]=inf_pkt->seq>>8; - dgram->data[ptr++]=inf_pkt->seq>>16; - dgram->data[ptr++]=inf_pkt->seq>>24; - - if ((int)(ptr + inf_pkt->ll.len) <= PACKET_DATA_SIZE) { - memcpy(&dgram->data[ptr], inf_pkt->ll.dgram, inf_pkt->ll.len); ptr+=inf_pkt->ll.len; - } else DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] payload overflow: ptr=%d len=%u max=%d", etcp->log_name, ptr, inf_pkt->ll.len, PACKET_DATA_SIZE); } - else { - if (link->burst_active) { - int buf_avail = PACKET_DATA_SIZE - ptr; - int filler_data = max_enc - ptr - FILLER_HDR_SIZE; - if (filler_data > buf_avail - FILLER_HDR_SIZE) filler_data = buf_avail - FILLER_HDR_SIZE; - if (filler_data > 0) { - dgram->data[ptr++] = ETCP_SECTION_FILLER; - dgram->data[ptr++] = filler_data & 0xFF; - dgram->data[ptr++] = (filler_data >> 8) & 0xFF; - memset(&dgram->data[ptr], 0, filler_data); ptr += filler_data; - } - } - int chk=queue_check_consistency(etcp->ack_q); - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] only ACK (size=%d) packet with %d bytes total (chk=%d) rem=%d", etcp->log_name, ack_q_size, ptr, chk, remain_len); + if (dgram->burst_packet && remaining > FILLER_HDR_SIZE) { + unsigned filler = remaining - FILLER_HDR_SIZE; + data[ptr++] = ETCP_SECTION_FILLER; write_le(data + ptr, filler, 2); ptr += 2; + memset(data + ptr, 0, filler); ptr += filler; } - if (ptr>=PACKET_DATA_SIZE-50) DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] SIZE ERROR!!! %d", etcp->log_name, ptr); - - dgram->data_len=ptr; - -// === Fill pkt_sz in MEAS_TS and advance burst state === - if (link->burst_active && pkt_sz_pos > 0) { - uint16_t pkt_sz = (uint16_t)dgram->data_len; - dgram->data[pkt_sz_pos] = pkt_sz & 0xFF; - dgram->data[pkt_sz_pos + 1] = (pkt_sz >> 8) & 0xFF; - link->burst_pkt_size = pkt_sz; - link->burst_seq++; - if (link->burst_seq >= link->burst_count) etcp_link_burst_finish(link); + if (packet) { + data[ptr++] = ETCP_SECTION_PAYLOAD; write_le(data + ptr, packet->seq, 4); ptr += 4; + write_le(data + ptr, dgram->attempt->attempt_id, 8); ptr += 8; + memcpy(data + ptr, packet->ll.dgram, packet->ll.len); ptr += packet->ll.len; + } + dgram->data_len = ptr; + if (pkt_size_offset) { + write_le(data + pkt_size_offset, ptr, 2); link->burst_pkt_size = ptr; + if (++link->burst_seq >= link->burst_count) etcp_link_burst_finish(link); } - etcp_dump_pkt_sections(dgram, link, 1); - return dgram; } @@ -1564,122 +1362,35 @@ void etcp_output_try_assembly(struct ETCP_CONN* etcp) { next_expected_id++; } - if (delivered_count>0) DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] delivered %u contiguous packets (%u bytes), last_delivered_id=%u, asm_queue=%d, output_queue=%d", - etcp->log_name, delivered_count, delivered_bytes, etcp->last_delivered_id, queue_entry_count(etcp->recv_q), queue_entry_count(etcp->output_queue)); + if (delivered_count > 0 && !etcp->close_requested && !etcp->reinit_pending) { + etcp->ack_till_pending = 1; ack_schedule(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); + } } // Process ACK receipt - remove acknowledged packet from inflight queues -void etcp_ack_recv(struct ETCP_CONN* etcp, uint32_t seq, uint16_t ts, uint16_t dts) { - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); - if (!etcp) return; - if (etcp->state == 2) { DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] ack_recv on deleted conn", etcp->log_name); return; } - -// DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] processing ACK for seq=%u, ts=%u, dts=%u", etcp->log_name, seq, ts, dts); - - // Find the acknowledged packet in the wait_ack queue - struct INFLIGHT_PACKET* acked_pkt; - acked_pkt = (struct INFLIGHT_PACKET*)queue_find_data_by_index(etcp->input_wait_ack, &seq); - if (acked_pkt) { - etcp->cnt_ack_hit_inf++; - queue_remove_data(etcp->input_wait_ack, (struct ll_entry*)acked_pkt); - } - else { - acked_pkt = (struct INFLIGHT_PACKET*)queue_find_data_by_index(etcp->input_send_q, &seq); - if (acked_pkt) { - etcp->cnt_ack_hit_sndq++; - queue_remove_data(etcp->input_send_q, (struct ll_entry*)acked_pkt); - } - else etcp->cnt_ack_miss++; +void etcp_ack_recv(struct ETCP_CONN* conn, uint32_t seq) { + if (!conn || conn->state == 2 || conn->close_requested) return; + struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)queue_find_data_by_index(conn->input_wait_ack, &seq); + struct ll_queue* queue = conn->input_wait_ack; + if (!packet) { packet = (struct INFLIGHT_PACKET*)queue_find_data_by_index(conn->input_send_q, &seq); queue = conn->input_send_q; } + if (!packet) { conn->cnt_ack_miss++; return; } + if (queue == conn->input_wait_ack) conn->cnt_ack_hit_inf++; else conn->cnt_ack_hit_sndq++; + queue_remove_data(queue, &packet->ll); + if (conn->unacked_bytes < packet->ll.len) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] logical ACK accounting mismatch seq=%u unacked=%u len=%u", + conn->log_name, seq, conn->unacked_bytes, packet->ll.len); + abort(); } - - if (!acked_pkt) { - // Packet might be already acknowledged or not found - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "packet seq=%u not found in wait_ack queue", seq); - return; - } - etcp_tx_record_ack(&etcp->instance->tx_history, get_time_us(), acked_pkt->ll.len); - - int cwnd_limited = acked_pkt->last_link && (acked_pkt->last_link->cwnd_limited || - acked_pkt->last_link->inflight_bytes >= acked_pkt->last_link->inflight_lim_bytes); - etcp_inflight_release(acked_pkt); - if (acked_pkt->state == INFLIGHT_STATE_WAIT_SEND) - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] late ACK seq=%u before retry, no window debit", etcp->log_name, seq); - if (acked_pkt->last_link && acked_pkt->last_link->bbr) { - struct ETCP_LINK* link = acked_pkt->last_link; - - // BBR: собираем rate_sample и вызываем bbr_main - uint64_t now_tb = get_time_tb(); - link->delivered_bytes += acked_pkt->ll.len; - uint64_t delivered = link->delivered_bytes - acked_pkt->delivered_at_send; - uint64_t ack_elapsed = now_tb - acked_pkt->delivered_time_at_send_tb; - uint64_t send_elapsed = acked_pkt->last_timestamp - acked_pkt->first_tx_at_send_tb; - uint64_t interval = (ack_elapsed > send_elapsed ? ack_elapsed : send_elapsed) * 100; - uint32_t interval_us = interval > UINT32_MAX ? UINT32_MAX : (uint32_t)interval; - if (acked_pkt->state == INFLIGHT_STATE_WAIT_SEND) interval_us = 0; // поздний ACK до retry, timestamp уже заменён RTO - struct bbr_rate_sample rs = { - .delivered = delivered > UINT32_MAX ? UINT32_MAX : (uint32_t)delivered, - .interval_us = interval_us, - .rtt_us = (uint32_t)link->rtt_last * 100, - .acked_sacked = acked_pkt->ll.len, - .prior_delivered = (uint32_t)acked_pkt->delivered_at_send, - .tx_in_flight = acked_pkt->inflight_at_send, - .lost = (int)link->bbr_loss_since_ack, - .is_app_limited = acked_pkt->is_app_limited, - }; - link->last_ack_time_tb = now_tb; - if (acked_pkt->state == INFLIGHT_STATE_WAIT_ACK && acked_pkt->last_timestamp > link->first_tx_time_tb) - link->first_tx_time_tb = acked_pkt->last_timestamp; - - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[L%u] BBR in: seq=%u del=%u acked=%u iv=%uus rtt=%uus pri=%u tx_infl=%u cur=%u/%u lost=%d app=%d", - link->local_link_id, seq, rs.delivered, rs.acked_sacked, rs.interval_us, rs.rtt_us, - rs.prior_delivered, rs.tx_in_flight, link->inflight_bytes, link->inflight_lim_bytes, - rs.lost, rs.is_app_limited); - - uint32_t old_cwnd = link->inflight_lim_bytes; - uint32_t old_pacing = link->bbr_pacing_rate; - uint32_t new_cwnd = old_cwnd; - uint32_t new_pacing = old_pacing; - bbr_main(link->bbr, &rs, &new_cwnd, &new_pacing, etcp->normalizer->frag_size, link->inflight_bytes, - cwnd_limited); - link->cwnd_limited = 0; - etcp_link_update_inflight_lim(link, new_cwnd); - etcp_loadbalancer_set_pacing_rate(link, new_pacing); - link->bbr_loss_since_ack = 0; - - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[L%u] BBR out: cwnd %u→%u pace %u→%u bw=%uK mode=%d cyc=%d infl=%u/%u", - link->local_link_id, old_cwnd, new_cwnd, old_pacing, new_pacing, - link->bandwidth, link->bbr->mode, link->bbr->cycle_idx, link->inflight_bytes, new_cwnd); - - link->acked_bytes += acked_pkt->ll.len; - link->acked_packets++; - } - - - // Update connection statistics - if (etcp->unacked_bytes >= acked_pkt->ll.len) etcp->unacked_bytes -= acked_pkt->ll.len; - else { - DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] unacked_bytes underflow prevented: %u < %u", etcp->log_name, etcp->unacked_bytes, acked_pkt->ll.len); - etcp->unacked_bytes = 0; - } - etcp->bytes_sent_total += acked_pkt->ll.len; - etcp->ack_packets_count++; - - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] TX removed packet seq=%u from wait_ack, unacked_bytes now %u total acked=%u", etcp->log_name, seq, etcp->unacked_bytes, etcp->ack_packets_count); - - if (acked_pkt->ll.dgram) { - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] TX-ACK free dgram=%p seq=%u", etcp->log_name, acked_pkt->ll.dgram, seq); - memory_pool_free(etcp->instance->data_pool, acked_pkt->ll.dgram); - } - memory_pool_free(etcp->inflight_pool, acked_pkt); - - // Try to resume sending more packets if window space opened up - input_queue_try_resume(etcp); - etcp_conn_wake_sender(etcp); - -// DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] completed for seq=%u", etcp->log_name, seq); + conn->unacked_bytes -= packet->ll.len; conn->bytes_sent_total += packet->ll.len; conn->ack_packets_count++; + etcp_tx_record_ack(&conn->instance->tx_history, get_time_us(), packet->ll.len); + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] logical delivery seq=%u bytes=%u active_attempts=%u unacked=%u", + conn->log_name, seq, packet->ll.len, packet->active_attempts, conn->unacked_bytes); + if (packet->ll.dgram) memory_pool_free(conn->instance->data_pool, packet->ll.dgram); + queue_entry_free(&packet->ll); + input_queue_try_resume(conn); etcp_conn_wake_sender(conn); } - - // Process incoming decrypted packet void etcp_conn_input(struct ETCP_DGRAM* pkt) { DEBUG_TRACE(DEBUG_CATEGORY_ETCP, ""); @@ -1712,37 +1423,38 @@ void etcp_conn_input(struct ETCP_DGRAM* pkt) { // Process sections as per protocol.txt switch (type) { case ETCP_SECTION_ACK: { - if (len < ETCP_ACK_BASE_SIZE) { len = 0; break; } - int elm_cnt=data[1]; - uint32_t till=data[2] | (data[3]<<8) | (data[4]<<16) | ((uint32_t)data[5]<<24); - uint32_t last_tx = etcp->next_tx_id - 1; - if ((int32_t)(till - last_tx) > 0) till = last_tx; // нельзя подтвердить ещё не назначенный seq - uint16_t rx_dup_count_16 = data[6] | (data[7]<<8); - uint16_t old_tx_dup_16 = etcp->tx_dup_count & 0xFFFF; - int16_t diff = (int16_t)(rx_dup_count_16 - old_tx_dup_16); - etcp->tx_dup_count += diff; - int ack_section_len = 8 + elm_cnt * 8; - if (ack_section_len > len) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] ACK section too large: elm_cnt=%d len=%d", etcp->log_name, elm_cnt, len); - len = 0; - break; + if (len < ETCP_ACK_BASE_SIZE) { DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] short ACK len=%u", etcp->log_name, len); len = 0; break; } + unsigned count = data[1], section_len = ETCP_ACK_BASE_SIZE + count * ETCP_ACK_ENTRY_SIZE; + if (section_len > len) { + DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] ACK exceeds packet count=%u len=%u", etcp->log_name, count, len); + len = 0; break; } - data+=ack_section_len; - len-=ack_section_len; - for (int i=0; inext_tx_id - 1; + if ((int32_t)(till - last_tx) > 0) till = last_tx; + uint16_t duplicates = (uint16_t)read_le(data + 6, 2); + etcp->tx_dup_count += (int16_t)(duplicates - (uint16_t)etcp->tx_dup_count); + for (unsigned i = 0; i < count; i++) { + const uint8_t* entry = data + ETCP_ACK_BASE_SIZE + i * ETCP_ACK_ENTRY_SIZE; + etcp_tx_ack(pkt->link, (uint32_t)read_le(entry, 4), read_le(entry + 4, 8), (uint32_t)read_le(entry + 12, 4)); + if (etcp->close_requested || etcp->reinit_pending) break; + } + uint32_t advanced = 0; + while (!etcp->close_requested && (int32_t)(till - etcp->rx_ack_till) > 0 && advanced < ETCP_ACK_GAP_MAX_PACKETS) { + etcp->rx_ack_till++; etcp_ack_recv(etcp, etcp->rx_ack_till); advanced++; + } + if (advanced) { + if (etcp->ack_window_timer && (uint32_t)(etcp->next_tx_id - etcp->rx_ack_till) <= MAX_INFLIGHT_SIZE) { + uasync_cancel_timeout(etcp->instance->ua, etcp->ack_window_timer); etcp->ack_window_timer = NULL; } - uint32_t ack_gap = 0; - while ((int32_t)(etcp->rx_ack_till-till)<0 && ack_gap < ETCP_ACK_GAP_MAX_PACKETS) { - etcp->rx_ack_till++; etcp_ack_recv(etcp, etcp->rx_ack_till, -1, -1); ack_gap++; - }// подтверждаем всё по till - /* Все записи могли уже уйти по selective ACK; тогда ack_recv не вызовет resume. */ - if (ack_gap) input_queue_try_resume(etcp); + input_queue_try_resume(etcp); + } + data += section_len; len -= section_len; break; } + case ETCP_SECTION_ACK_REQUEST: + etcp->ack_till_pending = 1; etcp_conn_wake_sender(etcp); + data++; len--; + break; case ETCP_SECTION_TIMESTAMP: { if (len < 5) { len = 0; break; } uint16_t cur_ts=get_current_timestamp(); @@ -1801,93 +1513,48 @@ void etcp_conn_input(struct ETCP_DGRAM* pkt) { break; } case ETCP_SECTION_PAYLOAD: { - - if (len>=5) { - // формируем ACK - uint32_t seq=data[1] | (data[2]<<8) | (data[3]<<16) | ((uint32_t)data[4]<<24); - if (etcp->got_initial_pkt == 0) { - if (seq==1) { etcp->got_initial_pkt = 1; DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] Initial packet seq=1 received", etcp->log_name); } - else { - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] Out-of-order pkt before init seq=%d, waiting seq=1", - etcp->log_name, seq); - len=0; - break; - } + if (len < ETCP_PAYLOAD_HEADER_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] short DATA len=%u", etcp->log_name, len); len = 0; break; + } + uint32_t seq = (uint32_t)read_le(data + 1, 4); + uint64_t attempt_id = read_le(data + 5, 8); + uint32_t payload_len = len - ETCP_PAYLOAD_HEADER_SIZE; + if (!attempt_id || (!etcp->got_initial_pkt && seq != 1)) { + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] DATA rejected before initial fragment seq=%u attempt=%llu", + etcp->log_name, seq, (unsigned long long)attempt_id); + len = 0; break; + } + int32_t distance = (int32_t)(seq - etcp->last_delivered_id); + if (distance > MAX_INFLIGHT_SIZE || distance < -MAX_INFLIGHT_SIZE || payload_len > PACKET_DATA_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] DATA out of bounds seq=%u till=%u len=%u", + etcp->log_name, seq, etcp->last_delivered_id, payload_len); + len = 0; break; + } + int duplicate = distance <= 0 || queue_find_data_by_index(etcp->recv_q, &seq) != NULL; + if (!duplicate) { + struct ETCP_FRAGMENT* fragment = (struct ETCP_FRAGMENT*)queue_entry_new_from_pool(etcp->io_pool); + uint8_t* payload = memory_pool_alloc(etcp->instance->data_pool); + if (!fragment || !payload) { + if (fragment) queue_entry_free(&fragment->ll); + if (payload) memory_pool_free(etcp->instance->data_pool, payload); + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] DATA admission OOM seq=%u attempt=%llu; no ACK", + etcp->log_name, seq, (unsigned long long)attempt_id); + len = 0; break; } - else { - int32_t d=seq - etcp->last_delivered_id; - if (d>MAX_INFLIGHT_SIZE || d<-MAX_INFLIGHT_SIZE) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] Received packet out of inflight bounds: seq=%d last delivered=%d", etcp->log_name, seq, etcp->last_delivered_id); - len=0; - break; - } + fragment->seq = seq; fragment->timestamp = pkt->timestamp; + fragment->ll.dgram = payload; fragment->ll.dgram_pool = etcp->instance->data_pool; fragment->ll.len = payload_len; + memcpy(payload, data + ETCP_PAYLOAD_HEADER_SIZE, payload_len); + if (queue_data_put_with_index(etcp->recv_q, &fragment->ll) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot retain DATA seq=%u; no ACK", etcp->log_name, seq); + len = 0; break; } - if (queue_find_data_by_index(etcp->ack_q, &seq) == NULL) { - struct ACK_PACKET* p = (struct ACK_PACKET*)queue_entry_new_from_pool(etcp->instance->ack_pool); - if (!p) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] failed to allocate ACK_PACKET", etcp->log_name); - len = 0; - break; - } - p->seq=seq; - p->pkt_timestamp=pkt->timestamp; - p->recv_timestamp=get_current_timestamp(); - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] RX add to ack_q seq=%d", etcp->log_name, seq); - - queue_data_put_with_index(etcp->ack_q, (struct ll_entry*)p); - if (etcp->ack_resp_timer == NULL) { - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] set ack_timer for delayed ACK send", etcp->log_name); - etcp->ack_resp_timer = uasync_set_timeout(etcp->instance->ua, ACK_DELAY_TB, etcp, ack_response_timer_cb, "etcp_ack_resp"); - } - } - if (((int32_t)(etcp->last_delivered_id-seq)<0) && (queue_find_data_by_index(etcp->recv_q, &seq)==NULL)) {// проверяем есть ли пакет с этим seq - uint32_t pkt_len=len-5; - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] adding packet seq=%u to recv_q (last_delivered_id=%u)", etcp->log_name, seq, etcp->last_delivered_id); - // отправляем пакет в очередь на сборку - uint8_t* payload_data = memory_pool_alloc(etcp->instance->data_pool); - if (!payload_data) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] failed to allocate payload_data from data_pool", etcp->log_name); - len=0; - break; - } - struct ETCP_FRAGMENT* rx_pkt = (struct ETCP_FRAGMENT*)queue_entry_new_from_pool(etcp->io_pool); - if (!rx_pkt) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] failed to allocate rx_pkt from io_pool", etcp->log_name); - memory_pool_free(etcp->instance->data_pool, payload_data); - len=0; - break; - } - rx_pkt->seq = seq; - rx_pkt->timestamp = pkt->timestamp; - rx_pkt->ll.dgram = payload_data; - rx_pkt->ll.len = pkt_len; - rx_pkt->ll.dgram_pool = etcp->instance->data_pool; - rx_pkt->ll.memlen = etcp->instance->data_pool->object_size; - if (pkt_len > (int)etcp->instance->data_pool->object_size) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] payload too large for data_pool: %u > %zu", etcp->log_name, pkt_len, etcp->instance->data_pool->object_size); - memory_pool_free(etcp->instance->data_pool, payload_data); - queue_entry_free(&rx_pkt->ll); - len = 0; - break; - } - if (data + len > pkt->data + pkt->data_len) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] payload section bounds overflow: data+len=%p > end=%p", etcp->log_name, (void*)(data + len), (void*)(pkt->data + pkt->data_len)); - memory_pool_free(etcp->instance->data_pool, payload_data); - queue_entry_free(&rx_pkt->ll); - len = 0; - break; - } - // Copy the actual payload data - memcpy(payload_data, data + 5, pkt_len); - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] RX seq=%u need=%u asm_len=%d", etcp->log_name, seq, etcp->last_delivered_id+1, etcp->recv_q->count); - queue_data_put_with_index(etcp->recv_q, (struct ll_entry*)rx_pkt); - if ((int32_t)(seq - etcp->last_delivered_id) == 1) etcp_output_try_assembly(etcp);// пробуем собрать выходную очередь из фрагментов - } else { - etcp->rx_dup_count++; - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] RX dup: seq=%u need=%u asm_len=%d", etcp->log_name, seq, etcp->last_delivered_id+1, etcp->recv_q->count); - } - } else DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "payload len %d < 5", len); - len=0; + } else etcp->rx_dup_count++; + etcp->got_initial_pkt = 1; + if (etcp_tx_queue_ack(pkt->link, seq, attempt_id) == 0) ack_schedule(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_id, pkt->link->local_link_id, payload_len, duplicate); + if (distance == 1) etcp_output_try_assembly(etcp); + len = 0; break; } diff --git a/src/transport_layer/etcp_connections.h b/src/transport_layer/etcp_connections.h index c632e4c7..ef1055e2 100644 --- a/src/transport_layer/etcp_connections.h +++ b/src/transport_layer/etcp_connections.h @@ -72,6 +72,7 @@ struct ETCP_DGRAM {// пакет (незашифрованный) uint16_t payload_len; // локальные метаданные request_pkt: DATA-байты; 0 — только ACK/control/FILLER, вне DATA-pacing struct ETCP_TX_ATTEMPT* attempt; // локальный резерв окна; complete снимает/фиксирует его после send uint8_t burst_packet; // Пакет создан измерительной серией; не зависит от последующей смены burst_active + uint8_t ack_request; // Пакет содержит запрос границы окна; сбрасывается только после успешного send uint16_t timestamp;// timestamp отправляющего узла при отправке пакета uint8_t flag_up:1;// bit0 = up/down (recv_keepalive) uint8_t data[];// данные пакета (без timestamp) diff --git a/src/transport_layer/etcp_tx.c b/src/transport_layer/etcp_tx.c index 30f177d4..e7d57c21 100644 --- a/src/transport_layer/etcp_tx.c +++ b/src/transport_layer/etcp_tx.c @@ -156,6 +156,10 @@ struct ETCP_TX_ATTEMPT* etcp_tx_prepare(struct ETCP_LINK* link, struct INFLIGHT_ } void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success) { + if (success) { + dgram->link->etcp->ack_till_pending = 0; + if (dgram->ack_request) dgram->link->etcp->ack_request_pending = 0; + } struct ETCP_TX_ATTEMPT* attempt = dgram->attempt; if (!attempt) return; dgram->attempt = NULL;