From 484b620016e52d74f2a9b3c4bad9ce35f3a2a785 Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 21 Oct 2026 18:08:50 +0200 Subject: [PATCH] Preserve broadcast UUIDs until TTL and document callback contracts --- src/broadcast.c | 29 ++++++++++++++++------------- src/broadcast.h | 23 ++++++++++++++++++----- 2 files changed, 34 insertions(+), 18 deletions(-) diff --git a/src/broadcast.c b/src/broadcast.c index 5b7ab2ce..dc85baa6 100644 --- a/src/broadcast.c +++ b/src/broadcast.c @@ -46,10 +46,13 @@ int broadcast_init(struct TOPO_GROUP* group) { ctx->messages = queue_new(group->instance->ua, 64, BMSG_UUID_OFFSET, BROADCAST_UUID_SIZE, "broadcast_msg"); if (!ctx->messages) { u_free(ctx); DEBUG_ERROR(DEBUG_CATEGORY_BGP, "broadcast_init: queue_new failed"); return -1; } - group->broadcast = ctx; - ctx->cleanup_timer = uasync_set_timeout(group->instance->ua, BROADCAST_CLEANUP_INTERVAL_TB, ctx, broadcast_cleanup_timer_cb, "broadcast_cleanup_timer"); - if (!ctx->cleanup_timer) DEBUG_ERROR(DEBUG_CATEGORY_BGP, "broadcast_init: cleanup_timer failed grp=%016llx", (unsigned long long)group->group_id); + if (!ctx->cleanup_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_BGP, "broadcast_init: cleanup_timer failed grp=%016llx", (unsigned long long)group->group_id); + queue_free(ctx->messages); u_free(ctx); + return -1; + } + group->broadcast = ctx; DEBUG_INFO(DEBUG_CATEGORY_BGP, "broadcast_init ok grp=%016llx", (unsigned long long)group->group_id); return 0; @@ -158,6 +161,7 @@ static void broadcast_fire_cbks(struct broadcast_ctx* ctx, const uint8_t* uuid, static void broadcast_cleanup_timer_cb(void* arg) { struct broadcast_ctx* ctx = (struct broadcast_ctx*)arg; if (!ctx || !ctx->group) return; + ctx->cleanup_timer = NULL; #ifdef UTUN_HAVE_STANDBY if (standby_get_sleep_tb() > 0) { ctx->cleanup_wait = standby_wait(ctx, broadcast_cleanup_timer_cb); @@ -168,20 +172,19 @@ static void broadcast_cleanup_timer_cb(void* arg) { uint64_t now_tb = get_time_tb(); int removed = 0; - while (1) { - struct ll_entry* e = queue_data_get(ctx->messages); - if (!e) break; + while (ctx->messages->head) { + struct ll_entry* e = ctx->messages->head; struct broadcast_msg* msg = (struct broadcast_msg*)e->data; - if (now_tb - msg->local_time_tb > BROADCAST_TTL_TB) { - queue_entry_free(e); removed++; - } else { - queue_data_put_first(ctx->messages, e); - break; - } + if (now_tb - msg->local_time_tb <= BROADCAST_TTL_TB) break; + queue_entry_free(queue_data_get(ctx->messages)); + removed++; } - if (removed > 0) DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "broadcast_cleanup: removed %d stale msgs grp=%016llx", removed, (unsigned long long)ctx->group->group_id); + DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "broadcast_cleanup: grp=%016llx removed=%d cached=%d", + (unsigned long long)ctx->group->group_id, removed, queue_entry_count(ctx->messages)); ctx->cleanup_timer = uasync_set_timeout(ctx->group->instance->ua, BROADCAST_CLEANUP_INTERVAL_TB, ctx, broadcast_cleanup_timer_cb, "broadcast_cleanup_timer"); + if (!ctx->cleanup_timer) + DEBUG_ERROR(DEBUG_CATEGORY_BGP, "broadcast_cleanup: timer rearm failed grp=%016llx", (unsigned long long)ctx->group->group_id); } void broadcast_recv(struct TOPO_GROUP* group, struct ETCP_CONN* from_conn, const uint8_t* payload, size_t payload_len) { diff --git a/src/broadcast.h b/src/broadcast.h index 0096f59f..45ab9f6c 100644 --- a/src/broadcast.h +++ b/src/broadcast.h @@ -2,7 +2,9 @@ * Контекст и кеш увиденных UUID принадлежат группе; TTL — срок хранения UUID, не счётчик сетевых хопов. * init_instance регистрирует общий ETCP-диспетчер, init(group) создаёт состояние отдельной группы. * Все вызовы/callbacks — uasync-поток; recv получает заимствованные uuid/data только на время callback. - * send копирует данные и возвращает 0 / -1; 0 не подтверждает доставку каждому участнику. */ + * send копирует данные и возвращает 0 / -1; 0 не подтверждает доставку каждому участнику. + * Получение нового UUID вызывает подписчиков; локальный send их не вызывает. Кеш общий для send/recv группы. + * Внутри recv-callback нельзя менять список подписчиков или уничтожать группу/контекст broadcast. */ #ifndef BROADCAST_H #define BROADCAST_H @@ -19,25 +21,36 @@ struct UTUN_INSTANCE; #define BROADCAST_MAX_DATA 1600 #define BROADCAST_UUID_SIZE 16 #define BROADCAST_CLEANUP_INTERVAL_TB 600000 // 1 минута -#define BROADCAST_TTL_TB 1200000 // 2 минуты +#define BROADCAST_TTL_TB 1200000 // 2 минуты; удаление при возрасте > TTL на периодической очистке +// uuid — 16 байт, data — data_len байт; заимствованы только на время вызова. arg — контекст из add_cbk. typedef void (*broadcast_recv_fn)(const uint8_t* uuid, const uint8_t* data, uint16_t data_len, void* arg); struct broadcast_cbk_entry { - broadcast_recv_fn fn; - void* arg; - struct broadcast_cbk_entry* next; + broadcast_recv_fn fn; // Функция подписчика + void* arg; // Заимствованный контекст, модуль его не освобождает + struct broadcast_cbk_entry* next; // Следующая регистрация (новые добавляются в начало) }; struct broadcast_ctx; +// Создаёт кеш и таймер живой group с instance/ua; 0 — успех, -1 — NULL group, повторный init или ошибка ресурсов. int broadcast_init(struct TOPO_GROUP* group); +// Отменяет таймер/standby-ожидание, освобождает кеш и регистрации, но не arg подписчиков; NULL/нет контекста допустимы. void broadcast_destroy(struct TOPO_GROUP* group); +// data не NULL даже при data_len=0; data_len <= BROADCAST_MAX_DATA, группа должна быть инициализирована. +// 0 — данные сохранены в кеше и рассылка предпринята; -1 — аргументы/отсутствие контекста/OOM записи кеша. +// Ошибки выделения/отправки отдельным соседям не меняют 0; собственные recv-callbacks не вызываются. int broadcast_send(struct TOPO_GROUP* group, const uint8_t* data, uint16_t data_len); +// Добавляет fn(arg) в начало; дубликаты разрешены и вызываются отдельно. При NULL/нет контекста/OOM не добавляет. +// arg должен жить до удаления регистрации или destroy; список не менять из recv-callback. void broadcast_add_cbk(struct TOPO_GROUP* group, broadcast_recv_fn fn, void* arg); +// Удаляет первое совпадение fn+arg (последнее добавленное); отсутствие совпадения ничего не меняет. void broadcast_remove_cbk(struct TOPO_GROUP* group, broadcast_recv_fn fn, void* arg); +// Регистрирует диспетчер ETCP_ID_BROADCAST для instance; возвращает результат etcp_bind или -1 при NULL. int broadcast_init_instance(struct UTUN_INSTANCE* instance); +// Снимает общий диспетчер, но не уничтожает контексты отдельных групп; NULL допустим. void broadcast_destroy_instance(struct UTUN_INSTANCE* instance); #ifdef __cplusplus