Separate video decoding from event-driven presentation timing

This commit is contained in:
evgeny committed 2026-10-19 21:12:02 +02:00
1 parent 93f6fa1d52
commit d9843b8cbd
3 files changed
+187 -99

No files matched your search

+172 -98
View File
@@ -2,6 +2,7 @@
#include "call_video_encoder.h"
#include "call_video_decoder.h"
#include "call_video_wake.h"
#include "call_video_queue.h"
#include "call_camera_capture.h"
#include "../../../src/call/call_video_port.h"
#include "../../../src/call/call_video_playout.h"
@@ -14,6 +15,11 @@
#include <atomic>
#include <deque>
#include <chrono>
#ifdef _WIN32
#include <QWinEventNotifier>
#else
#include <QSocketNotifier>
#endif
namespace {
int64_t nowUs() { return call_video_playout_now_us(); }
@@ -22,8 +28,21 @@ struct CallVideoEngine::Impl {
std::mutex lock;
std::condition_variable wake;
CallVideoWake rxWake;
CallVideoWake displayWake, guiWake;
QObject* guiNotifier = nullptr;
struct Picture { call_video_playout_frame timing; QImage image; };
CallVideoQueue<Picture, 3> pictures;
CallVideoQueue<call_video_playout_event, 16> logEvents;
struct TimerReport {
int64_t interval = 0, workMax = 0, wakeLate = 0;
unsigned busy = 0, ready = 0, total = 0, presented = 0;
};
CallVideoQueue<TimerReport, 4> timerReports;
std::atomic<uint64_t> discarded{0};
std::atomic<unsigned> pipeline{0}, lostLogs{0};
call_video_playout playout{}; // Единственный владелец после start — display thread.
uint64_t wakeSerial = 0;
std::thread tx, rx;
std::thread tx, rx, display;
std::atomic<bool> running{false};
uint64_t id = 0, revision = 1;
bool enabled = false;
@@ -34,15 +53,15 @@ struct CallVideoEngine::Impl {
QByteArray cameraId;
bool captureActive = false;
uint64_t captured = 0;
uint64_t publicationEpoch = 0;
bool publicationPending = false;
std::atomic<uint64_t> publicationEpoch{0}, pendingSerial{0}, ackSerial{0};
std::atomic<int64_t> presentedAt{0};
bool presentationDriven = false;
uint64_t publicationSerial = 0;
int64_t publicationQueuedAt = 0;
int64_t paintWaitMax = 0; // Только под lock: время до подтверждения рисования.
int64_t presentedAt = 0; // GUI ACK -> RX; часы меняет только RX worker.
int64_t presentedPts = -1;
int64_t paintWaitMax = 0; // Только GUI-поток.
int64_t publicationPts = -1;
bool publicationVideo = false;
uint64_t deliveredSerial = 0; // Только GUI-поток; повторное событие не повторяет кадр.
int64_t uiReportAt = 0, uiLastAt = 0, uiWaitMax = 0, uiGapMax = 0; // Только GUI-поток.
int64_t uiTickAt = 0, uiTickReportAt = 0, uiTickGapMax = 0, uiImageSlotMax = 0;
unsigned uiTicks = 0;
@@ -60,6 +79,16 @@ struct CallVideoEngine::Impl {
#endif
};
CallVideoEngine::CallVideoEngine(QObject* parent) : QObject(parent), m(std::make_unique<Impl>()) {
if (m->guiWake.valid()) {
#ifdef _WIN32
auto* notifier = new QWinEventNotifier(m->guiWake.nativeHandle(), this);
connect(notifier, &QWinEventNotifier::activated, this, [this] { m->guiWake.consume(); deliverImage(); });
#else
auto* notifier = new QSocketNotifier(m->guiWake.nativeHandle(), QSocketNotifier::Read, this);
connect(notifier, &QSocketNotifier::activated, this, [this] { m->guiWake.consume(); deliverImage(); });
#endif
m->guiNotifier = notifier;
}
m_timer.setInterval(20);
m_timer.setTimerType(Qt::PreciseTimer);
connect(&m_timer, &QTimer::timeout, this, [this] {
@@ -105,31 +134,32 @@ CallVideoEngine::CallVideoEngine(QObject* parent) : QObject(parent), m(std::make
});
#endif
}
CallVideoEngine::~CallVideoEngine() { stop(); }
CallVideoEngine::~CallVideoEngine() { stop(); delete m->guiNotifier; }
void CallVideoEngine::setPresentationDriven(bool enabled) {
std::lock_guard<std::mutex> guard(m->lock); m->presentationDriven = enabled;
}
void CallVideoEngine::imagePresented(quint64 serial) {
{ std::lock_guard<std::mutex> guard(m->lock);
if (!m->publicationPending || serial != m->publicationSerial) return;
m->paintWaitMax = std::max(m->paintWaitMax, nowUs() - m->publicationQueuedAt);
m->presentedAt = nowUs(); m->presentedPts = m->publicationPts;
m->publicationPending = false; }
m->wake.notify_all();
m->rxWake.notify();
if (!serial || serial != m->pendingSerial.load(std::memory_order_acquire) || serial == m->ackSerial.load()) return;
const int64_t now = nowUs();
m->paintWaitMax = std::max(m->paintWaitMax, now - m->publicationQueuedAt);
m->presentedAt.store(now, std::memory_order_relaxed); m->ackSerial.store(serial, std::memory_order_release);
m->displayWake.notify();
}
// Только один кадр до подтверждения рисования: GUI не заменяет ещё не показанное изображение.
// Только display thread. Payload публикуется release-store, GUI получает его через native notifier.
void CallVideoEngine::publishImage(const QImage& image, int64_t pts) {
uint64_t epoch, serial; int64_t queued;
{ std::unique_lock<std::mutex> guard(m->lock);
m->wake.wait(guard, [this] { return !m->running || !m->publicationPending; });
if (!m->running) return;
m->publicationPending = true; queued = m->publicationQueuedAt = nowUs(); epoch = m->publicationEpoch;
m->publicationPts = pts;
serial = ++m->publicationSerial; }
QMetaObject::invokeMethod(this, [this, epoch, serial, image, pts, queued] {
{ std::lock_guard<std::mutex> guard(m->lock);
if (epoch != m->publicationEpoch) return;
m->publicationQueuedAt = nowUs(); m->publicationPts = pts; m->publicationVideo = !image.isNull();
m->pendingSerial.store(++m->publicationSerial, std::memory_order_release); m->guiWake.notify();
}
void CallVideoEngine::deliverImage() {
const uint64_t serial = m->pendingSerial.load(std::memory_order_acquire);
if (!m->running || !serial || serial == m->deliveredSerial) return;
m->deliveredSerial = serial;
const int64_t queued = m->publicationQueuedAt, pts = m->publicationPts;
QImage image;
if (m->publicationVideo) {
const auto* slot = m->pictures.readSlot();
if (!slot) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video GUI missing published slot serial=%llu", (unsigned long long)serial); return; }
image = slot->image;
}
const int64_t now = nowUs(), wait = now - queued;
++m->uiPublished; m->uiWaitMax = std::max(m->uiWaitMax, wait);
@@ -141,13 +171,10 @@ void CallVideoEngine::publishImage(const QImage& image, int64_t pts) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "video GUI image slot slow call=%llu serial=%llu work_us=%lld mono_us=%lld",
(unsigned long long)m->id, (unsigned long long)serial, (long long)slotUs, (long long)now);
m->uiImageSlotMax = std::max(m->uiImageSlotMax, slotUs);
bool driven;
{ std::lock_guard<std::mutex> guard(m->lock); driven = m->presentationDriven; }
if (!driven || image.isNull()) imagePresented(serial);
if (!m->presentationDriven || image.isNull()) imagePresented(serial);
if (!m->uiReportAt) m->uiReportAt = now;
if (now - m->uiReportAt >= 1000000) {
int64_t paintWait;
{ std::lock_guard<std::mutex> guard(m->lock); paintWait = m->paintWaitMax; m->paintWaitMax = 0; }
const int64_t paintWait = m->paintWaitMax; m->paintWaitMax = 0;
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video GUI delivery call=%llu interval_us=%lld published=%u "
"notification_wait_max_us=%lld delivery_gap_max_us=%lld paint_wait_max_us=%lld image_slot_max_us=%lld mono_us=%lld",
(unsigned long long)m->id, (long long)(now - m->uiReportAt), m->uiPublished,
@@ -158,15 +185,30 @@ void CallVideoEngine::publishImage(const QImage& image, int64_t pts) {
if (!image.isNull() && pts >= 0 && pts % 1000000 < 34000)
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video timing stage=publish call=%llu pts=%lld mono_us=%lld ui_wait_us=%lld",
(unsigned long long)m->id, (long long)pts, (long long)now, (long long)wait);
}, Qt::QueuedConnection);
}
void CallVideoEngine::start(quint64 id) {
if (m->running && m->id == id) return;
stop(); m->id = id; m->running = true; m_timer.start();
stop();
if (!m->rxWake.valid() || !m->displayWake.valid() || !m->guiNotifier) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video worker start: wake handles unavailable"); emit statusChanged(tr("Video timer is unavailable")); return;
}
m->id = id; m->running = true; m_timer.start();
m->uiReportAt = m->uiLastAt = m->uiWaitMax = m->uiGapMax = 0; m->uiPublished = 0;
m->uiTickAt = m->uiTickReportAt = m->uiTickGapMax = m->uiImageSlotMax = 0; m->uiTicks = 0;
m->presentedAt = 0; m->presentedPts = -1;
try { m->tx = std::thread(&CallVideoEngine::transmit, this); m->rx = std::thread(&CallVideoEngine::receive, this); }
m->pictures.reset(); m->logEvents.reset(); m->timerReports.reset(); m->discarded = 0; m->pipeline = 0; m->lostLogs = 0;
m->pendingSerial = 0; m->ackSerial = 0; m->presentedAt = 0;
call_video_playout_init(&m->playout, id);
m->playout.log_arg = m.get();
m->playout.log_event = [](void* arg, const call_video_playout_event* event) {
auto* state = static_cast<Impl*>(arg);
if (auto* slot = state->logEvents.writeSlot()) { *slot = *event; state->logEvents.publish(); }
else state->lostLogs.fetch_add(1);
state->rxWake.notify();
};
try {
m->tx = std::thread(&CallVideoEngine::transmit, this); m->rx = std::thread(&CallVideoEngine::receive, this);
m->display = std::thread(&CallVideoEngine::present, this);
}
catch (const std::exception& error) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video worker start: %s", error.what()); stop(); emit statusChanged(error.what()); }
}
void CallVideoEngine::closeCamera() {
@@ -181,11 +223,13 @@ void CallVideoEngine::stop() {
m->running = false; m_timer.stop(); closeCamera();
m->wake.notify_all();
m->rxWake.notify();
m->displayWake.notify();
if (m->display.joinable()) m->display.join();
if (m->tx.joinable()) m->tx.join();
if (m->rx.joinable()) m->rx.join();
if (m->id) call_video_port_close(m->id);
{ std::lock_guard<std::mutex> guard(m->lock);
++m->publicationEpoch; m->publicationPending = false; m->paintWaitMax = 0; }
++m->publicationEpoch; m->pendingSerial = 0; m->paintWaitMax = 0;
while (const auto* event = m->logEvents.readSlot()) { call_video_playout_log_event(event); m->logEvents.release(); }
m->id = 0; m->enabled = false; emit imageChanged({}, 0);
}
void CallVideoEngine::setVideo(bool enabled, const QByteArray& cameraId, int width, int height, int codec) {
@@ -382,95 +426,80 @@ void CallVideoEngine::transmit() {
}
void CallVideoEngine::receive() {
struct Picture { call_video_playout_frame timing; QImage image; };
std::deque<Picture> pictures;
auto& pictures = m->pictures;
CallVideoDecoder decoder;
call_video_playout playout; call_video_playout_init(&playout, m->id);
int64_t lastPictureAt = 0;
int64_t reportAt = nowUs(), decodeMax = 0, rxAgeMax = 0, wakeLateMax = 0, requestedWake = 0;
unsigned decoded = 0;
unsigned submitted = 0, acknowledged = 0;
unsigned submitted = 0;
uint64_t reportedGeneration = 0, rejectedGeneration = 0;
call_video_frame input{};
bool wakeBound = false, waitingKey = false;
int64_t decoderProgressAt = nowUs();
unsigned decodeDeadlineMisses = 0;
int64_t decodeDeadlineMax = 0, decodeDeadlineWork = 0, decodeDeadlineMargin = 0;
const char* decodeDeadlineStage = "none";
auto recordDeadlineWork = [&](const char* stage, int64_t started, int64_t finished) {
// Только уже назначенный показ: ожидание аудио и отсутствие готовой головы сюда не относятся.
if (pictures.empty() || playout.awaiting || !playout.audio_valid
|| playout.head_pts != pictures.front().timing.pts_us || playout.deadline_us != playout.head_due_us
|| started >= playout.head_due_us || finished <= playout.head_due_us) return;
++decodeDeadlineMisses;
if (finished - playout.head_due_us > decodeDeadlineMax) {
decodeDeadlineMax = finished - playout.head_due_us; decodeDeadlineWork = finished - started;
decodeDeadlineMargin = playout.head_due_us - started; decodeDeadlineStage = stage;
}
};
DEBUG_INFO(DEBUG_CATEGORY_CALL, "video RX scheduling call=%llu decoder_and_playout_shared=1 ready_capacity=3 "
"deadline_overlap_diagnostics=1", (unsigned long long)m->id);
DEBUG_INFO(DEBUG_CATEGORY_CALL, "video RX scheduling call=%llu decoder_and_playout_shared=0 ready_capacity=3 "
"gui_publication=native-event timer_logs=deferred", (unsigned long long)m->id);
while (m->running) {
const uint64_t ticket = m->rxWake.ticket();
if (requestedWake) wakeLateMax = std::max(wakeLateMax, nowUs() - requestedWake);
if (!wakeBound) wakeBound = call_video_port_set_wake(m->id, [](void* arg) { static_cast<Impl*>(arg)->rxWake.notify(); }, m.get());
int64_t presentedAt, presentedPts;
{ std::lock_guard<std::mutex> guard(m->lock);
presentedAt = m->presentedAt; presentedPts = m->presentedPts; m->presentedAt = 0; }
if (presentedAt && playout.awaiting && presentedPts >= 0) {
call_video_playout_presented(&playout, presentedPts, presentedAt); ++acknowledged; lastPictureAt = presentedAt;
while (const auto* event = m->logEvents.readSlot()) { call_video_playout_log_event(event); m->logEvents.release(); }
while (const auto* r = m->timerReports.readSlot()) {
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video display worker call=%llu interval_us=%lld work_max_us=%lld wake_late_max_us=%lld "
"audio_busy=%u ready=%u total=%u presented=%u", (unsigned long long)m->id, (long long)r->interval,
(long long)r->workMax, (long long)r->wakeLate, r->busy, r->ready, r->total, r->presented);
m->timerReports.release();
}
int64_t now = nowUs(); call_clock_snapshot clock{};
call_audio_get_clock(m->id, now, &clock);
struct call_video_port_status status{}; call_video_port_status(m->id, &status);
unsigned total = status.rx_frames + (input.data != nullptr) + decoder.pendingFrames()
+ static_cast<unsigned>(pictures.size()) + playout.awaiting;
auto action = call_video_playout_poll(&playout, pictures.empty() ? nullptr : &pictures.front().timing,
now, &clock, total, 1 + decoder.pipelineFrames());
if (action != CALL_VIDEO_PLAYOUT_WAIT) {
if (action == CALL_VIDEO_PLAYOUT_SHOW) {
const auto& f = pictures.front(); publishImage(f.image, f.timing.pts_us);
}
pictures.pop_front(); requestedWake = 0; continue;
if (unsigned lost = m->lostLogs.exchange(0)) DEBUG_WARN(DEBUG_CATEGORY_CALL, "video deferred log overflow lost=%u", lost);
for (auto* wake : {&m->rxWake, &m->displayWake, &m->guiWake}) if (unsigned error = wake->takeError()) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video wake failed error=%u; stopping workers", error);
m->running = false; m->rxWake.notify(); m->displayWake.notify();
}
int64_t now = nowUs();
bool worked = false;
// Три места включают выбранный GUI кадр. Выход GPU читается независимо от новых входов.
while (pictures.size() + playout.awaiting < 3 && m->running) {
while (pictures.writeSlot() && m->running) {
call_video_frame packet{}; QImage image;
const int64_t started = nowUs(); int result = decoder.receive(packet, image);
const int64_t finished = nowUs();
decodeMax = std::max(decodeMax, finished - started); recordDeadlineWork("output", started, finished);
if (!result && !playout.awaiting && pictures.empty() && decoder.pendingFrames() > decoder.pipelineFrames()
decodeMax = std::max(decodeMax, finished - started);
if (!result && !pictures.size() && decoder.pendingFrames() > decoder.pipelineFrames()
&& nowUs() - decoderProgressAt > 2000000) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video decoder stalled call=%llu pending=%u pipeline=%u progress_age_us=%lld",
(unsigned long long)m->id, decoder.pendingFrames(), decoder.pipelineFrames(), (long long)(nowUs() - decoderProgressAt));
result = -1;
}
if (result < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video decoder recovery call=%llu lost_inputs=%u retained_ready=%zu awaiting=%d error=%s",
(unsigned long long)m->id, decoder.pendingFrames(), pictures.size(), playout.awaiting, qPrintable(decoder.error()));
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video decoder recovery call=%llu lost_inputs=%u retained_ready=%u error=%s",
(unsigned long long)m->id, decoder.pendingFrames(), pictures.size(), qPrintable(decoder.error()));
m->report(decoder.error().isEmpty() ? tr("Video decoder stopped producing frames") : decoder.error());
m->discarded.fetch_add(decoder.pendingFrames(), std::memory_order_release);
if (decoder.hardware()) decoder.useSoftware(); else decoder.close();
waitingKey = true; call_video_port_request_key(m->id); break;
}
if (!result) break;
decoderProgressAt = nowUs();
++decoded; worked = true; rxAgeMax = std::max(rxAgeMax, nowUs() - packet.received_us);
pictures.push_back({{packet.pts_us, packet.received_us, 1000000u / packet.format.fps}, image});
auto* slot = pictures.writeSlot();
slot->timing = {packet.pts_us, packet.received_us, 1000000u / packet.format.fps}; slot->image = std::move(image);
m->pipeline.store(decoder.pipelineFrames(), std::memory_order_release);
pictures.publish(); m->displayWake.notify();
if (reportedGeneration != packet.format.generation) {
reportedGeneration = packet.format.generation; call_video_port_decoded(m->id, reportedGeneration, 1);
}
}
if (pictures.size() + playout.awaiting < 3 && decoder.pendingFrames() < 8) {
if (pictures.writeSlot() && decoder.pendingFrames() < 8 && m->running) {
if (!input.data) call_video_port_pull(m->id, &input);
if (input.data) {
if (input.format.generation == rejectedGeneration || (waitingKey && !(input.flags & CALL_VIDEO_KEY))) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "video decoder rejects dependent input call=%llu frame=%llu generation=%llu waiting_key=%d",
(unsigned long long)m->id, (unsigned long long)input.frame_id,
(unsigned long long)input.format.generation, waitingKey);
m->discarded.fetch_add(1, std::memory_order_release);
u_free((void*)input.data); input = {}; requestedWake = 0; continue;
}
const int64_t started = nowUs();
int result = decoder.submit(input);
const int64_t finished = nowUs();
decodeMax = std::max(decodeMax, finished - started); recordDeadlineWork("input", started, finished);
decodeMax = std::max(decodeMax, finished - started);
if (result) {
worked = true;
if (result > 0) {
@@ -479,6 +508,7 @@ void CallVideoEngine::receive() {
}
else {
m->report(decoder.error());
m->discarded.fetch_add(decoder.pendingFrames(), std::memory_order_release);
if (decoder.hardware()) {
decoder.useSoftware(); waitingKey = true;
// Не принятый KEY остаётся у worker: CPU повторяет тот же кадр без потери.
@@ -489,37 +519,81 @@ void CallVideoEngine::receive() {
call_video_port_decoded(m->id, rejectedGeneration, 0); decoder.close();
}
waitingKey = true; call_video_port_request_key(m->id);
m->discarded.fetch_add(1, std::memory_order_release);
}
u_free((void*)input.data); input = {};
}
}
}
if (now - reportAt >= 1000000) {
struct call_video_port_status status{}; call_video_port_status(m->id, &status);
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video RX worker call=%llu interval_us=%lld decoded=%u decode_max_us=%lld "
"rx_to_decode_max_us=%lld wake_late_max_us=%lld ready=%zu audio_valid=%d audio_measured=%d "
"audio_pts=%lld played=%llu written=%llu mono_us=%lld submitted=%u presented=%u encoded=%u inflight=%u total=%u "
"decode_deadline_misses=%u decode_deadline_max_us=%lld decode_deadline_stage=%s "
"decode_deadline_work_us=%lld decode_deadline_margin_us=%lld",
"rx_to_decode_max_us=%lld wake_late_max_us=%lld ready=%u submitted=%u encoded=%u inflight=%u discarded=%llu",
(unsigned long long)m->id, (long long)(now - reportAt), decoded, (long long)decodeMax, (long long)rxAgeMax,
(long long)wakeLateMax, pictures.size(), clock.valid, clock.measured, (long long)clock.pts_us,
(unsigned long long)clock.played_samples, (unsigned long long)clock.written_samples, (long long)now,
submitted, acknowledged, status.rx_frames, decoder.pendingFrames(), total, decodeDeadlineMisses,
(long long)decodeDeadlineMax, decodeDeadlineStage, (long long)decodeDeadlineWork, (long long)decodeDeadlineMargin);
(long long)wakeLateMax, pictures.size(), submitted, status.rx_frames, decoder.pendingFrames(),
(unsigned long long)m->discarded.load());
reportAt = now; decoded = 0; decodeMax = rxAgeMax = wakeLateMax = 0;
submitted = acknowledged = 0;
decodeDeadlineMisses = 0; decodeDeadlineMax = decodeDeadlineWork = decodeDeadlineMargin = 0; decodeDeadlineStage = "none";
submitted = 0;
}
if (worked) { requestedWake = 0; continue; }
// VIDEO_OFF пока отсутствует в протоколе: убираем застывшую картинку после прекращения показа.
if (!playout.awaiting && lastPictureAt && now - lastPictureAt >= 5000000) {
publishImage({}, -1); lastPictureAt = 0;
DEBUG_INFO(DEBUG_CATEGORY_CALL, "video playback hidden: no playable frame for 5s");
}
const int64_t sleepAt = nowUs();
const int64_t waitUs = std::max<int64_t>(100, std::min<int64_t>(5000, playout.deadline_us - sleepAt));
const int64_t waitUs = decoder.pendingFrames() && pictures.writeSlot() ? 1000 : 5000;
requestedWake = sleepAt + waitUs;
m->rxWake.wait(requestedWake, ticket);
}
call_video_port_set_wake(m->id, nullptr, nullptr);
u_free((void*)input.data);
}
void CallVideoEngine::present() {
auto& playout = m->playout;
uint64_t completed = 0;
int64_t lastPictureAt = 0, reportAt = nowUs(), requestedWake = 0;
Impl::TimerReport metrics{};
while (m->running) {
const uint64_t ticket = m->displayWake.ticket();
const int64_t started = nowUs();
if (requestedWake) metrics.wakeLate = std::max(metrics.wakeLate, started - requestedWake);
uint64_t serial = m->pendingSerial.load(std::memory_order_acquire);
if (serial && m->ackSerial.load(std::memory_order_acquire) == serial) {
if (playout.awaiting) {
const auto* slot = m->pictures.readSlot();
lastPictureAt = m->presentedAt.load(std::memory_order_relaxed);
call_video_playout_presented(&playout, slot->timing.pts_us, lastPictureAt);
m->pictures.release(); ++completed; ++metrics.presented; m->rxWake.notify();
}
m->pendingSerial.store(0, std::memory_order_release); serial = 0;
}
const uint64_t discarded = m->discarded.load(std::memory_order_acquire);
const uint64_t received = call_video_port_received(m->id);
const unsigned ready = m->pictures.size();
const unsigned total = static_cast<unsigned>(std::max<uint64_t>(ready,
received >= completed + discarded ? received - completed - discarded : 0));
call_clock_snapshot clock{};
const int clockResult = call_audio_try_clock(m->id, started, &clock);
auto action = CALL_VIDEO_PLAYOUT_WAIT;
if (clockResult) {
const auto* slot = serial ? nullptr : m->pictures.readSlot();
action = call_video_playout_poll(&playout, slot ? &slot->timing : nullptr, started, &clock, total,
1 + m->pipeline.load(std::memory_order_acquire));
if (action == CALL_VIDEO_PLAYOUT_SHOW) publishImage(slot->image, slot->timing.pts_us);
else if (action == CALL_VIDEO_PLAYOUT_DROP_ORDER) {
m->pictures.release(); ++completed; m->rxWake.notify();
}
} else ++metrics.busy;
if (!serial && !playout.awaiting && lastPictureAt && started - lastPictureAt >= 5000000) {
publishImage({}, -1); lastPictureAt = 0;
}
metrics.workMax = std::max(metrics.workMax, nowUs() - started);
if (started - reportAt >= 1000000) {
metrics.interval = started - reportAt; metrics.ready = ready; metrics.total = total;
if (auto* slot = m->timerReports.writeSlot()) { *slot = metrics; m->timerReports.publish(); }
else m->lostLogs.fetch_add(1);
m->rxWake.notify(); reportAt = started; metrics = {};
}
if (action != CALL_VIDEO_PLAYOUT_WAIT) { requestedWake = 0; continue; }
const int64_t now = nowUs();
const int64_t waitUs = clockResult ? std::max<int64_t>(100, std::min<int64_t>(5000, playout.deadline_us - now)) : 500;
requestedWake = now + waitUs; m->displayWake.wait(requestedWake, ticket);
}
}
+3 -1
View File
@@ -5,7 +5,7 @@
#include <memory>
#include "call_camera_capture.h"
// Камера изолирована в процессе; TX/RX владеют codec и получают только CPU-изображения.
// Камера изолирована в процессе. TX/decoder владеют codec; отдельный playout получает готовые CPU-изображения.
class CallVideoEngine : public QObject {
Q_OBJECT
public:
@@ -28,5 +28,7 @@ private:
void closeCamera();
void transmit();
void receive();
void present();
void deliverImage(); // GUI notifier: чтение единственного опубликованного слота.
void publishImage(const QImage& image, int64_t pts);
};
+12
View File
@@ -61,6 +61,18 @@ public:
#endif
}
unsigned takeError() { return error.exchange(0); }
#ifdef _WIN32
HANDLE nativeHandle() const { return event; }
#else
int nativeHandle() const { return pipefd[0]; }
#endif
// Только GUI notifier вместо wait(); чтение ограничено одним неблокирующим вызовом.
void consume() {
#ifndef _WIN32
char bytes[128];
if (read(pipefd[0], bytes, sizeof(bytes)) < 0 && errno != EAGAIN) error.store(errno);
#endif
}
uint64_t ticket() const { return serial.load(std::memory_order_acquire); }
void notify() {
serial.fetch_add(1, std::memory_order_release);