Move captured frame conversion to persistent worker

tools/chatgui/src/stream_publish_engine.cpp

@@ -63,13 +63,19 @@ struct StreamPublishEngine::Impl {
     std::mutex lock;
     std::condition_variable wake;
     std::thread videoThread, audioThread, joinThread;
-    std::atomic<bool> workersJoined{false}, videoExited{false}, audioExited{false};
+    std::thread imageThread, imageJoinThread;
+    std::atomic<bool> workersJoined{false}, videoExited{false}, audioExited{false}, imageJoined{false};
     bool stopping = false, finSent = false, destroying = false;
     int64_t stopStarted = 0, stopReport = 0;
     std::atomic<bool> running{false}, muted{false};
     std::atomic<unsigned> audioDropped{0}, audioMissing{0};
     std::atomic<int64_t> lastAudioAt{0};
     AudioEventQueue<Audio, 32> audio;
+    std::condition_variable imageWake;
+    QVideoFrame imageInput;
+    uint64_t imageGeneration = 0;
+    int64_t imagePts = -1;
+    bool imagePending = false, imageStop = false;
     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;
     int64_t audioAnchor = -1;
@@ -142,17 +148,29 @@ StreamPublishEngine::StreamPublishEngine(QObject* parent) : QObject(parent), m(s
         if (m->running && m->wanted.audio) publicationFailed(m->id, text("The audio device was disconnected. Select a microphone and start again."));
         m_capture.close(); // Контекст нельзя освободить до завершения фонового закрытия устройства.
     });
+#ifdef HAVE_STREAM_CAPTURE
+    m->imageThread = std::thread(&StreamPublishEngine::imageLoop, this);
+#else
+    m->imageJoined = true;
+#endif
 }
 StreamPublishEngine::~StreamPublishEngine() {
     m->destroying = true; blockSignals(true);
     stop(); m_timer.stop(); m_previewTimer.stop();
-    if (!finishStop()) {
+    closeSource();
+#ifdef HAVE_STREAM_CAPTURE
+    { std::lock_guard<std::mutex> guard(m->lock); m->imageStop = true; m->imagePending = false; m->imageInput = {}; }
+    m->imageWake.notify_all();
+    m->imageJoinThread = std::thread([this] { m->imageThread.join(); m->imageJoined = true; });
+#endif
+    if (!finishStop() || !m->imageJoined) {
         // Qt освобождает thread-local графику через GUI. Даже при выходе приложения нельзя блокировать его join().
         DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop publisher destruction waits for worker resource cleanup with GUI events enabled");
         QEventLoop loop; QTimer timer;
-        connect(&timer, &QTimer::timeout, &loop, [this, &loop] { if (finishStop()) loop.quit(); });
+        connect(&timer, &QTimer::timeout, &loop, [this, &loop] { if (finishStop() && m->imageJoined) loop.quit(); });
         timer.start(10); loop.exec(QEventLoop::ExcludeUserInputEvents);
     }
+    if (m->imageJoinThread.joinable()) m->imageJoinThread.join();
 }
 bool StreamPublishEngine::supported() {
 #ifdef HAVE_STREAM_CAPTURE
@@ -267,33 +285,12 @@ bool StreamPublishEngine::openSource(const StreamPublishConfig& config) {
             if (m->framePts < 0 && m->wanted.source.id == "portal") m->started = monotonicUs();
             framePts = std::max<int64_t>(1, callbackAt - m->origin);
         }
-        /* QVideoFrame conversion stays on Qt's persistent multimedia callback thread.
-         * Qt worker-thread storage was preventing the short-lived publisher thread from exiting on Windows. */
-        int64_t convertAt = monotonicUs();
-        QImage image = frame.toImage();
-        if (image.isNull()) {
-            DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop captured frame conversion failed generation=%llu format=%d size=%dx%d",
-                (unsigned long long)generation, frame.pixelFormat(), frame.width(), frame.height());
-            QMetaObject::invokeMethod(this, [this, generation] { sourceFailed(generation, text("Video frame conversion failed")); });
-            return;
-        }
-#if QT_VERSION >= QT_VERSION_CHECK(6, 7, 0)
-        if (frame.rotation() != QtVideo::Rotation::None) image = image.transformed(QTransform().rotate(static_cast<int>(frame.rotation())));
-        if (frame.mirrored()) image = image.transformed(QTransform().scale(-1, 1));
-#else
-        if (frame.rotationAngle()) image = image.transformed(QTransform().rotate(frame.rotationAngle()));
-#endif
         {
             std::lock_guard<std::mutex> guard(m->lock);
             if (!m->running || generation != m->sourceGeneration) return;
-            m->latestImage = image; m->frame = std::move(image); m->framePts = framePts; m->frameGeneration = m->generation;
-            if (!m->convertedCaptureFrames++) DEBUG_INFO(DEBUG_CATEGORY_VIDEO,
-                "desktop first image conversion complete generation=%llu size=%dx%d callback_tid=%llu gui_thread=%d convert_us=%lld",
-                (unsigned long long)generation, m->latestImage.width(), m->latestImage.height(),
-                (unsigned long long)(quintptr)QThread::currentThreadId(), QThread::currentThread() == qApp->thread(),
-                (long long)(monotonicUs() - convertAt));
+            m->imageInput = frame; m->imageGeneration = generation; m->imagePts = framePts; m->imagePending = true;
         }
-        m->wake.notify_all();
+        m->imageWake.notify_one();
     }, Qt::DirectConnection);
     auto failed = [this, generation](const QString& error) {
         if (generation == m->sourceGeneration) sourceFailed(m->generation, error);
@@ -410,6 +407,11 @@ bool StreamPublishEngine::openSource(const StreamPublishConfig& config) {
 }
 void StreamPublishEngine::closeSource() {
 #ifdef HAVE_STREAM_CAPTURE
+    {
+        std::lock_guard<std::mutex> guard(m->lock);
+        ++m->sourceGeneration; m->imagePending = false; m->imageInput = {};
+        m->frame = {}; m->latestImage = {}; m->previewFrame = {}; m->presentedPreview = {}; m->framePts = -1;
+    }
     m->screenGrabTimer.reset();
 #ifdef UTUN_STREAM_PORTAL
     m->portal.reset();
@@ -423,8 +425,48 @@ void StreamPublishEngine::closeSource() {
     if (m->screen) { disconnect(m->screen.get(), nullptr, this, nullptr); m->screen->stop(); }
     if (m->window) { disconnect(m->window.get(), nullptr, this, nullptr); m->window->stop(); }
     m->session.reset(); m->camera.reset(); m->screen.reset(); m->window.reset(); m->sink.reset();
-    std::lock_guard<std::mutex> guard(m->lock);
-    m->frame = {}; m->latestImage = {}; m->previewFrame = {}; m->presentedPreview = {}; m->framePts = -1;
+#endif
+}
+
+/* Долгоживущий worker выполняет Qt Multimedia преобразование; video worker его TLS не создаёт. */
+void StreamPublishEngine::imageLoop() {
+#ifdef HAVE_STREAM_CAPTURE
+    for (;;) {
+        QVideoFrame frame; uint64_t generation; int64_t pts;
+        {
+            std::unique_lock<std::mutex> lock(m->lock);
+            m->imageWake.wait(lock, [this] { return m->imageStop || m->imagePending; });
+            if (m->imageStop) return;
+            frame = std::move(m->imageInput); generation = m->imageGeneration; pts = m->imagePts; m->imagePending = false;
+        }
+        int64_t convertAt = monotonicUs();
+        QImage image = frame.toImage();
+        if (image.isNull()) {
+            DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop captured frame conversion failed generation=%llu format=%d size=%dx%d",
+                (unsigned long long)generation, frame.pixelFormat(), frame.width(), frame.height());
+            QMetaObject::invokeMethod(this, [this, generation] { sourceFailed(generation, text("Video frame conversion failed")); });
+            continue;
+        }
+#if QT_VERSION >= QT_VERSION_CHECK(6, 7, 0)
+        if (frame.rotation() != QtVideo::Rotation::None) image = image.transformed(QTransform().rotate(static_cast<int>(frame.rotation())));
+        if (frame.mirrored()) image = image.transformed(QTransform().scale(-1, 1));
+#else
+        if (frame.rotationAngle()) image = image.transformed(QTransform().rotate(frame.rotationAngle()));
+#endif
+        {
+            std::lock_guard<std::mutex> guard(m->lock);
+            if (m->imageStop) return;
+            if (!m->running || generation != m->sourceGeneration) continue;
+            if (m->framePts < 0 && m->wanted.source.id == "portal") m->started = monotonicUs();
+            m->latestImage = image; m->frame = std::move(image); m->framePts = pts; m->frameGeneration = m->generation;
+            if (!m->convertedCaptureFrames++) DEBUG_INFO(DEBUG_CATEGORY_VIDEO,
+                "desktop first image conversion complete generation=%llu size=%dx%d callback_tid=%llu gui_thread=%d convert_us=%lld",
+                (unsigned long long)generation, m->latestImage.width(), m->latestImage.height(),
+                (unsigned long long)(quintptr)QThread::currentThreadId(), QThread::currentThread() == qApp->thread(),
+                (long long)(monotonicUs() - convertAt));
+        }
+        m->wake.notify_all();
+    }
 #endif
 }
 

tools/chatgui/src/stream_publish_engine.h

@@ -82,6 +82,7 @@ private:
     void publicationFailed(uint64_t id, const QString& error);
     void tick();
     bool finishStop();
+    void imageLoop();
     void videoLoop(StreamPublishConfig initial);
     void audioLoop();
     void captureAudio(const AudioCapture::Frame& frame);