From 650b65078a395a590889042b4340a6691793d52b Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 22:42:16 +0200 Subject: [PATCH] Switch desktop media delivery to bounded demand reads from the shared ring --- src/call/voice_sources.cmake | 1 + tools/chatgui/transport/stream_bridge.cpp | 90 ++++++++++++++++------- tools/chatgui/transport/stream_bridge.h | 6 +- 3 files changed, 71 insertions(+), 26 deletions(-) diff --git a/src/call/voice_sources.cmake b/src/call/voice_sources.cmake index 3061eea8..89c24bc8 100644 --- a/src/call/voice_sources.cmake +++ b/src/call/voice_sources.cmake @@ -22,6 +22,7 @@ function(utun_setup_voice LIB_DIR CALL_DIR) ${CALL_DIR}/call_jitter.cpp ${_src_dir}/radio/radio_jitter.cpp ${_src_dir}/radio/radio_audio.c + ${_src_dir}/stream/stream_playback_audio.cpp ${_st_dir}/source/SoundTouch/SoundTouch.cpp ${_st_dir}/source/SoundTouch/TDStretch.cpp ${_st_dir}/source/SoundTouch/RateTransposer.cpp diff --git a/tools/chatgui/transport/stream_bridge.cpp b/tools/chatgui/transport/stream_bridge.cpp index 6998b4df..54265bbb 100644 --- a/tools/chatgui/transport/stream_bridge.cpp +++ b/tools/chatgui/transport/stream_bridge.cpp @@ -19,10 +19,12 @@ struct Bridge { UTUN_INSTANCE* inst = nullptr; posted_task* pending = nullptr; uint64_t watch = 0; - bool watchPending = false, opening = false, stopping = false, overflow = false; + bool watchPending = false, opening = false, stopping = false; + bool wantRead = false, waiting = false, resync = false; + int64_t position = -1, videoAhead = STREAM_MEDIA_READ_AHEAD_US, audioAhead = STREAM_MEDIA_READ_AHEAD_US; + stream_media_status status{}; Viewer viewer, requested; STREAM_HANDLE* handle = nullptr; - stream_media_receiver* receiver = nullptr; std::deque output; size_t bytes = 0; } bridge; @@ -44,10 +46,10 @@ bool schedule() { /* Управляющие записи не теряются при media pressure; старые readers не накапливают очередь. */ void output(StreamRecord record) { - if (bridge.output.size() >= 256 || bridge.bytes + static_cast(record.data.size()) > 4u * 1024u * 1024u) { + 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.overflow = true; + 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; } @@ -58,10 +60,9 @@ void output(StreamRecord record) { void closeViewer(StreamRecordKind kind, int reason) { if (bridge.viewer.id) output({bridge.viewer.id, 0, kind, {}, 0, reason}); if (bridge.handle) stream_close(bridge.handle); - stream_media_receiver_destroy(bridge.receiver); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop viewer released id=%llu reason=%d", (unsigned long long)bridge.viewer.id, reason); - bridge.handle = nullptr; bridge.receiver = nullptr; bridge.viewer = {}; bridge.overflow = false; + 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) { @@ -77,24 +78,14 @@ void mediaFrame(void*, const stream_media_format* format, const stream_media_vid output(std::move(record)); } -/* Scratch receiver/stream не покидает callback; decoder получает собственные копии. */ +/* Сеть только сообщает готовность индекса; ближайшие кадры читает запрос плеера из общего кольца. */ void mediaEvent(UTUN_INSTANCE* inst, const stream_event* event, 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_DATA) { - int result = stream_media_receiver_feed(bridge.receiver, event->data, event->length); - if (result != STREAM_OK) { - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer media rejected result=%d", result); - closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_FAILED); - } else if (bridge.overflow) { - stream_media_receiver_gap(bridge.receiver); bridge.overflow = false; - } - } else if (event->type == STREAM_EVENT_GAP) { - stream_media_receiver_gap(bridge.receiver); output({bridge.viewer.id, 0, StreamRecordKind::Gap}); - } else if (event->type == STREAM_EVENT_END || !stream_handle_active(bridge.handle)) { - closeViewer(event->type == STREAM_EVENT_END ? StreamRecordKind::End : StreamRecordKind::Closed, event->reason); - } + if (event->type == STREAM_EVENT_CLOSED || !stream_handle_active(bridge.handle)) + closeViewer(StreamRecordKind::Closed, event->reason); + else if (event->type == STREAM_EVENT_AVAILABLE) { bridge.wantRead = true; bridge.waiting = false; schedule(); } } /* Сериализованный actor: старый owner закрывается перед открытием нового выбранного источника. */ @@ -111,8 +102,9 @@ void process(void*) { if (bridge.viewer.id) closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_STOPPED); bridge.stopping = false; } - if (!bridge.opening) return; + if (bridge.opening) { bridge.opening = false; bridge.viewer = bridge.requested; bridge.requested = {}; + bridge.status = {}; bridge.position = -1; bridge.waiting = false; stream_member_state member{}; int reason = STREAM_CLOSE_FAILED; int result = stream_member_load(bridge.inst, bridge.viewer.group, bridge.viewer.source, &member, nullptr); @@ -129,7 +121,7 @@ void process(void*) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop reader encountered stale local epoch"); result = STREAM_ERR_CONFLICT; reason = STREAM_CLOSE_SOURCE_CHANGED; } - if (result == STREAM_OK) result = stream_media_receiver_create(&bridge.receiver, mediaFrame, nullptr); + if (result == STREAM_OK) result = stream_media_enable(bridge.handle, STREAM_MEDIA_TRACK_VIDEO | STREAM_MEDIA_TRACK_AUDIO); 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, @@ -140,6 +132,35 @@ void process(void*) { 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 = true; + } + if (!bridge.handle) 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; + } + if (bridge.resync) { + bridge.resync = false; bridge.position = -1; + if (stream_media_resync(bridge.handle) != STREAM_OK) { closeViewer(StreamRecordKind::Closed, STREAM_CLOSE_FAILED); return; } + } + if (!bridge.wantRead || bridge.output.size() >= 2 || 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.videoAhead, bridge.audioAhead, &packet); + 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_WAIT) 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; + 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); + } } } @@ -200,11 +221,30 @@ void gui_stream_stop(uint64_t id) { } } -bool gui_stream_pull(uint64_t id, StreamRecord& record) { +bool gui_stream_pull(uint64_t id, StreamRecord& record, int64_t position, int64_t videoAhead, 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; }); - if (i == bridge.output.end()) return false; - bridge.bytes -= static_cast(i->data.size()); record = std::move(*i); bridge.output.erase(i); return true; + bool found = i != bridge.output.end(); + if (found) { bridge.bytes -= static_cast(i->data.size()); record = std::move(*i); bridge.output.erase(i); } + if (bridge.viewer.id == id && bridge.handle && (!bridge.waiting || bridge.position != position + || bridge.videoAhead != videoAhead || bridge.audioAhead != audioAhead || found)) { + bridge.position = position; bridge.videoAhead = videoAhead; bridge.audioAhead = audioAhead; bridge.wantRead = true; schedule(); + } + return found; +} + +bool gui_stream_status(uint64_t id, stream_media_status& status) { + std::lock_guard guard(bridge.lock); + if (bridge.viewer.id != id && !bridge.status.finished) return false; + status = bridge.status; return true; +} + +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.output.clear(); bridge.bytes = 0; + return schedule(); } 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 4f965552..7f0ded42 100644 --- a/tools/chatgui/transport/stream_bridge.h +++ b/tools/chatgui/transport/stream_bridge.h @@ -7,6 +7,7 @@ #include #include #include "../../../src/stream/stream_media.h" +#include "../../../src/stream/stream_media_buffer.h" struct UTUN_INSTANCE; enum class StreamRecordKind { Video, Audio, Gap, Opened, End, Closed }; @@ -23,5 +24,8 @@ void gui_stream_detach(UTUN_INSTANCE* inst); bool gui_stream_watch(uint64_t group); bool gui_stream_open(uint64_t id, uint64_t group, uint64_t source, uint64_t epoch); void gui_stream_stop(uint64_t id); -bool gui_stream_pull(uint64_t id, StreamRecord& record); +bool gui_stream_pull(uint64_t id, StreamRecord& record, int64_t position = -1, + int64_t videoAhead = STREAM_MEDIA_READ_AHEAD_US, int64_t audioAhead = STREAM_MEDIA_READ_AHEAD_US); +bool gui_stream_status(uint64_t id, stream_media_status& status); +bool gui_stream_resync(uint64_t id); bool gui_stream_ready();