PR23 forest: Add an all-or-nothing reader retry contract

src/engine/cachingreader/cachingreader.cpp

@@ -41,6 +41,7 @@
         UserSettingsPointer config,
         mixxx::audio::ChannelCount maxSupportedChannel)
         : m_pConfig(config),
+          m_retryOnCacheMiss(false),
           // Limit the number of in-flight requests to the worker. This should
           // prevent to overload the worker when it is not able to fetch those
           // requests from the FIFO timely. Otherwise outdated requests pile up
@@ -60,6 +61,7 @@
           m_lruCachingReaderChunk(nullptr),
           m_sampleBuffer(CachingReaderChunk::kFrames * maxSupportedChannel *
                   kNumberOfCachedChunksInMemory),
+          m_retryReadBuffer(CachingReaderChunk::kFrames * maxSupportedChannel),
           m_worker(group,
                   &m_chunkReadRequestFIFO,
                   &m_readerStatusUpdateFIFO,
@@ -305,6 +307,60 @@
         bool reverse,
         CSAMPLE* buffer,
         mixxx::audio::ChannelCount channelCount) {
+    return readInternal(startSample,
+            numSamples,
+            reverse,
+            buffer,
+            channelCount,
+            m_retryOnCacheMiss);
+}
+
+CachingReader::ReadResult CachingReader::readWithRetry(SINT startSample,
+        SINT numSamples,
+        bool reverse,
+        CSAMPLE* buffer,
+        mixxx::audio::ChannelCount channelCount) {
+    VERIFY_OR_DEBUG_ASSERT(numSamples >= 0 &&
+            static_cast<size_t>(numSamples) <= m_retryReadBuffer.size()) {
+        return ReadResult::UNAVAILABLE;
+    }
+    VERIFY_OR_DEBUG_ASSERT(buffer) {
+        return ReadResult::UNAVAILABLE;
+    }
+
+    DEBUG_ASSERT(!m_retryOnCacheMiss);
+    m_retryOnCacheMiss = true;
+    const auto retryResult = readWithRetryHook(startSample,
+            numSamples,
+            reverse,
+            m_retryReadBuffer.data(),
+            channelCount);
+    m_retryOnCacheMiss = false;
+
+    if (retryResult.retryPending ||
+            retryResult.result == ReadResult::UNAVAILABLE) {
+        return ReadResult::UNAVAILABLE;
+    }
+    SampleUtil::copy(buffer, m_retryReadBuffer.data(), numSamples);
+    return retryResult.result;
+}
+
+CachingReader::RetryReadResult CachingReader::readWithRetryHook(SINT startSample,
+        SINT numSamples,
+        bool reverse,
+        CSAMPLE* buffer,
+        mixxx::audio::ChannelCount channelCount) {
+    const auto result = read(
+            startSample, numSamples, reverse, buffer, channelCount);
+    return {result, result == ReadResult::UNAVAILABLE};
+}
+
+CachingReader::ReadResult CachingReader::readInternal(SINT startSample,
+        SINT numSamples,
+        bool reverse,
+        CSAMPLE* buffer,
+        mixxx::audio::ChannelCount channelCount,
+        bool retryOnCacheMiss) {
     // Check for bad inputs
     // Refuse to read from an invalid position
     VERIFY_OR_DEBUG_ASSERT(startSample % channelCount == 0) {
@@ -332,7 +388,6 @@
                 << "buffer =" << buffer;
         return ReadResult::UNAVAILABLE;
     }
-
     // If no track is loaded, don't do anything.
     if (atomicLoadRelaxed(m_state) != STATE_TRACK_LOADED) {
         return ReadResult::UNAVAILABLE;
@@ -365,6 +420,42 @@
                     CachingReaderChunk::samples2frames(numSamples, channelCount));
     DEBUG_ASSERT(!remainingFrameIndexRange.empty());
 
+    if (retryOnCacheMiss) {
+        auto preflightFrameIndexRange =
+                intersect(remainingFrameIndexRange, m_readableFrameIndexRange);
+        if (!preflightFrameIndexRange.empty()) {
+            const SINT firstChunkIndex =
+                    CachingReaderChunk::indexForFrame(preflightFrameIndexRange.start());
+            SINT lastChunkIndex =
+                    CachingReaderChunk::indexForFrame(preflightFrameIndexRange.end() - 1);
+            for (SINT chunkIndex = firstChunkIndex;
+                    chunkIndex <= lastChunkIndex;
+                    ++chunkIndex) {
+                process();
+                preflightFrameIndexRange = intersect(
+                        preflightFrameIndexRange, m_readableFrameIndexRange);
+                if (preflightFrameIndexRange.empty()) {
+                    break;
+                }
+                lastChunkIndex = CachingReaderChunk::indexForFrame(
+                        preflightFrameIndexRange.end() - 1);
+                if (lastChunkIndex < chunkIndex) {
+                    break;
+                }
+
+                const CachingReaderChunkForOwner* const pChunk =
+                        lookupChunkAndFreshen(chunkIndex);
+                if (!pChunk ||
+                        pChunk->getState() != CachingReaderChunkForOwner::READY) {
+                    DEBUG_ASSERT(!pChunk ||
+                            pChunk->getState() ==
+                                    CachingReaderChunkForOwner::READ_PENDING);
+                    return ReadResult::UNAVAILABLE;
+                }
+            }
+        }
+    }
+
     auto result = ReadResult::AVAILABLE;
     if (!intersect(remainingFrameIndexRange, m_readableFrameIndexRange).empty()) {
         // Fill the buffer up to the first readable sample with

src/engine/cachingreader/cachingreader.h

@@ -107,6 +107,14 @@ class CachingReader : public QObject {
             CSAMPLE* buffer,
             mixxx::audio::ChannelCount channelCount);
 
+    // Like read(), but treats a cache miss in any required chunk as an
+    // all-or-nothing failure. On ReadResult::UNAVAILABLE, buffer is untouched.
+    ReadResult readWithRetry(SINT startSample,
+            SINT numSamples,
+            bool reverse,
+            CSAMPLE* buffer,
+            mixxx::audio::ChannelCount channelCount);
+
     // Issue a list of hints, but check whether any of the hints request a chunk
     // that is not in the cache. If any hints do request a chunk not in cache,
     // then wake the reader so that it can process them. Must only be called
@@ -126,6 +134,23 @@ class CachingReader : public QObject {
         m_worker.setScheduler(pScheduler);
     }
 
+  protected:
+    struct RetryReadResult {
+        ReadResult result;
+        bool retryPending;
+    };
+
+    // Explicit retry hook. Implementations write only to the provided staging
+    // buffer and report whether the same absolute range must be retried. The
+    // default implementation accepts PARTIALLY_AVAILABLE as intentional
+    // padding from legacy readers. Retry-aware subclasses must override this
+    // hook to identify partial cache misses.
+    virtual RetryReadResult readWithRetryHook(SINT startSample,
+            SINT numSamples,
+            bool reverse,
+            CSAMPLE* buffer,
+            mixxx::audio::ChannelCount channelCount);
+
   signals:
     // Emitted once a new track is loaded and ready to be read from.
     void trackLoading();
@@ -136,7 +161,15 @@ class CachingReader : public QObject {
     void trackLoadFailed(TrackPointer pTrack, const QString& reason);
 
   private:
+    ReadResult readInternal(SINT startSample,
+            SINT numSamples,
+            bool reverse,
+            CSAMPLE* buffer,
+            mixxx::audio::ChannelCount channelCount,
+            bool retryOnCacheMiss);
+
     const UserSettingsPointer m_pConfig;
+    bool m_retryOnCacheMiss;
 
     // Thread-safe FIFOs for communication between the engine callback and
     // reader thread.
@@ -194,6 +227,10 @@ class CachingReader : public QObject {
     // The raw memory buffer which is divided up into chunks.
     mixxx::SampleBuffer m_sampleBuffer;
 
+    // Preallocated staging storage that preserves the caller's buffer until a
+    // retry read has completed atomically.
+    mixxx::SampleBuffer m_retryReadBuffer;
+
     // The readable frame index range as reported by the worker.
     mixxx::IndexRange m_readableFrameIndexRange;