Ichthyostega
835f468e1b
Add failure check when opening next URing slot for a submission; this ''can'' fail when the ring is fully populated -- which in theory should never happen in our setup, yet can always, in practice...
335 lines
11 KiB
C++
335 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.
|
||
* The lifecycle of each buffer »slot« is tracked by a #BuffState flag,
|
||
* used solely within the »Proactor« thread. A single [atomic pointer](\ref toDisplay_)
|
||
* is used as baton to hand the current frame to the GUI and to signal back
|
||
* when the display was published, thereby also establishing cross-thread
|
||
* visibility barriers for the buffer contents.
|
||
* @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 BufferAlloc = unique_ptr<Buffer[]>;
|
||
using BuffStateAlloc = unique_ptr<BuffState[]>;
|
||
|
||
/**
|
||
* 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}
|
||
{ }
|
||
};
|
||
|
||
size_t nextRead_{0};
|
||
size_t nextShow_{0};
|
||
BufferAlloc buffer_{};
|
||
BuffStateAlloc buffState_{};
|
||
std::atomic<Buffer*> toDisplay_{nullptr};
|
||
std::atomic_bool playing_{true};
|
||
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<BuffState[]> (RING_SIZ);
|
||
|
||
// initialise uRing Kernel interface
|
||
io_uring uRing;
|
||
io_uring_queue_init (RING_SIZ+1, &uRing, /*flags*/ 0 ); // Note: allocate 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 =============== */
|
||
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);
|
||
io_uring_queue_exit (&uRing);
|
||
}
|
||
|
||
|
||
/**
|
||
* 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);
|
||
if (nullptr == readSubmission)
|
||
__FAIL ("URing slot allocation failed");
|
||
|
||
io_uring_prep_read (readSubmission
|
||
,inputFD
|
||
,& buffer_[nextRead_ % RING_SIZ]
|
||
,BUFFSIZ
|
||
,BUFFSIZ * srcOffset
|
||
);
|
||
io_uring_sqe_set_data64 (readSubmission, nextRead_);
|
||
buffState_[nextRead_ % RING_SIZ] = READ;
|
||
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 = STOP; // Insufficient read, possibly end-of-file => signal stop of playback (to ourselves)
|
||
else
|
||
state = FEED; // 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 the cursor #nextShow_ walks the ring of buffers. It rests on the slot
|
||
* currently lent to the GUI-thread (marked #SHOW) until the latter did signal
|
||
* completion by clearing the atomic #toDisplay_. Only then is the slot reclaimed
|
||
* (#FREE) and the cursor advanced. The next slot reached thusly is displayed if
|
||
* its data was already loaded (#FEED); otherwise the frame slips forward to
|
||
* the next regular time tick.
|
||
* Thus #toDisplay_ also provides the back-pressure: a new frame is published
|
||
* only once the previous one was displayed and released.
|
||
*/
|
||
void
|
||
UringVideoPlayer::maybeDisplayFrame()
|
||
{
|
||
// check and possibly reclaim previously displayed buffer
|
||
if (SHOW == buffState_[nextShow_ % RING_SIZ] and
|
||
nullptr == toDisplay_.load (std::memory_order_acquire)) // read barrier
|
||
{ // GUI thread did mark completion
|
||
buffState_[nextShow_ % RING_SIZ] = FREE;
|
||
++nextShow_;
|
||
}
|
||
|
||
switch (buffState_[nextShow_ % RING_SIZ])
|
||
{
|
||
case FEED:
|
||
{ // hand over this frame to the VideoDisplay in the GUI-thread...
|
||
buffState_[nextShow_ % RING_SIZ] = SHOW;
|
||
Buffer& frame{buffer_[nextShow_ % RING_SIZ]};
|
||
toDisplay_.store (& frame, std::memory_order_release); // write barrier
|
||
dispatchFrame_(); // Notify GUI thread via Glib::Dispatcher
|
||
|
||
if (0 == nextShow_ % EMA_PERIOD)
|
||
std::cout << "Frame #"<<nextShow_<<" IO ∅ "<<timings.expMA()<<"µs"<< std::endl;
|
||
}
|
||
break;
|
||
|
||
case STOP:
|
||
playing_ = false;
|
||
std::cout << "Playback stopped due to short read" << std::endl;
|
||
break;
|
||
|
||
default: // READ (IO still pending) or SHOW (GUI not yet caught up)
|
||
std::cout << "WARNING: Frame slipped" << std::endl;
|
||
break;
|
||
}
|
||
}
|
||
|
||
|
||
/** pick up the buffer published by the proactor-thread */
|
||
UringVideoPlayer::Buffer*
|
||
UringVideoPlayer::BufferDispatcher::retrieveFrame()
|
||
{
|
||
return player_.toDisplay_.load (std::memory_order_acquire);
|
||
} // acquire-load also creates a read barrier
|
||
|
||
|
||
/** signal completion of display activity to the proactor-thread */
|
||
void
|
||
UringVideoPlayer::BufferDispatcher::completedFrame()
|
||
{
|
||
player_.toDisplay_.store (nullptr, std::memory_order_release);
|
||
} // release-store also creates a write barrier
|
||
|
||
|
||
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();
|
||
}
|