From 9500581e0b5393e398cb2e31e485795e55fb603c Mon Sep 17 00:00:00 2001 From: evgeny Date: Fri, 16 Oct 2026 16:11:04 +0200 Subject: [PATCH] Gate stream encoding on microphone input and CPU consent --- tools/chatgui/src/stream_publish_engine.cpp | 121 +++++++++++++++----- tools/chatgui/src/stream_publish_engine.h | 3 + tools/chatgui/src/stream_publish_window.cpp | 20 ++++ 3 files changed, 114 insertions(+), 30 deletions(-) diff --git a/tools/chatgui/src/stream_publish_engine.cpp b/tools/chatgui/src/stream_publish_engine.cpp index ebf8950d..0eb6ff4b 100644 --- a/tools/chatgui/src/stream_publish_engine.cpp +++ b/tools/chatgui/src/stream_publish_engine.cpp @@ -60,9 +60,10 @@ struct StreamPublishEngine::Impl { struct Audio { int64_t pts = 0, created = 0; uint64_t sequence = 0; unsigned frames = 0; int16_t pcm[1920]{}; }; std::mutex lock; std::condition_variable wake; - std::thread videoThread, audioThread, probeThread; + std::thread videoThread, audioThread; std::atomic running{false}, muted{false}; std::atomic audioDropped{0}, audioMissing{0}; + std::atomic lastAudioAt{0}; AudioEventQueue audio; uint64_t id = 0, group = 0, generation = 0, sourceGeneration = 0, confirmedGeneration = 0, frameGeneration = 0, audioSequence = 0; int64_t origin = 0, started = 0, framePts = -1, backendOrigin = -1, backendAnchor = -1; @@ -71,7 +72,10 @@ struct StreamPublishEngine::Impl { int channels = 2; uint64_t capturedFrames = 0; int64_t captureReport = 0; - bool probeReady = false, probeDelivered = false, hasGood = false, rollingBack = false; + bool hasGood = false, rollingBack = false, videoReady = false; + uint64_t softwareRequest = 0; + int softwareCodec = 0; + int64_t microphoneStarted = 0; StreamPublishConfig wanted, good; QList codecs; stream_publisher_status stats{}; @@ -118,21 +122,15 @@ StreamPublishEngine::StreamPublishEngine(QObject* parent) : QObject(parent), m(s }); m_previewTimer.start(); #endif - try { - m->probeThread = std::thread([this] { - auto codecs = supported() ? LiveVideoEncoder::available() : QList(); - std::lock_guard guard(m->lock); m->codecs = std::move(codecs); m->probeReady = true; - }); - } catch (const std::system_error& error) { - DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop codec probe worker failed: %s", error.what()); m->probeReady = true; - } + m->codecs = supported() ? LiveVideoEncoder::available() : QList(); auto* sound = SoundManager::instance(); - connect(sound, &SoundManager::contextAboutToReset, this, [this] { m_capture.close(); }); - connect(sound, &SoundManager::contextRestored, this, [this] { if (m->running && m->wanted.audio) openMicrophone(); }); + connect(sound, &SoundManager::contextAboutToReset, this, [this] { + if (m->running && m->wanted.audio) publicationFailed(m->id, text("The audio device was disconnected. Select a microphone and start again.")); + m_capture.close(); // Контекст нельзя освободить до завершения фонового закрытия устройства. + }); } StreamPublishEngine::~StreamPublishEngine() { stop(); m_timer.stop(); m_previewTimer.stop(); - if (m->probeThread.joinable()) m->probeThread.join(); } bool StreamPublishEngine::supported() { #ifdef HAVE_STREAM_CAPTURE @@ -385,6 +383,7 @@ bool StreamPublishEngine::start(uint64_t group, const StreamPublishConfig& confi } { std::lock_guard guard(m->lock); m->wanted = config; m->good = {}; m->hasGood = m->rollingBack = false; m->generation++; m->confirmedGeneration = 0; + m->videoReady = false; m->softwareRequest = 0; m->softwareCodec = 0; m->lastAudioAt = 0; m->origin = m->started = monotonicUs(); m->stats = {}; } m->id = QRandomGenerator::global()->generate64(); if (!m->id) m->id = 1; m->group = group; m->audio.reset(); m->audioSequence = 0; m->muted = false; @@ -398,9 +397,13 @@ bool StreamPublishEngine::start(uint64_t group, const StreamPublishConfig& confi m->videoThread = std::thread(&StreamPublishEngine::videoLoop, this, config); if (config.audio) m->audioThread = std::thread(&StreamPublishEngine::audioLoop, this); } catch (const std::system_error& error) { publicationFailed(m->id, QString::fromUtf8(error.what())); return false; } - if (!openSource(config) || (config.audio && !openMicrophone())) { + if (!openSource(config)) { sourceFailed(m->generation, text("Could not open the selected capture device")); return false; } + if (config.audio && !openMicrophone()) { + publicationFailed(m->id, text("Could not start the microphone. Select another microphone or disable stream audio.")); return false; + } + m->microphoneStarted = monotonicUs(); emit stateChanged(config.source.id == "portal" ? text("Select a screen in the system dialog") : text("Starting stream…")); emit configurationChanged(); return true; } @@ -420,7 +423,7 @@ bool StreamPublishEngine::reconfigure(const StreamPublishConfig& config) { DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop cursor selected enabled=%d supported=%d", config.showCursor, cursorSupported(config.source)); bool sourceChange = !config.source.sameSource(previous.source) || (config.source.kind == StreamVideoSource::Camera && (config.video.width != previous.video.width || config.video.height != previous.video.height || config.video.fps != previous.video.fps)); - { std::lock_guard guard(m->lock); m->wanted = config; m->generation++; m->started = monotonicUs(); + { std::lock_guard guard(m->lock); m->wanted = config; m->generation++; m->started = monotonicUs(); m->softwareRequest = 0; #ifdef HAVE_STREAM_CAPTURE if (!sourceChange) { m->frame = m->previewFrame; m->frameGeneration = m->generation; } #endif @@ -453,14 +456,31 @@ void StreamPublishEngine::publicationFailed(uint64_t id, const QString& error) { /* Сначала callbacks, затем workers; FIN не обгоняет уже принятые access units. */ void StreamPublishEngine::stop() { if (!m->running.exchange(false)) return; - closeSource(); m_capture.close(); m->wake.notify_all(); + { std::lock_guard guard(m->lock); m->softwareRequest = 0; } + m->wake.notify_all(); closeSource(); m_capture.closeAsync(); if (m->videoThread.joinable()) m->videoThread.join(); if (m->audioThread.joinable()) m->audioThread.join(); - m_microphone.release(); gui_stream_publish_stop(m->id); + gui_stream_publish_stop(m->id); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop capture stopped id=%llu; draining publisher", (unsigned long long)m->id); emit stateChanged(text("Finishing stream…")); } +uint64_t StreamPublishEngine::softwareEncodingRequest() const { + std::lock_guard guard(m->lock); return m->softwareRequest; +} + +/* Разрешение относится к кодеку текущей трансляции; устаревший ответ не запускает новый режим. */ +void StreamPublishEngine::answerSoftwareEncoding(uint64_t generation, bool allow) { + { std::lock_guard guard(m->lock); + if (!m->running || !generation || m->softwareRequest != generation || m->generation != generation) { + DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop CPU response ignored generation=%llu", (unsigned long long)generation); return; + } + if (allow) { m->softwareCodec = m->wanted.video.codec; m->softwareRequest = 0; m->started = monotonicUs(); } } + DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop CPU encoding decision generation=%llu allow=%d", (unsigned long long)generation, allow); + if (!allow) sourceFailed(generation, text("Hardware encoding is unavailable. CPU encoding was declined.")); + else { m->wake.notify_all(); emit stateChanged(text("Starting stream…")); } +} + bool StreamPublishEngine::openMicrophone() { return m_capture.open(config().microphone, AudioCapture::Role::Stream, 2, m->id, [this](int channels, int native) { m->audioAnchor = -1; m->audioFrames = 0; m->channels = channels; @@ -475,12 +495,15 @@ void StreamPublishEngine::captureAudio(const AudioCapture::Frame& frame) { const auto* pcm = frame.pcm; unsigned frames = frame.frames; if (!m->running) return; + if (!pcm || !frames || frame.missing) { m->audioMissing++; return; } + m->lastAudioAt.store(monotonicUs(), std::memory_order_release); + m->wake.notify_all(); + { std::lock_guard guard(m->lock); if (!m->videoReady) return; } int64_t first = frame.timeUs - m->origin; if (frame.measured || m->audioAnchor < 0 || frame.missing) { m->audioAnchor = first; m->audioFrames = 0; } first = m->audioAnchor + (int64_t)m->audioFrames * 1000000 / 48000; m->audioFrames += frames; - if (!pcm) m->audioMissing++; for (unsigned offset = 0; offset < frames;) { Impl::Audio record; record.frames = std::min(960u, frames - offset); record.pts = std::max(0, first + (int64_t)offset * 1000000 / 48000); record.created = monotonicUs(); @@ -497,6 +520,9 @@ void StreamPublishEngine::captureAudio(const AudioCapture::Frame& frame) { void StreamPublishEngine::audioLoop() { const uint64_t id = m->id; + { std::unique_lock lock(m->lock); + m->wake.wait(lock, [this] { return !m->running || m->videoReady; }); } + if (!m->running) return; auto* encoder = opus_codec_encoder_create(48000, 2); auto* compressor = audio_compressor_create(); int result = encoder && compressor ? opus_codec_encoder_application_set(encoder, OPUS_CODEC_APP_AUDIO) : -1; if (result >= 0) result = opus_codec_encoder_bitrate_set(encoder, 128000); @@ -562,6 +588,9 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { while (m->running) { StreamPublishConfig config; uint64_t generation; QVideoFrame frame; int64_t capture; { std::unique_lock lock(m->lock); config = m->wanted; generation = m->generation; + if (config.audio && !m->lastAudioAt.load(std::memory_order_acquire)) { + m->wake.wait_for(lock, std::chrono::milliseconds(5)); continue; + } int64_t now = monotonicUs(); if (next > now) { m->wake.wait_for(lock, std::chrono::microseconds(next - now)); continue; } if (m->frameGeneration == generation) { frame = m->frame; capture = m->framePts; m->frame = {}; } @@ -603,8 +632,31 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { }); } else { auto candidate = std::make_unique(); - if (!candidate->open(config.video)) { - QString error = candidate->error(); + bool opened = candidate->open(config.video, LiveVideoEncoder::Backend::HardwareOnly); + if (!opened && LiveVideoEncoder::available(LiveVideoEncoder::Backend::SoftwareOnly).contains(config.video.codec)) { + QString details = candidate->error(); + std::unique_lock lock(m->lock); + if (!m->running || generation != m->generation) continue; + if (m->softwareCodec != config.video.codec) { + m->softwareRequest = generation; + DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "desktop CPU confirmation required generation=%llu codec=%d reason=%s", + (unsigned long long)generation, config.video.codec, qPrintable(details)); + QMetaObject::invokeMethod(this, [this, generation, codec = config.video.codec, details] { + if (softwareEncodingRequest() != generation || !m->running) return; + emit stateChanged(text("Waiting for permission to encode on the CPU…")); + emit softwareEncodingRequested(generation, LiveVideoEncoder::codecName(codec), details); + }); + m->wake.wait(lock, [this, generation] { + return !m->running || m->generation != generation || m->softwareRequest != generation; + }); + } + if (!m->running || generation != m->generation) continue; + lock.unlock(); + opened = candidate->open(config.video, LiveVideoEncoder::Backend::SoftwareOnly); + } + if (!opened) { + QString error = text("Could not start video encoding for %1. Choose another codec or video mode.") + .arg(LiveVideoEncoder::codecName(config.video.codec)); QMetaObject::invokeMethod(this, [this, generation, error] { sourceFailed(generation, error); }); std::unique_lock lock(m->lock); m->wake.wait_for(lock, std::chrono::milliseconds(100), [this, generation] { @@ -615,6 +667,10 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { // Начальный формат publisher уже зарегистрирован; уточнение размера требует новой revision даже до первого KEY. revision = revision ? revision + 1 : config.video == initial.video ? 1 : 2; encoder = std::move(candidate); current = generation; encoding = config; lastImage = {}; lastKey = -1000000; + { std::lock_guard guard(m->lock); + if (!m->running || generation != m->generation) continue; + m->videoReady = true; } + m->wake.notify_all(); DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop video prepared generation=%llu revision=%u encoder=%s", (unsigned long long)generation, revision, qPrintable(encoder->name())); } @@ -680,26 +736,34 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { } void StreamPublishEngine::tick() { - bool ready; - { std::lock_guard guard(m->lock); ready = m->probeReady && !m->probeDelivered; m->probeDelivered |= ready; } - if (ready) { if (m->probeThread.joinable()) m->probeThread.join(); emit codecsChanged(); } if (!active()) return; + if (!m->running && m_capture.closing()) return; stream_publisher_status status; if (!gui_stream_publish_status(m->id, status)) { - stop(); m->id = m->group = 0; emit stateChanged(text("Chat service is stopped")); return; + stop(); + if (m_capture.closing()) return; + m_microphone.release(); m->id = m->group = 0; emit stateChanged(text("Chat service is stopped")); return; } { std::lock_guard guard(m->lock); m->stats = status; } emit telemetryChanged(status); if (status.state == STREAM_PUBLISHER_ENDED || status.state == STREAM_PUBLISHER_FAILED) { - stop(); m->id = m->group = 0; + stop(); + if (m_capture.closing()) return; + m_microphone.release(); m->id = m->group = 0; if (status.state == STREAM_PUBLISHER_FAILED) emit errorOccurred(text("Stream publication failed")); emit stateChanged(text("Stopped")); emit configurationChanged(); return; } if (m->running) { + int64_t audioAt = m->lastAudioAt.load(std::memory_order_acquire); + if (m->wanted.audio && monotonicUs() - (audioAt ? audioAt : m->microphoneStarted) >= 1000000) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "desktop microphone has no PCM id=%llu received=%d idle_us=%lld", + (unsigned long long)m->id, audioAt != 0, (long long)(monotonicUs() - (audioAt ? audioAt : m->microphoneStarted))); + publicationFailed(m->id, text("The microphone is not delivering audio. Select another microphone or disable stream audio.")); return; + } bool confirmed, waitingSelection; int64_t started; { std::lock_guard guard(m->lock); confirmed = m->confirmedGeneration == m->generation; started = m->started; waitingSelection = m->wanted.source.id == "portal" && m->framePts < 0; } - if (!confirmed && !waitingSelection && monotonicUs() - started > 8000000) + if (!confirmed && !waitingSelection && !softwareEncodingRequest() && monotonicUs() - started > 8000000) sourceFailed(m->generation, text("Timed out waiting for the first video frame")); { std::lock_guard guard(m->lock); if (monotonicUs() - m->captureReport >= 1000000) { @@ -708,12 +772,9 @@ void StreamPublishEngine::tick() { (unsigned long long)m->sourceGeneration, (unsigned long long)m->generation, (unsigned long long)m->confirmedGeneration, (unsigned long long)m->capturedFrames, (long long)m->framePts); } } - if (m->wanted.audio) { - if (!m_capture.subscribed()) openMicrophone(); else m_capture.recover(); - } } unsigned dropped = m->audioDropped.exchange(0); if (dropped) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop audio capture discarded blocks=%u", dropped); unsigned missing = m->audioMissing.exchange(0); - if (missing) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop microphone callbacks lack PCM count=%u; sent silence", missing); + if (missing) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop microphone callbacks lack PCM count=%u; waiting for input", missing); } diff --git a/tools/chatgui/src/stream_publish_engine.h b/tools/chatgui/src/stream_publish_engine.h index 57bcbd52..9b56a4e7 100644 --- a/tools/chatgui/src/stream_publish_engine.h +++ b/tools/chatgui/src/stream_publish_engine.h @@ -56,6 +56,8 @@ public: static bool cursorSupported(const StreamVideoSource& source); void setMuted(bool muted); bool muted() const; + uint64_t softwareEncodingRequest() const; + void answerSoftwareEncoding(uint64_t generation, bool allow); signals: void codecsChanged(); void sourcesChanged(); @@ -65,6 +67,7 @@ signals: void telemetryChanged(const stream_publisher_status& status); void errorOccurred(const QString& error); void audioChannelsChanged(int channels); + void softwareEncodingRequested(uint64_t generation, const QString& codec, const QString& details); private: struct Impl; std::unique_ptr m; diff --git a/tools/chatgui/src/stream_publish_window.cpp b/tools/chatgui/src/stream_publish_window.cpp index 9693fb5b..34d1758b 100644 --- a/tools/chatgui/src/stream_publish_window.cpp +++ b/tools/chatgui/src/stream_publish_window.cpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -118,6 +119,25 @@ StreamPublishWindow::StreamPublishWindow(StreamPublishEngine* engine, uint64_t g connect(engine, &StreamPublishEngine::errorOccurred, this, [this](const QString& value) { m_status->setProperty("streamError", value); m_status->setText(value); }); + connect(engine, &StreamPublishEngine::softwareEncodingRequested, this, + [this](uint64_t generation, const QString& codec, const QString& details) { + auto* question = new QMessageBox(QMessageBox::Warning, text("Hardware encoding is unavailable"), + text("Hardware encoding for %1 is unavailable in the selected video mode. Check the graphics driver or choose another codec.\n\n" + "Use CPU encoding instead? This may cause high CPU usage and slow down the computer.").arg(codec), + QMessageBox::Yes | QMessageBox::No, this); + question->setObjectName(QStringLiteral("streamCpuConfirmation")); + question->setDefaultButton(QMessageBox::No); question->setEscapeButton(QMessageBox::No); + question->setDetailedText(details); question->setAttribute(Qt::WA_DeleteOnClose); + auto* engine = m_engine; + connect(question, &QMessageBox::finished, engine, [engine, generation](int result) { + engine->answerSoftwareEncoding(generation, result == QMessageBox::Yes); + }); + connect(question, &QObject::destroyed, engine, [engine, generation] { engine->answerSoftwareEncoding(generation, false); }); + connect(engine, &StreamPublishEngine::stateChanged, question, [engine, generation, question] { + if (engine->softwareEncodingRequest() != generation) question->reject(); + }); + question->open(); + }); connect(engine, &StreamPublishEngine::audioChannelsChanged, this, [this](int channels) { m_inputChannels->setText(channels == 2 ? text("Stereo") : text("Mono")); }); connect(engine, &StreamPublishEngine::telemetryChanged, this, [this](const stream_publisher_status& stats) { m_stats->setText(text("Video queue: %1 · Inflight: %2 · Send buffer: %3\nACK, all connections, last second: %4")