Drive desktop ring playback from corrected history and physical audio time

This commit is contained in:
evgeny committed 2026-10-13 23:30:06 +02:00
1 parent e1abcb5d69
commit a12665bbd1
1 file changed
+147 -152
+147 -152
View File
@@ -14,15 +14,15 @@
#include <thread> #include <thread>
#include <mutex> #include <mutex>
#include <deque> #include <deque>
#include <array>
#include <chrono> #include <chrono>
#include <algorithm> #include <algorithm>
#include <cstring> #include <cstring>
extern "C" { extern "C" {
#include "../../../lib/debug_config.h" #include "../../../lib/debug_config.h"
#include "../../../lib/opus_codec.h"
#include "../../../src/stream/stream.h" #include "../../../src/stream/stream.h"
#include "../../../src/stream/stream_playback_clock.h"
#include "../../../src/stream/stream_playback_audio.h"
#ifdef HAVE_FFMPEG #ifdef HAVE_FFMPEG
#include <libavcodec/avcodec.h> #include <libavcodec/avcodec.h>
#include <libswscale/swscale.h> #include <libswscale/swscale.h>
@@ -39,7 +39,6 @@ QString text(const char* value) { return QCoreApplication::translate("ChatGui",
struct StreamPlayerWindow::Playback { struct StreamPlayerWindow::Playback {
struct Video { QImage image; int64_t pts; }; struct Video { QImage image; int64_t pts; };
struct Audio { std::array<int16_t, 1920> pcm; int64_t pts; unsigned offset = 0; };
uint64_t id = ++nextId, group = 0, source = 0, epoch = 0; uint64_t id = ++nextId, group = 0, source = 0, epoch = 0;
std::atomic<bool> running{true}, audioAvailable{false}; std::atomic<bool> running{true}, audioAvailable{false};
std::atomic<float> volume{1.0f}; std::atomic<float> volume{1.0f};
@@ -47,14 +46,27 @@ struct StreamPlayerWindow::Playback {
std::mutex lock; std::mutex lock;
std::deque<Video> videos; std::deque<Video> videos;
size_t videoBytes = 0; size_t videoBytes = 0;
std::deque<Audio> audios; stream_playback_audio* sound = nullptr;
int64_t prefillUntil = 0, videoBase = -1, videoAt = 0, audioPts = -1, audioAt = 0; stream_playback_clock adaptation{};
uint64_t shown = 0, underruns = 0; stream_playback_state timing{};
bool ended = false; stream_media_status media{};
int64_t position = -1, clockAt = 0, ready = -1, audioDacAt = 0, deviceLead = 0, lastShownPts = -1;
uint64_t audioFrameStart = 0, shown = 0, underruns = 0, endFrames = UINT64_MAX;
bool started = false, ended = false, activeAudio = false, holdAudio = false, resync = false;
~Playback() { stream_playback_audio_destroy(sound); }
int64_t clock(int64_t now) const { uint64_t played(int64_t now) const {
if (audioAvailable && audioPts >= 0 && now - audioAt < 200000) return audioPts + now - audioAt; if (!sound || !audioDacAt) return 0;
return videoBase >= 0 ? videoBase + now - videoAt : 0; long double frames = (long double)audioFrameStart + (long double)(now - audioDacAt) * 48000 / 1000000;
uint64_t count = frames <= 0 ? 0 : (uint64_t)frames;
return std::min(count, stream_playback_audio_written(sound));
}
int64_t clock(int64_t now) {
if (!started) return 0;
if (activeAudio && sound && audioAvailable)
return std::max(position, stream_playback_audio_position(sound, played(now)));
int64_t proposed = position + (now - clockAt) * adaptation.tempo_milli / 1000;
return std::max(position, std::min(ready, proposed));
} }
}; };
@@ -62,8 +74,7 @@ StreamPlayerWindow::StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t
: QWidget(parent), m_playback(std::make_unique<Playback>()) { : QWidget(parent), m_playback(std::make_unique<Playback>()) {
m_playback->group = group; m_playback->source = source; m_playback->epoch = epoch; m_playback->group = group; m_playback->source = source; m_playback->epoch = epoch;
setWindowTitle(text("Live stream: %1").arg(name)); resize(800, 500); setWindowTitle(text("Live stream: %1").arg(name)); resize(800, 500);
auto* layout = new QVBoxLayout(this); auto* layout = new QVBoxLayout(this); layout->addWidget(new QLabel(name, this));
layout->addWidget(new QLabel(name, this));
m_status = new QLabel(text("Connecting…"), this); layout->addWidget(m_status); m_status = new QLabel(text("Connecting…"), this); layout->addWidget(m_status);
m_render = new VideoRenderWidget(this); layout->addWidget(m_render, 1); m_render = new VideoRenderWidget(this); layout->addWidget(m_render, 1);
auto* controls = new QHBoxLayout; auto* controls = new QHBoxLayout;
@@ -77,12 +88,14 @@ StreamPlayerWindow::StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t
connect(this, &StreamPlayerWindow::statusChanged, m_status, &QLabel::setText, Qt::QueuedConnection); connect(this, &StreamPlayerWindow::statusChanged, m_status, &QLabel::setText, Qt::QueuedConnection);
connect(SoundManager::instance(), &SoundManager::contextAboutToReset, this, [this] { connect(SoundManager::instance(), &SoundManager::contextAboutToReset, this, [this] {
m_audio.close(); m_playback->audioAvailable = false; m_audio.close(); m_playback->audioAvailable = false;
std::lock_guard<std::mutex> guard(m_playback->lock); std::lock_guard<std::mutex> guard(m_playback->lock); m_playback->resync = true;
m_playback->audios.clear(); m_playback->audioPts = -1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream audio context reset id=%llu", (unsigned long long)m_playback->id);
}); });
connect(SoundManager::instance(), &SoundManager::contextRestored, this, &StreamPlayerWindow::openAudio); connect(SoundManager::instance(), &SoundManager::contextRestored, this, &StreamPlayerWindow::openAudio);
connect(SoundManager::instance(), &SoundManager::devicesChanged, this, [this] { m_audio.close(); openAudio(); }); connect(SoundManager::instance(), &SoundManager::devicesChanged, this, [this] {
m_audio.close(); m_playback->audioAvailable = false;
{ std::lock_guard<std::mutex> guard(m_playback->lock); m_playback->resync = true; }
openAudio();
});
openAudio(); openAudio();
connect(&m_timer, &QTimer::timeout, this, &StreamPlayerWindow::tick); m_timer.start(15); connect(&m_timer, &QTimer::timeout, this, &StreamPlayerWindow::tick); m_timer.start(15);
m_playback->worker = std::thread(&StreamPlayerWindow::decode, this); m_playback->worker = std::thread(&StreamPlayerWindow::decode, this);
@@ -90,7 +103,7 @@ StreamPlayerWindow::StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t
StreamPlayerWindow::~StreamPlayerWindow() { stop(); } StreamPlayerWindow::~StreamPlayerWindow() { stop(); }
/* Output принадлежит GUI и закрывается до shutdown общего miniaudio-контекста. */ /* Hardware callback заполняет только ближайшую порцию; source time тишины остаётся неподвижным. */
void StreamPlayerWindow::openAudio() { void StreamPlayerWindow::openAudio() {
if (!m_playback->running || m_audio.initialized()) return; if (!m_playback->running || m_audio.initialized()) return;
ma_context* context = SoundManager::instance()->context(); ma_context* context = SoundManager::instance()->context();
@@ -103,63 +116,62 @@ void StreamPlayerWindow::openAudio() {
auto* pcm = static_cast<int16_t*>(output); std::fill(pcm, pcm + frames * 2, 0); auto* pcm = static_cast<int16_t*>(output); std::fill(pcm, pcm + frames * 2, 0);
auto& p = *m_playback; std::lock_guard<std::mutex> guard(p.lock); auto& p = *m_playback; std::lock_guard<std::mutex> guard(p.lock);
int64_t now = nowUs(); int64_t now = nowUs();
if (!p.prefillUntil || now < p.prefillUntil) return; p.deviceLead = std::max<int64_t>(0, m_audio.callbackTimeUs() - now);
unsigned written = 0; float volume = p.volume.load(); int64_t firstPts = -1; if (!p.sound || !p.started || p.resync) return;
while (written < frames && !p.audios.empty()) { p.audioFrameStart = stream_playback_audio_written(p.sound); p.audioDacAt = m_audio.callbackTimeUs();
auto& packet = p.audios.front(); int count = p.holdAudio ? 0 : stream_playback_audio_pull(p.sound, pcm, frames, p.ended ? INT64_MAX : p.ready);
if (firstPts < 0) firstPts = packet.pts + packet.offset * 1000000LL / 48000 - written * 1000000LL / 48000; if (count < 0) { p.resync = true; count = 0; }
unsigned count = std::min(frames - written, 960 - packet.offset); if ((unsigned)count < frames) {
for (unsigned i = 0; i < count * 2; i++) pcm[written * 2 + i] = static_cast<int16_t>(packet.pcm[packet.offset * 2 + i] * volume); stream_playback_audio_silence(p.sound, frames - (unsigned)count);
written += count; packet.offset += count; if (!p.ended) p.underruns++;
if (packet.offset == 960) p.audios.pop_front();
} }
if (firstPts >= 0) { p.audioPts = firstPts; p.audioAt = m_audio.callbackTimeUs(); } float volume = p.volume.load();
else if (p.audioPts >= 0) { for (int i = 0; i < count * 2; i++) pcm[i] = static_cast<int16_t>(pcm[i] * volume);
p.audioPts += m_audio.callbackTimeUs() - p.audioAt; p.audioAt = m_audio.callbackTimeUs(); if (p.ended && p.endFrames == UINT64_MAX && stream_playback_audio_empty(p.sound))
} p.endFrames = stream_playback_audio_written(p.sound);
if (written < frames && p.audioPts >= 0 && !p.ended) p.underruns++;
}); });
if (!initialized || !m_audio.start()) { if (!initialized || !m_audio.start()) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream audio device failed id=%llu", (unsigned long long)m_playback->id); DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop stream audio device failed id=%llu", (unsigned long long)m_playback->id);
m_audio.close(); m_playback->audioAvailable = false; return; m_audio.close(); m_playback->audioAvailable = false; return;
} }
m_playback->audioAvailable = true; m_playback->audioAvailable = true;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream audio output ready id=%llu rate=48000 channels=2", (unsigned long long)m_playback->id);
} }
/* GUI забирает последний готовый кадр по аппаратным часам, не накапливая Qt-сигналы с QImage. */ /* GUI держит только ближайшие готовые изображения; накопленный поток остаётся в RS-кольце. */
void StreamPlayerWindow::tick() { void StreamPlayerWindow::tick() {
auto& p = *m_playback; auto& p = *m_playback;
if (m_audio.needsRecovery()) { m_audio.close(); p.audioAvailable = false; openAudio(); } if (m_audio.needsRecovery()) {
m_audio.close(); p.audioAvailable = false;
{ std::lock_guard<std::mutex> guard(p.lock); p.resync = true; }
openAudio();
}
QImage image; bool finished = false, first = false; QImage image; bool finished = false, first = false;
{ {
std::lock_guard<std::mutex> guard(p.lock); std::lock_guard<std::mutex> guard(p.lock);
int64_t now = nowUs(), clock = p.clock(now); int64_t clock = p.clock(nowUs());
if (now >= p.prefillUntil) while (!p.videos.empty() && p.videos.front().pts <= clock) { while (p.started && !p.videos.empty() && p.videos.front().pts <= clock) {
p.videoBytes -= static_cast<size_t>(p.videos.front().image.sizeInBytes()); p.videoBytes -= static_cast<size_t>(p.videos.front().image.sizeInBytes());
image = std::move(p.videos.front().image); p.videos.pop_front(); p.lastShownPts = p.videos.front().pts; image = std::move(p.videos.front().image); p.videos.pop_front();
} }
if (!image.isNull()) first = p.shown++ == 0; if (!image.isNull()) first = p.shown++ == 0;
finished = p.ended && p.videos.empty() && (p.audios.empty() || !p.audioAvailable); bool audioDone = !p.activeAudio || !p.audioAvailable || (p.endFrames != UINT64_MAX && p.played(nowUs()) >= p.endFrames);
finished = p.ended && p.videos.empty() && audioDone;
} }
if (!image.isNull()) { if (!image.isNull()) {
m_render->setFrame(image); m_render->setFrame(image); m_status->setText(text("Live"));
m_status->setText(text("Live")); if (first) DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop first frame id=%llu %dx%d", (unsigned long long)p.id,
if (first) { image.width(), image.height());
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop first frame id=%llu %dx%d", (unsigned long long)p.id, image.width(), image.height());
}
emit framePresented(); emit framePresented();
} }
if (finished) { m_status->setText(text("Stream ended")); m_audio.close(); m_timer.stop(); } if (finished) { m_status->setText(text("Stream ended")); m_audio.close(); m_timer.stop(); }
} }
/* Закрытие может происходить до выполнения open actor; session_id отменяет именно эту попытку. */ /* Callback устройства прекращается до освобождения DSP; worker снимает только свой reader. */
void StreamPlayerWindow::stop() { void StreamPlayerWindow::stop() {
if (!m_playback->running.exchange(false) && !m_playback->worker.joinable()) return; if (!m_playback->running.exchange(false) && !m_playback->worker.joinable()) return;
m_timer.stop(); gui_stream_stop(m_playback->id); m_timer.stop(); gui_stream_stop(m_playback->id); m_audio.close(); m_playback->audioAvailable = false;
if (m_playback->worker.joinable()) m_playback->worker.join(); if (m_playback->worker.joinable()) m_playback->worker.join();
m_audio.close(); m_playback->audioAvailable = false; DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop player stopped id=%llu shown=%llu underruns=%llu",
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stream player stopped id=%llu shown=%llu underruns=%llu",
(unsigned long long)m_playback->id, (unsigned long long)m_playback->shown, (unsigned long long)m_playback->underruns); (unsigned long long)m_playback->id, (unsigned long long)m_playback->shown, (unsigned long long)m_playback->underruns);
} }
@@ -169,7 +181,7 @@ void StreamPlayerWindow::keyPressEvent(QKeyEvent* event) {
else QWidget::keyPressEvent(event); else QWidget::keyPressEvent(event);
} }
/* Worker владеет codec contexts. GAP/pressure сбрасывает refs и часы до следующего ключевого кадра. */ /* Worker декодирует по запросу ближайшего времени; общая минутная адаптация не зависит от сетевого cursor. */
void StreamPlayerWindow::decode() { void StreamPlayerWindow::decode() {
auto& p = *m_playback; auto& p = *m_playback;
#ifndef HAVE_FFMPEG #ifndef HAVE_FFMPEG
@@ -178,34 +190,77 @@ void StreamPlayerWindow::decode() {
#else #else
AVCodecContext* decoder = nullptr; AVFrame* frame = av_frame_alloc(); AVPacket* packet = av_packet_alloc(); AVCodecContext* decoder = nullptr; AVFrame* frame = av_frame_alloc(); AVPacket* packet = av_packet_alloc();
SwsContext* converter = nullptr; SwsContext* converter = nullptr;
opus_codec_decoder_t* opus = nullptr;
bool waitKey = true, opened = false; bool waitKey = true, opened = false;
int64_t deadline = nowUs() + 10000000, reportAt = nowUs(), expectedAudio = -1; uint32_t revision = 0;
uint64_t decoded = 0, discarded = 0, late = 0, pressureDropped = 0, previousPressure = 0, previousDecoded = 0, previousShown = 0; uint64_t decoded = 0, discarded = 0, reportedShown = 0;
uint64_t prerollSamples = 0; int64_t deadline = nowUs() + 10000000, report = nowUs();
uint32_t videoRevision = 0; auto reset = [&](bool request) {
auto reset = [&] { avcodec_free_context(&decoder); waitKey = true;
avcodec_free_context(&decoder); {
opus_codec_decoder_destroy(opus); opus = nullptr; expectedAudio = -1; waitKey = true; prerollSamples = 0; std::lock_guard<std::mutex> guard(p.lock);
std::lock_guard<std::mutex> guard(p.lock); stream_playback_audio_destroy(p.sound); p.sound = nullptr;
p.videos.clear(); p.videoBytes = 0; p.audios.clear(); p.prefillUntil = 0; p.videoBase = p.audioPts = -1; p.videos.clear(); p.videoBytes = 0; p.started = false; p.position = p.ready = -1;
emit statusChanged(text("Waiting for a key frame…")); p.audioDacAt = 0; p.audioFrameStart = 0; p.holdAudio = p.resync = false; p.endFrames = UINT64_MAX;
}
if (request && !gui_stream_resync(p.id)) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop playback resync refused");
deadline = nowUs() + 10000000; emit statusChanged(text("Waiting for a key frame…"));
}; };
auto fail = [&](const QString& reason) { auto fail = [&](const QString& reason) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer failed id=%llu: %s", (unsigned long long)p.id, qPrintable(reason)); DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop viewer failed id=%llu: %s", (unsigned long long)p.id, qPrintable(reason));
emit statusChanged(reason); emit statusChanged(reason);
}; };
auto observe = [&] {
stream_media_status status{};
if (!gui_stream_status(p.id, status)) return;
std::lock_guard<std::mutex> guard(p.lock);
p.media = status;
if (!status.initialized) return;
if (!p.started) {
p.activeAudio = status.audio_enabled && p.audioAvailable;
if (p.activeAudio) p.sound = stream_playback_audio_create(status.start_us);
if (p.activeAudio && !p.sound) { p.running = false; return; }
p.position = status.start_us; p.clockAt = nowUs(); p.started = true;
stream_playback_clock_reset(&p.adaptation, p.id);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop playback starts id=%llu pts=%lld fixed_prefill=0",
(unsigned long long)p.id, (long long)p.position);
}
p.ready = p.activeAudio ? status.ready_us : status.video_us;
};
if (!frame || !packet) fail(text("Stream decoder allocation failed")); 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)) fail(text("Could not connect to the stream"));
else while (p.running) { else while (p.running) {
bool resync;
{ std::lock_guard<std::mutex> guard(p.lock); resync = p.resync; }
if (resync) reset(true);
observe();
int64_t position, videoAhead, audioAhead;
{
std::lock_guard<std::mutex> guard(p.lock);
int64_t now = nowUs(); position = p.started ? p.clock(now) : -1;
videoAhead = p.videos.size() < 2 ? 80000 : -1;
audioAhead = !p.activeAudio || !p.sound || stream_playback_audio_needs_input(p.sound) ? p.deviceLead + 80000 : -1;
if (p.started && p.ready >= 0) {
uint64_t played = p.played(now);
p.holdAudio = !p.videos.empty() && p.videos.front().pts < position - 40000;
int starved = p.holdAudio || (p.sound && stream_playback_audio_starved(p.sound, played));
int64_t guardUs = std::max<int64_t>(33333, p.activeAudio ? p.deviceLead + 10000 : 33333);
int64_t pending = p.sound ? stream_playback_audio_pending_gain(p.sound, played) : 0;
if (stream_playback_clock_update(&p.adaptation, now, std::max(p.position, position), p.ready,
guardUs, pending, !p.ended && starved, &p.timing) < 0) p.resync = true;
else {
p.position = std::max(p.position, position); p.clockAt = now;
if (p.sound) stream_playback_audio_set_tempo(p.sound, p.timing.tempo_milli);
}
}
}
StreamRecord record; StreamRecord record;
if (!gui_stream_pull(p.id, record)) { if (!gui_stream_pull(p.id, record, position, videoAhead, audioAhead)) {
if (!gui_stream_ready()) { fail(text("Chat service is stopped")); break; } if (!gui_stream_ready()) { fail(text("Chat service is stopped")); break; }
if ((!opened || waitKey) && nowUs() > deadline) { fail(text("Stream connection timed out")); break; } if ((!opened || waitKey) && nowUs() > deadline) { fail(text("Stream connection timed out")); break; }
std::this_thread::sleep_for(std::chrono::milliseconds(5)); continue; std::this_thread::sleep_for(std::chrono::milliseconds(5)); continue;
} }
if (record.kind == StreamRecordKind::Opened) { opened = true; continue; } if (record.kind == StreamRecordKind::Opened) { opened = true; continue; }
if (record.kind == StreamRecordKind::Gap) { reset(); deadline = nowUs() + 10000000; continue; } if (record.kind == StreamRecordKind::Gap) { reset(false); continue; }
if (record.kind == StreamRecordKind::Closed) { if (record.kind == StreamRecordKind::Closed) {
const char* reason = record.reason == STREAM_CLOSE_SOURCE_CHANGED ? "The stream restarted. Select a source again." const char* reason = record.reason == STREAM_CLOSE_SOURCE_CHANGED ? "The stream restarted. Select a source again."
: record.reason == STREAM_CLOSE_MEMBER ? "The source is no longer a channel member" : record.reason == STREAM_CLOSE_MEMBER ? "The source is no longer a channel member"
@@ -215,45 +270,22 @@ void StreamPlayerWindow::decode() {
} }
if (record.kind == StreamRecordKind::End) { if (record.kind == StreamRecordKind::End) {
std::lock_guard<std::mutex> guard(p.lock); p.ended = true; std::lock_guard<std::mutex> guard(p.lock); p.ended = true;
if (p.videoBase < 0) p.audios.clear(); if (p.sound) stream_playback_audio_finish(p.sound);
break; break;
} }
observe();
if (record.pts > INT64_MAX) { fail(text("Invalid stream timestamp")); break; } if (record.pts > INT64_MAX) { fail(text("Invalid stream timestamp")); break; }
if (record.kind == StreamRecordKind::Audio) { if (record.kind == StreamRecordKind::Audio) {
if (!p.audioAvailable) continue; int result = 0;
if (!opus) opus = opus_codec_decoder_create(48000, 2); { std::lock_guard<std::mutex> guard(p.lock);
if (!opus) { fail(text("Stream audio decoder failed")); break; } if (p.sound) result = stream_playback_audio_feed(p.sound, (int64_t)record.pts,
if (expectedAudio >= 0 && (static_cast<int64_t>(record.pts) < expectedAudio - 10000 reinterpret_cast<const uint8_t*>(record.data.constData()), (size_t)record.data.size()); }
|| static_cast<int64_t>(record.pts) > expectedAudio + 100000)) { if (result == -2) reset(true);
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop audio discontinuity id=%llu expected=%lld pts=%llu", else if (result < 0) { fail(text("Stream audio decoder failed")); break; }
(unsigned long long)p.id, (long long)expectedAudio, (unsigned long long)record.pts);
reset(); deadline = nowUs() + 10000000; continue;
}
int missing = expectedAudio < 0 ? 0 : static_cast<int>((static_cast<int64_t>(record.pts) - expectedAudio + 10000) / 20000);
bool failed = false, pressure = false;
for (int i = 0; i <= missing; i++) {
Playback::Audio audio{}; audio.pts = i == missing ? static_cast<int64_t>(record.pts) : expectedAudio + i * 20000;
int count = opus_codec_decode(opus, i == missing ? reinterpret_cast<const uint8_t*>(record.data.constData()) : nullptr,
i == missing ? record.data.size() : 0, audio.pcm.data(), 960);
if (count != 960) { fail(text("Stream audio decoder failed")); failed = true; break; }
std::lock_guard<std::mutex> guard(p.lock);
if (p.audios.size() >= 25) {
if (p.videoBase >= 0) { pressure = true; break; }
prerollSamples += 960; p.audios.pop_front();
}
p.audios.push_back(std::move(audio));
}
if (failed) break;
if (pressure) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop decoded audio pressure id=%llu; wait KEY", (unsigned long long)p.id);
reset(); deadline = nowUs() + 10000000;
} else expectedAudio = static_cast<int64_t>(record.pts) + 20000;
continue; continue;
} }
if (videoRevision != record.format.revision) { if (revision != record.format.revision) {
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop video revision=%u->%u dimensions=%ux%u rotation=%u; preserve audio", avcodec_free_context(&decoder); waitKey = true; revision = record.format.revision;
videoRevision, record.format.revision, record.format.width, record.format.height, record.format.rotation);
avcodec_free_context(&decoder); av_frame_unref(frame); waitKey = true; videoRevision = record.format.revision;
{ std::lock_guard<std::mutex> guard(p.lock); p.videos.clear(); p.videoBytes = 0; } { std::lock_guard<std::mutex> guard(p.lock); p.videos.clear(); p.videoBytes = 0; }
deadline = nowUs() + 10000000; deadline = nowUs() + 10000000;
} }
@@ -263,87 +295,50 @@ void StreamPlayerWindow::decode() {
decoder = codec ? avcodec_alloc_context3(codec) : nullptr; decoder = codec ? avcodec_alloc_context3(codec) : nullptr;
if (decoder) { decoder->thread_count = 1; decoder->flags |= AV_CODEC_FLAG_LOW_DELAY; } if (decoder) { decoder->thread_count = 1; decoder->flags |= AV_CODEC_FLAG_LOW_DELAY; }
if (!decoder || avcodec_open2(decoder, codec, nullptr) < 0) { fail(text("Stream video decoder failed")); break; } if (!decoder || avcodec_open2(decoder, codec, nullptr) < 0) { fail(text("Stream video decoder failed")); break; }
decoder->pkt_timebase = AVRational{1, 1000000}; decoder->pkt_timebase = AVRational{1, 1000000}; waitKey = false;
waitKey = false;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop H264 decoder started id=%llu", (unsigned long long)p.id);
} }
av_packet_unref(packet); av_packet_unref(packet);
if (av_new_packet(packet, record.data.size()) < 0) { fail(text("Stream decoder allocation failed")); break; } if (av_new_packet(packet, record.data.size()) < 0) { fail(text("Stream decoder allocation failed")); break; }
std::memcpy(packet->data, record.data.constData(), static_cast<size_t>(record.data.size())); std::memcpy(packet->data, record.data.constData(), (size_t)record.data.size());
packet->pts = packet->dts = static_cast<int64_t>(record.pts); packet->pts = packet->dts = (int64_t)record.pts;
int sent = avcodec_send_packet(decoder, packet); int sent = avcodec_send_packet(decoder, packet);
if (sent < 0) { if (sent < 0) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop H264 packet rejected result=%d", sent); reset(true); continue; }
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop H264 packet rejected result=%d id=%llu", sent, (unsigned long long)p.id);
reset(); deadline = nowUs() + 10000000; continue;
}
int received; int received;
while ((received = avcodec_receive_frame(decoder, frame)) == 0) { while ((received = avcodec_receive_frame(decoder, frame)) == 0) {
decoded++; decoded++;
if (frame->width <= 0 || frame->height <= 0 || frame->width > 4096 || frame->height > 4096) { int64_t pts = frame->best_effort_timestamp == AV_NOPTS_VALUE ? (int64_t)record.pts : frame->best_effort_timestamp;
fail(text("Invalid stream video dimensions")); p.running = false; break; { std::lock_guard<std::mutex> guard(p.lock); if (pts < p.clock(nowUs()) - 100000) { discarded++; continue; } }
} converter = sws_getCachedContext(converter, frame->width, frame->height, (AVPixelFormat)frame->format,
int64_t pts = frame->best_effort_timestamp == AV_NOPTS_VALUE ? static_cast<int64_t>(record.pts) : frame->best_effort_timestamp;
{
std::lock_guard<std::mutex> guard(p.lock);
if (p.videoBase < 0) {
while (!p.audios.empty() && p.audios.front().pts + 20000 <= pts) {
prerollSamples += 960; p.audios.pop_front();
}
if (!p.audios.empty() && p.audios.front().pts < pts) {
p.audios.front().offset = static_cast<unsigned>((pts - p.audios.front().pts) * 48000 + 999999) / 1000000;
prerollSamples += p.audios.front().offset;
}
p.prefillUntil = nowUs() + 150000; p.videoBase = pts; p.videoAt = p.prefillUntil;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop playback start id=%llu pts=%lld prefill_ms=150 preroll_frames=%llu audioQ=%zu",
(unsigned long long)p.id, (long long)pts, (unsigned long long)prerollSamples, p.audios.size());
prerollSamples = 0;
}
if (pts < p.clock(nowUs()) - 100000) { discarded++; late++; continue; }
}
converter = sws_getCachedContext(converter, frame->width, frame->height, static_cast<AVPixelFormat>(frame->format),
frame->width, frame->height, AV_PIX_FMT_RGB24, SWS_BILINEAR, nullptr, nullptr, nullptr); frame->width, frame->height, AV_PIX_FMT_RGB24, SWS_BILINEAR, nullptr, nullptr, nullptr);
QImage image(frame->width, frame->height, QImage::Format_RGB888); QImage image(frame->width, frame->height, QImage::Format_RGB888);
if (!converter || image.isNull()) { fail(text("Stream decoder allocation failed")); p.running = false; break; } if (!converter || image.isNull()) { fail(text("Stream video conversion failed")); p.running = false; break; }
uint8_t* destination[] = {image.bits(), nullptr, nullptr, nullptr}; uint8_t* destination[] = {image.bits(), nullptr, nullptr, nullptr};
int strides[] = {static_cast<int>(image.bytesPerLine()), 0, 0, 0}; int strides[] = {(int)image.bytesPerLine(), 0, 0, 0};
if (sws_scale(converter, frame->data, frame->linesize, 0, frame->height, destination, strides) != frame->height) { if (sws_scale(converter, frame->data, frame->linesize, 0, frame->height, destination, strides) != frame->height) {
fail(text("Stream video conversion failed")); p.running = false; break; fail(text("Stream video conversion failed")); p.running = false; break;
} }
if (record.format.rotation) image = image.transformed(QTransform().rotate(record.format.rotation)); if (record.format.rotation) image = image.transformed(QTransform().rotate(record.format.rotation));
std::lock_guard<std::mutex> guard(p.lock); std::lock_guard<std::mutex> guard(p.lock);
size_t bytes = static_cast<size_t>(image.sizeInBytes()); p.videoBytes += (size_t)image.sizeInBytes(); p.videos.push_back({std::move(image), pts});
// Будущие кадры ждут аудиочасы; при пределе памяти сохраняем ближайшие к показу.
if (p.videos.size() >= 128 || p.videoBytes + bytes > 256u * 1024u * 1024u) {
discarded++; pressureDropped++; continue;
}
p.videoBytes += bytes;
p.videos.push_back({std::move(image), pts});
} }
if (received != AVERROR(EAGAIN) && received != AVERROR_EOF && p.running) { if (received != AVERROR(EAGAIN) && received != AVERROR_EOF && p.running) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop H264 decode rejected result=%d id=%llu", received, (unsigned long long)p.id); DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop H264 decode rejected result=%d", received); reset(true);
reset(); deadline = nowUs() + 10000000;
} }
if (nowUs() - reportAt >= 1000000) { if (nowUs() - report >= 1000000) {
int64_t now = nowUs(); double seconds = (now - reportAt) / 1000000.0; reportAt = now; int64_t now = nowUs(); double seconds = (now - report) / 1000000.0; report = now;
std::lock_guard<std::mutex> guard(p.lock); int64_t clock = p.clock(now); std::lock_guard<std::mutex> guard(p.lock);
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, DEBUG_DEBUG(DEBUG_CATEGORY_STREAM,
"desktop viewer id=%llu decoded=%llu discarded=%llu late=%llu pressure=%llu shown=%llu decodeFps=%.1f showFps=%.1f " "desktop viewer id=%llu decoded=%llu discarded=%llu shown=%llu showFps=%.1f videoQ=%zu "
"videoQ=%zu videoBytes=%zu firstAheadUs=%lld lastAheadUs=%lld audioQ=%zu audioAtDeltaUs=%lld underruns=%llu", "clock=%lld reserve_us=%lld minimum_us=%lld tempo=%d lead_us=%lld underruns=%llu",
(unsigned long long)p.id, (unsigned long long)decoded, (unsigned long long)discarded, (unsigned long long)p.id, (unsigned long long)decoded, (unsigned long long)discarded,
(unsigned long long)late, (unsigned long long)pressureDropped, (unsigned long long)p.shown, (unsigned long long)p.shown, (p.shown - reportedShown) / seconds, p.videos.size(), (long long)p.position,
(decoded - previousDecoded) / seconds, (p.shown - previousShown) / seconds, p.videos.size(), p.videoBytes, (long long)p.timing.reserve_us, (long long)p.timing.minimum_us, p.timing.tempo_milli,
(long long)(p.videos.empty() ? 0 : p.videos.front().pts - clock), (long long)p.deviceLead, (unsigned long long)p.underruns);
(long long)(p.videos.empty() ? 0 : p.videos.back().pts - clock), p.audios.size(), reportedShown = p.shown;
(long long)(p.audioPts < 0 ? 0 : now - p.audioAt), (unsigned long long)p.underruns);
if (pressureDropped != previousPressure)
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop video queue pressure id=%llu dropped=%llu; preserve due frames",
(unsigned long long)p.id, (unsigned long long)pressureDropped);
previousDecoded = decoded; previousShown = p.shown; previousPressure = pressureDropped;
} }
} }
avcodec_free_context(&decoder); av_frame_free(&frame); av_packet_free(&packet); avcodec_free_context(&decoder); av_frame_free(&frame); av_packet_free(&packet);
if (converter) sws_freeContext(converter); if (converter) sws_freeContext(converter);
opus_codec_decoder_destroy(opus);
#endif #endif
gui_stream_stop(p.id); gui_stream_stop(p.id);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop decoder released id=%llu", (unsigned long long)p.id); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop decoder released id=%llu", (unsigned long long)p.id);