jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Retain and release media replication connection ownership

master
evgeny 7 days ago
parent
57d2828496
commit
035a2eb6d0
4 changed files with 21 additions and 33 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      doc/stream.txt
  2. 43
      src/media_delivery/media_delivery.c
  3. 2
      src/media_delivery/media_delivery.h
  4. 7
      tests/test_stream_service.c

2
doc/stream.txt
Unescape View File

@ -537,7 +537,7 @@ Stop/destroy выполняются по правилам ядра, вне callb
Лог общего прогона: /tmp/stream-service-check.log (107 с).
[+] Сетевая выдача — 11.10.2026:
stream_wire.c/h — big-endian формат, raw ETCP_ID_STREAM=0x07, routed ETCP_RT_ID_STREAM=0x39,
stream_wire.c/h — big-endian формат, raw ETCP_ID_STREAM=0x07, routed ETCP_RT_ID_STREAM=0x3C,
INFO/ACCEPT с границами окна и путём, 64-битные эпохи/handles, отдельная подпись источника на END.
stream_network.c/h — собственный NCD owner пира, общий для его потоков/групп; ожидание его UP,
прямые SUBSCRIBE/ACCEPT/REJECT/DATA/GAP/RENEW/UNSUBSCRIBE/CLOSED/END, routed REVERSE_REQ/CLOSED.

43
src/media_delivery/media_delivery.c
Unescape View File

@ -227,7 +227,9 @@ static void md_super_peer_remove(struct media_delivery_ctx* md, uint64_t node_id
if (!e) return;
struct media_super_peer* sp = (struct media_super_peer*)e->data;
if (sp->timeout_timer) { uasync_cancel_timeout(md->inst->ua, sp->timeout_timer); sp->timeout_timer = NULL; }
if (sp->connect_timer) { uasync_cancel_timeout(md->inst->ua, sp->connect_timer); sp->connect_timer = NULL; }
if (sp->connection) { conn_mgr_close(sp->connection); sp->connection = NULL; }
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "%s: super peer released node=%llx group=%llx", MD_ID,
(unsigned long long)sp->peer_node_id, (unsigned long long)sp->group_id);
queue_remove_data(md->super_peers, e);
queue_entry_free(e);
}
@ -345,18 +347,6 @@ static void md_repl_schedule_retrans(struct media_delivery_ctx* md, struct media
md_repl_retrans_cb, "md_repl_retrans");
}
static void md_reconnect_cb(void* arg) {
struct media_delivery_ctx* md = (struct media_delivery_ctx*)arg;
struct media_super_peer* peer = (struct media_super_peer*)md->super_peers;
while (peer) {
if (!peer->connected && peer->connect_timer) {
peer->connect_timer = NULL;
md_super_connect(md, peer->peer_node_id, peer->group_id);
}
peer = peer->connected ? NULL : peer;
}
}
/* ── replication send ── */
static void md_super_repl_send(struct media_delivery_ctx* md, struct media_super_peer* peer) {
@ -687,10 +677,10 @@ static void md_super_conn_cb(struct CONN_MGR_HANDLE* h, uint64_t node_id, uint64
if (!peer) { DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: conn_cb for unknown peer 0x%016llx", MD_ID, (unsigned long long)node_id); return; }
if (event != CONN_EVENT_UP) {
peer->connected = peer->hello_done = 0;
if (peer->timeout_timer) { uasync_cancel_timeout(md->inst->ua, peer->timeout_timer); peer->timeout_timer = NULL; }
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "%s: connect to supernode 0x%016llx failed rc=%d",
MD_ID, (unsigned long long)node_id, event);
peer->connect_timer = uasync_set_timeout(md->inst->ua,
MEDIA_RECONNECT_COOLDOWN_TB, md, md_reconnect_cb, "md_reconnect");
return;
}
@ -1106,6 +1096,7 @@ void media_delivery_stream_done(struct UTUN_INSTANCE* inst) {
static void md_super_connect(struct media_delivery_ctx* md, uint64_t peer_node_id, uint64_t group_id) {
struct media_super_peer* peer = md_super_peer_add(md, peer_node_id, group_id);
if (!peer) { DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "%s: super_peer_add failed for 0x%016llx", MD_ID, (unsigned long long)peer_node_id); return; }
if (peer->connection) return;
struct TOPO_GROUP* grp = topo_groups_find(md->inst->topo_groups, group_id);
if (!grp || !grp->conn_mgr) {
@ -1115,7 +1106,9 @@ static void md_super_connect(struct media_delivery_ctx* md, uint64_t peer_node_i
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "%s: connecting to supernode 0x%016llx",
MD_ID, (unsigned long long)peer_node_id);
conn_mgr_open(grp->instance, grp->group_id, peer_node_id, md_super_conn_cb, md, NULL);
if (conn_mgr_open(grp->instance, grp->group_id, peer_node_id, md_super_conn_cb, md, &peer->connection) < 0)
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "%s: supernode CM request failed node=%llx group=%llx", MD_ID,
(unsigned long long)peer_node_id, (unsigned long long)group_id);
}
/* ── supernode start/stop ── */
@ -1166,13 +1159,8 @@ static void md_super_stop(struct media_delivery_ctx* md) {
/* clear super_peers */
if (md->super_peers) {
struct ll_entry* se = md->super_peers->head;
while (se) {
struct media_super_peer* sp = (struct media_super_peer*)se->data;
if (sp->timeout_timer) { uasync_cancel_timeout(md->inst->ua, sp->timeout_timer); sp->timeout_timer = NULL; }
if (sp->connect_timer) { uasync_cancel_timeout(md->inst->ua, sp->connect_timer); sp->connect_timer = NULL; }
se = se->next;
}
while (md->super_peers->head)
md_super_peer_remove(md, ((struct media_super_peer*)md->super_peers->head->data)->peer_node_id);
queue_free(md->super_peers);
md->super_peers = NULL;
}
@ -1460,13 +1448,8 @@ void media_delivery_destroy(struct UTUN_INSTANCE* inst) {
}
if (md->super_peers) {
struct ll_entry* se = md->super_peers->head;
while (se) {
struct media_super_peer* sp = (struct media_super_peer*)se->data;
if (sp->timeout_timer) { uasync_cancel_timeout(md->inst->ua, sp->timeout_timer); sp->timeout_timer = NULL; }
if (sp->connect_timer) { uasync_cancel_timeout(md->inst->ua, sp->connect_timer); sp->connect_timer = NULL; }
se = se->next;
}
while (md->super_peers->head)
md_super_peer_remove(md, ((struct media_super_peer*)md->super_peers->head->data)->peer_node_id);
queue_free(md->super_peers);
md->super_peers = NULL;
}

2
src/media_delivery/media_delivery.h
Unescape View File

@ -41,7 +41,7 @@ struct media_super_peer {
uint8_t connected; // 1 = прямое подключение установлено
uint8_t hello_done; // 1 = SUPER_HELLO обмен завершён, можно реплицировать
void* timeout_timer; // таймер ретрансмита
void* connect_timer; // таймер переподключения (1 час при ошибке)
struct CONN_MGR_HANDLE* connection; // собственное CM-владение; восстановлением пути управляет CM
struct media_delivery_ctx* md; // обратная ссылка
};

7
tests/test_stream_service.c
Unescape View File

@ -1040,7 +1040,7 @@ static void relay_test_event(struct UTUN_INSTANCE *inst, const struct stream_eve
/* Все проверки помощи используют подписанное управление и настоящий raw ETCP, не прямой ingest. */
static void test_supernode_relay(void) {
int route_level = g_debug_config.category_levels[DEBUG_CATEGORY_ETCPROUTE];
debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_TRACE);
debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_INFO);
int diagnostic_level = g_debug_config.category_levels[DEBUG_CATEGORY_DEBUG];
debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_DEBUG);
struct TOPO_PEER_REQUEST *connections[3] = {0};
@ -1238,6 +1238,11 @@ int main(int argc, char **argv) {
for (unsigned i = 0; i < 3; i++) topo_group_peer_close(connections[i]);
for (unsigned i = 0; i < 4; i++) { stream_set_event_callback(nodes[i], NULL, NULL); utun_instance_destroy(nodes[i]); }
uasync_destroy(ua, 1); test_clock_finish();
if (u_get_allocated_count() != allocations) {
debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_ERROR);
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "relay tracked allocations before=%zu after=%zu", allocations, u_get_allocated_count());
u_report_unfreed_blocks();
}
CHECK(u_get_allocated_count() == allocations);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] isolated supernode relay checks complete, no tracked allocations remain");
return 0;

Write Preview
Loading…
Cancel
Save