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

Discard failed ETCP streams without allocating or replaying fragments

Browse Source
This commit is contained in:
evgeny committed 2026-10-23 12:22:22 +02:00
1 parent 466a56b499
commit 5f085b2086
2 files changed
+47 -232
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/transport_layer/etcp.c
+42 -231
View File
@@ -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");
}
src/transport_layer/etcp.h
+5 -1
View File
@@ -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, своя эпоха сохраняется. Возврата нет. */