From 688d55978728d88482b818955f7daa4d8542e157 Mon Sep 17 00:00:00 2001 From: evgeny Date: Fri, 16 Oct 2026 17:31:20 +0200 Subject: [PATCH] Wait for native thread exit outside GUI and cover FIN drain in tests --- tools/chatgui/src/stream_publish_engine.cpp | 48 ++++++++++----------- tools/chatgui/tests/test_stream_publish.cpp | 28 +++++++----- 2 files changed, 39 insertions(+), 37 deletions(-) diff --git a/tools/chatgui/src/stream_publish_engine.cpp b/tools/chatgui/src/stream_publish_engine.cpp index b903bb43..deb825b2 100644 --- a/tools/chatgui/src/stream_publish_engine.cpp +++ b/tools/chatgui/src/stream_publish_engine.cpp @@ -12,7 +12,6 @@ #include #include #include -#include #include #include #include @@ -62,8 +61,8 @@ 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; - std::future videoExited, audioExited; + std::thread videoThread, audioThread, joinThread; + std::atomic workersJoined{false}, videoExited{false}, audioExited{false}; bool stopping = false, destroying = false; int64_t stopStarted = 0, stopReport = 0; std::atomic running{false}, muted{false}; @@ -415,20 +414,25 @@ bool StreamPublishEngine::start(uint64_t group, const StreamPublishConfig& confi m->id = m->group = 0; m_microphone.release(); emit errorOccurred(text("Could not start the stream")); return false; } m->running = true; + m->workersJoined = m->videoExited = m->audioExited = false; try { - std::promise videoDone; m->videoExited = videoDone.get_future(); - m->videoThread = std::thread([this, config, done = std::move(videoDone)]() mutable { + // Отдельный join ждёт и native TLS Qt; отметки выхода из функции/thread_local для этого недостаточно. + m->joinThread = std::thread([this] { + { std::unique_lock lock(m->lock); m->wake.wait(lock, [this] { return !m->running; }); } + if (m->videoThread.joinable()) m->videoThread.join(); + m->videoExited = true; + DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop stop video thread exited"); + if (m->audioThread.joinable()) m->audioThread.join(); + m->audioExited = true; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stop audio thread exited"); + m->workersJoined = true; + }); + m->videoThread = std::thread([this, config] { videoLoop(config); m->videoProgress("release-thread-resources"); DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop video loop returned; releasing thread resources"); - done.set_value_at_thread_exit(); }); - if (config.audio) { - std::promise audioDone; m->audioExited = audioDone.get_future(); - m->audioThread = std::thread([this, done = std::move(audioDone)]() mutable { - audioLoop(); done.set_value_at_thread_exit(); - }); - } + 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)) { sourceFailed(m->generation, text("Could not open the selected capture device")); return false; @@ -505,26 +509,18 @@ void StreamPublishEngine::stop() { emit stateChanged(text("Finishing stream…")); } -/* Future готов только после TLS destructors. FIN не обгоняет принятые access units; owner жив до обоих join. */ +/* FIN не обгоняет принятые access units; owner жив до обоих join, включая native TLS destructors. */ bool StreamPublishEngine::finishStop() { if (!m->stopping) return true; if (monotonicUs() - m->stopReport >= 5000000) { m->stopReport = monotonicUs(); DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "desktop stop still waiting elapsed_ms=%lld video_stage=%s video=%d audio=%d microphone=%d", - (long long)((m->stopReport - m->stopStarted) / 1000), m->videoStage.load(), m->videoThread.joinable(), - m->audioThread.joinable(), m_capture.closing()); - } - if (m->videoThread.joinable()) { - if (m->videoExited.wait_for(std::chrono::seconds(0)) != std::future_status::ready) return false; - m->videoThread.join(); - DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop stop video thread exited elapsed_ms=%lld", - (long long)((monotonicUs() - m->stopStarted) / 1000)); + (long long)((m->stopReport - m->stopStarted) / 1000), m->videoStage.load(), !m->videoExited.load(), + !m->audioExited.load(), m_capture.closing()); } - if (m->audioThread.joinable()) { - if (m->audioExited.wait_for(std::chrono::seconds(0)) != std::future_status::ready) return false; - m->audioThread.join(); - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop stop audio thread exited elapsed_ms=%lld", - (long long)((monotonicUs() - m->stopStarted) / 1000)); + if (m->joinThread.joinable()) { + if (!m->workersJoined) return false; + m->joinThread.join(); } if (m_capture.closing()) return false; m_microphone.release(); m->stopping = false; diff --git a/tools/chatgui/tests/test_stream_publish.cpp b/tools/chatgui/tests/test_stream_publish.cpp index 5c16786e..ae09e415 100644 --- a/tools/chatgui/tests/test_stream_publish.cpp +++ b/tools/chatgui/tests/test_stream_publish.cpp @@ -17,6 +17,9 @@ #include #include #include +#ifdef TEST_NULL_AUDIO +#include +#endif #include "../src/live_video_encoder.h" #include "../src/stream_publish_engine.h" #include "../src/stream_publish_window.h" @@ -61,20 +64,18 @@ static std::atomic softwareOpens{0}, hardwareOpens{0}; static std::atomic failHardware{false}; static std::atomic requireGuiAtThreadExit{false}; static std::atomic threadExitArmed{0}, threadExitCompleted{0}; -struct GuiThreadExit { - bool armed = false; - ~GuiThreadExit() { - if (!armed) return; - std::this_thread::sleep_for(std::chrono::milliseconds(200)); - QMetaObject::invokeMethod(qApp, [] { threadExitCompleted++; }, Qt::BlockingQueuedConnection); - } -}; +static pthread_key_t guiThreadExitKey; +static void guiThreadExit(void*) { + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + QMetaObject::invokeMethod(qApp, [] { threadExitCompleted++; }, Qt::BlockingQueuedConnection); +} extern "C" int __real_avcodec_open2(AVCodecContext*, const AVCodec*, AVDictionary**); extern "C" int __wrap_avcodec_open2(AVCodecContext* context, const AVCodec* codec, AVDictionary** options) { if (av_codec_is_encoder(codec)) { if (requireGuiAtThreadExit && QThread::currentThread() != qApp->thread()) { - thread_local GuiThreadExit cleanup; - if (!cleanup.armed) { cleanup.armed = true; threadExitArmed++; } + if (!pthread_getspecific(guiThreadExitKey)) { + pthread_setspecific(guiThreadExitKey, &guiThreadExitKey); threadExitArmed++; + } } QByteArray name(codec->name); bool hardware = name.endsWith("_nvenc") || name.endsWith("_qsv") || name.endsWith("_amf") || name.endsWith("_vaapi"); @@ -538,14 +539,18 @@ private slots: if (!qEnvironmentVariableIsSet("UTUN_TEST_SCREEN_CAPTURE")) QSKIP("Run under isolated Xvfb"); QFETCH(bool, destroy); requireGuiAtThreadExit = true; failHardware = true; threadExitArmed = threadExitCompleted = 0; - auto reset = qScopeGuard([] { requireGuiAtThreadExit = false; failHardware = false; }); + QCOMPARE(pthread_key_create(&guiThreadExitKey, guiThreadExit), 0); + auto reset = qScopeGuard([] { requireGuiAtThreadExit = false; failHardware = false; pthread_key_delete(guiThreadExitKey); }); auto engine = std::make_unique(); + uint64_t id = 0; + connect(engine.get(), &StreamPublishEngine::telemetryChanged, this, [&](const stream_publisher_status& status) { id = status.id; }); connect(engine.get(), &StreamPublishEngine::softwareEncodingRequested, engine.get(), [&](uint64_t generation, const QString&, const QString&) { engine->answerSoftwareEncoding(generation, true); }); StreamPublishConfig config; config.audio = false; config.video.width = 320; config.video.height = 180; config.video.fps = 15; for (const auto& source : engine->sources()) if (source.kind == StreamVideoSource::Screen) { config.source = source; break; } QVERIFY(!config.source.id.isEmpty()); QVERIFY(engine->start(group, config)); QTRY_COMPARE_WITH_TIMEOUT(threadExitArmed.load(), 1u, 4000); + QTRY_VERIFY(id != 0); unsigned heartbeats = 0; QTimer heartbeat; connect(&heartbeat, &QTimer::timeout, this, [&] { heartbeats++; }); heartbeat.start(10); QElapsedTimer elapsed; elapsed.start(); @@ -559,6 +564,7 @@ private slots: QCOMPARE(threadExitCompleted.load(), 1u); QVERIFY(heartbeats >= 5); DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "[PASS] publisher stop destroy=%d GUI-dependent TLS cleanup completed heartbeat=%u elapsed_ms=%lld", destroy, heartbeats, (long long)elapsed.elapsed()); + QTRY_VERIFY_WITH_TIMEOUT(ended(id), 4000); // Core продолжает FIN drain после удаления desktop engine. #else QSKIP("Capture test requires FFmpeg and null audio wrappers"); #endif