diff --git a/tools/chatgui/src/call_video_queue.h b/tools/chatgui/src/call_video_queue.h new file mode 100644 index 00000000..f71823ca --- /dev/null +++ b/tools/chatgui/src/call_video_queue.h @@ -0,0 +1,36 @@ +/* Ограниченный SPSC FIFO: один producer и один consumer, без mutex и аллокаций. + * Producer заполняет writeSlot(), затем publish(); consumer читает readSlot(), затем release(). + * До release слот неизменяем, в том числе во время ожидания GUI ACK. + * reset/destruction — только после остановки обоих потоков. Старый payload заменяет producer. */ +#pragma once +#include +#include +#include + +template class CallVideoQueue { + static_assert(Capacity > 0); + static_assert(std::atomic::is_always_lock_free); + std::array slots{}; + alignas(64) std::atomic written{0}; + alignas(64) std::atomic consumed{0}; +public: + T* writeSlot() { + const uint64_t tail = written.load(std::memory_order_relaxed); + return tail - consumed.load(std::memory_order_acquire) < Capacity ? &slots[tail % Capacity] : nullptr; + } + void publish() { written.fetch_add(1, std::memory_order_release); } + const T* readSlot() const { + const uint64_t head = consumed.load(std::memory_order_relaxed); + return head < written.load(std::memory_order_acquire) ? &slots[head % Capacity] : nullptr; + } + void release() { consumed.fetch_add(1, std::memory_order_release); } + // Вызывается только producer/consumer: собственный индекс во время снимка не меняется. + unsigned size() const { + const uint64_t head = consumed.load(std::memory_order_acquire); + return static_cast(written.load(std::memory_order_acquire) - head); + } + void reset() { + for (auto& slot : slots) slot = {}; + written.store(0, std::memory_order_relaxed); consumed.store(0, std::memory_order_relaxed); + } +}; diff --git a/tools/chatgui/src/call_video_wake.h b/tools/chatgui/src/call_video_wake.h index 6cb74426..c7d0ee31 100644 --- a/tools/chatgui/src/call_video_wake.h +++ b/tools/chatgui/src/call_video_wake.h @@ -1,27 +1,28 @@ -/* Прерываемое ожидание RX по абсолютному MONOTONIC deadline. - * notify не вызывает API порта; может исполняться под mutex producer. - * Windows: auto-reset event + high-resolution waitable timer, без busy-spin/timeBeginPeriod. - * Остальные платформы: condition_variable. ticket исключает потерю уведомления перед wait. */ +/* Ожидание по абсолютному MONOTONIC deadline, отдельный объект на каждого waiter. + * notify не захватывает mutex и допустим под mutex порта. ticket закрывает гонку до wait. + * Windows: event + high-resolution timer; POSIX: неблокирующий pipe + pselect. + * Ошибки исполнения забирает reporter через takeError(), I/O логов в таймере нет. */ #pragma once #include "../../../src/call/call_video_playout.h" #include "debug_config.h" -#include -#include -#include -#ifdef _WIN32 +#include +#include +#include #ifndef NOMINMAX #define NOMINMAX #endif #include "../../../lib/platform_compat.h" +#ifndef _WIN32 +#include #endif class CallVideoWake { - std::mutex lock; - std::condition_variable changed; - uint64_t serial = 0; + std::atomic serial{0}; + std::atomic error{0}; #ifdef _WIN32 HANDLE event = nullptr, timer = nullptr; - bool failed = false; +#else + int pipefd[2] = {-1, -1}; #endif public: CallVideoWake() { @@ -36,38 +37,64 @@ public: 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); +#else + if (pipe(pipefd) || fcntl(pipefd[0], F_SETFL, O_NONBLOCK) < 0 || fcntl(pipefd[1], F_SETFL, O_NONBLOCK) < 0 + || fcntl(pipefd[0], F_SETFD, FD_CLOEXEC) < 0 || fcntl(pipefd[1], F_SETFD, FD_CLOEXEC) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video wake pipe setup failed errno=%d", errno); + for (int& fd : pipefd) { if (fd >= 0) close(fd); fd = -1; } + } #endif } ~CallVideoWake() { #ifdef _WIN32 if (timer) CloseHandle(timer); if (event) CloseHandle(event); +#else + for (int fd : pipefd) if (fd >= 0) close(fd); #endif } - uint64_t ticket() { std::lock_guard guard(lock); return serial; } - void notify() { - { std::lock_guard guard(lock); ++serial; } - changed.notify_one(); + bool valid() const { #ifdef _WIN32 - if (event && !SetEvent(event)) DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video RX SetEvent failed error=%lu", GetLastError()); + return event && timer; +#else + return pipefd[0] >= 0 && pipefd[0] < FD_SETSIZE; +#endif + } + unsigned takeError() { return error.exchange(0); } + uint64_t ticket() const { return serial.load(std::memory_order_acquire); } + void notify() { + serial.fetch_add(1, std::memory_order_release); +#ifdef _WIN32 + if (event && !SetEvent(event)) error.store(GetLastError()); +#else + if (pipefd[1] >= 0) { + const char byte = 0; + if (write(pipefd[1], &byte, 1) < 0 && errno != EAGAIN) error.store(errno); + } #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) { + if (event && timer) { 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()); + error.store(GetLastError()); } +#else + if (!valid()) return; + fd_set fds; FD_ZERO(&fds); FD_SET(pipefd[0], &fds); + timespec timeout{remaining / 1000000, (remaining % 1000000) * 1000}; + int result = pselect(pipefd[0] + 1, &fds, nullptr, nullptr, &timeout, nullptr); + if (result > 0) { + char bytes[128]; + if (read(pipefd[0], bytes, sizeof(bytes)) < 0 && errno != EAGAIN) error.store(errno); + } else if (result < 0 && errno != EINTR) error.store(errno); #endif - std::unique_lock guard(lock); - changed.wait_for(guard, std::chrono::microseconds(remaining), [&] { return serial != expected; }); } };