Preserve broadcast UUIDs until TTL and document callback contracts
This commit is contained in:
evgeny committed
1 parent
16f81575ce
commit
484b620016
2 files changed
+34
-18
No files matched your search
+16
-13
@@ -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) {
|
||||
|
||||
+18
-5
@@ -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
|
||||
|
||||
Reference in new issue
Block a user