Drive desktop video from presentation ACKs with three decoded slots and interruptible deadlines

tools/chatgui/src/call_video_engine.cpp

@@ -1,6 +1,7 @@
 #include "call_video_engine.h"
 #include "call_video_encoder.h"
 #include "call_video_decoder.h"
+#include "call_video_wake.h"
 #include "call_camera_capture.h"
 #include "../../../src/call/call_video_port.h"
 #include "../../../src/call/call_video_playout.h"
@@ -20,6 +21,7 @@ int64_t nowUs() { return call_video_playout_now_us(); }
 struct CallVideoEngine::Impl {
     std::mutex lock;
     std::condition_variable wake;
+    CallVideoWake rxWake;
     uint64_t wakeSerial = 0;
     std::thread tx, rx;
     std::atomic<bool> running{false};
@@ -38,6 +40,9 @@ struct CallVideoEngine::Impl {
     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 publicationPts = -1;
     int64_t uiReportAt = 0, uiLastAt = 0, uiWaitMax = 0, uiGapMax = 0; // Только GUI-поток.
     int64_t uiTickAt = 0, uiTickReportAt = 0, uiTickGapMax = 0, uiImageSlotMax = 0;
     unsigned uiTicks = 0;
@@ -108,8 +113,10 @@ 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();
 }
 // Только один кадр до подтверждения рисования: GUI не заменяет ещё не показанное изображение.
 void CallVideoEngine::publishImage(const QImage& image, int64_t pts) {
@@ -118,6 +125,7 @@ void CallVideoEngine::publishImage(const QImage& image, int64_t pts) {
       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);
@@ -157,6 +165,7 @@ void CallVideoEngine::start(quint64 id) {
     stop(); 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); }
     catch (const std::exception& error) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video worker start: %s", error.what()); stop(); emit statusChanged(error.what()); }
 }
@@ -171,6 +180,7 @@ void CallVideoEngine::closeCamera() {
 void CallVideoEngine::stop() {
     m->running = false; m_timer.stop(); closeCamera();
     m->wake.notify_all();
+    m->rxWake.notify();
     if (m->tx.joinable()) m->tx.join();
     if (m->rx.joinable()) m->rx.join();
     if (m->id) call_video_port_close(m->id);
@@ -372,68 +382,88 @@ void CallVideoEngine::receive() {
     int64_t lastPictureAt = 0;
     int64_t reportAt = nowUs(), decodeMax = 0, rxAgeMax = 0, wakeLateMax = 0, requestedWake = 0;
     unsigned decoded = 0;
+    unsigned submitted = 0, acknowledged = 0;
     uint64_t reportedGeneration = 0, rejectedGeneration = 0;
+    call_video_frame input{};
+    bool wakeBound = false;
     while (m->running) {
+        const uint64_t ticket = m->rxWake.ticket();
         if (requestedWake) wakeLateMax = std::max(wakeLateMax, nowUs() - requestedWake);
-        // Малый decoded lookahead; накопленный долг остаётся в encoded FIFO.
-        while (pictures.size() < 2 && m->running) {
-            call_video_frame packet{};
-            if (call_video_port_pull(m->id, &packet) == 1) {
-                if (packet.format.generation == rejectedGeneration) { u_free((void*)packet.data); continue; }
-                QImage image;
-                const int64_t decodeStarted = nowUs();
-                if (decoder.decode(packet, image)) {
-                    const int64_t decodedAt = nowUs();
-                    ++decoded; decodeMax = std::max(decodeMax, decodedAt - decodeStarted);
-                    rxAgeMax = std::max(rxAgeMax, decodedAt - packet.received_us);
-                    pictures.push_back({{packet.pts_us, packet.received_us, 1000000u / packet.format.fps}, image});
-                    if (packet.pts_us % 1000000 < 34000)
-                        DEBUG_DEBUG(DEBUG_CATEGORY_CALL,
-                            "video timing stage=decode call=%llu frame=%llu pts=%lld mono_us=%lld decode_us=%lld rx_age_us=%lld queue=%zu",
-                            (unsigned long long)m->id, (unsigned long long)packet.frame_id, (long long)packet.pts_us,
-                            (long long)decodedAt, (long long)(decodedAt - decodeStarted),
-                            (long long)(decodedAt - packet.received_us), pictures.size());
-                    if (reportedGeneration != packet.format.generation) {
-                        reportedGeneration = packet.format.generation; call_video_port_decoded(m->id, reportedGeneration, 1);
-                    }
-                } else {
-                    rejectedGeneration = packet.format.generation; call_video_port_decoded(m->id, rejectedGeneration, 0);
-                }
-                u_free((void*)packet.data);
-            } else break;
+        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;
         }
         int64_t now = nowUs(); call_clock_snapshot clock{};
         call_audio_get_clock(m->id, now, &clock);
+        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;
+        }
+        bool worked = false;
+        // Три места включают выбранный GUI кадр. Выход GPU читается независимо от новых входов.
+        while (pictures.size() + playout.awaiting < 3 && m->running) {
+            call_video_frame packet{}; QImage image;
+            const int64_t started = nowUs(); int result = decoder.receive(packet, image);
+            decodeMax = std::max(decodeMax, nowUs() - started);
+            if (result < 0) {
+                m->report(decoder.error()); decoder.close(); call_video_port_request_key(m->id); break;
+            }
+            if (!result) break;
+            ++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});
+            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 (!input.data) call_video_port_pull(m->id, &input);
+            if (input.data) {
+                const int64_t started = nowUs();
+                int result = decoder.submit(input);
+                decodeMax = std::max(decodeMax, nowUs() - started);
+                if (result) {
+                    worked = true;
+                    if (result > 0) ++submitted;
+                    else {
+                        rejectedGeneration = input.format.generation;
+                        call_video_port_decoded(m->id, rejectedGeneration, 0); m->report(decoder.error()); decoder.close();
+                    }
+                    u_free((void*)input.data); input = {};
+                }
+            }
+        }
         if (now - reportAt >= 1000000) {
             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",
+                "audio_pts=%lld played=%llu written=%llu mono_us=%lld submitted=%u presented=%u encoded=%u inflight=%u total=%u",
                 (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);
+                (unsigned long long)clock.played_samples, (unsigned long long)clock.written_samples, (long long)now,
+                submitted, acknowledged, status.rx_frames, decoder.pendingFrames(), total);
             reportAt = now; decoded = 0; decodeMax = rxAgeMax = wakeLateMax = 0;
+            submitted = acknowledged = 0;
         }
-        auto action = call_video_playout_poll(&playout, pictures.empty() ? nullptr : &pictures.front().timing,
-            pictures.size() > 1 ? &pictures[1].timing : nullptr, now, &clock);
-        if (action != CALL_VIDEO_PLAYOUT_WAIT) {
-            if (action == CALL_VIDEO_PLAYOUT_SHOW) {
-                const auto& f = pictures.front(); publishImage(f.image, f.timing.pts_us); lastPictureAt = now;
-                if (f.timing.pts_us % 1000000 < 34000)
-                    DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video timing stage=select call=%llu pts=%lld mono_us=%lld audio_pts=%lld",
-                        (unsigned long long)m->id, (long long)f.timing.pts_us, (long long)now, (long long)clock.pts_us);
-            }
-            pictures.pop_front();
-            requestedWake = 0;
-            continue;
-        }
+        if (worked) { requestedWake = 0; continue; }
         // VIDEO_OFF пока отсутствует в протоколе: убираем застывшую картинку после прекращения показа.
-        if (lastPictureAt && now - lastPictureAt >= 5000000) {
+        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));
         requestedWake = sleepAt + waitUs;
-        std::this_thread::sleep_for(std::chrono::microseconds(waitUs));
+        m->rxWake.wait(requestedWake, ticket);
     }
+    call_video_port_set_wake(m->id, nullptr, nullptr);
+    u_free((void*)input.data);
 }

tools/chatgui/src/call_video_wake.h

@@ -0,0 +1,72 @@
+/* Прерываемое ожидание RX по абсолютному MONOTONIC deadline.
+ * notify не вызывает API порта; может исполняться под mutex producer.
+ * Windows: auto-reset event + high-resolution waitable timer, без busy-spin/timeBeginPeriod.
+ * Остальные платформы: condition_variable. ticket исключает потерю уведомления перед wait. */
+#pragma once
+#include "../../../src/call/call_video_playout.h"
+#include "debug_config.h"
+#include <condition_variable>
+#include <mutex>
+#include <chrono>
+#ifdef _WIN32
+#ifndef NOMINMAX
+#define NOMINMAX
+#endif
+#include "../../../lib/platform_compat.h"
+#endif
+
+class CallVideoWake {
+    std::mutex lock;
+    std::condition_variable changed;
+    uint64_t serial = 0;
+#ifdef _WIN32
+    HANDLE event = nullptr, timer = nullptr;
+    bool failed = false;
+#endif
+public:
+    CallVideoWake() {
+#ifdef _WIN32
+        event = CreateEventW(nullptr, FALSE, FALSE, nullptr);
+        timer = CreateWaitableTimerExW(nullptr, nullptr, 0x00000002, TIMER_ALL_ACCESS);
+        if (!timer) {
+            DEBUG_WARN(DEBUG_CATEGORY_CALL, "video high resolution timer unavailable error=%lu; using regular timer", GetLastError());
+            timer = CreateWaitableTimerW(nullptr, FALSE, nullptr);
+        }
+        if (!event || !timer) DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video wait handles unavailable event=%p timer=%p error=%lu",
+            event, timer, GetLastError());
+        else DEBUG_INFO(DEBUG_CATEGORY_CALL, "video RX waitable timer ready event=%p timer=%p", event, timer);
+#endif
+    }
+    ~CallVideoWake() {
+#ifdef _WIN32
+        if (timer) CloseHandle(timer);
+        if (event) CloseHandle(event);
+#endif
+    }
+    uint64_t ticket() { std::lock_guard<std::mutex> guard(lock); return serial; }
+    void notify() {
+        { std::lock_guard<std::mutex> guard(lock); ++serial; }
+        changed.notify_one();
+#ifdef _WIN32
+        if (event && !SetEvent(event)) DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video RX SetEvent failed error=%lu", GetLastError());
+#endif
+    }
+    void wait(int64_t deadline, uint64_t expected) {
+        const int64_t remaining = deadline - call_video_playout_now_us();
+        if (remaining <= 0 || ticket() != expected) return;
+#ifdef _WIN32
+        if (event && timer && !failed) {
+            LARGE_INTEGER due; due.QuadPart = -remaining * 10;
+            if (SetWaitableTimer(timer, &due, 0, nullptr, nullptr, FALSE)) {
+                HANDLE handles[] = {event, timer};
+                DWORD result = WaitForMultipleObjects(2, handles, FALSE, INFINITE);
+                if (result == WAIT_OBJECT_0 || result == WAIT_OBJECT_0 + 1) return;
+            }
+            failed = true;
+            DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video RX waitable timer failed error=%lu; using condition_variable", GetLastError());
+        }
+#endif
+        std::unique_lock<std::mutex> guard(lock);
+        changed.wait_for(guard, std::chrono::microseconds(remaining), [&] { return serial != expected; });
+    }
+};