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

Wait for native thread exit outside GUI and cover FIN drain in tests

stream-multiview
evgeny 1 week ago
parent
0f5a5580c5
commit
688d559787
2 changed files with 39 additions and 37 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 48
      tools/chatgui/src/stream_publish_engine.cpp
  2. 28
      tools/chatgui/tests/test_stream_publish.cpp

48
tools/chatgui/src/stream_publish_engine.cpp
Unescape View File

@ -12,7 +12,6 @@
#include <QJsonDocument>
#include <QEventLoop>
#include <thread>
#include <future>
#include <mutex>
#include <condition_variable>
#include <chrono>
@ -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<void> videoExited, audioExited;
std::thread videoThread, audioThread, joinThread;
std::atomic<bool> workersJoined{false}, videoExited{false}, audioExited{false};
bool stopping = false, destroying = false;
int64_t stopStarted = 0, stopReport = 0;
std::atomic<bool> 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<void> 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<std::mutex> 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<void> 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;

28
tools/chatgui/tests/test_stream_publish.cpp
Unescape View File

@ -17,6 +17,9 @@
#include <atomic>
#include <thread>
#include <cmath>
#ifdef TEST_NULL_AUDIO
#include <pthread.h>
#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<unsigned> softwareOpens{0}, hardwareOpens{0};
static std::atomic<bool> failHardware{false};
static std::atomic<bool> requireGuiAtThreadExit{false};
static std::atomic<unsigned> 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<StreamPublishEngine>();
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

Write Preview
Loading…
Cancel
Save