Decrease render event processing latency [msema]

Sources/CSFBAudioEngine/Player/AudioPlayer.h

@@ -150,12 +150,15 @@ class AudioPlayer final {
     /// Thread used for decoding
     std::jthread decodingThread_;
     /// Semaphore used for communication with the decoding thread
-    dsema::Semaphore decodingSemaphore_{0};
+    msema::Semaphore decodingSemaphore_{0};
 
     /// Thread used for event processing
     std::jthread eventThread_;
     /// Semaphore used for communication with the event processing thread
-    dsema::Semaphore eventSemaphore_{0};
+    msema::Semaphore eventSemaphore_{0};
+    /// Whether `eventSemaphore_` has been signaled since the event processing thread last began draining events
+    std::atomic_bool eventSignaled_{false};
+    static_assert(std::atomic_bool::is_always_lock_free, "Lock-free std::atomic_bool required");
 
     /// Message queue communicating events to the event processing thread
     mpsc::MessageQueue<256, 48> events_;
@@ -454,17 +457,25 @@ class AudioPlayer final {
     /// Returns the appropriate decoding semaphore timeout
     std::chrono::nanoseconds decodingTimeout(DecoderState *_Nullable decoderState) const noexcept;
 
+#ifndef NDEBUG
+    /// Returns `true` if the decoding thread has a decoder to process and should not be idle
+    /// - note: Used only to detect missed signals when a wait with no decoder to process times out
+    bool shouldBeProcessingDecoders() const noexcept;
+#endif /* !NDEBUG */
+
     // MARK: - Rendering
 
     /// Render block implementation
     OSStatus render(BOOL &isSilence, const AudioTimeStamp &timestamp, AVAudioFrameCount frameCount,
                     AudioBufferList &outputData) noexcept;
 
     /// Enqueues frames rendered event(s) for rendered audio
-    void enqueueFramesRenderedEvents(uint32_t framesRead, const AudioTimeStamp &timestamp) noexcept;
+    /// - returns: `true` if an event was enqueued
+    [[nodiscard]] bool enqueueFramesRenderedEvents(uint32_t framesRead, const AudioTimeStamp &timestamp) noexcept;
 
     /// Enqueues an empty frames rendered event if an empty decoded chunk descriptor is present
-    void enqueueEmptyFramesRenderedEvent(const AudioTimeStamp &timestamp) noexcept;
+    /// - returns: `true` if an event was enqueued
+    [[nodiscard]] bool enqueueEmptyFramesRenderedEvent(const AudioTimeStamp &timestamp) noexcept;
 
     /// The current rendering chunk descriptor
     std::optional<detail::RenderingChunkDescriptor> renderingChunk_{};
@@ -535,6 +546,9 @@ class AudioPlayer final {
     /// Called when the final audio frame from a decoder will render.
     void handleRenderingWillComplete(Decoder _Nonnull decoder, uint64_t hostTime, uint64_t outputGeneration) noexcept;
 
+    /// Signals the event semaphore if required
+    void signalEventSemaphore() noexcept;
+
     // MARK: - Transport Snapshots
 
     /// Publishes a playback snapshot.
@@ -660,6 +674,12 @@ inline AVAudioMixerNode *_Nonnull AudioPlayer::mainMixerNode() const noexcept {
 
 inline AVAudioOutputNode *_Nonnull AudioPlayer::outputNode() const noexcept { return engine_.outputNode; }
 
+inline void AudioPlayer::signalEventSemaphore() noexcept {
+    if (!eventSignaled_.exchange(true, std::memory_order_acq_rel)) {
+        eventSemaphore_.signal();
+    }
+}
+
 // MARK: Transport Snapshot
 
 inline bool detail::TransportSnapshot::getPlaybackPositionAndTime(SFBPlaybackPosition *position,

Sources/CSFBAudioEngine/Player/AudioPlayer.mm

@@ -29,6 +29,7 @@
 #import "SFBAudioPlayer+Internal.h"
 #import "SFBCStringForOSType.h"
 #import "host_time.hpp"
+#import "scope_exit.hpp"
 
 namespace {
 
@@ -1158,7 +1159,12 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
 
         // Wait for an event signal; ring buffer space availability is polled using the timeout
         const auto timeout = decodingTimeout(decoderState);
-        decodingSemaphore_.wait(timeout);
+        [[maybe_unused]] const auto signaled = decodingSemaphore_.wait(timeout);
+#ifndef NDEBUG
+        if (!signaled && decoderState == nullptr && shouldBeProcessingDecoders()) [[unlikely]] {
+            os_log_debug(log_, "Decoding thread woke on timeout with decoder to process: possible missed signal");
+        }
+#endif /* !NDEBUG */
     }
 
 #ifndef NDEBUG
@@ -1208,7 +1214,7 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
 
     // Signal the event thread if any decoders were canceled
     if (signal) {
-        eventSemaphore_.signal();
+        signalEventSemaphore();
     }
 
     // All requested cancelations have been processed and any stale audio is marked
@@ -1270,7 +1276,7 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
     pendingSeek_.compare_exchange_strong(pendingSeek, {}, std::memory_order_acq_rel, std::memory_order_relaxed);
 
     if (events_.enqueue(EventCommand::seekComplete, decoderState->sequenceNumber_, framePosition)) [[likely]] {
-        eventSemaphore_.signal();
+        signalEventSemaphore();
     } else {
         os_log_fault(log_, "Error writing seek complete event");
     }
@@ -1396,7 +1402,7 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
     } catch (const std::exception &e) {
         os_log_error(log_, "Error allocating decoder state for %{public}@: %{public}s", decoder, e.what());
         if (events_.enqueue(EventCommand::allocationFailure)) [[likely]] {
-            eventSemaphore_.signal();
+            signalEventSemaphore();
         } else {
             os_log_fault(log_, "Error writing allocation failure event");
         }
@@ -1576,7 +1582,7 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
 #endif /* !NDEBUG */
                 // Submit the decoding started event for the initial start only
                 if (events_.enqueue(EventCommand::decodingStarted, decoderState->sequenceNumber_)) [[likely]] {
-                    eventSemaphore_.signal();
+                    signalEventSemaphore();
                 } else {
                     os_log_fault(log_, "Error writing decoding started event");
                 }
@@ -1607,7 +1613,7 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
             // Submit the decoding complete event for the first completion only
             if (!resumed) {
                 if (events_.enqueue(EventCommand::decodingComplete, decoderState->sequenceNumber_)) [[likely]] {
-                    eventSemaphore_.signal();
+                    signalEventSemaphore();
                 } else {
                     os_log_fault(log_, "Error writing decoding complete event");
                 }
@@ -1667,7 +1673,7 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
 
     if (decoderState == nullptr) {
         // Idling or waiting on a decoder to complete rendering for a pending format change
-        return 500ms;
+        return 5s;
     }
 
     // Attempt to keep the ring buffer 75% full
@@ -1685,6 +1691,36 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
     return std::chrono::duration_cast<std::chrono::nanoseconds>(duration);
 }
 
+#ifndef NDEBUG
+bool sfb::AudioPlayer::shouldBeProcessingDecoders() const noexcept {
+    const auto flags = loadFlags();
+
+    // Cancellation was requested but not processed
+    if (bits::is_set(flags, Flags::cancelationPending)) {
+        return true;
+    }
+
+    std::scoped_lock lock{queuedDecodersMutex_, activeDecodersMutex_};
+
+    // A decoder is waiting in the queue and there is room for it to become active
+    if (!queuedDecoders_.empty() && activeDecoders_.size() < maximumActiveDecoders) {
+        return true;
+    }
+
+    // While a format change is pending the next decoder is active but waits until the previous decoder's
+    // audio has finished rendering; only the gate opening is signaled
+    if (bits::is_set(flags, Flags::formatChangePending)) {
+        return activeDecoders_.size() == 1 && scheduledRenderingCompleteNotifications_ == 0;
+    }
+
+    // An active decoder has not finished decoding
+    return std::ranges::any_of(activeDecoders_, [](const auto &state) noexcept {
+        return bits::has_none(state->loadFlags(),
+                              DecoderState::Flags::canceled | DecoderState::Flags::decodingComplete);
+    });
+}
+#endif /* !NDEBUG */
+
 // MARK: - Rendering
 
 OSStatus sfb::AudioPlayer::render(BOOL &isSilence, const AudioTimeStamp &timestamp, AVAudioFrameCount frameCount,
@@ -1698,14 +1734,24 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
         audioMetadata_.discardAll();
         renderingChunk_ = {};
         clearFlags(Flags::audioStale);
+        decodingSemaphore_.signal();
     }
 
+    // Whether this render cycle enqueued an event and should signal the event semaphore
+    auto eventEnqueued = false;
+    const auto sendSignal = util::scope_exit{[&]() noexcept {
+        if (eventEnqueued) {
+            signalEventSemaphore();
+        }
+    }};
+
     // Process pending `AVAudioEngine` pause requests
     if (bits::is_set_and_is_clear(flags, Flags::active, Flags::playing)) {
         if (const auto generation = pauseRequestGeneration_.load(std::memory_order_acquire);
             generation != processedPauseGeneration_) [[unlikely]] {
             if (events_.enqueue(EventCommand::pauseEngine, timestamp.mHostTime, generation)) [[likely]] {
                 processedPauseGeneration_ = generation;
+                eventEnqueued = true;
             } else {
                 setFlags(Flags::renderEventDropped);
             }
@@ -1727,10 +1773,14 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
 
     // Enqueue frames rendered event(s)
     if (framesRead > 0) [[likely]] {
-        enqueueFramesRenderedEvents(framesRead, timestamp);
+        if (enqueueFramesRenderedEvents(framesRead, timestamp)) [[likely]] {
+            eventEnqueued = true;
+        }
     } else {
         isSilence = YES;
-        enqueueEmptyFramesRenderedEvent(timestamp);
+        if (enqueueEmptyFramesRenderedEvent(timestamp)) {
+            eventEnqueued = true;
+        }
     }
 
     // Suppress underrun notifications while a non-gapless format change is pending; the ring buffer
@@ -1740,8 +1790,10 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
         // Throttle underrun events to prevent saturating the event message queue
         constexpr auto throttleInterval = 100 * nanosecondsPerMillisecond;
         if (host_time::toNanoseconds(timestamp.mHostTime - lastUnderrunHostTime_) >= throttleInterval) {
-            if (!events_.enqueue(EventCommand::renderBufferUnderrun, timestamp.mHostTime, framesRead, frameCount))
-                    [[unlikely]] {
+            if (events_.enqueue(EventCommand::renderBufferUnderrun, timestamp.mHostTime, framesRead, frameCount))
+                    [[likely]] {
+                eventEnqueued = true;
+            } else {
                 setFlags(Flags::renderEventDropped);
             }
             lastUnderrunHostTime_ = timestamp.mHostTime;
@@ -1751,9 +1803,10 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
     return noErr;
 }
 
-void sfb::AudioPlayer::enqueueFramesRenderedEvents(uint32_t framesRead, const AudioTimeStamp &timestamp) noexcept {
+bool sfb::AudioPlayer::enqueueFramesRenderedEvents(uint32_t framesRead, const AudioTimeStamp &timestamp) noexcept {
     assert(framesRead > 0);
 
+    auto eventEnqueued = false;
     auto framesRemaining = framesRead;
     do {
         // Read the next chunk descriptor if needed
@@ -1782,9 +1835,11 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
         const auto eventTime = hostTimeForFrameOffset(frameOffset, timestamp, sampleRate);
         const auto completionTime =
                 isFinal ? hostTimeForFrameOffset(frameOffset + chunkFramesConsumed, timestamp, sampleRate) : 0;
-        if (!events_.enqueue(EventCommand::framesRendered, chunkSequenceNumber, chunkFramesConsumed,
-                             renderingChunk_->descriptor_.playbackGeneration_, eventTime, completionTime, isFinal))
-                [[unlikely]] {
+        if (events_.enqueue(EventCommand::framesRendered, chunkSequenceNumber, chunkFramesConsumed,
+                            renderingChunk_->descriptor_.playbackGeneration_, eventTime, completionTime, isFinal))
+                [[likely]] {
+            eventEnqueued = true;
+        } else {
             setFlags(Flags::renderEventDropped);
         }
 
@@ -1797,11 +1852,15 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
             renderingChunk_.reset();
         }
     } while (framesRemaining > 0);
+
+    return eventEnqueued;
 }
 
-void sfb::AudioPlayer::enqueueEmptyFramesRenderedEvent(const AudioTimeStamp &timestamp) noexcept {
+bool sfb::AudioPlayer::enqueueEmptyFramesRenderedEvent(const AudioTimeStamp &timestamp) noexcept {
     assert(!renderingChunk_);
 
+    auto eventEnqueued = false;
+
     // Check for an empty decoded chunk descriptor, and if found enqueue an empty frames rendered event to signify
     // rendering complete
     if (detail::DecodedChunkDescriptor chunkDescriptor{};
@@ -1810,14 +1869,18 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
 
         // Rendering is complete
         const auto eventTime = hostTimeForFrameOffset(0, timestamp, audioBuffer_.format().mSampleRate);
-        if (!events_.enqueue(EventCommand::framesRendered, chunkDescriptor.sequenceNumber_, uint32_t{0},
-                             chunkDescriptor.playbackGeneration_, eventTime, eventTime, true)) [[unlikely]] {
+        if (events_.enqueue(EventCommand::framesRendered, chunkDescriptor.sequenceNumber_, uint32_t{0},
+                            chunkDescriptor.playbackGeneration_, eventTime, eventTime, true)) [[likely]] {
+            eventEnqueued = true;
+        } else {
             setFlags(Flags::renderEventDropped);
         }
 
         // Chunk processing complete
         audioMetadata_.discard();
     }
+
+    return eventEnqueued;
 }
 
 // MARK: - Event Processing
@@ -1833,6 +1896,9 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
 #endif /* !NDEBUG */
 
     while (!stoken.stop_requested()) {
+        // Clear before draining so an event enqueued during the drain is followed by a new signal
+        eventSignaled_.exchange(false, std::memory_order_acq_rel);
+
 #ifndef NDEBUG
         if (const auto occupancy =
                     static_cast<double>(events_.occupiedSlots()) / static_cast<double>(events_.slotCount);
@@ -1877,27 +1943,21 @@ Flags clearFlags(Flags flags, std::memory_order order = std::memory_order_acq_re
             }
         }
 
-        const auto prevFlags = clearFlags(Flags::renderEventDropped);
-        if (bits::is_set(prevFlags, Flags::renderEventDropped)) {
+        if (const auto prevFlags = clearFlags(Flags::renderEventDropped);
+            bits::is_set(prevFlags, Flags::renderEventDropped)) {
             os_log_fault(log_, "Missing rendering event(s): event message queue overrun");
         }
 
         using namespace std::chrono_literals;
 
-        // Poll more frequently when the render block can produce events
-        std::chrono::nanoseconds timeout = 500ms;
-        if (bits::is_set(prevFlags, Flags::playing)) {
-            std::lock_guard lock{activeDecodersMutex_};
-            if (firstActiveDecoderState() != nullptr) {
-                timeout = 7500us;
-            }
-        } else if (bits::is_set(prevFlags, Flags::active) && engine_.isRunning) {
-            // A paused player's engine keeps running until the delayed engine pause takes effect
-            timeout = 7500us;
+        // Wait for an event; the timeout is protection against a missed signal
+        [[maybe_unused]] const auto signaled = eventSemaphore_.wait(5s);
+#ifndef NDEBUG
+        if (!signaled && !events_.isEmpty()) [[unlikely]] {
+            // If an event is enqueued just as the wait times out the message will still be logged, hence "possible"
+            os_log_debug(log_, "Event processing thread woke on timeout with events pending: possible missed signal");
         }
-
-        // Decoding events will be signaled; render events are polled using the timeout
-        eventSemaphore_.wait(timeout);
+#endif /* !NDEBUG */
     }
 
 #ifndef NDEBUG

Sources/CSFBAudioEngine/Player/scope_exit.hpp

@@ -0,0 +1,35 @@
+//
+// SPDX-FileCopyrightText: 2026 Stephen F. Booth <contact@sbooth.dev>
+// SPDX-License-Identifier: MIT
+//
+// Part of https://github.com/sbooth/SFBAudioEngine
+//
+
+#pragma once
+
+#include <concepts>
+#include <type_traits>
+#include <utility>
+
+namespace util {
+
+template <typename F, typename... Args>
+concept nothrow_invocable = std::invocable<F &, Args...> && std::is_nothrow_invocable_v<F &, Args...>;
+
+/// A simple scope guard
+template <nothrow_invocable F> class scope_exit final {
+  public:
+    [[nodiscard]] explicit scope_exit(F f) noexcept(std::is_nothrow_move_constructible_v<F>) : f_{std::move(f)} {}
+    ~scope_exit() noexcept { f_(); }
+
+    scope_exit(const scope_exit &) = delete;
+    scope_exit &operator=(const scope_exit &) = delete;
+
+    scope_exit(scope_exit &&) = delete;
+    scope_exit &operator=(scope_exit &&) = delete;
+
+  private:
+    F f_;
+};
+
+} /* namespace util */