auto-connect: cursor-based, no close on timeout, GC every 1s, cancel API
- ca_timeout_cb: removed etcp_connection_close, ETCP lives past 3s timeout - ca_state.cancelled flag + chat_core_connect_auto_cancel() for external cancel - chat_core_connect_auto returns ca_state* via out_state - Rewritten auto_connect: cursor (ch,peer) instead of bulk collect - GC every 1s: closes flights with no link_status after 3s - ac_result_cb no longer clears ca_state (GC handles cleanup) - stop: cancels all flights via chat_core_connect_auto_cancel
This commit is contained in:
1 parent
dccf54b30b
commit
33aa6bbd5a
3 files changed
+195
-205
No files matched your search
@@ -891,7 +891,8 @@ struct ca_state {
|
||||
struct UTUN_INSTANCE* inst;
|
||||
int addr_count;
|
||||
int pending_count;
|
||||
int completed; /* 0=pending, 1=delivered */
|
||||
int delivered; /* 0=pending, 1=result already delivered */
|
||||
int cancelled; /* 1=externally cancelled, do not deliver result */
|
||||
uint64_t node_id;
|
||||
struct ETCP_CONN** conns;
|
||||
void** timers;
|
||||
@@ -910,8 +911,8 @@ static void ca_cleanup(struct ca_state* st) {
|
||||
static void ca_ready_cb(struct ETCP_CONN* conn, void* arg) {
|
||||
struct ca_ctx* ctx = (struct ca_ctx*)arg;
|
||||
struct ca_state* st = ctx->state;
|
||||
if (st->completed) { u_free(ctx); return; }
|
||||
st->completed = 1;
|
||||
if (st->delivered || st->cancelled) { u_free(ctx); return; }
|
||||
st->delivered = 1;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: auto_connect SUCCESS idx=%d peer=0x%016llx",
|
||||
CC_ID, ctx->addr_index, (unsigned long long)st->node_id);
|
||||
for (int i = 0; i < st->addr_count; i++) {
|
||||
@@ -926,24 +927,37 @@ static void ca_ready_cb(struct ETCP_CONN* conn, void* arg) {
|
||||
static void ca_timeout_cb(void* arg) {
|
||||
struct ca_ctx* ctx = (struct ca_ctx*)arg;
|
||||
struct ca_state* st = ctx->state;
|
||||
if (st->completed) { u_free(ctx); return; }
|
||||
if (st->conns[ctx->addr_index]) { etcp_connection_close(st->conns[ctx->addr_index]); st->conns[ctx->addr_index] = NULL; }
|
||||
if (st->delivered || st->cancelled) { u_free(ctx); return; }
|
||||
st->timers[ctx->addr_index] = NULL;
|
||||
st->pending_count--;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: auto_connect TIMEOUT idx=%d pending=%d/%d peer=0x%016llx",
|
||||
CC_ID, ctx->addr_index, st->pending_count, st->addr_count, (unsigned long long)st->node_id);
|
||||
if (st->pending_count <= 0 && !st->completed) {
|
||||
st->completed = 1;
|
||||
if (st->pending_count <= 0 && !st->delivered) {
|
||||
st->delivered = 1;
|
||||
st->result_cb(CONN_MGR_ERR_TIMEOUT, st->node_id, st->result_arg);
|
||||
ca_cleanup(st);
|
||||
}
|
||||
u_free(ctx);
|
||||
}
|
||||
|
||||
void chat_core_connect_auto_cancel(void* state) {
|
||||
if (!state) return;
|
||||
struct ca_state* st = (struct ca_state*)state;
|
||||
st->cancelled = 1;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: auto_connect CANCELLED peer=0x%016llx",
|
||||
CC_ID, (unsigned long long)st->node_id);
|
||||
for (int i = 0; i < st->addr_count; i++) {
|
||||
if (st->timers[i]) { uasync_cancel_timeout(st->inst->ua, st->timers[i]); st->timers[i] = NULL; }
|
||||
if (st->conns[i]) { etcp_connection_close(st->conns[i]); st->conns[i] = NULL; }
|
||||
}
|
||||
ca_cleanup(st);
|
||||
}
|
||||
|
||||
void chat_core_connect_auto(uint64_t node_id,
|
||||
void (*cb)(int result, uint64_t node_id, void* arg),
|
||||
void* arg) {
|
||||
void* arg,
|
||||
void** out_state) {
|
||||
if (!g_cc.initialized || !g_cc.inst || !cb) return;
|
||||
if (out_state) *out_state = NULL;
|
||||
|
||||
struct ETCP_SOCKET* best_socket = g_cc.inst->etcp_sockets;
|
||||
while (best_socket && best_socket->local_addr.ss_family != AF_INET) best_socket = best_socket->next;
|
||||
@@ -1048,8 +1062,11 @@ void chat_core_connect_auto(uint64_t node_id,
|
||||
addrs[i].port);
|
||||
}
|
||||
|
||||
if (pst->pending_count <= 0 && !pst->completed) {
|
||||
if (out_state) *out_state = pst;
|
||||
|
||||
if (pst->pending_count <= 0 && !pst->delivered) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTIVITY, "%s: auto_connect all %d attempts failed to start", CC_ID, addr_count);
|
||||
if (out_state) *out_state = NULL;
|
||||
cb(CONN_MGR_ERR_UNREACHABLE, node_id, arg);
|
||||
ca_cleanup(pst);
|
||||
}
|
||||
|
||||
@@ -55,10 +55,16 @@ struct chat_invite {
|
||||
void chat_core_connect_from_invite(struct chat_invite* inv);
|
||||
|
||||
/* Прямое ETCP-подключение к узлу без BGP: pubkey+адреса из SQLite,
|
||||
* коллбэк вызывается с CONN_MGR_OK / CONN_MGR_ERR_TIMEOUT / CONN_MGR_ERR_NO_ADDRESSES */
|
||||
* коллбэк вызывается один раз с CONN_MGR_OK / CONN_MGR_ERR_*.
|
||||
* При out_state != NULL — возвращает opaque handle для отмены. */
|
||||
void chat_core_connect_auto(uint64_t node_id,
|
||||
void (*cb)(int result, uint64_t node_id, void* arg),
|
||||
void* arg);
|
||||
void* arg,
|
||||
void** out_state);
|
||||
|
||||
/* Отмена активного auto-connect (закрывает ETCP-соединения, освобождает память).
|
||||
* Коллбэк chat_core_connect_auto после cancel не вызывается. */
|
||||
void chat_core_connect_auto_cancel(void* state);
|
||||
|
||||
/* ── Создание канала (GUI → uasync) ── */
|
||||
|
||||
|
||||
+160
-193
@@ -24,219 +24,194 @@
|
||||
static struct chat_sync* g_cs = NULL;
|
||||
|
||||
/* ═══════════════════════════════════════════════════════════════════════
|
||||
* Auto-connect: parallel connect to SQLite peers via direct ETCP (no BGP)
|
||||
* Auto-connect: cursor-based peer iteration with GC every 1s
|
||||
* ══════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
#define AC_MAX_PARALLEL 10
|
||||
#define AC_TARGET_SUCCESS 3
|
||||
#define AC_RETRY_INTERVAL_MS 10000
|
||||
#define AC_MAX_FLIGHTS 10
|
||||
#define AC_RETRY_MS 1000
|
||||
#define AC_GC_TIMEOUT_MS 3000
|
||||
#define AC_ID "auto_connect"
|
||||
|
||||
struct auto_connect;
|
||||
|
||||
struct ac_flight {
|
||||
struct auto_connect* ac;
|
||||
uint64_t node_id;
|
||||
void* ca_state; /* opaque, owned by chat_core_connect_auto */
|
||||
uint64_t created_tb; /* get_time_tb() when launched */
|
||||
};
|
||||
|
||||
struct auto_connect {
|
||||
struct UTUN_INSTANCE* inst;
|
||||
uint64_t* node_ids;
|
||||
int node_count;
|
||||
int next_index;
|
||||
int in_flight;
|
||||
int success_count;
|
||||
int total_tried;
|
||||
void* retry_timer;
|
||||
uint8_t active;
|
||||
|
||||
char** channel_ids; /* loaded once, refreshed on cursor wrap */
|
||||
int channel_count;
|
||||
int ch_cursor; /* current channel index */
|
||||
int peer_cursor; /* current peer index within channel */
|
||||
|
||||
struct ac_flight flights[AC_MAX_FLIGHTS];
|
||||
};
|
||||
|
||||
static struct auto_connect* g_ac = NULL;
|
||||
|
||||
static void ac_result_cb(int result, uint64_t node_id, void* arg);
|
||||
static void ac_retry_timer_cb(void* arg);
|
||||
static int ac_connect_one(struct auto_connect* ac, uint64_t nid);
|
||||
static void ac_launch_batch(struct auto_connect* ac);
|
||||
static int ac_collect_nodes(struct UTUN_INSTANCE* inst, uint64_t** out, int* out_count);
|
||||
static void ac_result_cb(int result, uint64_t node_id, void* arg);
|
||||
|
||||
/* ── helper: count connected nodes (excl. self) ── */
|
||||
/* ── load channel IDs into auto_connect ── */
|
||||
|
||||
static int ac_count_connected(struct UTUN_INSTANCE* inst) {
|
||||
int cnt = 0;
|
||||
struct ETCP_CONN* c = inst->connections;
|
||||
while (c) {
|
||||
if (c->peer_node_id && c->peer_node_id != inst->node_id) {
|
||||
struct ETCP_LINK* l = c->links;
|
||||
while (l) { if (l->initialized && l->link_status) { cnt++; break; } l = l->next; }
|
||||
}
|
||||
c = c->next;
|
||||
static int ac_load_channels(struct auto_connect* ac) {
|
||||
if (ac->channel_ids) {
|
||||
for (int i = 0; i < ac->channel_count; i++) u_free(ac->channel_ids[i]);
|
||||
u_free(ac->channel_ids);
|
||||
ac->channel_ids = NULL;
|
||||
}
|
||||
return cnt;
|
||||
}
|
||||
|
||||
/* ── helper: check if node already has active ETCP connection ── */
|
||||
|
||||
static int ac_node_busy(struct UTUN_INSTANCE* inst, uint64_t nid) {
|
||||
struct ETCP_CONN* c = inst->connections;
|
||||
while (c) {
|
||||
if (c->peer_node_id == nid) {
|
||||
struct ETCP_LINK* l = c->links;
|
||||
while (l) { if (l->initialized && l->link_status) return 1; l = l->next; }
|
||||
return 1; /* conn exists even without link_status yet */
|
||||
}
|
||||
c = c->next;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* ── sort: read RTT from node_addresses, sort nodes ascending ── */
|
||||
|
||||
static void ac_sort_by_rtt(sqlite3* db, uint64_t* ids, int count) {
|
||||
if (!db || count < 2) return;
|
||||
struct ac_rtt_item { uint64_t id; int32_t rtt; };
|
||||
struct ac_rtt_item* items = u_malloc((size_t)count * sizeof(*items));
|
||||
if (!items) return;
|
||||
for (int i = 0; i < count; i++) {
|
||||
items[i].id = ids[i]; items[i].rtt = INT32_MAX;
|
||||
}
|
||||
sqlite3_stmt* st = NULL;
|
||||
if (sqlite3_prepare_v2(db,
|
||||
"SELECT node_id, rtt FROM node_addresses WHERE rtt IS NOT NULL ORDER BY rtt ASC",
|
||||
-1, &st, NULL) == SQLITE_OK) {
|
||||
while (sqlite3_step(st) == SQLITE_ROW) {
|
||||
uint64_t nid = (uint64_t)sqlite3_column_int64(st, 0);
|
||||
int32_t rtt = sqlite3_column_int(st, 1);
|
||||
for (int i = 0; i < count; i++) if (items[i].id == nid) { items[i].rtt = rtt; break; }
|
||||
}
|
||||
sqlite3_finalize(st);
|
||||
}
|
||||
for (int i = 1; i < count; i++) {
|
||||
struct ac_rtt_item key = items[i]; int j = i - 1;
|
||||
while (j >= 0 && items[j].rtt > key.rtt) { items[j+1] = items[j]; j--; }
|
||||
items[j+1] = key;
|
||||
}
|
||||
for (int i = 0; i < count; i++) ids[i] = items[i].id;
|
||||
u_free(items);
|
||||
}
|
||||
|
||||
/* ── collect + sort nodes from SQLite peers (all channels) ── */
|
||||
|
||||
static int ac_collect_nodes(struct UTUN_INSTANCE* inst, uint64_t** out, int* out_count) {
|
||||
int cap = 32, cnt = 0;
|
||||
uint64_t* ids = u_malloc((size_t)cap * sizeof(uint64_t));
|
||||
if (!ids) return -1;
|
||||
|
||||
uint8_t ch_buf[4096]; size_t ch_len;
|
||||
if (chat_core_list_channels(ch_buf, sizeof(ch_buf), &ch_len) != 0 || ch_len < 2)
|
||||
{ if (cnt == 0) { u_free(ids); *out = NULL; *out_count = 0; return 0; } }
|
||||
|
||||
uint16_t ch_cnt; memcpy(&ch_cnt, ch_buf, 2);
|
||||
const uint8_t* p = ch_buf + 2; size_t rem = ch_len - 2;
|
||||
for (uint16_t ci = 0; ci < ch_cnt && rem >= 1; ci++) {
|
||||
ac->channel_count = 0;
|
||||
uint8_t buf[4096]; size_t buf_len;
|
||||
if (chat_core_list_channels(buf, sizeof(buf), &buf_len) != 0 || buf_len < 2) return 0;
|
||||
uint16_t cnt; memcpy(&cnt, buf, 2);
|
||||
const uint8_t* p = buf + 2; size_t rem = buf_len - 2;
|
||||
ac->channel_ids = u_calloc(cnt, sizeof(char*));
|
||||
if (!ac->channel_ids) return 0;
|
||||
for (uint16_t i = 0; i < cnt && rem >= 1; i++) {
|
||||
uint8_t id_len = *p++; rem--;
|
||||
if (rem < id_len) break;
|
||||
char cid[64]; memcpy(cid, p, id_len); cid[id_len] = '\0'; p += id_len; rem -= id_len;
|
||||
|
||||
uint8_t pb[2048]; size_t plen;
|
||||
if (chat_core_list_peers(cid, pb, sizeof(pb), &plen) != 0 || plen < 2) continue;
|
||||
uint16_t pc; memcpy(&pc, pb, 2);
|
||||
for (uint16_t pj = 0; pj < pc && (size_t)(2 + (pj+1)*8) <= plen; pj++) {
|
||||
uint64_t nid; memcpy(&nid, pb + 2 + pj*8, 8);
|
||||
if (nid == 0 || nid == inst->node_id) continue;
|
||||
if (ac_node_busy(inst, nid)) continue;
|
||||
int dup = 0;
|
||||
for (int j = 0; j < cnt; j++) if (ids[j] == nid) { dup = 1; break; }
|
||||
if (dup) continue;
|
||||
if (cnt >= cap) { cap *= 2; ids = u_realloc(ids, (size_t)cap * sizeof(uint64_t)); if (!ids) return -1; }
|
||||
ids[cnt++] = nid;
|
||||
}
|
||||
ac->channel_ids[i] = u_malloc(id_len + 1);
|
||||
if (ac->channel_ids[i]) { memcpy(ac->channel_ids[i], p, id_len); ac->channel_ids[i][id_len] = '\0'; ac->channel_count++; }
|
||||
p += id_len; rem -= id_len;
|
||||
}
|
||||
|
||||
if (cnt > 0) {
|
||||
sqlite3* db = inst->topo_groups ? inst->topo_groups->topo_sqlite_db : NULL;
|
||||
if (db) ac_sort_by_rtt(db, ids, cnt);
|
||||
}
|
||||
*out = ids; *out_count = cnt;
|
||||
return cnt;
|
||||
return ac->channel_count;
|
||||
}
|
||||
|
||||
/* ── connect to node via direct ETCP (timeout handled by chat_core_connect_auto) ── */
|
||||
/* ── GC: close expired flights (no link_status after AC_GC_TIMEOUT_MS) ── */
|
||||
|
||||
static int ac_connect_one(struct auto_connect* ac, uint64_t nid) {
|
||||
struct ac_flight* f = u_calloc(1, sizeof(*f));
|
||||
if (!f) return -1;
|
||||
f->ac = ac; f->node_id = nid;
|
||||
ac->in_flight++;
|
||||
chat_core_connect_auto(nid, ac_result_cb, f);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: connecting to node 0x%016llx (in_flight=%d)",
|
||||
AC_ID, (unsigned long long)nid, ac->in_flight);
|
||||
static void ac_gc(struct auto_connect* ac) {
|
||||
uint64_t now = get_time_tb();
|
||||
uint64_t deadline = (uint64_t)AC_GC_TIMEOUT_MS * 10;
|
||||
for (int i = 0; i < AC_MAX_FLIGHTS; i++) {
|
||||
if (!ac->flights[i].ca_state) continue;
|
||||
if (now - ac->flights[i].created_tb < deadline) continue;
|
||||
uint64_t nid = ac->flights[i].node_id;
|
||||
/* check if link already UP — if so, just free slot (already connected) */
|
||||
struct ETCP_CONN* c = ac->inst->connections;
|
||||
int found_up = 0;
|
||||
while (c) {
|
||||
if (c->peer_node_id == nid) {
|
||||
struct ETCP_LINK* l = c->links;
|
||||
while (l) { if (l->initialized && l->link_status) { found_up = 1; break; } l = l->next; }
|
||||
if (found_up) break;
|
||||
}
|
||||
c = c->next;
|
||||
}
|
||||
if (found_up) {
|
||||
chat_core_connect_auto_cancel(ac->flights[i].ca_state);
|
||||
memset(&ac->flights[i], 0, sizeof(ac->flights[i]));
|
||||
continue;
|
||||
}
|
||||
/* expired and not up — cancel */
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: GC closing flight node=0x%016llx (expired)", AC_ID, (unsigned long long)nid);
|
||||
chat_core_connect_auto_cancel(ac->flights[i].ca_state);
|
||||
memset(&ac->flights[i], 0, sizeof(ac->flights[i]));
|
||||
}
|
||||
}
|
||||
|
||||
/* ── count occupied flight slots ── */
|
||||
|
||||
static int ac_flight_count(struct auto_connect* ac) {
|
||||
int n = 0;
|
||||
for (int i = 0; i < AC_MAX_FLIGHTS; i++) if (ac->flights[i].ca_state) n++;
|
||||
return n;
|
||||
}
|
||||
|
||||
/* ── find first free flight slot ── */
|
||||
|
||||
static int ac_find_free_slot(struct auto_connect* ac) {
|
||||
for (int i = 0; i < AC_MAX_FLIGHTS; i++) if (!ac->flights[i].ca_state) return i;
|
||||
return -1;
|
||||
}
|
||||
|
||||
/* ── check if node already has active ETCP connection ── */
|
||||
|
||||
static int ac_node_has_conn(struct UTUN_INSTANCE* inst, uint64_t nid) {
|
||||
struct ETCP_CONN* c = inst->connections;
|
||||
while (c) {
|
||||
if (c->peer_node_id == nid) return 1;
|
||||
c = c->next;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* ── fill up to AC_MAX_FLIGHTS by advancing cursor ── */
|
||||
|
||||
static void ac_fill(struct auto_connect* ac) {
|
||||
if (ac->channel_count == 0) return;
|
||||
/* try all channels up to 2 full cycles without finding a candidate → give up */
|
||||
int max_iter = ac->channel_count * 2 + 10;
|
||||
while (ac_flight_count(ac) < AC_MAX_FLIGHTS && max_iter-- > 0) {
|
||||
int slot = ac_find_free_slot(ac);
|
||||
if (slot < 0) break;
|
||||
char* ch_id = ac->channel_ids[ac->ch_cursor];
|
||||
uint8_t pb[2048]; size_t plen;
|
||||
if (chat_core_list_peers(ch_id, pb, sizeof(pb), &plen) != 0 || plen < 2) {
|
||||
/* channel has no peers or error — skip to next */
|
||||
ac->ch_cursor++; ac->peer_cursor = 0;
|
||||
if (ac->ch_cursor >= ac->channel_count) { ac->ch_cursor = 0; ac_load_channels(ac); }
|
||||
continue;
|
||||
}
|
||||
uint16_t pc; memcpy(&pc, pb, 2);
|
||||
if (ac->peer_cursor >= (int)pc) {
|
||||
ac->ch_cursor++; ac->peer_cursor = 0;
|
||||
if (ac->ch_cursor >= ac->channel_count) { ac->ch_cursor = 0; ac_load_channels(ac); }
|
||||
continue;
|
||||
}
|
||||
/* get next peer */
|
||||
uint64_t nid; memcpy(&nid, pb + 2 + ac->peer_cursor * 8, 8);
|
||||
ac->peer_cursor++;
|
||||
if (nid == 0 || nid == ac->inst->node_id) continue;
|
||||
|
||||
/* already connected? */
|
||||
if (ac_node_has_conn(ac->inst, nid)) continue;
|
||||
|
||||
/* already in a flight slot? */
|
||||
int dup = 0;
|
||||
for (int i = 0; i < AC_MAX_FLIGHTS; i++)
|
||||
if (ac->flights[i].ca_state && ac->flights[i].node_id == nid) { dup = 1; break; }
|
||||
if (dup) continue;
|
||||
|
||||
/* launch */
|
||||
struct ac_flight* f = &ac->flights[slot];
|
||||
f->node_id = nid;
|
||||
f->created_tb = get_time_tb();
|
||||
chat_core_connect_auto(nid, ac_result_cb, f, &f->ca_state);
|
||||
if (!f->ca_state) {
|
||||
f->node_id = 0; f->created_tb = 0;
|
||||
continue;
|
||||
}
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: flight[%d] launched node=0x%016llx", AC_ID, slot, (unsigned long long)nid);
|
||||
uint8_t evt[7]; evt[0] = 0; uint16_t t = 0, n = (uint16_t)ac->channel_count, s = 0;
|
||||
memcpy(evt + 1, &t, 2); memcpy(evt + 3, &n, 2); memcpy(evt + 5, &s, 2);
|
||||
gui_bridge_post(GUI_EVT_AUTO_CONNECT_STATUS, evt, 7);
|
||||
}
|
||||
}
|
||||
|
||||
/* ── result callback (fired by chat_core_connect_auto) ── */
|
||||
|
||||
static void ac_result_cb(int result, uint64_t node_id, void* arg) {
|
||||
struct ac_flight* f = (struct ac_flight*)arg;
|
||||
struct auto_connect* ac = f->ac;
|
||||
if (!ac->active) { u_free(f); return; }
|
||||
ac->in_flight--; ac->total_tried++;
|
||||
if (result == CONN_MGR_OK) ac->success_count++;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: result node=0x%016llx result=%d succ=%d in_flight=%d total=%d",
|
||||
AC_ID, (unsigned long long)node_id, result, ac->success_count, ac->in_flight, ac->total_tried);
|
||||
uint8_t evt[7]; evt[0] = ac->success_count >= AC_TARGET_SUCCESS ? 1 : 0;
|
||||
uint16_t t = (uint16_t)ac->total_tried, n = (uint16_t)ac->node_count, s = (uint16_t)ac->success_count;
|
||||
memcpy(evt + 1, &t, 2); memcpy(evt + 3, &n, 2); memcpy(evt + 5, &s, 2);
|
||||
gui_bridge_post(GUI_EVT_AUTO_CONNECT_STATUS, evt, 7);
|
||||
u_free(f);
|
||||
if (!ac->active) return;
|
||||
while (ac->in_flight < AC_MAX_PARALLEL && ac->next_index < ac->node_count) {
|
||||
uint64_t nid = ac->node_ids[ac->next_index++];
|
||||
ac_connect_one(ac, nid);
|
||||
}
|
||||
if (ac->in_flight == 0 && ac->success_count < AC_TARGET_SUCCESS)
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: all tries done, %d/%d connected, waiting retry",
|
||||
AC_ID, ac->success_count, ac->total_tried);
|
||||
if (!g_ac || !g_ac->active) return;
|
||||
const char* rs = result == CONN_MGR_OK ? "OK" : "FAIL";
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: result node=0x%016llx %s", AC_ID, (unsigned long long)node_id, rs);
|
||||
/* DON'T clear ca_state — GC will close connection and free slot after AC_GC_TIMEOUT_MS */
|
||||
}
|
||||
|
||||
/* ── launch initial batch ── */
|
||||
|
||||
static void ac_launch_batch(struct auto_connect* ac) {
|
||||
while (ac->in_flight < AC_MAX_PARALLEL && ac->next_index < ac->node_count) {
|
||||
uint64_t nid = ac->node_ids[ac->next_index++];
|
||||
ac_connect_one(ac, nid);
|
||||
}
|
||||
}
|
||||
|
||||
/* ── retry timer callback ── */
|
||||
/* ── retry timer callback (every AC_RETRY_MS) ── */
|
||||
|
||||
static void ac_retry_timer_cb(void* arg) {
|
||||
struct auto_connect* ac = (struct auto_connect*)arg;
|
||||
ac->retry_timer = NULL;
|
||||
if (!ac->active) return;
|
||||
int connected = ac_count_connected(ac->inst);
|
||||
if (connected >= AC_TARGET_SUCCESS) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: %d connected, target reached, stopping",
|
||||
AC_ID, connected);
|
||||
chat_sync_auto_connect_stop();
|
||||
return;
|
||||
}
|
||||
if (ac->in_flight > 0) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: retry, %d in_flight, waiting", AC_ID, ac->in_flight);
|
||||
} else {
|
||||
if (ac->node_ids) { u_free(ac->node_ids); ac->node_ids = NULL; }
|
||||
ac->node_count = 0; ac->next_index = 0;
|
||||
ac_collect_nodes(ac->inst, &ac->node_ids, &ac->node_count);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: retry, %d nodes to try, %d connected",
|
||||
AC_ID, ac->node_count, connected);
|
||||
if (ac->node_count > 0) {
|
||||
uint8_t evt[7]; evt[0] = 0;
|
||||
uint16_t t = 0, n = (uint16_t)ac->node_count, s = (uint16_t)ac->success_count;
|
||||
memcpy(evt + 1, &t, 2); memcpy(evt + 3, &n, 2); memcpy(evt + 5, &s, 2);
|
||||
gui_bridge_post(GUI_EVT_AUTO_CONNECT_STATUS, evt, 7);
|
||||
ac_launch_batch(ac);
|
||||
}
|
||||
}
|
||||
ac->retry_timer = uasync_set_timeout(ac->inst->ua, AC_RETRY_INTERVAL_MS * 10, ac, ac_retry_timer_cb, "ac_retry");
|
||||
if (ac->channel_count == 0) ac_load_channels(ac);
|
||||
ac_gc(ac);
|
||||
ac_fill(ac);
|
||||
ac->retry_timer = uasync_set_timeout(ac->inst->ua, AC_RETRY_MS * 10, ac, ac_retry_timer_cb, "ac_retry");
|
||||
}
|
||||
|
||||
/* ═══════════════════════════════════════════════════════════════════════
|
||||
@@ -251,22 +226,11 @@ void chat_sync_auto_connect_start(struct UTUN_INSTANCE* inst) {
|
||||
ac->inst = inst;
|
||||
ac->active = 1;
|
||||
g_ac = ac;
|
||||
int connected = ac_count_connected(inst);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: start, %d already connected", AC_ID, connected);
|
||||
if (connected >= AC_TARGET_SUCCESS) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: already have %d connections, idle", AC_ID, connected);
|
||||
ac->retry_timer = uasync_set_timeout(inst->ua, AC_RETRY_INTERVAL_MS * 10, ac, ac_retry_timer_cb, "ac_retry");
|
||||
return;
|
||||
}
|
||||
ac_collect_nodes(inst, &ac->node_ids, &ac->node_count);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: collected %d candidate nodes", AC_ID, ac->node_count);
|
||||
if (ac->node_count > 0) {
|
||||
uint8_t evt[7]; evt[0] = 0; uint16_t z = 0, n = (uint16_t)ac->node_count;
|
||||
memcpy(evt + 1, &z, 2); memcpy(evt + 3, &n, 2); memcpy(evt + 5, &z, 2);
|
||||
gui_bridge_post(GUI_EVT_AUTO_CONNECT_STATUS, evt, 7);
|
||||
ac_launch_batch(ac);
|
||||
}
|
||||
ac->retry_timer = uasync_set_timeout(inst->ua, AC_RETRY_INTERVAL_MS * 10, ac, ac_retry_timer_cb, "ac_retry");
|
||||
ac_load_channels(ac);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: start, %d channels loaded", AC_ID, ac->channel_count);
|
||||
ac_gc(ac);
|
||||
ac_fill(ac);
|
||||
ac->retry_timer = uasync_set_timeout(inst->ua, AC_RETRY_MS * 10, ac, ac_retry_timer_cb, "ac_retry");
|
||||
}
|
||||
|
||||
void chat_sync_auto_connect_stop(void) {
|
||||
@@ -274,15 +238,18 @@ void chat_sync_auto_connect_stop(void) {
|
||||
if (!ac) return;
|
||||
ac->active = 0; g_ac = NULL;
|
||||
if (ac->retry_timer) { uasync_cancel_timeout(ac->inst->ua, ac->retry_timer); ac->retry_timer = NULL; }
|
||||
uint8_t evt[7]; evt[0] = 2;
|
||||
uint16_t z = 0; memcpy(evt + 1, &z, 2); memcpy(evt + 3, &z, 2); memcpy(evt + 5, &z, 2);
|
||||
for (int i = 0; i < AC_MAX_FLIGHTS; i++)
|
||||
if (ac->flights[i].ca_state) { chat_core_connect_auto_cancel(ac->flights[i].ca_state); memset(&ac->flights[i], 0, sizeof(ac->flights[i])); }
|
||||
if (ac->channel_ids) { for (int i = 0; i < ac->channel_count; i++) u_free(ac->channel_ids[i]); u_free(ac->channel_ids); }
|
||||
uint8_t evt[7]; evt[0] = 2; uint16_t z = 0;
|
||||
memcpy(evt + 1, &z, 2); memcpy(evt + 3, &z, 2); memcpy(evt + 5, &z, 2);
|
||||
gui_bridge_post(GUI_EVT_AUTO_CONNECT_STATUS, evt, 7);
|
||||
/* DON'T free ac — in-flight callbacks still reference it via ac_flight */
|
||||
u_free(ac);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: stopped", AC_ID);
|
||||
}
|
||||
|
||||
void chat_sync_auto_connect_switch_group(struct UTUN_INSTANCE* inst, uint64_t new_group_id) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: switching to group 0x%016llx", AC_ID, (unsigned long long)new_group_id);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CONNECTIVITY, "%s: switch group 0x%016llx", AC_ID, (unsigned long long)new_group_id);
|
||||
chat_sync_auto_connect_stop();
|
||||
chat_sync_auto_connect_start(inst);
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user