jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1934 Commits
15 Branches
2 Tags
152 MiB
 
 
 
 
 
 
Tree: a97e8bcc1f
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from 'a97e8bcc1f'
${ noResults }
utun2/src/routing_layer/conn_mgr_core.c

752 lines
43 KiB
Raw Blame History

/* conn_mgr_core.c — init/destroy, open/close/handles, DIRECT, REVERSE, invite */
#include "conn_mgr_priv.h"
#include <stdlib.h>
#include <string.h>
#ifdef _WIN32
#include <winsock2.h>
#include <ws2tcpip.h>
#else
#include <arpa/inet.h>
#endif
#include "../lib/platform_compat.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "../lib/u_async.h"
#include "utun_instance.h"
#include "etcp.h"
#include "etcp_connections.h"
#include "stcp_client.h"
#include "etcp_router.h"
#include "config_parser.h"
#include "topo_node.h"
#include "topo_node_sqlite.h"
#include "sock_match.h"
#include "topo_group.h"
#include "route_ping.h"
#ifdef UTUN_HAVE_STANDBY
#include "standby.h"
#endif
/* ═══════ forward-декларации ═══════ */
void conn_mgr_router_recv_handler(struct ETCP_CONN* conn, struct ll_entry* entry);
static void cm_attempt_tick(void* arg);
static void cm_start_attempt(struct CONN_MGR_ENTRY* entry, int recovery);
static const char* cm_path_name(uint8_t type) {
static const char* names[] = { "NONE", "DIRECT", "REVERSE", "INDIRECT" };
return type <= CONN_TYPE_INDIRECT ? names[type] : "INVALID";
}
/* ═══════ утилиты ═══════ */
static struct CONN_MGR_HANDLE* cm_handle_new(struct CONN_MGR_ENTRY* entry, uint64_t group_id,
conn_mgr_cb_t cb, void* cb_arg) {
struct CONN_MGR_HANDLE* h = u_calloc(1, sizeof(*h));
if (!h) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: handle allocation failed"); return NULL; }
h->node_id = entry->node_id; h->group_id = group_id; h->cb = cb; h->cb_arg = cb_arg;
h->ua = entry->mgr->instance->ua;
h->entry = entry; h->next = entry->handles; entry->handles = h;
return h;
}
static void cm_deliver_up_cb(void* arg) {
struct CONN_MGR_HANDLE* h = (struct CONN_MGR_HANDLE*)arg;
h->deliver_up_token = NULL;
if (!h->closed && h->entry && h->entry->state == CONN_MGR_STATE_CONNECTED && h->cb)
h->cb(h, h->node_id, h->group_id, CONN_EVENT_UP, h->cb_arg);
}
/* Закрытые handles освобождаются через call_soon, поэтому next переживает callbacks.
* Рассылает событие ВСЕМ handle'ам entry. Одно соединение — много слушателей:
* каждый conn_mgr_open создал свой handle, все получают UP/DOWN/TIMEOUT.
* Все поля handle'а читаются до вызова коллбэка — коллбэк может закрыть
* handle (conn_mgr_close), после вызова к handle не обращаемся. */
void cm_deliver_event(struct CONN_MGR_ENTRY* entry, enum conn_mgr_event event) {
if (!entry || !entry->mgr || !entry->mgr->group) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "cm_deliver_event: BAD entry=%p mgr=%p", entry, entry ? (void*)entry->mgr : NULL);
return;
}
uint64_t gid = entry->mgr->group->group_id;
struct CONN_MGR_HANDLE* h = entry->handles;
while (h) {
struct CONN_MGR_HANDLE* next = h->next;
uint64_t node_id = h->node_id;
conn_mgr_cb_t cb = h->cb;
void* cb_arg = h->cb_arg;
if (event != CONN_EVENT_UP && h->deliver_up_token) {
uasync_call_soon_cancel(h->ua, h->deliver_up_token); h->deliver_up_token = NULL;
}
if (!h->closed && h->entry == entry && cb) cb(h, node_id, gid, event, cb_arg);
h = next;
}
}
/* ═══════ NAT / адресные утилиты ═══════ */
/* Классификация IPv6-адреса: link-local (fe80), локальный ULA (fc/fd),
* глобальный/прямой, другое (loopback, multicast, ::). */
uint8_t cm_classify_v6_addr(const uint8_t addr[16]) {
if (addr[0] == 0xfe && (addr[1] & 0xc0) == 0x80) return CM_V6_LL;
if (addr[0] == 0xfc || addr[0] == 0xfd) return CM_V6_LOC;
if (addr[0] == 0xff) return CM_V6_OTH;
{ static const uint8_t z[16]; if (!memcmp(addr, z, 16)) return CM_V6_OTH; }
{ static const uint8_t lb[16]={0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,1}; if (!memcmp(addr, lb, 16)) return CM_V6_OTH; }
return CM_V6_DIR;
}
/* Та же классификация, что при создании линков; EIM и прямой IPv6 доступны извне. */
int cm_has_direct_ip(struct CONN_MGR* mgr, struct TOPO_GROUP_NODE* nq) {
struct TOPO_NODE* ni = nq ? topo_node_registry_find(mgr->instance->topo_groups, nq->node_id) : NULL;
if (!ni) return 0;
for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) {
if (!(a->protocol & (TOPO_PROTO_UDP | TOPO_PROTO_TCP))) continue;
for (const struct TOPO_SOCKMETA4* m = ni->v4_sock_meta; m; m = m->next)
if (m->id == a->socket_id && sock_target_class(a->type, m->config_type, m->nat_type) == SOCK_TC_PUB) return 1;
}
for (const struct TOPO_ADDR6* a = ni->v6_addrs; a; a = a->next)
if ((a->protocol & (TOPO_PROTO_UDP | TOPO_PROTO_TCP)) && cm_classify_v6_addr(a->addr) == CM_V6_DIR) return 1;
return 0;
}
/* ═══════ entries ═══════ */
struct CONN_MGR_ENTRY* cm_find_entry(struct CONN_MGR* mgr, uint64_t node_id) {
if (!mgr || !mgr->entries) return NULL;
struct ll_entry* e = queue_find_data_by_index(mgr->entries, &node_id);
return e ? (struct CONN_MGR_ENTRY*)e : NULL;
}
struct CONN_MGR_ENTRY* cm_ensure_entry(struct CONN_MGR* mgr, uint64_t node_id) {
struct CONN_MGR_ENTRY* e = cm_find_entry(mgr, node_id);
if (e) return e;
struct ll_entry* qe = queue_entry_new(sizeof(struct CONN_MGR_ENTRY) - sizeof(struct ll_entry));
if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: queue_entry_new failed"); return NULL; }
struct CONN_MGR_ENTRY* new_e = (struct CONN_MGR_ENTRY*)qe;
memset((uint8_t*)new_e + sizeof(struct ll_entry), 0, sizeof(*new_e) - sizeof(struct ll_entry));
new_e->node_id = node_id; new_e->mgr = mgr;
if (queue_data_put_with_index(mgr->entries, qe) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: entry insertion failed peer=%016llx", (unsigned long long)node_id);
queue_entry_free(qe); return NULL;
}
return new_e;
}
static void cm_entry_free_cb(void* arg) { queue_entry_free(arg); }
static void cm_handle_free_cb(void* arg) { u_free(arg); }
void cm_entry_cleanup(struct CONN_MGR_ENTRY* entry) {
if (!entry) return;
struct CONN_MGR* mgr = entry->mgr;
if (cm_find_entry(mgr, entry->node_id) != entry) return;
entry->state = CONN_MGR_STATE_DISCONNECTED;
queue_remove_data(mgr->entries, &entry->ll);
if (entry->timer) { uasync_cancel_timeout(mgr->instance->ua, entry->timer); entry->timer = NULL; }
if (entry->refresh_token) { uasync_call_soon_cancel(mgr->instance->ua, entry->refresh_token); entry->refresh_token = NULL; }
for (struct CONN_MGR_HANDLE* h = entry->handles; h; h = h->next) {
if (h->deliver_up_token) { uasync_call_soon_cancel(h->ua, h->deliver_up_token); h->deliver_up_token = NULL; }
h->entry = NULL; // Владелец обязан закрыть отсоединённый handle.
}
entry->handles = NULL;
struct NODE_CONN_DIRECT* ncd = entry->ncd_handle; entry->ncd_handle = NULL;
if (ncd) node_conn_direct_force_close(ncd);
cm_clear_nodeinfo(mgr, entry->node_id);
cm_cleanup_db_node(entry);
DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "conn_mgr: cleanup group=%016llx peer=%016llx",
(unsigned long long)mgr->group->group_id, (unsigned long long)entry->node_id);
uasync_call_soon(mgr->instance->ua, entry, cm_entry_free_cb);
}
void cm_clear_nodeinfo(struct CONN_MGR* mgr, uint64_t node_id) {
struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(mgr->group, node_id);
if (!nq) return;
nq->conn_mgr_type = CONN_TYPE_NONE; nq->conn_mgr_intermediariy_count = 0;
nq->conn_presence &= ~NCONN_INDIRECT; nq->conn_up &= ~NCONN_INDIRECT;
}
void cm_update_nodeinfo(struct CONN_MGR_ENTRY* entry) {
if (!entry->mgr->group) return;
struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(entry->mgr->group, entry->node_id);
if (!nq) return;
nq->conn_mgr_type = entry->conn_type;
memcpy(nq->conn_mgr_intermediaries, entry->intermediaries, sizeof(entry->intermediaries));
nq->conn_mgr_intermediariy_count = entry->intermediariy_count;
if (entry->conn_type == CONN_TYPE_INDIRECT) { nq->conn_presence |= NCONN_INDIRECT; nq->conn_up |= NCONN_INDIRECT; }
else { nq->conn_presence &= ~NCONN_INDIRECT; nq->conn_up &= ~NCONN_INDIRECT; }
topo_fire_nodeinfo_cbk(entry->mgr->instance, entry->mgr->group, nq);
}
void cm_cleanup_db_node(struct CONN_MGR_ENTRY* entry) {
if (!entry || !entry->mgr || !entry->mgr->group || !entry->db_loaded) return;
struct TOPO_GROUP* group = entry->mgr->group;
struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(group, entry->node_id);
if (!nq) return;
if (nq->paths && queue_entry_count(nq->paths) > 0) {
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: db_node 0x%016llx got paths, keep in group", (unsigned long long)entry->node_id);
return;
}
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: removing db_node 0x%016llx from group (no paths)", (unsigned long long)entry->node_id);
topo_nodeq_remove_node(group, nq);
}
/* Прямой транспорт пригоден только вместе с READY-сессией этой группы. */
struct ETCP_CONN* cm_direct_conn(struct CONN_MGR* mgr, uint64_t node_id) {
struct ETCP_CONN* conn = topo_group_find_conn_for_node(mgr->group, node_id);
return conn && conn->peer_node_id == node_id && conn->links_up && !conn->close_requested &&
topo_group_peer_ready(mgr->group, node_id) ? conn : NULL;
}
/* READY первого плеча недостаточно: посредник должен ещё объявлять маршрут к получателю. */
static struct ETCP_CONN* cm_intermediary_conn(struct CONN_MGR_ENTRY* entry, uint64_t node_id) {
struct ETCP_CONN* conn = cm_direct_conn(entry->mgr, node_id);
struct TOPO_GROUP_NODE* target = topo_node_find_by_id(entry->mgr->group, entry->node_id);
for (struct ll_entry* e = conn && target && target->paths ? target->paths->head : NULL; e; e = e->next)
if (((struct TOPO_NODEPATH*)e)->conn == conn) return conn;
return NULL;
}
uint16_t cm_candidate_rtt(struct CONN_MGR* mgr, uint64_t node_id) {
struct ETCP_CONN* conn = cm_direct_conn(mgr, node_id);
if (!conn) return UINT16_MAX;
uint16_t rtt = conn->rtt_last;
if (!rtt) {
struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(mgr->group, node_id);
// Нулевой текущий RTT на loopback допустим, если успешный замер уже был.
if (conn->last_rtt_cb_time) return 1;
if (nq && cm_is_rtt_fresh(nq, get_time_tb())) {
rtt = nq->connectivity.real_min_rtt;
if (!rtt) rtt = nq->connectivity.nat_min_rtt;
if (!rtt) rtt = nq->connectivity.interface_min_rtt;
}
if (!rtt) DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "conn_mgr: READY candidate lacks RTT peer=%016llx",
(unsigned long long)node_id);
}
return rtt && rtt != UINT16_MAX ? rtt : UINT16_MAX;
}
struct ETCP_CONN* conn_mgr_route_conn(struct CONN_MGR* mgr, uint64_t node_id, int allow_probe) {
struct CONN_MGR_ENTRY* entry = cm_find_entry(mgr, node_id);
if (!entry || entry->state == CONN_MGR_STATE_DISCONNECTED) return NULL;
struct ETCP_CONN* direct = cm_direct_conn(mgr, node_id);
if (direct) return direct;
for (uint8_t i = 0; entry->indirect_confirmed && i < entry->intermediariy_count; i++) {
struct ETCP_CONN* conn = cm_intermediary_conn(entry, entry->intermediaries[i]);
if (conn) return conn;
}
// Пока нет ACK, через этот путь идёт проверка; сервис ещё не получил UP.
for (uint8_t i = 0; allow_probe && i < entry->selection.count; i++) {
struct ETCP_CONN* conn = cm_intermediary_conn(entry, entry->selection.selected[i].node_id);
if (conn) return conn;
}
return NULL;
}
/* Исходящие и входящие линки могут сосуществовать; тип не зависит от порядка их создания. */
uint8_t conn_mgr_connection_type(const struct ETCP_CONN* conn, uint64_t node_id) {
if (!conn || !conn->links_up || conn->close_requested || conn->state == 2) return CONN_TYPE_NONE;
if (conn->peer_node_id != node_id) return CONN_TYPE_INDIRECT;
int incoming = 0;
for (const struct ETCP_LINK* link = conn->links; link; link = link->next) {
if (!link->initialized || !link->link_status) continue;
if (!link->is_server) return CONN_TYPE_DIRECT;
incoming = 1;
}
return incoming ? CONN_TYPE_REVERSE : CONN_TYPE_NONE;
}
/* Смена пути не сбрасывает router-очереди и не повторяет UP при улучшении. */
void cm_refresh_path(struct CONN_MGR_ENTRY* entry) {
if (entry->state == CONN_MGR_STATE_DISCONNECTED) return;
uint8_t type = CONN_TYPE_NONE;
struct ETCP_CONN* direct = cm_direct_conn(entry->mgr, entry->node_id);
if (direct) type = conn_mgr_connection_type(direct, entry->node_id);
if (entry->indirect_confirmed) {
int reserve_up = 0;
for (uint8_t i = 0; i < entry->intermediariy_count; i++)
if (cm_intermediary_conn(entry, entry->intermediaries[i])) { reserve_up = 1; break; }
if (reserve_up && !direct) type = CONN_TYPE_INDIRECT;
if (!reserve_up) {
entry->indirect_confirmed = 0;
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: reserve unavailable peer=%016llx", (unsigned long long)entry->node_id);
}
}
int was_up = entry->state == CONN_MGR_STATE_CONNECTED;
if (type == entry->conn_type && was_up == (type != CONN_TYPE_NONE)) return;
struct ETCP_CONN* next = conn_mgr_route_conn(entry->mgr, entry->node_id, 0);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: path group=%016llx peer=%016llx %s->%s via=%016llx elapsed=%llums",
(unsigned long long)entry->mgr->group->group_id, (unsigned long long)entry->node_id,
cm_path_name(entry->conn_type), cm_path_name(type), (unsigned long long)(next ? next->peer_node_id : 0),
(unsigned long long)((get_time_tb() - entry->start_tb) / 10));
int lost_direct = entry->conn_type == CONN_TYPE_DIRECT || entry->conn_type == CONN_TYPE_REVERSE;
lost_direct &= type != CONN_TYPE_DIRECT && type != CONN_TYPE_REVERSE;
entry->conn_type = type;
entry->state = type ? CONN_MGR_STATE_CONNECTED : CONN_MGR_STATE_CONNECTING;
if (entry->handles && (lost_direct || (was_up && !type))) {
entry->attempt_active = 0; entry->retry_tb = get_time_tb(); entry->retry_delay_ms = CONN_MGR_RETRY_MIN_MS;
}
cm_update_nodeinfo(entry);
if (entry->state == CONN_MGR_STATE_DISCONNECTED) return;
if (!was_up && type) { entry->ever_up = 1; cm_deliver_event(entry, CONN_EVENT_UP); }
else if (was_up && !type) cm_deliver_event(entry, CONN_EVENT_DOWN);
}
static void cm_refresh_cb(void* arg) {
struct CONN_MGR_ENTRY* entry = arg; entry->refresh_token = NULL;
struct TOPO_NODE* ni = topo_node_registry_find(entry->mgr->instance->topo_groups, entry->node_id);
if (entry->handles && ni && ni->timestamp != entry->node_timestamp && !cm_direct_conn(entry->mgr, entry->node_id)) {
entry->attempt_active = 0; entry->retry_tb = get_time_tb(); entry->retry_delay_ms = CONN_MGR_RETRY_MIN_MS;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: addresses changed peer=%016llx version=%llu; retry now",
(unsigned long long)entry->node_id, (unsigned long long)ni->timestamp);
entry->node_timestamp = ni->timestamp;
}
cm_refresh_path(entry);
if (entry->state != CONN_MGR_STATE_DISCONNECTED) cm_arm_timer(entry);
}
void conn_mgr_routes_changed(struct CONN_MGR* mgr) {
if (!mgr || !mgr->initialized) return;
// Не вызываем сервисы внутри обхода entries/изменения таблицы: callback может закрыть соседний entry.
for (struct ll_entry* e = mgr->entries->head; e; e = e->next) {
struct CONN_MGR_ENTRY* entry = (struct CONN_MGR_ENTRY*)e;
if (entry->refresh_token) continue;
entry->refresh_token = uasync_call_soon(mgr->instance->ua, entry, cm_refresh_cb);
if (!entry->refresh_token) DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: refresh scheduling failed peer=%016llx",
(unsigned long long)entry->node_id);
}
}
/* Изменение собственных сокетов обходит backoff, но не прерывает рабочую доставку. */
void conn_mgr_network_changed(struct CONN_MGR* mgr) {
if (!mgr || !mgr->initialized) return;
for (struct ll_entry* e = mgr->entries->head; e; e = e->next) {
struct CONN_MGR_ENTRY* entry = (struct CONN_MGR_ENTRY*)e;
if (!entry->handles || cm_direct_conn(mgr, entry->node_id)) continue;
entry->attempt_active = 0; entry->retry_tb = get_time_tb(); entry->retry_delay_ms = CONN_MGR_RETRY_MIN_MS;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: local network changed peer=%016llx; retry now",
(unsigned long long)entry->node_id);
}
conn_mgr_routes_changed(mgr);
}
/* TIMEOUT первого NCD не закрывает транспорт: поздний UP всё ещё может улучшить путь. */
void cm_ncd_callback(struct NODE_CONN_DIRECT* ncd_h, enum ncd_event ev, void* arg) {
struct CONN_MGR_ENTRY* entry = arg;
if (!entry || entry->state == CONN_MGR_STATE_DISCONNECTED) return;
DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "conn_mgr: NCD event=%d peer=%016llx path=%u elapsed=%llums",
ev, (unsigned long long)entry->node_id, entry->conn_type,
(unsigned long long)((get_time_tb() - entry->start_tb) / 10));
if (ev == NCD_EVENT_UP) {
entry->direct_up = 1; entry->main_connect_state = CM_TRY_OK;
struct ETCP_CONN* conn = node_conn_direct_get_conn(ncd_h);
if (conn && topo_group_new_conn(entry->mgr->group, conn) < 0)
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: group negotiation unavailable peer=%016llx",
(unsigned long long)entry->node_id);
} else if (ev == NCD_EVENT_TIMEOUT) {
entry->main_connect_state = CM_TRY_FAILED;
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: direct probe timeout peer=%016llx; retaining transport path=%u",
(unsigned long long)entry->node_id, entry->conn_type);
} else {
entry->direct_up = 0;
if (ev == NCD_EVENT_CLOSED) { entry->ncd_handle = NULL; node_conn_direct_close(ncd_h); }
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: direct transport lost peer=%016llx event=%d",
(unsigned long long)entry->node_id, ev);
if (entry->handles) {
entry->attempt_active = 0; entry->retry_tb = get_time_tb(); entry->retry_delay_ms = CONN_MGR_RETRY_MIN_MS;
}
}
cm_refresh_path(entry);
if (entry->state != CONN_MGR_STATE_DISCONNECTED) cm_arm_timer(entry);
}
int cm_send_control(struct CONN_MGR* mgr, uint64_t node_id, const void* packet, size_t len) {
struct ll_entry* qe = ll_alloc_lldgram(1 + len);
if (!qe) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: control allocation failed peer=%016llx", (unsigned long long)node_id); return -1; }
qe->dgram[0] = ETCP_RT_ID_CONN_MGR; memcpy(qe->dgram + 1, packet, len); qe->len = (uint16_t)(1 + len);
int ret = etcp_route_send(mgr->instance, mgr->group->group_id, node_id, qe, 1, 0);
if (ret < 0) DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: control send failed peer=%016llx ret=%d",
(unsigned long long)node_id, ret);
return ret;
}
void cm_arm_timer(struct CONN_MGR_ENTRY* entry) {
if (entry->state == CONN_MGR_STATE_DISCONNECTED) return;
if (entry->timer) { uasync_cancel_timeout(entry->mgr->instance->ua, entry->timer); entry->timer = NULL; }
uint64_t now = get_time_tb(), when = UINT64_MAX;
int direct_ready = entry->conn_type == CONN_TYPE_DIRECT || entry->conn_type == CONN_TYPE_REVERSE;
if (!entry->handles) when = entry->deadline_tb;
else if (direct_ready && (!entry->indirect_started || entry->indirect_confirmed)) {
entry->attempt_active = 0; entry->retry_tb = 0; entry->retry_delay_ms = CONN_MGR_RETRY_MIN_MS;
} else if (entry->attempt_active) when = entry->deadline_tb;
else if (!direct_ready) {
if (!entry->retry_tb) entry->retry_tb = now;
when = entry->retry_tb;
}
if (entry->handles && !entry->ever_up && !entry->timeout_notified && entry->initial_deadline_tb < when)
when = entry->initial_deadline_tb;
if (entry->handles && entry->attempt_active && !direct_ready) {
uint64_t reverse = entry->reverse_started ? entry->reverse_next_tb : entry->reverse_start_tb;
if (reverse < when) when = reverse;
}
if (entry->handles && entry->attempt_active && !entry->indirect_confirmed && (!direct_ready || entry->indirect_started)) {
uint64_t indirect = entry->indirect_started ? entry->indirect_next_tb : entry->indirect_start_tb;
if (indirect < when) when = indirect;
}
if (when == UINT64_MAX) return;
entry->timer = uasync_set_timeout(entry->mgr->instance->ua, when > now ? (int)(when - now) : 1,
entry, cm_attempt_tick, "conn_mgr_attempt");
if (!entry->timer) DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: timer allocation failed peer=%016llx",
(unsigned long long)entry->node_id);
}
static void cm_attempt_tick(void* arg) {
struct CONN_MGR_ENTRY* entry = arg; entry->timer = NULL;
cm_refresh_path(entry);
if (entry->state == CONN_MGR_STATE_DISCONNECTED) return;
uint64_t now = get_time_tb();
int direct_ready = entry->conn_type == CONN_TYPE_DIRECT || entry->conn_type == CONN_TYPE_REVERSE;
if (!entry->handles) {
if (now >= entry->deadline_tb) cm_entry_cleanup(entry);
else cm_arm_timer(entry);
return;
}
if (!entry->ever_up && !entry->timeout_notified && now >= entry->initial_deadline_tb) {
entry->timeout_notified = 1;
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: initial connection timeout peer=%016llx; retaining demand",
(unsigned long long)entry->node_id);
cm_deliver_event(entry, CONN_EVENT_TIMEOUT);
if (entry->state == CONN_MGR_STATE_DISCONNECTED) return;
}
if (!direct_ready && !entry->attempt_active && now >= entry->retry_tb) cm_start_attempt(entry, 1);
if (entry->state == CONN_MGR_STATE_DISCONNECTED) return;
if (entry->attempt_active && now < entry->deadline_tb) {
if (!entry->ncd_handle && !direct_ready) cm_start_phase_direct(entry);
if (!direct_ready && now >= entry->reverse_start_tb && now >= entry->reverse_next_tb) cm_start_phase_reverse(entry);
if ((!direct_ready || entry->indirect_started) && !entry->indirect_confirmed &&
now >= entry->indirect_start_tb && now >= entry->indirect_next_tb)
cm_start_phase_indirect(entry);
}
if (entry->state == CONN_MGR_STATE_DISCONNECTED) return;
if (entry->attempt_active && now >= entry->deadline_tb) {
entry->attempt_active = 0;
if (!entry->retry_delay_ms) entry->retry_delay_ms = CONN_MGR_RETRY_MIN_MS;
entry->retry_tb = now + entry->retry_delay_ms * 10;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: attempt exhausted peer=%016llx path=%s retry=%ums",
(unsigned long long)entry->node_id, cm_path_name(entry->conn_type), entry->retry_delay_ms);
unsigned next = entry->retry_delay_ms * 2;
entry->retry_delay_ms = next > CONN_MGR_RETRY_MAX_MS ? CONN_MGR_RETRY_MAX_MS : next;
}
cm_arm_timer(entry);
}
/* Новый цикл меняет только пробы; NCD, подтверждённый резерв и очереди router остаются живы. */
static void cm_start_attempt(struct CONN_MGR_ENTRY* entry, int recovery) {
struct CONN_MGR* mgr = entry->mgr;
struct TOPO_GROUP_NODE* target = topo_node_find_by_id(mgr->group, entry->node_id);
struct TOPO_NODE* ni = topo_node_registry_find(mgr->instance->topo_groups, entry->node_id);
entry->start_tb = get_time_tb(); entry->attempt_active = 1; entry->retry_tb = 0;
entry->node_timestamp = ni ? ni->timestamp : 0;
entry->reverse_started = 0; entry->reverse_next_tb = 0; entry->indirect_started = 0; entry->indirect_next_tb = 0;
entry->exchange_request_id = 0;
memset(&entry->selection, 0, sizeof(entry->selection));
int local_public = cm_has_direct_ip(mgr, mgr->group->local_node), peer_public = cm_has_direct_ip(mgr, target);
entry->reverse_start_tb = entry->start_tb + (local_public && !peer_public ? 0 : CONN_MGR_REVERSE_DELAY_MS * 10);
entry->indirect_start_tb = entry->start_tb + (recovery || (!local_public && !peer_public) ? 0 : CONN_MGR_INDIRECT_DELAY_MS * 10);
entry->deadline_tb = entry->indirect_start_tb + CONN_MGR_INTERM_EXCHANGE_TIMEOUT_MS * 10;
if (!recovery) entry->initial_deadline_tb = entry->deadline_tb;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: %s group=%016llx peer=%016llx public=%d/%d reverse=%llums indirect=%llums reserve=%d",
recovery ? "recover" : "start", (unsigned long long)mgr->group->group_id, (unsigned long long)entry->node_id,
local_public, peer_public, (unsigned long long)((entry->reverse_start_tb - entry->start_tb) / 10),
(unsigned long long)((entry->indirect_start_tb - entry->start_tb) / 10), entry->indirect_confirmed);
cm_start_phase_direct(entry);
if (entry->state == CONN_MGR_STATE_DISCONNECTED) return;
if (!local_public && !peer_public) cm_start_local_scan(entry);
if (entry->reverse_start_tb == entry->start_tb) cm_start_phase_reverse(entry);
if (!entry->indirect_confirmed && entry->indirect_start_tb == entry->start_tb) cm_start_phase_indirect(entry);
}
/* ═══════ init / destroy ═══════ */
struct CONN_MGR* conn_mgr_init(struct TOPO_GROUP* group) {
if (!group || !group->instance) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "NULL group"); return NULL; }
struct CONN_MGR* mgr = u_calloc(1, sizeof(struct CONN_MGR));
if (!mgr) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "alloc failed"); return NULL; }
mgr->instance = group->instance; mgr->group = group;
mgr->entries = queue_new(mgr->instance->ua, 256,
offsetof(struct CONN_MGR_ENTRY, node_id) - sizeof(struct ll_entry),
8, "conn_mgr_entries");
if (!mgr->entries) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "entries alloc failed"); u_free(mgr); return NULL; }
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");
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");
mgr->bg_ping_cycle_start_tb = get_time_tb(); mgr->initialized = 1;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: initialized for group %016llx", (unsigned long long)group->group_id);
return mgr;
}
void conn_mgr_destroy(struct CONN_MGR* mgr) {
if (!mgr) return;
mgr->initialized = 0;
if (mgr->bg_ping_timer) uasync_cancel_timeout(mgr->instance->ua, mgr->bg_ping_timer);
if (mgr->candidate_ping_timer) uasync_cancel_timeout(mgr->instance->ua, mgr->candidate_ping_timer);
#ifdef UTUN_HAVE_STANDBY
if (mgr->bg_ping_wait) standby_wait_cancel(mgr->bg_ping_wait);
if (mgr->candidate_ping_wait) standby_wait_cancel(mgr->candidate_ping_wait);
#endif
for (int i = 0; i < CONN_MGR_MAX_CANDIDATES; i++)
if (mgr->cand_tcp_pings[i]) stcp_client_destroy(mgr->cand_tcp_pings[i]);
while (mgr->entries->head) cm_entry_cleanup((struct CONN_MGR_ENTRY*)mgr->entries->head);
queue_free(mgr->entries);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: destroyed group=%016llx", (unsigned long long)mgr->group->group_id);
u_free(mgr);
}
/* ═══════ публичное API ═══════ */
int cm_open(struct CONN_MGR* mgr, uint64_t node_id,
conn_mgr_cb_t cb, void* cb_arg, struct CONN_MGR_HANDLE** out_handle) {
if (out_handle) *out_handle = NULL;
if (!mgr) { if (cb) cb(NULL, node_id, 0, CONN_EVENT_TIMEOUT, cb_arg); return -1; }
struct TOPO_GROUP* group = mgr->group;
if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "BGP not initialized"); return -1; }
if (node_id == 0) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: cm_open node_id=0 — rejected grp=%016llx type=%d ch=%s",
(unsigned long long)group->group_id, group->group_type,
group->channel_id[0] ? group->channel_id : "-");
if (cb) cb(NULL, node_id, 0, CONN_EVENT_TIMEOUT, cb_arg);
return -1;
}
uint64_t gid = group->group_id;
struct TOPO_GROUP_NODE* target = topo_node_find_by_id(group, node_id);
int loaded_from_db = 0;
if (!target && mgr->instance->topo_sqlite_db && mgr->instance->topo_groups) {
struct TOPO_NODE* ni = topo_node_sqlite_node_load(mgr->instance->topo_sqlite_db, mgr->instance->topo_groups, node_id);
if (ni) {
ni = topo_node_registry_store(mgr->instance->topo_groups, ni);
struct ll_entry* qe = queue_entry_new(sizeof(struct TOPO_GROUP_NODE) - sizeof(struct ll_entry));
if (qe) {
target = (struct TOPO_GROUP_NODE*)qe;
memset((uint8_t*)target + sizeof(struct ll_entry), 0, sizeof(*target) - sizeof(struct ll_entry));
target->node_id = node_id; target->conn_mgr_type = CONN_TYPE_NONE;
queue_data_put_with_index(group->nodes, &target->ll); loaded_from_db = 1;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: node 0x%016llx loaded from DB", (unsigned long long)node_id);
} else DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: failed to alloc TOPO_GROUP_NODE for DB node 0x%016llx", (unsigned long long)node_id);
}
}
if (!target) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "node 0x%016llx skipped — not in routing, no active connections", (unsigned long long)node_id); return -1; }
struct CONN_MGR_ENTRY* entry = cm_find_entry(mgr, node_id);
if (entry && entry->handles) {
struct CONN_MGR_HANDLE* h = cm_handle_new(entry, gid, cb, cb_arg);
if (!h) return -1; if (out_handle) *out_handle = h;
if (entry->state == CONN_MGR_STATE_CONNECTED)
h->deliver_up_token = uasync_call_soon(mgr->instance->ua, h, cm_deliver_up_cb);
DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "conn_mgr: joined demand peer=%016llx state=%u retry=%llu",
(unsigned long long)node_id, entry->state, (unsigned long long)entry->retry_tb);
return 0;
}
entry = cm_ensure_entry(mgr, node_id);
if (!entry) return -1;
int already_up = entry->state == CONN_MGR_STATE_CONNECTED;
if (!already_up) entry->state = CONN_MGR_STATE_CONNECTING;
struct CONN_MGR_HANDLE* h = cm_handle_new(entry, gid, cb, cb_arg);
if (!h) { cm_entry_cleanup(entry); return -1; } if (out_handle) *out_handle = h;
// Первый владелец входящего пути начинает свои пробы, сохраняя рабочую доставку.
if (entry->timer) { uasync_cancel_timeout(mgr->instance->ua, entry->timer); entry->timer = NULL; }
if (already_up) h->deliver_up_token = uasync_call_soon(mgr->instance->ua, h, cm_deliver_up_cb);
entry->last_traffic_tb = get_time_tb();
entry->timeout_notified = 0;
entry->retry_delay_ms = CONN_MGR_RETRY_MIN_MS;
entry->db_loaded |= (uint8_t)loaded_from_db;
cm_start_attempt(entry, 0);
if (entry->state == CONN_MGR_STATE_DISCONNECTED) return 0;
cm_arm_timer(entry);
return 0;
}
void conn_mgr_close(struct CONN_MGR_HANDLE* h) {
if (!h || h->closed) return;
h->closed = 1;
if (h->deliver_up_token) { uasync_call_soon_cancel(h->ua, h->deliver_up_token); h->deliver_up_token = NULL; }
struct CONN_MGR_ENTRY* entry = h->entry; h->entry = NULL;
DEBUG_DEBUG(DEBUG_CATEGORY_GENERAL, "conn_mgr: close handle=%p node=%016llx attached=%d", (void*)h,
(unsigned long long)h->node_id, entry != NULL);
if (entry) {
struct CONN_MGR_HANDLE** pp = &entry->handles;
while (*pp && *pp != h) pp = &(*pp)->next;
if (*pp) *pp = h->next;
if (!entry->handles) cm_entry_cleanup(entry);
}
// Не посылаем DISCONNECT: удалённые владельцы и групповой транспорт живут независимо.
uasync_call_soon(h->ua, h, cm_handle_free_cb);
}
struct ETCP_CONN* conn_mgr_get_conn(struct CONN_MGR_HANDLE* h) {
if (!h || h->closed || !h->entry || h->entry->state != CONN_MGR_STATE_CONNECTED ||
h->entry->conn_type == CONN_TYPE_INDIRECT) return NULL;
return cm_direct_conn(h->entry->mgr, h->entry->node_id);
}
int conn_mgr_send(struct CONN_MGR_HANDLE* h, struct ll_entry* e) {
if (!h || !e) return -1;
struct CONN_MGR_ENTRY* entry = h->entry;
if (!entry || entry->state != CONN_MGR_STATE_CONNECTED) { queue_entry_free(e); return -1; }
entry->last_traffic_tb = get_time_tb();
struct ETCP_ROUTER_CONN* r = etcp_router_conn_get(entry->mgr->instance, entry->mgr->group->group_id, entry->node_id, ETCP_RT_ID_CONN_MGR);
if (!r) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: no route to 0x%016llx, dropping %u bytes", (unsigned long long)entry->node_id, (unsigned)e->len); queue_entry_free(e); return -1; }
int ret = etcp_router_conn_send(r, e->dgram, e->len, 0); queue_entry_free(e); return ret;
}
/* Локальные пробы уже созданы DIRECT через NCD. Проверяем все LAN-адреса,
* не дублируя handshake и не создавая отдельные ping-контексты. */
void cm_start_local_scan(struct CONN_MGR_ENTRY* entry) {
struct TOPO_NODE* ni = topo_node_registry_find(entry->mgr->instance->topo_groups, entry->node_id);
if (!ni || !entry->ncd_handle) {
entry->local_scan_state = CM_TRY_FAILED;
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: LOCAL probe missing node/transport peer=%016llx",
(unsigned long long)entry->node_id); return;
}
int addresses = 0;
for (const struct TOPO_ADDR4* a = ni->v4_addrs; a; a = a->next) {
if (!(a->protocol & (TOPO_PROTO_UDP | TOPO_PROTO_TCP))) continue;
for (const struct TOPO_SOCKMETA4* m = ni->v4_sock_meta; m; m = m->next)
if (m->id == a->socket_id && sock_target_class(a->type, m->config_type, m->nat_type) == SOCK_TC_PRIV) {
addresses++; break;
}
}
for (const struct TOPO_ADDR6* a = ni->v6_addrs; a; a = a->next) {
uint8_t cls = cm_classify_v6_addr(a->addr);
if ((a->protocol & (TOPO_PROTO_UDP | TOPO_PROTO_TCP)) && (cls == CM_V6_LL || cls == CM_V6_LOC)) addresses++;
}
entry->local_scan_state = addresses ? CM_TRY_PENDING : CM_TRY_FAILED;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: LOCAL probes peer=%016llx addresses=%d shared_with_DIRECT=1 elapsed=%llums",
(unsigned long long)entry->node_id, addresses, (unsigned long long)((get_time_tb() - entry->start_tb) / 10));
if (!addresses) DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: no advertised local addresses peer=%016llx",
(unsigned long long)entry->node_id);
}
void cm_start_phase_direct(struct CONN_MGR_ENTRY* entry) {
if (!entry->ncd_handle && node_conn_direct_open(entry->mgr->instance, entry->node_id, cm_ncd_callback, entry,
&entry->ncd_handle, NULL) == NCD_ERR) {
DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "conn_mgr: NCD open failed peer=%016llx", (unsigned long long)entry->node_id);
return;
}
// Входящий REVERSE добавляет новые адреса, не заменяя уже идущие пробы.
if (node_conn_direct_update_node(entry->mgr->instance, entry->node_id) < 0)
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: address update failed peer=%016llx", (unsigned long long)entry->node_id);
if (!entry->direct_up) entry->main_connect_state = CM_TRY_PENDING;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: DIRECT probes peer=%016llx elapsed=%llums transport=%p",
(unsigned long long)entry->node_id, (unsigned long long)((get_time_tb() - entry->start_tb) / 10),
(void*)node_conn_direct_get_conn(entry->ncd_handle));
}
/* Запрос использует подписанный снимок отправителя: одинаковые правила IPv4/IPv6, UDP/TCP и Reality с DIRECT. */
void cm_start_phase_reverse(struct CONN_MGR_ENTRY* entry) {
uint64_t now = get_time_tb();
if (!entry->reverse_started) {
entry->reverse_started = 1; entry->reverse_request_id = ++entry->mgr->next_request_id;
entry->reverse_deadline_tb = now + CONN_MGR_CONNECT_REVERSE_TIMEOUT_MS * 10;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: REVERSE start peer=%016llx elapsed=%llums",
(unsigned long long)entry->node_id, (unsigned long long)((now - entry->start_tb) / 10));
}
if (now >= entry->reverse_deadline_tb) { entry->reverse_next_tb = UINT64_MAX; return; }
struct CM_DIRECT_REQ req = { ETCP_RT_ID_CONN_MGR, CM_SUBCMD_DIRECT_REQ, entry->reverse_request_id };
cm_send_control(entry->mgr, entry->node_id, &req, sizeof(req));
entry->reverse_next_tb = now + CONN_MGR_REVERSE_RETRANSMIT_MS * 10;
}
void cm_handle_direct_req(uint64_t src, struct CONN_MGR* mgr, const uint8_t* data, size_t len) {
if (!mgr || len != sizeof(struct CM_DIRECT_REQ)) {
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: malformed DIRECT_REQ peer=%016llx len=%zu", (unsigned long long)src, len);
return;
}
struct CM_DIRECT_REQ req; memcpy(&req, data, sizeof(req));
struct TOPO_NODE* ni = topo_node_registry_find(mgr->instance->topo_groups, src);
if (!ni) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: REVERSE unknown peer=%016llx", (unsigned long long)src); return; }
struct CONN_MGR_ENTRY* entry = cm_ensure_entry(mgr, src);
if (!entry) return;
if (entry->state == CONN_MGR_STATE_DISCONNECTED) {
entry->state = CONN_MGR_STATE_CONNECTING; entry->start_tb = get_time_tb();
entry->deadline_tb = entry->start_tb + CONN_MGR_INTERM_EXCHANGE_TIMEOUT_MS * 10;
}
if (entry->peer_reverse_request_id == req.request_id) return;
entry->peer_reverse_request_id = req.request_id;
if (!entry->handles) entry->deadline_tb = get_time_tb() + CONN_MGR_INTERM_EXCHANGE_TIMEOUT_MS * 10;
if (entry->direct_up && cm_direct_conn(mgr, src)) return;
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: REVERSE request peer=%016llx path=%u existing_transport=%d",
(unsigned long long)src, entry->conn_type, entry->ncd_handle != NULL);
cm_start_phase_direct(entry);
cm_arm_timer(entry);
}
/* ═══════ disconnect / recv ═══════ */
void cm_handle_disconnect(struct CONN_MGR* mgr, uint64_t node_id) {
struct CONN_MGR_ENTRY* entry = cm_find_entry(mgr, node_id); if (!entry) return;
cm_clear_nodeinfo(mgr, node_id); cm_entry_cleanup(entry);
DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "conn_mgr: remote disconnect from 0x%016llx", (unsigned long long)node_id);
}
void conn_mgr_router_recv_handler(struct ETCP_CONN* conn, struct ll_entry* entry) {
if (!entry || !entry->dgram || entry->len < ROUTER_SVC_PAYLOAD_OFF + 2) {
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
return;
}
if (!conn || !conn->instance) {
DEBUG_ERROR(DEBUG_CATEGORY_ROUTING, "conn_mgr_router_recv: conn=%p", (void*)conn);
queue_dgram_free(entry); queue_entry_free(entry);
return;
}
uint8_t* d = entry->dgram + ROUTER_SVC_PAYLOAD_OFF; size_t len = entry->len - ROUTER_SVC_PAYLOAD_OFF;
if (d[0] != ETCP_RT_ID_CONN_MGR) { queue_dgram_free(entry); queue_entry_free(entry); return; }
uint8_t sub = d[1];
/* conn_mgr — per-group: резолвим группу из заголовка доставки, а не instance->conn_mgr */
uint64_t group_id; memcpy(&group_id, entry->dgram + ROUTER_SVC_GROUP_OFF, 8);
uint64_t src; memcpy(&src, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
struct TOPO_GROUP* grp = topo_groups_find(conn->instance->topo_groups, group_id);
struct CONN_MGR* mgr = grp ? grp->conn_mgr : NULL;
if (!mgr) {
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: packet for unknown group=%016llx peer=%016llx",
(unsigned long long)group_id, (unsigned long long)src);
queue_dgram_free(entry); queue_entry_free(entry); return;
}
if (mgr) { struct CONN_MGR_ENTRY* e = cm_find_entry(mgr, src); if (e && e->state == CONN_MGR_STATE_CONNECTED) e->last_traffic_tb = get_time_tb(); }
switch (sub) {
case CM_SUBCMD_DIRECT_REQ: cm_handle_direct_req(src, mgr, d, len); break;
case CM_SUBCMD_DIRECT_RESP: break;
case CM_SUBCMD_INTERM_EXCHANGE_REQ:
if (len == CM_EXCHANGE_REQ_SIZE) cm_handle_interm_exchange_req(src, mgr, (struct CM_EXCHANGE_REQ*)d);
else DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: malformed EXCHANGE_REQ peer=%016llx len=%zu", (unsigned long long)src, len);
break;
case CM_SUBCMD_INTERM_EXCHANGE_RESP: if (mgr) cm_handle_interm_exchange_resp(mgr, src, d, len); break;
case CM_SUBCMD_INTERM_SELECTED: if (mgr) cm_handle_interm_selected(mgr, src, d, len); break;
case CM_SUBCMD_INTERM_ACK: if (mgr) cm_handle_interm_ack(mgr, src, d, len); break;
case CM_SUBCMD_DISCONNECT:
if (len == CM_DISCONNECT_SIZE && ((struct CM_DISCONNECT*)d)->node_id == src) cm_handle_disconnect(mgr, src);
else DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: malformed DISCONNECT peer=%016llx len=%zu", (unsigned long long)src, len);
break;
default: DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "conn_mgr: unknown command=%u peer=%016llx", sub, (unsigned long long)src); break;
}
queue_dgram_free(entry); queue_entry_free(entry);
}
int conn_mgr_open(struct UTUN_INSTANCE* inst,
uint64_t group_id,
uint64_t node_id,
conn_mgr_cb_t cb, void* cb_arg,
struct CONN_MGR_HANDLE** out_handle) {
if (out_handle) *out_handle = NULL;
if (!inst || !inst->topo_groups || !cb) return -1;
struct TOPO_GROUP* group = topo_groups_find(inst->topo_groups, group_id);
if (!group) {
char ch[32]; snprintf(ch, sizeof(ch), "%llu", (unsigned long long)group_id);
group = topo_groups_create_group(inst->topo_groups, group_id, TOPO_GROUP_TYPE_CHAT, ch);
if (!group) return -1;
}
struct CONN_MGR* mgr = group->conn_mgr;
if (!mgr) return -1;
return cm_open(mgr, node_id, cb, cb_arg, out_handle);
}