/* 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(); }