jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Switch desktop media delivery to bounded demand reads from the shared ring

master
evgeny 6 days ago
parent
2c7bd78a64
commit
650b65078a
3 changed files with 71 additions and 26 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 1
      src/call/voice_sources.cmake
  2. 90
      tools/chatgui/transport/stream_bridge.cpp
  3. 6
      tools/chatgui/transport/stream_bridge.h

1
src/call/voice_sources.cmake
Unescape View File

@ -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

90
tools/chatgui/transport/stream_bridge.cpp
Unescape View File

@ -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<StreamRecord> 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<size_t>(record.data.size()) > 4u * 1024u * 1024u) {
if (bridge.output.size() >= 16 || bridge.bytes + static_cast<size_t>(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<std::mutex> 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<std::mutex> 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<size_t>(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<size_t>(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<std::mutex> 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<std::mutex> 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<std::mutex> guard(bridge.lock); return bridge.inst != nullptr; }

6
tools/chatgui/transport/stream_bridge.h
Unescape View File

@ -7,6 +7,7 @@
#include <QByteArray>
#include <cstdint>
#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();

Write Preview
Loading…
Cancel
Save