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

Guard TCP link transitions across callbacks and unify reconnect backoff

Browse Source
This commit is contained in:
evgeny committed 2026-10-23 21:48:11 +02:00
1 parent 42e7b9b1b6
commit f37221107d
2 files changed
+76 -47
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/transport_layer/etcp.h
-1
View File
@@ -121,7 +121,6 @@ struct ETCP_CONN {
// Links (channels) - linked list
struct ETCP_LINK* links;
struct ETCP_LINK* last_rr_link; // последний линк, выбранный round-robin
uint8_t tcp_link_count; // количество UP TCP линков
uint8_t last_assigned_link_id; // последний выданный link_id (для циклического перебора)
struct ETCP_PADDING padding; // диапазон и независимый таймер всех линков этого соединения
src/transport_layer/etcp_connections.c
+76 -46
View File
@@ -65,6 +65,7 @@ static int socket_identity_init(struct ETCP_SOCKET *socket, struct UTUN_INSTANCE
* Только static-функции, используемые раньше своего определения.
* Не-static функции уже объявлены в etcp_connections.h. */
static void tcp_link_close_cb(struct stcp_link *sl, int err, void *arg);
static int etcp_link_rx_alive(struct ETCP_CONN* conn, struct ETCP_LINK* link);
/* Wire fields may start at any byte offset. */
static uint16_t wire_read16(const void* p) { uint16_t v; memcpy(&v,p,2); return be16toh(v); }
static uint64_t wire_read64(const void* p) { uint64_t v; memcpy(&v,p,8); return be64toh(v); }
@@ -1188,7 +1189,12 @@ static void etcp_inflight_nullify_link(struct ll_queue* q, struct ETCP_LINK* dea
// зануление ссылок inflight (защита от use-after-free) и освобождение памяти.
void etcp_link_close(struct ETCP_LINK* link) {
DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "");
if (!link) return;
if (!link || link->closing) return;
link->closing = 1;
struct ETCP_CONN *etcp = link->etcp;
etcp->callbacks_running++;
int old_state = link->link_state, old_status = link->link_status;
link->link_status = link->recv_keepalive = link->initialized = 0;
if (link->rx_timeout_timer) {
uasync_cancel_timeout(link->etcp->instance->ua, link->rx_timeout_timer); link->rx_timeout_timer = NULL;
}
@@ -1205,22 +1211,8 @@ void etcp_link_close(struct ETCP_LINK* link) {
uasync_cancel_timeout(link->etcp->instance->ua, link->burst_resp_timer);
link->burst_resp_timer = NULL;
}
etcp_tx_link_destroy(link);
if (!link->conn) {
struct ETCP_LINK **pp = &link->etcp->links;
while (*pp && *pp != link) pp = &(*pp)->next;
if (*pp) *pp = link->next;
if (link->init_timer) uasync_cancel_timeout(link->etcp->instance->ua, link->init_timer);
if (link->shaper_timer) uasync_cancel_timeout(link->etcp->instance->ua, link->shaper_timer);
if (link->keepalive_timer) uasync_cancel_timeout(link->etcp->instance->ua, link->keepalive_timer);
etcp_conn_on_inflight_lim_changed(link->etcp);
etcp_inflight_nullify_link(link->etcp->input_wait_ack, link);
etcp_inflight_nullify_link(link->etcp->input_send_q, link);
callring_put(CR_LINK_CLOSE_FREE, (uintptr_t)link, 0);
u_free(link->bbr);
u_free(link);
return;
if (link->ka_sleep_announce_timer) {
uasync_cancel_timeout(etcp->instance->ua, link->ka_sleep_announce_timer); link->ka_sleep_announce_timer = NULL;
}
// Cancel init timer if active
@@ -1252,18 +1244,19 @@ void etcp_link_close(struct ETCP_LINK* link) {
}
remove_link_from_queue(link);
etcp_tx_link_destroy(link);
etcp_conn_on_inflight_lim_changed(link->etcp);
DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "[%s] Link %d closed: addr=%s rcvd=%zub ack=%llub infl=%ub/%upkt",
link->etcp->log_name, link->local_link_id, sockaddr_storage_to_str(&link->remote_addr).str, link->total_decrypted,
(unsigned long long)link->acked_bytes, link->inflight_bytes, link->inflight_packets);
etcp_fire_link_status_cbk(link, link->link_state, link->link_status);
etcp_inflight_nullify_link(link->etcp->input_wait_ack, link);
etcp_inflight_nullify_link(link->etcp->input_send_q, link);
etcp_on_link_down(link->etcp, link);
etcp_fire_link_status_cbk(link, old_state, old_status);
if (!etcp->close_requested) etcp_on_link_down(etcp, NULL);
callring_put(CR_LINK_CLOSE_FREE, (uintptr_t)link, 0);
u_free(link->bbr);
u_free(link);
etcp->callbacks_running--;
}
// ====== TCP link helpers ======
@@ -1291,17 +1284,26 @@ static void tcp_send_observed_ip(struct ETCP_LINK *link) {
memory_pool_free(inst->pkt_pool, pkt);
}
/* STCP-handle живёт до следующего poll; его отвязка обнаруживает удаление/замену линка, включая переиспользование адреса. */
static int tcp_link_attached(struct ETCP_CONN *conn, struct ETCP_LINK *link, struct stcp_link *transport) {
if (conn->close_requested || conn->state == 2) return 0;
if (transport && stcp_link_get_etcp_link(transport) != link) return 0;
return etcp_link_rx_alive(conn, link) && !link->closing && link->tcp_link == transport;
}
// Поднять TCP-линк в ready: синхронизация reset-эпохи (gop/rid), keepalive, ed25519,
// mark initialized + link_status=1, уведомить коннект и loadbalancer.
void etcp_link_enter_ready_tcp(struct ETCP_LINK *link) {
if (!link || !link->etcp) return;
if (!link || !link->etcp || link->closing || link->etcp->close_requested) return;
struct ETCP_CONN *etcp = link->etcp;
struct stcp_link *transport = link->tcp_link;
etcp->callbacks_running++;
/* reinit только при смене reset_id пира — синхронизация через etcp_conn_apply_peer_reset_id */
if (link->tcp_link) {
uint64_t peer_rid = stcp_link_get_peer_reset_id(link->tcp_link);
etcp_conn_apply_peer_reset_id(etcp, peer_rid);
if (etcp->state == 2) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "enter_ready_tcp: conn DELETED after apply_reset, link=%p", (void*)link); return; }
if (!tcp_link_attached(etcp, link, transport)) goto done;
link->peer_device_type = stcp_link_get_peer_device_type(link->tcp_link);
{ uint16_t peer_ka = stcp_link_get_peer_keepalive_interval(link->tcp_link);
@@ -1309,25 +1311,32 @@ void etcp_link_enter_ready_tcp(struct ETCP_LINK *link) {
etcp->instance->keepalive_interval, link->peer_device_type, peer_ka); }
}
int old_state = link->link_state;
int old_state = link->link_state, old_status = link->link_status;
link->initialized = 1; link->link_state = LINK_STATE_CONNECTED; link->link_status = 1;
link->recv_keepalive = 1;
link->last_recv_local_time = get_time_tb();
if (!link->mtu_remote) link->mtu_remote = link->mtu;
etcp->tcp_link_count++;
link->tcp_reconnect_delay_ms = 0;
if (link->tcp_link) { const uint8_t* ed = stcp_link_get_peer_ed25519_pubkey(link->tcp_link); if (ed) memcpy(etcp->peer_ed25519_pubkey, ed, SC_PUBKEY_SIZE); }
// Известная эпоха пира не подтверждает новый поток; следующий линк не перезапускает согласование.
if (!etcp->initialized && !etcp->session_required) etcp_session_start(etcp);
if (!tcp_link_attached(etcp, link, transport)) goto done;
if (etcp->initialized == 0) etcp_conn_ready(etcp);
if (etcp->state == 2) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "enter_ready_tcp: conn DELETED after conn_ready, link=%p", (void*)link); return; }
if (!tcp_link_attached(etcp, link, transport)) goto done;
etcp_keepalive_link_ready(link);
if (!tcp_link_attached(etcp, link, transport)) goto done;
loadbalancer_link_ready(link);
etcp_fire_link_status_cbk(link, old_state, link->link_status);
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] TCP link %d UP (mtu=%d init=%d up=%d tcp_links=%d)",
if (!tcp_link_attached(etcp, link, transport)) goto done;
etcp_fire_link_status_cbk(link, old_state, old_status);
if (!tcp_link_attached(etcp, link, transport)) goto done;
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] TCP link %d UP (mtu=%d init=%d up=%d)",
etcp->log_name, link->local_link_id, link->mtu,
etcp->initialized, etcp->links_up, etcp->tcp_link_count);
etcp->initialized, etcp->links_up);
if (etcp->instance->nat_det) nat_detection_link_ready(etcp->instance->nat_det, link);
if (!tcp_link_attached(etcp, link, transport)) goto done;
tcp_send_observed_ip(link);
done:
etcp->callbacks_running--;
}
// Повторная попытка STCP-подключения с экспоненциальным backoff (до 30 с).
@@ -1335,24 +1344,27 @@ static void tcp_link_reconnect_cb(void *arg) {
struct ETCP_LINK *link = (struct ETCP_LINK *)arg;
if (!link || !link->etcp || !link->etcp->instance) return;
link->tcp_reconnect_timer = NULL;
if (link->tcp_reconnect_delay_ms == 0) link->tcp_reconnect_delay_ms = 1000;
if (link->closing || link->etcp->close_requested || link->etcp->state == 2) return;
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] TCP link %d reconnect attempt (delay=%ums)", link->etcp->log_name, link->local_link_id, link->tcp_reconnect_delay_ms);
uint16_t port = ntohs(((struct sockaddr_in *)&link->remote_addr)->sin_port);
struct stcp_link *sl = stcp_link_connect(link, &link->remote_addr, port);
if (!sl) { link->tcp_reconnect_delay_ms *= 2; if (link->tcp_reconnect_delay_ms > 30000) link->tcp_reconnect_delay_ms = 30000;
link->tcp_reconnect_timer = uasync_set_timeout(link->etcp->instance->ua, (int)(link->tcp_reconnect_delay_ms * 10), link, tcp_link_reconnect_cb, "tcp_rct"); return; }
link->tcp_link = sl;
stcp_link_set_on_close(sl, tcp_link_close_cb, link);
etcp_tcp_link_start_connect(link, &link->remote_addr, ss_get_port(&link->remote_addr));
}
// Исходящее TCP-подключение линка: stcp_link_connect + установка close-callback.
void etcp_tcp_link_start_connect(struct ETCP_LINK *link, struct sockaddr_storage *addr, uint16_t port) {
if (!link || !link->etcp || !addr) return;
if (!link || !link->etcp || !addr || link->closing || link->etcp->close_requested || link->etcp->state == 2) return;
if (link->tcp_link || link->tcp_reconnect_timer) {
DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] TCP connect already pending path=%u", link->etcp->log_name, link->local_link_id);
return;
}
/* link->conn опционален: NULL = без локального bind (ОС выбирает source IP/интерфейс).
* Задан — bind к interface_addr этого сокета (выбор интерфейса/ip). */
memcpy(&link->remote_addr, addr, sizeof(*addr));
if (addr != &link->remote_addr) memcpy(&link->remote_addr, addr, sizeof(*addr));
link->link_state = LINK_STATE_HANDSHAKE;
struct stcp_link *sl = stcp_link_connect(link, &link->remote_addr, port);
if (!sl) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "etcp_tcp_link_start_connect: stcp_link_connect failed"); return; }
if (!sl) {
DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] TCP connect startup failed path=%u", link->etcp->log_name, link->local_link_id);
etcp_tcp_link_start_reconnect(link); return;
}
link->tcp_link = sl;
if (link->is_server == 0) stcp_link_set_on_close(sl, tcp_link_close_cb, link);
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] TCP link %d → stcp_link_connect %s:%u rc=%p bind=%s",
@@ -1363,25 +1375,39 @@ void etcp_tcp_link_start_connect(struct ETCP_LINK *link, struct sockaddr_storage
// Запланировать реконнект TCP-линка после обрыва (через tcp_reconnect_timer).
void etcp_tcp_link_start_reconnect(struct ETCP_LINK *link) {
if (!link || !link->etcp || !link->etcp->instance || link->is_server) return;
if (link->etcp->state == 2) return;
if (link->closing || link->etcp->close_requested || link->etcp->state == 2) return;
if (link->tcp_reconnect_timer) return;
if (link->tcp_link) { stcp_link_close(link->tcp_link); link->tcp_link = NULL; }
link->initialized = link->recv_keepalive = link->link_status = 0;
link->link_state = LINK_STATE_HANDSHAKE;
if (link->etcp->tcp_link_count > 0) link->etcp->tcp_link_count--;
uint32_t previous = link->tcp_reconnect_delay_ms;
link->tcp_reconnect_delay_ms = !previous ? 1000 : previous >= 15000 ? 30000 : previous * 2;
link->tcp_reconnect_timer = uasync_set_timeout(link->etcp->instance->ua, (int)(link->tcp_reconnect_delay_ms * 10), link, tcp_link_reconnect_cb, "tcp_rct");
if (!link->tcp_reconnect_timer) {
struct ETCP_CONN *conn = link->etcp;
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] TCP reconnect timer allocation failed path=%u", conn->log_name, link->local_link_id);
conn->callbacks_running++; etcp_connection_close(conn); conn->callbacks_running--;
return;
}
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] TCP reconnect scheduled path=%u delay_ms=%u previous_ms=%u",
link->etcp->log_name, link->local_link_id, link->tcp_reconnect_delay_ms, previous);
}
// Callback закрытия STCP-линка: серверный линк закрываем, клиентский — отправляем на реконнект.
static void tcp_link_close_cb(struct stcp_link *sl, int err, void *arg) {
struct ETCP_LINK *link = (struct ETCP_LINK *)arg;
callring_put(CR_TCPCB_IN, (uintptr_t)sl, (uintptr_t)arg);
if (!link || !link->etcp) return;
if (link->etcp->state == 2) return;
if (!link || !link->etcp || link->closing) return;
if (link->tcp_link != sl) {
DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] stale TCP close ignored path=%u", link->etcp->log_name, link->local_link_id);
stcp_link_close(sl); return;
}
/* Сохраняем до каскада: etcp_on_link_down может закрыть conn и освободить link */
int old_state = link->link_state;
int old_state = link->link_state, old_status = link->link_status;
int is_server = link->is_server;
int local_link_id = link->local_link_id;
struct ETCP_CONN *etcp = link->etcp;
etcp->callbacks_running++;
stcp_link_close(sl);
link->tcp_link = NULL;
@@ -1395,11 +1421,15 @@ static void tcp_link_close_cb(struct stcp_link *sl, int err, void *arg) {
callring_put(CR_TCPCB_FREE, (uintptr_t)link, 0);
etcp_link_close(link);
} else {
etcp_fire_link_status_cbk(link, old_state, 0);
link->initialized = link->recv_keepalive = link->link_status = 0;
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] TCP link %d down err=%d, scheduling reconnect", etcp->log_name, local_link_id, err);
etcp_on_link_down(etcp, link);
if (etcp->state != 2) etcp_tcp_link_start_reconnect(link);
etcp_tcp_link_start_reconnect(link);
if (!etcp->close_requested && (old_state != link->link_state || old_status))
etcp_fire_link_status_cbk(link, old_state, old_status);
/* Callback мог удалить link или соседний путь; агрегат пересчитывается только по живому списку. */
if (!etcp->close_requested) etcp_on_link_down(etcp, NULL);
}
etcp->callbacks_running--;
}
// Единый plaintext для UDP/STCP. Эпохи DATA/ACK подписываются вместе с содержимым.