fix: reconnect test — cancel timers in conn_reset, add init_timer cancel on link_state=3, add reinit resilience to test
- etcp_conn_reset: cancel retrans_timer/ack_resp_timer via uasync_cancel_timeout before nulling (was leaking timer handles, confirmed by 'Timer leak: diff') - etcp_connections: cancel link->init_timer explicitly when link_state becomes 3 on both client and server INIT_RESPONSE paths (prevents spurious INIT after connection is established that could trigger cascading reinits) - etcp_connections: remove duplicate conn->session_id assignment in new conn path (session_id is set later via conn_reinit flow) - pkt_normalizer: resume output queue callback after pn_reset - Add test_etcp_reconnect with reinit detection — monitors conn->reinit_count and restarts data transfer if reinit occurs during active phase - Add etcp_dump.c/h — ETCP connection state dump utility for debugging
This commit is contained in:
Evgeny committed
1 parent
0f8df44b2b
commit
dea3a39598
8 files changed
+668
-5
No files matched your search
@@ -21,6 +21,7 @@ utun_CORE_SOURCES = \
|
||||
etcp_connections.c \
|
||||
etcp_loadbalancer.c \
|
||||
etcp_debug.c \
|
||||
etcp_dump.c \
|
||||
secure_channel.c \
|
||||
crc32.c \
|
||||
pkt_normalizer.c \
|
||||
|
||||
+9
-3
@@ -290,9 +290,15 @@ void etcp_conn_reset(struct ETCP_CONN* etcp) {
|
||||
l = l->next;
|
||||
}
|
||||
|
||||
// Reset timers (just clear the pointers - timers will expire naturally)
|
||||
etcp->retrans_timer = NULL;
|
||||
etcp->ack_resp_timer = NULL;
|
||||
// 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;
|
||||
}
|
||||
|
||||
etcp->reset_count++;
|
||||
|
||||
|
||||
@@ -1536,7 +1536,6 @@ void etcp_connections_read_callback_socket(socket_t sock, void* arg) {
|
||||
if (!conn) { errorcode=55; DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "failed to create connection"); goto ec_fr; }
|
||||
memcpy(&conn->crypto_ctx, &sc, sizeof(sc));
|
||||
conn->peer_node_id=peer_id;
|
||||
conn->session_id = session_id;
|
||||
etcp_update_log_name(conn);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "New connection received on socket %s: log_name=%s peer_id=%lu peer:%s", e_sock->name, conn->log_name, (unsigned long)peer_id, sockaddr_storage_to_str(&addr).str);
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTION, "New connection from %s peer_id=%ld etcp=%p", ip_to_str(&addr, addr.ss_family).str, peer_id, conn);
|
||||
@@ -1698,6 +1697,10 @@ void etcp_connections_read_callback_socket(socket_t sock, void* arg) {
|
||||
memory_pool_free(e_sock->instance->pkt_pool, pkt);
|
||||
link->initialized = 1;
|
||||
link->link_state = 3;
|
||||
if (link->init_timer) {
|
||||
uasync_cancel_timeout(link->etcp->instance->ua, link->init_timer);
|
||||
link->init_timer = NULL;
|
||||
}
|
||||
if (link->etcp->initialized == 0) {
|
||||
etcp_conn_ready(link->etcp);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "Connection established: log_name=%s socket=%s link_id=%d status=UP", link->etcp->log_name, e_sock->name, link->local_link_id);
|
||||
@@ -1826,6 +1829,10 @@ process_decrypted:
|
||||
|
||||
link->initialized = 1;// получен init response (client)
|
||||
link->link_state = 3; // connected
|
||||
if (link->init_timer) {
|
||||
uasync_cancel_timeout(link->etcp->instance->ua, link->init_timer);
|
||||
link->init_timer = NULL;
|
||||
}
|
||||
|
||||
if (code == ETCP_INIT_RESPONSE) {
|
||||
DEBUG_TRACE(DEBUG_CATEGORY_CONNECTION, "do reinit 3 %p", link->etcp);
|
||||
|
||||
+255
@@ -0,0 +1,255 @@
|
||||
#include "etcp_dump.h"
|
||||
#include "etcp.h"
|
||||
#include "pkt_normalizer.h"
|
||||
#include "../lib/ll_queue.h"
|
||||
#include "../lib/memory_pool.h"
|
||||
#include "../lib/socket_compat.h"
|
||||
#include "../lib/swm_min.h"
|
||||
#include <stdio.h>
|
||||
#include <string.h>
|
||||
|
||||
#define DUMP_FP stdout
|
||||
|
||||
#define DUMP_LINE(fmt, ...) do { \
|
||||
fprintf(DUMP_FP, " " fmt "\n", ##__VA_ARGS__); \
|
||||
} while(0)
|
||||
|
||||
#define DUMP_HDR(fmt, ...) do { \
|
||||
fprintf(DUMP_FP, fmt "\n", ##__VA_ARGS__); \
|
||||
} while(0)
|
||||
|
||||
static void dump_queue(const char* name, struct ll_queue* q) {
|
||||
if (!q) { DUMP_LINE("%-14s NULL", name); return; }
|
||||
DUMP_LINE("%-14s %d pkts / %zu bytes", name, queue_entry_count(q), queue_total_bytes(q));
|
||||
}
|
||||
|
||||
static void dump_pool(const char* name, struct memory_pool* pool) {
|
||||
if (!pool) return;
|
||||
size_t allocs, reuse;
|
||||
memory_pool_get_stats(pool, &allocs, &reuse);
|
||||
DUMP_LINE(" %-12s alloc=%zu reuse=%zu", name, allocs, reuse);
|
||||
}
|
||||
|
||||
static void dump_link_timer(const char* name, void* timer) {
|
||||
DUMP_LINE(" %-14s %s", name, timer ? "ACTIVE" : "free");
|
||||
}
|
||||
|
||||
static const char* ip_to_str_buf(uint32_t ip, char* buf, size_t sz) {
|
||||
uint8_t* b = (uint8_t*)&ip;
|
||||
snprintf(buf, sz, "%u.%u.%u.%u", b[0], b[1], b[2], b[3]);
|
||||
return buf;
|
||||
}
|
||||
|
||||
static const char* sockaddr_to_str(const struct sockaddr_storage* addr, char* buf, size_t sz) {
|
||||
if (!addr || addr->ss_family != AF_INET) { snprintf(buf, sz, "?"); return buf; }
|
||||
const struct sockaddr_in* sin = (const struct sockaddr_in*)addr;
|
||||
uint8_t* b = (uint8_t*)&sin->sin_addr.s_addr;
|
||||
snprintf(buf, sz, "%u.%u.%u.%u:%u", b[0], b[1], b[2], b[3], ntohs(sin->sin_port));
|
||||
return buf;
|
||||
}
|
||||
|
||||
void etcp_dump_conn_state(struct ETCP_CONN* conn) {
|
||||
if (!conn) { DUMP_HDR("etcp_dump_conn_state: NULL conn"); return; }
|
||||
|
||||
char addr_buf[32], nat_buf[32];
|
||||
|
||||
DUMP_HDR("=== ETCP CONN STATE [%s] ===", conn->log_name);
|
||||
|
||||
/* --- GENERAL --- */
|
||||
DUMP_LINE("GENERAL: %s peer=0x%llx initialized=%d links_up=%d tx_state=%d session=0x%08x mtu=%d routing_ex=%d",
|
||||
conn->name ? conn->name : "",
|
||||
(unsigned long long)conn->peer_node_id, conn->initialized, conn->links_up,
|
||||
conn->tx_state, conn->session_id, conn->mtu, conn->routing_exchange_active);
|
||||
|
||||
/* --- IDS --- */
|
||||
DUMP_LINE("IDS: next_tx=%u last_rx=%u last_del=%u rx_ack_till=%u",
|
||||
conn->next_tx_id, conn->last_rx_id, conn->last_delivered_id, conn->rx_ack_till);
|
||||
|
||||
/* --- RTT --- */
|
||||
DUMP_LINE("RTT: last=%u avg10=%u avg100=%u jitter=%u",
|
||||
conn->rtt_last, conn->rtt_avg_10, conn->rtt_avg_100, conn->jitter);
|
||||
|
||||
/* --- STATS --- */
|
||||
DUMP_LINE("STATS: bytes_sent=%u retrans=%u reinit=%u reset=%u ack_pkts=%u",
|
||||
conn->bytes_sent_total, conn->retransmissions_count,
|
||||
conn->reinit_count, conn->reset_count, conn->ack_packets_count);
|
||||
|
||||
/* --- INFLIGHT --- */
|
||||
DUMP_LINE("INFLIGHT: unacked=%u optimal=%u",
|
||||
conn->unacked_bytes, conn->optimal_inflight);
|
||||
|
||||
/* --- ACK DEBUG --- */
|
||||
DUMP_LINE("ACK_DEBUG: hit_inf=%u hit_sndq=%u miss=%u link_wait=%u rx_dup=%u tx_dup=%u",
|
||||
conn->cnt_ack_hit_inf, conn->cnt_ack_hit_sndq, conn->cnt_ack_miss,
|
||||
conn->cnt_link_wait, conn->rx_dup_count, conn->tx_dup_count);
|
||||
|
||||
/* --- DEBUG[8] --- */
|
||||
DUMP_LINE("DEBUG: [0]=%u [1]=%u [2]=%u [3]=%u [4]=%u [5]=%u [6]=%u [7]=%u",
|
||||
conn->debug[0], conn->debug[1], conn->debug[2], conn->debug[3],
|
||||
conn->debug[4], conn->debug[5], conn->debug[6], conn->debug[7]);
|
||||
|
||||
/* --- TIMERS --- */
|
||||
DUMP_LINE("TIMERS: retrans=%s ack_resp=%s",
|
||||
conn->retrans_timer ? "ACTIVE" : "free",
|
||||
conn->ack_resp_timer ? "ACTIVE" : "free");
|
||||
|
||||
/* --- QUEUES --- */
|
||||
DUMP_HDR(" --- QUEUES ---");
|
||||
dump_queue("input", conn->input_queue);
|
||||
dump_queue("input_send_q", conn->input_send_q);
|
||||
dump_queue("input_wait_ack", conn->input_wait_ack);
|
||||
dump_queue("ack_q", conn->ack_q);
|
||||
dump_queue("recv_q", conn->recv_q);
|
||||
dump_queue("output", conn->output_queue);
|
||||
|
||||
/* --- NORMALIZER --- */
|
||||
if (conn->normalizer) {
|
||||
struct PKTNORM* pn = (struct PKTNORM*)conn->normalizer;
|
||||
DUMP_HDR(" --- NORMALIZER ---");
|
||||
DUMP_LINE("frag=%u data_ptr=%u/%u flush=%s pending=%s recvpart=%s",
|
||||
pn->frag_size, pn->data_ptr, pn->data_size,
|
||||
pn->flush_timer ? "ACTIVE" : "free",
|
||||
pn->pending ? "ACTIVE" : "free",
|
||||
pn->recvpart ? "ACTIVE" : "free");
|
||||
dump_queue("pn-input", pn->input);
|
||||
dump_queue("pn-output", pn->output);
|
||||
DUMP_LINE("alloc_err=%u logic_err=%u in=%llu/%llu out=%llu/%llu",
|
||||
pn->alloc_errors, pn->logic_errors,
|
||||
(unsigned long long)pn->in_total_pkts, (unsigned long long)pn->in_total_bytes,
|
||||
(unsigned long long)pn->out_total_pkts, (unsigned long long)pn->out_total_bytes);
|
||||
} else {
|
||||
DUMP_HDR(" --- NORMALIZER --- NULL");
|
||||
}
|
||||
|
||||
/* --- POOLS --- */
|
||||
DUMP_HDR(" --- POOLS ---");
|
||||
dump_pool("inflight", conn->inflight_pool);
|
||||
dump_pool("io", conn->io_pool);
|
||||
if (conn->instance) {
|
||||
dump_pool("data", conn->instance->data_pool);
|
||||
dump_pool("ack", conn->instance->ack_pool);
|
||||
dump_pool("pkt", conn->instance->pkt_pool);
|
||||
}
|
||||
|
||||
/* --- LINKS --- */
|
||||
int link_idx = 0;
|
||||
struct ETCP_LINK* link = conn->links;
|
||||
while (link) {
|
||||
DUMP_HDR(" --- LINK %d ---", link_idx);
|
||||
|
||||
sockaddr_to_str(&link->remote_addr, addr_buf, sizeof(addr_buf));
|
||||
|
||||
/* basic link state */
|
||||
DUMP_LINE("BASIC: id=%d/%d is_server=%d link_state=%d link_status=%d initialized=%d mtu=%d",
|
||||
link->local_link_id, link->remote_link_id, link->is_server,
|
||||
link->link_state, link->link_status, link->initialized, link->mtu);
|
||||
DUMP_LINE("ADDR: remote=%s remote_sock=%d remote_type=%d remote_only_local=%d",
|
||||
addr_buf, link->remote_socket_id, link->remote_type, link->remote_only_local);
|
||||
|
||||
/* keepalive */
|
||||
DUMP_LINE("KA: recv=%d remote=%d sent=%u recv=%u interval=%u timeout=%u",
|
||||
link->recv_keepalive, link->remote_keepalive,
|
||||
link->keepalive_sent_count, link->keepalive_recv_count,
|
||||
link->keepalive_interval, link->keepalive_timeout);
|
||||
|
||||
/* timers */
|
||||
DUMP_LINE("TIMERS: init=%s(%u/%u) ka=%s shaper=%s stats=%s burst=%s keepalive=%s",
|
||||
link->init_timer ? "ACTIVE" : "free",
|
||||
link->init_timeout, link->init_retry_count,
|
||||
link->keepalive_timer ? "ACTIVE" : "free",
|
||||
link->shaper_timer ? "ACTIVE" : "free",
|
||||
link->stats_timer ? "ACTIVE" : "free",
|
||||
link->burst_resp_timer ? "ACTIVE" : "free",
|
||||
link->keepalive_sent_count > 0 ? "has_sent" : "idle");
|
||||
|
||||
/* inflight */
|
||||
DUMP_LINE("INFLIGHT: bytes=%u pkts=%u lim=%u blocked=%d phase=%d sst=%u last_win_tb=%llu",
|
||||
link->inflight_bytes, link->inflight_packets, link->inflight_lim_bytes,
|
||||
link->send_blocked_inflight, link->inflight_phase,
|
||||
link->slow_start_threshold, (unsigned long long)link->last_window_update_tb);
|
||||
|
||||
/* rtt */
|
||||
char hist_str[128] = "";
|
||||
int hist_pos = 0;
|
||||
for (int i = 0; i < 10 && i < (int)link->rtt_history_count; i++) {
|
||||
hist_pos += snprintf(hist_str + hist_pos, sizeof(hist_str) - hist_pos,
|
||||
"%s%u", i == 0 ? "" : ",", link->rtt_history[i]);
|
||||
}
|
||||
DUMP_LINE("RTT: last=%u avg10=%u min=%u max=%u jitter=%u hist=[%s] cnt=%d swm=%u",
|
||||
link->rtt_last, link->rtt_avg10, link->rtt_min, link->rtt_max_val,
|
||||
link->jitter, hist_str, link->rtt_history_count,
|
||||
link->rtt_swm ? swm_get_min(link->rtt_swm) : 0);
|
||||
|
||||
/* tt/rt/bandwidth */
|
||||
DUMP_LINE("TT/RT: tt=%u rt=%u recv_dt_tx=%u recv_dt_rx=%u bw=%u",
|
||||
link->tt_last, link->rt_last,
|
||||
link->recv_dt_avg_tx, link->recv_dt_avg_rx,
|
||||
link->bandwidth);
|
||||
|
||||
/* errors */
|
||||
DUMP_LINE("MTU: %u local=%u remote=%u", link->mtu, link->mtu_local, link->mtu_remote);
|
||||
DUMP_LINE("ERRORS: enc=%zu dec=%zu snd=%zu rcv=%zu total_enc=%zu total_dec=%zu retrans=%u",
|
||||
link->encrypt_errors, link->decrypt_errors,
|
||||
link->send_errors, link->recv_errors,
|
||||
link->total_encrypted, link->total_decrypted,
|
||||
link->total_retransmissions);
|
||||
|
||||
/* nat */
|
||||
ip_to_str_buf(link->nat_ip, nat_buf, sizeof(nat_buf));
|
||||
DUMP_LINE("NAT: ip=%s:%u chg=%u hits=%u check=%d type=%d",
|
||||
nat_buf, link->nat_port,
|
||||
link->nat_changes_count, link->nat_hits_count,
|
||||
link->nat_check_status, link->nat_type);
|
||||
|
||||
/* stats win */
|
||||
DUMP_LINE("WIN: ptr=%u tb=%u tx=%u retrans=%u",
|
||||
link->win_ptr, link->win_timebase,
|
||||
link->window_pkt_transmitted, link->window_retransmissions);
|
||||
|
||||
/* last recv */
|
||||
DUMP_LINE("LAST_RECV: time=%llu ts=%u updated=%d",
|
||||
(unsigned long long)link->last_recv_local_time,
|
||||
link->last_recv_timestamp, link->last_recv_updated);
|
||||
|
||||
/* handshake */
|
||||
DUMP_LINE("HANDSHAKE: min=%u max=%u", link->handshake_minsize, link->handshake_maxsize);
|
||||
|
||||
/* burst */
|
||||
DUMP_LINE("BURST: active=%d id=%u seq=%u/%u last_tb=%llu resp=%s",
|
||||
link->burst_active, link->burst_id,
|
||||
link->burst_seq, link->burst_count,
|
||||
(unsigned long long)link->burst_last_time_tb,
|
||||
link->burst_resp_timer ? "ACTIVE" : "free");
|
||||
|
||||
/* shaper */
|
||||
DUMP_LINE("SHAPER: load_tb=%llu sub=%llu state=%d",
|
||||
(unsigned long long)link->shaper_load_time_tb,
|
||||
(unsigned long long)link->shaper_sub_nanotime,
|
||||
link->shaper_state);
|
||||
|
||||
link = link->next;
|
||||
link_idx++;
|
||||
}
|
||||
if (link_idx == 0) DUMP_HDR(" --- NO LINKS ---");
|
||||
|
||||
DUMP_HDR("=== END [%s] ===", conn->log_name);
|
||||
}
|
||||
|
||||
void etcp_dump_all_conns(struct UTUN_INSTANCE* instance) {
|
||||
if (!instance) {
|
||||
DUMP_HDR("etcp_dump_all_conns: NULL instance");
|
||||
return;
|
||||
}
|
||||
DUMP_HDR("=== DUMP ALL CONNS for instance node=0x%llx ===",
|
||||
(unsigned long long)instance->node_id);
|
||||
struct ETCP_CONN* conn = instance->connections;
|
||||
int idx = 0;
|
||||
while (conn) {
|
||||
DUMP_HDR("--- CONN %d ---", idx);
|
||||
etcp_dump_conn_state(conn);
|
||||
conn = conn->next;
|
||||
idx++;
|
||||
}
|
||||
if (idx == 0) DUMP_HDR("--- NO CONNECTIONS ---");
|
||||
DUMP_HDR("=== END DUMP ALL ===");
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
#ifndef ETCP_DUMP_H
|
||||
#define ETCP_DUMP_H
|
||||
|
||||
#include "etcp.h"
|
||||
#include "utun_instance.h"
|
||||
|
||||
void etcp_dump_conn_state(struct ETCP_CONN* conn);
|
||||
void etcp_dump_all_conns(struct UTUN_INSTANCE* instance);
|
||||
|
||||
#endif
|
||||
@@ -180,6 +180,7 @@ void pn_reset(struct PKTNORM* pn) {
|
||||
queue_entry_free(entry);
|
||||
}
|
||||
queue_resume_callback(pn->input);
|
||||
queue_resume_callback(pn->output);
|
||||
}
|
||||
|
||||
// Send data to packer (copies and adds to input queue or pending, triggering callback) используется только в юниттесте
|
||||
|
||||
Reference in new issue
Block a user