framestream2 (with rigtorp/SPSCQueue)

This commit is contained in:
Green Sky 2024-04-25 14:13:51 +02:00
parent d52a1a44f8
commit f46d0a713d
No known key found for this signature in database
9 changed files with 632 additions and 3 deletions

View File

@ -76,9 +76,14 @@ add_executable(tomato
./chat_gui4.cpp
./content/content.hpp
./content/stream_reader.hpp
./content/stream_reader_sdl_audio.hpp
./content/stream_reader_sdl_audio.cpp
#./content/stream_reader.hpp
#./content/stream_reader_sdl_audio.hpp
#./content/stream_reader_sdl_audio.cpp
#./content/stream_reader_sdl_video.hpp
#./content/stream_reader_sdl_video.cpp
./content/frame_stream2.hpp
./content/sdl_video_frame_stream2.hpp
./content/sdl_video_frame_stream2.cpp
)
target_compile_features(tomato PUBLIC cxx_std_17)

237
src/content/SPSCQueue.h Normal file
View File

@ -0,0 +1,237 @@
/*
Copyright (c) 2020 Erik Rigtorp <erik@rigtorp.se>
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/
#pragma once
#include <atomic>
#include <cassert>
#include <cstddef>
#include <memory> // std::allocator
#include <new> // std::hardware_destructive_interference_size
#include <stdexcept>
#include <type_traits> // std::enable_if, std::is_*_constructible
#ifdef __has_cpp_attribute
#if __has_cpp_attribute(nodiscard)
#define RIGTORP_NODISCARD [[nodiscard]]
#endif
#endif
#ifndef RIGTORP_NODISCARD
#define RIGTORP_NODISCARD
#endif
namespace rigtorp {
template <typename T, typename Allocator = std::allocator<T>> class SPSCQueue {
#if defined(__cpp_if_constexpr) && defined(__cpp_lib_void_t)
template <typename Alloc2, typename = void>
struct has_allocate_at_least : std::false_type {};
template <typename Alloc2>
struct has_allocate_at_least<
Alloc2, std::void_t<typename Alloc2::value_type,
decltype(std::declval<Alloc2 &>().allocate_at_least(
size_t{}))>> : std::true_type {};
#endif
public:
explicit SPSCQueue(const size_t capacity,
const Allocator &allocator = Allocator())
: capacity_(capacity), allocator_(allocator) {
// The queue needs at least one element
if (capacity_ < 1) {
capacity_ = 1;
}
capacity_++; // Needs one slack element
// Prevent overflowing size_t
if (capacity_ > SIZE_MAX - 2 * kPadding) {
capacity_ = SIZE_MAX - 2 * kPadding;
}
#if defined(__cpp_if_constexpr) && defined(__cpp_lib_void_t)
if constexpr (has_allocate_at_least<Allocator>::value) {
auto res = allocator_.allocate_at_least(capacity_ + 2 * kPadding);
slots_ = res.ptr;
capacity_ = res.count - 2 * kPadding;
} else {
slots_ = std::allocator_traits<Allocator>::allocate(
allocator_, capacity_ + 2 * kPadding);
}
#else
slots_ = std::allocator_traits<Allocator>::allocate(
allocator_, capacity_ + 2 * kPadding);
#endif
static_assert(alignof(SPSCQueue<T>) == kCacheLineSize, "");
static_assert(sizeof(SPSCQueue<T>) >= 3 * kCacheLineSize, "");
assert(reinterpret_cast<char *>(&readIdx_) -
reinterpret_cast<char *>(&writeIdx_) >=
static_cast<std::ptrdiff_t>(kCacheLineSize));
}
~SPSCQueue() {
while (front()) {
pop();
}
std::allocator_traits<Allocator>::deallocate(allocator_, slots_,
capacity_ + 2 * kPadding);
}
// non-copyable and non-movable
SPSCQueue(const SPSCQueue &) = delete;
SPSCQueue &operator=(const SPSCQueue &) = delete;
template <typename... Args>
void emplace(Args &&...args) noexcept(
std::is_nothrow_constructible<T, Args &&...>::value) {
static_assert(std::is_constructible<T, Args &&...>::value,
"T must be constructible with Args&&...");
auto const writeIdx = writeIdx_.load(std::memory_order_relaxed);
auto nextWriteIdx = writeIdx + 1;
if (nextWriteIdx == capacity_) {
nextWriteIdx = 0;
}
while (nextWriteIdx == readIdxCache_) {
readIdxCache_ = readIdx_.load(std::memory_order_acquire);
}
new (&slots_[writeIdx + kPadding]) T(std::forward<Args>(args)...);
writeIdx_.store(nextWriteIdx, std::memory_order_release);
}
template <typename... Args>
RIGTORP_NODISCARD bool try_emplace(Args &&...args) noexcept(
std::is_nothrow_constructible<T, Args &&...>::value) {
static_assert(std::is_constructible<T, Args &&...>::value,
"T must be constructible with Args&&...");
auto const writeIdx = writeIdx_.load(std::memory_order_relaxed);
auto nextWriteIdx = writeIdx + 1;
if (nextWriteIdx == capacity_) {
nextWriteIdx = 0;
}
if (nextWriteIdx == readIdxCache_) {
readIdxCache_ = readIdx_.load(std::memory_order_acquire);
if (nextWriteIdx == readIdxCache_) {
return false;
}
}
new (&slots_[writeIdx + kPadding]) T(std::forward<Args>(args)...);
writeIdx_.store(nextWriteIdx, std::memory_order_release);
return true;
}
void push(const T &v) noexcept(std::is_nothrow_copy_constructible<T>::value) {
static_assert(std::is_copy_constructible<T>::value,
"T must be copy constructible");
emplace(v);
}
template <typename P, typename = typename std::enable_if<
std::is_constructible<T, P &&>::value>::type>
void push(P &&v) noexcept(std::is_nothrow_constructible<T, P &&>::value) {
emplace(std::forward<P>(v));
}
RIGTORP_NODISCARD bool
try_push(const T &v) noexcept(std::is_nothrow_copy_constructible<T>::value) {
static_assert(std::is_copy_constructible<T>::value,
"T must be copy constructible");
return try_emplace(v);
}
template <typename P, typename = typename std::enable_if<
std::is_constructible<T, P &&>::value>::type>
RIGTORP_NODISCARD bool
try_push(P &&v) noexcept(std::is_nothrow_constructible<T, P &&>::value) {
return try_emplace(std::forward<P>(v));
}
RIGTORP_NODISCARD T *front() noexcept {
auto const readIdx = readIdx_.load(std::memory_order_relaxed);
if (readIdx == writeIdxCache_) {
writeIdxCache_ = writeIdx_.load(std::memory_order_acquire);
if (writeIdxCache_ == readIdx) {
return nullptr;
}
}
return &slots_[readIdx + kPadding];
}
void pop() noexcept {
static_assert(std::is_nothrow_destructible<T>::value,
"T must be nothrow destructible");
auto const readIdx = readIdx_.load(std::memory_order_relaxed);
assert(writeIdx_.load(std::memory_order_acquire) != readIdx &&
"Can only call pop() after front() has returned a non-nullptr");
slots_[readIdx + kPadding].~T();
auto nextReadIdx = readIdx + 1;
if (nextReadIdx == capacity_) {
nextReadIdx = 0;
}
readIdx_.store(nextReadIdx, std::memory_order_release);
}
RIGTORP_NODISCARD size_t size() const noexcept {
std::ptrdiff_t diff = writeIdx_.load(std::memory_order_acquire) -
readIdx_.load(std::memory_order_acquire);
if (diff < 0) {
diff += capacity_;
}
return static_cast<size_t>(diff);
}
RIGTORP_NODISCARD bool empty() const noexcept {
return writeIdx_.load(std::memory_order_acquire) ==
readIdx_.load(std::memory_order_acquire);
}
RIGTORP_NODISCARD size_t capacity() const noexcept { return capacity_ - 1; }
private:
#ifdef __cpp_lib_hardware_interference_size
static constexpr size_t kCacheLineSize =
std::hardware_destructive_interference_size;
#else
static constexpr size_t kCacheLineSize = 64;
#endif
// Padding to avoid false sharing between slots_ and adjacent allocations
static constexpr size_t kPadding = (kCacheLineSize - 1) / sizeof(T) + 1;
private:
size_t capacity_;
T *slots_;
#if defined(__has_cpp_attribute) && __has_cpp_attribute(no_unique_address)
Allocator allocator_ [[no_unique_address]];
#else
Allocator allocator_;
#endif
// Align to cache line size in order to avoid false sharing
// readIdxCache_ and writeIdxCache_ is used to reduce the amount of cache
// coherency traffic
alignas(kCacheLineSize) std::atomic<size_t> writeIdx_ = {0};
alignas(kCacheLineSize) size_t readIdxCache_ = 0;
alignas(kCacheLineSize) std::atomic<size_t> readIdx_ = {0};
alignas(kCacheLineSize) size_t writeIdxCache_ = 0;
};
} // namespace rigtorp

View File

@ -0,0 +1,99 @@
#pragma once
#include <cstdint>
#include <memory>
#include <optional>
#include <vector>
#include <mutex>
#include "./SPSCQueue.h"
// Frames ofen consist of:
// - seq id // incremental sequential id, gaps in ids can be used to detect loss
// - data // the frame data
// eg:
//struct ExampleFrame {
//int64_t seq_id {0};
//std::vector<uint8_t> data;
//};
template<typename FrameType>
struct FrameStream2Reader {
// get number of available frames
virtual int32_t getSize(void) = 0;
// get next frame
// TODO: optional instead?
// data sharing? -> no, data is copied for each fsr, if concurency supported
virtual std::optional<FrameType> getNext(void) = 0;
};
// needs count frames queue size
// having ~1-2sec buffer size is often sufficent
template<typename FrameType>
struct QueuedFrameStream2Reader : public FrameStream2Reader<FrameType> {
using frame_type = FrameType;
rigtorp::SPSCQueue<FrameType> _queue;
explicit QueuedFrameStream2Reader(size_t queue_size) : _queue(queue_size) {}
[[nodiscard]] int32_t getSize(void) override {
return _queue.size();
}
[[nodiscard]] std::optional<FrameType> getNext(void) override {
auto* ret_ptr = _queue.front();
if (ret_ptr == nullptr) {
return std::nullopt;
}
// move away
FrameType ret = std::move(*ret_ptr);
_queue.pop();
return ret;
}
};
template<typename FrameType>
struct MultiplexedQueuedFrameStream2Writer {
using ReaderType = QueuedFrameStream2Reader<FrameType>;
// TODO: expose
const size_t _queue_size {10};
// pointer stability
std::vector<std::unique_ptr<ReaderType>> _readers;
std::mutex _readers_lock; // accessing the _readers array needs to be exclusive
// a simple lock here is ok, since this tends to be a rare operation,
// except for the push, which is always on the same thread
ReaderType* aquireReader(void) {
std::lock_guard lg{_readers_lock};
return _readers.emplace_back(std::make_unique<ReaderType>(_queue_size)).get();
}
void releaseReader(ReaderType* reader) {
std::lock_guard lg{_readers_lock};
for (auto it = _readers.begin(); it != _readers.end(); it++) {
if (it->get() == reader) {
_readers.erase(it);
break;
}
}
}
// returns true if there are readers
bool pushValue(const FrameType& value) {
std::lock_guard lg{_readers_lock};
bool have_readers{false};
for (auto& it : _readers) {
[[maybe_unused]] auto _ = it->_queue.try_emplace(value);
have_readers = true; // even if queue full, we still continue believing in them
}
return have_readers;
}
};

View File

@ -0,0 +1,81 @@
#include "./sdl_video_frame_stream2.hpp"
#include <chrono>
#include <cstdint>
#include <iostream>
#include <memory>
#include <thread>
SDLVideoCameraContent::SDLVideoCameraContent(void) {
int devcount = 0;
SDL_CameraDeviceID *devices = SDL_GetCameraDevices(&devcount);
if (devices == nullptr || devcount < 1) {
throw int(1); // TODO: proper exception?
}
std::cout << "### found cameras:\n";
for (int i = 0; i < devcount; i++) {
const SDL_CameraDeviceID device = devices[i];
char *name = SDL_GetCameraDeviceName(device);
std::cout << " - Camera #" << i << ": " << name << "\n";
SDL_free(name);
}
_camera = {
SDL_OpenCameraDevice(devices[0], nullptr),
&SDL_CloseCamera
};
if (!static_cast<bool>(_camera)) {
throw int(2);
}
SDL_CameraSpec spec;
float interval {0.1f};
if (SDL_GetCameraFormat(_camera.get(), &spec) < 0) {
// meh
} else {
// interval
interval = float(spec.interval_numerator)/float(spec.interval_denominator);
std::cout << "camera interval: " << interval*1000 << "ms\n";
}
_thread = std::thread([this, interval](void) {
while (!_thread_should_quit) {
Uint64 timestampNS = 0;
SDL_Surface* sdl_frame_next = SDL_AcquireCameraFrame(_camera.get(), &timestampNS);
// no new frame yet, or error
if (sdl_frame_next == nullptr) {
// only sleep 1/10, we expected a frame
std::this_thread::sleep_for(std::chrono::milliseconds(int64_t(interval*1000 / 10)));
continue;
}
SDLVideoFrame new_frame {
timestampNS,
sdl_frame_next
};
bool someone_listening = pushValue(new_frame);
SDL_ReleaseCameraFrame(_camera.get(), sdl_frame_next);
if (someone_listening) {
// TODO: maybe double the freq?
std::this_thread::sleep_for(std::chrono::milliseconds(int64_t(interval*1000)));
} else {
// we just sleep 4x as long, bc no one is listening
std::this_thread::sleep_for(std::chrono::milliseconds(int64_t(interval*1000*4)));
}
}
});
}
SDLVideoCameraContent::~SDLVideoCameraContent(void) {
_thread_should_quit = true;
_thread.join();
// TODO: what to do if readers are still present?
}

View File

@ -0,0 +1,66 @@
#pragma once
#include "./frame_stream2.hpp"
#include "SDL_surface.h"
#include <SDL3/SDL.h>
#include <cstdint>
#include <thread>
inline void nopSurfaceDestructor(SDL_Surface*) {}
// this is very sdl specific
struct SDLVideoFrame {
// TODO: sequence numbering?
uint64_t timestampNS {0};
std::unique_ptr<SDL_Surface, decltype(&SDL_DestroySurface)> surface {nullptr, &SDL_DestroySurface};
// special non-owning constructor?
SDLVideoFrame(
uint64_t ts,
SDL_Surface* surf
) {
timestampNS = ts;
surface = {surf, &nopSurfaceDestructor};
}
// copy
SDLVideoFrame(const SDLVideoFrame& other) {
timestampNS = other.timestampNS;
if (static_cast<bool>(other.surface)) {
surface = {
SDL_CreateSurface(
other.surface->w,
other.surface->h,
SDL_PIXELFORMAT_RGBA8888 // meh
),
&SDL_DestroySurface
};
SDL_BlitSurface(other.surface.get(), nullptr, surface.get(), nullptr);
}
}
SDLVideoFrame& operator=(const SDLVideoFrame& other) = delete;
};
using SDLVideoFrameStream2Writer = MultiplexedQueuedFrameStream2Writer<SDLVideoFrame>;
using SDLVideoFrameStream2Reader = SDLVideoFrameStream2Writer::ReaderType;
struct SDLVideoCameraContent : protected SDLVideoFrameStream2Writer {
// meh, empty default
std::unique_ptr<SDL_Camera, decltype(&SDL_CloseCamera)> _camera {nullptr, &SDL_CloseCamera};
std::atomic<bool> _thread_should_quit {false};
std::thread _thread;
// construct source and start thread
// TODO: optimize so the thread is not always running
SDLVideoCameraContent(void);
// stops the thread and closes the camera
~SDLVideoCameraContent(void);
// make only some of writer public
using SDLVideoFrameStream2Writer::aquireReader;
using SDLVideoFrameStream2Writer::releaseReader;
};

View File

@ -0,0 +1,84 @@
#include "./stream_reader_sdl_video.hpp"
#include <iostream>
SDLVideoFrameStreamReader::SDLVideoFrameStreamReader() : _camera{nullptr, &SDL_CloseCamera}, _surface{nullptr, &SDL_DestroySurface} {
// enumerate
int devcount = 0;
SDL_CameraDeviceID *devices = SDL_GetCameraDevices(&devcount);
if (devices == nullptr || devcount < 1) {
throw int(1); // TODO: proper exception?
}
std::cout << "### found cameras:\n";
for (int i = 0; i < devcount; i++) {
const SDL_CameraDeviceID device = devices[i];
char *name = SDL_GetCameraDeviceName(device);
std::cout << " - Camera #" << i << ": " << name << "\n";
SDL_free(name);
}
_camera = {
SDL_OpenCameraDevice(devices[0], nullptr),
&SDL_CloseCamera
};
if (!static_cast<bool>(_camera)) {
throw int(2);
}
SDL_CameraSpec spec;
if (SDL_GetCameraFormat(_camera.get(), &spec) < 0) {
// meh
} else {
// interval
float interval = float(spec.interval_numerator)/float(spec.interval_denominator);
std::cout << "camera interval: " << interval*1000 << "ms\n";
}
}
SDLVideoFrameStreamReader::VideoFrame SDLVideoFrameStreamReader::getNextVideoFrameRGBA(void) {
if (!static_cast<bool>(_camera)) {
return {};
}
Uint64 timestampNS = 0;
SDL_Surface* frame_next = SDL_AcquireCameraFrame(_camera.get(), &timestampNS);
// no new frame yet, or error
if (frame_next == nullptr) {
//std::cout << "failed acquiring frame\n";
return {};
}
// TODO: investigate zero copy
_surface = {
SDL_ConvertSurfaceFormat(frame_next, SDL_PIXELFORMAT_RGBA8888),
&SDL_DestroySurface
};
SDL_ReleaseCameraFrame(_camera.get(), frame_next);
SDL_LockSurface(_surface.get());
return {
_surface->w,
_surface->h,
timestampNS,
{
reinterpret_cast<const uint8_t*>(_surface->pixels),
uint64_t(_surface->w*_surface->h*4) // rgba
}
};
}
int64_t SDLVideoFrameStreamReader::have(void) {
return -1;
}
ByteSpan SDLVideoFrameStreamReader::getNext(void) {
return {};
}

View File

@ -0,0 +1,34 @@
#pragma once
#include "./stream_reader.hpp"
#include <SDL3/SDL.h>
#include <cstdint>
#include <cstdio>
#include <memory>
struct SDLVideoFrameStreamReader : public RawFrameStreamReaderI {
// meh, empty default
std::unique_ptr<SDL_Camera, decltype(&SDL_CloseCamera)> _camera;
std::unique_ptr<SDL_Surface, decltype(&SDL_DestroySurface)> _surface;
SDLVideoFrameStreamReader(void);
struct VideoFrame {
int32_t width {0};
int32_t height {0};
uint64_t timestampNS {0};
ByteSpan data;
};
// data owned by StreamReader, overwritten by next call to getNext*()
VideoFrame getNextVideoFrameRGBA(void);
public: // interface
int64_t have(void) override;
ByteSpan getNext(void) override;
};

View File

@ -27,6 +27,8 @@ struct ImageLoaderI {
// only positive values are valid
ImageResult crop(int32_t c_x, int32_t c_y, int32_t c_w, int32_t c_h) const;
// TODO: scale
};
virtual ImageResult loadFromMemoryRGBA(const uint8_t* data, uint64_t data_size) = 0;
};

View File

@ -9,6 +9,7 @@
#include "./chat_gui/theme.hpp"
#include "./start_screen.hpp"
#include "./content/sdl_video_frame_stream2.hpp"
#include <memory>
#include <iostream>
@ -58,6 +59,26 @@ int main(int argc, char** argv) {
}
}
// optionally init audio and camera
if (SDL_Init(SDL_INIT_AUDIO) < 0) {
std::cerr << "SDL_Init AUDIO failed (" << SDL_GetError() << ")\n";
}
if (SDL_Init(SDL_INIT_CAMERA) < 0) {
std::cerr << "SDL_Init CAMERA failed (" << SDL_GetError() << ")\n";
} else { // HACK
SDLVideoCameraContent vcc;
auto* reader = vcc.aquireReader();
for (size_t i = 0; i < 200; i++) {
std::this_thread::sleep_for(std::chrono::milliseconds(50));
auto new_frame_opt = reader->getNext();
if (new_frame_opt.has_value()) {
std::cout << "video frame was " << new_frame_opt.value().surface->w << "x" << new_frame_opt.value().surface->h << " " << new_frame_opt.value().timestampNS << "ns\n";
}
}
vcc.releaseReader(reader);
}
std::cout << "after sdl video stuffery\n";
IMGUI_CHECKVERSION();
ImGui::CreateContext();