diff --git a/tools/chatgui/src/stream_publish_engine.cpp b/tools/chatgui/src/stream_publish_engine.cpp index b965c592..b903bb43 100644 --- a/tools/chatgui/src/stream_publish_engine.cpp +++ b/tools/chatgui/src/stream_publish_engine.cpp @@ -10,7 +10,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -61,6 +63,9 @@ struct StreamPublishEngine::Impl { std::mutex lock; std::condition_variable wake; std::thread videoThread, audioThread; + std::future videoExited, audioExited; + bool stopping = false, destroying = false; + int64_t stopStarted = 0, stopReport = 0; std::atomic running{false}, muted{false}; std::atomic audioDropped{0}, audioMissing{0}; std::atomic lastAudioAt{0}; @@ -136,7 +141,15 @@ StreamPublishEngine::StreamPublishEngine(QObject* parent) : QObject(parent), m(s }); } StreamPublishEngine::~StreamPublishEngine() { + m->destroying = true; blockSignals(true); stop(); m_timer.stop(); m_previewTimer.stop(); + if (!finishStop()) { + // 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(); }); + timer.start(10); loop.exec(QEventLoop::ExcludeUserInputEvents); + } } bool StreamPublishEngine::supported() { #ifdef HAVE_STREAM_CAPTURE @@ -378,6 +391,7 @@ void StreamPublishEngine::closeSource() { } bool StreamPublishEngine::start(uint64_t group, const StreamPublishConfig& config) { + if (m->destroying) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop publication start during destruction"); return false; } if (active()) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop publication already active"); emit errorOccurred(text("A stream is already running")); return false; } if (!supported() || !codecs().contains(config.video.codec)) { DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "desktop capture/encoder unavailable codec=%d", config.video.codec); @@ -402,8 +416,19 @@ bool StreamPublishEngine::start(uint64_t group, const StreamPublishConfig& confi } m->running = true; try { - m->videoThread = std::thread(&StreamPublishEngine::videoLoop, this, config); - if (config.audio) m->audioThread = std::thread(&StreamPublishEngine::audioLoop, this); + std::promise videoDone; m->videoExited = videoDone.get_future(); + m->videoThread = std::thread([this, config, done = std::move(videoDone)]() mutable { + 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(); + }); + } } 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; @@ -461,10 +486,11 @@ void StreamPublishEngine::publicationFailed(uint64_t id, const QString& error) { emit errorOccurred(error); stop(); } -/* Сначала callbacks, затем workers; FIN не обгоняет уже принятые access units. */ +/* Сначала callbacks; ожидание workers оставляет GUI доступным для освобождения Qt thread-local ресурсов. */ void StreamPublishEngine::stop() { if (!m->running.exchange(false)) return; int64_t started = monotonicUs(); + m->stopping = true; m->stopStarted = m->stopReport = started; DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop stop begin id=%llu video_stage=%s stage_age_ms=%lld converted=%llu encoded=%llu delivered=%llu", (unsigned long long)m->id, m->videoStage.load(), (long long)((started - m->videoStageAt.load()) / 1000), (unsigned long long)m->convertedFrames.load(), (unsigned long long)m->encodedFrames.load(), @@ -475,17 +501,39 @@ void StreamPublishEngine::stop() { closeSource(); DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop stop video source closed elapsed_ms=%lld", (long long)((monotonicUs() - started) / 1000)); m_capture.closeAsync(); - DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop stop joining video worker stage=%s stage_age_ms=%lld", - m->videoStage.load(), (long long)((monotonicUs() - m->videoStageAt.load()) / 1000)); - if (m->videoThread.joinable()) m->videoThread.join(); - DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop stop video joined; joining audio elapsed_ms=%lld", (long long)((monotonicUs() - started) / 1000)); - if (m->audioThread.joinable()) m->audioThread.join(); - DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop stop workers joined; sending FIN elapsed_ms=%lld", (long long)((monotonicUs() - started) / 1000)); - gui_stream_publish_stop(m->id); - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop capture stopped id=%llu; draining publisher", (unsigned long long)m->id); + DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop stop waiting asynchronously for workers stage=%s", m->videoStage.load()); emit stateChanged(text("Finishing stream…")); } +/* Future готов только после TLS destructors. FIN не обгоняет принятые access units; owner жив до обоих join. */ +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)); + } + 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_capture.closing()) return false; + m_microphone.release(); m->stopping = false; + gui_stream_publish_stop(m->id); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop capture stopped id=%llu elapsed_ms=%lld; draining publisher", + (unsigned long long)m->id, (long long)((monotonicUs() - m->stopStarted) / 1000)); + return true; +} + uint64_t StreamPublishEngine::softwareEncodingRequest() const { std::lock_guard guard(m->lock); return m->softwareRequest; } @@ -650,7 +698,7 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { { std::lock_guard guard(m->lock); if (m->generation == generation) { m->good = config; m->hasGood = true; m->confirmedGeneration = generation; } } QMetaObject::invokeMethod(this, [this, generation] { - if (generation == m->generation) { emit configurationChanged(); emit stateChanged(text("Live")); } + if (m->running && generation == m->generation) { emit configurationChanged(); emit stateChanged(text("Live")); } }); } else { auto candidate = std::make_unique(); @@ -750,7 +798,7 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop first KEY committed generation=%llu revision=%u PTS=%llu bytes=%lld", (unsigned long long)generation, revision, (unsigned long long)encoded.pts, (long long)encoded.data.size()); QMetaObject::invokeMethod(this, [this, generation] { - if (generation != m->generation) return; + if (!m->running || generation != m->generation) return; emit configurationChanged(); emit stateChanged(text("Live")); }); } @@ -775,19 +823,19 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { void StreamPublishEngine::tick() { if (!active()) return; - if (!m->running && m_capture.closing()) return; + if (!finishStop()) return; if (!m->running) m_microphone.release(); stream_publisher_status status; if (!gui_stream_publish_status(m->id, status)) { stop(); - if (m_capture.closing()) return; + if (!finishStop()) 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(); - if (m_capture.closing()) return; + if (!finishStop()) 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; diff --git a/tools/chatgui/src/stream_publish_engine.h b/tools/chatgui/src/stream_publish_engine.h index 9b56a4e7..bf445015 100644 --- a/tools/chatgui/src/stream_publish_engine.h +++ b/tools/chatgui/src/stream_publish_engine.h @@ -1,5 +1,5 @@ /* Desktop publisher: GUI управляет источниками и AudioDevice, workers кодируют видео/Opus. - * Окно не владеет трансляцией. Stop закрывает callbacks, joins workers и отправляет FIN. + * Окно не владеет трансляцией. Stop закрывает callbacks; GUI ждёт workers асинхронно, затем отправляет FIN. * Captured video/preview хранят только последний кадр; публикация использует общий C actor. */ #pragma once #include @@ -81,6 +81,7 @@ private: void sourceFailed(uint64_t generation, const QString& error); void publicationFailed(uint64_t id, const QString& error); void tick(); + bool finishStop(); void videoLoop(StreamPublishConfig initial); void audioLoop(); void captureAudio(const AudioCapture::Frame& frame);