From 05f1fe485c65b14141b402684f8c93d187ed9bde Mon Sep 17 00:00:00 2001 From: evgeny Date: Fri, 16 Oct 2026 16:47:25 +0200 Subject: [PATCH] Trace stream video progress and detect missing microphone PCM --- tools/chatgui/src/audio_device.cpp | 45 +++++++++++++++++- tools/chatgui/src/audio_device.h | 7 +++ tools/chatgui/src/stream_publish_engine.cpp | 52 +++++++++++++++++++-- 3 files changed, 98 insertions(+), 6 deletions(-) diff --git a/tools/chatgui/src/audio_device.cpp b/tools/chatgui/src/audio_device.cpp index 9e61b1e2..88aaf84c 100644 --- a/tools/chatgui/src/audio_device.cpp +++ b/tools/chatgui/src/audio_device.cpp @@ -4,11 +4,19 @@ #include #include #include +#ifdef _WIN32 +#include "platform_compat.h" +#include +#endif extern "C" int64_t audio_stream_frame_time(ma_device* device, unsigned int frames, int* valid); extern "C" uint64_t audio_stream_route(ma_device* device, int capture); extern "C" int audio_stream_capture_missing(ma_device* device); extern "C" void audio_stream_log_route(ma_device* device, int capture); +static int64_t captureNowUs() { + return std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()).count(); +} + AudioDevice::AudioDevice() { m_clock.start(); } AudioDevice::~AudioDevice() { close(); } @@ -25,6 +33,7 @@ void AudioDevice::requestLowLatency(ma_context* context, ma_device_config& confi } bool AudioDevice::init(ma_context* context, ma_device_config config, const char* role, uint64_t session, Callback callback) { + if (closing()) { DEBUG_WARN(DEBUG_CATEGORY_CALL, "audio: %s previous device is still closing", role); return false; } close(); m_role = role; m_callback = std::move(callback); @@ -59,6 +68,7 @@ bool AudioDevice::init(ma_context* context, ma_device_config config, const char* bool AudioDevice::start() { if (!m_device) return false; m_progressAt = m_clock.elapsed(); + m_captureProgress = captureNowUs(); ma_result result = ma_device_start(m_device); if (result != MA_SUCCESS) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "audio: %s start failed result=%d", m_role, result); @@ -71,6 +81,7 @@ bool AudioDevice::start() { } void AudioDevice::close() { + if (m_closeThread.joinable()) m_closeThread.join(); if (m_device) { ma_device_uninit(m_device); /* uninit включает stop/join, в том числе после неудачного start. */ delete m_device; @@ -81,6 +92,37 @@ void AudioDevice::close() { m_route = 0; } +/* Устройство отсоединено от GUI, но callback хранится до завершения аппаратного потока. */ +void AudioDevice::closeAsync() { + if (closing()) return; + if (m_closeThread.joinable()) m_closeThread.join(); + auto* device = std::exchange(m_device, nullptr); + if (!device) return; + m_closeDone.store(false, std::memory_order_release); + DEBUG_INFO(DEBUG_CATEGORY_CALL, "audio: %s asynchronous close started", m_role); + try { + m_closeThread = std::thread([this, device] { +#ifdef _WIN32 + HRESULT com = CoInitializeEx(nullptr, COINIT_MULTITHREADED); + if (FAILED(com)) DEBUG_ERROR(DEBUG_CATEGORY_CALL, "audio: close worker COM init failed HRESULT=0x%08lx", (unsigned long)com); +#endif + ma_device_uninit(device); delete device; m_callback = {}; m_route = 0; +#ifdef _WIN32 + if (SUCCEEDED(com)) CoUninitialize(); +#endif + DEBUG_INFO(DEBUG_CATEGORY_CALL, "audio: %s asynchronous close completed", m_role); + m_closeDone.store(true, std::memory_order_release); + }); + } catch (const std::system_error& error) { + DEBUG_ERROR(DEBUG_CATEGORY_CALL, "audio: %s close worker failed: %s", m_role, error.what()); + m_device = device; close(); m_closeDone.store(true, std::memory_order_release); + } +} + +bool AudioDevice::captureTimedOut() const { + return m_device && m_device->type == ma_device_type_capture && captureNowUs() - m_captureProgress.load() >= 1000000; +} + void AudioDevice::pause() { if (!m_device) return; ma_result result = ma_device_stop(m_device); @@ -102,6 +144,7 @@ void AudioDevice::data(ma_device* device, void* out, const void* in, ma_uint32 f uint64_t route = audio_stream_route(device, capture); if (route != self->m_route.exchange(route, std::memory_order_relaxed)) audio_stream_log_route(device, capture); self->m_captureMissing.store(audio_stream_capture_missing(device), std::memory_order_relaxed); + if (capture && in && frames && !self->m_captureMissing.load(std::memory_order_relaxed)) self->m_captureProgress = now; if (self->m_callback) self->m_callback(in, out, frames); self->m_frames.fetch_add(frames, std::memory_order_relaxed); } @@ -136,7 +179,7 @@ bool AudioDevice::needsRecovery() { return true; } -bool AudioDevice::retryReady() const { return !m_device && m_clock.elapsed() >= m_retryAt; } +bool AudioDevice::retryReady() const { return !m_device && !closing() && m_clock.elapsed() >= m_retryAt; } void AudioDevice::retryLater() { m_retryAt = m_clock.elapsed() + m_retryDelay; DEBUG_WARN(DEBUG_CATEGORY_CALL, "audio: %s retry in %dms", m_role, m_retryDelay); diff --git a/tools/chatgui/src/audio_device.h b/tools/chatgui/src/audio_device.h index 88e2aa30..e4354934 100644 --- a/tools/chatgui/src/audio_device.h +++ b/tools/chatgui/src/audio_device.h @@ -4,6 +4,7 @@ #include #include #include +#include /* Один владелец в GUI-потоке. Callback публикует только прогресс и уведомления; * close() дожидается worker перед освобождением callback и его состояния. */ @@ -19,6 +20,9 @@ public: bool start(); void pause(); // GUI: завершает текущий callback, сохраняя backend stream и его физический route. void close(); + void closeAsync(); // Callback и владелец живут до завершения uninit; close() дожидается worker. + bool closing() const { return !m_closeDone.load(std::memory_order_acquire); } + bool captureTimedOut() const; // Секунда без корректного PCM; нулевые сэмплы считаются данными. bool needsRecovery(); bool retryReady() const; void retryLater(); @@ -41,6 +45,9 @@ private: std::atomic m_timingMeasured{false}; std::atomic m_route{0}; std::atomic m_captureMissing{false}; + std::atomic m_captureProgress{0}; + std::thread m_closeThread; + std::atomic m_closeDone{true}; uint64_t m_reportedRoute = 0; int m_reportedTiming = -1; std::atomic m_suspended{false}; diff --git a/tools/chatgui/src/stream_publish_engine.cpp b/tools/chatgui/src/stream_publish_engine.cpp index 725d9004..b965c592 100644 --- a/tools/chatgui/src/stream_publish_engine.cpp +++ b/tools/chatgui/src/stream_publish_engine.cpp @@ -72,6 +72,12 @@ struct StreamPublishEngine::Impl { int channels = 2; uint64_t capturedFrames = 0; int64_t captureReport = 0; + std::atomic videoStage{"stopped"}; + std::atomic videoStageAt{0}; + std::atomic convertedFrames{0}, encodedFrames{0}, deliveredFrames{0}, repeatedFrames{0}; + void videoProgress(const char* stage) { + videoStageAt.store(monotonicUs()); videoStage.store(stage); + } bool hasGood = false, rollingBack = false, videoReady = false; uint64_t softwareRequest = 0; int softwareCodec = 0; @@ -385,6 +391,8 @@ bool StreamPublishEngine::start(uint64_t group, const StreamPublishConfig& confi m->wanted = config; m->good = {}; m->hasGood = m->rollingBack = false; m->generation++; m->confirmedGeneration = 0; m->videoReady = false; m->softwareRequest = 0; m->softwareCodec = 0; m->lastAudioAt = 0; m->origin = m->started = monotonicUs(); m->stats = {}; } + m->convertedFrames = m->encodedFrames = m->deliveredFrames = m->repeatedFrames = 0; + m->videoProgress("starting"); m->captureReport = 0; m->id = QRandomGenerator::global()->generate64(); if (!m->id) m->id = 1; m->group = group; m->audio.reset(); m->audioSequence = 0; m->muted = false; stream_media_format format{(uint16_t)config.video.width, (uint16_t)config.video.height, (uint16_t)config.video.rotation, @@ -456,10 +464,23 @@ void StreamPublishEngine::publicationFailed(uint64_t id, const QString& error) { /* Сначала callbacks, затем workers; FIN не обгоняет уже принятые access units. */ void StreamPublishEngine::stop() { if (!m->running.exchange(false)) return; + int64_t started = monotonicUs(); + 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(), + (unsigned long long)m->deliveredFrames.load()); { std::lock_guard guard(m->lock); m->softwareRequest = 0; } - m->wake.notify_all(); closeSource(); m_capture.closeAsync(); + m->wake.notify_all(); + DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop stop closing video source"); + 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); emit stateChanged(text("Finishing stream…")); @@ -586,6 +607,7 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { QImage lastImage; uint64_t current = 0; uint32_t revision = 0; int64_t lastPts = -1, lastKey = -1000000, next = 0, report = 0; unsigned count = 0, skipped = 0; while (m->running) { + m->videoProgress("waiting-frame"); StreamPublishConfig config; uint64_t generation; QVideoFrame frame; int64_t capture; { std::unique_lock lock(m->lock); config = m->wanted; generation = m->generation; if (config.audio && !m->lastAudioAt.load(std::memory_order_acquire)) { @@ -632,6 +654,7 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { }); } else { auto candidate = std::make_unique(); + m->videoProgress("open-hardware-encoder"); bool opened = candidate->open(config.video, LiveVideoEncoder::Backend::HardwareOnly); if (!opened && LiveVideoEncoder::available(LiveVideoEncoder::Backend::SoftwareOnly).contains(config.video.codec)) { QString details = candidate->error(); @@ -639,6 +662,7 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { if (!m->running || generation != m->generation) continue; if (m->softwareCodec != config.video.codec) { m->softwareRequest = generation; + m->videoProgress("waiting-cpu-consent"); DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "desktop CPU confirmation required generation=%llu codec=%d reason=%s", (unsigned long long)generation, config.video.codec, qPrintable(details)); QMetaObject::invokeMethod(this, [this, generation, codec = config.video.codec, details] { @@ -652,6 +676,7 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { } if (!m->running || generation != m->generation) continue; lock.unlock(); + m->videoProgress("open-software-encoder"); opened = candidate->open(config.video, LiveVideoEncoder::Backend::SoftwareOnly); } if (!opened) { @@ -666,6 +691,7 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { } // Начальный формат publisher уже зарегистрирован; уточнение размера требует новой revision даже до первого KEY. revision = revision ? revision + 1 : config.video == initial.video ? 1 : 2; + m->videoProgress("replace-encoder"); encoder = std::move(candidate); current = generation; encoding = config; lastImage = {}; lastKey = -1000000; { std::lock_guard guard(m->lock); if (!m->running || generation != m->generation) continue; @@ -678,7 +704,9 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { bool screen = config.source.kind != StreamVideoSource::Camera; if (frame.isValid()) { if (!screen && (capture <= lastPts || monotonicUs() - m->origin - capture > 250000)) { skipped++; continue; } + m->videoProgress("convert-frame-to-image"); lastImage = frame.toImage(); + m->convertedFrames++; if (lastImage.isNull()) { DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop captured frame conversion failed format=%d", frame.pixelFormat()); QMetaObject::invokeMethod(this, [this, generation] { sourceFailed(generation, text("Video frame conversion failed")); }); @@ -692,20 +720,25 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { #endif } else if (!screen || lastImage.isNull()) { std::unique_lock lock(m->lock); m->wake.wait_for(lock, std::chrono::milliseconds(5)); continue; - } + } else m->repeatedFrames++; + m->videoProgress("publisher-status"); int64_t now = monotonicUs(), pts = screen ? now - m->origin : capture; next = now + 1000000 / config.video.fps; stream_publisher_status status; bool needsKey = gui_stream_publish_status(m->id, status) && status.need_key; bool key = pts - lastKey >= 1000000 || needsKey; QList output; + m->videoProgress("encode-frame"); if (!encoder->encode(lastImage, std::max(lastPts + 1, pts), key, output)) { QString error = encoder->error(); QMetaObject::invokeMethod(this, [this, generation, error] { sourceFailed(generation, error); }); continue; } + m->encodedFrames += output.size(); + m->videoProgress("publish-frame"); stream_media_format format{(uint16_t)config.video.width, (uint16_t)config.video.height, (uint16_t)config.video.rotation, (uint8_t)config.video.codec, (uint8_t)(config.audio ? STREAM_MEDIA_OPUS_STEREO : 0), revision}; for (const auto& encoded : output) { if (gui_stream_publish_feed(m->id, true, encoded.pts, encoded.key, encoded.data, &format) == 1) { lastPts = encoded.pts; if (encoded.key) lastKey = encoded.pts; count++; + m->deliveredFrames++; if (encoded.key) { bool first; { std::lock_guard guard(m->lock); first = !m->hasGood || m->good.video != config.video || m->rollingBack; @@ -729,7 +762,12 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) { (unsigned long long)generation, count, skipped, (long long)(screen ? 0 : now - m->origin - capture), qPrintable(encoder->name())); report = now; count = skipped = 0; } + m->videoProgress("release-frame"); } + m->videoProgress("close-encoder"); + encoder.reset(); + m->videoProgress("stopped"); + DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop video worker stopped"); #else Q_UNUSED(initial); #endif @@ -767,11 +805,15 @@ void StreamPublishEngine::tick() { if (!confirmed && !waitingSelection && !softwareEncodingRequest() && monotonicUs() - started > 8000000) sourceFailed(m->generation, text("Timed out waiting for the first video frame")); { std::lock_guard guard(m->lock); - if (monotonicUs() - m->captureReport >= 1000000) { + if (monotonicUs() - m->captureReport >= 5000000) { m->captureReport = monotonicUs(); - DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop capture state source_generation=%llu video_generation=%llu confirmed=%llu raw_frames=%llu raw_PTS=%lld", + DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop capture progress source_generation=%llu video_generation=%llu confirmed=%llu raw_frames=%llu raw_PTS=%lld " + "converted=%llu encoded=%llu delivered=%llu repeated=%llu stage=%s stage_age_ms=%lld", (unsigned long long)m->sourceGeneration, (unsigned long long)m->generation, (unsigned long long)m->confirmedGeneration, - (unsigned long long)m->capturedFrames, (long long)m->framePts); + (unsigned long long)m->capturedFrames, (long long)m->framePts, + (unsigned long long)m->convertedFrames.load(), (unsigned long long)m->encodedFrames.load(), + (unsigned long long)m->deliveredFrames.load(), (unsigned long long)m->repeatedFrames.load(), + m->videoStage.load(), (long long)((monotonicUs() - m->videoStageAt.load()) / 1000)); } } } unsigned dropped = m->audioDropped.exchange(0);