wegen/FrOSCon26/examples/demo-uring.cpp

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