diff --git a/tools/chatgui/src/call_video_engine.cpp b/tools/chatgui/src/call_video_engine.cpp index ac0c101b..19995122 100644 --- a/tools/chatgui/src/call_video_engine.cpp +++ b/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 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 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 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(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; } 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(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(100, std::min(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); } diff --git a/tools/chatgui/src/call_video_wake.h b/tools/chatgui/src/call_video_wake.h new file mode 100644 index 00000000..b2304e73 --- /dev/null +++ b/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 +#include +#include +#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 guard(lock); return serial; } + void notify() { + { std::lock_guard 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 guard(lock); + changed.wait_for(guard, std::chrono::microseconds(remaining), [&] { return serial != expected; }); + } +};