jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Keep GUI responsive until publisher threads finish resource cleanup

Browse Source
This commit is contained in:
evgeny committed 2026-10-16 17:26:30 +02:00
1 parent fb2a615165
commit 85101765eb
2 files changed
+66 -17
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tools/chatgui/src/stream_publish_engine.cpp
+64 -16
View File
@@ -10,7 +10,9 @@
#include <QPixmap>
#include <QJsonArray>
#include <QJsonDocument>
#include <QEventLoop>
#include <thread>
#include <future>
#include <mutex>
#include <condition_variable>
#include <chrono>
@@ -61,6 +63,9 @@ struct StreamPublishEngine::Impl {
std::mutex lock;
std::condition_variable wake;
std::thread videoThread, audioThread;
std::future<void> videoExited, audioExited;
bool stopping = 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};
@@ -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<void> 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<void> 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<std::mutex> guard(m->lock); return m->softwareRequest;
}
@@ -650,7 +698,7 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) {
{ std::lock_guard<std::mutex> 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<LiveVideoEncoder>();
@@ -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<std::mutex> 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;
tools/chatgui/src/stream_publish_engine.h
+2 -1
View File
@@ -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 <QObject>
@@ -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);