From 5f085b2086b8a535f5aff23dd8e499d3422dea91 Mon Sep 17 00:00:00 2001 From: evgeny Date: Fri, 23 Oct 2026 12:22:22 +0200 Subject: [PATCH] Discard failed ETCP streams without allocating or replaying fragments --- src/transport_layer/etcp.c | 273 ++++++------------------------------- src/transport_layer/etcp.h | 6 +- 2 files changed, 47 insertions(+), 232 deletions(-) diff --git a/src/transport_layer/etcp.c b/src/transport_layer/etcp.c index 93c02c30..93ddded2 100644 --- a/src/transport_layer/etcp.c +++ b/src/transport_layer/etcp.c @@ -86,132 +86,13 @@ static void etcp_cancel_send(struct ETCP_CONN* conn) { } -static void clear_queue(struct ll_queue* q) { - if (!q) return; - struct ll_entry* pkt; - while ((pkt = queue_data_get(q)) != NULL) { - queue_dgram_free(pkt); - queue_entry_free(pkt); - } -} - static void drain_and_free_queue(struct ll_queue** q) { if (!*q) return; - clear_queue(*q); + queue_clear(*q); queue_free(*q); *q = NULL; } -static int inflight_seq_cmp(const void* a, const void* b) { - const struct INFLIGHT_PACKET* pa = *(const struct INFLIGHT_PACKET**)a; - const struct INFLIGHT_PACKET* pb = *(const struct INFLIGHT_PACKET**)b; - if (pa->seq < pb->seq) return -1; - if (pa->seq > pb->seq) return 1; - return 0; -} - -static void feed_dgram_to_asm(struct ETCP_CONN* etcp, struct PKTNORM* pn, - uint8_t* dgram, uint16_t dgram_len, - uint8_t** asm_buf, uint16_t* asm_len, uint32_t* asm_cap, - uint32_t* returned) { - uint32_t need = (uint32_t)*asm_len + dgram_len; - if (need >= ASM_BUF_MAX_SIZE) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] assembly buffer would exceed max size %d", etcp->log_name, ASM_BUF_MAX_SIZE); - return; - } - if (need > *asm_cap) { - uint32_t new_cap = *asm_cap ? *asm_cap : 256; - while (new_cap < need) new_cap *= 2; - if (new_cap > 0xFFFF) new_cap = 0xFFFF; - uint8_t* new_buf = u_realloc(*asm_buf, (uint16_t)new_cap); - if (!new_buf) return; - *asm_buf = new_buf; - *asm_cap = new_cap; - } - memcpy(*asm_buf + *asm_len, dgram, dgram_len); - *asm_len = need; - - while (*asm_len >= 2) { - uint16_t pkt_len = (*asm_buf)[0] | ((*asm_buf)[1] << 8); - if (pkt_len == 0 || pkt_len > 16384 || *asm_len < 2 + pkt_len) break; - - const uint8_t* pkt_data = *asm_buf + 2; - if (pkt_data[0] != ETCP_ID_TOPO_ENTRY) { - struct ll_entry* e = ll_alloc_lldgram(pkt_len); - if (e) { - memcpy(e->dgram, pkt_data, pkt_len); - e->len = pkt_len; - queue_data_put(pn->input, e); - (*returned)++; - } - } - - uint16_t consumed = 2 + pkt_len; - *asm_len -= consumed; - if (*asm_len > 0) memmove(*asm_buf, *asm_buf + consumed, *asm_len); - } -} - -static void etcp_return_inflight_to_normalizer(struct ETCP_CONN* etcp) { - struct PKTNORM* pn = etcp->normalizer; - if (!pn) return; - - struct INFLIGHT_PACKET** inflight = NULL; - int inflight_count = 0; - struct ll_entry** input_frags = NULL; - int input_count = 0; - struct ll_entry* entry; - - while ((entry = queue_data_get(etcp->input_wait_ack))) { - inflight = u_realloc(inflight, (inflight_count + 1) * sizeof(*inflight)); - inflight[inflight_count++] = (struct INFLIGHT_PACKET*)entry; - } - while ((entry = queue_data_get(etcp->input_send_q))) { - inflight = u_realloc(inflight, (inflight_count + 1) * sizeof(*inflight)); - inflight[inflight_count++] = (struct INFLIGHT_PACKET*)entry; - } - while ((entry = queue_data_get(etcp->input_queue))) { - input_frags = u_realloc(input_frags, (input_count + 1) * sizeof(*input_frags)); - input_frags[input_count++] = entry; - } - - if (inflight_count + input_count == 0 && (!pn->data || pn->data_ptr == 0)) return; - - if (inflight_count > 0) qsort(inflight, inflight_count, sizeof(*inflight), inflight_seq_cmp); - - uint8_t* asm_buf = NULL; - uint16_t asm_len = 0; - uint32_t asm_cap = 0; - uint32_t returned_packets = 0; - - for (int i = 0; i < inflight_count; i++) { - feed_dgram_to_asm(etcp, pn, inflight[i]->ll.dgram, inflight[i]->ll.len, - &asm_buf, &asm_len, &asm_cap, &returned_packets); - memory_pool_free(etcp->instance->data_pool, inflight[i]->ll.dgram); - memory_pool_free(etcp->inflight_pool, inflight[i]); - } - u_free(inflight); - - for (int i = 0; i < input_count; i++) { - feed_dgram_to_asm(etcp, pn, input_frags[i]->dgram, input_frags[i]->len, - &asm_buf, &asm_len, &asm_cap, &returned_packets); - memory_pool_free(etcp->instance->data_pool, input_frags[i]->dgram); - memory_pool_free(etcp->io_pool, input_frags[i]); - } - u_free(input_frags); - - if (pn->data && pn->data_ptr > 0) { - feed_dgram_to_asm(etcp, pn, pn->data, pn->data_ptr, - &asm_buf, &asm_len, &asm_cap, &returned_packets); - } - - if (asm_buf) { u_free(asm_buf); } - - if (returned_packets > 0) { - DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] returned %u packets to normalizer input during reinit", - etcp->log_name, returned_packets); - } -} uint16_t get_current_timestamp() { return (uint16_t)get_time_tb(); @@ -520,6 +401,10 @@ void etcp_connection_close(struct ETCP_CONN* etcp) { if (etcp->close_token) return; DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] close requested inside callback chain — deferring", etcp->log_name); etcp->close_token = uasync_call_soon(etcp->instance->ua, etcp, etcp_connection_close_deferred); + if (!etcp->close_token) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot defer mandatory close", etcp->log_name); + abort(); + } return; } @@ -597,112 +482,35 @@ void etcp_conn_ref_free(struct ETCP_CONN* conn) { } // Reset connection -void etcp_conn_reset(struct ETCP_CONN* etcp) { - if (!etcp || etcp->state == 2) { - if (etcp) DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] conn_reset on deleted conn", etcp->log_name); +void etcp_conn_reset(struct ETCP_CONN* conn) { + if (!conn || conn->state == 2) { + if (conn) DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] conn_reset on deleted conn", conn->log_name); return; } - // 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; - etcp->rx_ack_till = 0; - - // Устанавливаем флаг ожидания первого пакета - etcp->got_initial_pkt = 0; - - // Reset metrics - etcp->unacked_bytes = 0; - etcp->rtt_last = 0; - etcp->rtt_avg_10 = 0; - etcp->jitter = 0; - etcp->last_rtt_cb_time = 0; - etcp->bytes_sent_total = 0; - etcp->retransmissions_count = 0; - etcp->ack_packets_count = 0; - etcp->rx_dup_count = 0; - etcp->tx_dup_count = 0; - - // Reset RTT history - memset(etcp->rtt_history, 0, sizeof(etcp->rtt_history)); - etcp->rtt_history_idx = 0; - etcp->last_rr_link = NULL; - - // Return unconfirmed inflight data to normalizer input (до нормалайзера) - if (etcp->normalizer) { - etcp->normalizer->input->callback_suspended = 1; - etcp_return_inflight_to_normalizer(etcp); - } - - // Clear queues (keep queue structures) - clear_queue(etcp->input_queue); - clear_queue(etcp->output_queue); - clear_queue(etcp->input_send_q); - clear_queue(etcp->input_wait_ack); - clear_queue(etcp->recv_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); - if (etcp->normalizer && etcp->normalizer->input) { - clear_queue(etcp->normalizer->input); - queue_resume_callback(etcp->normalizer->input); - } - -// В etcp_conn_reset(), после очистки очередей добавьте: - struct ETCP_LINK* l = etcp->links; - while (l) { - l->inflight_bytes = 0; - l->inflight_packets = 0; - l->cwnd_limited = 0; - l->delivered_bytes = 0; - l->acked_bytes = 0; - l->acked_packets = 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->ack_resp_timer) { - uasync_cancel_timeout(etcp->instance->ua, etcp->ack_resp_timer); - etcp->ack_resp_timer = NULL; - } - - etcp->reset_count++; - - // Reset normalizer (packer/unpacker state for reconnection) - if (etcp->normalizer) { - pn_reset(etcp->normalizer); - } - - queue_resume_callback(etcp->input_queue); - queue_resume_callback(etcp->output_queue); - queue_resume_callback(etcp->input_send_q); - queue_resume_callback(etcp->input_wait_ack); - queue_resume_callback(etcp->recv_q); - - etcp->tx_state=ETCP_TX_STATE_DATA_WAIT; - - - // Сброс потока не меняет состояние физических линков. Владельцы получают REINIT - // после согласования сессии; DOWN/UP принадлежат только переходам link_status. - DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] stream reset: keeping links_up=%u", etcp->log_name, etcp->links_up); - - clear_queue(etcp->input_send_q); - clear_queue(etcp->input_wait_ack); - queue_resume_callback(etcp->input_send_q); - queue_resume_callback(etcp->input_wait_ack); - - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "end"); + DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] discard stream: unacked=%u send=%d wait=%d recv=%d output=%d", + conn->log_name, conn->unacked_bytes, queue_entry_count(conn->input_send_q), queue_entry_count(conn->input_wait_ack), + queue_entry_count(conn->recv_q), queue_entry_count(conn->output_queue)); + etcp_cancel_send(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; } + 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->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; + conn->bytes_sent_total = conn->retransmissions_count = conn->ack_packets_count = 0; + conn->rx_dup_count = conn->tx_dup_count = 0; + memset(conn->rtt_history, 0, sizeof(conn->rtt_history)); conn->rtt_history_idx = 0; + conn->last_rr_link = NULL; conn->reset_count++; + queue_resume_callback(conn->input_queue); queue_resume_callback(conn->output_queue); + queue_resume_callback(conn->input_send_q); queue_resume_callback(conn->input_wait_ack); queue_resume_callback(conn->recv_q); + conn->tx_state = ETCP_TX_STATE_DATA_WAIT; + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] stream discarded without replay: reset_count=%u links_up=%u attempts=%u", + conn->log_name, conn->reset_count, conn->links_up, conn->active_attempts); } void etcp_links_reset(struct ETCP_CONN* etcp) {// Если сбой в обмене или ребутнулась одна из сторон -> необходимо заново переинициализировать соединение @@ -721,15 +529,18 @@ void etcp_conn_reinit(struct ETCP_CONN* etcp, const char* reason) { // Фатальный ресет (normalizer desync / bad fragment size): меняем эпоху reset_id void etcp_conn_fatal_reinit(struct ETCP_CONN* etcp, const char* reason) { - if (!etcp || etcp->close_requested) return; + if (!etcp || etcp->close_requested || etcp->state == 2) return; + etcp->callbacks_running++; uint64_t new_id; do { if (random_bytes((uint8_t*)&new_id,8) != 0) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP,"[%s] reset epoch RNG failed",etcp->log_name); return; + DEBUG_ERROR(DEBUG_CATEGORY_ETCP,"[%s] reset epoch RNG failed; closing",etcp->log_name); + etcp_connection_close(etcp); etcp->callbacks_running--; return; } } while (!new_id || new_id == etcp->reset_id); - etcp_session_start(etcp); etcp_conn_reinit_id(etcp, reason, new_id); + if (!etcp->close_requested) etcp_session_start(etcp); + etcp->callbacks_running--; } void etcp_conn_reinit_id(struct ETCP_CONN* etcp, const char* reason, uint64_t reset_id) {// Если сбой в обмене или ребутнулась одна из сторон -> необходимо заново переинициализировать соединение @@ -745,17 +556,17 @@ void etcp_conn_reinit_id(struct ETCP_CONN* etcp, const char* reason, uint64_t re etcp->reinit_pending = 1; etcp->reset_done = 0; etcp->initialized = 0;// еще раз придёт conn_ready_callback + etcp->callbacks_running++; // Отменяем висящие NAT-ping'и для этого соединения if (etcp->instance && etcp->instance->nat_det) { nat_detection_cancel_for_conn(etcp->instance->nat_det, etcp); } - // Сбрасываем ретрансмиты роутера ДО очистки ETCP-очередей (избегаем гонки с data_pool) - if (etcp->peer_node_id) etcp_router_pause_retrans_for_node(etcp->instance, etcp->peer_node_id); - - // Вызываем etcp_conn_reset для сброса состояния + // Очистка не выделяет память; повторные передачи router остаются ответственностью его собственного inflight. etcp_conn_reset(etcp); + if (!etcp->close_requested && etcp->peer_node_id) etcp_router_pause_retrans_for_node(etcp->instance, etcp->peer_node_id); + etcp->callbacks_running--; DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "end"); } diff --git a/src/transport_layer/etcp.h b/src/transport_layer/etcp.h index 07aace13..65e06e74 100644 --- a/src/transport_layer/etcp.h +++ b/src/transport_layer/etcp.h @@ -296,6 +296,8 @@ int etcp_conn_ref_take(struct ETCP_CONN* conn); */ void etcp_conn_ref_free(struct ETCP_CONN* conn); +/* Очистить старый поток без выделений памяти/вызова callbacks очередей; DATA и ожидания отбрасываются. + * Эпохи/физические каналы сохраняются; это внутренняя очистка, внешний перезапуск — fatal_reinit(). */ void etcp_conn_reset(struct ETCP_CONN* etcp); void etcp_links_reset(struct ETCP_CONN* etcp); @@ -305,7 +307,9 @@ void etcp_conn_reinit(struct ETCP_CONN* etcp, const char* reason); * Не генерирует эпоху и не запускает подтверждение. При подтверждённой смене эпохи пира передают текущий etcp->reset_id. */ void etcp_conn_reinit_id(struct ETCP_CONN* etcp, const char* reason, uint64_t reset_id); /* Сгенерировать новую локальную эпоху, сбросить поток etcp и запустить подтверждение сессии. - * reason — причина для лога; DATA ждёт подтверждения, физические линки сохраняются. Возврата нет. */ + * reason — причина для лога; DATA ждёт подтверждения, физические линки сохраняются. Возврата нет. + * Старые DATA, в том числе ACKed транспортом, не воспроизводятся. После согласования владельцы получают REINIT + * и завершают/повторяют свои операции. При невозможности согласования (RNG/таймер/таймаут) соединение закрывается. */ void etcp_conn_fatal_reinit(struct ETCP_CONN* etcp, const char* reason); /* Учесть эпоху peer_reset_id из handshake соединения conn: первую сохранить, смену предложить для challenge. * Нулевая и прежняя эпохи игнорируются. Сброс при смене выполняется после CONFIRM, своя эпоха сохраняется. Возврата нет. */