Browse Source
Добавлено: - conn_mgr_connect_from_invite() — подключение к новому узлу по invite-данным (INIT → INVITE_INFO_REQ/RESP → проверка членства в группе → сохранение в БД) - topo_group_connect.c — авто-подключение при старте CHAT-группы: Phase 1: одновременный connect всех connected=1 пиров, таймаут, отмена незавершённых Phase 2: последовательный перебор публичных адресов до первого успеха - Счётчик active_conn_count с проверкой nodeinfo-путей (UP/DOWN) - restart() при phase=DONE и count==0 БД: - connected INTEGER NOT NULL DEFAULT 0 в peers_<ch> + ALTER TABLE для существующих - topo_node_sqlite_set_connected / get_connected_peers / get_public_peers Chatgui: - Включён topo_group (убран utun_instance_set_topo_group_enabled(0)) - Создание TOPO_GROUP_TYPE_CHAT при channel_ready и при старте - Invite connect через conn_mgr_connect_from_invite - Удалён chat_conn_mgr (из сборки) + весь старый auto-connect кодtopo_upd
Evgeny
2 months ago
16 changed files with 834 additions and 271 deletions
@ -0,0 +1,260 @@
|
||||
/*
|
||||
* topo_group_connect.c — авто-подключение к узлам группы при старте |
||||
* |
||||
* Phase 1: одновременный запуск conn_mgr_connect_node для пиров с connected=1. |
||||
* Таймаут = direct_timeout_ms. По таймауту отменяем незавершённые и помечаем |
||||
* connected=0 в БД. Если хоть один подключился → done. |
||||
* |
||||
* Phase 2: последовательный перебор пиров с публичными адресами до первого успеха. |
||||
* |
||||
* Счётчик active_conn_count — число уникальных peer-узлов с реальными соединениями. |
||||
* UP: инкремент только для первого соединения к peer_node_id. |
||||
* DOWN: декремент только если нет других conn к peer_node_id И нет indirect-путей в nodeinfo. |
||||
* |
||||
* restart: перезапуск авто-подключения при phase=done и active_conn_count==0. |
||||
*/ |
||||
|
||||
#include "topo_group_connect.h" |
||||
#include "topo_group.h" |
||||
#include "topo_node.h" |
||||
#include "topo_node_sqlite.h" |
||||
#include "conn_mgr.h" |
||||
#include "../lib/debug_config.h" |
||||
#include "../lib/mem.h" |
||||
#include "../lib/u_async.h" |
||||
#include "../lib/ll_queue.h" |
||||
#include "etcp.h" |
||||
|
||||
/* ─── внутренние константы ─── */ |
||||
#define TGC_ID "topo_group_connect" |
||||
#define TGC_PHASE_ONE 0 |
||||
#define TGC_PHASE_TWO 1 |
||||
#define TGC_PHASE_DONE 2 |
||||
|
||||
struct TOPO_GROUP_CONNECT { |
||||
struct TOPO_GROUP* group; |
||||
void* phase_timer; |
||||
uint8_t phase; |
||||
uint8_t active; |
||||
int pending; |
||||
int connected_count; |
||||
int active_conn_count; // число уникальных peer-узлов с живыми соединениями
|
||||
int cursor; |
||||
uint64_t* candidate_ids; |
||||
int candidate_count; |
||||
}; |
||||
|
||||
static void tgc_phase2_try_next(struct TOPO_GROUP_CONNECT* gc); |
||||
static void tgc_phase2_result(int result, uint64_t node_id, void* arg); |
||||
static void tgc_phase1_timeout(void* arg); |
||||
static void tgc_phase1_result(int result, uint64_t node_id, void* arg); |
||||
|
||||
/* ═══════════════════════════════════════════════════════════════════════
|
||||
* Жизненный цикл |
||||
* ══════════════════════════════════════════════════════════════════════ */ |
||||
|
||||
int topo_group_connect_init(struct TOPO_GROUP* group) { |
||||
if (!group || group->group_type != TOPO_GROUP_TYPE_CHAT || !group->channel_id[0] || group->connect) |
||||
return -1; |
||||
struct TOPO_GROUP_CONNECT* gc = u_calloc(1, sizeof(*gc)); |
||||
if (!gc) return -1; |
||||
gc->group = group; gc->active = 1; |
||||
group->connect = gc; |
||||
|
||||
uint64_t* ids = NULL; int count = 0; |
||||
sqlite3* db = group->instance->topo_sqlite_db; |
||||
if (!db || topo_node_sqlite_get_connected_peers(db, group->channel_id, &ids, &count) != 0 || count == 0) { |
||||
if (ids) { u_free(ids); ids = NULL; } |
||||
gc->phase = TGC_PHASE_TWO; |
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: ch=%s no connected peers → Phase 2", TGC_ID, group->channel_id); |
||||
tgc_phase2_try_next(gc); |
||||
return 0; |
||||
} |
||||
|
||||
gc->phase = TGC_PHASE_ONE; gc->pending = count; |
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: ch=%s Phase 1 launching %d connects", TGC_ID, group->channel_id, count); |
||||
for (int i = 0; i < count; i++) |
||||
conn_mgr_connect_node(group->conn_mgr, ids[i], 0, tgc_phase1_result, gc); |
||||
u_free(ids); |
||||
|
||||
gc->phase_timer = uasync_set_timeout(group->instance->ua, |
||||
(int)group->conn_mgr->direct_timeout_ms * 10, gc, tgc_phase1_timeout, "tgc_phase1"); |
||||
return 0; |
||||
} |
||||
|
||||
void topo_group_connect_destroy(struct TOPO_GROUP* group) { |
||||
struct TOPO_GROUP_CONNECT* gc = group->connect; |
||||
if (!gc) return; |
||||
group->connect = NULL; |
||||
if (gc->phase_timer) { uasync_cancel_timeout(group->instance->ua, gc->phase_timer); gc->phase_timer = NULL; } |
||||
u_free(gc->candidate_ids); |
||||
u_free(gc); |
||||
} |
||||
|
||||
void topo_group_connect_restart(struct TOPO_GROUP* group) { |
||||
struct TOPO_GROUP_CONNECT* gc = group->connect; |
||||
if (!gc) { topo_group_connect_init(group); return; } |
||||
if (gc->phase != TGC_PHASE_DONE) return; |
||||
if (gc->active_conn_count > 0) return; |
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: restart ch=%s", TGC_ID, group->channel_id); |
||||
topo_group_connect_destroy(group); |
||||
topo_group_connect_init(group); |
||||
} |
||||
|
||||
int topo_group_connect_active_count(struct TOPO_GROUP* group) { |
||||
struct TOPO_GROUP_CONNECT* gc = group->connect; |
||||
return gc ? gc->active_conn_count : 0; |
||||
} |
||||
|
||||
/* ═══════════════════════════════════════════════════════════════════════
|
||||
* ON UP / DOWN |
||||
* ══════════════════════════════════════════════════════════════════════ */ |
||||
|
||||
void topo_group_connect_on_up(struct TOPO_GROUP* group, struct ETCP_CONN* conn) { |
||||
if (!group || group->group_type != TOPO_GROUP_TYPE_CHAT || !group->channel_id[0] || !conn) return; |
||||
uint64_t peer = conn->peer_node_id; |
||||
struct TOPO_GROUP_CONNECT* gc = group->connect; |
||||
if (!gc) return; |
||||
|
||||
/* уже есть другое прямое соединение к этому peer? */ |
||||
struct ll_entry* e = group->senders_list ? group->senders_list->head : NULL; |
||||
while (e) { |
||||
struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data; |
||||
if (item->conn && item->conn != conn && item->conn->peer_node_id == peer) { |
||||
DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "%s: UP peer=0x%016llx already connected via other conn, skip count", |
||||
TGC_ID, (unsigned long long)peer); |
||||
return; |
||||
} |
||||
e = e->next; |
||||
} |
||||
|
||||
gc->active_conn_count++; |
||||
sqlite3* db = group->instance->topo_sqlite_db; |
||||
if (db) topo_node_sqlite_set_connected(db, group->channel_id, peer, 1); |
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: UP peer=0x%016llx ch=%s active=%d", |
||||
TGC_ID, (unsigned long long)peer, group->channel_id, gc->active_conn_count); |
||||
} |
||||
|
||||
void topo_group_connect_on_down(struct TOPO_GROUP* group, struct ETCP_CONN* conn) { |
||||
if (!group || group->group_type != TOPO_GROUP_TYPE_CHAT || !group->channel_id[0] || !conn) return; |
||||
uint64_t peer = conn->peer_node_id; |
||||
struct TOPO_GROUP_CONNECT* gc = group->connect; |
||||
if (!gc) return; |
||||
|
||||
/* есть ли другие прямые соединения к этому peer? */ |
||||
struct ll_entry* e = group->senders_list ? group->senders_list->head : NULL; |
||||
while (e) { |
||||
struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data; |
||||
if (item->conn && item->conn != conn && item->conn->peer_node_id == peer) { |
||||
DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "%s: DOWN peer=0x%016llx still has other direct conn, skip", |
||||
TGC_ID, (unsigned long long)peer); |
||||
return; |
||||
} |
||||
e = e->next; |
||||
} |
||||
|
||||
/* есть ли indirect-пути в nodeinfo? */ |
||||
struct TOPO_NODEQ* nq = topo_node_find_by_id(group, peer); |
||||
if (nq && nq->paths) { |
||||
struct ll_entry* pe = nq->paths->head; |
||||
while (pe) { |
||||
struct TOPO_NODEPATH* path = (struct TOPO_NODEPATH*)pe; |
||||
if (path->conn && path->conn != conn) { |
||||
DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "%s: DOWN peer=0x%016llx has indirect path, skip", |
||||
TGC_ID, (unsigned long long)peer); |
||||
return; |
||||
} |
||||
pe = pe->next; |
||||
} |
||||
} |
||||
|
||||
gc->active_conn_count--; |
||||
sqlite3* db = group->instance->topo_sqlite_db; |
||||
if (db) topo_node_sqlite_set_connected(db, group->channel_id, peer, 0); |
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: DOWN peer=0x%016llx ch=%s active=%d", |
||||
TGC_ID, (unsigned long long)peer, group->channel_id, gc->active_conn_count); |
||||
} |
||||
|
||||
/* ═══════════════════════════════════════════════════════════════════════
|
||||
* Phase 1 |
||||
* ══════════════════════════════════════════════════════════════════════ */ |
||||
|
||||
static void tgc_phase1_result(int result, uint64_t node_id, void* arg) { |
||||
struct TOPO_GROUP_CONNECT* gc = (struct TOPO_GROUP_CONNECT*)arg; |
||||
if (gc->phase != TGC_PHASE_ONE) return; |
||||
gc->pending--; |
||||
if (result == CONN_MGR_OK) gc->connected_count++; |
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase1 result node=0x%016llx %s pending=%d connected=%d", |
||||
TGC_ID, (unsigned long long)node_id, result == CONN_MGR_OK ? "OK" : "FAIL", gc->pending, gc->connected_count); |
||||
} |
||||
|
||||
static void tgc_phase1_timeout(void* arg) { |
||||
struct TOPO_GROUP_CONNECT* gc = (struct TOPO_GROUP_CONNECT*)arg; |
||||
gc->phase_timer = NULL; |
||||
if (gc->phase != TGC_PHASE_ONE) return; |
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase1 timeout connected=%d pending=%d ch=%s", |
||||
TGC_ID, gc->connected_count, gc->pending, gc->group->channel_id); |
||||
|
||||
sqlite3* db = gc->group->instance->topo_sqlite_db; |
||||
uint64_t* ids = NULL; int count = 0; |
||||
topo_node_sqlite_get_connected_peers(db, gc->group->channel_id, &ids, &count); |
||||
for (int i = 0; i < count; i++) { |
||||
uint8_t status; uint8_t conn_type; |
||||
conn_mgr_get_status(gc->group->conn_mgr, ids[i], &status, &conn_type); |
||||
if (status != CONN_MGR_STATE_CONNECTED) { |
||||
conn_mgr_cancel_callback(gc->group->conn_mgr, ids[i], tgc_phase1_result, gc); |
||||
topo_node_sqlite_set_connected(db, gc->group->channel_id, ids[i], 0); |
||||
} |
||||
} |
||||
u_free(ids); |
||||
|
||||
if (gc->connected_count > 0) { |
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase1 done — %d connected, skipping Phase 2", TGC_ID, gc->connected_count); |
||||
gc->phase = TGC_PHASE_DONE; |
||||
return; |
||||
} |
||||
gc->phase = TGC_PHASE_TWO; |
||||
tgc_phase2_try_next(gc); |
||||
} |
||||
|
||||
/* ═══════════════════════════════════════════════════════════════════════
|
||||
* Phase 2 |
||||
* ══════════════════════════════════════════════════════════════════════ */ |
||||
|
||||
static void tgc_phase2_try_next(struct TOPO_GROUP_CONNECT* gc) { |
||||
if (gc->phase != TGC_PHASE_TWO) return; |
||||
if (gc->candidate_count == 0) { |
||||
sqlite3* db = gc->group->instance->topo_sqlite_db; |
||||
topo_node_sqlite_get_public_peers(db, gc->group->channel_id, &gc->candidate_ids, &gc->candidate_count); |
||||
gc->cursor = 0; |
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase2 loaded %d public peers for ch=%s", |
||||
TGC_ID, gc->candidate_count, gc->group->channel_id); |
||||
} |
||||
while (gc->cursor < gc->candidate_count) { |
||||
uint64_t nid = gc->candidate_ids[gc->cursor++]; |
||||
uint8_t status; uint8_t conn_type; |
||||
conn_mgr_get_status(gc->group->conn_mgr, nid, &status, &conn_type); |
||||
if (status == CONN_MGR_STATE_CONNECTED) { |
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase2 skip 0x%016llx already connected", TGC_ID, (unsigned long long)nid); |
||||
continue; |
||||
} |
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase2 trying 0x%016llx (%d/%d)", TGC_ID, |
||||
(unsigned long long)nid, gc->cursor, gc->candidate_count); |
||||
conn_mgr_connect_node(gc->group->conn_mgr, nid, 0, tgc_phase2_result, gc); |
||||
return; |
||||
} |
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase2 exhausted — no connections, ch=%s", TGC_ID, gc->group->channel_id); |
||||
gc->phase = TGC_PHASE_DONE; |
||||
} |
||||
|
||||
static void tgc_phase2_result(int result, uint64_t node_id, void* arg) { |
||||
struct TOPO_GROUP_CONNECT* gc = (struct TOPO_GROUP_CONNECT*)arg; |
||||
if (gc->phase != TGC_PHASE_TWO) return; |
||||
if (result == CONN_MGR_OK) { |
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase2 connected to 0x%016llx — done", TGC_ID, (unsigned long long)node_id); |
||||
gc->phase = TGC_PHASE_DONE; |
||||
return; |
||||
} |
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase2 0x%016llx FAIL — try next", TGC_ID, (unsigned long long)node_id); |
||||
tgc_phase2_try_next(gc); |
||||
} |
||||
@ -0,0 +1,30 @@
|
||||
/**
|
||||
* @file topo_group_connect.h |
||||
* @brief Авто-подключение к узлам группы при старте и отслеживание состояния connected в БД. |
||||
* |
||||
* При старте CHAT-группы: |
||||
* Phase 1 — одновременный запуск подключений ко всем пирам с connected=1. |
||||
* Если хоть один подключился → done. |
||||
* Phase 2 — последовательный перебор по пирам с публичными адресами. |
||||
* До первого успешного подключения. |
||||
* |
||||
* Отслеживание: на UP → peers_<ch>.connected=1, на DOWN (последний conn) → 0. |
||||
* active_conn_count — количество уникальных peer-узлов с живыми соединениями. |
||||
* restart — перезапуск авто-подключения (если phase=done и нет подключений). |
||||
*/ |
||||
#ifndef TOPO_GROUP_CONNECT_H |
||||
#define TOPO_GROUP_CONNECT_H |
||||
|
||||
#include <stdint.h> |
||||
|
||||
struct TOPO_GROUP; |
||||
struct ETCP_CONN; |
||||
|
||||
int topo_group_connect_init(struct TOPO_GROUP* group); |
||||
void topo_group_connect_destroy(struct TOPO_GROUP* group); |
||||
void topo_group_connect_on_up(struct TOPO_GROUP* group, struct ETCP_CONN* conn); |
||||
void topo_group_connect_on_down(struct TOPO_GROUP* group, struct ETCP_CONN* conn); |
||||
int topo_group_connect_active_count(struct TOPO_GROUP* group); |
||||
void topo_group_connect_restart(struct TOPO_GROUP* group); |
||||
|
||||
#endif |
||||
Loading…
Reference in new issue