AsyncStreams: Return true from AsyncReadableStream::push if the caller can continue pushing

Author: Pagghiu

Libraries/AsyncStreams/AsyncRequestStreams.cpp

@@ -102,16 +102,20 @@ void SC::AsyncRequestReadableStream<AsyncReadRequest>::afterRead(typename AsyncR
         }
         else
         {
-            AsyncReadableStream::push(bufferID, data.sizeInBytes());
+            const bool continuePushing = AsyncReadableStream::push(bufferID, data.sizeInBytes());
             SC_ASSERT_RELEASE(result.getAsync().isFree());
             getBuffersPool().unrefBuffer(bufferID);
-            if (getBufferOrPause(0, bufferID, result.getAsync().buffer))
+            // Check if we're still pushing (so not, paused, destroyed or errored etc.)
+            if (continuePushing)
             {
-                request.callback = [this, bufferID](typename AsyncReadRequest::Result& result)
-                { afterRead(result, bufferID); };
-                result.reactivateRequest(true);
-                // Stream is in AsyncPushing mode and SC::AsyncResult::reactivateRequest(true) will cause more
-                // data to be delivered here, so it's not necessary calling AsyncReadableStream::reactivate(true).
+                if (getBufferOrPause(0, bufferID, result.getAsync().buffer))
+                {
+                    request.callback = [this, bufferID](typename AsyncReadRequest::Result& result)
+                    { afterRead(result, bufferID); };
+                    result.reactivateRequest(true);
+                    // Stream is in AsyncPushing mode and SC::AsyncResult::reactivateRequest(true) will cause more
+                    // data to be delivered here, so it's not necessary calling AsyncReadableStream::reactivate(true).
+                }
             }
         }
     }

Libraries/AsyncStreams/AsyncStreams.cpp

@@ -189,12 +189,12 @@ void AsyncReadableStream::emitOnData()
     }
 }
 
-void AsyncReadableStream::push(AsyncBufferView::ID bufferID, size_t newSize)
+bool AsyncReadableStream::push(AsyncBufferView::ID bufferID, size_t newSize)
 {
     if (newSize == 0)
     {
         emitError(Result::Error("AsyncReadableStream::push zero sized buffer is not allowed"));
-        return;
+        return false;
     }
     // Push buffer to the queue
     buffers->setNewBufferSize(bufferID, newSize);
@@ -204,7 +204,7 @@ void AsyncReadableStream::push(AsyncBufferView::ID bufferID, size_t newSize)
     {
         state = State::Errored;
         emitError(Result::Error("AsyncReadableStream::push dropping buffer"));
-        return;
+        return false;
     }
     buffers->refBuffer(bufferID); // 1a. unrefBuffer in emitOnData()
 
@@ -219,7 +219,10 @@ void AsyncReadableStream::push(AsyncBufferView::ID bufferID, size_t newSize)
     case State::AsyncPushing:
     case State::AsyncReading: {
         emitOnData();
-        state = State::AsyncPushing;
+        if (state == State::AsyncReading)
+        {
+            state = State::AsyncPushing;
+        }
     }
     break;
     case State::Pausing: {
@@ -232,6 +235,7 @@ void AsyncReadableStream::push(AsyncBufferView::ID bufferID, size_t newSize)
     }
     break;
     }
+    return state == State::AsyncPushing or state == State::SyncPushing;
 }
 
 void AsyncReadableStream::reactivate(bool doReactivate)
@@ -652,7 +656,8 @@ void AsyncTransformStream::afterProcess(Span<const char> inputAfter, Span<char>
     const size_t consumedOutput = outputData.sizeInBytes() - outputAfter.sizeInBytes();
     if (consumedOutput > 0)
     {
-        AsyncReadableStream::push(outputBufferID, consumedOutput);
+        // Ignore whatever push returns because later on the stream is either finalizing or pausing either way
+        (void)AsyncReadableStream::push(outputBufferID, consumedOutput);
     }
     AsyncReadableStream::getBuffersPool().unrefBuffer(outputBufferID);
     if (inputAfter.empty())
@@ -681,7 +686,8 @@ void AsyncTransformStream::afterFinalize(Span<char> outputAfter, bool streamEnde
     const size_t consumedOutput = outputData.sizeInBytes() - outputAfter.sizeInBytes();
     if (consumedOutput > 0)
     {
-        AsyncReadableStream::push(outputBufferID, consumedOutput);
+        // Ignore whatever push returns because later on the stream is either finalizing or pausing either way
+        (void)AsyncReadableStream::push(outputBufferID, consumedOutput);
     }
     AsyncReadableStream::getBuffersPool().unrefBuffer(outputBufferID);
     if (streamEnded)

Libraries/AsyncStreams/AsyncStreams.h

@@ -208,7 +208,8 @@ struct AsyncReadableStream
     AsyncBuffersPool& getBuffersPool();
 
     /// @brief Use push from inside AsyncReadableStream::asyncRead function to queue received data
-    void push(AsyncBufferView::ID bufferID, size_t newSize);
+    /// @return `true` if the caller can continue pushing
+    [[nodiscard]] bool push(AsyncBufferView::ID bufferID, size_t newSize);
 
     /// @brief Use pushEnd from inside AsyncReadableStream::asyncRead to signal production end
     void pushEnd();

Libraries/AsyncStreams/ZLibTransformStreams.cpp

@@ -23,9 +23,11 @@ SC::Result SC::SyncZLibTransformStream::transform(AsyncBufferView::ID bufferID,
     SC_TRY(sourceData.sliceStart(consumedInputBytes, inputData));
     while (not inputData.empty())
     {
-        Span<char>          outputData;
         AsyncBufferView::ID outputBufferID;
-        if (getBufferOrPause(0, outputBufferID, outputData))
+
+        Span<char> outputData;
+        bool       continuePushing = true;
+        if (getBufferOrPause(0, outputBufferID, outputData) and continuePushing)
         {
             const size_t outputBefore = outputData.sizeInBytes();
             const size_t inputBefore  = inputData.sizeInBytes();
@@ -40,7 +42,7 @@ SC::Result SC::SyncZLibTransformStream::transform(AsyncBufferView::ID bufferID,
             consumedInputBytes += consumedInput;
             if (consumedOutput > 0)
             {
-                AsyncReadableStream::push(outputBufferID, consumedOutput);
+                continuePushing = AsyncReadableStream::push(outputBufferID, consumedOutput);
             }
             AsyncReadableStream::getBuffersPool().unrefBuffer(outputBufferID);
         }
@@ -65,8 +67,10 @@ bool SC::SyncZLibTransformStream::canEndTransform()
     // we need to hold the "Ending" state of the state machine, to finish
     // writing this last trail of transformed data.
     AsyncBufferView::ID outputBufferID;
-    Span<char>          outputBefore;
-    while (getBufferOrPause(0, outputBufferID, outputBefore))
+
+    Span<char> outputBefore;
+    bool       continuePushing = true;
+    while (getBufferOrPause(0, outputBufferID, outputBefore) and continuePushing)
     {
         Span<char> outputData = outputBefore;
 
@@ -80,7 +84,7 @@ bool SC::SyncZLibTransformStream::canEndTransform()
         const size_t outputBytes = outputBefore.sizeInBytes() - outputData.sizeInBytes();
         if (outputBytes > 0)
         {
-            AsyncReadableStream::push(outputBufferID, outputBytes);
+            continuePushing = AsyncReadableStream::push(outputBufferID, outputBytes);
         }
         AsyncReadableStream::getBuffersPool().unrefBuffer(outputBufferID);
         if (streamEnded)

Tests/Libraries/AsyncStreams/AsyncStreamsTest.cpp

@@ -154,7 +154,7 @@ void SC::AsyncStreamsTest::readableSyncStream()
     } context = {readable, 0, 100, {}};
 
     (void)readable.eventError.addListener([this](Result res) { SC_TEST_EXPECT(res); });
-    readable.asyncRead = [&context]() -> Result
+    readable.asyncRead = [this, &context]() -> Result
     {
         if (context.idx < context.max)
         {
@@ -163,7 +163,7 @@ void SC::AsyncStreamsTest::readableSyncStream()
             if (context.readable.getBufferOrPause(sizeof(context.idx), bufferID, data))
             {
                 memcpy(data.data(), &context.idx, sizeof(context.idx));
-                context.readable.push(bufferID, sizeof(context.idx));
+                SC_TEST_EXPECT(context.readable.push(bufferID, sizeof(context.idx)));
                 context.readable.getBuffersPool().unrefBuffer(bufferID);
                 context.idx += 1;
                 context.readable.reactivate(true);
@@ -231,14 +231,14 @@ void SC::AsyncStreamsTest::readableAsyncStream()
 
     SC_TEST_EXPECT(loop.create());
     AsyncLoopTimeout timeout;
-    timeout.callback = [&context](AsyncLoopTimeout::Result&)
+    timeout.callback = [this, &context](AsyncLoopTimeout::Result&)
     {
         AsyncBufferView::ID bufferID;
         Span<char>          data;
         if (context.readable.getBufferOrPause(sizeof(context.idx), bufferID, data))
         {
             memcpy(data.data(), &context.idx, sizeof(context.idx));
-            context.readable.push(bufferID, sizeof(context.idx));
+            SC_TEST_EXPECT(context.readable.push(bufferID, sizeof(context.idx)));
             context.readable.getBuffersPool().unrefBuffer(bufferID);
             context.idx += 1;
             context.readable.reactivate(true);