diff --git a/src/routing_layer/conn_mgr_monitor.c b/src/routing_layer/conn_mgr_monitor.c index 8560abd8..a733affe 100644 --- a/src/routing_layer/conn_mgr_monitor.c +++ b/src/routing_layer/conn_mgr_monitor.c @@ -37,23 +37,31 @@ int cm_is_rtt_fresh(struct TOPO_GROUP_NODE* nq, uint64_t now_tb) { * из route_connectivity после успешного probe. INDIRECT-фаза использует этот * список (best_candidates) для выбора посредника с минимальным суммарным RTT. */ void conn_mgr_update_best_candidates(struct CONN_MGR* mgr, uint64_t node_id, uint16_t rtt) { - if (!mgr || node_id == mgr->instance->node_id || !rtt) return; - for (uint8_t i = 0; i < mgr->best_candidate_count; i++) { - if (mgr->best_candidates[i].node_id != node_id) continue; - mgr->best_candidates[i].rtt = rtt; - for (uint8_t j = i; j > 0 && mgr->best_candidates[j].rtt < mgr->best_candidates[j-1].rtt; j--) - { struct CONN_MGR_CANDIDATE t = mgr->best_candidates[j]; mgr->best_candidates[j]=mgr->best_candidates[j-1]; mgr->best_candidates[j-1]=t; } - for (uint8_t j = i; j + 1 < mgr->best_candidate_count && mgr->best_candidates[j].rtt > mgr->best_candidates[j+1].rtt; j++) - { struct CONN_MGR_CANDIDATE t = mgr->best_candidates[j]; mgr->best_candidates[j]=mgr->best_candidates[j+1]; mgr->best_candidates[j+1]=t; } - return; + if (!mgr || !node_id || node_id == mgr->instance->node_id || !rtt || rtt == UINT16_MAX) return; + uint8_t slot = 0; + while (slot < mgr->best_candidate_count && mgr->best_candidates[slot].node_id != node_id) slot++; + if (slot == mgr->best_candidate_count) { + if (slot < CONN_MGR_MAX_CANDIDATES) mgr->best_candidate_count++; + else { + slot = CONN_MGR_MAX_CANDIDATES - 1; + /* Быстрый, но недоступный кандидат не блокирует рабочий READY-путь. */ + int ready = cm_direct_conn(mgr, node_id) != NULL; + int previous_ready = cm_direct_conn(mgr, mgr->best_candidates[slot].node_id) != NULL; + for (int i = slot - 1; i >= 0 && previous_ready; i--) + if (!cm_direct_conn(mgr, mgr->best_candidates[i].node_id)) { slot = (uint8_t)i; previous_ready = 0; } + if (ready < previous_ready || (ready == previous_ready && (rtt > mgr->best_candidates[slot].rtt || + (rtt == mgr->best_candidates[slot].rtt && node_id >= mgr->best_candidates[slot].node_id)))) return; + } + DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "conn_mgr: candidate admitted peer=%016llx rtt=%u replacing=%016llx", + (unsigned long long)node_id, rtt, (unsigned long long)mgr->best_candidates[slot].node_id); + } + mgr->best_candidates[slot] = (struct CONN_MGR_CANDIDATE){node_id, rtt}; + for (uint8_t i = 0; i < mgr->best_candidate_count; i++) for (uint8_t j = i + 1; j < mgr->best_candidate_count; j++) { + struct CONN_MGR_CANDIDATE *a = &mgr->best_candidates[i], *b = &mgr->best_candidates[j]; + if (b->rtt < a->rtt || (b->rtt == a->rtt && b->node_id < a->node_id)) { + struct CONN_MGR_CANDIDATE tmp = *a; *a = *b; *b = tmp; + } } - if (mgr->best_candidate_count < CONN_MGR_MAX_CANDIDATES) - { mgr->best_candidates[mgr->best_candidate_count++]=(struct CONN_MGR_CANDIDATE){node_id,rtt}; } - else if (rtt < mgr->best_candidates[CONN_MGR_MAX_CANDIDATES-1].rtt) - { mgr->best_candidates[CONN_MGR_MAX_CANDIDATES-1]=(struct CONN_MGR_CANDIDATE){node_id,rtt}; } - else return; - for (uint8_t i = mgr->best_candidate_count - 1; i > 0 && mgr->best_candidates[i].rtt < mgr->best_candidates[i-1].rtt; i--) - { struct CONN_MGR_CANDIDATE t = mgr->best_candidates[i]; mgr->best_candidates[i]=mgr->best_candidates[i-1]; mgr->best_candidates[i-1]=t; } } /* Маршрут может вести через посредника. Для прямой пробы нужны transport peer и его живой линк. */ @@ -105,7 +113,7 @@ static void cm_bg_ping_to_node(struct CONN_MGR* mgr, struct TOPO_NODE* ni) { } } -/* Фоновый ping: один узел группы за тик (~100ms), полный цикл ≥ 10с. +/* Фоновый ping: один узел группы за тик (~100ms), полный цикл ≥ 60с. * Пингует только если нет активного ETCP-соединения с узлом. Цель: поддерживать * свежие RTT-замеры для ВСЕХ узлов группы — без этого best_candidates протухают * и INDIRECT-фаза не может выбрать посредника. */ @@ -122,7 +130,8 @@ void cm_bg_ping_timer_cb(void* arg) { #ifdef UTUN_HAVE_STANDBY if (standby_get_sleep_tb() > 0) { mgr->bg_ping_wait = standby_wait(mgr, cm_bg_ping_timer_cb); - return; + if (mgr->bg_ping_wait) return; + DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: background standby wait failed; retrying by timer"); goto rearm; } mgr->bg_ping_wait = NULL; #endif @@ -135,7 +144,12 @@ void cm_bg_ping_timer_cb(void* arg) { if (mgr->bg_ping_cursor >= total) { mgr->bg_ping_cursor = 0; if ((now - mgr->bg_ping_cycle_start_tb) < CONN_MGR_BG_PING_CYCLE_MIN_TB) - { mgr->bg_ping_timer = uasync_set_timeout(mgr->instance->ua, (int)(CONN_MGR_BG_PING_CYCLE_MIN_TB - (now - mgr->bg_ping_cycle_start_tb)), mgr, cm_bg_ping_timer_cb, "conn_mgr_bg_ping"); return; } + { + mgr->bg_ping_timer = uasync_set_timeout(mgr->instance->ua, + (int)(CONN_MGR_BG_PING_CYCLE_MIN_TB - (now - mgr->bg_ping_cycle_start_tb)), mgr, cm_bg_ping_timer_cb, "conn_mgr_bg_ping"); + if (!mgr->bg_ping_timer) DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: background cycle timer allocation failed"); + return; + } mgr->bg_ping_cycle_start_tb = now; } @@ -154,6 +168,7 @@ void cm_bg_ping_timer_cb(void* arg) { } rearm: mgr->bg_ping_timer = uasync_set_timeout(mgr->instance->ua, CONN_MGR_BG_PING_INTERVAL_TB, mgr, cm_bg_ping_timer_cb, "conn_mgr_bg_ping"); + if (!mgr->bg_ping_timer) DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: background timer allocation failed"); } /* ═══════ candidate_ping ═══════ */ @@ -178,7 +193,8 @@ void cm_candidate_ping_timer_cb(void* arg) { #ifdef UTUN_HAVE_STANDBY if (standby_get_sleep_tb() > 0) { mgr->candidate_ping_wait = standby_wait(mgr, cm_candidate_ping_timer_cb); - return; + if (mgr->candidate_ping_wait) return; + DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: candidate standby wait failed; retrying by timer"); goto rearm; } mgr->candidate_ping_wait = NULL; #endif @@ -224,5 +240,7 @@ void cm_candidate_ping_timer_cb(void* arg) { } } else route_connectivity_probe_node(mgr->instance, mgr->group, nq); } +rearm: mgr->candidate_ping_timer = uasync_set_timeout(mgr->instance->ua, CONN_MGR_CANDIDATE_PING_TB, mgr, cm_candidate_ping_timer_cb, "conn_mgr_cand_ping"); + if (!mgr->candidate_ping_timer) DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: candidate timer allocation failed"); } diff --git a/tests/Makefile.am b/tests/Makefile.am index 04d49c84..d2a05e2c 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -678,6 +678,9 @@ test_conn_mgr_phases_LDFLAGS = $(test_clock_ldflags) test_conn_mgr_handles_SOURCES = test_conn_mgr_handles.c test_conn_mgr_handles_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib test_conn_mgr_handles_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) +if OS_LINUX +test_conn_mgr_handles_LDFLAGS = -Wl,--wrap=uasync_call_soon +endif test_conn_mgr_already_connected_SOURCES = test_conn_mgr_already_connected.c test_conn_mgr_already_connected_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib diff --git a/tests/test_conn_mgr_handles.c b/tests/test_conn_mgr_handles.c index 76546e15..472500fe 100644 --- a/tests/test_conn_mgr_handles.c +++ b/tests/test_conn_mgr_handles.c @@ -39,6 +39,14 @@ static uint64_t g_nid_a = 0, g_nid_b = 0; static volatile int cb1_ok = 0, cb2_ok = 0; static int close_in_callback, unexpected_callback; static struct CONN_MGR_HANDLE *gh1 = NULL, *gh2 = NULL; +#ifdef __linux__ +static int fail_soon, rejected_soon; +void* __real_uasync_call_soon(struct UASYNC*, void*, timeout_callback_t); +void* __wrap_uasync_call_soon(struct UASYNC* loop, void* arg, timeout_callback_t callback) { + if (fail_soon) { rejected_soon++; return NULL; } + return __real_uasync_call_soon(loop, arg, callback); +} +#endif static int wf(const char* p, const char* f, ...) { va_list ap; FILE* fp = fopen(p, "w"); if (!fp) return -1; @@ -56,8 +64,14 @@ static void to_cb(void* arg) { (void)arg; fail("timeout"); done(); } static void ccb1(struct CONN_MGR_HANDLE* h, uint64_t id, uint64_t gid, enum conn_mgr_event ev, void* arg) { (void)h; (void)arg; (void)gid; if (close_in_callback) { +#ifdef __linux__ + fail_soon = 1; +#endif conn_mgr_close(gh1); gh1 = NULL; conn_mgr_close(gh2); gh2 = NULL; +#ifdef __linux__ + fail_soon = 0; +#endif close_in_callback = 0; return; } @@ -101,6 +115,26 @@ static void test1_check(void* arg) { if (!cb1_ok || !cb2_ok) { fail("test1: both callbacks should be UP"); done(); return; } uint8_t ty = entry_type(); if (ty != CONN_TYPE_DIRECT) { fail("test1: conn_type != DIRECT"); done(); return; } + struct CONN_MGR* mgr = topo_groups_get_default(g_a->topo_groups)->conn_mgr; + struct CONN_MGR_CANDIDATE saved[3]; memcpy(saved, mgr->best_candidates, sizeof(saved)); + uint8_t saved_count = mgr->best_candidate_count; + mgr->best_candidate_count = 3; + for (int i = 0; i < 3; i++) mgr->best_candidates[i] = (struct CONN_MGR_CANDIDATE){UINT64_MAX - i, i + 1}; + conn_mgr_update_best_candidates(mgr, g_nid_b, 500); + conn_mgr_update_best_candidates(mgr, UINT64_MAX - 100, 1); + int ready_present = 0; + for (int i = 0; i < 3; i++) if (mgr->best_candidates[i].node_id == g_nid_b) ready_present = 1; + if (!ready_present) { fail("test1: unavailable candidates evicted a READY intermediary"); done(); return; } + memcpy(mgr->best_candidates, saved, sizeof(saved)); mgr->best_candidate_count = saved_count; +#ifdef __linux__ + struct CONN_MGR_HANDLE* extra = NULL; + rejected_soon = 0; fail_soon = 1; + int rc = conn_mgr_open(g_a, TOPO_GROUP_UTUN, g_nid_b, ccb1, NULL, &extra); + fail_soon = 0; + if (rc != -1 || extra || rejected_soon != 1 || !gh1->entry || !gh2->entry) { + fail("test1: failed initial UP scheduling did not roll back handle"); done(); return; + } +#endif fprintf(stderr, " OK: both handles UP, conn_type=DIRECT\n"); fflush(stderr); uasync_call_soon(ua, NULL, test2_check); } @@ -136,8 +170,14 @@ static void test3_verify(void* arg) { fail("test4: open failed"); return; } cb2_ok = 0; close_in_callback = 1; +#ifdef __linux__ + rejected_soon = 0; +#endif cm_deliver_event(cm_find_entry(mgr, g_nid_b), CONN_EVENT_UP); if (gh1 || gh2 || cb2_ok || unexpected_callback) { fail("test4: closed handle received callback"); return; } +#ifdef __linux__ + if (rejected_soon) { fail("test4: cleanup attempted to allocate call_soon"); return; } +#endif fprintf(stderr, "Test 5: remote cleanup detaches handles and cancels pending UP\n"); if (conn_mgr_open(g_a, TOPO_GROUP_UTUN, g_nid_b, ccb1, NULL, &gh1) != 0) { fail("test5: open failed"); return;