From 8533dfc30375bb329870fe4363f9bd718e4887a0 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 26 Oct 2026 12:45:31 +0100 Subject: [PATCH] Clarify asynchronous publisher and viewer session contracts --- src/stream/stream_publisher.h | 32 ++++++++++++++++++++++++++------ src/stream/stream_viewer.h | 30 ++++++++++++++++++++---------- 2 files changed, 46 insertions(+), 16 deletions(-) diff --git a/src/stream/stream_publisher.h b/src/stream/stream_publisher.h index 8cef3770..7ee134b4 100644 --- a/src/stream/stream_publisher.h +++ b/src/stream/stream_publisher.h @@ -18,23 +18,43 @@ struct stream_publisher; enum stream_publisher_state { STREAM_PUBLISHER_OPENING = 1, STREAM_PUBLISHER_LIVE, STREAM_PUBLISHER_STOPPING, STREAM_PUBLISHER_DRAINING, STREAM_PUBLISHER_ENDED, STREAM_PUBLISHER_FAILED }; struct stream_publisher_status { - uint64_t id, epoch, sample_us, video_bytes, audio_bytes, stream_bytes, transport_bytes, inflight_bytes, acked_1s, total_bytes; - unsigned connections; - int state, reason, need_key; - int compressor_enabled, compressor_max_gain_db, compressor_rise_rate; - int audio_enabled; + uint64_t id, epoch, // Session ID caller и эпоха публикации; epoch=0 до успешного открытия + sample_us, // Монотонное время последнего 100-мс снимка очередей + video_bytes, // Непереданный остаток очереди access units этого publisher + audio_bytes, // Байты целых Opus-пакетов его очереди + stream_bytes, // Логические RS-байты ожидающих выдач этого запуска + transport_bytes, // ETCP-очереди всех соединений instance, включая другие сервисы + inflight_bytes, // Их же неподтверждённые байты + acked_1s, // Их же подтверждённые за секунду байты + total_bytes; // Сумма video/audio/stream/transport/inflight для диагностики + unsigned connections; // Число соединений в общем ETCP-снимке instance + int state, reason, need_key; // stream_publisher_state, stream_close_reason либо 0, запрос нового KEY + int compressor_enabled, compressor_max_gain_db, compressor_rise_rate; // Снимок общих настроек компрессора + int audio_enabled; // Последнее применённое actor состояние; codec эпохи от mute не меняется struct stream_buffer_positions buffer; // Позиции получателей в кольце; обновление раз в 500 мс }; +/* out=NULL, inst с работающим stream-контекстом; 0/-1. Inst/uasync живут до destroy. */ int stream_publisher_create(struct stream_publisher **out, struct UTUN_INSTANCE *inst); +/* Снять producers, вызвать в uasync вне stream callbacks: отменить actor/таймеры и освободить publisher; NULL допустим. */ void stream_publisher_destroy(struct stream_publisher *publisher); +/* Ненулевые id/group и корректный format; 0 — поставлен OPENING, -1 — аргументы/busy/scheduling. + * FEC берётся из настроек actor при фактическом open, результат читается snapshot. + * После ENDED/FAILED разрешён новый open; снимок предыдущего ID заменяется. */ int stream_publisher_open(struct stream_publisher *publisher, uint64_t id, uint64_t group, const struct stream_media_format *format); -/* Producer заглушает свою копию PCM; actor публикует состояние без смены аудиотрека/эпохи. */ +/* enabled=0/1; producer заглушает свою копию PCM, actor публикует состояние без смены аудиотрека/эпохи. + * 0 — команда принята в OPENING/LIVE; -1 — старый/закрытый ID, включение без audio codec или scheduling. */ int stream_publisher_set_audio_enabled(struct stream_publisher *publisher, uint64_t id, int enabled); /* 1 скопировано в очередь, 0 требуется KEY/перегрузка, -1 ошибка или старый/закрытый session ID. + * pts — микросекунды, <=INT64_MAX, строго растёт отдельно для video/audio; data заимствован только на вызов. + * video!=0 — полный access unit до MAX_FRAME, format обязателен; иначе Opus до MAX_OPUS, format/key игнорируются. + * Новый video revision допускается лишь с KEY, атомарно заменяя очередь видео; audio codec эпохи неизменен. * Actor определяет ёмкость при open; слишком крупный Opus удаляет с WARN. Принятые записи также ограничены TTL очереди. */ int stream_publisher_feed(struct stream_publisher *publisher, uint64_t id, int video, uint64_t pts, int key, const uint8_t *data, size_t length, const struct stream_media_format *format); +/* Асинхронно запретить feed, дренировать очередь с её TTL, подписать END и держать owner ещё 2 с. + * В OPENING отменяет публикацию; повтор/старый ID допустим. После закрытия publisher кольцо живёт при других owners. */ void stream_publisher_stop(struct stream_publisher *publisher, uint64_t id); +/* Скопировать последний снимок, включая ENDED/FAILED: 0/-1. Ошибка не меняет status. */ int stream_publisher_snapshot(struct stream_publisher *publisher, uint64_t id, struct stream_publisher_status *status); #ifdef __cplusplus } diff --git a/src/stream/stream_viewer.h b/src/stream/stream_viewer.h index 470bda67..ad1ed723 100644 --- a/src/stream/stream_viewer.h +++ b/src/stream/stream_viewer.h @@ -1,6 +1,7 @@ /* Общие viewers desktop/Android: независимый media cursor и ограниченный батч на session ID. * create/destroy — uasync вне callbacks; внешний bridge защищает destroy от параллельных callers. - * Остальные API потокобезопасны. Один manager занимает stream event callback ядра. + * Остальные API потокобезопасны. Один manager занимает stream event callback ядра, заменяя прежний callback; + * другие владельцы не должны менять его до destroy manager. Instance/uasync живут до destroy. * stop снимает только своего reader; медленные/остановленные viewers не задерживают остальных. * pull передаёт владение цепочкой records вызывающему: освобождать records_free(). Кольца RS остаются общими. */ #ifndef STREAM_VIEWER_H @@ -24,19 +25,24 @@ enum stream_viewer_kind { }; struct stream_viewer_record { struct stream_viewer_record *next; /* Следующая запись батча; NULL завершает цепочку. */ - enum stream_viewer_kind kind; - struct stream_media_format format; + enum stream_viewer_kind kind; /* Тип данных/управляющего события. */ + struct stream_media_format format; /* Формат только VIDEO/AUDIO; у событий нули. */ uint64_t pts; /* PTS только этого источника, микросекунды. */ size_t length; /* Длина owned payload после структуры. */ int flags, reason; /* VIDEO: KEY; CLOSED: stream_close_reason. */ - uint8_t data[]; + uint8_t data[]; /* Полный access unit либо один Opus; у событий length=0. */ }; /* 0 — успех, -1 — ошибка; out заранее NULL. */ int stream_viewers_create(struct stream_viewers **out, struct UTUN_INSTANCE *inst); +/* После остановки внешних callers отменить actor, снять callback, закрыть все readers; NULL допустим. + * Выполняется в uasync вне callbacks; notify здесь вызывается без lock при исключительном владении manager. */ void stream_viewers_destroy(struct stream_viewers *viewers); -/* Уникальный ненулевой ID принадлежит caller до stop, включая завершённый/ошибочный сеанс. - * Открытие асинхронно: OPENED/CLOSED возвращается через pull. tracks — VIDEO или VIDEO|AUDIO. */ +/* Уникальный ненулевой ID занят до обработки stop actor, включая завершённый/ошибочный сеанс. + * group/source/epoch ненулевые; tracks — VIDEO либо VIDEO|AUDIO. 0 — открытие поставлено, -1 — отказ. + * Открытие асинхронно: OPENED/CLOSED возвращается через pull без предварительного request. */ int stream_viewer_open(struct stream_viewers *viewers, uint64_t id, uint64_t group, uint64_t source, uint64_t epoch, unsigned tracks); +/* Асинхронно освободить reader/батч; синхронно разбудить и снять notify. Старый ID/NULL допустимы. + * После возврата ID ещё может быть занят до исполнения actor; уже полученные records остаются у caller. */ void stream_viewer_stop(struct stream_viewers *viewers, uint64_t id); /* Owned батч или NULL. Выдача батча погашает demand, включая повторный request до pull. * NULL сохраняет ожидающий запрос. Следующие данные требуют нового request; idle не копирует кольцо. @@ -46,19 +52,23 @@ struct stream_viewer_record *stream_viewer_pull(struct stream_viewers *viewers, /* Освобождает всю owned цепочку и payload каждой записи; NULL допустим. */ void stream_viewer_records_free(struct stream_viewer_record *records); /* Позиция и audio_ahead относятся только к этому reader; повторный demand заменяет прежний. - * video — максимум видеокадров; records — максимум всех записей, 1..STREAM_VIEWER_BATCH_RECORDS. + * position<0 прогревает звук перед KEY; audio_ahead=-1 отключает выдачу звука, иначе 0..60000000 мкс. + * video — максимум видеокадров, 0..records; records — максимум всех записей, 1..STREAM_VIEWER_BATCH_RECORDS. * Батч отдаётся сразу при WAIT/MORE или после бюджета 2ms/256KiB, без ожидания заполнения. * Последняя запись может пересечь порог байтов, но не превышает STREAM_MEDIA_MAX_FRAME. * 0 — принято, -1 — неизвестный/закрытый ID или ошибка scheduling. */ int stream_viewer_request(struct stream_viewers *viewers, uint64_t id, int64_t position, unsigned video, int64_t audio_ahead, unsigned records); -/* Уведомление о готовом батче/закрытии, под lock manager: только разбудить worker, API ядра не вызывать. +/* Уведомление о готовом батче/закрытии, под lock manager (кроме destroy): только разбудить worker, + * не обращаться повторно к manager или API ядра. * arg живёт до stop/destroy или снятия callback (notify=NULL); после возврата старый callback уже не выполняется. * Уже готовый батч уведомляет сразу. 0 — установлен, -1 — ID отсутствует/остановлен. */ int stream_viewer_set_notify(struct stream_viewers *viewers, uint64_t id, void (*notify)(void *), void *arg); -/* Последний снимок сохраняется после END до stop. 0 — получен, -1 — ID отсутствует. */ +/* Последний снимок сохраняется после END/CLOSED до stop; до первого обновления actor и после resync он нулевой. + * 0 — получен, -1 — аргументы/ID отсутствует/остановлен, выход не меняется. */ int stream_viewer_status(struct stream_viewers *viewers, uint64_t id, struct stream_media_status *status); -/* Последний снимок цепочки; чтение раз в 200ms запрашивает обновление в uasync даже при остановке DATA. */ +/* Последний снимок цепочки с актуализированным возрастом, до первого снимка нули; 0/-1, ошибка не меняет выход. + * Чтение раз в 200ms запрашивает обновление в uasync даже при остановке DATA; текущий вызов его не ждёт. */ int stream_viewer_get_telemetry(struct stream_viewers *viewers, uint64_t id, struct stream_viewer_telemetry *telemetry); /* Очистить ответ и искать ближайший KEY только этого reader. 0 — принято, -1 — отказ. */ int stream_viewer_resync(struct stream_viewers *viewers, uint64_t id);