diff --git a/tools/chatgui/tests/test_stream_view.cpp b/tools/chatgui/tests/test_stream_view.cpp index dc4c8678..a89a6411 100644 --- a/tools/chatgui/tests/test_stream_view.cpp +++ b/tools/chatgui/tests/test_stream_view.cpp @@ -346,6 +346,30 @@ private slots: stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u); } + void independentReaders() { + QVERIFY(gui_stream_open(201, group, inst->node_id, epoch)); + QVERIFY(gui_stream_open(202, group, inst->node_id, epoch)); + StreamRecord a, b; + QVERIFY(receive(201, a)); QCOMPARE(a.kind, StreamRecordKind::Opened); + QVERIFY(receive(202, b)); QCOMPARE(b.kind, StreamRecordKind::Opened); + stream_info info{}; QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 3u); + QVERIFY(receive(201, a)); QCOMPARE(a.kind, StreamRecordKind::Video); + QTest::qWait(30); // Reader 202 ещё не запросил media и не тормозит 201. + QVERIFY(receive(202, b)); QCOMPARE(b.kind, StreamRecordKind::Video); + QCOMPARE(a.pts, b.pts); QCOMPARE(a.data, b.data); + stream_media_status before{}, after{}; + QVERIFY(gui_stream_status(202, before)); + QVERIFY(gui_stream_resync(201)); QTest::qWait(20); + QVERIFY(gui_stream_status(202, after)); + QCOMPARE(after.generation, before.generation); QCOMPARE(after.start_us, before.start_us); + gui_stream_stop(201); QTest::qWait(20); + QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 2u); + const QByteArray nextFrame(1500, 'm'); publish(nextFrame); + QVERIFY(receive(202, b)); QCOMPARE(b.kind, StreamRecordKind::Video); QCOMPARE(b.data, nextFrame); + gui_stream_stop(202); QTest::qWait(20); + QCOMPARE(stream_get_info(publisher, &info), STREAM_OK); QCOMPARE(info.owners, 1u); + } + void realH264RenderingAndClose() { #ifdef HAVE_FFMPEG /* Фикстура протокола выше содержит фиктивный H264; реальный decoder получает отдельную эпоху. */ diff --git a/tools/chatgui/transport/stream_bridge.cpp b/tools/chatgui/transport/stream_bridge.cpp index f8b46311..13c1f7b8 100644 --- a/tools/chatgui/transport/stream_bridge.cpp +++ b/tools/chatgui/transport/stream_bridge.cpp @@ -1,7 +1,5 @@ #include "stream_bridge.h" -#include #include -#include extern "C" { #include "../../../lib/debug_config.h" @@ -9,159 +7,25 @@ extern "C" { #include "../../../lib/u_async.h" #include "../../../src/utun_instance.h" #include "../../../src/stream/stream.h" +#include "../../../src/stream/stream_viewer.h" } namespace { -struct Viewer { uint64_t id = 0, group = 0, source = 0, epoch = 0; }; struct Bridge { - std::mutex lock; + std::mutex lock; // Ограничивает lifetime managers для callers GUI/workers. UTUN_INSTANCE* inst = nullptr; posted_task* pending = nullptr; uint64_t watch = 0; - bool watchPending = false, opening = false, stopping = false; - bool wantRead = false, waiting = false, resync = false; - bool wantVideo = false; - int64_t position = -1, audioAhead = STREAM_MEDIA_READ_AHEAD_US; - stream_media_status status{}; - uint64_t statusId = 0; - Viewer viewer, requested; - STREAM_HANDLE* handle = nullptr; + bool watchPending = false; + stream_viewers* viewers = nullptr; stream_publisher* publisher = nullptr; - std::deque output; - size_t bytes = 0; } bridge; -void process(void*); - -/* Вызывается под lock; резервирование делает отказ команды наблюдаемым до изменения состояния. */ -bool schedule() { - if (!bridge.inst) { - DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop stream command rejected: core unavailable"); return false; - } - if (bridge.pending) return true; - auto* task = static_cast(u_calloc(1, sizeof(posted_task))); - if (!task) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream task allocation failed"); return false; } - task->callback = process; bridge.pending = task; - uasync_post_reserved(bridge.inst->ua, task); - return true; -} - -/* Управляющие записи не теряются при media pressure; старые readers не накапливают очередь. */ -void output(StreamRecord record) { - if (bridge.output.size() >= 16 || bridge.bytes + static_cast(record.data.size()) > 4u * 1024u * 1024u) { - DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop stream RX pressure id=%llu records=%zu bytes=%zu; wait KEY", - (unsigned long long)record.id, bridge.output.size(), bridge.bytes); - bridge.output.clear(); bridge.bytes = 0; bridge.resync = true; - bridge.output.push_back({record.id, 0, StreamRecordKind::Gap}); - if (record.kind == StreamRecordKind::Video || record.kind == StreamRecordKind::Audio) return; - } - bridge.bytes += static_cast(record.data.size()); bridge.output.push_back(std::move(record)); -} - -/* Только uasync под lock; снимает собственное владение подпиской, не транспорт других владельцев. */ -void closeViewer(StreamRecordKind kind, int reason) { - if (kind == StreamRecordKind::Closed) { - for (auto i = bridge.output.begin(); i != bridge.output.end();) { - if (i->id == bridge.viewer.id && (i->kind == StreamRecordKind::Video || i->kind == StreamRecordKind::Audio)) { - bridge.bytes -= static_cast(i->data.size()); i = bridge.output.erase(i); - } else ++i; - } - } - if (bridge.viewer.id) output({bridge.viewer.id, 0, kind, {}, 0, reason}); - if (bridge.handle) stream_close(bridge.handle); - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop viewer released id=%llu reason=%d", - (unsigned long long)bridge.viewer.id, reason); - bridge.handle = nullptr; bridge.viewer = {}; bridge.wantRead = bridge.waiting = bridge.resync = false; -} - -void mediaFrame(void*, const stream_media_format* format, const stream_media_video* video, const stream_media_audio* audio) { - StreamRecord record; record.id = bridge.viewer.id; record.format = *format; - if (video) { - record.kind = StreamRecordKind::Video; record.pts = video->pts_us; record.flags = video->flags; - record.data = QByteArray(reinterpret_cast(video->data), static_cast(video->length)); - } else { - record.kind = StreamRecordKind::Audio; record.pts = audio->pts_us; - record.data = QByteArray(reinterpret_cast(audio->data), static_cast(audio->length)); - } - output(std::move(record)); -} - -/* Сеть только сообщает готовность индекса; ближайшие кадры читает запрос плеера из общего кольца. */ -void mediaEvent(UTUN_INSTANCE* inst, const stream_event* event, void*) { +void watch(void*) { std::lock_guard guard(bridge.lock); - if (inst != bridge.inst || !bridge.handle || event->descriptor.group_id != bridge.viewer.group - || event->descriptor.source_node_id != bridge.viewer.source || event->descriptor.source_epoch != bridge.viewer.epoch) return; - if (event->type == STREAM_EVENT_CLOSED || !stream_handle_active(bridge.handle)) - closeViewer(StreamRecordKind::Closed, event->reason); - else if (event->type == STREAM_EVENT_AVAILABLE) { bridge.waiting = false; schedule(); } -} - -/* Сериализованный actor: старый owner закрывается перед открытием нового выбранного источника. */ -void process(void*) { - std::lock_guard guard(bridge.lock); - bridge.pending = nullptr; - if (!bridge.inst) return; - if (bridge.watchPending) { - bridge.watchPending = false; - if (stream_watch_sources(bridge.inst, bridge.watch) != STREAM_OK) - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream source watch failed group=%llx", (unsigned long long)bridge.watch); - } - if (bridge.stopping || bridge.opening) { - if (bridge.viewer.id) closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_STOPPED); - bridge.stopping = false; - } - if (bridge.opening) { - bridge.opening = false; bridge.viewer = bridge.requested; bridge.requested = {}; - bridge.status = {}; bridge.statusId = bridge.viewer.id; bridge.position = -1; bridge.waiting = false; - int result = stream_media_open(bridge.inst, bridge.viewer.group, bridge.viewer.source, bridge.viewer.epoch, - STREAM_MEDIA_TRACK_VIDEO | STREAM_MEDIA_TRACK_AUDIO, &bridge.handle); - int reason = result == STREAM_ERR_CONFLICT ? STREAM_CLOSE_SOURCE_CHANGED : STREAM_CLOSE_FAILED; - if (result != STREAM_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer open failed id=%llu group=%llx source=%llx result=%d", - (unsigned long long)bridge.viewer.id, (unsigned long long)bridge.viewer.group, - (unsigned long long)bridge.viewer.source, result); - closeViewer(StreamRecordKind::Closed, reason); return; - } - output({bridge.viewer.id, 0, StreamRecordKind::Opened}); - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop viewer opened id=%llu group=%llx source=%llx epoch=%llx", - (unsigned long long)bridge.viewer.id, (unsigned long long)bridge.viewer.group, - (unsigned long long)bridge.viewer.source, (unsigned long long)bridge.viewer.epoch); - bridge.wantRead = false; - } - if (!bridge.handle) return; - if (bridge.resync) { - bridge.resync = false; bridge.position = -1; - if (stream_media_resync(bridge.handle) != STREAM_OK) { closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_FAILED); return; } - } - int statusResult = stream_media_get_status(bridge.handle, &bridge.status); - if (statusResult < 0) { - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop media index failed result=%d", statusResult); - closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_FAILED); return; - } - bool prepared = std::any_of(bridge.output.begin(), bridge.output.end(), [](const StreamRecord& r) { return r.id == bridge.viewer.id; }); - if (!bridge.wantRead || prepared || bridge.bytes + STREAM_MEDIA_MAX_FRAME > 4u * 1024u * 1024u) return; - bridge.wantRead = false; - stream_media_packet packet{}; - int result = stream_media_read(bridge.handle, bridge.position, bridge.wantVideo, bridge.audioAhead, &packet); - DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "desktop media read id=%llu PTS=%lld video=%d audio_ahead=%lld result=%d kind=%d records=%zu", - (unsigned long long)bridge.viewer.id, (long long)bridge.position, bridge.wantVideo, - (long long)bridge.audioAhead, result, result == STREAM_OK ? (int)packet.kind : -1, bridge.output.size()); - if (stream_media_get_status(bridge.handle, &bridge.status) < 0) result = STREAM_ERR_FORMAT; - if (result == STREAM_OK) { - if (packet.kind == STREAM_MEDIA_READ_END) closeViewer(StreamRecordKind::End, 0); - else mediaFrame(nullptr, &packet.format, packet.kind == STREAM_MEDIA_READ_VIDEO ? &packet.video : nullptr, - packet.kind == STREAM_MEDIA_READ_AUDIO ? &packet.audio : nullptr); - } else if (result == STREAM_MORE) { bridge.wantRead = true; schedule(); } - else if (result == STREAM_WAIT) bridge.wantRead = bridge.waiting = true; - else if (result == STREAM_GAP || result == STREAM_EXPIRED) { - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop playback resync id=%llu result=%d", (unsigned long long)bridge.viewer.id, result); - stream_media_resync(bridge.handle); bridge.position = -1; - bridge.status = {}; - output({bridge.viewer.id, 0, StreamRecordKind::Gap}); - } else { - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop playback read failed result=%d", result); - closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_FAILED); - } + bridge.pending = nullptr; bridge.watchPending = false; + if (stream_watch_sources(bridge.inst, bridge.watch) != STREAM_OK) + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop source watch failed group=%llx", (unsigned long long)bridge.watch); } } @@ -171,8 +35,10 @@ bool gui_stream_attach(UTUN_INSTANCE* inst) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream attach rejected"); return false; } if (stream_publisher_create(&bridge.publisher, inst) < 0) return false; - bridge.inst = inst; bridge.output.clear(); bridge.bytes = 0; - stream_set_event_callback(inst, mediaEvent, nullptr); + if (stream_viewers_create(&bridge.viewers, inst) < 0) { + stream_publisher_destroy(bridge.publisher); bridge.publisher = nullptr; return false; + } + bridge.inst = inst; if (bridge.watchPending) { bridge.watchPending = false; if (stream_watch_sources(inst, bridge.watch) != STREAM_OK) @@ -185,72 +51,61 @@ void gui_stream_detach(UTUN_INSTANCE* inst) { std::lock_guard guard(bridge.lock); if (!inst || bridge.inst != inst) return; if (bridge.pending) uasync_cancel_post(inst->ua, bridge.pending); - stream_watch_sources(inst, 0); - bridge.pending = nullptr; stream_set_event_callback(inst, nullptr, nullptr); - if (bridge.viewer.id) closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_SERVICE); - if (bridge.opening) output({bridge.requested.id, 0, StreamRecordKind::Closed, {}, 0, STREAM_CLOSE_SERVICE}); + stream_watch_sources(inst, 0); bridge.pending = nullptr; + stream_viewers_destroy(bridge.viewers); bridge.viewers = nullptr; stream_publisher_destroy(bridge.publisher); bridge.publisher = nullptr; - bridge.inst = nullptr; bridge.requested = {}; bridge.opening = bridge.stopping = bridge.watchPending = false; + bridge.inst = nullptr; bridge.watchPending = bridge.watch != 0; DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream bridge detached"); } bool gui_stream_watch(uint64_t group) { std::lock_guard guard(bridge.lock); - if (!bridge.inst) { - bridge.watch = group; bridge.watchPending = true; - DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "desktop source watch queued until attach group=%llx", (unsigned long long)group); return true; + if (bridge.inst && !bridge.pending) { + auto* task = static_cast(u_calloc(1, sizeof(posted_task))); + if (!task) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop watch task allocation failed"); return false; } + task->callback = watch; bridge.pending = task; + uasync_post_reserved(bridge.inst->ua, task); } - if (!schedule()) return false; bridge.watch = group; bridge.watchPending = true; return true; } bool gui_stream_open(uint64_t id, uint64_t group, uint64_t source, uint64_t epoch) { std::lock_guard guard(bridge.lock); - if (!id || !group || !source || !epoch) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer invalid identity"); return false; } - if (!schedule()) return false; - if (bridge.opening) output({bridge.requested.id, 0, StreamRecordKind::Closed, {}, 0, STREAM_CLOSE_STOPPED}); - bridge.requested = {id, group, source, epoch}; bridge.opening = true; return true; + return stream_viewer_open(bridge.viewers, id, group, source, epoch, STREAM_MEDIA_TRACK_VIDEO | STREAM_MEDIA_TRACK_AUDIO) == 0; } void gui_stream_stop(uint64_t id) { - std::lock_guard guard(bridge.lock); - if (bridge.opening && bridge.requested.id == id) { - bridge.opening = false; bridge.requested = {}; output({id, 0, StreamRecordKind::Closed, {}, 0, STREAM_CLOSE_STOPPED}); - } - if (bridge.viewer.id == id && schedule()) bridge.stopping = true; - for (auto i = bridge.output.begin(); i != bridge.output.end();) { - if (i->id == id) { bridge.bytes -= static_cast(i->data.size()); i = bridge.output.erase(i); } - else ++i; - } + std::lock_guard guard(bridge.lock); stream_viewer_stop(bridge.viewers, id); } bool gui_stream_pull(uint64_t id, StreamRecord& record, int64_t position, bool wantVideo, int64_t audioAhead) { std::lock_guard guard(bridge.lock); - auto i = std::find_if(bridge.output.begin(), bridge.output.end(), [id](const StreamRecord& r) { return r.id == id; }); - bool found = i != bridge.output.end(); - if (found) { bridge.bytes -= static_cast(i->data.size()); record = std::move(*i); bridge.output.erase(i); } - if (!found && bridge.viewer.id == id && bridge.handle) { - bool changed = bridge.position != position || bridge.wantVideo != wantVideo || bridge.audioAhead != audioAhead; - bridge.position = position; bridge.wantVideo = wantVideo; bridge.audioAhead = audioAhead; - bridge.wantRead = wantVideo || audioAhead >= 0; - if (bridge.wantRead && (!bridge.waiting || changed)) { bridge.waiting = false; schedule(); } + auto* r = stream_viewer_pull(bridge.viewers, id); + if (!r) { + if (bridge.viewers) stream_viewer_request(bridge.viewers, id, position, wantVideo, audioAhead); + return false; } - return found; + switch (r->kind) { + case STREAM_VIEWER_VIDEO: record.kind = StreamRecordKind::Video; break; + case STREAM_VIEWER_AUDIO: record.kind = StreamRecordKind::Audio; break; + case STREAM_VIEWER_GAP: record.kind = StreamRecordKind::Gap; break; + case STREAM_VIEWER_OPENED: record.kind = StreamRecordKind::Opened; break; + case STREAM_VIEWER_END: record.kind = StreamRecordKind::End; break; + case STREAM_VIEWER_CLOSED: record.kind = StreamRecordKind::Closed; break; + } + record.id = id; record.pts = r->pts; record.format = r->format; record.flags = r->flags; record.reason = r->reason; + record.data = QByteArray(reinterpret_cast(r->data), static_cast(r->length)); + u_free(r); return true; } bool gui_stream_status(uint64_t id, stream_media_status& status) { std::lock_guard guard(bridge.lock); - if (bridge.statusId != id) return false; - status = bridge.status; return true; + return stream_viewer_status(bridge.viewers, id, &status) == 0; } bool gui_stream_resync(uint64_t id) { std::lock_guard guard(bridge.lock); - if (bridge.viewer.id != id || !bridge.handle) return false; - bridge.resync = bridge.wantRead = true; bridge.waiting = false; bridge.position = -1; - bridge.status = {}; - bridge.output.clear(); bridge.bytes = 0; - return schedule(); + return stream_viewer_resync(bridge.viewers, id) == 0; } bool gui_stream_ready() { std::lock_guard guard(bridge.lock); return bridge.inst != nullptr; } diff --git a/tools/chatgui/transport/stream_bridge.h b/tools/chatgui/transport/stream_bridge.h index 0fbaf333..6cfc284f 100644 --- a/tools/chatgui/transport/stream_bridge.h +++ b/tools/chatgui/transport/stream_bridge.h @@ -1,6 +1,6 @@ /* Desktop live media: attach/detach в uasync до destroy ядра; остальные вызовы потокобезопасны. - * Один общий publisher и один viewer; команды содержат IDs, не borrowed pointers ядра. - * Один ответ на запрос плеера; следующий кадр не читается автоматически. Общий предел RX 4MiB/16 записей. + * Один общий publisher и независимые viewers; команды содержат IDs, не borrowed pointers ядра. + * Один ответ на запрос каждого плеера; следующий кадр не читается автоматически. * Watch до attach запоминает выбранную группу. Каталог идёт через CHAT_EVT_STREAM_SOURCES, * media pull читает ближайшие кадры из общего RS-кольца; копия нужна только для передачи между потоками. */ #pragma once