Define descriptor-based multicast API and usage contract
This commit is contained in:
evgeny committed
1 parent
8d5459f8fa
commit
e416d69212
1 file changed
+163
-40
+163
-40
@@ -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
|
||||
|
||||
Reference in new issue
Block a user