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

Test incoming ETCP expiry, reconnect, sleep and graceful close

master
evgeny 14 hours ago
parent
f090d697c4
commit
0ba8380db4
3 changed files with 195 additions and 4 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 5
      src/transport_layer/etcp.c
  2. 2
      src/transport_layer/etcp_connections.c
  3. 192
      tests/test_etcp_server_lifecycle.c

5
src/transport_layer/etcp.c
Unescape View File

@ -215,7 +215,6 @@ static int etcp_down_timer_start(struct ETCP_CONN* conn) {
conn->down_timer = uasync_set_timeout(conn->instance->ua, ETCP_DOWN_TIMEOUT_TB, conn, etcp_down_timeout_cb, "etcp_down");
if (!conn->down_timer) {
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "[%s] cannot schedule DOWN timeout", conn->log_name);
etcp_connection_close(conn);
return -1;
}
DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTION, "[%s] DOWN timer started timeout_ms=%d refs=%d",
@ -317,7 +316,7 @@ struct ETCP_CONN* etcp_connection_create(struct UTUN_INSTANCE* instance, char* n
return NULL;
}
if (etcp_down_timer_start(etcp) < 0) return NULL;
if (etcp_down_timer_start(etcp) < 0) { etcp_connection_close(etcp); return NULL; }
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] connection initialized. ETCP=%p mtu=%d, next_tx_id=%u state=pending",
etcp->log_name, etcp, etcp->mtu, etcp->next_tx_id);
@ -1113,7 +1112,7 @@ void etcp_on_link_down(struct ETCP_CONN* etcp, struct ETCP_LINK* down_link) {
}
int was_up = etcp->links_up;
etcp->links_up = up;
if (!up) etcp_down_timer_start(etcp);
if (!up && etcp_down_timer_start(etcp) < 0) { etcp_connection_close(etcp); return; }
else etcp_down_timer_cancel(etcp);
if (up == 0 && was_up != 0) {
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "All links fall down");

2
src/transport_layer/etcp_connections.c
Unescape View File

@ -2906,7 +2906,7 @@ int init_connections(struct UTUN_INSTANCE* instance) {
}
}
node_conn_direct_init(instance);
if (node_conn_direct_init(instance) < 0) return -1;
etcp_keepalive_register(instance);
// Return 1 if there was a partial socket initialization error

192
tests/test_etcp_server_lifecycle.c
Unescape View File

@ -0,0 +1,192 @@
/* Входящий ETCP: реальные очереди, таймеры, NCD и управляемые часы без сетевого I/O. */
#include <stdlib.h>
#include <string.h>
#include "etcp.h"
#include "etcp_api.h"
#include "etcp_keepalive.h"
#include "node_conn_direct.h"
#include "utun_instance.h"
#include "config_parser.h"
#include "test_clock.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#define CHECK(x) do { if (!(x)) { DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "FAIL %s:%d: %s", __func__, __LINE__, #x); abort(); } } while (0)
static struct UTUN_INSTANCE inst;
static struct utun_config cfg;
static struct ETCP_SOCKET sock;
static unsigned deleted, closed, down;
static int release_on_down;
static void deleted_cb(struct ETCP_CONN* conn, int event, void* arg) {
(void)conn; (void)event; (void)arg; deleted++;
}
/* Освобождение последнего владельца внутри DOWN не сокращает срок сохранения ETCP. */
static void owner_cb(struct NODE_CONN_DIRECT* handle, enum ncd_event event, void* arg) {
(void)arg;
if (event == NCD_EVENT_DOWN) {
down++;
if (release_on_down) node_conn_direct_close(handle);
} else if (event == NCD_EVENT_CLOSED) {
closed++; CHECK(node_conn_direct_get_conn(handle) == NULL);
}
}
static void setup(void) {
memset(&inst, 0, sizeof(inst)); memset(&cfg, 0, sizeof(cfg)); memset(&sock, 0, sizeof(sock));
deleted = closed = down = 0; release_on_down = 0;
inst.ua = uasync_create(); CHECK(inst.ua); inst.config = &cfg;
cfg.global.bbr_max_cwnd = 100000; cfg.global.keepalive_interval = 200; cfg.global.keepalive_timeout = 2000;
inst.connections = queue_new(inst.ua, 16, 0, 8, "server lifecycle"); CHECK(inst.connections);
inst.pkt_pool = memory_pool_init(sizeof(struct ETCP_DGRAM), "server lifecycle packets"); CHECK(inst.pkt_pool);
sock.instance = &inst; strcpy(sock.name, "synthetic server");
CHECK(init_connections(&inst) == 0); // чисто входящий узел тоже принимает NCD CLOSE
CHECK(inst.ncd_control_bound && !inst.ncd_registry);
}
static void teardown(void) {
while (inst.connections->head) {
struct conn_queue_entry* entry = (void*)inst.connections->head->data;
etcp_connection_close(entry->conn);
}
uasync_drain_immediate(inst.ua); CHECK(!inst.ncd_registry);
while (inst.activity_cbks) utun_remove_activity_cbk(&inst, inst.activity_cbks->fn, inst.activity_cbks->arg);
queue_free(inst.connections); memory_pool_destroy(inst.pkt_pool); uasync_poll(inst.ua, 0);
CHECK(inst.ua->timer_alloc_count == inst.ua->timer_free_count);
uasync_destroy(inst.ua, 0);
}
static struct ETCP_CONN* new_conn(void) {
struct ETCP_CONN* conn = etcp_connection_create(&inst, "incoming"); CHECK(conn);
conn->peer_node_id = 42; etcp_conn_add_cbk(conn, deleted_cb, NULL, ETCP_CBK_EVENT_DELETE);
queue_set_callback(conn->send_input_q, NULL, NULL);
return conn;
}
/* Фикстура владеет только сокетом; линки выделены штатно освобождаемой памятью ETCP. */
static struct ETCP_LINK* add_link(struct ETCP_CONN* conn, int tcp) {
struct ETCP_LINK* link = u_calloc(1, sizeof(*link)); CHECK(link);
link->etcp = conn; link->conn = &sock; link->is_server = 1; link->is_tcp = tcp;
link->initialized = link->link_status = link->recv_keepalive = link->remote_keepalive = 1;
link->link_state = LINK_STATE_CONNECTED; link->keepalive_interval = link->ka_period_ms = 200;
link->keepalive_timeout = 2000; link->last_recv_local_time = get_time_tb();
link->local_link_id = conn->links ? conn->links->local_link_id + 1 : 1;
link->next = conn->links; conn->links = link;
return link;
}
static struct ETCP_CONN* ready_conn(int tcp) {
struct ETCP_CONN* conn = new_conn(); add_link(conn, tcp); etcp_conn_ready(conn);
CHECK(conn->links_up && !conn->down_timer);
return conn;
}
static void receive_control(struct ETCP_CONN* conn, uint8_t subcmd) {
struct ncd_control_msg msg = { ETCP_RT_ID_NCD_CONTROL, subcmd, conn->peer_node_id };
struct ll_entry* entry = ll_alloc_lldgram(sizeof(msg)); CHECK(entry);
memcpy(entry->dgram, &msg, sizeof(msg)); entry->len = sizeof(msg);
inst.api_bindings.callbacks[ETCP_RT_ID_NCD_CONTROL](conn, entry);
uasync_drain_immediate(inst.ua);
}
/* Pending имеет конечный срок; внешняя ссылка задерживает только освобождение памяти. */
static void check_pending(void) {
setup(); struct ETCP_CONN* conn = new_conn(); CHECK(etcp_conn_ref_take(conn) == 0);
test_clock_run_for(inst.ua, ETCP_DOWN_TIMEOUT_TB - 10000); CHECK(!deleted && inst.connections->count == 1);
test_clock_run_for(inst.ua, 20000); CHECK(deleted == 1 && conn->state == 2 && !conn->down_timer);
CHECK(!inst.connections->count && !conn->free_scheduled);
etcp_conn_ref_free(conn); teardown();
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS initial DOWN timeout and external reference");
}
/* Два пути, потеря последнего, неизменный deadline, восстановление и новая потеря. */
static void check_reconnect(void) {
setup(); struct ETCP_CONN* conn = ready_conn(0); struct ETCP_LINK* second = add_link(conn, 1);
etcp_link_close(second); CHECK(conn->links_up && !conn->down_timer);
conn->links->link_status = 0; etcp_on_link_down(conn, conn->links); CHECK(conn->down_timer);
void* timer = conn->down_timer;
test_clock_run_for(inst.ua, 200000); etcp_on_link_down(conn, conn->links); CHECK(conn->down_timer == timer);
conn->links->link_status = 1; conn->link_ready_for_send_fn(conn); CHECK(conn->links_up && !conn->down_timer);
test_clock_run_for(inst.ua, ETCP_DOWN_TIMEOUT_TB + 10000); CHECK(!deleted && inst.connections->count == 1);
etcp_link_close(conn->links); CHECK(!conn->links && !conn->links_up && conn->down_timer);
test_clock_run_for(inst.ua, ETCP_DOWN_TIMEOUT_TB + 10000); CHECK(deleted == 1 && !inst.connections->count);
teardown(); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS multiple links, reconnect cancellation and fresh DOWN deadline");
}
/* Локальный handle не препятствует удалению; release внутри DOWN сохраняет таймаут. */
static void check_owner(int release) {
setup(); struct ETCP_CONN* conn = ready_conn(0); struct NODE_CONN_DIRECT* handle = NULL;
CHECK(node_conn_direct_open(&inst, 42, owner_cb, NULL, &handle, NULL) == NCD_REUSED);
uasync_poll(inst.ua, 0); release_on_down = release;
conn->links->link_status = 0; etcp_on_link_down(conn, conn->links);
CHECK(down == 1 && !deleted && conn->down_timer && !conn->close_requested);
test_clock_run_for(inst.ua, ETCP_DOWN_TIMEOUT_TB + 10000);
CHECK(deleted == 1 && closed == (release ? 0 : 1) && !inst.connections->count);
if (!release) node_conn_direct_close(handle);
teardown(); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS DOWN timeout with local owner release_on_down=%d", release);
}
/* RX продлевает watchdog, затем молчание даёт DOWN и ещё 30 секунд до DELETE. */
static void check_silent_peer(int tcp) {
setup(); struct ETCP_CONN* conn = ready_conn(tcp); struct ETCP_LINK* link = conn->links;
etcp_keepalive_watch_rx(link); CHECK(link->rx_timeout_timer);
test_clock_run_for(inst.ua, 15000); link->last_recv_local_time = get_time_tb(); etcp_keepalive_watch_rx(link);
test_clock_run_for(inst.ua, 10000); CHECK(conn->links_up && !conn->down_timer);
test_clock_run_for(inst.ua, 15000); CHECK(!conn->links_up && conn->down_timer && !deleted);
test_clock_run_for(inst.ua, ETCP_DOWN_TIMEOUT_TB + 10000); CHECK(deleted == 1 && !inst.connections->count);
teardown(); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS silent incoming peer tcp=%d", tcp);
}
/* Сон не даёт преждевременный DOWN, но локальные фазовые таймеры не продлевают жизнь пира. */
static void check_sleep(void) {
setup(); struct ETCP_CONN* conn = ready_conn(1); struct ETCP_LINK* link = conn->links;
struct ETCP_DGRAM* pkt = memory_pool_alloc(inst.pkt_pool); CHECK(pkt);
memset(pkt, 0, sizeof(*pkt)); pkt->data_len = 8; pkt->data[0] = ETCP_KEEPALIVE; pkt->data[1] = KA_FLAG_SLEEP;
pkt->data[2] = 0xe8; pkt->data[3] = 3; pkt->data[4] = 5; pkt->data[6] = 200; // 1 s period, 0.5 s ACTIVE, 20 s SLEEP
CHECK(etcp_keepalive_on_recv(&sock, pkt, link, 11) == 1);
CHECK(link->ka_peer_sleeping && !link->keepalive_timer && link->rx_timeout_timer);
test_clock_run_for(inst.ua, 60000); CHECK(conn->links_up && !conn->down_timer);
struct NODE_CONN_DIRECT* handle = NULL;
CHECK(node_conn_direct_open(&inst, 42, NULL, NULL, &handle, NULL) == NCD_REUSED);
node_conn_direct_close(handle); CHECK(conn->fin_wait);
test_clock_run_for(inst.ua, 60000); CHECK(!deleted && conn->links_up && conn->fin_wait);
receive_control(conn, NCD_SUBCMD_KEEP_ALIVE); CHECK(!conn->fin_wait);
test_clock_run_for(inst.ua, 120000); CHECK(!conn->links_up && conn->down_timer && !deleted);
test_clock_run_for(inst.ua, ETCP_DOWN_TIMEOUT_TB + 10000); CHECK(deleted == 1 && !inst.connections->count);
teardown(); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS sleeping peer deadline, FIN_WAIT and disappearance");
}
/* Мягкое закрытие работает без NCD entry; KEEP_ALIVE сохраняет транспорт дольше прежних 5 секунд. */
static void check_graceful(void) {
setup(); struct ETCP_CONN* conn = ready_conn(0);
receive_control(conn, NCD_SUBCMD_CLOSE); CHECK(deleted == 1 && !inst.connections->count); teardown();
setup(); conn = ready_conn(0); struct NODE_CONN_DIRECT* handle = NULL;
CHECK(node_conn_direct_open(&inst, 42, NULL, NULL, &handle, NULL) == NCD_REUSED);
receive_control(conn, NCD_SUBCMD_CLOSE); CHECK(!deleted && !conn->fin_wait);
CHECK(conn->send_input_q->count == 1);
struct ncd_control_msg* response = (void*)conn->send_input_q->head->dgram;
CHECK(response->subcmd == NCD_SUBCMD_KEEP_ALIVE);
node_conn_direct_close(handle); CHECK(conn->fin_wait);
receive_control(conn, NCD_SUBCMD_KEEP_ALIVE); CHECK(!conn->fin_wait);
test_clock_run_for(inst.ua, 70000); CHECK(!deleted && conn->links_up);
receive_control(conn, NCD_SUBCMD_CLOSE); CHECK(deleted == 1 && !inst.connections->count); teardown();
DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS inbound CLOSE, owned refusal and durable KEEP_ALIVE");
}
int main(void) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); test_clock_init(10000);
debug_set_category_level(DEBUG_CATEGORY_CONNECTION, DEBUG_LEVEL_DEBUG);
debug_set_category_level(DEBUG_CATEGORY_KEEPALIVE, DEBUG_LEVEL_DEBUG);
size_t baseline = u_get_allocated_count();
check_pending(); CHECK(u_get_allocated_count() == baseline);
check_reconnect(); CHECK(u_get_allocated_count() == baseline);
check_owner(0); CHECK(u_get_allocated_count() == baseline);
check_owner(1); CHECK(u_get_allocated_count() == baseline);
check_silent_peer(0); CHECK(u_get_allocated_count() == baseline);
check_silent_peer(1); CHECK(u_get_allocated_count() == baseline);
check_sleep(); CHECK(u_get_allocated_count() == baseline);
check_graceful(); CHECK(u_get_allocated_count() == baseline);
test_clock_finish(); return 0;
}
Write Preview
Loading…
Cancel
Save