Browse Source

Keep a finite RX deadline while ETCP keepalive sending is suspended

master
evgeny 19 hours ago
parent
commit
3cfdb9e6ec
  1. 2
      src/transport_layer/etcp_connections.h
  2. 86
      src/transport_layer/etcp_keepalive.c
  3. 3
      src/transport_layer/etcp_keepalive.h

2
src/transport_layer/etcp_connections.h

@ -225,6 +225,8 @@ struct ETCP_LINK {
// uint64_t last_activity; // Время последней активности // uint64_t last_activity; // Время последней активности
uint64_t last_recv_local_time; // x0.1 ms uint64_t last_recv_local_time; // x0.1 ms
uint64_t rx_deadline_tb; // конечный срок ожидания RX, включая объявленный сон
void* rx_timeout_timer; // проверка жизни независимо от отправки keepalive
uint16_t last_recv_timestamp; uint16_t last_recv_timestamp;
uint8_t last_recv_updated; // =1 при обновлении timestamp, =0 при отправке (чтобы не дублировать отправки при отсутствии обновлений) uint8_t last_recv_updated; // =1 при обновлении timestamp, =0 при отправке (чтобы не дублировать отправки при отсутствии обновлений)

86
src/transport_layer/etcp_keepalive.c

@ -8,6 +8,7 @@
#include "../lib/debug_config.h" #include "../lib/debug_config.h"
#include "../lib/memory_pool.h" #include "../lib/memory_pool.h"
#include <string.h> #include <string.h>
#include <limits.h>
#ifdef UTUN_HAVE_STANDBY #ifdef UTUN_HAVE_STANDBY
#include "standby.h" #include "standby.h"
#endif #endif
@ -129,7 +130,49 @@ static void restart_keepalive_timer_ms(struct ETCP_LINK* link, uint16_t ms) {
link->keepalive_timer = uasync_set_timeout(link->etcp->instance->ua, (uint32_t)ms * 10, link, keepalive_timer_cb, "link_keepalive"); link->keepalive_timer = uasync_set_timeout(link->etcp->instance->ua, (uint32_t)ms * 10, link, keepalive_timer_cb, "link_keepalive");
} }
// Keepalive timer callback // RX проверяется независимо от отправки keepalive, в том числе при объявленном сне.
static void ka_rx_timeout_cb(void* arg) {
struct ETCP_LINK* link = arg;
link->rx_timeout_timer = NULL;
if (link->etcp->close_requested || !link->recv_keepalive) return;
uint64_t now = get_time_tb();
if (now < link->rx_deadline_tb) {
uint64_t remaining = link->rx_deadline_tb - now;
int delay = remaining > INT_MAX ? INT_MAX : (int)remaining;
link->rx_timeout_timer = uasync_set_timeout(link->etcp->instance->ua, delay, link, ka_rx_timeout_cb, "link_rx_timeout");
if (!link->rx_timeout_timer)
DEBUG_ERROR(DEBUG_CATEGORY_KEEPALIVE, "[%s] cannot rearm RX timeout link=%u", link->etcp->log_name, link->local_link_id);
return;
}
DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "[%s] RX timeout link=%u server=%u tcp=%u age_ms=%llu my_sleep=%u peer_sleep=%u",
link->etcp->log_name, link->local_link_id, link->is_server, link->is_tcp,
(unsigned long long)((now - link->last_recv_local_time) / 10), link->ka_my_sleeping, link->ka_peer_sleeping);
ka_link_down(link); // callbacks могут удалить линк; дальше его не используем
}
// При RX меняется только deadline: уже поставленный таймер сам проверит продление.
void etcp_keepalive_watch_rx(struct ETCP_LINK* link) {
if (!link || !link->etcp || link->etcp->close_requested || !link->initialized || !link->recv_keepalive) return;
uint64_t timeout_ms = link->keepalive_timeout;
if (link->ka_my_sleeping || link->ka_peer_sleeping) {
uint64_t sleep_ms = ((uint64_t)link->ka_active_remaining + link->ka_sleep_duration) * 100;
sleep_ms += (uint64_t)link->keepalive_interval * KA_TIMEOUT_MULT;
if (sleep_ms > timeout_ms) timeout_ms = sleep_ms;
}
link->rx_deadline_tb = link->last_recv_local_time + timeout_ms * 10;
if (link->rx_timeout_timer) return;
uint64_t now = get_time_tb();
uint64_t remaining = link->rx_deadline_tb > now ? link->rx_deadline_tb - now : 1;
int delay = remaining > INT_MAX ? INT_MAX : (int)remaining;
link->rx_timeout_timer = uasync_set_timeout(link->etcp->instance->ua, delay, link, ka_rx_timeout_cb, "link_rx_timeout");
if (!link->rx_timeout_timer)
DEBUG_ERROR(DEBUG_CATEGORY_KEEPALIVE, "[%s] cannot schedule RX timeout link=%u", link->etcp->log_name, link->local_link_id);
else DEBUG_DEBUG(DEBUG_CATEGORY_KEEPALIVE, "[%s] RX watchdog started link=%u timeout_ms=%llu my_sleep=%u peer_sleep=%u",
link->etcp->log_name, link->local_link_id, (unsigned long long)timeout_ms,
link->ka_my_sleeping, link->ka_peer_sleeping);
}
// Таймер отправки keepalive и восстановления исходящих линков.
static void keepalive_timer_cb(void* arg) { static void keepalive_timer_cb(void* arg) {
DEBUG_TRACE(DEBUG_CATEGORY_KEEPALIVE, ""); DEBUG_TRACE(DEBUG_CATEGORY_KEEPALIVE, "");
struct ETCP_LINK* link = (struct ETCP_LINK*)arg; struct ETCP_LINK* link = (struct ETCP_LINK*)arg;
@ -167,38 +210,6 @@ static void keepalive_timer_cb(void* arg) {
goto restart_timer; goto restart_timer;
} }
// Check keepalive timeout
uint64_t now = get_time_tb();
uint64_t timeout_units = (uint64_t)link->keepalive_timeout * 10; // ms -> 0.1ms units
uint64_t elapsed = now - link->last_recv_local_time;
if (elapsed > timeout_units) {
if (link->recv_keepalive != 0) {
link->recv_keepalive = 0;
int old_link_status = link->link_status;
link->link_status = 0;
etcp_fire_link_status_cbk(link, link->link_state, old_link_status);
/* Снапшот до каскада: etcp_on_link_down может синхронно закрыть conn/линки
* (topo_recovery → ncd_revive_entry) — link может быть освобождён. */
struct ETCP_CONN *etcp = link->etcp;
int is_tcp = link->is_tcp;
int local_link_id = link->local_link_id;
struct sockaddr_storage raddr = link->remote_addr;
uint8_t remote_ka = link->remote_keepalive;
int link_state = link->link_state, initialized = link->initialized;
etcp_on_link_down(etcp, link);
if (etcp->state == 2) return; /* conn+link освобождены каскадом, link трогать нельзя */
/* каскад мог закрыть линк и без state==2 — проверить, что link ещё в списке */
int alive = 0;
for (struct ETCP_LINK *x = etcp->links; x; x = x->next) if (x == link) { alive = 1; break; }
if (!alive) return;
if (is_tcp && link->tcp_link) { stcp_link_close(link->tcp_link); link->tcp_link = NULL; }
if (old_link_status) {
DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "[%s] Link %d down: addr=%s ka=%d remote_ka=%d state=%d init=%d tmo: %llu>%llu els=%llums", etcp->log_name, local_link_id, sockaddr_storage_to_str(&raddr).str, 0, remote_ka, link_state, initialized, (unsigned long long)timeout_units, (unsigned long long)elapsed, (unsigned long long)(elapsed/10));
}
}
}
// Период: адаптивный от базового (normal-режим). // Период: адаптивный от базового (normal-режим).
if (link->pkt_sent_since_keepalive) if (link->pkt_sent_since_keepalive)
link->ka_period_ms = (uint16_t)link->keepalive_interval; link->ka_period_ms = (uint16_t)link->keepalive_interval;
@ -282,6 +293,9 @@ static uint8_t ka_announce_max_calc(struct ETCP_LINK* link) {
// Разрыв линка (мёртвый пир в спячке). Может освободить link — после вызова link не трогать. // Разрыв линка (мёртвый пир в спячке). Может освободить link — после вызова link не трогать.
static void ka_link_down(struct ETCP_LINK* link) { static void ka_link_down(struct ETCP_LINK* link) {
if (!link || !link->etcp) return; if (!link || !link->etcp) return;
if (link->rx_timeout_timer) {
uasync_cancel_timeout(link->etcp->instance->ua, link->rx_timeout_timer); link->rx_timeout_timer = NULL;
}
link->recv_keepalive = 0; link->recv_keepalive = 0;
int old_link_status = link->link_status; int old_link_status = link->link_status;
link->link_status = 0; link->link_status = 0;
@ -299,7 +313,7 @@ static void ka_link_down(struct ETCP_LINK* link) {
if (!alive) return; if (!alive) return;
if (is_tcp && link->tcp_link) { stcp_link_close(link->tcp_link); link->tcp_link = NULL; } if (is_tcp && link->tcp_link) { stcp_link_close(link->tcp_link); link->tcp_link = NULL; }
if (old_link_status) { if (old_link_status) {
DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "[%s] Link %d down (sleep unresponsive): addr=%s remote_ka=%d state=%d init=%d", DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "[%s] Link %d down (peer unresponsive): addr=%s remote_ka=%d state=%d init=%d",
etcp->log_name, local_link_id, sockaddr_storage_to_str(&raddr).str, remote_ka, link_state, initialized); etcp->log_name, local_link_id, sockaddr_storage_to_str(&raddr).str, remote_ka, link_state, initialized);
} }
} }
@ -371,6 +385,7 @@ static void ka_enter_sleep(struct ETCP_LINK* link) {
ka_stop_keepalive_timer(link); ka_stop_keepalive_timer(link);
DEBUG_INFO(DEBUG_CATEGORY_KEEPALIVE, "[%s] entering sleep (period=%dms)", link->etcp->log_name, period); DEBUG_INFO(DEBUG_CATEGORY_KEEPALIVE, "[%s] entering sleep (period=%dms)", link->etcp->log_name, period);
ka_announce_burst(link); ka_announce_burst(link);
etcp_keepalive_watch_rx(link);
} }
// Пробуждение (форграунд) — обычный keepalive, возврат к normal-режиму. // Пробуждение (форграунд) — обычный keepalive, возврат к normal-режиму.
@ -389,6 +404,7 @@ static void ka_wake(struct ETCP_LINK* link) {
etcp_link_send_keepalive(link); etcp_link_send_keepalive(link);
if (!link->ka_peer_sleeping) if (!link->ka_peer_sleeping)
restart_keepalive_timer_ms(link, link->ka_period_ms); restart_keepalive_timer_ms(link, link->ka_period_ms);
etcp_keepalive_watch_rx(link);
} }
typedef void (*ka_link_fn)(struct ETCP_LINK* link); typedef void (*ka_link_fn)(struct ETCP_LINK* link);
@ -555,6 +571,7 @@ int etcp_keepalive_on_recv(struct ETCP_SOCKET* e_sock, struct ETCP_DGRAM* pkt,
} }
link->keepalive_recv_count++; link->keepalive_recv_count++;
etcp_keepalive_watch_rx(link);
memory_pool_free(e_sock->instance->pkt_pool, pkt); memory_pool_free(e_sock->instance->pkt_pool, pkt);
return 1; return 1;
} }
@ -592,6 +609,7 @@ void etcp_keepalive_link_ready(struct ETCP_LINK* link) {
etcp_link_send_keepalive(link); etcp_link_send_keepalive(link);
if (!link->ka_peer_sleeping) start_keepalive_timer(link); if (!link->ka_peer_sleeping) start_keepalive_timer(link);
} }
etcp_keepalive_watch_rx(link);
} }
void etcp_keepalive_register(struct UTUN_INSTANCE* inst) { void etcp_keepalive_register(struct UTUN_INSTANCE* inst) {

3
src/transport_layer/etcp_keepalive.h

@ -59,6 +59,9 @@ uint16_t negotiate_keepalive(uint8_t my_type, uint16_t my_ka,
void etcp_link_send_keepalive(struct ETCP_LINK* link); void etcp_link_send_keepalive(struct ETCP_LINK* link);
void start_keepalive_timer(struct ETCP_LINK* link); void start_keepalive_timer(struct ETCP_LINK* link);
// Обновить срок ожидания RX после принятого пакета или изменения режима сна.
void etcp_keepalive_watch_rx(struct ETCP_LINK* link);
// --- новое --- // --- новое ---
// Обработка keepalive-пакетов (ETCP_KEEPALIVE) из etcp_packet_decrypted. // Обработка keepalive-пакетов (ETCP_KEEPALIVE) из etcp_packet_decrypted.
// Возвращает 1 если пакет обработан (и освобождён), 0 — если это не keepalive. // Возвращает 1 если пакет обработан (и освобождён), 0 — если это не keepalive.

Loading…
Cancel
Save