From d9843b8cbdd4bc33b54472e6ac1dfa2d4b762351 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 19 Oct 2026 21:12:02 +0200 Subject: [PATCH] Separate video decoding from event-driven presentation timing --- tools/chatgui/src/call_video_engine.cpp | 270 +++++++++++++++--------- tools/chatgui/src/call_video_engine.h | 4 +- tools/chatgui/src/call_video_wake.h | 12 ++ 3 files changed, 187 insertions(+), 99 deletions(-) diff --git a/tools/chatgui/src/call_video_engine.cpp b/tools/chatgui/src/call_video_engine.cpp index d75736dc..6fda6866 100644 --- a/tools/chatgui/src/call_video_engine.cpp +++ b/tools/chatgui/src/call_video_engine.cpp @@ -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 #include #include +#ifdef _WIN32 +#include +#else +#include +#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 pictures; + CallVideoQueue logEvents; + struct TimerReport { + int64_t interval = 0, workMax = 0, wakeLate = 0; + unsigned busy = 0, ready = 0, total = 0, presented = 0; + }; + CallVideoQueue timerReports; + std::atomic discarded{0}; + std::atomic 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 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 publicationEpoch{0}, pendingSerial{0}, ackSerial{0}; + std::atomic 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()) { + 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 guard(m->lock); m->presentationDriven = enabled; } void CallVideoEngine::imagePresented(quint64 serial) { - { std::lock_guard 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 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 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 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 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(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 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 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(arg)->rxWake.notify(); }, m.get()); - int64_t presentedAt, presentedPts; - { std::lock_guard 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(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(100, std::min(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(std::max(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(100, std::min(5000, playout.deadline_us - now)) : 500; + requestedWake = now + waitUs; m->displayWake.wait(requestedWake, ticket); + } +} diff --git a/tools/chatgui/src/call_video_engine.h b/tools/chatgui/src/call_video_engine.h index d909bc2d..481fde01 100644 --- a/tools/chatgui/src/call_video_engine.h +++ b/tools/chatgui/src/call_video_engine.h @@ -5,7 +5,7 @@ #include #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); }; diff --git a/tools/chatgui/src/call_video_wake.h b/tools/chatgui/src/call_video_wake.h index c7d0ee31..dceef37d 100644 --- a/tools/chatgui/src/call_video_wake.h +++ b/tools/chatgui/src/call_video_wake.h @@ -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);