From f5d112a712c2c09d96b13210ba917943a92cbe82 Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 14 Oct 2026 02:32:14 +0200 Subject: [PATCH] Share bounded media publication and send telemetry across frontends --- src/Makefile.am | 2 + src/stream/stream_publisher.c | 301 ++++++++++++++++++++++++++++++++++ src/stream/stream_publisher.h | 32 ++++ 3 files changed, 335 insertions(+) create mode 100644 src/stream/stream_publisher.c create mode 100644 src/stream/stream_publisher.h diff --git a/src/Makefile.am b/src/Makefile.am index 02e5b7b1..92ea30bb 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -19,6 +19,8 @@ stream_core_files = \ stream/stream_proto.h \ stream/stream_media.h \ stream/stream_media.c \ + stream/stream_publisher.h \ + stream/stream_publisher.c \ stream/stream_media_buffer.h \ stream/stream_media_buffer.c \ stream/stream_playback_clock.h \ diff --git a/src/stream/stream_publisher.c b/src/stream/stream_publisher.c new file mode 100644 index 00000000..c4399463 --- /dev/null +++ b/src/stream/stream_publisher.c @@ -0,0 +1,301 @@ +#include +#include +#include "../../lib/mem.h" +#include "../../lib/debug_config.h" +#include "../../lib/u_async.h" +#include "../utun_instance.h" +#include "stream.h" +#include "stream_network.h" +#include "stream_publisher.h" + +#define TX_LIMIT (2u * 1024u * 1024u) +#define TX_RECORDS 128 +#define MEDIA_AGE_US 250000u + +struct publisher_record { + struct publisher_record *next; + uint64_t pts, frame_id, created; + size_t length, offset; + int key; + uint8_t data[]; +}; +struct publisher_queue { struct publisher_record *head, *tail; size_t bytes; unsigned count; }; +struct stream_publisher { + pthread_mutex_t lock; + struct UTUN_INSTANCE *inst; + struct STREAM_HANDLE *handle; + struct posted_task *post; + void *timer, *stats_timer; + struct stream_media_format format; + struct stream_publisher_status status; + struct publisher_queue video, audio; + uint64_t group, next_frame, last_pts[2], drain_until, report_us; + int seen[2], stopping; +}; +static void publisher_process(void *arg); + +/* Очередь producer/actor защищена mutex; LL_QUEUE между потоками не применяется. */ +static void record_append(struct publisher_queue *q, struct publisher_record *r) { + r->next = NULL; + if (q->tail) q->tail->next = r; else q->head = r; + q->tail = r; q->bytes += r->length; q->count++; +} +static struct publisher_record *record_take(struct publisher_queue *q) { + struct publisher_record *r = q->head; + if (!r) return NULL; + q->head = r->next; if (!q->head) q->tail = NULL; + q->bytes -= r->length; q->count--; return r; +} +static void records_clear(struct publisher_queue *q) { + struct publisher_record *r; + while ((r = record_take(q))) u_free(r); +} + +/* Под lock; резервированный post не может потерять принятую команду. */ +static int publisher_schedule(struct stream_publisher *p) { + if (p->post || p->timer) return 0; + if (!p->inst || !uasync_is_running(p->inst->ua)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher schedule: core stopped"); return -1; + } + p->post = u_calloc(1, sizeof(*p->post)); + if (!p->post) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher post allocation failed"); return -1; } + p->post->callback = publisher_process; p->post->arg = p; + uasync_post_reserved(p->inst->ua, p->post); return 0; +} + +/* Только uasync под lock; терминальный снимок остаётся доступен до следующего open. */ +static void publisher_close(struct stream_publisher *p, int reason) { + if (p->timer) uasync_cancel_timeout(p->inst->ua, p->timer); + if (p->stats_timer) uasync_cancel_timeout(p->inst->ua, p->stats_timer); + p->timer = p->stats_timer = NULL; + if (p->handle) stream_close(p->handle); + p->handle = NULL; records_clear(&p->video); records_clear(&p->audio); + p->status.state = reason ? STREAM_PUBLISHER_FAILED : STREAM_PUBLISHER_ENDED; p->status.reason = reason; + p->status.video_bytes = p->status.audio_bytes = p->status.total_bytes = 0; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "publisher closed id=%llu epoch=%llx reason=%d", + (unsigned long long)p->status.id, (unsigned long long)p->status.epoch, reason); +} + +/* Один снимок всех соединений, без повторного учёта input_send_q внутри inflight. */ +static void publisher_stats(void *arg) { + struct stream_publisher *p = arg; + pthread_mutex_lock(&p->lock); p->stats_timer = NULL; + if (p->handle && stream_handle_active(p->handle)) { + uint64_t now = get_time_us(); struct etcp_tx_snapshot transport; struct stream_info info; + if (etcp_tx_get_snapshot(p->inst, now, &transport) == 0 && stream_get_info(p->handle, &info) == STREAM_OK) { + p->status.sample_us = now; p->status.video_bytes = p->video.bytes - (p->video.head ? p->video.head->offset : 0); + p->status.audio_bytes = p->audio.bytes; p->status.stream_bytes = stream_network_pending_bytes(p->inst, &info.descriptor); + p->status.transport_bytes = transport.queued_bytes; p->status.inflight_bytes = transport.inflight_bytes; + p->status.acked_1s = transport.acked_1s; p->status.connections = transport.connections; + p->status.total_bytes = p->status.video_bytes + p->status.audio_bytes + p->status.stream_bytes + + p->status.transport_bytes + p->status.inflight_bytes; + if (now - p->report_us >= 1000000) { + p->report_us = now; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, + "publisher TX id=%llu video=%llu audio=%llu RS=%llu ETCP=%llu inflight=%llu total=%llu ACK_1s=%llu conns=%u KEY=%d", + (unsigned long long)p->status.id, (unsigned long long)p->status.video_bytes, (unsigned long long)p->status.audio_bytes, + (unsigned long long)p->status.stream_bytes, (unsigned long long)p->status.transport_bytes, + (unsigned long long)p->status.inflight_bytes, (unsigned long long)p->status.total_bytes, + (unsigned long long)p->status.acked_1s, p->status.connections, p->status.need_key); + } + } + p->stats_timer = uasync_set_timeout(p->inst->ua, 1000, p, publisher_stats, "publisher_stats"); + if (!p->stats_timer) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher stats timer allocation failed"); publisher_close(p, STREAM_CLOSE_FAILED); } + } else if (p->handle) publisher_close(p, stream_handle_reason(p->handle)); + pthread_mutex_unlock(&p->lock); +} + +/* Упаковка общая для обеих платформ: Opus целиком, видео с ограниченными продолжениями. */ +static int publisher_block(struct stream_publisher *p) { + struct stream_media_block block = {.format = p->format}; + uint8_t wire[STREAM_DEFAULT_INPUT_CAPACITY]; + size_t capacity = stream_publish_capacity(p->handle), used = STREAM_MEDIA_HEADER_SIZE; + if (capacity > sizeof(wire) || capacity < 256) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher invalid capacity=%zu", capacity); return STREAM_ERR_STATE; + } + for (struct publisher_record *r = p->audio.head; r && block.audio_count < 8; r = r->next) { + if (used + STREAM_MEDIA_AUDIO_SIZE + r->length + STREAM_MEDIA_VIDEO_SIZE >= capacity) break; + block.audio[block.audio_count++] = (struct stream_media_audio){r->pts, (uint32_t)r->length, 960, r->data}; + used += STREAM_MEDIA_AUDIO_SIZE + r->length; + } + for (struct publisher_record *r = p->video.head; r && block.video_count < STREAM_MEDIA_MAX_ENTRIES; r = r->next) { + if (used + STREAM_MEDIA_VIDEO_SIZE >= capacity) break; + size_t take = capacity - used - STREAM_MEDIA_VIDEO_SIZE; + if (take > r->length - r->offset) take = r->length - r->offset; + block.video[block.video_count++] = (struct stream_media_video){r->frame_id, r->pts, (uint32_t)r->length, + (uint32_t)r->offset, (uint32_t)take, (uint8_t)r->key, r->data + r->offset}; + used += STREAM_MEDIA_VIDEO_SIZE + take; + if (take < r->length - r->offset) break; + } + if (!block.video_count && !block.audio_count) return STREAM_WAIT; + size_t length = 0; + int result = stream_media_encode(&block, wire, capacity, &length); + if (result == STREAM_OK) result = stream_publish_block(p->handle, wire, length); + if (result != STREAM_OK) return result; + for (unsigned i = 0; i < block.audio_count; i++) u_free(record_take(&p->audio)); + for (unsigned i = 0; i < block.video_count; i++) { + p->video.head->offset += block.video[i].length; + if (p->video.head->offset == p->video.head->length) u_free(record_take(&p->video)); + } + return STREAM_OK; +} + +/* Не более двух RS-пачек за проход; таймер продолжает работу, не занимая сетевой поток. */ +static void publisher_process(void *arg) { + struct stream_publisher *p = arg; + pthread_mutex_lock(&p->lock); p->post = NULL; p->timer = NULL; + if (p->status.state == STREAM_PUBLISHER_OPENING) { + int result = p->stopping ? STREAM_ERR_STATE : stream_publish_open(p->inst, p->group, NULL, &p->handle); + if (result == STREAM_OK) result = stream_media_enable(p->handle, 0); + struct stream_info info; + if (result == STREAM_OK) result = stream_get_info(p->handle, &info); + if (result != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher open failed id=%llu group=%llx result=%d", + (unsigned long long)p->status.id, (unsigned long long)p->group, result); + publisher_close(p, p->stopping ? 0 : STREAM_CLOSE_FAILED); + } else { + p->status.epoch = info.descriptor.source_epoch; p->status.state = STREAM_PUBLISHER_LIVE; + p->stats_timer = uasync_set_timeout(p->inst->ua, 1000, p, publisher_stats, "publisher_stats"); + if (!p->stats_timer) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher stats start failed"); publisher_close(p, STREAM_CLOSE_FAILED); } + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "publisher opened id=%llu group=%llx epoch=%llx codec=%u size=%ux%u audio=%u", + (unsigned long long)p->status.id, (unsigned long long)p->group, (unsigned long long)p->status.epoch, + p->format.video_codec, p->format.width, p->format.height, p->format.audio_codec); + } + } + uint64_t now = get_time_us(); + if (p->handle && !stream_handle_active(p->handle)) publisher_close(p, stream_handle_reason(p->handle)); + if (p->video.head && now - p->video.head->created > MEDIA_AGE_US) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "publisher video expired bytes=%zu records=%u; require KEY", p->video.bytes, p->video.count); + records_clear(&p->video); p->status.need_key = 1; + } + while (p->audio.head && now - p->audio.head->created > MEDIA_AGE_US) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "publisher audio expired pts=%llu", (unsigned long long)p->audio.head->pts); + u_free(record_take(&p->audio)); + } + if (p->handle && p->status.state != STREAM_PUBLISHER_DRAINING) { + for (unsigned i = 0; i < 2 && (p->video.head || p->audio.head); i++) { + size_t pending = p->video.bytes + p->audio.bytes - (p->video.head ? p->video.head->offset : 0); + uint64_t oldest = p->video.head ? p->video.head->created : p->audio.head->created; + if (p->audio.head && p->audio.head->created < oldest) oldest = p->audio.head->created; + int immediate = p->video.head && p->video.head->key; + if (!p->stopping && !immediate && pending + STREAM_MEDIA_HEADER_SIZE + STREAM_MEDIA_VIDEO_SIZE < stream_publish_capacity(p->handle) + && now - oldest < 20000) break; + int result = publisher_block(p); + if (result < 0) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher block failed result=%d", result); publisher_close(p, STREAM_CLOSE_FAILED); break; } + } + if (p->handle && p->stopping && !p->video.head && !p->audio.head) { + if (stream_publish_finish(p->handle) != STREAM_OK) publisher_close(p, STREAM_CLOSE_FAILED); + else { p->status.state = STREAM_PUBLISHER_DRAINING; p->drain_until = now + 2000000; } + } + } + if (p->handle && p->status.state == STREAM_PUBLISHER_DRAINING && now >= p->drain_until) publisher_close(p, 0); + if (p->handle && (p->video.head || p->audio.head || p->status.state == STREAM_PUBLISHER_DRAINING)) { + p->timer = uasync_set_timeout(p->inst->ua, p->status.state == STREAM_PUBLISHER_DRAINING ? 200 : 50, + p, publisher_process, "publisher_media"); + if (!p->timer) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher media timer allocation failed"); publisher_close(p, STREAM_CLOSE_FAILED); } + } + pthread_mutex_unlock(&p->lock); +} + +/* Контекст принадлежит bridge, который снимает producers перед destroy. */ +int stream_publisher_create(struct stream_publisher **out, struct UTUN_INSTANCE *inst) { + if (!out || *out || !inst || !inst->stream) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher create invalid owner"); return -1; } + struct stream_publisher *p = u_calloc(1, sizeof(*p)); + if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher allocation failed"); return -1; } + if (pthread_mutex_init(&p->lock, NULL) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher mutex failed"); u_free(p); return -1; } + p->inst = inst; *out = p; return 0; +} +void stream_publisher_destroy(struct stream_publisher *p) { + if (!p) return; + pthread_mutex_lock(&p->lock); + if (p->post) uasync_cancel_post(p->inst->ua, p->post); + publisher_close(p, STREAM_CLOSE_SERVICE); p->inst = NULL; + pthread_mutex_unlock(&p->lock); pthread_mutex_destroy(&p->lock); u_free(p); +} + +/* Wire формат не допускает смену audio codec внутри одной эпохи. */ +static int format_valid(const struct stream_media_format *f) { + return f && f->revision && f->width && f->width <= 4096 && f->height && f->height <= 4096 + && f->rotation <= 270 && f->rotation % 90 == 0 && f->video_codec >= STREAM_MEDIA_H264 + && f->video_codec <= STREAM_MEDIA_H266 && f->audio_codec <= STREAM_MEDIA_OPUS_STEREO; +} +int stream_publisher_open(struct stream_publisher *p, uint64_t id, uint64_t group, const struct stream_media_format *f) { + if (!p || !id || !group || !format_valid(f)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher open invalid configuration"); return -1; } + pthread_mutex_lock(&p->lock); int result = -1; + if (p->handle || p->post || p->status.state == STREAM_PUBLISHER_OPENING) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "publisher open busy id=%llu owner=%llu", (unsigned long long)id, (unsigned long long)p->status.id); + } else if (publisher_schedule(p) == 0) { + p->status = (struct stream_publisher_status){.id = id, .state = STREAM_PUBLISHER_OPENING, .need_key = 1}; + p->group = group; p->format = *f; p->next_frame = 0; p->stopping = 0; memset(p->seen, 0, sizeof(p->seen)); result = 0; + } + pthread_mutex_unlock(&p->lock); return result; +} + +/* Admission сохраняет PTS только принятого кадра; новая revision коммитится атомарно с KEY. */ +int stream_publisher_feed(struct stream_publisher *p, uint64_t id, int video, uint64_t pts, int key, + const uint8_t *data, size_t length, const struct stream_media_format *f) { + if (!p || !data || !length || pts > INT64_MAX || length > (video ? STREAM_MEDIA_MAX_FRAME : STREAM_MEDIA_MAX_OPUS)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher feed invalid data length=%zu", length); return -1; + } + pthread_mutex_lock(&p->lock); int result = -1, track = video ? 0 : 1, change = 0; + if (p->status.id != id || p->stopping || (p->status.state != STREAM_PUBLISHER_OPENING && p->status.state != STREAM_PUBLISHER_LIVE)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "publisher feed stale/stopped id=%llu owner=%llu state=%d", + (unsigned long long)id, (unsigned long long)p->status.id, p->status.state); goto done; + } + if (video) { + if (!format_valid(f) || f->audio_codec != p->format.audio_codec) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher invalid video format/audio codec"); goto done; + } + change = f->revision > p->format.revision; + if (f->revision < p->format.revision || (change && !key)) { result = 0; goto done; } + if (!change && (f->width != p->format.width || f->height != p->format.height + || f->rotation != p->format.rotation || f->video_codec != p->format.video_codec)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher changed format without revision=%u", f->revision); goto done; + } + } + if ((!video && !p->format.audio_codec) || (p->seen[track] && pts <= p->last_pts[track])) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher invalid track/PTS pts=%llu", (unsigned long long)pts); goto done; + } + if (video && !change && (p->video.bytes + p->audio.bytes + length > TX_LIMIT || p->video.count + p->audio.count >= TX_RECORDS)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "publisher TX overload video=%zu audio=%zu; require KEY", p->video.bytes, p->audio.bytes); + records_clear(&p->video); p->status.need_key = 1; + } + if (video && p->status.need_key && !key) { result = 0; goto done; } + if ((change ? 0 : p->video.bytes) + p->audio.bytes + length > TX_LIMIT + || (change ? 0 : p->video.count) + p->audio.count >= TX_RECORDS) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "publisher admission refused video=%d bytes=%zu", video, length); result = 0; goto done; + } + struct publisher_record *r = u_calloc(1, sizeof(*r) + length); + if (!r) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher record allocation failed bytes=%zu", length); goto done; } + if (publisher_schedule(p) < 0) { u_free(r); goto done; } + if (change) { + records_clear(&p->video); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "publisher revision committed id=%llu %u->%u codec=%u size=%ux%u rotation=%u", + (unsigned long long)id, p->format.revision, f->revision, f->video_codec, f->width, f->height, f->rotation); + p->format = *f; + } + r->length = length; r->pts = pts; r->created = get_time_us(); r->key = key ? STREAM_MEDIA_KEY : 0; + if (video) { r->frame_id = ++p->next_frame; p->status.need_key = 0; } + memcpy(r->data, data, length); record_append(video ? &p->video : &p->audio, r); + p->seen[track] = 1; p->last_pts[track] = pts; result = 1; +done: + pthread_mutex_unlock(&p->lock); return result; +} + +/* Stop идемпотентен; принятые кадры дренируются actor перед FIN. */ +void stream_publisher_stop(struct stream_publisher *p, uint64_t id) { + if (!p) return; + pthread_mutex_lock(&p->lock); + if (p->status.id == id && p->status.state < STREAM_PUBLISHER_DRAINING) { + p->stopping = 1; + if (p->status.state != STREAM_PUBLISHER_OPENING) p->status.state = STREAM_PUBLISHER_STOPPING; + if (publisher_schedule(p) < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher stop scheduling failed"); + } + pthread_mutex_unlock(&p->lock); +} +int stream_publisher_snapshot(struct stream_publisher *p, uint64_t id, struct stream_publisher_status *out) { + if (!p || !out) return -1; + pthread_mutex_lock(&p->lock); int result = p->status.id == id && id ? 0 : -1; + if (!result) *out = p->status; + pthread_mutex_unlock(&p->lock); return result; +} diff --git a/src/stream/stream_publisher.h b/src/stream/stream_publisher.h new file mode 100644 index 00000000..f8050793 --- /dev/null +++ b/src/stream/stream_publisher.h @@ -0,0 +1,32 @@ +/* Общий publisher Android/desktop: принимает полные access units и Opus, публикует uAV2/RS. + * create/destroy выполняются в uasync вне callbacks, внешний bridge сериализует destroy с producers. + * open/feed/stop/snapshot потокобезопасны; вызывающие используют session ID, не pointers ядра. + * Один publisher на контекст. Snapshot содержит состояние и 100ms телеметрию, callbacks не устанавливаются. + * TX ограничен 2MiB/128 records/250ms; потеря видео требует KEY. Stop завершает очередь и держит FIN 2s. */ +#ifndef STREAM_PUBLISHER_H +#define STREAM_PUBLISHER_H +#include "stream_media.h" +#ifdef __cplusplus +extern "C" { +#endif +struct UTUN_INSTANCE; +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 stream_publisher_create(struct stream_publisher **out, struct UTUN_INSTANCE *inst); +void stream_publisher_destroy(struct stream_publisher *publisher); +int stream_publisher_open(struct stream_publisher *publisher, uint64_t id, uint64_t group, const struct stream_media_format *format); +/* 1 принято, 0 требуется KEY/перегрузка, -1 ошибка или старый/закрытый session ID. */ +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); +void stream_publisher_stop(struct stream_publisher *publisher, uint64_t id); +int stream_publisher_snapshot(struct stream_publisher *publisher, uint64_t id, struct stream_publisher_status *status); +#ifdef __cplusplus +} +#endif +#endif