Keep READY intermediaries eligible and test CM allocation failures

This commit is contained in:
evgeny committed 2026-10-24 16:14:48 +02:00
1 parent 80451c43be
commit 2f3cf0933c
3 files changed
+81 -20

No files matched your search

+38 -20
View File
@@ -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");
}
+3
View File
@@ -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
+40
View File
@@ -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;