Async: Make AsyncPipeline and AsyncRequest{Readable | Writable}Stream…

… public

- Refactor both classes out of the test
- Add some initial documentation

Bindings/cpp/SC.cpp

@@ -1,6 +1,7 @@
 // Copyright (c) Stefano Cristiano
 // SPDX-License-Identifier: MIT
 #include "../../Libraries/Async/Async.cpp"
+#include "../../Libraries/Async/AsyncRequestStreams.cpp"
 #include "../../Libraries/Async/AsyncStreams.cpp"
 #include "../../Libraries/Build/Build.cpp"
 #include "../../Libraries/File/FileDescriptor.cpp"

Documentation/Libraries/Async.md

@@ -8,6 +8,9 @@ Async is a multi-platform / event-driven asynchronous I/O library.
 
 @copydetails group_async
 
+@note
+Check @ref library_async_streams for an higher level construct when streaming data
+
 # Features
 
 This is the list of supported async operations:

Documentation/Libraries/AsyncStreams.md

@@ -0,0 +1,43 @@
+@page library_async_streams Async Streams
+
+@brief 🟨 Concurrently read and write a byte stream staying inside fixed buffers
+
+[TOC]
+
+Async Streams read and write data concurrently from async sources to destinations.
+
+@note Even if the state machine is not strictly depending on @ref library_async, most practical uses of this library will be using it, so it can be considered an extension of @ref library_async
+
+@copydetails group_async_streams
+
+# Features
+
+Async Streams support reading from an async source and placing such reads in a request queue. This queue is bounded, so it will pause the stream when it becomes full.
+Data is pushed downstream to listeners of data events.
+Such listeners can be for example writers and they will eventually emit a `drain` event that resumes the readable streams that may have been paused.
+
+| Async Stream                                          | Description                           |
+|-------------------------------------------------------|---------------------------------------|
+| [AsyncReadableStream](@ref SC::AsyncReadableStream)   | @copybrief SC::AsyncReadableStream    | 
+| [AsyncWritableStream](@ref SC::AsyncWritableStream)   | @copybrief SC::AsyncWritableStream    |
+| [AsyncPipeline](@ref SC::AsyncPipeline)               | @copybrief SC::AsyncPipeline          |
+
+# Implementation
+
+Async streams is heavily inspired by [Node.js streams](https://nodejs.org/api/stream.html) but drops a few features to concentrate on the most useful abstraction.
+
+
+## Memory allocation
+Async streams do not allocate any memory, but use caller provided buffers for handling data and request queues.
+
+# Roadmap
+
+🟩 Usable features:
+- Transform Streams
+
+🟦 Complete Features:
+- Duplex Streams
+
+💡 Unplanned Features:
+- Object Mode
+- readable + read mode

Documentation/Pages/Libraries.md

@@ -6,6 +6,7 @@ Library                                     | Description
 :-------------------------------------------|:-----------------------------------------------
 @subpage library_algorithms                 | @copybrief library_algorithms
 @subpage library_async                      | @copybrief library_async
+@subpage library_async_streams              | @copybrief library_async_streams
 @subpage library_build                      | @copybrief library_build
 @subpage library_containers                 | @copybrief library_containers
 @subpage library_file                       | @copybrief library_file

Libraries/Async/AsyncRequestStreams.cpp

@@ -0,0 +1,160 @@
+// Copyright (c) Stefano Cristiano
+// SPDX-License-Identifier: MIT
+#include "AsyncRequestStreams.h"
+
+//-------------------------------------------------------------------------------------------------------
+// AsyncRequestReadableStream
+//-------------------------------------------------------------------------------------------------------
+template <typename AsyncReadRequest>
+struct SC::AsyncRequestReadableStream<AsyncReadRequest>::Internal
+{
+    static bool isEnded(AsyncFileRead::Result& result) { return result.completionData.endOfFile; }
+    static bool isEnded(AsyncSocketReceive::Result& result) { return result.completionData.disconnected; }
+    static SocketDescriptor::Handle& getDescriptor(AsyncSocketReceive& async) { return async.handle; }
+    static FileDescriptor::Handle&   getDescriptor(AsyncFileRead& async) { return async.fileDescriptor; }
+};
+
+template <typename AsyncReadRequest>
+SC::AsyncRequestReadableStream<AsyncReadRequest>::AsyncRequestReadableStream()
+{
+    AsyncReadableStream::asyncRead.bind<AsyncRequestReadableStream, &AsyncRequestReadableStream::read>(*this);
+}
+
+template <typename AsyncReadRequest>
+template <typename DescriptorType>
+SC::Result SC::AsyncRequestReadableStream<AsyncReadRequest>::init(AsyncBuffersPool& buffersPool, Span<Request> requests,
+                                                                  AsyncEventLoop&       loop,
+                                                                  const DescriptorType& descriptor)
+{
+    request.cacheInternalEventLoop(loop);
+    SC_TRY(descriptor.get(Internal::getDescriptor(request), Result::Error("Missing descriptor")));
+    return AsyncReadableStream::init(buffersPool, requests);
+}
+
+template <typename AsyncReadRequest>
+SC::Result SC::AsyncRequestReadableStream<AsyncReadRequest>::read()
+{
+    if (request.isFree())
+    {
+        AsyncBufferView::ID bufferID;
+        SC_TRY(getBuffersPool().requestNewBuffer(0, bufferID, request.buffer))
+        request.callback = [this, bufferID](typename AsyncReadRequest::Result& result) { afterRead(result, bufferID); };
+        const Result startResult = request.start(*request.getEventLoop());
+        if (startResult)
+        {
+            return Result(true); // started successfully
+        }
+        else
+        {
+            getBuffersPool().unrefBuffer(bufferID);
+            return startResult; // Error occurred during request start
+        }
+    }
+    else
+    {
+        // read is already in progress from a previous callback that has called reactivateRequest(true)
+        return Result(true);
+    }
+}
+
+template <typename AsyncReadRequest>
+void SC::AsyncRequestReadableStream<AsyncReadRequest>::afterRead(typename AsyncReadRequest::Result& result,
+                                                                 AsyncBufferView::ID                bufferID)
+{
+    SC_ASSERT_RELEASE(result.getAsync().isFree());
+    Span<char> data;
+    if (result.get(data))
+    {
+        if (Internal::isEnded(result))
+        {
+            getBuffersPool().unrefBuffer(bufferID);
+            AsyncReadableStream::pushEnd();
+        }
+        else
+        {
+            AsyncReadableStream::push(bufferID, data.sizeInBytes());
+            SC_ASSERT_RELEASE(result.getAsync().isFree());
+            getBuffersPool().unrefBuffer(bufferID);
+            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).
+            }
+        }
+    }
+    else
+    {
+        getBuffersPool().unrefBuffer(bufferID);
+        AsyncReadableStream::emitError(result.isValid());
+    }
+}
+
+//-------------------------------------------------------------------------------------------------------
+// AsyncRequestWritableStream
+//-------------------------------------------------------------------------------------------------------
+
+template <typename AsyncWriteRequest>
+struct SC::AsyncRequestWritableStream<AsyncWriteRequest>::Internal
+{
+    static SocketDescriptor::Handle& getDescriptor(AsyncSocketSend& async) { return async.handle; }
+    static FileDescriptor::Handle&   getDescriptor(AsyncFileWrite& async) { return async.fileDescriptor; }
+};
+
+template <typename AsyncWriteRequest>
+SC::AsyncRequestWritableStream<AsyncWriteRequest>::AsyncRequestWritableStream()
+{
+    AsyncWritableStream::asyncWrite.bind<AsyncRequestWritableStream, &AsyncRequestWritableStream::write>(*this);
+}
+
+template <typename AsyncWriteRequest>
+template <typename DescriptorType>
+SC::Result SC::AsyncRequestWritableStream<AsyncWriteRequest>::init(AsyncBuffersPool& buffersPool,
+                                                                   Span<Request> requests, AsyncEventLoop& loop,
+                                                                   const DescriptorType& descriptor)
+{
+    request.cacheInternalEventLoop(loop);
+    SC_TRY(descriptor.get(Internal::getDescriptor(request), Result::Error("Missing descriptor")));
+    return AsyncWritableStream::init(buffersPool, requests);
+}
+
+template <typename AsyncWriteRequest>
+SC::Result SC::AsyncRequestWritableStream<AsyncWriteRequest>::write(AsyncBufferView::ID                 bufferID,
+                                                                    Function<void(AsyncBufferView::ID)> cb)
+{
+    SC_ASSERT_RELEASE(not callback.isValid());
+    callback = move(cb);
+    SC_TRY(getBuffersPool().getData(bufferID, request.buffer));
+    request.callback = [this, bufferID](typename AsyncWriteRequest::Result& result)
+    {
+        getBuffersPool().unrefBuffer(bufferID);
+        auto callbackCopy = move(callback);
+        callback          = {};
+        AsyncWritableStream::finishedWriting(bufferID, move(callbackCopy), result.isValid());
+    };
+    const Result res = request.start(*request.getEventLoop());
+    if (res)
+    {
+        getBuffersPool().refBuffer(bufferID);
+    }
+    return res;
+}
+namespace SC
+{
+SC_COMPILER_EXTERN template struct AsyncRequestReadableStream<AsyncSocketReceive>;
+SC_COMPILER_EXTERN template struct AsyncRequestReadableStream<AsyncFileRead>;
+SC_COMPILER_EXTERN template struct AsyncRequestWritableStream<AsyncFileWrite>;
+SC_COMPILER_EXTERN template struct AsyncRequestWritableStream<AsyncSocketSend>;
+
+SC_COMPILER_EXTERN template SC::Result SC::AsyncRequestReadableStream<AsyncSocketReceive>::init(
+    AsyncBuffersPool& buffersPool, Span<Request> requests, AsyncEventLoop& loop, const SocketDescriptor& descriptor);
+SC_COMPILER_EXTERN template SC::Result SC::AsyncRequestWritableStream<AsyncSocketSend>::init(
+    AsyncBuffersPool& buffersPool, Span<Request> requests, AsyncEventLoop& loop, const SocketDescriptor& descriptor);
+SC_COMPILER_EXTERN template SC::Result SC::AsyncRequestReadableStream<AsyncFileRead>::init(
+    AsyncBuffersPool& buffersPool, Span<Request> requests, AsyncEventLoop& loop, const FileDescriptor& descriptor);
+SC_COMPILER_EXTERN template SC::Result SC::AsyncRequestWritableStream<AsyncFileWrite>::init(
+    AsyncBuffersPool& buffersPool, Span<Request> requests, AsyncEventLoop& loop, const FileDescriptor& descriptor);
+
+} // namespace SC

Libraries/Async/AsyncRequestStreams.h

@@ -0,0 +1,52 @@
+// Copyright (c) Stefano Cristiano
+// SPDX-License-Identifier: MIT
+#pragma once
+#include "Async.h"
+#include "AsyncStreams.h"
+
+//! @addtogroup group_async_streams
+//! @{
+namespace SC
+{
+template <typename AsyncRequestType>
+struct AsyncRequestReadableStream : public AsyncReadableStream
+{
+    AsyncRequestReadableStream();
+
+    template <typename DescriptorType>
+    Result init(AsyncBuffersPool& buffersPool, Span<Request> requests, AsyncEventLoop& loop,
+                const DescriptorType& descriptor);
+
+  private:
+    struct Internal;
+    AsyncRequestType request;
+
+    Result read();
+    void   afterRead(typename AsyncRequestType::Result& result, AsyncBufferView::ID bufferID);
+};
+
+template <typename AsyncRequestType>
+struct AsyncRequestWritableStream : public AsyncWritableStream
+{
+    AsyncRequestWritableStream();
+
+    template <typename DescriptorType>
+    Result init(AsyncBuffersPool& buffersPool, Span<Request> requests, AsyncEventLoop& loop,
+                const DescriptorType& descriptor);
+
+  private:
+    struct Internal;
+    AsyncRequestType request;
+
+    Function<void(AsyncBufferView::ID)> callback;
+
+    Result write(AsyncBufferView::ID bufferID, Function<void(AsyncBufferView::ID)> cb);
+};
+
+using ReadableFileStream   = AsyncRequestReadableStream<AsyncFileRead>;
+using WritableFileStream   = AsyncRequestWritableStream<AsyncFileWrite>;
+using ReadableSocketStream = AsyncRequestReadableStream<AsyncSocketReceive>;
+using WritableSocketStream = AsyncRequestWritableStream<AsyncSocketSend>;
+
+} // namespace SC
+//! @}

Libraries/Async/AsyncStreams.cpp

@@ -459,3 +459,46 @@ void SC::AsyncWritableStream::end()
 }
 
 SC::AsyncBuffersPool& SC::AsyncWritableStream::getBuffersPool() { return *buffers; }
+
+//-------------------------------------------------------------------------------------------------------
+// AsyncPipeline
+//-------------------------------------------------------------------------------------------------------
+
+SC::Result SC::AsyncPipeline::start()
+{
+    SC_TRY_MSG(source != nullptr, "AsyncPipeline::start - Missing source");
+
+    AsyncBuffersPool& buffers = source->getBuffersPool();
+    for (Sink sink : sinks)
+    {
+        if (&sink.sink->getBuffersPool() != &buffers)
+        {
+            return Result::Error("AsyncPipeline::start - all streams must use the same AsyncBuffersPool");
+        }
+    }
+
+    // TODO: Register also onErrors
+    SC_TRY((source->eventData.addListener<AsyncPipeline, &AsyncPipeline::onBufferRead>(*this)));
+    return source->start();
+}
+
+void SC::AsyncPipeline::onBufferRead(AsyncBufferView::ID bufferID)
+{
+    for (Sink sink : sinks)
+    {
+        source->getBuffersPool().refBuffer(bufferID); // 4a. AsyncPipeline::onBufferWritten
+        Function<void(AsyncBufferView::ID)> cb;
+        cb.bind<AsyncPipeline, &AsyncPipeline::onBufferWritten>(*this);
+        Result res = sink.sink->write(bufferID, move(cb));
+        if (not res)
+        {
+            eventError.emit(res);
+        }
+    }
+}
+
+void SC::AsyncPipeline::onBufferWritten(AsyncBufferView::ID bufferID)
+{
+    source->getBuffersPool().unrefBuffer(bufferID); // 4b. AsyncPipeline::onBufferRead
+    source->resume();
+}

Libraries/Async/AsyncStreams.h

@@ -230,5 +230,36 @@ struct AsyncWritableStream
 
     CircularQueue<Request> writeQueue;
 };
+
+/// @brief Pipes reads on SC::AsyncReadableStream to SC::AsyncWritableStream.
+/// Back-pressure happens when the source provides data at a faster rate than what the sink (writable)
+/// is able to process.
+/// When this happens, AsyncPipeline will AsyncReadableStream::pause the (source).
+/// It will also AsyncReadableStream::resume it when some writable has finished writing, freeing one buffer.
+/// Caller needs to set AsyncPipeline::source field and AsyncPipeline::sinks with valid streams.
+/// @note It's crucial to use the same AsyncBuffersPool for the AsyncReadableStream and all AsyncWritableStream
+struct AsyncPipeline
+{
+    static constexpr int        MaxListeners = 8;
+    Event<MaxListeners, Result> eventError; /// Emitted when an error occurs
+
+    // TODO: Make all these private
+    AsyncReadableStream* source = nullptr; /// User specified source
+
+    struct Sink
+    {
+        AsyncWritableStream* sink = nullptr;
+    };
+    Span<Sink> sinks; /// User specified sinks
+
+    /// @brief Starts the pipeline
+    /// @note Both source and sinks must have been already setup by the caller
+    Result start();
+
+    // TODO: Add a pause and cancel/step
+  private:
+    void onBufferRead(AsyncBufferView::ID bufferID);
+    void onBufferWritten(AsyncBufferView::ID bufferID);
+};
 } // namespace SC
 //! @}