From e416d69212de519f4f7305e5bc14b34d72fb6b08 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 20 Oct 2026 14:44:23 +0200 Subject: [PATCH] Define descriptor-based multicast API and usage contract --- src/routing_layer/multicast.h | 203 +++++++++++++++++++++++++++------- 1 file changed, 163 insertions(+), 40 deletions(-) diff --git a/src/routing_layer/multicast.h b/src/routing_layer/multicast.h index b6bf467f..f44ec18f 100644 --- a/src/routing_layer/multicast.h +++ b/src/routing_layer/multicast.h @@ -1,24 +1,69 @@ -/* multicast — адресная рассылка по маршрутам topo-группы, с разделением по next hop. - * send копирует данные; descriptor задаёт приложение источника, это не номер пакета. - * Одна FIFO на ETCP-соединение обслуживает все группы/источники/дескрипторы. - * При переполнении теряется новая ветка; источник получает OVERLOAD через router. - * Все вызовы и callbacks — uasync-поток. В callbacks допустима отправка, но нельзя - * уничтожать ядро/группу/модуль. Данные callbacks заимствованы только на время вызова. +/* multicast — отправка одного payload явно указанному списку узлов topo-группы. + * Общая часть пути передаётся одним пакетом; при расхождении маршрутов пакет + * разделяется на ветки с соответствующими подмножествами получателей. * - * Wire DATA, все числа little-endian: - * cmd:1 | group:8 | descriptor:2 | count:2 | source:8 | recipients:8*count | - * payload | visited:8*hops | hops:1. - * Получатели строго возрастают. Источник начинает visited собой, каждый транзит - * добавляет себя ровно один раз. Первый visited=source, последний=ETCP peer. - * Полный резерв visited учитывается до отправки, payload на транзите не меняется. + * Как отправлять: + * 1. multicast_open(group, on_loss, arg) выделяет исходящий дескриптор. + * Номер 1..65535 уникален среди открытых дескрипторов данного ядра. + * Выдача последовательная: 1, 2, ... 65535, затем снова с 1, пропуская + * занятые номера. Счётчик общий для всех групп ядра; close его не откатывает. + * Группа и обработчик потерь закрепляются за ним до закрытия. + * 2. Передать получателям свой node_id и номер дескриптора через протокол + * использующего сервиса. Получатели должны установить подписки до DATA. + * 3. multicast_send(inst, descriptor, recipients, count, payload, len). + * Можно отправлять много пакетов; список адресатов задаётся для каждого. + * Список и payload копируются: после возврата их можно менять/освобождать. + * 4. Перегрузки и локальные ошибки приходят в on_loss этого дескриптора. + * loss->descriptor указывает исходящий дескриптор, loss->recipients — + * только пострадавшую часть списка. Решение о снижении нагрузки принимает сервис. + * 5. multicast_close(inst, descriptor) завершает использование дескриптора. * - * Wire OVERLOAD — payload отдельного router-сервиса: - * descriptor:2 | src:8 | dst:8 | packets:4 | count:2 | prefix_len:1 | - * prefix:16 | recipients:8*count. - * src/dst — концы перегруженного участка; router dst — источник DATA. - * packets относится к КАЖДОМУ указанному получателю. Части длинного отчёта имеют - * непересекающиеся списки: складывать их packets как число исходных пакетов нельзя. - * Отчёт сохраняет префикс первого пакета; разные наборы получателей не смешиваются. + * Как принимать: + * 1. Получить node_id источника и ЕГО исходящий дескриптор от сервиса. + * 2. multicast_subscribe(group, src_node, src_descriptor, on_data, arg). + * Подписка локальная, ключ — (group_id, src_node, src_descriptor). + * Она не посылает JOIN и не добавляет нас в список адресатов у источника. + * Собственный исходящий дескриптор для приёма выделять не требуется. + * 3. on_data вызывается только для пакетов, где мы указаны получателем. + * Без подписки локальная доставка тихо отбрасывается, без лога и отчёта. + * Транзит другим получателям продолжается независимо от наших подписок. + * 4. multicast_unsubscribe(subscription) прекращает приём и освобождает handle. + * + * Минимальный порядок (на разных узлах; on_loss/on_data — callbacks сервиса): + * int fd = multicast_open(send_group, on_loss, sender_ctx); + * if (fd < 0) ...; // дескриптор не создан + * // сервис передаёт source_node_id и (uint16_t)fd получателю + * struct multicast_subscription* sub = + * multicast_subscribe(recv_group, source_node_id, (uint16_t)fd, on_data, receiver_ctx); + * if (!sub) ...; // подписка не создана + * // после готовности подписки, на источнике: + * int rc = multicast_send(send_inst, (uint16_t)fd, receivers, count, data, size); + * if (rc < 0) ...; // этот вызов не принят + * // при завершении, каждый на своём узле: + * multicast_unsubscribe(sub); + * multicast_close(send_inst, (uint16_t)fd); + * + * Условия использования: + * - Все вызовы и callbacks выполняются в uasync-потоке соответствующего ядра. + * - Callbacks отложены: open/subscribe/send/close/unsubscribe их не вызывают. + * В callback можно отправлять, закрывать свой дескриптор и отписывать себя. + * Уничтожать ядро/группу/модуль из callback нельзя. + * - payload, loss и loss->recipients заимствованы до возврата из callback. + * arg принадлежит сервису: он должен жить до close/unsubscribe и завершения + * уже выполняющегося callback. После закрытия новых callbacks не будет. + * - Подпиской владеет вызывающий: unsubscribe обязателен даже после удаления + * группы. Удаление группы/ядра деактивирует подписку, но handle остаётся + * валиден до unsubscribe; после unsubscribe использовать его нельзя. + * - Удаление группы автоматически закрывает её исходящие дескрипторы. + * - Успех send означает принятие к обработке, а не доставку. Возможны частичные + * потери; сквозных ACK/повторов нет, порядок при смене маршрута не гарантируется. + * Сам отчёт тоже может потеряться: отсутствие on_loss не подтверждает доставку. + * + * Состояние API: этот заголовок описывает новый контракт; multicast.c пока + * использует прежнюю модель с общими callbacks и требует адаптации. + * Поколений нет: DATA, отчёты и подписки используют 16-битный номер дескриптора. + * Новое ядро начинает выдачу с 1. После повторной выдачи номера запоздалый + * сетевой пакет с тем же источником и номером неотличим от пакета нового владельца. */ #ifndef UTUN_MULTICAST_H #define UTUN_MULTICAST_H @@ -33,10 +78,14 @@ extern "C" { struct UTUN_INSTANCE; struct TOPO_GROUP; struct multicast_ctx; +struct multicast_subscription; /* непрозрачный handle локальной подписки; освобождает вызывающий */ -#define MULTICAST_QUEUE_BYTES (256u * 1024u) -#define MULTICAST_REPORT_BYTES (16u * 1024u) -#define MULTICAST_REPORT_INTERVAL_TB 3000 +#define MULTICAST_DESCRIPTOR_INVALID 0 +#define MULTICAST_QUEUE_BYTES (256u * 1024u) /* одна FIFO на ETCP-соединение для всех рассылок */ +#define MULTICAST_REPORT_BYTES (16u * 1024u) /* общий бюджет накопленных отчётов и окон троттлинга */ +#define MULTICAST_REPORT_INTERVAL_TB 3000 /* 300 мс; timebase = 0.1 мс */ + +/* Размеры служебной оболочки; payload вместе с ней должен помещаться в 16 КиБ ETCP. */ #define MULTICAST_MAX_VISITED 64 #define MULTICAST_HEADER_SIZE 21 #define MULTICAST_PATH_RESERVE (8 * MULTICAST_MAX_VISITED + 1) @@ -53,42 +102,116 @@ struct multicast_loss { uint64_t group_id; /* группа исходной рассылки */ uint64_t source_node_id; /* источник исходного DATA */ uint64_t src, dst; /* узел потери и выбранный next hop; dst=0 без маршрута */ - uint16_t descriptor; /* идентификатор приложения источника */ + uint16_t descriptor; /* исходящий дескриптор, callback которого получает этот отчёт */ uint16_t recipient_count; /* число ID в recipients */ const uint64_t* recipients; /* заимствованный массив в host byte order */ uint32_t packets; /* потери для каждого получателя этого списка */ - enum multicast_loss_reason reason; + enum multicast_loss_reason reason; /* перегрузка либо локальная причина потери */ uint8_t prefix_len; /* 0..16 значимых байт */ uint8_t prefix[16]; /* начало payload первого потерянного пакета */ }; -typedef void (*multicast_recv_fn)(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t source, - uint16_t descriptor, const uint8_t* payload, size_t len, void* arg); +/* Приём для конкретной подписки. len может быть 0; payload нельзя сохранять без копирования. + * subscription можно передать в unsubscribe прямо здесь; после этого handle недействителен. + * src_node/src_descriptor/group известны сервису из subscribe и могут храниться в arg. */ +typedef void (*multicast_recv_fn)(struct multicast_subscription* subscription, const uint8_t* payload, size_t len, void* arg); + +/* Уведомление конкретному исходящему дескриптору. loss->src/dst — перегруженный + * участок, не пара «исходный отправитель / конечный получатель». + * packets относится к каждому ID в recipients. При разделении отчёта на части + * нельзя складывать packets как число уникальных исходных пакетов. + * Допустим multicast_close(inst, loss->descriptor) из этого callback. */ typedef void (*multicast_loss_fn)(struct UTUN_INSTANCE* inst, const struct multicast_loss* loss, void* arg); struct multicast_stats { - uint64_t sent, delivered, overload_packets, no_route_packets, loop_packets, invalid_packets, internal_errors; - uint64_t reports_sent, reports_received, reports_evicted; /* отправленные части / принятые / вытесненные записи */ - size_t queued_bytes, report_bytes; /* данные в FIFO / память отчётов и окон, без уже переданного транспорту */ + uint64_t sent; /* ветки DATA, переданные ETCP */ + uint64_t delivered; /* вызовы обработчиков входящих подписок */ + uint64_t unsubscribed; /* тихие отбрасывания локальной доставки без подписки */ + uint64_t overload_packets; /* ветки, отброшенные из-за лимита исходящей FIFO */ + uint64_t no_route_packets; /* ветки без доступного маршрута */ + uint64_t loop_packets; /* петли и превышения длины пройденного пути */ + uint64_t invalid_packets; /* некорректные DATA/отчёты */ + uint64_t internal_errors; /* отказы выделения памяти или передачи транспорту */ + uint64_t reports_sent; /* переданные части отчётов, включая локальные уведомления */ + uint64_t reports_received; /* принятые части отчётов */ + uint64_t reports_evicted; /* записи отчётов, вытесненные из бюджета 16 КиБ */ + size_t queued_bytes; /* DATA в FIFO модуля, без уже переданного ETCP */ + size_t report_bytes; /* память накопленных отчётов и окон, без переданного router */ unsigned queues; /* лениво созданные FIFO, в том числе временно опустевшие */ + unsigned descriptors; /* открытые исходящие дескрипторы */ + unsigned subscriptions; /* активные входящие подписки */ }; -/* Инициализация после etcp_router_init; destroy до router/ETCP. 0/-1; повторный init — ошибка. */ +/* Открыть исходящий дескриптор в работающей group. on_loss обязателен; arg может быть NULL. + * Выдать следующий свободный номер по циклическому порядку 1..65535, начиная после + * последнего выданного; 0 не выдаётся. Закрытые номера используются только при обходе. + * Возврат: 1..65535 — выделенный номер, <0 — ошибка (включая отсутствие свободных номеров). + * До проверки результата не приводить его к uint16_t. Подписку на приём не создаёт. */ +int multicast_open(struct TOPO_GROUP* group, multicast_loss_fn on_loss, void* arg); + +/* Отправить через открытый descriptor, принадлежащий inst; группа берётся из дескриптора. + * recipients: count>0 ненулевых node_id; повторы удаляются, собственный ID допустим. + * len может быть 0; при len>0 payload обязателен. Размер payload, списка и полного + * резерва пройденного пути должен помещаться в 16 КиБ. + * 0=принято к обработке; <0=вызов не принят, ничего не отправлено. + * После принятия отдельные ветки могут потеряться; уведомления идут в on_loss. + * Обе входные области копируются и после возврата больше не используются. */ +int multicast_send(struct UTUN_INSTANCE* inst, uint16_t descriptor, const uint64_t* recipients, + uint16_t count, const void* payload, size_t len); + +/* Закрыть исходящий descriptor данного inst: убрать его из поиска, отменить + * ещё не переданные DATA и ожидающие локальные callbacks этого дескриптора. + * Уже переданное транспорту/другим узлам отозвать нельзя. Удалённые подписки + * автоматически не закрываются. Сохранённый номер после close не использовать. + * Для отсутствующего дескриптора — диагностируемая ошибка без побочных действий. */ +void multicast_close(struct UTUN_INSTANCE* inst, uint16_t descriptor); + +/* Подписаться на локальную доставку (group_id, src_node, src_descriptor). + * src_node!=0, src_descriptor!=0, on_data обязателен; arg может быть NULL. + * Возврат: принадлежащий вызывающему handle либо NULL с диагностикой ошибки. + * Повторная подписка на тот же ключ запрещена; сначала unsubscribe прежней. + * Никаких сетевых запросов не отправляет и пропущенные ранее DATA не воспроизводит. */ +struct multicast_subscription* multicast_subscribe(struct TOPO_GROUP* group, uint64_t src_node, uint16_t src_descriptor, + multicast_recv_fn on_data, void* arg); + +/* Прекратить доставку, отменить ожидающие callbacks и освободить handle. + * NULL допустим. Работает и для подписки, деактивированной удалением группы/ядра. + * После вызова handle недействителен; повторный вызов с тем же указателем запрещён. + * Из собственного on_data допускается; память освобождается после его завершения. */ +void multicast_unsubscribe(struct multicast_subscription* subscription); + +/* Снимок общих счётчиков ядра в *out. 0=успех, <0=неверный аргумент/модуль не создан. + * Не открывает дескрипторов или подписок, счётчики не сбрасывает. */ +int multicast_get_stats(struct UTUN_INSTANCE* inst, struct multicast_stats* out); + +/* Далее — интеграция с ядром; сервисы используют open/send/close и subscribe/unsubscribe. + * init вызывается ядром один раз после etcp_router_init; 0=успех, <0=ошибка. + * destroy отменяет работу, закрывает дескрипторы, деактивирует подписки; повтор допустим. + * Вызывать до уничтожения router/ETCP и вне callbacks модуля. */ int multicast_init(struct UTUN_INSTANCE* inst); void multicast_destroy(struct UTUN_INSTANCE* inst); -/* Один приёмник и обработчик потерь на ядро; NULL отключает соответствующий callback. */ -void multicast_set_callbacks(struct UTUN_INSTANCE* inst, multicast_recv_fn recv, multicast_loss_fn loss, void* arg); -/* count>0, ID ненулевые; повторы удаляются. len может быть 0. 0=принято к обработке, - * -1=неверный аргумент/размер или отказ начального выделения. Частичные потери — через loss. - * Доставка best effort: повторов/сквозных ACK нет; порядок при смене маршрутов не гарантируется. */ -int multicast_send(struct TOPO_GROUP* group, uint16_t descriptor, const uint64_t* recipients, - uint16_t count, const void* payload, size_t len); -/* Снимок диагностики; 0/-1, не создаёт состояния. */ -int multicast_get_stats(struct UTUN_INSTANCE* inst, struct multicast_stats* out); -/* Внутренние hooks топологии: отложенный пересмотр маршрутов / отмена данных удаляемой группы. */ +/* Запланировать пересмотр ожидающих DATA после изменения маршрутов; повторные вызовы объединяются. */ void multicast_routes_changed(struct UTUN_INSTANCE* inst); +/* Перед удалением группы: закрыть её дескрипторы, деактивировать подписки, + * убрать её DATA/отчёты из общих очередей, сохранив работу остальных групп. */ void multicast_close_group(struct UTUN_INSTANCE* inst, uint64_t group_id); +/* Wire DATA, все числа little-endian: + * cmd:1 | group:8 | descriptor:2 | count:2 | source:8 | recipients:8*count | + * payload | visited:8*hops | hops:1. + * Получатели строго возрастают. Первый visited=source, последний=ETCP peer. + * Транзит добавляет себя один раз; повторное посещение запрещено. + * + * Wire OVERLOAD — payload отдельного router-сервиса: + * descriptor:2 | src:8 | dst:8 | packets:4 | count:2 | prefix_len:1 | + * prefix:16 | recipients:8*count. + * src/dst — концы перегруженного участка; router dst — исходный источник DATA. + * Разные наборы потерянных получателей не смешиваются. Префикс относится + * к первому потерянному пакету записи; для payload короче 16 байт остаток нулевой. + * Первая потеря сообщается сразу, последующие накапливаются до конца окна 300 мс. + * Таймер отправляет остаток независимо от поступления новых DATA. + */ + #ifdef __cplusplus } #endif