Browse Source

Add signed FEC stream core and source buffering

master
evgeny 4 days ago
parent
commit
ec82d95a03
  1. 517
      doc/stream.txt
  2. 8
      doc/tasks.md
  3. 1
      lib/debug_config.c
  4. 3
      lib/debug_config.h
  5. 13
      src/Makefile.am
  6. 180
      src/stream/stream_block.c
  7. 48
      src/stream/stream_block.h
  8. 330
      src/stream/stream_buffer.c
  9. 53
      src/stream/stream_buffer.h
  10. 136
      src/stream/stream_metrics.c
  11. 64
      src/stream/stream_metrics.h
  12. 73
      src/stream/stream_proto.h
  13. 181
      src/stream/stream_source.c
  14. 36
      src/stream/stream_source.h
  15. 5
      tests/Makefile.am
  16. 326
      tests/test_stream_core.c

517
doc/stream.txt

@ -1,20 +1,511 @@
Модуль стриминга. Архитектурные нюансы. Модуль стриминга. Техническое задание, этап 1.
главная цель: узлы ищут более оптимальные пути доставки контента от источника до себя.
этап 1. 1. Цель и принятые решения
- реализуем в utun отдельным каталогом: src/stream
- источников стриминга может быть несколько в группе. для каждого источника в группе создается свой экземпляр. В этом экземпляре:
- отдельные списки хендлов подключений (ncd) для стриминга.
- узлы откуда берем трафик
- узлы куда отдаём трафик
Модуль размещается в src/stream. Узлы получают поток данных от источника через других подписчиков группы,
передают его своим подписчикам и постоянно ищут более качественные пути доставки отдельных субпотоков.
Этап 1 использует CHAT-группы с таблицей мемберов и member_sync/merkle_sync.
формат передачи: Принятые решения:
Источник формирует несколько потоков с разным битрейтом (до 3х штук). - Транспорт — существующий ETCP с быстрыми переповторами, мультилинком, шифрованием и управлением нагрузкой.
- Параметры FEC задаёт источник; 6+2 — значение по умолчанию, а не ограничение протокола.
- Источник подписывает каждый блок после разбиения на субпотоки, включая проверочные блоки.
- В источник добавляется циклический timestamp uint16_t в миллисекундах; он сохраняется при всей ретрансляции.
Задержка оценивается относительно предыдущих данных по дельтам timestamp и времени прихода, включая полную остановку.
- Передаём произвольные данные как есть, без обработки форматов видео, кадров и ключевых точек.
- Размер одной кодограммы после разбиения — около 1300 байт; источник сбрасывает неполную пачку через 100 мс.
- На каждый локальный экземпляр трансляции — кольцевой буфер 32 МБ для всех её субпотоков вместе.
Вся выдача идёт из этого буфера. Вытесненные данные недоступны: не успел получить — опоздал.
- Отдельных дозапросов диапазонов нет. Единственный способ получения — запросить субпоток с индекса или с конца.
- Обратное подключение — NCD + REVERSE_REQ стриминга через маршрутизатор группы.
- Статус источника и список подписок — опции самого мембера, распространяемые через Merkle.
Удалённая подписка актуальна только пока узел онлайн в BGP данной группы.
- Экземпляры определяются локальными подписками; узлы без подписки не принимают и не ретранслируют поток.
Активный источник также имеет собственный экземпляр и подписку на свою трансляцию.
- Подтверждение перегрузки — 500 мс. Число отключаемых выдач определяется ростом очереди и rate отдельных выдач.
- При запуске параллельно пробуем до 32 совместимых кандидатов; далее одна проба в секунду,
новый полный цикл не начинается раньше чем через 30 секунд от начала предыдущего.
Обратная совместимость с прежними форматами не требуется.
2. Трансляция, подписки и владение ресурсами
Надо учитывать что узлы могут быть перегружены трафиком. 2 сигнала перегрузки: очередь отправки и поток uasync Не справляется с нагрузкой (uasync thread load). Трансляция определяется ключом (group_id, source_node_id, source_epoch).
source_epoch — новый случайный ненулевой 64-битный идентификатор при каждом запуске источника.
В этапе 1 у одного источника в одной группе активна одна трансляция. Старые и новые запуски различаются по эпохе.
Пакеты и уведомления старого запуска не могут менять состояние нового.
- узлы учавствующие в стриминге анонсируют себя аналогично как сделано в рации. Локальная подписка на трансляцию создаёт один экземпляр, общий для получения и ретрансляции.
У модуля свой фоновый опросчик связности и поиска хороших линков - он пробует связь с узлами стриминна кроме мобильных, и обновляет их rtt и флаг прямой доступности. Подписка на трансляцию отличается от сетевой выдачи конкретного субпотока по конкретному handle.
Один экземпляр может получать субпотоки от нескольких пиров и выдавать их нескольким получателям.
Повторная локальная подписка не создаёт второй буфер; разные локальные потребители используют один экземпляр.
Закрытие последнего локального владельца подписки останавливает экземпляр и все его сетевые выдачи.
Экземпляр хранит:
- Дескриптор источника, FEC-контекст, состояние и номер эпохи.
- Собственные NCD handles, независимо от владельцев транспорта в других модулях и трансляциях.
- Upstream-выдачи: поставщик, handle, субпоток, текущая позиция, путь, прогресс и метрики.
- Downstream-выдачи: получатель, handle, субпоток, курсор в кольце и срок аренды.
- Кольцевой буфер, индекс наличия блоков, состояния сборки пачек и статистику.
- Лимит числа одновременно выдаваемых субпотоков.
Одна выдача — один субпоток одному получателю. Один субпоток двум получателям занимает два исходящих слота.
Есть также общий лимит исходящих выдач модуля: лимиты разных экземпляров не должны суммироваться без ограничения.
Модуль хранит общий контекст пира в пределах UTUN_INSTANCE: одну очередь отправки стриминга для всех его
трансляций и групп, waiter на транспорт, stream-сессию, метрики и текущий запрос разгрузки.
Увиденная удалённая опция мембера не создаёт локальную подписку, буфер или NCD-подключение.
Без локального экземпляра dispatcher отбрасывает DATA после минимального разбора ключа трансляции:
не проверяет подпись содержимого, не выполняет FEC, не буферизует и не пересылает блок.
INFO/SUBSCRIBE/REVERSE_REQ к неподписанному узлу получает явный отказ, а не создаёт неявную подписку.
Операции с экземплярами, NCD, очередями, Merkle и таймерами выполняются в uasync-потоке ядра.
Внешние producers передают байты через существующий механизм post с ограничением ожидающего объёма.
При нехватке ресурсов создание подписки завершается явной ошибкой с очисткой частично созданного состояния.
3. FEC, номера блоков и размер кодограммы
Источник формирует пачки и кодирует их lib/reed_solomon:
- n — полное число субпотоков, r — число проверочных, k = n-r — число исходных.
- По умолчанию n=8, r=2, k=6. Любые k кодограмм одной пачки восстанавливают её содержимое.
- subblock от 0 до n-1 определяет субпоток; первые k блоков систематические, остальные проверочные.
- batch_seq — общий 64-битный индекс пачки; блок субпотока X определяется парой (batch_seq, X).
Позиция N в запросе — batch_seq, а не байтовое смещение или отдельный счётчик субпотока.
- Номера возрастают без повторов; UINT64_MAX зарезервирован для запроса с конца.
Переполнение требует новой source_epoch.
Дескриптор содержит ключ трансляции, n, r и размер RS-кодограммы block_size.
Параметры задаёт только источник и не меняет в пределах source_epoch; изменение создаёт новый запуск.
Дескриптор подтверждается подписанной опцией источника; поля FEC также входят в подписанный блок.
Ограничения кодека: 1 <= n <= 255, 0 <= r < n, 10 <= block_size <= 4096.
Кроме допустимости RS проверяется расход памяти и корректность полного размера блока.
Целевой размер неизменяемого блока субпотока — около 1300 байт, включая метаданные источника и его 64-байтную подпись.
block_size для RS рассчитывается из этого бюджета; он включает 9-байтный заголовок RS seq/subblock.
Handle получателя, поколение stream-сессии, изменяемый путь и служебные заголовки ETCP идут сверх этого бюджета.
Не обещаем соответствия одной кодограммы одному UDP-пакету: необходимую фрагментацию выполняет ETCP normalizer.
На первой порции данных неполной пачки источник запускает таймер 100 мс.
При заполнении пачка немедленно кодируется, таймер отменяется; для следующей неполной пачки начинается новый срок.
По таймауту кодируем накопленные данные с дополнением нулями, сохраняя их действительную длину.
Без накопленных данных пустые пачки не создаём. При штатном окончании источник сразу сбрасывает остаток.
Этот таймер есть только у источника; ретрансляторы не накапливают данные для повторного формирования пачек.
В этапе 1 ретранслятор пересылает полученные блоки; отсутствующие блоки для downstream заново не генерирует.
Локальный decode восстанавливает содержимое для потребителя, но не создаёт подписи источника на отсутствующие блоки.
Восстанавливаем при наличии k проверенных блоков, не ожидая остальных.
Полученные байты выдаются как есть с удалением padding, по порядку batch_seq.
Невосстановимый пропуск обозначается событием GAP; нельзя молча склеивать данные по обе стороны разрыва или выдавать нули за исходные байты.
Окончательно пропускаем пачку, когда её данных уже нет в окне либо все нужные выдачи прошли её индекс/сообщили пропуск,
а k блоков не собрано. Получение более поздней пачки от одного поставщика само по себе не закрывает ожидание остальных.
Модуль не интерпретирует содержимое и не занимается видео, кодированием, воспроизведением или поиском ключевых кадров.
4. Подпись источника и timestamp
Каждый полученный после RS блок, в том числе проверочный, подписывается Ed25519-ключом самого источника.
Подписывается каноническое сообщение с маркером назначения подписи stream_block, включающее:
group_id, source_node_id, source_epoch, параметры FEC, batch_seq, subblock, source_ts_ms,
действительную длину содержимого пачки, флаги источника и все байты RS-кодограммы.
Byte order и точная раскладка фиксируются в stream_proto.h. Дублирующиеся поля seq/subblock проверяются на совпадение.
Это связывает блок с группой, запуском, индексом, временем и параметрами кодирования.
Подпись создаётся один раз в источнике после FEC. Она хранится с блоком и пересылается без изменений.
Handle, stream-сессия и hop-list находятся вне подписанного тела, поскольку меняются при выдаче и ретрансляции.
Перед помещением нового блока в кольцо, FEC или ретрансляцией проверяется подпись по ключу источника
из проверенной записи мембера данной группы. Ключ непосредственного поставщика не заменяет ключ источника.
Неверная подпись, неизвестный ключ или конфликт метаданных — отказ с диагностикой, без изменения индекса наличия.
Для одинакового уже проверенного блока допустимо повторно не выполнять crypto, только если совпадает всё неизменяемое тело.
Одинаковый индекс с другим содержимым нельзя считать обычным дубликатом без проверки.
source_ts_ms = (uint16_t)monotonic_time_ms источника в момент формирования пачки.
У всех её субпотоков одинаковые timestamp, длина и флаги. Ретранслятор не переписывает их своим временем.
Timestamp циклический с периодом 65536 мс; обычный переход через ноль обрабатывается модульной разностью.
Это не Unix-время; NTP и сравнение абсолютных часов разных узлов для метрики задержки не требуются.
16 бит не позволяют однозначно восстановить произвольное число полных оборотов.
Источник помечает в подписанных флагах разрыв временной шкалы, если интервал между его пачками >= 65536 мс.
При таком флаге, смене поставщика/эпохи, потере непрерывности индексов или длинном перерыве приёма
получатель начинает новую базу измерения, не придумывая число оборотов.
Первый блок устанавливает базу и не измеряет абсолютную задержку пути.
5. Кольцевой буфер и единственный способ выдачи
На экземпляр — 32 * 1024 * 1024 байт под неизменяемые блоки всех субпотоков вместе, включая их подписи.
Метаданные индекса, путей и сборки учитываются отдельно и ограничиваются.
Буфер общий для локального потребителя и всех downstream; копии кольца на каждого получателя не создаются.
Источник сначала помещает сформированные блоки в своё кольцо, затем делает их доступными выдачам.
Ретранслятор сначала проверяет и сохраняет блок, затем планирует выдачу подписчикам.
Индекс записи — (batch_seq, subblock) внутри конкретной эпохи.
Новые пачки вытесняют старейшие; при вытеснении пачки удаляются все её блоки и метаданные.
Дубликаты не занимают повторную память и не продлевают хранение.
Блоки старше нижней границы окна не вытесняют более свежие данные.
По каждому субпотоку известны доступные границы и наличие блоков внутри них; между границами возможны дырки.
Длительность истории определяется битрейтом и FEC; фиксированного времени хранения нет.
Есть только SUBSCRIBE(subblock, N):
- Конкретный N: начать непрерывную выдачу этого субпотока с указанного индекса, затем выдавать новые блоки.
- N=UINT64_MAX: начать с последнего доступного блока запрошенного субпотока, затем продолжать выдачу.
Нет REPAIR_REQ, выдачи диапазонов и отдельной системы повторного запроса потерянных блоков.
Выдача хранит курсор и читает только кольцо; запрос не закрепляет данные в памяти и не резервирует историю.
Если начальный блок уже вытеснен или недоступен, возвращаем явный отказ с границами окна.
Недоступную позицию не заменяем последней молча.
Если курсор работающей выдачи отстал за вытесненные данные, сообщаем EXPIRED/GAP для пропуска и продолжаем с доступного индекса.
Если внутри окна нет блока, но уже доступны более поздние, сообщаем GAP и продолжаем.
Восстановление получения выполняется выбором поставщика/новым SUBSCRIBE с нужного индекса, пока он ещё есть в его кольце.
Никакого отдельного архивирования, гарантии хранения до доставки или ожидания медленного получателя нет.
Общая очередь хранит ключ блока и идентификатор выдачи; долговременно указатели внутрь переиспользуемого кольца не сохраняются.
Перед передачей в ETCP наличие проверяется заново, пакет получает самостоятельное содержимое.
Вытесненное задание учитывается как пропуск, отменённое не отправляется.
После CLOSED не отправляются новые DATA/GAP по старому handle.
Уже принятые транспортом пакеты принадлежат ETCP и не ссылаются на память кольца.
6. Транспорт и NCD + reverse
Данные и управление установленной выдачей идут непосредственно между поставщиком и получателем через raw ETCP.
Для стриминга выделяются новые IDs в едином реестре src/transport_layer/etcp_api.h:
raw ID прямого обмена и router service ID управления через группу, без конфликтов в соответствующих пространствах.
ETCP обеспечивает переповторы, порядок, мультилинк и backpressure; второй механизм транспортных ACK поверх него не создаётся.
До отправки требуется UP от собственного NCD handle. Наличие ETCP_CONN само по себе недостаточно.
Согласование BGP-сессий остаётся в topo_group; stream приобретает своё NCD-владение и не дублирует этот протокол.
Членство проверяется отдельно; NCD такую проверку не выполняет.
На стороне отправителя до ACCEPT приобретается собственное NCD-владение на срок выдачи.
Прямое подключение запускается через node_conn_direct_open().
При необходимости инициатор отправляет REVERSE_REQ через маршрутизатор группы;
получатель проверяет членство, наличие своей подписки на эту трансляцию и открывает встречный NCD handle.
Запрос содержит ID попытки и ключ трансляции; адреса и ключ пира берутся из проверенного реестра.
Повтор одной попытки не создаёт дополнительные handles; сроки и повторы ограничены.
Завершение/отмена попытки убирает её таймеры и временное владение.
Совместимость учитывает оба узла, локальные сокеты, типы NAT, LAN/публичные IPv4/IPv6 и direct/reverse.
Используются существующие критерии sock_match/NCD; отбор не строится только по наличию EIM у одного пира.
Код адресного отбора, handshake, шифрования и линков в stream не дублируется.
Маршрутизатор группы обслуживает REVERSE_REQ и уведомления при отсутствии прямой связи,
а не большую INDIRECT-выдачу через conn_mgr.
7. Протокол и сроки выдач
Основные команды:
- INFO_REQ/INFO_RESP — дескриптор, доступные субпотоки, диапазоны кольца, пути, rates и метрики кандидата.
- SUBSCRIBE/ACCEPT/REJECT — непрерывная выдача с индекса либо последнего доступного блока.
- RENEW — продление списка существующих выдач, период 2 секунды.
- UNSUBSCRIBE/CLOSED — закрытие выдачи с указанием handle и причины.
- DATA/GAP/EXPIRED — подписанный блок либо уведомление о пропуске.
- OVERLOAD — запрос сокращения исходящей нагрузки через закрытие нескольких выдач.
- REVERSE_REQ — встречное NCD-подключение через маршрутизатор группы.
- END — последняя пачка и штатное завершение содержимого источником.
handle — ненулевой 64-битный ID, назначаемый получателем.
Предлагаемый простой механизм защиты от повторов: возрастающие handles в пределах stream-сессии пира,
новые SUBSCRIBE отправляются по порядку в одной управляющей FIFO.
Номер выделяется непосредственно при постановке нового SUBSCRIBE в эту FIFO, после готовности подключения;
параллельные ожидания NCD/INFO не резервируют номера, которые позже нарушили бы порядок отправки.
Ключ выдачи — peer, stream key, поколение stream-сессии и handle.
Повтор известного действующего SUBSCRIBE с прежними параметрами возвращает прежний результат,
не создаёт вторую выдачу, не меняет курсор и продлевает аренду.
Другие параметры по тому же handle отвергаются.
Для закрытого/неизвестного handle не выше уже обработанного номера SUBSCRIBE — STALE_REQUEST.
Водяной знак сохраняется до конца сессии; не нужен неограниченный список всех ранее закрытых handles.
RENEW создаёт только продление существующей выдачи. Повторное закрытие безопасно.
Для другой позиции создаётся новый handle; после переполнения счётчика требуется новое поколение сессии.
Поколения пиров согласуются при прямом INFO-обмене до SUBSCRIBE.
При DOWN, REINIT или закрытии старое поколение инвалидируется вместе с его выдачами.
Поздние ACCEPT/DATA/routed CLOSED не восстанавливают их. ETCP_CONN может пережить REINIT, его адрес не является ID сессии.
Управляющие поля и размеры списков имеют явный byte order и пределы.
ACCEPT содержит handle, фактический начальный индекс, границы кольца и путь субпотока.
Реальное получение блока, а не только ACCEPT, подтверждает пригодность выдачи.
Отказы различают отсутствие подписки/трансляции, эпоху, FEC, позицию, слот, перегрузку, цикл и право доступа.
DATA содержит поколение сессии, handle, неизменяемый подписанный блок источника и изменяемый путь.
Источник подписывает и END с ключом трансляции и индексом последней пачки, чтобы ретранслятор не мог подделать окончание.
Нет продления 6 секунд — отправитель закрывает выдачу и сообщает CLOSED, при необходимости через маршрутизатор группы.
Аренда, пробы и timers используют монотонное локальное время. Таймаут аренды очищает ресурсы;
реакция на отсутствие прогресса начинается раньше по метрикам потока.
Потеря отдельного ETCP-линка не закрывает выдачи, пока NCD остаётся UP.
DOWN закрывает выдачи соответствующей сессии; REINIT требует их согласовать заново.
Локальное освобождение ресурсов не ждёт доставки уведомления.
8. Начальная позиция и перенос субпотоков
Начальный запуск:
1) По опциям мемберов и BGP выбрать первый подходящий онлайн-узел, включая сам источник.
2) Запросить один субпоток с N=UINT64_MAX.
3) Получить и проверить первый блок, зафиксировать его source_epoch и batch_seq=N.
4) У остальных поставщиков запросить субпотоки этой же эпохи с конкретного N.
5) Собрать достаточный набор блоков одной пачки, затем продолжать непрерывное получение.
Не выбирать независимый последний индекс у каждого поставщика: пачки должны совпадать.
Если N успел исчезнуть из колец или набор собрать нельзя, явно начать новую попытку с конца одного поставщика.
Не смешивать старую и новую попытки и не выдавать их за непрерывное содержимое.
Обычный перенос одного субпотока:
1) Выбрать первую ещё нужную пачку N и поставщика, способного её выдать.
2) Создать новую выдачу, продолжая старую; временное перекрытие также занимает исходящий слот нового поставщика.
3) Получить блоки и подтвердить продвижение нового поставщика.
4) Закрыть старую выдачу. При отказе новой сохранить старую и отменить неудачную попытку.
Дедупликация — по (stream key, batch_seq, subblock), независимо от handles.
При перекрытии качество поставщиков учитывается отдельно, но блок потребителю выдаётся один раз.
При явном отказе разрешено одновременно восстанавливать несколько субпотоков.
Точная задержка обычного переноса и гистерезис задаются настройками; небольшое колебание RTT не запускает перестройку.
9. Пути и защита от циклов
У каждого субпотока свой stream-путь от источника до поставщика; BGP-путь его не заменяет.
INFO_RESP и ACCEPT содержат путь запрошенного субпотока.
Путь начинается источником, не содержит нулей и повторов; подписка запрещена, если получатель уже находится в нём.
Эта проверка повторяется при принятии запроса, а не только при предварительном выборе.
DATA несёт фактически пройденные узлы; начальный путь содержит источник, последний — непосредственного проверенного отправителя.
При пересылке добавляется свой node_id. Повторное посещение и превышение максимума хопов запрещены.
При выдаче из кольца сохраняется история пути блока; запрещено отправлять его уже пройденному узлу.
В связи с этим отсутствие пригодного блока для конкретного получателя также обозначается пропуском.
Смена upstream обновляет объявляемый путь и повторно проверяет downstream-выдачи; возникший цикл закрывается с причиной LOOP.
При гонках метрик временно возможна неработающая цепочка, но DATA не циркулирует, а отсутствие прогресса запускает поиск.
Старые данные в кольце сами по себе не делают узел качественным live-поставщиком.
10. Относительная задержка, остановка и rate
Метрики ведутся отдельно для каждого поставщика и субпотока.
Обрабатываются новые индексы, а не дубликаты; timestamp всех субпотоков одной пачки одинаковый.
Для последовательных блоков:
source_delta_ms = (uint16_t)(source_ts_current - source_ts_previous)
receive_delta_ms = local_receive_current_ms - local_receive_previous_ms
lag_ms = max(0, lag_ms + receive_delta_ms - source_delta_ms)
Первый блок устанавливает базу. Обычный оборот timestamp обрабатывается без скачка lag.
Так измеряется накопленное отставание относительно скорости источника, а не абсолютное время от источника до получателя.
Ускоренный приход уменьшает lag, но значение не становится отрицательным.
Источник timestamp не изменяется по пути, поэтому изменение доставки учитывает задержки на всей цепочке.
Базу сбрасываем при отмеченном источником разрыве времени, неоднозначной дельте, новой выдаче/эпохе или разрыве последовательности.
Резкое изменение позиции выдачи не считается доказательством улучшения задержки.
Ожидаемый межблочный интервал и rate источника оцениваются по окну его timestamp и объёмам блоков.
Если несколько блоков имеют один timestamp, агрегируем объём до ненулевой дельты, не делим на ноль.
Не используем время формирования всех n блоков одной пачки как n независимых временных интервалов.
При отсутствии новых данных продолжаем измерять отставание от ожидаемого следующего прихода:
stall_lag_ms = lag_ms + max(0, now_ms - last_receive_ms - expected_interval_ms)
Превышение 500 мс является проблемой доставки; полная остановка не остаётся незамеченной только потому, что новых timestamp нет.
Отсутствие данных источника и остановка на маршруте по одной этой метрике неразличимы:
фиксируем отсутствие прогресса, не приписывая ему недоказанную причину. END прекращает ожидание новых блоков.
История проблем — 60 секундных корзин: число запаздывающих блоков и отдельные эпизоды остановки.
Блок считается запоздавшим один раз для соответствующей выдачи; один эпизод остановки не считается заново на каждом тике.
Сохраняются число наблюдаемых блоков, rates и доля проблем, чтобы сравнение не искажалось разным битрейтом.
FEC-восстановление не считается доставкой отсутствующего субпотока; частичные пачки и пропуски считаются отдельно.
Скорость продвижения по старой истории при запуске/переносе отличается от steady live и явно отмечается.
INFO_RESP также сообщает число свободных исходящих слотов, диапазоны буфера, актуальность прогресса, очередь и нагрузку.
Суммарный RTT = RTT непосредственного соединения + объявленный RTT stream-пути кандидата, отдельно по субпотоку.
Поле суммарного RTT — не менее 32 бит с явными единицами и признаком неизвестного/устаревшего значения.
11. Поиск поставщиков и расписание проб
Кандидаты — онлайн в BGP данной группы, с подпиской на нужную трансляцию/эпоху в опциях мембера.
Начальный поиск учитывает источник. Фоновые пробы ретрансляторов исключают мобильных.
Мобильный узел может быть источником и получателем; выбранный мобильный источник допускается к подключению.
Активное получение/публикация согласовывается с активностью ядра, чтобы standby не прерывал поток.
При запуске экземпляра одновременно пробуем до 32 совместимых по типу подключения кандидатов.
Если подходящих меньше, пробуем имеющихся. Это узлы, а не 32 отдельных адреса или субпотока.
Совместимость определяется существующими транспортными правилами; известный рабочий транспорт переиспользуется.
После стартовой серии запускаем не более одной новой пробы в секунду.
Один кандидат пробуется один раз за цикл; новый полный цикл начинается не раньше 30 секунд от старта предыдущего.
Если кандидатов мало и цикл исчерпан раньше, до этого срока ждём. Если их много, цикл занимает фактическое время обхода.
Стартовые пробы учитываются в первом цикле, а не повторяются сразу после него.
Новые кандидаты из Merkle/BGP могут добавляться в текущий цикл с темпом 1/с, без повторного опроса уже проверенных.
Неуспех имеет backoff; одинаковые активные пробы одного пира для разных экземпляров переиспользуются.
Временные NCD handles невыбранных кандидатов освобождаются; рабочие экземпляры удерживают свои handles.
Используются route_connectivity, sock_match/NCD и метрики существующих соединений; второй адресный probe-движок не создаётся.
Прямая доступность — UNKNOWN/REACHABLE/UNREACHABLE с временем замера, а не вечный boolean.
Порядок выбора: доступный субпоток/индекс и отсутствие цикла; затем проблемы доставки и RTT;
далее устойчивость набора поставщиков, совпадение промежуточных узлов и число хопов.
Стремимся получить n субпотоков, хотя k достаточно для decode.
При возможности не более r субпотоков от одного поставщика: потеря его при получении всех n оставляет не менее k.
При r=0 такой защиты нет; при недостатке кандидатов ограничение ослабляется с отражением устойчивости в статусе.
Учитываем совпадение путей разных поставщиков, кроме общего источника.
12. Общая отправка и перегрузка
На удалённого пира — одна ограниченная очередь стриминга для всех групп и источников.
Управление имеет небольшой приоритет; выдачи данных чередуются, чтобы один источник не захватил всю очередь.
При старте с прошлого индекса не помещаем сразу всё кольцо в очередь: читаем его по курсору по мере готовности отправки.
Темп задают имеющиеся данные и backpressure ETCP, независимый транспортный shaper не создаётся.
Отправка ждёт освобождения conn->send_input_q через queue_waiter_wait с существующим порогом.
Не переполняем очередь, не меняем чужой общий порог и не заменяем waiter активным опросом.
Handle waiter обнуляется полностью, callback может быть синхронным; отменяем waiter до освобождения владельца.
При автозаборе ll_queue соблюдается queue_data_get -> обработка -> queue_resume_callback.
При успехе etcp_send пакет принадлежит транспорту; при ошибке его освобождает stream.
Принятие стеком не означает доставку; для диагностики учитываются нижние очереди/inflight и продвижение приёмника.
Уже принятые ETCP пакеты не извлекаются обратно; общий транспорт ради одного потока не сбрасывается.
Раз в 200 мс сохраняется интервал, история — 10 секунд.
Длительность интервала фактическая, включая задержки uasync.
По каждому пиру и выдаче считаем пакеты/байты:
A — поступившая нагрузка, D — передано в ETCP, X — отменено/вытеснено, Q — ожидающий объём;
баланс Q_next = Q + A - D - X, возраст старейшего задания, rate и скорость роста.
Для заданий-ключей байтовый объём — размер ожидающего блока, память самих заданий учитывается отдельно.
Также сохраняются отказы при постановке, чтобы ограничение очереди не скрывало реальный избыток нагрузки.
Подтверждение перегрузки — непрерывные 500 мс наблюдаемого отставания, по монотонному времени, не по округлённому числу тиков.
Возраст данных >200 мс вместе с устойчивым ростом — сигнал недостаточной скорости.
Полная остановка D также обрабатывается. Краткий глитч с последующей быстрой разгрузкой не вызывает немедленного отключения.
Нагрузка uasync берётся из uasync_get_load(); дополнительно измеряется опоздание callbacks.
Проблема одного пира отличается от одновременного роста многих очередей и локального отставания event loop.
Высокий CPU отдельно от отставания/очередей не является достаточным основанием.
Скорость роста g оценивается в байтах/с по увеличению Q до искусственных удалений:
в интервале без смены выдач g = max(0, (A-D)/elapsed_seconds).
Rate выдачи — её средний вклад в поступающую нагрузку, а не урезанная скорость выхода из перегруженного канала.
Для разгрузки нужно снять суммарный rate больше g, чтобы очередь начала убывать.
Число выдач выбирается по их rates: закрываем достаточно выдач для требуемого сокращения;
при сильном росте это сразу несколько, не фиксированная одна и не случайное постоянное число.
OVERLOAD содержит request_id, расчётное число выдач и требуемое сокращение rate.
Получатель выбирает handles с учётом FEC и альтернативных поставщиков.
При разных rates одного количества закрытий недостаточно: суммарный снятый rate должен покрыть требование.
Rate закрываемых выдач фиксируется по оценке на момент запроса, чтобы условие исполнения не менялось на каждом тике.
Срок реакции — 2 секунды с первого запроса; повторы не сбрасывают срок и не начинают новое требование.
Новые выдачи, компенсирующие запрошенное сокращение, до разгрузки не принимаются.
Если требование не выполнено, отправитель сам закрывает достаточный остаток выдач и сообщает handles/причину.
Далее снова измеряется рост; автоматический выбор учитывает освобождаемую нагрузку, а при равенстве допустима случайность.
При глобальном отставании сокращение выбирается по всем пирам модуля.
Память и возраст очереди остаются ограниченными во время ожидания; просроченные задания удаляются с GAP/EXPIRED.
13. Статусы через опции мембера и Merkle
Статус источника и список подписок принадлежат самому мемберу в конкретной группе.
Они входят в его подписанный блок A/userinfo и распространяются существующим member_sync/merkle_sync.
adm_tags — блок владельца, для самоуправляемых stream-опций он не используется.
Отдельного STREAM_ANNOUNCE, broadcast-обновлений и собственных TTL/версий анонсов нет.
Опции содержат:
- stream_source: активна ли собственная трансляция; если активна — её source_epoch и параметры FEC.
- stream_subscriptions: список (source_node_id, source_epoch), на которые узел локально подписан.
Формат — совместимые с существующим плоским userinfo строковые значения; список кодируется однозначно,
со строгим разбором 64-битных IDs без потери точности через JSON numbers.
Источник включает свою трансляцию в собственные подписки.
Список относится к трансляциям, а не к сетевым handles или выбранным upstream-поставщикам.
Изменение подписки/запуск/остановка источника обновляет свой блок мембера:
userinfo + возрастающий update_ts + update_sig, атомарно с Merkle-хешами, после commit запускается распространение.
Другие опции (имя, аватар и т.д.) сохраняются; запись одного модуля не должна стирать состояние другого.
Размер укладывается в MS_USERINFO_MAX; переполнение — явная ошибка подписки, а не молчаливое обрезание.
При запуске ядра свои stream-опции приводятся к реально восстановленному локальному состоянию,
чтобы старый источник/подписка после аварии не объявлялись работающими без экземпляра.
Подписка удалённого мембера актуальна только если есть его живой BGP-путь в этой же группе.
Недостаточно присутствия записи в SQLite/реестре или произвольного ETCP-соединения из другой группы.
При исчезновении BGP-пути узел исключается из кандидатов; его подписанная опция в Merkle не удаляется третьей стороной.
При возвращении в BGP опции читаются заново и текущая доступность подтверждается INFO.
Статус источника также проверяется совместно с его доступностью; наличие старого source_epoch в БД не доказывает поступление данных.
Stream следит за двумя независимыми событиями: изменением блока мембера и изменением BGP-доступности.
Текущий props callback передаёт adm_tags; stream-опции читаются из принятого userinfo после commit,
а не трактуются как поля adm_tags.
Merkle уже обеспечивает версии, подписи и обмен для поздних участников; второй механизм хранения объявлений не создаётся.
Rates, RTT и очереди запрашиваются прямо у кандидата, а не записываются в Merkle каждые 200 мс.
Входящий SUBSCRIBE проходит авторизацию по членству; задержка распространения его опции не должна создавать взаимное ожидание.
Сам получатель обязан иметь локальную подписку до запросов, а выдающий узел — свою до обработки потока.
14. Жизненный цикл и диагностика
При остановке последнего локального владельца прекращаем producers/поиск, снимаем опцию подписки,
отменяем callbacks/таймеры/waiters, закрываем сетевые выдачи, убираем задания и освобождаем свои NCD handles.
Кольцо и FEC-контекст освобождаются после прекращения обращений.
Работа других трансляций/модулей на общем ETCP сохраняется.
Удаление группы, исключение мембера и смена source_epoch отменяют относящиеся к ним выдачи.
Stop/destroy выполняются по правилам ядра, вне callbacks останавливаемого сервиса.
Категория stream с конфигурируемым уровнем:
- INFO: запуск/остановка с FEC/размером/лимитами, выдачи, перенос, подтверждённая перегрузка и её результат, итоговые счётчики.
- DEBUG: выбор кандидата с метриками/причиной, окно кольца, timestamp-дельты/lag/rates, сводки очередей и циклов проб.
- TRACE: отдельные блоки, протокольные команды и важные ветви при включённой подробной диагностике.
- WARN/ERROR: отказные и нештатные ветки с контекстом, включая подпись, эпоху, LOOP, EXPIRED, память, аренду и транспорт.
Указываются группа/источник/эпоха, peer, handle/subblock, индекс, состояние, причина и значимые величины.
При перегрузке — A/D/X/Q, возраст, длительность подтверждения, g, rates выбранных выдач и требуемое сокращение.
При timestamp-разрыве — предыдущий/новый индекс, timestamp и причина сброса базы.
Ключи и содержимое данных не выводятся; частые события агрегируются с полными счётчиками.
Не выдавать принятие в очередь за доставку, относительный lag за абсолютную задержку, старую опцию за онлайн-подписку.
15. Реализация и проверки
Порядок:
1) Подписки/экземпляры, member-опции и BGP-фильтр, дескриптор, FEC/подпись/timestamp, кольцо.
2) Raw-протокол, NCD + reverse, выдачи с индекса/конца, аренда и защита от повторов.
3) Общая очередь/backpressure, относительные метрики, rates и разгрузка.
4) Стартовые 32 пробы/последующий цикл, выбор путей, защита от циклов и перенос.
5) API непрозрачных данных и потребителя, без интеграции видеокодека в этом этапе.
Проверки:
- Несколько подписок/источников/групп на одном ETCP; stop одной сохраняет остальные и чужое владение.
- Неподписанный узел не создаёт экземпляр, не выполняет crypto/FEC для DATA и не ретранслирует его.
- Источник задаёт различные FEC; decode из k проверенных блоков и невозможность при k-1.
- Подпись каждого исходного/проверочного блока; подмена payload/timestamp/индекса/FEC/эпохи/группы отвергается.
- Кодограммы около 1300 байт, полный/неполный буфер, источник сбрасывает за 100 мс, ретранслятор не имеет такого таймера.
- Оборот timestamp, нулевая дельта, постепенное отставание, догон, полная остановка и длинный разрыв без ложного числа оборотов.
- Переполнение 32-МБ кольца, дырки, EXPIRED до отправки, курсор медленного получателя и ограниченность памяти.
- Первый подписанный блок с конца задаёт общий N; остальные выдачи начинают с него, а не с независимых концов.
- Идемпотентные запросы/продления/закрытия, водяной знак handles, поздние сообщения, REINIT и новая эпоха источника.
- Reverse/неуспех/отмена; потеря одного линка сохраняет выдачи, DOWN всего NCD запускает восстановление.
- Перекрытие поставщиков без дублирования потребителю, rollback, одновременное восстановление нескольких субпотоков.
- Цикл по устаревшим метрикам, одновременная перестройка и выдача сохранённых путей: DATA не циркулирует.
- Глитч короче подтверждения, устойчивое отставание 500 мс, нулевой выход и локальная перегрузка uasync.
- Выбор числа закрытий по g/rates, разные rates, повтор OVERLOAD, частичное выполнение и закрытие остатка через 2 секунды.
- 32 совместимых кандидата на старте, далее 1/с, маленький/большой список и запрет повторного полного цикла раньше 30 с.
- Merkle-опции до/после BGP-события, офлайн-опция не кандидат, поздний участник, перезапуск и сохранение других полей userinfo.
- Мобильный источник/получатель и исключение мобильных из фоновых проб ретрансляторов.
Перед кодированием фиксируются точная раскладка stream_proto.h, строковая схема member-опций и численные
лимиты собственной очереди/служебной памяти. Это детализация реализации принятых решений, а не повторный выбор транспорта,
подписи, часов, формата содержимого, анонсов, отдельного дозапроса или расписания проб.
Гистерезис обычного переноса и небольшой запас rate сверх g подбираются на стенде с сохранением заданных сроков.
Исходники добавляются в Autotools и применимые desktop/Android CMake-сборки общего ядра.
Тесты добавляются через tests/Makefile.am. Диагностика должна однозначно показывать выбор путей,
пропуски, сбросы timestamp-базы, переключения и фактический результат разгрузки.
16. План работ и текущий статус реализации
Реализация начата 10.10.2026. Работа разбита на проверяемые этапы:
[+] Основа данных в src/stream:
stream_proto.h — канонический big-endian формат подписанного блока; дефолтный полный блок 1300 байт.
stream_block — RS и Ed25519-подпись каждого блока; проверка ключа трансляции, параметров и содержимого.
stream_buffer — общий бюджет 32 МБ по фактически сохранённым блокам, индекс пачек и вытеснение старейшей пачки.
stream_metrics — относительное отставание, rate, timestamp wrap/reset и минутная история проблем/остановок.
stream_source — приём непрозрачных байтов, немедленная выдача полной пачки, таймер остатка 100 мс и отмена при destroy.
Добавлены категория stream, Autotools-сборка и test_stream_core.
Desktop/Android CMake уже используют GLOB_RECURSE для src/*.c и автоматически включают новые компоненты.
Проверки: подписи и подмены полей, все 28 пар потерь при 6+2, другие FEC-параметры, padding,
дырки/переполнение кольца, timestamp через ноль/длинный перерыв, остановка и отмена таймеров, учёт утечек.
./build.sh --no-gui -j4 и ./check.sh прошли: 143 passed, 0 failed, 0 skipped;
отдельные proxy/full, burst и load также PASS. Лог: /tmp/stream-check.log.
[ ] Включение в жизненный цикл чат-сервиса:
контекст модуля в UTUN_INSTANCE, публичные локальные подписки/публикация, события потребителя,
stream_source/stream_subscriptions в блоке мембера, сохранение остальных userinfo и BGP-фильтр кандидатов.
[ ] Сетевая выдача:
raw/router IDs, INFO и поколения сессий, 64-битные handles, SUBSCRIBE/RENEW/CLOSED,
собственное владение NCD, REVERSE_REQ, курсоры кольца и общий начальный индекс от первого поставщика.
[ ] Планировщик отправки и разгрузка:
одна очередь на пира, backpressure, ограничение памяти/возраста, A/D/X/Q,
подтверждение 500 мс, сокращение по g/rates и двухсекундный срок исполнения OVERLOAD.
[ ] Оптимизация сети и интеграционные проверки:
пути/защита от циклов, стартовые 32 пробы, дальнейший темп 1/с и полный цикл не чаще 30 с,
постепенный перенос с перекрытием, failover, несколько источников и независимое закрытие на общем ETCP.
Текущие компоненты — основа модуля данных; сетевые bindings, Merkle-опции и вызовы из UTUN_INSTANCE ещё не подключены.

8
doc/tasks.md

@ -4,6 +4,14 @@
## Текущие задачи ## Текущие задачи
[ ] **Стриминг: реализация по stream.txt** — 10.10.2026:
ТЗ и этапы: doc/stream.txt, разделы 15–16. Основа данных реализована в src/stream:
подписанные FEC-блоки 1300 байт, общий буфер 32 МБ, относительные timestamp-метрики,
источник с таймером 100 мс и категория stream. test_stream_core прошёл;
./check.sh: 143 passed, 0 failed, 0 skipped; proxy/full, burst и load — PASS.
Следующий этап — жизненный цикл CHAT и опции мембера через Merkle с фильтром BGP,
затем NCD-выдачи/reverse, общий отправитель/перегрузка и оптимизация путей.
[+] **test_reality_tls: коллизии порта слушателя** — 10.10.2026: [+] **test_reality_tls: коллизии порта слушателя** — 10.10.2026:
В общем прогоне 140 passed, 1 failed: stcp_server_create не смог занять порт 34922, bind errno=98 (EADDRINUSE). В общем прогоне 140 passed, 1 failed: stcp_server_create не смог занять порт 34922, bind errno=98 (EADDRINUSE).
Причина: выбор порта как 34000 + time(NULL) % 1000 без гарантии его доступности. Причина: выбор порта как 34000 + time(NULL) % 1000 без гарантии его доступности.

1
lib/debug_config.c

@ -136,6 +136,7 @@ static const struct {
{"radio", DEBUG_CATEGORY_RADIO}, {"radio", DEBUG_CATEGORY_RADIO},
{"vad", DEBUG_CATEGORY_VAD}, {"vad", DEBUG_CATEGORY_VAD},
{"aec", DEBUG_CATEGORY_AEC}, {"aec", DEBUG_CATEGORY_AEC},
{"stream", DEBUG_CATEGORY_STREAM},
{"all", DEBUG_CATEGORY_ALL}, {"all", DEBUG_CATEGORY_ALL},
{NULL, DEBUG_CATEGORY_NONE} {NULL, DEBUG_CATEGORY_NONE}
}; };

3
lib/debug_config.h

@ -74,7 +74,8 @@ typedef int debug_category_t;
#define DEBUG_CATEGORY_RADIO 33 // Radio — PTT walkie-talkie multicast (voice) #define DEBUG_CATEGORY_RADIO 33 // Radio — PTT walkie-talkie multicast (voice)
#define DEBUG_CATEGORY_VAD 34 // VAD — Silero voice activity detector (ONNX Runtime) #define DEBUG_CATEGORY_VAD 34 // VAD — Silero voice activity detector (ONNX Runtime)
#define DEBUG_CATEGORY_AEC 35 // AEC — acoustic echo cancellation (SpeexDSP) #define DEBUG_CATEGORY_AEC 35 // AEC — acoustic echo cancellation (SpeexDSP)
#define DEBUG_CATEGORY_COUNT 36 // Total number of categories #define DEBUG_CATEGORY_STREAM 36 // Подписанные потоки, FEC, буферы и выбор поставщиков
#define DEBUG_CATEGORY_COUNT 37 // Total number of categories
#define DEBUG_CATEGORY_ALL (-1) // special value for all categories #define DEBUG_CATEGORY_ALL (-1) // special value for all categories
/* Debug configuration structure */ /* Debug configuration structure */

13
src/Makefile.am

@ -3,8 +3,20 @@ noinst_LIBRARIES = libutun.a
EXTRA_DIST = lwip_tcp/COPYING dnsmasq/COPYING dnsmasq/COPYING-v3 EXTRA_DIST = lwip_tcp/COPYING dnsmasq/COPYING dnsmasq/COPYING-v3
stream_core_files = \
stream/stream_proto.h \
stream/stream_block.h \
stream/stream_block.c \
stream/stream_buffer.h \
stream/stream_buffer.c \
stream/stream_metrics.h \
stream/stream_metrics.c \
stream/stream_source.h \
stream/stream_source.c
# Sources that are always compiled # Sources that are always compiled
utun_CORE_SOURCES = \ utun_CORE_SOURCES = \
$(stream_core_files) \
utun.c \ utun.c \
utun_instance.c \ utun_instance.c \
config_parser.c \ config_parser.c \
@ -152,6 +164,7 @@ utun_CORE_SOURCES = \
# libutun: all core sources except main() # libutun: all core sources except main()
libutun_a_SOURCES = \ libutun_a_SOURCES = \
$(stream_core_files) \
utun_instance.c \ utun_instance.c \
config_parser.c \ config_parser.c \
config_updater.c \ config_updater.c \

180
src/stream/stream_block.c

@ -0,0 +1,180 @@
#include <string.h>
#include <openssl/crypto.h>
#include <openssl/evp.h>
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "stream_block.h"
struct stream_encoder {
struct stream_descriptor descriptor;
struct rs_codec *rs;
uint8_t private_key[SC_PRIVKEY_SIZE];
uint8_t public_key[SC_PUBKEY_SIZE];
uint8_t *input;
uint8_t *wire;
};
/* Проверить ключ трансляции и параметры до расчёта размеров/выделения памяти. */
int stream_descriptor_validate(const struct stream_descriptor *d) {
if (!d || !d->group_id || !d->source_node_id || !d->source_epoch || !d->n || d->r >= d->n
|| d->block_size <= RS_HEADER_SIZE || d->block_size > RS_MAX_BLOCK_SIZE) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "descriptor invalid: present=%d group=%llx source=%llx epoch=%llx n=%u r=%u size=%u",
d != NULL, d ? (unsigned long long)d->group_id : 0, d ? (unsigned long long)d->source_node_id : 0,
d ? (unsigned long long)d->source_epoch : 0, d ? d->n : 0, d ? d->r : 0, d ? d->block_size : 0);
return STREAM_ERR_ARGUMENT;
}
return STREAM_OK;
}
size_t stream_block_size(const struct stream_descriptor *d) {
return stream_descriptor_validate(d) == STREAM_OK ? d->block_size + STREAM_BLOCK_OVERHEAD : 0;
}
size_t stream_input_size(const struct stream_descriptor *d) {
return stream_descriptor_validate(d) == STREAM_OK ? (size_t)(d->n - d->r) * (d->block_size - RS_HEADER_SIZE) : 0;
}
/* Создать рабочую память RS и получить публичный ключ именно переданного Ed25519 seed. */
int stream_encoder_create(struct stream_encoder **out, const struct stream_descriptor *d, const uint8_t private_key[32]) {
if (!out || *out || !private_key || stream_descriptor_validate(d) != STREAM_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encoder create invalid arguments out=%p key=%p", (void *)out, (const void *)private_key);
return STREAM_ERR_ARGUMENT;
}
struct stream_encoder *encoder = u_calloc(1, sizeof(*encoder));
if (!encoder) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encoder allocation failed source=%llx", (unsigned long long)d->source_node_id);
return STREAM_ERR_MEMORY;
}
encoder->descriptor = *d;
memcpy(encoder->private_key, private_key, sizeof(encoder->private_key));
EVP_PKEY *key = EVP_PKEY_new_raw_private_key(EVP_PKEY_ED25519, NULL, private_key, SC_PRIVKEY_SIZE);
size_t public_length = SC_PUBKEY_SIZE;
int key_ok = key && EVP_PKEY_get_raw_public_key(key, encoder->public_key, &public_length) == 1 && public_length == SC_PUBKEY_SIZE;
EVP_PKEY_free(key);
if (!key_ok) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encoder Ed25519 key initialization failed source=%llx",
(unsigned long long)d->source_node_id);
stream_encoder_destroy(encoder);
return STREAM_ERR_CRYPTO;
}
if (rs_init(&encoder->rs, d->n, d->r, d->block_size) != RS_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encoder RS initialization failed n=%u r=%u size=%u", d->n, d->r, d->block_size);
stream_encoder_destroy(encoder);
return STREAM_ERR_FEC;
}
encoder->input = u_malloc((uint32_t)stream_input_size(d));
encoder->wire = u_malloc((uint32_t)(d->n * stream_block_size(d)));
if (!encoder->input || !encoder->wire) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encoder working memory allocation failed n=%u size=%u", d->n, d->block_size);
stream_encoder_destroy(encoder);
return STREAM_ERR_MEMORY;
}
*out = encoder;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "encoder created group=%llx source=%llx epoch=%llx FEC=%u+%u block=%zu input=%zu",
(unsigned long long)d->group_id, (unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch,
d->n - d->r, d->r, stream_block_size(d), stream_input_size(d));
return STREAM_OK;
}
/* Очистить секретный seed и рабочее содержимое до возврата памяти allocator. */
void stream_encoder_destroy(struct stream_encoder *encoder) {
if (!encoder) return;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "encoder destroy source=%llx epoch=%llx",
(unsigned long long)encoder->descriptor.source_node_id, (unsigned long long)encoder->descriptor.source_epoch);
OPENSSL_cleanse(encoder->private_key, sizeof(encoder->private_key));
if (encoder->input) OPENSSL_cleanse(encoder->input, stream_input_size(&encoder->descriptor));
u_free(encoder->input);
u_free(encoder->wire);
rs_free(encoder->rs);
u_free(encoder);
}
const uint8_t *stream_encoder_public_key(const struct stream_encoder *encoder) {
if (!encoder) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encoder public key: null context"); return NULL; }
return encoder->public_key;
}
/* Сформировать канонический заголовок, RS-кодограммы и подпись каждого блока. */
int stream_encoder_encode(struct stream_encoder *encoder, uint64_t seq, uint16_t timestamp_ms, uint8_t flags,
const uint8_t *input, size_t length, const uint8_t **blocks) {
if (!encoder || !input || !blocks || !length || seq == UINT64_MAX || (flags & ~STREAM_BLOCK_TIME_RESET)
|| length > stream_input_size(&encoder->descriptor)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encode invalid arguments seq=%llu len=%zu flags=%u", (unsigned long long)seq, length, flags);
return STREAM_ERR_ARGUMENT;
}
const struct stream_descriptor *d = &encoder->descriptor;
size_t wire_size = stream_block_size(d);
uint8_t *rs_blocks[RS_MAX_BLOCKS];
memcpy(encoder->input, input, length);
memset(encoder->input + length, 0, stream_input_size(d) - length);
for (unsigned i = 0; i < d->n; i++) {
uint8_t *wire = encoder->wire + i * wire_size;
memcpy(wire, STREAM_BLOCK_DOMAIN, STREAM_BLOCK_DOMAIN_SIZE);
stream_write_be(wire + STREAM_BLOCK_GROUP_OFF, d->group_id, 8);
stream_write_be(wire + STREAM_BLOCK_SOURCE_OFF, d->source_node_id, 8);
stream_write_be(wire + STREAM_BLOCK_EPOCH_OFF, d->source_epoch, 8);
wire[STREAM_BLOCK_N_OFF] = d->n;
wire[STREAM_BLOCK_R_OFF] = d->r;
stream_write_be(wire + STREAM_BLOCK_SIZE_OFF, d->block_size, 2);
stream_write_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, timestamp_ms, 2);
stream_write_be(wire + STREAM_BLOCK_LENGTH_OFF, length, 4);
wire[STREAM_BLOCK_FLAGS_OFF] = flags;
rs_blocks[i] = wire + STREAM_BLOCK_RS_OFF;
}
if (rs_encode(encoder->rs, seq, encoder->input, rs_blocks) != RS_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encode RS failed seq=%llu", (unsigned long long)seq);
return STREAM_ERR_FEC;
}
for (unsigned i = 0; i < d->n; i++) {
uint8_t *wire = encoder->wire + i * wire_size;
if (sc_ed25519_sign(encoder->private_key, wire, wire_size - SC_SIGN_SIZE, wire + wire_size - SC_SIGN_SIZE) != SC_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "encode signature failed seq=%llu sub=%u", (unsigned long long)seq, i);
return STREAM_ERR_CRYPTO;
}
}
for (unsigned i = 0; i < d->n; i++) blocks[i] = encoder->wire + i * wire_size;
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "encoded seq=%llu timestamp=%u len=%zu n=%u flags=%u",
(unsigned long long)seq, timestamp_ms, length, d->n, flags);
return STREAM_OK;
}
/* Проверить размеры, область подписи, дескриптор, метаданные и подпись до публикации view. */
int stream_block_verify(const struct stream_descriptor *d, const uint8_t public_key[32], const uint8_t *wire,
size_t length, struct stream_block_view *view) {
if (!public_key || !wire || !view || stream_descriptor_validate(d) != STREAM_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "verify invalid arguments length=%zu", length);
return STREAM_ERR_ARGUMENT;
}
if (length != stream_block_size(d) || memcmp(wire, STREAM_BLOCK_DOMAIN, STREAM_BLOCK_DOMAIN_SIZE)
|| stream_read_be(wire + STREAM_BLOCK_GROUP_OFF, 8) != d->group_id
|| stream_read_be(wire + STREAM_BLOCK_SOURCE_OFF, 8) != d->source_node_id
|| stream_read_be(wire + STREAM_BLOCK_EPOCH_OFF, 8) != d->source_epoch
|| wire[STREAM_BLOCK_N_OFF] != d->n || wire[STREAM_BLOCK_R_OFF] != d->r
|| stream_read_be(wire + STREAM_BLOCK_SIZE_OFF, 2) != d->block_size) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "verify format/descriptor mismatch source=%llx epoch=%llx length=%zu expected=%zu",
(unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch, length, stream_block_size(d));
return STREAM_ERR_FORMAT;
}
struct stream_block_view parsed = {
.seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8),
.data_length = (uint32_t)stream_read_be(wire + STREAM_BLOCK_LENGTH_OFF, 4),
.timestamp_ms = (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2),
.subblock = wire[STREAM_BLOCK_SUB_OFF], .flags = wire[STREAM_BLOCK_FLAGS_OFF], .rs = wire + STREAM_BLOCK_RS_OFF
};
if (parsed.seq == UINT64_MAX || parsed.subblock >= d->n || !parsed.data_length
|| parsed.data_length > stream_input_size(d) || (parsed.flags & ~STREAM_BLOCK_TIME_RESET)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "verify invalid metadata seq=%llu sub=%u length=%u flags=%u",
(unsigned long long)parsed.seq, parsed.subblock, parsed.data_length, parsed.flags);
return STREAM_ERR_FORMAT;
}
if (sc_ed25519_verify(public_key, wire, length - SC_SIGN_SIZE, wire + length - SC_SIGN_SIZE) != SC_OK) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "verify source signature failed source=%llx epoch=%llx seq=%llu sub=%u",
(unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch,
(unsigned long long)parsed.seq, parsed.subblock);
return STREAM_ERR_SIGNATURE;
}
*view = parsed;
return STREAM_OK;
}

48
src/stream/stream_block.h

@ -0,0 +1,48 @@
/* Формирование и проверка подписанных FEC-блоков без сокетов и внешних callbacks.
* Контексты одного uasync-потока; descriptor и ключ источника уже проверены членством группы.
* encode возвращает заимствованные блоки до следующего encode/destroy; ошибочные результаты не публикуют.
* verify не меняет выходной view при отказе. Ключ непосредственного ретранслятора здесь не используется. */
#ifndef STREAM_BLOCK_H
#define STREAM_BLOCK_H
#include "stream_proto.h"
#ifdef __cplusplus
extern "C" {
#endif
struct stream_encoder;
struct stream_block_view {
uint64_t seq;
uint32_t data_length;
uint16_t timestamp_ms;
uint8_t subblock;
uint8_t flags;
const uint8_t *rs; /* Заимствованный диапазон descriptor.block_size байт. */
};
int stream_descriptor_validate(const struct stream_descriptor *descriptor);
size_t stream_block_size(const struct stream_descriptor *descriptor);
size_t stream_input_size(const struct stream_descriptor *descriptor);
/* out заранее NULL; ключ копируется и очищается при destroy. */
int stream_encoder_create(struct stream_encoder **out, const struct stream_descriptor *descriptor,
const uint8_t private_key[SC_PRIVKEY_SIZE]);
void stream_encoder_destroy(struct stream_encoder *encoder);
/* Заимствованный публичный ключ, полученный из private_key этого encoder. */
const uint8_t *stream_encoder_public_key(const struct stream_encoder *encoder);
/* blocks — массив descriptor.n указателей; input не перекрывается с рабочими буферами encoder.
* 0 < length <= stream_input_size; остаток пачки дополняется нулями. seq != UINT64_MAX.
* При ошибке blocks не изменяется; собственные рабочие буферы могут измениться. */
int stream_encoder_encode(struct stream_encoder *encoder, uint64_t seq, uint16_t timestamp_ms, uint8_t flags,
const uint8_t *input, size_t length, const uint8_t **blocks);
int stream_block_verify(const struct stream_descriptor *descriptor, const uint8_t public_key[SC_PUBKEY_SIZE],
const uint8_t *wire, size_t length, struct stream_block_view *view);
#ifdef __cplusplus
}
#endif
#endif

330
src/stream/stream_buffer.c

@ -0,0 +1,330 @@
#include <string.h>
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "stream_buffer.h"
#define STREAM_SLOT_NONE UINT32_MAX
struct stream_cached_slot {
uint32_t next;
uint8_t subblock;
};
struct stream_cached_batch {
struct stream_cached_batch *hash_next;
uint64_t seq;
uint32_t first_slot;
uint32_t data_length;
uint16_t timestamp_ms;
uint16_t count;
uint8_t flags;
};
struct stream_buffer {
struct stream_descriptor descriptor;
uint8_t public_key[SC_PUBKEY_SIZE];
struct rs_codec *rs;
uint8_t *decode_output;
uint8_t *data;
struct stream_cached_slot *slots;
struct stream_cached_batch **hash;
struct stream_cached_batch **heap;
uint64_t latest[RS_MAX_BLOCKS];
uint8_t latest_valid[RS_MAX_BLOCKS];
size_t block_bytes;
uint32_t capacity;
uint32_t hash_size;
uint32_t free_slot;
struct stream_buffer_stats stats;
};
/* Индекс пачки не зависит от порядка поступления; множество дырок не требует линейного поиска. */
static uint32_t buffer_hash(const struct stream_buffer *b, uint64_t seq) {
seq ^= seq >> 33;
seq *= UINT64_C(0xff51afd7ed558ccd);
seq ^= seq >> 33;
return (uint32_t)seq & (b->hash_size - 1);
}
static struct stream_cached_batch *buffer_batch(const struct stream_buffer *b, uint64_t seq) {
for (struct stream_cached_batch *batch = b->hash[buffer_hash(b, seq)]; batch; batch = batch->hash_next)
if (batch->seq == seq) return batch;
return NULL;
}
/* Поиск не более n блоков пачки; пустые субпотоки не занимают отдельные позиции в массиве данных. */
static uint32_t buffer_slot(const struct stream_buffer *b, const struct stream_cached_batch *batch, uint8_t subblock) {
for (uint32_t slot = batch->first_slot; slot != STREAM_SLOT_NONE; slot = b->slots[slot].next)
if (b->slots[slot].subblock == subblock) return slot;
return STREAM_SLOT_NONE;
}
/* Обмен узлов min-heap, упорядоченного по индексу пачки. */
static void buffer_heap_swap(struct stream_buffer *b, uint32_t first, uint32_t second) {
struct stream_cached_batch *batch = b->heap[first];
b->heap[first] = b->heap[second];
b->heap[second] = batch;
}
/* Добавить новую пачку в хеш и heap; память уже выделена, отказов здесь нет. */
static void buffer_insert_batch(struct stream_buffer *b, struct stream_cached_batch *batch) {
uint32_t hash = buffer_hash(b, batch->seq);
batch->hash_next = b->hash[hash];
b->hash[hash] = batch;
uint32_t index = b->stats.batches++;
b->heap[index] = batch;
while (index && b->heap[(index - 1) / 2]->seq > batch->seq) {
buffer_heap_swap(b, index, (index - 1) / 2);
index = (index - 1) / 2;
}
}
/* Удалить целую минимальную пачку; выданные наружу указатели становятся недействительными. */
static void buffer_evict_oldest(struct stream_buffer *b) {
struct stream_cached_batch *batch = b->heap[0];
uint32_t hash = buffer_hash(b, batch->seq);
struct stream_cached_batch **link = &b->hash[hash];
while (*link != batch) link = &(*link)->hash_next;
*link = batch->hash_next;
uint32_t remaining = --b->stats.batches;
if (remaining) {
b->heap[0] = b->heap[remaining];
uint32_t index = 0;
while (index * 2 + 1 < remaining) {
uint32_t child = index * 2 + 1;
if (child + 1 < remaining && b->heap[child + 1]->seq < b->heap[child]->seq) child++;
if (b->heap[index]->seq <= b->heap[child]->seq) break;
buffer_heap_swap(b, index, child);
index = child;
}
}
for (uint32_t slot = batch->first_slot; slot != STREAM_SLOT_NONE;) {
uint32_t next = b->slots[slot].next;
unsigned sub = b->slots[slot].subblock;
if (b->latest_valid[sub] && b->latest[sub] == batch->seq) b->latest_valid[sub] = 0;
b->slots[slot].next = b->free_slot;
b->free_slot = slot;
slot = next;
}
b->stats.expired_before = batch->seq + 1;
b->stats.blocks -= batch->count;
b->stats.stored_bytes -= batch->count * b->block_bytes;
b->stats.evicted_batches++;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "buffer evicted seq=%llu blocks=%u retained=%u bytes=%zu low=%llu",
(unsigned long long)batch->seq, batch->count, b->stats.blocks, b->stats.stored_bytes,
(unsigned long long)b->stats.expired_before);
u_free(batch);
}
/* Выделить ограниченные массивы данных/индексов; память пачек растёт только по фактическому поступлению. */
int stream_buffer_create(struct stream_buffer **out, const struct stream_descriptor *d, const uint8_t public_key[32], size_t bytes) {
if (!out || *out || !public_key || stream_descriptor_validate(d) != STREAM_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer create invalid arguments out=%p key=%p", (void *)out, (const void *)public_key);
return STREAM_ERR_ARGUMENT;
}
size_t block_bytes = stream_block_size(d);
if (!bytes) bytes = STREAM_BUFFER_BYTES;
if (bytes > STREAM_BUFFER_BYTES || bytes < d->n * block_bytes) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer budget invalid bytes=%zu batch=%zu max=%u", bytes, d->n * block_bytes, STREAM_BUFFER_BYTES);
return STREAM_ERR_ARGUMENT;
}
struct stream_buffer *b = u_calloc(1, sizeof(*b));
if (!b) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer context allocation failed bytes=%zu", bytes); return STREAM_ERR_MEMORY; }
b->descriptor = *d;
memcpy(b->public_key, public_key, SC_PUBKEY_SIZE);
b->block_bytes = block_bytes;
b->capacity = (uint32_t)(bytes / block_bytes);
b->hash_size = 1;
while (b->hash_size < b->capacity) b->hash_size *= 2;
b->stats.capacity_bytes = b->capacity * block_bytes;
b->data = u_malloc((uint32_t)b->stats.capacity_bytes);
b->slots = u_calloc(b->capacity, sizeof(*b->slots));
b->heap = u_calloc(b->capacity, sizeof(*b->heap));
b->hash = u_calloc(b->hash_size, sizeof(*b->hash));
b->decode_output = u_malloc((uint32_t)stream_input_size(d));
if (!b->data || !b->slots || !b->heap || !b->hash || !b->decode_output) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer arrays allocation failed capacity=%u bytes=%zu", b->capacity, b->stats.capacity_bytes);
stream_buffer_destroy(b);
return STREAM_ERR_MEMORY;
}
for (uint32_t i = 0; i < b->capacity; i++) b->slots[i].next = i + 1;
b->slots[b->capacity - 1].next = STREAM_SLOT_NONE;
if (rs_init(&b->rs, d->n, d->r, d->block_size) != RS_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer decode RS initialization failed n=%u r=%u", d->n, d->r);
stream_buffer_destroy(b);
return STREAM_ERR_FEC;
}
*out = b;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "buffer created source=%llx epoch=%llx bytes=%zu block=%zu slots=%u",
(unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch,
b->stats.capacity_bytes, b->block_bytes, b->capacity);
return STREAM_OK;
}
/* Прекращение владельцем всех обращений предшествует destroy; кольцо ничего не знает о чужих NCD handles. */
void stream_buffer_destroy(struct stream_buffer *b) {
if (!b) return;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "buffer destroy source=%llx blocks=%u batches=%u bytes=%zu evicted=%llu duplicates=%llu",
(unsigned long long)b->descriptor.source_node_id, b->stats.blocks, b->stats.batches, b->stats.stored_bytes,
(unsigned long long)b->stats.evicted_batches, (unsigned long long)b->stats.duplicates);
for (uint32_t i = 0; i < b->stats.batches; i++) u_free(b->heap[i]);
rs_free(b->rs);
u_free(b->decode_output);
u_free(b->data);
u_free(b->slots);
u_free(b->heap);
u_free(b->hash);
u_free(b);
}
/* Вставить только подписанный блок; OOM/конфликт проверяются до вытеснения ранее сохранённых данных. */
int stream_buffer_put(struct stream_buffer *b, const uint8_t *wire, size_t length) {
if (!b || !wire) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer put invalid arguments length=%zu", length); return STREAM_ERR_ARGUMENT; }
struct stream_cached_batch *batch = NULL;
if (length == b->block_bytes) {
batch = buffer_batch(b, stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8));
uint8_t sub = wire[STREAM_BLOCK_SUB_OFF];
if (batch && sub < b->descriptor.n) {
uint32_t slot = buffer_slot(b, batch, sub);
if (slot != STREAM_SLOT_NONE && !memcmp(b->data + slot * b->block_bytes, wire, length)) {
b->stats.duplicates++;
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer duplicate seq=%llu sub=%u", (unsigned long long)batch->seq, sub);
return STREAM_DUPLICATE;
}
}
}
struct stream_block_view view;
int result = stream_block_verify(&b->descriptor, b->public_key, wire, length, &view);
if (result != STREAM_OK) return result;
if (view.seq < b->stats.expired_before) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer expired incoming seq=%llu sub=%u low=%llu full=%d",
(unsigned long long)view.seq, view.subblock, (unsigned long long)b->stats.expired_before, b->stats.blocks == b->capacity);
return STREAM_EXPIRED;
}
if (batch && (batch->timestamp_ms != view.timestamp_ms || batch->data_length != view.data_length || batch->flags != view.flags
|| buffer_slot(b, batch, view.subblock) != STREAM_SLOT_NONE)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer source conflict seq=%llu sub=%u timestamp=%u/%u length=%u/%u flags=%u/%u",
(unsigned long long)view.seq, view.subblock, batch->timestamp_ms, view.timestamp_ms,
batch->data_length, view.data_length, batch->flags, view.flags);
return STREAM_ERR_CONFLICT;
}
if (b->stats.blocks == b->capacity && view.seq <= b->heap[0]->seq) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "buffer full: incoming seq=%llu sub=%u cannot replace newer data oldest=%llu",
(unsigned long long)view.seq, view.subblock, (unsigned long long)b->heap[0]->seq);
return STREAM_EXPIRED;
}
int new_batch = batch == NULL;
if (new_batch) {
batch = u_calloc(1, sizeof(*batch));
if (!batch) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer batch allocation failed seq=%llu", (unsigned long long)view.seq);
return STREAM_ERR_MEMORY;
}
batch->seq = view.seq;
batch->first_slot = STREAM_SLOT_NONE;
batch->timestamp_ms = view.timestamp_ms;
batch->data_length = view.data_length;
batch->flags = view.flags;
}
if (b->stats.blocks == b->capacity) buffer_evict_oldest(b);
if (new_batch) buffer_insert_batch(b, batch);
uint32_t slot = b->free_slot;
b->free_slot = b->slots[slot].next;
b->slots[slot].subblock = view.subblock;
b->slots[slot].next = batch->first_slot;
batch->first_slot = slot;
batch->count++;
memcpy(b->data + slot * b->block_bytes, wire, length);
b->stats.blocks++;
b->stats.stored_bytes += length;
if (view.seq > b->stats.highest_seq) b->stats.highest_seq = view.seq;
if (!b->latest_valid[view.subblock] || view.seq > b->latest[view.subblock]) {
b->latest[view.subblock] = view.seq;
b->latest_valid[view.subblock] = 1;
}
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer stored seq=%llu sub=%u present=%u/%u bytes=%zu",
(unsigned long long)view.seq, view.subblock, batch->count, b->descriptor.n, b->stats.stored_bytes);
return STREAM_OK;
}
/* Найти точную позицию либо последний блок субпотока, не меняя его историю/срок хранения. */
int stream_buffer_read(struct stream_buffer *b, uint64_t seq, uint8_t subblock,
const uint8_t **wire, size_t *length, uint64_t *actual_seq) {
if (!b || !wire || !length || !actual_seq || subblock >= b->descriptor.n) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer read invalid arguments seq=%llu sub=%u", (unsigned long long)seq, subblock);
return STREAM_ERR_ARGUMENT;
}
if (seq == UINT64_MAX) {
if (!b->latest_valid[subblock]) return STREAM_WAIT;
seq = b->latest[subblock];
}
if (seq < b->stats.expired_before) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "buffer read expired seq=%llu low=%llu", (unsigned long long)seq,
(unsigned long long)b->stats.expired_before);
return STREAM_EXPIRED;
}
if (!b->stats.blocks || seq > b->stats.highest_seq) return STREAM_WAIT;
struct stream_cached_batch *batch = buffer_batch(b, seq);
uint32_t slot = batch ? buffer_slot(b, batch, subblock) : STREAM_SLOT_NONE;
if (slot == STREAM_SLOT_NONE) {
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer read gap seq=%llu sub=%u", (unsigned long long)seq, subblock);
return STREAM_GAP;
}
*wire = b->data + slot * b->block_bytes;
*length = b->block_bytes;
*actual_seq = seq;
return STREAM_OK;
}
/* Границы наличия конкретного субпотока не обещают заполнения всех промежуточных индексов. */
int stream_buffer_range(struct stream_buffer *b, uint8_t subblock, uint64_t *first, uint64_t *last) {
if (!b || !first || !last || subblock >= b->descriptor.n) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer range invalid sub=%u", subblock);
return STREAM_ERR_ARGUMENT;
}
if (!b->latest_valid[subblock]) return STREAM_WAIT;
uint64_t low = b->latest[subblock];
for (uint32_t i = 0; i < b->stats.batches; i++) {
struct stream_cached_batch *batch = b->heap[i];
if (batch->seq < low && buffer_slot(b, batch, subblock) != STREAM_SLOT_NONE) low = batch->seq;
if (low == b->heap[0]->seq) break;
}
*first = low;
*last = b->latest[subblock];
return STREAM_OK;
}
/* RS пишет в собственный scratch; ошибка или недостающие блоки не повреждают выход потребителя. */
int stream_buffer_decode(struct stream_buffer *b, uint64_t seq, uint8_t *output, size_t capacity, size_t *length) {
if (!b || !output || !length || seq == UINT64_MAX) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer decode invalid arguments seq=%llu capacity=%zu", (unsigned long long)seq, capacity);
return STREAM_ERR_ARGUMENT;
}
if (seq < b->stats.expired_before) return STREAM_EXPIRED;
struct stream_cached_batch *batch = buffer_batch(b, seq);
if (!batch) return !b->stats.blocks || seq > b->stats.highest_seq ? STREAM_WAIT : STREAM_GAP;
if (batch->count < b->descriptor.n - b->descriptor.r) return STREAM_WAIT;
if (capacity < batch->data_length) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer decode output too small seq=%llu capacity=%zu needed=%u",
(unsigned long long)seq, capacity, batch->data_length);
return STREAM_ERR_ARGUMENT;
}
const uint8_t *blocks[RS_MAX_BLOCKS] = {0};
for (uint32_t slot = batch->first_slot; slot != STREAM_SLOT_NONE; slot = b->slots[slot].next)
blocks[b->slots[slot].subblock] = b->data + slot * b->block_bytes + STREAM_BLOCK_RS_OFF;
if (rs_decode(b->rs, blocks, b->decode_output) != RS_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer decode RS failed seq=%llu present=%u", (unsigned long long)seq, batch->count);
return STREAM_ERR_FEC;
}
memcpy(output, b->decode_output, batch->data_length);
*length = batch->data_length;
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "buffer decoded seq=%llu present=%u length=%u", (unsigned long long)seq, batch->count, batch->data_length);
return STREAM_OK;
}
int stream_buffer_get_stats(const struct stream_buffer *b, struct stream_buffer_stats *stats) {
if (!b || !stats) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "buffer stats invalid arguments"); return STREAM_ERR_ARGUMENT; }
*stats = b->stats;
return STREAM_OK;
}

53
src/stream/stream_buffer.h

@ -0,0 +1,53 @@
/* Кольцо проверенных блоков одного запуска: общий бюджет данных для всех субпотоков.
* Пустые субпотоки не резервируют блоки: 32 МБ занимают реально сохранённые кодограммы.
* Вытеснение целой старейшей пачки не закрепляет данные для медленного получателя.
* Все вызовы — один uasync-поток. read возвращает память до следующего put/destroy.
* public_key должен происходить из проверенной записи источника в группе. */
#ifndef STREAM_BUFFER_H
#define STREAM_BUFFER_H
#include "stream_block.h"
#ifdef __cplusplus
extern "C" {
#endif
struct stream_buffer;
struct stream_buffer_stats {
size_t capacity_bytes;
size_t stored_bytes;
uint32_t blocks;
uint32_t batches;
uint64_t expired_before;
uint64_t highest_seq;
uint64_t evicted_batches;
uint64_t duplicates;
};
/* out заранее NULL; bytes=0 выбирает 32 МБ, меньший бюджет допускается для стенда.
* Должна помещаться хотя бы одна полная пачка n блоков; верхний предел — STREAM_BUFFER_BYTES. */
int stream_buffer_create(struct stream_buffer **out, const struct stream_descriptor *descriptor,
const uint8_t public_key[SC_PUBKEY_SIZE], size_t bytes);
void stream_buffer_destroy(struct stream_buffer *buffer);
/* Новый блок проверяется до изменения кольца. Повтор неизменного блока: STREAM_DUPLICATE.
* STREAM_EXPIRED не меняет более свежие данные; неверная подпись/конфликт возвращает отрицательный код. */
int stream_buffer_put(struct stream_buffer *buffer, const uint8_t *wire, size_t length);
/* seq=UINT64_MAX читает последний имеющийся блок именно запрошенного субпотока.
* WAIT — блок впереди головы/пустое кольцо; GAP — дыра; EXPIRED — вытеснен.
* При отсутствии блока выходные указатели/номер не меняются. */
int stream_buffer_read(struct stream_buffer *buffer, uint64_t seq, uint8_t subblock,
const uint8_t **wire, size_t *length, uint64_t *actual_seq);
int stream_buffer_range(struct stream_buffer *buffer, uint8_t subblock, uint64_t *first, uint64_t *last);
/* Восстановить пачку из проверенных блоков; достаточно capacity >= действительной длины.
* Выход не меняется при WAIT/GAP/EXPIRED/ошибке; восстановленные отсутствующие блоки в кольцо не добавляются. */
int stream_buffer_decode(struct stream_buffer *buffer, uint64_t seq, uint8_t *output, size_t capacity, size_t *length);
int stream_buffer_get_stats(const struct stream_buffer *buffer, struct stream_buffer_stats *stats);
#ifdef __cplusplus
}
#endif
#endif

136
src/stream/stream_metrics.c

@ -0,0 +1,136 @@
#include <string.h>
#include "../../lib/debug_config.h"
#include "stream_metrics.h"
/* Единственная корзина секундного интервала; old данные не переживают оборот окна. */
static struct stream_metric_bucket *metrics_bucket(struct stream_metrics *m, uint64_t now_ms) {
uint64_t second = now_ms / 1000;
struct stream_metric_bucket *bucket = &m->buckets[second % STREAM_METRIC_SECONDS];
if (!bucket->valid || bucket->second != second) {
memset(bucket, 0, sizeof(*bucket));
bucket->second = second;
bucket->valid = 1;
}
return bucket;
}
/* Переполнение вычисленного отставания не может превращать длительный stall в нулевую задержку. */
static uint64_t metrics_add(uint64_t first, uint64_t second) {
return second > UINT64_MAX - first ? UINT64_MAX : first + second;
}
/* Фиксировать одну остановку только после отсутствия нового блока сверх ожидаемого интервала + 500 мс. */
static void metrics_detect_stop(struct stream_metrics *m, uint64_t now_ms) {
if (!m->initialized || m->ended || m->stopped) return;
uint64_t threshold = metrics_add(m->expected_interval_ms, STREAM_LATE_MS);
uint64_t elapsed = now_ms - m->last_receive_ms;
if (elapsed <= threshold) return;
uint64_t started = metrics_add(m->last_receive_ms, metrics_add(threshold, 1));
metrics_bucket(m, started)->stops++;
m->stopped = 1;
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "subflow stopped after seq=%llu elapsed=%llums expected=%llums lag=%llums",
(unsigned long long)m->last_seq, (unsigned long long)elapsed,
(unsigned long long)m->expected_interval_ms, (unsigned long long)m->lag_ms);
}
/* Начальная ожидаемая пауза — source flush 100 мс; после замеров используется фактический темп источника. */
void stream_metrics_init(struct stream_metrics *m) {
if (!m) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "metrics init null context"); return; }
memset(m, 0, sizeof(*m));
m->expected_interval_ms = STREAM_SOURCE_FLUSH_TB / 10;
}
/* Дельты относительно предшествующего блока исключают смещение часов и учитывают отставание по всей цепочке. */
int stream_metrics_observe(struct stream_metrics *m, uint64_t seq, uint16_t timestamp_ms, uint8_t flags,
uint32_t bytes, uint64_t now_ms) {
if (!m || seq == UINT64_MAX || !bytes || (flags & ~STREAM_BLOCK_TIME_RESET)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "metrics observe invalid seq=%llu bytes=%u flags=%u", (unsigned long long)seq, bytes, flags);
return STREAM_ERR_ARGUMENT;
}
if (m->ended || (m->initialized && now_ms < m->last_receive_ms)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "metrics state/clock invalid seq=%llu now=%llu previous=%llu ended=%d",
(unsigned long long)seq, (unsigned long long)now_ms, (unsigned long long)m->last_receive_ms, m->ended);
return STREAM_ERR_STATE;
}
if (m->initialized && seq <= m->last_seq) return STREAM_DUPLICATE;
metrics_detect_stop(m, now_ms);
uint64_t receive_delta = m->initialized ? now_ms - m->last_receive_ms : 0;
int rebase = !m->initialized || (flags & STREAM_BLOCK_TIME_RESET) || seq != m->last_seq + 1
|| receive_delta >= STREAM_TIMESTAMP_PERIOD_MS;
uint16_t source_delta = (uint16_t)(timestamp_ms - m->last_timestamp_ms);
struct stream_metric_bucket *bucket = metrics_bucket(m, now_ms);
bucket->blocks++;
if (rebase) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "metrics rebase seq=%llu previous=%llu timestamp=%u/%u receive_delta=%llu flags=%u",
(unsigned long long)seq, (unsigned long long)m->last_seq, m->last_timestamp_ms, timestamp_ms,
(unsigned long long)receive_delta, flags);
m->lag_ms = 0;
m->expected_interval_ms = STREAM_SOURCE_FLUSH_TB / 10;
} else {
if (receive_delta >= source_delta) m->lag_ms = metrics_add(m->lag_ms, receive_delta - source_delta);
else m->lag_ms = source_delta - receive_delta >= m->lag_ms ? 0 : m->lag_ms - (source_delta - receive_delta);
bucket->bytes += bytes;
bucket->source_ms += source_delta;
bucket->receive_ms += receive_delta;
bucket->intervals++;
if (bucket->source_ms) {
m->expected_interval_ms = bucket->source_ms / bucket->intervals;
if (!m->expected_interval_ms) m->expected_interval_ms = 1;
}
if (m->lag_ms > STREAM_LATE_MS) bucket->late_blocks++;
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "metrics seq=%llu source_delta=%u receive_delta=%llu lag=%llu expected=%llu",
(unsigned long long)seq, source_delta, (unsigned long long)receive_delta,
(unsigned long long)m->lag_ms, (unsigned long long)m->expected_interval_ms);
}
m->last_seq = seq;
m->last_timestamp_ms = timestamp_ms;
m->last_receive_ms = now_ms;
m->initialized = 1;
m->stopped = 0;
return STREAM_OK;
}
/* Скорость суммируется по объёму и времени, а не усреднением частных rates разных интервалов. */
int stream_metrics_poll(struct stream_metrics *m, uint64_t now_ms, struct stream_metric_snapshot *snapshot) {
if (!m || !snapshot) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "metrics poll invalid arguments"); return STREAM_ERR_ARGUMENT; }
if (m->initialized && now_ms < m->last_receive_ms) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "metrics poll clock regression now=%llu previous=%llu",
(unsigned long long)now_ms, (unsigned long long)m->last_receive_ms);
return STREAM_ERR_STATE;
}
struct stream_metric_snapshot result = {0};
if (!m->initialized) { *snapshot = result; return STREAM_WAIT; }
metrics_detect_stop(m, now_ms);
result.lag_ms = m->lag_ms;
result.expected_interval_ms = m->expected_interval_ms;
result.stopped = m->stopped;
if (!m->ended && now_ms - m->last_receive_ms > m->expected_interval_ms)
result.lag_ms = metrics_add(result.lag_ms, now_ms - m->last_receive_ms - m->expected_interval_ms);
uint64_t second = now_ms / 1000;
uint64_t bytes = 0, source_ms = 0, receive_ms = 0;
for (unsigned i = 0; i < STREAM_METRIC_SECONDS; i++) {
const struct stream_metric_bucket *bucket = &m->buckets[i];
if (!bucket->valid || bucket->second > second || second - bucket->second >= STREAM_METRIC_SECONDS) continue;
result.blocks += bucket->blocks;
result.late_blocks += bucket->late_blocks;
result.stops += bucket->stops;
bytes += bucket->bytes;
source_ms += bucket->source_ms;
receive_ms += bucket->receive_ms;
}
result.source_rate_valid = source_ms != 0;
result.receive_rate_valid = receive_ms != 0;
if (source_ms) result.source_rate_bytes_sec = bytes * 1000 / source_ms;
if (receive_ms) result.receive_rate_bytes_sec = bytes * 1000 / receive_ms;
*snapshot = result;
return STREAM_OK;
}
/* END останавливает ожидание будущих данных, история качества остаётся доступной. */
void stream_metrics_end(struct stream_metrics *m) {
if (!m) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "metrics end null context"); return; }
m->ended = 1;
m->stopped = 0;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "metrics ended at seq=%llu", (unsigned long long)m->last_seq);
}

64
src/stream/stream_metrics.h

@ -0,0 +1,64 @@
/* Относительное отставание одного субпотока от одного поставщика, без NTP и сокетов.
* now_ms — монотонные часы получателя; timestamp — неизменяемые 16-битные часы источника.
* Первый блок/разрыв устанавливает новую базу, история проблем остаётся за последние 60 секунд.
* observe/poll вызываются одним uasync-потоком. DUPLICATE не меняет ни часы, ни статистику. */
#ifndef STREAM_METRICS_H
#define STREAM_METRICS_H
#include "stream_proto.h"
#ifdef __cplusplus
extern "C" {
#endif
#define STREAM_METRIC_SECONDS 60
struct stream_metric_bucket {
uint64_t second;
uint64_t blocks;
uint64_t late_blocks;
uint64_t stops;
uint64_t bytes;
uint64_t source_ms;
uint64_t receive_ms;
uint64_t intervals;
int valid;
};
struct stream_metrics {
struct stream_metric_bucket buckets[STREAM_METRIC_SECONDS];
uint64_t last_seq;
uint64_t last_receive_ms;
uint64_t lag_ms;
uint64_t expected_interval_ms;
uint16_t last_timestamp_ms;
int initialized;
int stopped;
int ended;
};
struct stream_metric_snapshot {
uint64_t lag_ms;
uint64_t blocks;
uint64_t late_blocks;
uint64_t stops;
uint64_t source_rate_bytes_sec;
uint64_t receive_rate_bytes_sec;
uint64_t expected_interval_ms;
int stopped;
int source_rate_valid;
int receive_rate_valid;
};
void stream_metrics_init(struct stream_metrics *metrics);
int stream_metrics_observe(struct stream_metrics *metrics, uint64_t seq, uint16_t timestamp_ms, uint8_t flags,
uint32_t bytes, uint64_t now_ms);
/* Также обнаруживает полную остановку; episode учитывается в момент превышения порога, не на каждом poll.
* До первого блока возвращает WAIT и обнулённый snapshot; при ошибке snapshot остаётся прежним. */
int stream_metrics_poll(struct stream_metrics *metrics, uint64_t now_ms, struct stream_metric_snapshot *snapshot);
void stream_metrics_end(struct stream_metrics *metrics);
#ifdef __cplusplus
}
#endif
#endif

73
src/stream/stream_proto.h

@ -0,0 +1,73 @@
/* Неизменяемый блок источника: все числа big-endian, подпись охватывает тело до signature.
* Данные ETCP-выдачи (handle/сессия/путь) добавляются снаружи и в подпись не входят.
* seq/subblock берутся из заголовка RS; второго независимого номера блока нет. */
#ifndef STREAM_PROTO_H
#define STREAM_PROTO_H
#include <stddef.h>
#include <stdint.h>
#include "../../lib/reed_solomon.h"
#include "../transport_layer/secure_channel.h"
#define STREAM_BLOCK_DOMAIN "stream_block"
#define STREAM_BLOCK_DOMAIN_SIZE 12
#define STREAM_BLOCK_GROUP_OFF 12
#define STREAM_BLOCK_SOURCE_OFF 20
#define STREAM_BLOCK_EPOCH_OFF 28
#define STREAM_BLOCK_N_OFF 36
#define STREAM_BLOCK_R_OFF 37
#define STREAM_BLOCK_SIZE_OFF 38
#define STREAM_BLOCK_TIMESTAMP_OFF 40
#define STREAM_BLOCK_LENGTH_OFF 42
#define STREAM_BLOCK_FLAGS_OFF 46
#define STREAM_BLOCK_RS_OFF 47
#define STREAM_BLOCK_SEQ_OFF STREAM_BLOCK_RS_OFF
#define STREAM_BLOCK_SUB_OFF (STREAM_BLOCK_RS_OFF + 8)
#define STREAM_BLOCK_OVERHEAD (STREAM_BLOCK_RS_OFF + SC_SIGN_SIZE)
#define STREAM_BLOCK_DEFAULT_SIZE 1300
#define STREAM_RS_DEFAULT_SIZE (STREAM_BLOCK_DEFAULT_SIZE - STREAM_BLOCK_OVERHEAD)
#define STREAM_BLOCK_MAX_SIZE (RS_MAX_BLOCK_SIZE + STREAM_BLOCK_OVERHEAD)
#define STREAM_BLOCK_TIME_RESET 0x01
#define STREAM_BUFFER_BYTES (32u * 1024u * 1024u)
#define STREAM_SOURCE_FLUSH_TB 1000 /* 100 мс, timebase 0.1 мс */
#define STREAM_LATE_MS 500
#define STREAM_TIMESTAMP_PERIOD_MS 65536u
struct stream_descriptor {
uint64_t group_id;
uint64_t source_node_id;
uint64_t source_epoch;
uint16_t block_size; /* Полный размер RS-кодограммы, без stream-метаданных/подписи. */
uint8_t n;
uint8_t r;
};
enum stream_result {
STREAM_OK = 0,
STREAM_DUPLICATE = 1,
STREAM_WAIT = 2,
STREAM_GAP = 3,
STREAM_EXPIRED = 4,
STREAM_ERR_ARGUMENT = -1,
STREAM_ERR_MEMORY = -2,
STREAM_ERR_FORMAT = -3,
STREAM_ERR_SIGNATURE = -4,
STREAM_ERR_CONFLICT = -5,
STREAM_ERR_STATE = -6,
STREAM_ERR_CRYPTO = -7,
STREAM_ERR_FEC = -8
};
/* Короткие примитивы без доступа за границы: размер проверяет вызывающий parser. */
static inline uint64_t stream_read_be(const uint8_t *data, unsigned bytes) {
uint64_t result = 0;
for (unsigned i = 0; i < bytes; i++) result = (result << 8) | data[i];
return result;
}
static inline void stream_write_be(uint8_t *data, uint64_t value, unsigned bytes) {
for (unsigned i = 0; i < bytes; i++) data[bytes - i - 1] = (uint8_t)(value >> (8 * i));
}
#endif

181
src/stream/stream_source.c

@ -0,0 +1,181 @@
#include <string.h>
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "../../lib/u_async.h"
#include "stream_source.h"
struct stream_source {
struct UASYNC *ua;
struct stream_descriptor descriptor;
struct stream_encoder *encoder;
struct stream_buffer *buffer;
uint8_t *input;
size_t capacity;
size_t used;
uint64_t next_seq;
uint64_t last_batch_ms;
uint64_t emitted_bytes;
uint64_t report_ms;
void *timer;
stream_source_ready_fn ready;
void *arg;
int busy;
int has_batch;
int failed;
int ended;
};
/* Ошибка не скрывает уже принятый префикс: прекращаем таймер и дальнейшие write. */
static int source_fail(struct stream_source *source, int result) {
source->failed = 1;
if (source->timer) { uasync_cancel_timeout(source->ua, source->timer); source->timer = NULL; }
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source failed source=%llx seq=%llu pending=%zu result=%d",
(unsigned long long)source->descriptor.source_node_id, (unsigned long long)source->next_seq, source->used, result);
return result;
}
/* Сформировать блоки и сохранить их до уведомления потребителей; busy установлен внешней операцией. */
static int source_flush(struct stream_source *source) {
if (!source->used) return STREAM_OK;
if (source->timer) { uasync_cancel_timeout(source->ua, source->timer); source->timer = NULL; }
uint64_t now_ms = get_time_tb() / 10;
if (source->next_seq == UINT64_MAX || (source->has_batch && now_ms < source->last_batch_ms)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source sequence/clock invalid seq=%llu now=%llu previous=%llu",
(unsigned long long)source->next_seq, (unsigned long long)now_ms, (unsigned long long)source->last_batch_ms);
return source_fail(source, STREAM_ERR_STATE);
}
uint8_t flags = source->has_batch && now_ms - source->last_batch_ms >= STREAM_TIMESTAMP_PERIOD_MS ? STREAM_BLOCK_TIME_RESET : 0;
const uint8_t *blocks[RS_MAX_BLOCKS];
int result = stream_encoder_encode(source->encoder, source->next_seq, (uint16_t)now_ms, flags, source->input, source->used, blocks);
if (result != STREAM_OK) return source_fail(source, result);
for (unsigned i = 0; i < source->descriptor.n; i++) {
result = stream_buffer_put(source->buffer, blocks[i], stream_block_size(&source->descriptor));
if (result != STREAM_OK) return source_fail(source, result < 0 ? result : STREAM_ERR_STATE);
}
uint64_t seq = source->next_seq++;
source->emitted_bytes += source->used;
source->used = 0;
source->has_batch = 1;
source->last_batch_ms = now_ms;
if (!source->report_ms || now_ms - source->report_ms >= 1000) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "source progress source=%llx seq=%llu bytes=%llu timestamp=%u flags=%u",
(unsigned long long)source->descriptor.source_node_id, (unsigned long long)seq,
(unsigned long long)source->emitted_bytes, (uint16_t)now_ms, flags);
source->report_ms = now_ms;
}
if (source->ready) source->ready(source, seq, source->arg);
return STREAM_OK;
}
/* Срабатывание единственного таймера неполной пачки; handle больше не принадлежит таймерному heap. */
static void source_flush_timer(void *arg) {
struct stream_source *source = arg;
source->timer = NULL;
source->busy = 1;
int result = source_flush(source);
source->busy = 0;
if (result != STREAM_OK) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source timed flush failed result=%d", result);
}
/* Создать только реальную локальную публикацию; память удалённых анонсов сюда не попадает. */
int stream_source_create(struct stream_source **out, struct UASYNC *ua, const struct stream_descriptor *d,
const uint8_t private_key[32], stream_source_ready_fn ready, void *arg) {
if (!out || *out || !ua || !private_key || stream_descriptor_validate(d) != STREAM_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source create invalid arguments out=%p ua=%p", (void *)out, (void *)ua);
return STREAM_ERR_ARGUMENT;
}
struct stream_source *source = u_calloc(1, sizeof(*source));
if (!source) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source context allocation failed"); return STREAM_ERR_MEMORY; }
source->ua = ua;
source->descriptor = *d;
source->ready = ready;
source->arg = arg;
source->capacity = stream_input_size(d);
source->input = u_malloc((uint32_t)source->capacity);
int result = source->input ? stream_encoder_create(&source->encoder, d, private_key) : STREAM_ERR_MEMORY;
if (result == STREAM_OK)
result = stream_buffer_create(&source->buffer, d, stream_encoder_public_key(source->encoder), STREAM_BUFFER_BYTES);
if (result != STREAM_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source initialization failed source=%llx result=%d", (unsigned long long)d->source_node_id, result);
stream_source_destroy(source);
return result;
}
*out = source;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "source created group=%llx source=%llx epoch=%llx input=%zu flush=100ms",
(unsigned long long)d->group_id, (unsigned long long)d->source_node_id,
(unsigned long long)d->source_epoch, source->capacity);
return STREAM_OK;
}
void stream_source_destroy(struct stream_source *source) {
if (!source) return;
if (source->busy) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source destroy inside callback rejected"); return; }
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "source destroy source=%llx batches=%llu bytes=%llu pending=%zu ended=%d failed=%d",
(unsigned long long)source->descriptor.source_node_id, (unsigned long long)source->next_seq,
(unsigned long long)source->emitted_bytes, source->used, source->ended, source->failed);
if (source->timer) uasync_cancel_timeout(source->ua, source->timer);
stream_buffer_destroy(source->buffer);
stream_encoder_destroy(source->encoder);
u_free(source->input);
u_free(source);
}
size_t stream_source_input_capacity(const struct stream_source *source) {
if (!source) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source capacity null context"); return 0; }
return source->capacity;
}
struct stream_buffer *stream_source_buffer(struct stream_source *source) {
if (!source) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source buffer null context"); return NULL; }
return source->buffer;
}
/* Принимать ограниченную порцию, дополнять старую пачку и заводить таймер только на новый остаток. */
int stream_source_write(struct stream_source *source, const uint8_t *data, size_t length, size_t *accepted) {
if (accepted) *accepted = 0;
if (!source || !accepted || (length && !data) || length > source->capacity) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source write invalid arguments length=%zu", length);
return STREAM_ERR_ARGUMENT;
}
if (source->busy || source->failed || source->ended) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "source write state rejected busy=%d failed=%d ended=%d", source->busy, source->failed, source->ended);
return STREAM_ERR_STATE;
}
if (!length) return STREAM_OK;
source->busy = 1;
int result = STREAM_OK;
while (*accepted < length) {
size_t take = source->capacity - source->used;
if (take > length - *accepted) take = length - *accepted;
memcpy(source->input + source->used, data + *accepted, take);
source->used += take;
*accepted += take;
if (source->used == source->capacity && (result = source_flush(source)) != STREAM_OK) break;
}
if (result == STREAM_OK && source->used && !source->timer) {
source->timer = uasync_set_timeout(source->ua, STREAM_SOURCE_FLUSH_TB, source, source_flush_timer, "stream_source_flush");
if (!source->timer) result = source_fail(source, STREAM_ERR_MEMORY);
}
source->busy = 0;
return result;
}
/* Finish отменяет таймер и публикует остаток сразу; дальнейшие байты требуют нового source_epoch. */
int stream_source_finish(struct stream_source *source) {
if (!source || source->busy || source->failed) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source finish invalid state present=%d", source != NULL);
return source ? STREAM_ERR_STATE : STREAM_ERR_ARGUMENT;
}
if (source->ended) return STREAM_OK;
source->busy = 1;
int result = source_flush(source);
source->busy = 0;
if (result == STREAM_OK) {
source->ended = 1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "source finished source=%llx batches=%llu bytes=%llu",
(unsigned long long)source->descriptor.source_node_id, (unsigned long long)source->next_seq,
(unsigned long long)source->emitted_bytes);
}
return result;
}

36
src/stream/stream_source.h

@ -0,0 +1,36 @@
/* Источник непрозрачных байтов: одна подписанная FEC-пачка за ограниченный write,
* неполная пачка сбрасывается таймером 100 мс, все блоки сначала сохраняются в кольце.
* Один uasync-поток; источник сам владеет encoder/кольцом/таймером. NCD здесь ещё нет.
* ready вызывается после помещения всей пачки и может читать кольцо; write/finish/destroy
* из ready запрещены. destroy выполняется вне callbacks источника. */
#ifndef STREAM_SOURCE_H
#define STREAM_SOURCE_H
#include "stream_buffer.h"
#ifdef __cplusplus
extern "C" {
#endif
struct UASYNC;
struct stream_source;
typedef void (*stream_source_ready_fn)(struct stream_source *source, uint64_t seq, void *arg);
int stream_source_create(struct stream_source **out, struct UASYNC *ua, const struct stream_descriptor *descriptor,
const uint8_t private_key[SC_PRIVKEY_SIZE], stream_source_ready_fn ready, void *arg);
void stream_source_destroy(struct stream_source *source);
size_t stream_source_input_capacity(const struct stream_source *source);
/* Только заимствованное чтение/decode; вставлять блоки/освобождать это кольцо должен сам источник. */
struct stream_buffer *stream_source_buffer(struct stream_source *source);
/* length <= input_capacity: вызов кодирует не более одной пачки, не захватывая uasync большим внешним буфером.
* accepted получает принятую часть даже при ошибке; нулевой write допустим и не создаёт пустую пачку.
* Ошибка публикации/таймера делает источник failed; его уничтожают, не повторяют принятую часть как новую трансляцию. */
int stream_source_write(struct stream_source *source, const uint8_t *data, size_t length, size_t *accepted);
/* Сбросить остаток и остановить приём; повторный finish допустим, кольцо доступно до destroy. */
int stream_source_finish(struct stream_source *source);
#ifdef __cplusplus
}
#endif
#endif

5
tests/Makefile.am

@ -93,6 +93,7 @@ endif
# Короткие интеграционные проверки (до 1с). # Короткие интеграционные проверки (до 1с).
check_PROGRAMS += \ check_PROGRAMS += \
test_stream_core \
test_u_async_comprehensive \ test_u_async_comprehensive \
test_nat_detection \ test_nat_detection \
test_audio_compressor \ test_audio_compressor \
@ -444,6 +445,10 @@ test_reed_solomon_SOURCES = test_reed_solomon.c
test_reed_solomon_CFLAGS = -I$(top_srcdir)/lib test_reed_solomon_CFLAGS = -I$(top_srcdir)/lib
test_reed_solomon_LDADD = $(COMMON_LIBS) test_reed_solomon_LDADD = $(COMMON_LIBS)
test_stream_core_SOURCES = test_stream_core.c $(test_clock_sources)
test_stream_core_LDADD = $(LIBUTUN) $(COMMON_LIBS)
test_stream_core_LDFLAGS = $(test_clock_ldflags)
test_intensive_memory_pool_SOURCES = test_intensive_memory_pool.c test_intensive_memory_pool_SOURCES = test_intensive_memory_pool.c
test_intensive_memory_pool_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib test_intensive_memory_pool_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib
test_intensive_memory_pool_LDADD = $(COMMON_LIBS) test_intensive_memory_pool_LDADD = $(COMMON_LIBS)

326
tests/test_stream_core.c

@ -0,0 +1,326 @@
/* Подпись источника, RS-потери, кольцо, циклическое время и реальные uasync-таймеры источника. */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "../src/stream/stream_source.h"
#include "../src/stream/stream_metrics.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "test_clock.h"
#define CHECK(expr) do { if (!(expr)) { \
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "test_stream_core failed line=%d: %s", __LINE__, #expr); exit(1); \
} } while (0)
static const struct stream_descriptor descriptor = {
.group_id = UINT64_C(0x0123456789abcdef), .source_node_id = UINT64_C(0xfedcba9876543210),
.source_epoch = UINT64_C(0x0102030405060708), .block_size = STREAM_RS_DEFAULT_SIZE, .n = 8, .r = 2
};
static uint8_t private_key[32];
/* Воспроизводимые данные, не совпадающие с zero-padding и проверочными кодограммами. */
static void pattern(uint8_t *data, size_t length) {
uint32_t state = 0x12345678;
for (size_t i = 0; i < length; i++) {
state ^= state << 13; state ^= state >> 17; state ^= state << 5;
data[i] = (uint8_t)state;
}
}
/* Подмена любого класса неизменяемых полей отвергается, view и входной блок сохраняются. */
static void test_signed_blocks(void) {
struct stream_encoder *encoder = NULL;
CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK);
CHECK(stream_block_size(&descriptor) == 1300 && stream_input_size(&descriptor) == 7080);
uint8_t input[7080]; pattern(input, sizeof(input));
const uint8_t *blocks[RS_MAX_BLOCKS];
uint64_t seq = UINT64_C(0x1122334455667788);
CHECK(stream_encoder_encode(encoder, seq, 65530, STREAM_BLOCK_TIME_RESET, input, sizeof(input), blocks) == STREAM_OK);
const uint8_t *public_key = stream_encoder_public_key(encoder);
for (unsigned i = 0; i < descriptor.n; i++) {
struct stream_block_view view;
CHECK(stream_block_verify(&descriptor, public_key, blocks[i], 1300, &view) == STREAM_OK);
CHECK(view.seq == seq && view.subblock == i && view.timestamp_ms == 65530 && view.data_length == sizeof(input));
CHECK(view.flags == STREAM_BLOCK_TIME_RESET && view.rs == blocks[i] + STREAM_BLOCK_RS_OFF);
CHECK(!memcmp(blocks[i], "stream_block", 12));
CHECK(!memcmp(blocks[i] + STREAM_BLOCK_GROUP_OFF, "\x01\x23\x45\x67\x89\xab\xcd\xef", 8));
CHECK(!memcmp(blocks[i] + STREAM_BLOCK_SEQ_OFF, "\x11\x22\x33\x44\x55\x66\x77\x88", 8));
CHECK(sc_ed25519_verify(public_key, blocks[i], 1300 - 64, blocks[i] + 1300 - 64) == SC_OK);
}
const unsigned changed[] = {0, 12, 20, 28, 36, 37, 38, 40, 45, 46, 47, 55, 56, 1235, 1299};
uint8_t wire[STREAM_BLOCK_MAX_SIZE];
for (unsigned i = 0; i < sizeof(changed) / sizeof(changed[0]); i++) {
memcpy(wire, blocks[0], 1300); wire[changed[i]] ^= 1;
struct stream_block_view view; memset(&view, 0xa5, sizeof(view));
uint8_t before[sizeof(view)]; memcpy(before, &view, sizeof(view));
CHECK(stream_block_verify(&descriptor, public_key, wire, 1300, &view) < 0);
CHECK(!memcmp(before, &view, sizeof(view)));
}
struct stream_block_view view;
CHECK(stream_block_verify(&descriptor, public_key, blocks[0], 0, &view) == STREAM_ERR_FORMAT);
CHECK(stream_block_verify(&descriptor, public_key, blocks[0], 1299, &view) == STREAM_ERR_FORMAT);
CHECK(stream_block_verify(&descriptor, public_key, blocks[0], 1301, &view) == STREAM_ERR_FORMAT);
uint8_t wrong_key[32]; memcpy(wrong_key, public_key, 32); wrong_key[0] ^= 1;
CHECK(stream_block_verify(&descriptor, wrong_key, blocks[0], 1300, &view) == STREAM_ERR_SIGNATURE);
CHECK(stream_encoder_encode(encoder, UINT64_MAX, 0, 0, input, 1, blocks) == STREAM_ERR_ARGUMENT);
CHECK(stream_encoder_encode(encoder, 1, 0, 0, input, 0, blocks) == STREAM_ERR_ARGUMENT);
CHECK(stream_encoder_encode(encoder, 1, 0, 2, input, 1, blocks) == STREAM_ERR_ARGUMENT);
struct stream_descriptor invalid = descriptor; invalid.n = 0;
CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT);
invalid = descriptor; invalid.r = invalid.n;
CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT);
invalid = descriptor; invalid.block_size = 9;
CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT);
invalid = descriptor; invalid.source_epoch = 0;
CHECK(stream_descriptor_validate(&invalid) == STREAM_ERR_ARGUMENT);
stream_encoder_destroy(encoder);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] signed blocks: canonical bytes, all field classes, wrong key and malformed sizes");
}
/* Любая пара потерь при 6+2 восстанавливается из подписанных блоков, отсутствующие блоки не регенерируются. */
static void test_fec_and_padding(void) {
struct stream_encoder *encoder = NULL;
CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK);
uint8_t input[7080], output[7080]; pattern(input, sizeof(input));
const uint8_t *blocks[RS_MAX_BLOCKS];
CHECK(stream_encoder_encode(encoder, 0, 15, 0, input, sizeof(input), blocks) == STREAM_OK);
for (unsigned first = 0; first < 8; first++) {
for (unsigned second = first + 1; second < 8; second++) {
struct stream_buffer *buffer = NULL;
CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 8 * 1300) == STREAM_OK);
for (unsigned i = 0; i < 8; i++)
if (i != first && i != second) CHECK(stream_buffer_put(buffer, blocks[i], 1300) == STREAM_OK);
size_t length = 0;
memset(output, 0xcc, sizeof(output));
size_t allocations = u_get_allocated_count();
CHECK(stream_buffer_decode(buffer, 0, output, sizeof(output), &length) == STREAM_OK);
CHECK(u_get_allocated_count() == allocations && length == sizeof(input) && !memcmp(input, output, length));
const uint8_t *missing = NULL; size_t missing_length = 0; uint64_t actual = 100;
CHECK(stream_buffer_read(buffer, 0, (uint8_t)first, &missing, &missing_length, &actual) == STREAM_GAP);
CHECK(!missing && missing_length == 0 && actual == 100);
struct stream_buffer_stats stats; CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK);
CHECK(stats.blocks == 6 && stats.stored_bytes == 6 * 1300);
stream_buffer_destroy(buffer);
}
}
struct stream_buffer *buffer = NULL;
CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 8 * 1300) == STREAM_OK);
for (unsigned i = 0; i < 5; i++) CHECK(stream_buffer_put(buffer, blocks[i], 1300) == STREAM_OK);
memset(output, 0xcc, sizeof(output)); size_t length = 123;
CHECK(stream_buffer_decode(buffer, 0, output, sizeof(output), &length) == STREAM_WAIT);
CHECK(length == 123);
for (unsigned i = 0; i < sizeof(output); i++) CHECK(output[i] == 0xcc);
CHECK(stream_buffer_put(buffer, blocks[5], 1300) == STREAM_OK);
CHECK(stream_buffer_decode(buffer, 0, output, 1, &length) == STREAM_ERR_ARGUMENT);
stream_buffer_destroy(buffer);
stream_encoder_destroy(encoder);
const struct { uint8_t n, r; uint16_t size; } variants[] = {{1, 0, 1189}, {3, 1, 20}, {16, 5, 4096}, {255, 254, 10}};
for (unsigned v = 0; v < sizeof(variants) / sizeof(variants[0]); v++) {
struct stream_descriptor d = descriptor; d.n = variants[v].n; d.r = variants[v].r; d.block_size = variants[v].size;
encoder = NULL; buffer = NULL;
CHECK(stream_encoder_create(&encoder, &d, private_key) == STREAM_OK);
size_t size = stream_block_size(&d), wanted = stream_input_size(&d) > 17 ? 17 : stream_input_size(&d);
CHECK(stream_encoder_encode(encoder, 37, 0, 0, input, wanted, blocks) == STREAM_OK);
CHECK(stream_buffer_create(&buffer, &d, stream_encoder_public_key(encoder), d.n * size) == STREAM_OK);
for (unsigned i = d.r; i < d.n; i++) CHECK(stream_buffer_put(buffer, blocks[i], size) == STREAM_OK);
uint8_t partial[17]; length = 0;
CHECK(stream_buffer_decode(buffer, 37, partial, wanted, &length) == STREAM_OK);
CHECK(length == wanted && !memcmp(partial, input, wanted));
stream_buffer_destroy(buffer); stream_encoder_destroy(encoder);
}
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] FEC: all 28 double losses, insufficient set, partial data, n=1..255 and r=0");
}
/* Вставлять подписанную пачку в тестовое кольцо до повторного использования буферов encoder. */
static void put_batch(struct stream_encoder *encoder, struct stream_buffer *buffer, uint64_t seq, unsigned mask) {
uint8_t input[37]; pattern(input, sizeof(input));
const uint8_t *blocks[RS_MAX_BLOCKS];
CHECK(stream_encoder_encode(encoder, seq, (uint16_t)seq, 0, input, sizeof(input), blocks) == STREAM_OK);
for (unsigned i = 0; i < 8; i++) if (mask & (1u << i)) CHECK(stream_buffer_put(buffer, blocks[i], 1300) == STREAM_OK);
}
/* Общее кольцо хранит реальные блоки, принимает порядок 10/12/11 и вытесняет целую минимальную пачку. */
static void test_ring_and_conflicts(void) {
struct stream_encoder *encoder = NULL; struct stream_buffer *buffer = NULL;
CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK);
CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 16 * 1300) == STREAM_OK);
const uint8_t *wire = NULL; size_t length = 0; uint64_t actual = 100;
CHECK(stream_buffer_read(buffer, UINT64_MAX, 0, &wire, &length, &actual) == STREAM_WAIT);
put_batch(encoder, buffer, 10, 255); put_batch(encoder, buffer, 12, 255);
CHECK(stream_buffer_read(buffer, 10, 0, &wire, &length, &actual) == STREAM_OK);
CHECK(stream_buffer_put(buffer, wire, length) == STREAM_DUPLICATE);
uint8_t bad[1300]; memcpy(bad, wire, length); bad[STREAM_BLOCK_RS_OFF + RS_HEADER_SIZE] ^= 1;
CHECK(stream_buffer_put(buffer, bad, length) == STREAM_ERR_SIGNATURE);
CHECK(sc_ed25519_sign(private_key, bad, sizeof(bad) - 64, bad + sizeof(bad) - 64) == SC_OK);
CHECK(stream_buffer_put(buffer, bad, length) == STREAM_ERR_CONFLICT);
put_batch(encoder, buffer, 11, 1);
CHECK(stream_buffer_read(buffer, 10, 0, &wire, &length, &actual) == STREAM_EXPIRED);
CHECK(stream_buffer_read(buffer, 11, 1, &wire, &length, &actual) == STREAM_GAP);
CHECK(stream_buffer_read(buffer, 13, 0, &wire, &length, &actual) == STREAM_WAIT);
CHECK(stream_buffer_read(buffer, UINT64_MAX, 0, &wire, &length, &actual) == STREAM_OK && actual == 12);
uint64_t first = 0, last = 0;
CHECK(stream_buffer_range(buffer, 0, &first, &last) == STREAM_OK && first == 11 && last == 12);
struct stream_buffer_stats stats; CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK);
CHECK(stats.blocks == 9 && stats.batches == 2 && stats.stored_bytes == 9 * 1300 && stats.expired_before == 11);
const uint8_t *blocks[RS_MAX_BLOCKS]; uint8_t input[37]; pattern(input, sizeof(input));
CHECK(stream_encoder_encode(encoder, 10, 10, 0, input, sizeof(input), blocks) == STREAM_OK);
CHECK(stream_buffer_put(buffer, blocks[0], 1300) == STREAM_EXPIRED);
put_batch(encoder, buffer, 13, 255);
CHECK(stream_buffer_read(buffer, 11, 0, &wire, &length, &actual) == STREAM_EXPIRED);
put_batch(encoder, buffer, UINT64_MAX - 1, 1);
CHECK(stream_buffer_read(buffer, UINT64_MAX, 0, &wire, &length, &actual) == STREAM_OK && actual == UINT64_MAX - 1);
CHECK(stream_buffer_read(buffer, UINT64_MAX, 7, &wire, &length, &actual) == STREAM_OK && actual == 13);
CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK && stats.stored_bytes <= stats.capacity_bytes);
stream_buffer_destroy(buffer);
buffer = NULL;
CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 16 * 1300) == STREAM_OK);
for (unsigned seq = 0; seq < 16; seq++) put_batch(encoder, buffer, seq, 1);
CHECK(stream_buffer_get_stats(buffer, &stats) == STREAM_OK && stats.blocks == 16 && stats.batches == 16);
CHECK(stats.stored_bytes == stats.capacity_bytes); /* Ни один из семи пустых субпотоков не резервирует место. */
put_batch(encoder, buffer, 16, 1);
CHECK(stream_buffer_read(buffer, 0, 0, &wire, &length, &actual) == STREAM_EXPIRED);
CHECK(stream_buffer_range(buffer, 0, &first, &last) == STREAM_OK && first == 1 && last == 16);
stream_buffer_destroy(buffer); stream_encoder_destroy(encoder);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] ring: shared data budget, sparse batches, out-of-order indices, whole-batch eviction and conflicts");
}
/* Подлинное время источника разворачивается через ноль; постоянный lag не считается полной остановкой. */
static void test_relative_metrics(void) {
struct stream_metrics m; struct stream_metric_snapshot snapshot;
stream_metrics_init(&m);
CHECK(stream_metrics_poll(&m, 0, &snapshot) == STREAM_WAIT);
CHECK(stream_metrics_observe(&m, 100, 65520, 0, 1300, 1000) == STREAM_OK);
CHECK(stream_metrics_observe(&m, 101, 4, 0, 1300, 1020) == STREAM_OK);
CHECK(stream_metrics_poll(&m, 1020, &snapshot) == STREAM_OK);
CHECK(snapshot.lag_ms == 0 && snapshot.source_rate_bytes_sec == 65000 && snapshot.expected_interval_ms == 20);
CHECK(snapshot.receive_rate_bytes_sec == 65000 && snapshot.blocks == 2 && !snapshot.stopped);
CHECK(stream_metrics_observe(&m, 102, 24, 0, 1300, 1320) == STREAM_OK);
CHECK(stream_metrics_observe(&m, 103, 44, 0, 1300, 1620) == STREAM_OK);
CHECK(stream_metrics_poll(&m, 1620, &snapshot) == STREAM_OK);
CHECK(snapshot.lag_ms == 560 && snapshot.late_blocks == 1 && snapshot.stops == 0 && !snapshot.stopped);
uint64_t seq = 103, now = 1620; uint16_t timestamp = 44;
for (unsigned i = 0; i < 30; i++) {
seq++; now++; timestamp = (uint16_t)(timestamp + 20);
CHECK(stream_metrics_observe(&m, seq, timestamp, 0, 1300, now) == STREAM_OK);
}
CHECK(stream_metrics_poll(&m, now, &snapshot) == STREAM_OK && snapshot.lag_ms == 0);
CHECK(stream_metrics_observe(&m, seq, timestamp, 0, 1300, now + 300) == STREAM_DUPLICATE);
CHECK(stream_metrics_poll(&m, now + 521, &snapshot) == STREAM_OK);
CHECK(snapshot.stopped && snapshot.stops == 1 && snapshot.lag_ms == 501);
CHECK(stream_metrics_poll(&m, now + 800, &snapshot) == STREAM_OK && snapshot.stops == 1);
seq++; timestamp = (uint16_t)(timestamp + 20); now += 800;
CHECK(stream_metrics_observe(&m, seq, timestamp, 0, 1300, now) == STREAM_OK);
CHECK(stream_metrics_poll(&m, now, &snapshot) == STREAM_OK && !snapshot.stopped && snapshot.stops == 1);
CHECK(stream_metrics_observe(&m, seq + 1, timestamp, 0, 1300, now - 1) == STREAM_ERR_STATE);
stream_metrics_end(&m);
CHECK(stream_metrics_poll(&m, now + 70000, &snapshot) == STREAM_OK);
CHECK(!snapshot.blocks && !snapshot.late_blocks && !snapshot.stops && !snapshot.stopped && !snapshot.source_rate_valid);
stream_metrics_init(&m);
CHECK(stream_metrics_observe(&m, 0, 0, 0, 1300, 0) == STREAM_OK);
CHECK(stream_metrics_observe(&m, 1, 0, 0, 1300, 0) == STREAM_OK);
CHECK(stream_metrics_poll(&m, 0, &snapshot) == STREAM_OK && !snapshot.source_rate_valid);
CHECK(stream_metrics_observe(&m, 2, 1, 0, 1300, 1) == STREAM_OK);
CHECK(stream_metrics_poll(&m, 1, &snapshot) == STREAM_OK && snapshot.source_rate_bytes_sec == 2600000);
CHECK(stream_metrics_observe(&m, 3, 9, STREAM_BLOCK_TIME_RESET, 1300, 2) == STREAM_OK);
CHECK(stream_metrics_poll(&m, 2, &snapshot) == STREAM_OK && snapshot.lag_ms == 0);
CHECK(stream_metrics_observe(&m, 5, 10, 0, 1300, 1000) == STREAM_OK);
CHECK(stream_metrics_poll(&m, 1000, &snapshot) == STREAM_OK && snapshot.lag_ms == 0);
CHECK(stream_metrics_observe(&m, 6, 11, 0, 1300, 71000) == STREAM_OK);
CHECK(stream_metrics_poll(&m, 71000, &snapshot) == STREAM_OK && snapshot.lag_ms == 0 && !snapshot.stopped);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] metrics: timestamp wrap, zero deltas, lag/catchup, complete stall, one episode and window expiry");
}
struct source_events {
unsigned count;
uint64_t seq;
uint64_t time_tb;
uint16_t timestamp_ms;
uint8_t flags;
};
/* Callback видит уже всю пачку, а не частично опубликованные подписи. */
static void source_ready(struct stream_source *source, uint64_t seq, void *arg) {
struct source_events *events = arg;
events->count++; events->seq = seq; events->time_tb = get_time_tb();
struct stream_buffer_stats stats;
CHECK(stream_buffer_get_stats(stream_source_buffer(source), &stats) == STREAM_OK);
CHECK(stats.blocks == events->count * 8 && stats.highest_seq == seq);
const uint8_t *wire = NULL; size_t length = 0; uint64_t actual = UINT64_MAX;
CHECK(stream_buffer_read(stream_source_buffer(source), seq, 0, &wire, &length, &actual) == STREAM_OK);
CHECK(actual == seq && length == STREAM_BLOCK_DEFAULT_SIZE);
events->timestamp_ms = (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2);
events->flags = wire[STREAM_BLOCK_FLAGS_OFF];
for (unsigned sub = 1; sub < 8; sub++) {
CHECK(stream_buffer_read(stream_source_buffer(source), seq, (uint8_t)sub, &wire, &length, &actual) == STREAM_OK);
CHECK(stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2) == events->timestamp_ms);
CHECK(wire[STREAM_BLOCK_FLAGS_OFF] == events->flags);
}
}
/* Настоящий uasync timer: не сбрасывает до 100 мс, полный блок отменяет timer, destroy отменяет остаток. */
static void test_source_timer(void) {
test_clock_init(50);
struct UASYNC *ua = uasync_create(); CHECK(ua != NULL);
struct stream_source *source = NULL; struct source_events events = {0};
CHECK(stream_source_create(&source, ua, &descriptor, private_key, source_ready, &events) == STREAM_OK);
uint8_t input[7080], output[7080]; pattern(input, sizeof(input));
CHECK(stream_source_input_capacity(source) == sizeof(input));
size_t accepted = 999;
CHECK(stream_source_write(source, NULL, 0, &accepted) == STREAM_OK && accepted == 0);
uint64_t start = get_time_tb();
CHECK(stream_source_write(source, input, 17, &accepted) == STREAM_OK && accepted == 17);
test_clock_run_for(ua, 500);
CHECK(events.count == 0);
test_clock_next_timer(ua);
CHECK(events.count == 1 && events.seq == 0 && events.time_tb - start >= STREAM_SOURCE_FLUSH_TB);
CHECK((uint16_t)(events.timestamp_ms - (uint16_t)(start / 10)) >= 100 && !events.flags);
size_t length = 0;
CHECK(stream_buffer_decode(stream_source_buffer(source), 0, output, 17, &length) == STREAM_OK);
CHECK(length == 17 && !memcmp(input, output, length));
CHECK(stream_source_write(source, input, sizeof(input), &accepted) == STREAM_OK && accepted == sizeof(input));
CHECK(events.count == 2 && events.seq == 1);
test_clock_run_for(ua, 1100);
CHECK(events.count == 2); /* Нет лишней пустой пачки. */
CHECK(stream_source_write(source, input, 13, &accepted) == STREAM_OK && accepted == 13);
CHECK(stream_source_write(source, input + 13, sizeof(input) - 13, &accepted) == STREAM_OK);
CHECK(events.count == 3 && events.seq == 2);
test_clock_run_for(ua, 1100);
CHECK(events.count == 3); /* Timer неполной пачки отменён заполнением. */
CHECK(stream_buffer_decode(stream_source_buffer(source), 2, output, sizeof(output), &length) == STREAM_OK);
CHECK(length == sizeof(input) && !memcmp(input, output, length));
test_clock_run_for(ua, STREAM_TIMESTAMP_PERIOD_MS * 10);
CHECK(stream_source_write(source, input, 23, &accepted) == STREAM_OK);
CHECK(stream_source_finish(source) == STREAM_OK && events.count == 4);
CHECK(events.flags == STREAM_BLOCK_TIME_RESET); /* Источник явно пометил неоднозначное число оборотов. */
CHECK(stream_source_finish(source) == STREAM_OK && events.count == 4);
CHECK(stream_source_write(source, input, 1, &accepted) == STREAM_ERR_STATE && accepted == 0);
stream_source_destroy(source);
source = NULL; events.count = 0;
CHECK(stream_source_create(&source, ua, &descriptor, private_key, source_ready, &events) == STREAM_OK);
CHECK(stream_source_write(source, input, 17, &accepted) == STREAM_OK);
stream_source_destroy(source);
test_clock_run_for(ua, 1100);
CHECK(events.count == 0);
uasync_destroy(ua, 1); test_clock_finish();
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] source: 100ms timer, immediate full batch, padding, finish and cancellation on destroy");
}
int main(void) {
debug_config_init();
debug_set_level(DEBUG_LEVEL_ERROR);
debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_DISABLED);
CHECK(get_category_by_name("stream") == DEBUG_CATEGORY_STREAM);
for (unsigned i = 0; i < sizeof(private_key); i++) private_key[i] = (uint8_t)(i * 37 + 11);
size_t allocations = u_get_allocated_count();
test_signed_blocks(); CHECK(u_get_allocated_count() == allocations);
test_fec_and_padding(); CHECK(u_get_allocated_count() == allocations);
test_ring_and_conflicts(); CHECK(u_get_allocated_count() == allocations);
test_relative_metrics(); CHECK(u_get_allocated_count() == allocations);
test_source_timer(); CHECK(u_get_allocated_count() == allocations);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] stream core: all checks, no tracked allocations remain");
return 0;
}
Loading…
Cancel
Save