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

Consume desktop media batches with per-track backpressure and wakeups

Browse Source
This commit is contained in:
evgeny committed 2026-10-20 23:31:50 +02:00
1 parent 179f248db8
commit 039c4a9ed4
2 files changed
+64 -20
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/stream/stream_viewer.c
+8 -2
View File
@@ -171,7 +171,13 @@ static void process(void *arg) {
struct stream_media_packet p;
int result = stream_media_read(s->handle, s->position, video > 0, s->audio_ahead, &p);
if (result == STREAM_OK) {
if (p.kind == STREAM_MEDIA_READ_END) { close_viewer(s, STREAM_VIEWER_END, 0); break; }
if (p.kind == STREAM_MEDIA_READ_END) {
if (stream_media_get_status(s->handle, &s->status) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer final status failed id=%llu", (unsigned long long)s->id);
close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED);
} else close_viewer(s, STREAM_VIEWER_END, 0);
break;
}
int kind = p.kind == STREAM_MEDIA_READ_VIDEO ? STREAM_VIEWER_VIDEO : STREAM_VIEWER_AUDIO;
if (output(s, kind, 0, &p) < 0) { close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); break; }
count++; bytes += s->tail->length;
@@ -193,7 +199,7 @@ static void process(void *arg) {
close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); break;
}
} while (count < s->records && bytes < 256u * 1024u && get_time_us() - began < 2000 && (video || s->audio_ahead >= 0));
if (!s->handle) continue;
if (!s->handle || (s->tail && s->tail->kind == STREAM_VIEWER_GAP)) continue;
if (stream_media_get_status(s->handle, &s->status) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer batch status failed id=%llu", (unsigned long long)s->id);
close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED);
tools/chatgui/src/streamplayer_window.cpp
+56 -18
View File
@@ -14,6 +14,7 @@
#include <atomic>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <deque>
#include <chrono>
#include <algorithm>
@@ -57,6 +58,9 @@ struct StreamPlayerWindow::Playback {
uint64_t audioGeneration = 0;
std::thread worker;
std::mutex lock;
std::mutex wakeLock;
std::condition_variable wake;
uint64_t wakeVersion = 0, dataSignals = 0;
std::deque<Video> videos;
size_t videoBytes = 0;
stream_playback_audio* sound = nullptr;
@@ -71,6 +75,12 @@ struct StreamPlayerWindow::Playback {
int64_t audioFedAt = 0, audioFedPts = -1, workerAt = 0, guiAt = 0, guiGapMax = 0;
~Playback() { stream_playback_audio_destroy(sound); }
// Отдельный lock: уведомление из ядра не ждёт codec/PCM lock и не вызывает bridge.
void notify(bool data = false) {
{ std::lock_guard<std::mutex> guard(wakeLock); wakeVersion++; if (data) dataSignals++; }
wake.notify_one();
}
uint64_t played(int64_t now) const {
if (!sound || !audioDacAt) return 0;
long double frames = (long double)audioFrameStart + (long double)(now - audioDacAt) * 48000 / 1000000;
@@ -134,6 +144,7 @@ void StreamPlayerWindow::attachAudio() {
[this](float* output, unsigned frames, int64_t firstTimeUs, uint64_t generation) {
int16_t pcm[960 * 2] = {};
auto& p = *m_playback; std::lock_guard<std::mutex> guard(p.lock);
p.notify();
int64_t now = nowUs();
bool restored = !p.audioAvailable.exchange(true);
if ((p.audioGeneration && p.audioGeneration != generation) || (restored && p.started)) p.resync = true;
@@ -191,6 +202,7 @@ void StreamPlayerWindow::tick() {
bool audioDone = !p.activeAudio || !p.audioAvailable || (p.endFrames != UINT64_MAX && p.played(nowUs()) >= p.endFrames);
finished = p.ended && p.videoEnded && p.videos.empty() && audioDone;
}
p.notify();
if (!image.isNull()) {
m_render->setFrame(image); m_status->setText(text("Live"));
if (first) DEBUG_INFO(DEBUG_CATEGORY_STREAM,
@@ -212,6 +224,7 @@ void StreamPlayerWindow::tick() {
/* Callback устройства прекращается до освобождения DSP; worker снимает только свой reader. */
void StreamPlayerWindow::stop() {
if (!m_playback->running.exchange(false) && !m_playback->worker.joinable()) return;
m_playback->notify();
m_timer.stop(); gui_stream_stop(m_playback->id);
SoundManager::instance()->detachStream(m_playback->id); m_audioAttached = false; m_playback->audioAvailable = false;
if (m_playback->worker.joinable()) m_playback->worker.join();
@@ -237,19 +250,21 @@ void StreamPlayerWindow::decode() {
SwsContext* converter = nullptr;
bool waitKey = true, opened = false, finishing = false, eosSent = false;
stream_media_format videoFormat{};
std::deque<StreamRecord> records; // Только worker: кодированные записи двух треков, управление — барьер.
uint32_t revision = 0;
uint64_t decoded = 0, discarded = 0, reportedShown = 0;
int64_t deadline = nowUs() + 10000000, report = nowUs(), decoderProgress = -1, lastInputAt = -1;
uint64_t reportedGaps = 0;
int64_t gapReportAt = 0;
struct {
uint64_t loops = 0, video = 0, audio = 0, empty = 0;
uint64_t loops = 0, video = 0, audio = 0, empty = 0, batches = 0, batchMax = 0;
int64_t codec = 0, codecMax = 0, convert = 0, convertMax = 0;
int64_t audioFeed = 0, audioFeedMax = 0;
int64_t pull = 0, pullMax = 0, sleep = 0, sleepMax = 0, loopMax = 0;
int64_t pull = 0, pullMax = 0, wait = 0, waitMax = 0, loopMax = 0;
} work;
int64_t lastLoopAt = report;
auto reset = [&](bool request) {
records.clear();
avcodec_free_context(&decoder); av_packet_unref(packet); waitKey = true; finishing = eosSent = false;
decoderProgress = lastInputAt = -1;
{
@@ -343,26 +358,33 @@ void StreamPlayerWindow::decode() {
}
};
if (!frame || !packet) fail(text("Stream decoder allocation failed"));
else if (!gui_stream_open(p.id, p.group, p.source, p.epoch)) fail(text("Could not connect to the stream"));
else if (!gui_stream_open(p.id, p.group, p.source, p.epoch,
[](void* arg) { static_cast<Playback*>(arg)->notify(true); }, &p)) fail(text("Could not connect to the stream"));
else while (p.running) {
uint64_t wakeVersion;
{ std::lock_guard<std::mutex> guard(p.wakeLock); wakeVersion = p.wakeVersion; }
int64_t loopAt = nowUs();
work.loops++; work.loopMax = std::max(work.loopMax, loopAt - lastLoopAt); lastLoopAt = loopAt;
if (loopAt - report >= 1000000) {
double seconds = (loopAt - report) / 1000000.0;
std::lock_guard<std::mutex> guard(p.lock);
uint64_t dataSignals;
{ std::lock_guard<std::mutex> wakeGuard(p.wakeLock); dataSignals = p.dataSignals; p.dataSignals = 0; }
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
"desktop viewer work id=%llu interval_us=%lld loops=%llu video=%llu audio=%llu empty=%llu "
"codec_us=%lld/%lld convert_us=%lld/%lld audio_feed_us=%lld/%lld pull_us=%lld/%lld sleep_us=%lld/%lld "
"codec_us=%lld/%lld convert_us=%lld/%lld audio_feed_us=%lld/%lld pull_us=%lld/%lld wait_us=%lld/%lld "
"loop_max_us=%lld gui_gap_max_us=%lld gui_age_us=%lld decoded=%llu shown=%llu show_fps=%.1f "
"clock_us=%lld ready_us=%lld videoQ=%zu hold=%d underruns=%llu timing=total/max",
"clock_us=%lld ready_us=%lld videoQ=%zu hold=%d underruns=%llu batches=%llu batch_max=%llu "
"pending_records=%zu data_signals=%llu timing=total/max",
(unsigned long long)p.id, (long long)(loopAt - report), (unsigned long long)work.loops,
(unsigned long long)work.video, (unsigned long long)work.audio, (unsigned long long)work.empty,
(long long)work.codec, (long long)work.codecMax, (long long)work.convert, (long long)work.convertMax,
(long long)work.audioFeed, (long long)work.audioFeedMax,
(long long)work.pull, (long long)work.pullMax, (long long)work.sleep, (long long)work.sleepMax,
(long long)work.pull, (long long)work.pullMax, (long long)work.wait, (long long)work.waitMax,
(long long)work.loopMax, (long long)p.guiGapMax, (long long)(p.guiAt ? nowUs() - p.guiAt : -1),
(unsigned long long)decoded, (unsigned long long)p.shown, (p.shown - reportedShown) / seconds,
(long long)p.position, (long long)p.ready, p.videos.size(), p.holdAudio, (unsigned long long)p.underruns);
(long long)p.position, (long long)p.ready, p.videos.size(), p.holdAudio, (unsigned long long)p.underruns,
(unsigned long long)work.batches, (unsigned long long)work.batchMax, records.size(), (unsigned long long)dataSignals);
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM,
"desktop viewer id=%llu decoded=%llu discarded=%llu shown=%llu showFps=%.1f videoQ=%zu "
"clock=%lld reserve_us=%lld minimum_us=%lld tempo=%d lead_us=%lld underruns=%llu "
@@ -420,11 +442,12 @@ void StreamPlayerWindow::decode() {
(unsigned long long)p.id, (long long)p.lastInputPts, (long long)p.lastDecodedPts, packet->size, finishing);
fail(text("Stream video decoder failed")); break;
} }
int64_t position, audioAhead; bool wantVideo;
int64_t position, audioAhead; bool wantVideo; unsigned videoBudget;
{
std::lock_guard<std::mutex> guard(p.lock);
int64_t now = nowUs(); position = p.started ? p.clock(now) : -1;
wantVideo = !packet->size && p.videos.size() < 2;
videoBudget = static_cast<unsigned>(std::max(0, 2 - (int)p.videos.size() - (packet->size ? 1 : 0)));
audioAhead = !p.activeAudio || !p.sound || stream_playback_audio_needs_input(p.sound) ? p.deviceLead + 80000 : -1;
if (p.presenting && p.ready >= 0) {
uint64_t played = p.played(now);
@@ -443,20 +466,35 @@ void StreamPlayerWindow::decode() {
}
}
}
StreamRecord record;
int64_t pullAt = nowUs();
bool received = !finishing && gui_stream_pull(p.id, record, position, wantVideo, audioAhead);
int64_t pullUs = nowUs() - pullAt;
work.pull += pullUs; work.pullMax = std::max(work.pullMax, pullUs);
if (!received) {
// Каждый трек расходуется по своей готовности. GAP/FIN/CLOSED не обгоняют предыдущие media.
auto next = records.end();
unsigned pendingVideo = 0, pendingAudio = 0; bool barrier = false;
for (auto it = records.begin(); it != records.end(); ++it) {
bool video = it->kind == StreamRecordKind::Video, audio = it->kind == StreamRecordKind::Audio;
pendingVideo += video; pendingAudio += audio;
if (!video && !audio) { barrier = true; if (it == records.begin()) next = it; break; }
if (next == records.end() && ((video && wantVideo) || (audio && audioAhead >= 0))) next = it;
}
if (next == records.end()) {
size_t previous = records.size();
int64_t pullAt = nowUs();
bool received = !finishing && !barrier && gui_stream_pull_batch(p.id, records, position,
videoBudget > pendingVideo ? videoBudget - pendingVideo : 0, pendingAudio ? -1 : audioAhead);
int64_t pullUs = nowUs() - pullAt;
work.pull += pullUs; work.pullMax = std::max(work.pullMax, pullUs);
if (received) {
work.batches++; work.batchMax = std::max<uint64_t>(work.batchMax, records.size() - previous); continue;
}
work.empty++;
if (!gui_stream_ready()) { fail(text("Chat service is stopped")); break; }
if ((!opened || waitKey) && nowUs() > deadline) { fail(text("Stream connection timed out")); break; }
int64_t sleepAt = nowUs();
std::this_thread::sleep_for(std::chrono::milliseconds(5));
int64_t sleepUs = nowUs() - sleepAt;
work.sleep += sleepUs; work.sleepMax = std::max(work.sleepMax, sleepUs); continue;
int64_t waitAt = nowUs();
{ std::unique_lock<std::mutex> guard(p.wakeLock);
p.wake.wait_for(guard, std::chrono::milliseconds(50), [&] { return !p.running || p.wakeVersion != wakeVersion; }); }
int64_t waitUs = nowUs() - waitAt;
work.wait += waitUs; work.waitMax = std::max(work.waitMax, waitUs); continue;
}
StreamRecord record = std::move(*next); records.erase(next);
if (record.kind == StreamRecordKind::Opened) { opened = true; continue; }
if (record.kind == StreamRecordKind::Gap) { reset(false); continue; }
if (record.kind == StreamRecordKind::Closed) {