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

Trace stream video progress and detect missing microphone PCM

master
evgeny 2 days ago
parent
349869f732
commit
05f1fe485c
3 changed files with 98 additions and 6 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. 45
      tools/chatgui/src/audio_device.cpp
  2. 7
      tools/chatgui/src/audio_device.h
  3. 52
      tools/chatgui/src/stream_publish_engine.cpp

45
tools/chatgui/src/audio_device.cpp
Unescape View File

@ -4,11 +4,19 @@
#include <algorithm>
#include <utility>
#include <chrono>
#ifdef _WIN32
#include "platform_compat.h"
#include <objbase.h>
#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::microseconds>(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);

7
tools/chatgui/src/audio_device.h
Unescape View File

@ -4,6 +4,7 @@
#include <QElapsedTimer>
#include <atomic>
#include <functional>
#include <thread>
/* Один владелец в 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<bool> m_timingMeasured{false};
std::atomic<uint64_t> m_route{0};
std::atomic<bool> m_captureMissing{false};
std::atomic<int64_t> m_captureProgress{0};
std::thread m_closeThread;
std::atomic<bool> m_closeDone{true};
uint64_t m_reportedRoute = 0;
int m_reportedTiming = -1;
std::atomic<bool> m_suspended{false};

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

@ -72,6 +72,12 @@ struct StreamPublishEngine::Impl {
int channels = 2;
uint64_t capturedFrames = 0;
int64_t captureReport = 0;
std::atomic<const char*> videoStage{"stopped"};
std::atomic<int64_t> videoStageAt{0};
std::atomic<uint64_t> 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<std::mutex> 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<std::mutex> 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<LiveVideoEncoder>();
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<std::mutex> 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<std::mutex> 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<LiveEncodedFrame> 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<std::mutex> 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<std::mutex> 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);

Write Preview
Loading…
Cancel
Save