Add bounded video SPSC slots and mutex-free wake events

This commit is contained in:
evgeny committed 2026-10-19 21:08:41 +02:00
1 parent 9ca0e61361
commit 402c0c0611
2 files changed
+85 -22

No files matched your search

+36
View File
@@ -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 <array>
#include <atomic>
#include <cstdint>
template<class T, unsigned Capacity> class CallVideoQueue {
static_assert(Capacity > 0);
static_assert(std::atomic<uint64_t>::is_always_lock_free);
std::array<T, Capacity> slots{};
alignas(64) std::atomic<uint64_t> written{0};
alignas(64) std::atomic<uint64_t> 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<unsigned>(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);
}
};
+49 -22
View File
@@ -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 <condition_variable>
#include <mutex>
#include <chrono>
#ifdef _WIN32
#include <atomic>
#include <algorithm>
#include <cerrno>
#ifndef NOMINMAX
#define NOMINMAX
#endif
#include "../../../lib/platform_compat.h"
#ifndef _WIN32
#include <sys/select.h>
#endif
class CallVideoWake {
std::mutex lock;
std::condition_variable changed;
uint64_t serial = 0;
std::atomic<uint64_t> serial{0};
std::atomic<unsigned> 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<std::mutex> guard(lock); return serial; }
void notify() {
{ std::lock_guard<std::mutex> 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<std::mutex> guard(lock);
changed.wait_for(guard, std::chrono::microseconds(remaining), [&] { return serial != expected; });
}
};