Ichthyostega
72e1838b36
- use a helper class with a series of timer variables - these can be addressed like the buffer ring »slots« - commit each individual measurement into an EMA
344 lines
11 KiB
C++
344 lines
11 KiB
C++
/*
|
||
demo-uring.cpp - async file IO with io_uring
|
||
|
||
Copyright (C)
|
||
2026, Hermann Vosseler <Ichthyostega@web.de>
|
||
|
||
This program is free software; you can redistribute it and/or modify it
|
||
under the terms of the GNU GPL version 2+ See the LICENSE file for details.
|
||
|
||
* ************************************************************************/
|
||
|
||
|
||
#include "commons.hpp"
|
||
#include "egl-display.hpp"
|
||
#include "gtk-uring-app.hpp"
|
||
|
||
#include <iostream>
|
||
#include <memory>
|
||
#include <thread>
|
||
#include <atomic>
|
||
|
||
#include <fcntl.h>
|
||
#include <liburing.h>
|
||
|
||
using std::unique_ptr;
|
||
using std::make_unique;
|
||
using std::chrono::operator ""ms;
|
||
|
||
|
||
//-----------------hard-coded-config--
|
||
const size_t RING_SIZ = 5;
|
||
const string RAW_VIDEO_INPUT_FILE = "demovideo.raw";
|
||
const uint VIDEO_WIDTH = 1200;
|
||
const uint VIDEO_HEIGHT = 675;
|
||
const size_t START_FRAME = 40;
|
||
const size_t MAX_LOOP_FRAME = 960;
|
||
|
||
auto calcSrcFrame = [](size_t runningNr){ return START_FRAME + runningNr % MAX_LOOP_FRAME; };
|
||
|
||
const uint EMA_PERIOD = 20;
|
||
//-----------------hard-coded-config--
|
||
|
||
|
||
namespace {
|
||
/** a set of timers, one for each buffer »slot«
|
||
* to observe average IO times */
|
||
OverlappingTimers<RING_SIZ, EMA_PERIOD> timings;
|
||
}
|
||
|
||
|
||
|
||
/**
|
||
* Read video frames from a raw video file and pass them to OpenGL display.
|
||
* Using a »Proactor« thread to coordinate reading via IO_URING.
|
||
* Allocates a set of frame-sized buffers (using #RING_SIZ buffers)
|
||
* and uses them cyclically, first adding new tasks to read a frame-block
|
||
* from the file, receiving completed tasks to hand-over the next frame
|
||
* driven by a frame-beat timer, that is also based on IO_URING.
|
||
* Each buffer »slot« is associated with a #BuffState flag, implemented
|
||
* as `std::atomic` to also serve for cross-thread visibility.
|
||
* @note UringVideoPlayer must be constructed in the GTK UI thread,
|
||
* yet will immediately launch the »Proactor« thread from the ctor.
|
||
* @remarks
|
||
* - the source video and layout is hard wired
|
||
* - playback will either be looped (if the loop fits into the source video)
|
||
* - otherwise, when encountering end-of-file, or when the GTK window is closed,
|
||
* the »Proactor« thread will terminate.
|
||
* - frames will be submitted to display at the regular timer beats,
|
||
* whenever they are ready; frames delayed by IO will slip to the next beat.
|
||
*/
|
||
class UringVideoPlayer
|
||
: NonCopyable
|
||
{
|
||
using VideoDisplay = EglDisplay<uint8_t, VIDEO_WIDTH,VIDEO_HEIGHT>;
|
||
using Buffer = VideoDisplay::Buffer;
|
||
|
||
enum BuffState { FREE, READ, FEED, SHOW, STOP };
|
||
using AtomicBuffState = std::atomic<BuffState>;
|
||
|
||
using BufferAlloc = unique_ptr<Buffer[]>;
|
||
using BuffStateAlloc = unique_ptr<AtomicBuffState[]>;
|
||
|
||
/**
|
||
* Connector to the VideoDisplay for dispatching video data.
|
||
* @note inherits from Glib::Dispatcher -> signal frame readiness
|
||
* @remark the callback functions allow the VideoDisplay to interact
|
||
* with the buffer data and state in a threadsafe manner
|
||
*/
|
||
class BufferDispatcher
|
||
: public VideoDisplay::PixelFeed
|
||
{
|
||
Buffer* retrieveFrame() override;
|
||
void completedFrame() override;
|
||
|
||
UringVideoPlayer& player_;
|
||
|
||
public:
|
||
BufferDispatcher (UringVideoPlayer& player)
|
||
: player_{player}
|
||
{ }
|
||
|
||
private:
|
||
Buffer& getCurrentFrame();
|
||
AtomicBuffState& getCurrentFrameState();
|
||
};
|
||
|
||
size_t nextRead_{0};
|
||
size_t nextShow_{0};
|
||
size_t currShow_{0};
|
||
BufferAlloc buffer_{};
|
||
BuffStateAlloc buffState_{};
|
||
std::atomic_bool playing_{false};
|
||
BufferDispatcher dispatchFrame_;
|
||
VideoDisplay videoDisplayer_;
|
||
std::thread proactorThread_;
|
||
|
||
public:
|
||
UringVideoPlayer (Gtk::Window& appWindow)
|
||
: dispatchFrame_{*this}
|
||
, videoDisplayer_{appWindow, dispatchFrame_}
|
||
, proactorThread_{[this]{ playVideo(); }}
|
||
{ }
|
||
|
||
void terminate();
|
||
|
||
private:
|
||
void playVideo();
|
||
void addNextReadSubmissions (io_uring&, int inputFD);
|
||
void handleCompletedRead (io_uring_cqe&);
|
||
void maybeDisplayFrame();
|
||
};
|
||
|
||
|
||
/**
|
||
* Logic of the »Proactor Thread«:
|
||
* - maintain the working buffers
|
||
* - open the raw video file to retrieve data
|
||
* - establish an IO_URING connection for reading
|
||
* - periodic processing to fill and hand-off buffers
|
||
*/
|
||
void
|
||
UringVideoPlayer::playVideo()
|
||
{
|
||
/* =============== initial-setup =============== */
|
||
int inputFD = open (RAW_VIDEO_INPUT_FILE.c_str(), O_RDONLY);
|
||
if (inputFD < 0)
|
||
__FAIL ("open file \""+RAW_VIDEO_INPUT_FILE+"\"");
|
||
|
||
// Allocate several data buffers (used as ring)
|
||
buffer_ = make_unique<Buffer[]> (RING_SIZ);
|
||
buffState_ = make_unique<AtomicBuffState[]> (RING_SIZ);
|
||
|
||
// initialise uRing Kernel interface
|
||
io_uring uRing;
|
||
io_uring_queue_init (RING_SIZ+1, &uRing, /*flags*/ 0 ); // Note: one additional »slot« for the timer event(s)
|
||
|
||
// Setup periodic frame beat, firing infinitely
|
||
const auto TICK_TAG = UINT64_MAX; // special marker for timer events (differs from possible frame numbers)
|
||
auto timespec = asKernelTimespec (20ms); // 50fps ≙ 20ms per frame
|
||
io_uring_sqe* timerSubmission{io_uring_get_sqe (&uRing)};
|
||
io_uring_prep_timeout (timerSubmission, ×pec, /*no count limit*/ 0, IORING_TIMEOUT_MULTISHOT);
|
||
io_uring_sqe_set_data64(timerSubmission, TICK_TAG);
|
||
|
||
addNextReadSubmissions (uRing, inputFD);
|
||
io_uring_submit (&uRing);
|
||
|
||
/* =============== proactor-loop =============== */
|
||
playing_ = true;
|
||
while (playing_.load (std::memory_order_acquire))
|
||
{
|
||
io_uring_cqe* completion;
|
||
// blocking wait for the next completion entry
|
||
int res = io_uring_wait_cqe (&uRing, &completion);
|
||
if (res < 0)
|
||
__FAIL ("retrieval of io_uring completion failed");
|
||
|
||
uint64_t tag = completion->user_data;
|
||
if (tag == TICK_TAG)
|
||
{
|
||
maybeDisplayFrame();
|
||
addNextReadSubmissions (uRing, inputFD);
|
||
io_uring_submit (&uRing);
|
||
}
|
||
else
|
||
handleCompletedRead (*completion);
|
||
|
||
// mark completion as treated (Barrier)
|
||
io_uring_cqe_seen (&uRing, completion);
|
||
}
|
||
close (inputFD);
|
||
}
|
||
|
||
|
||
/**
|
||
* Instruct the next read operations,
|
||
* using all free buffer slots ahead.
|
||
*/
|
||
void
|
||
UringVideoPlayer::addNextReadSubmissions (io_uring& uRing, int inputFD)
|
||
{
|
||
assert (0 < inputFD);
|
||
const size_t BUFFSIZ = sizeof(Buffer);
|
||
|
||
while (FREE == buffState_[nextRead_ % RING_SIZ])
|
||
{
|
||
size_t srcOffset = calcSrcFrame (nextRead_);
|
||
io_uring_sqe* readSubmission = io_uring_get_sqe (&uRing);
|
||
io_uring_prep_read (readSubmission
|
||
,inputFD
|
||
,& buffer_[nextRead_ % RING_SIZ]
|
||
,BUFFSIZ
|
||
,BUFFSIZ * srcOffset
|
||
);
|
||
io_uring_sqe_set_data64 (readSubmission, nextRead_);
|
||
buffState_[nextRead_ % RING_SIZ].store (READ, std::memory_order_relaxed);
|
||
timings.markStart (nextRead_);
|
||
++nextRead_;
|
||
}
|
||
}
|
||
|
||
|
||
/** transition state of the buffer slot due to reported result */
|
||
void
|
||
UringVideoPlayer::handleCompletedRead (io_uring_cqe& completion)
|
||
{
|
||
if (0 > completion.res)
|
||
__FAIL ("async read of frame from video file failed");
|
||
|
||
size_t frameNr = completion.user_data; // Note: completion was marked with the frameNr as »user data«
|
||
auto& state = buffState_[frameNr % RING_SIZ]; // Can thus pick buffer slot based on frame number, cycling through the ring
|
||
|
||
timings.markStop(frameNr);
|
||
|
||
if (completion.res != sizeof(Buffer))
|
||
state.store(STOP, std::memory_order_relaxed); // Insufficient read, possibly end-of-file => signal stop of playback (to ourselves)
|
||
else
|
||
state.store(FEED, std::memory_order_relaxed); // Success; frame data is in the buffer => mark ready for use at the next tick event
|
||
}
|
||
|
||
|
||
/**
|
||
* At each timer tick: see if the next frame can be displayed.
|
||
* @remark determine the buffer slot where the frame data was placed,
|
||
* based on the cursor for the next frame to display, wrapping
|
||
* through the ring of buffers. Depending on the state marked
|
||
* for this buffer, the frame can be displayed if the data
|
||
* was already loaded, or otherwise the frame will slip
|
||
* to the next regular time tick.
|
||
*/
|
||
void
|
||
UringVideoPlayer::maybeDisplayFrame()
|
||
{
|
||
auto prevState = buffState_[currShow_ % RING_SIZ].load (std::memory_order_acquire);
|
||
auto nextState = buffState_[nextShow_ % RING_SIZ].load (std::memory_order_relaxed);
|
||
|
||
if (nextState == FEED and prevState != SHOW)
|
||
{ // can hand over this next frame to the VideoDisplay in the GUI-thread
|
||
currShow_ = nextShow_;
|
||
++nextShow_;
|
||
// mark state to wait for display completion; this also creates a write barrier
|
||
buffState_[currShow_ % RING_SIZ].store (SHOW, std::memory_order_release);
|
||
// Notify GUI thread via Glib::Dispatcher
|
||
dispatchFrame_();
|
||
|
||
if (0 == currShow_ % EMA_PERIOD)
|
||
std::cout << "Frame #"<<currShow_<<" IO ∅ "<<timings.expMA()<<"µs"<< std::endl;
|
||
}
|
||
else
|
||
if (nextState == STOP)
|
||
{
|
||
playing_ = false;
|
||
std::cout << "Playback stopped due to short read" << std::endl;
|
||
}
|
||
else
|
||
std::cout << "WARNING: Frame slipped" << std::endl;
|
||
}
|
||
|
||
|
||
UringVideoPlayer::Buffer*
|
||
UringVideoPlayer::BufferDispatcher::retrieveFrame()
|
||
{
|
||
// retrieve buffer state from proactor-thread; this also creates a read barrier
|
||
auto frameState = getCurrentFrameState().load (std::memory_order_acquire);
|
||
if (frameState == SHOW)
|
||
return & getCurrentFrame();
|
||
else
|
||
return nullptr;
|
||
// (can happen only in boundary cases)
|
||
}
|
||
|
||
|
||
void
|
||
UringVideoPlayer::BufferDispatcher::completedFrame()
|
||
{
|
||
// mark buffer as display complete for the proactor-thread; this also creates a write barrier
|
||
getCurrentFrameState().store (FREE, std::memory_order_release);
|
||
}
|
||
|
||
|
||
UringVideoPlayer::Buffer&
|
||
UringVideoPlayer::BufferDispatcher::getCurrentFrame()
|
||
{
|
||
auto currFrame = player_.currShow_;
|
||
return player_.buffer_[currFrame % RING_SIZ];
|
||
}
|
||
|
||
UringVideoPlayer::AtomicBuffState&
|
||
UringVideoPlayer::BufferDispatcher::getCurrentFrameState()
|
||
{
|
||
auto currFrame = player_.currShow_;
|
||
return player_.buffState_[currFrame % RING_SIZ];
|
||
}
|
||
|
||
|
||
void
|
||
UringVideoPlayer::terminate()
|
||
{
|
||
playing_.store (false, std::memory_order_release);
|
||
proactorThread_.join();
|
||
videoDisplayer_.cleanUp();
|
||
}
|
||
|
||
|
||
|
||
unique_ptr<UringVideoPlayer>
|
||
launchPlayer (Gtk::Window& appWindow)
|
||
{ // force minimum size to fit the player...
|
||
appWindow.set_size_request (VIDEO_WIDTH, VIDEO_HEIGHT);
|
||
appWindow.set_title ("powered by io_uring");
|
||
appWindow.set_resizable (false);
|
||
|
||
return std::make_unique<UringVideoPlayer> (appWindow);
|
||
}
|
||
|
||
|
||
|
||
int
|
||
main (int, const char*[])
|
||
{
|
||
return GtkUringApp<UringVideoPlayer>{"io-uring-demo"}
|
||
.onClick (launchPlayer)
|
||
.onClose ([](auto& player){ player.terminate();})
|
||
.run();
|
||
}
|