From 0cb388d4a7cda90425c1e04dc5580262628a7040 Mon Sep 17 00:00:00 2001 From: Ichthyostega Date: Thu, 16 Jul 2026 03:28:48 +0200 Subject: [PATCH] Prototype: implement complete Proactor buffer state-engine Turns out we need to add further states and transitions, to establish full sync-link (backpressure) in both directions... - need to introduce a third cursor since the callback from GUI-Thread can be late - consequently also need a further state transition SHOW -> WAIT to create the barrier - furthermore, need marker state STOP to halt playback after a short read (end-of-file) --- FrOSCon26/FrOSCon.26.mm | 156 +++++++++++++++++++++++++++++- FrOSCon26/examples/demo-uring.cpp | 90 ++++++++++++++++- 2 files changed, 239 insertions(+), 7 deletions(-) diff --git a/FrOSCon26/FrOSCon.26.mm b/FrOSCon26/FrOSCon.26.mm index fd48f7c..43cf312 100644 --- a/FrOSCon26/FrOSCon.26.mm +++ b/FrOSCon26/FrOSCon.26.mm @@ -334,6 +334,7 @@ + @@ -361,6 +362,49 @@ + + + + + + + + + + + +

+ IO_URING kann errno nicht sinnvoll direkt verwenden (errno ist eine globale variable) +

+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + @@ -1963,7 +2007,13 @@ - + + + + + + + @@ -2222,7 +2272,22 @@ - + + + + + + + + + + + + + + + + @@ -2231,7 +2296,92 @@ - + + + + + + + + +

+ verwende hier den zweiten Cursor: nextShow_ +

+ + + + + + + + + + + +

+ ...und zwar via Glib::Dispatcher ⟶ GUI-Thread. +

+

+ Der zugehörige Handler-Code läuft dann bei nächster Gelegenheit in der GUI-Event-Loop, und holt sich per Callback (Interface PixelFeed) die Daten; in diesem Callback muß die Read-Barrier realisiert sein, damit der GUI-Thread alle vor diesem Punkt erfolgten Datenänderungen auch tatsächlich sieht +

+ + + + + + + + + + +

+ Beachte: das darf erst hier passieren, weil wir hier den nächst niedersten erwarteten Frame anschauen; da wir immer etwas im Voraus lesen, werden wir auch ein end-of-File schon einige Ticks früher markieren, müssen aber die Schleife noch solange weiterlaufen lassen, bis die letzten erfolgreich gelesenen Frames in die Anzeige übertragen wurden +

+ + + + + + + + + + + + + + + + + +

+ brauche doch einen 3.Cursor: currShow_ +

+ + + + + + + + + + + + + + + +

+ also brauche ich die cross-Thread-Barriere erst hier +

+ + + + + + + diff --git a/FrOSCon26/examples/demo-uring.cpp b/FrOSCon26/examples/demo-uring.cpp index e9daa77..54bf4ea 100644 --- a/FrOSCon26/examples/demo-uring.cpp +++ b/FrOSCon26/examples/demo-uring.cpp @@ -46,7 +46,7 @@ class UringVideoPlayer using VideoDisplay = EglDisplay; using Buffer = VideoDisplay::Buffer; - enum BuffState { FREE, READ, SHOW }; + enum BuffState { FREE=0, READ, SHOW, WAIT, STOP }; using AtomicBuffState = std::atomic; using BufferAlloc = unique_ptr; @@ -73,6 +73,7 @@ class UringVideoPlayer size_t nextRead_{0}; size_t nextShow_{0}; + size_t currShow_{0}; BufferAlloc buffer_{}; BuffStateAlloc buffState_{}; std::atomic_bool playing_{false}; @@ -92,6 +93,8 @@ class UringVideoPlayer private: void playVideo(); void addNextReadSubmissions (io_uring&, int inputFD); + void handleCompletedRead (io_uring_cqe&); + void maybeDisplayFrame(); }; @@ -120,7 +123,7 @@ UringVideoPlayer::playVideo() // Setup periodic frame beat, firing infinitely const auto TICK_TAG = UINT64_MAX; // special marker for timer events (different from possible frame numbers) - auto timespec = asKernelTimespec (40ms); // 40ms ≈ 25fps + auto timespec = asKernelTimespec (40ms); // 25fps ≙ 40ms 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); @@ -129,7 +132,27 @@ UringVideoPlayer::playVideo() 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 (0 > res) + __FAIL ("retrieval of io_uring completion failed"); + uint64_t tag = completion->user_data; + if (tag == TICK_TAG) + { + maybeDisplayFrame(); + addNextReadSubmissions (uRing, inputFD); + } + else + handleCompletedRead (*completion); + + // mark completion as treated (Barrier) + io_uring_cqe_seen (&uRing, completion); + } close (inputFD); } @@ -160,17 +183,76 @@ UringVideoPlayer::addNextReadSubmissions (io_uring& uRing, int inputFD) } +/** 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, modulus ring size + + 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(SHOW, 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_relaxed); + auto nextState = buffState_[nextShow_ % RING_SIZ].load (std::memory_order_relaxed); + + if (nextState == SHOW and prevState == FREE) + { // can hand over this next frame to the VideoDisplay in the GUI-thread + currShow_ = nextShow_; + ++nextShow_; + // mark state to wait for display to complete; this also creates a write barrier + buffState_[currShow_ % RING_SIZ].store (WAIT, std::memory_order_release); + // Notify GUI thread via Glib::Dispatcher + dispatchFrame_(); + } + 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() { - ///////OOO threadsafe hand-over of current frame data + // retrieve buffer state from proactor-thread; this also creates a read barrier + auto frameState = player_.buffState_[player_.currShow_ % RING_SIZ].load (std::memory_order_acquire); + if (frameState == WAIT) + return & player_.buffer_[player_.currShow_ % RING_SIZ]; + else + return nullptr; + // (can happen only in boundary cases) } void UringVideoPlayer::BufferDispatcher::completedFrame() { - ///////OOO mark end of buffer access + // mark buffer as display complete for the proactor-thread; this also creates a write barrier + player_.buffState_[player_.currShow_ % RING_SIZ].store (FREE, std::memory_order_release); }