Gate stream encoding on microphone input and CPU consent

2 days ago · 9500581e0b
3 changed files with 114 additions and 30 deletions
--- 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<bool> running{false}, muted{false};
    std::atomic<unsigned> audioDropped{0}, audioMissing{0};
+    std::atomic<int64_t> lastAudioAt{0};
    AudioEventQueue<Audio, 32> 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<int> 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<int>();
-            std::lock_guard<std::mutex> 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<int>();
    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<std::mutex> 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<std::mutex> guard(m->lock); m->wanted = config; m->generation++; m->started = monotonicUs();
+    { std::lock_guard<std::mutex> 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<std::mutex> 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<std::mutex> guard(m->lock); return m->softwareRequest;
+}
+
+/* Разрешение относится к кодеку текущей трансляции; устаревший ответ не запускает новый режим. */
+void StreamPublishEngine::answerSoftwareEncoding(uint64_t generation, bool allow) {
+    { std::lock_guard<std::mutex> 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<std::mutex> 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<int64_t>(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<std::mutex> 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<std::mutex> 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<LiveVideoEncoder>();
-                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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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);
 }
--- 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<Impl> m;
--- a/tools/chatgui/src/stream_publish_window.cpp
+++ b/tools/chatgui/src/stream_publish_window.cpp
@ -8,6 +8,7 @@
 #include <QComboBox>
 #include <QCheckBox>
 #include <QLabel>
+#include <QMessageBox>
 #include <QPushButton>
 #include <QButtonGroup>
 #include <QPainter>
@ -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")