AsyncStreams.h

// Copyright (c) Stefano Cristiano
// SPDX-License-Identifier: MIT
#pragma once
 
#include "../Common/CompilerMacrosExport.h"
#include "../Common/PlatformMacrosInstructionSet.h"
#ifndef SC_EXPORT_LIBRARY_ASYNC_STREAMS
#define SC_EXPORT_LIBRARY_ASYNC_STREAMS 0
#endif
#define SC_ASYNC_STREAMS_EXPORT SC_COMPILER_LIBRARY_EXPORT(SC_EXPORT_LIBRARY_ASYNC_STREAMS)
 
#include "../Common/AlignedStorage.h"
#include "../Common/Assert.h"
#include "../Common/Function.h"
#include "../Common/IGrowableBufferSpan.h"
#include "../Common/PlatformMacrosInstructionSet.h"
#include "../Common/Result.h"
#include "../Common/Span.h"
#include "Internal/CircularQueue.h"
#include "Internal/Event.h"
 
 
namespace SC
{
SC_DECLARE_ASSERT_PROVIDER(AsyncStreamsAssert, SC_ASYNC_STREAMS_EXPORT);
 
#define SC_ASYNC_STREAMS_ASSERT_RELEASE(e)        SC_ASSERT_PROVIDER_RELEASE(SC::AsyncStreamsAssert, e)
#define SC_ASYNC_STREAMS_ASSERT_DEBUG(e)          SC_ASSERT_PROVIDER_DEBUG(SC::AsyncStreamsAssert, e)
#define SC_ASYNC_STREAMS_TRUST_RESULT(expression) SC_ASYNC_STREAMS_ASSERT_RELEASE(expression)
 
struct SC_ASYNC_STREAMS_EXPORT AsyncBufferView
{
    struct SC_ASYNC_STREAMS_EXPORT ID
    {
        using NumericType = int32_t;
 
        static constexpr NumericType InvalidValue = -1;
 
        NumericType identifier;
 
        constexpr ID() : identifier(InvalidValue) {}
        explicit constexpr ID(int32_t value) : identifier(value) {}
 
        [[nodiscard]] constexpr bool isValid() const { return identifier != InvalidValue; }
        [[nodiscard]] constexpr bool operator==(ID other) const { return identifier == other.identifier; }
    };
    enum class Type : uint8_t
    {
        Empty,
        Writable,
        ReadOnly,
        Growable,
        Child,
    };
 
    AsyncBufferView() : writableData(), offset(0), length(0), refs(0), type(Type::Empty), reUse(false) {}
    AsyncBufferView(Span<char> data) : writableData(data)
    {
        type     = Type::Writable;
        offset   = 0;
        length   = data.sizeInBytes();
        parentID = ID();
    }
    AsyncBufferView(Span<const char> data) : readonlyData(data)
    {
        type     = Type::ReadOnly;
        offset   = 0;
        length   = data.sizeInBytes();
        parentID = ID();
    }
 
    void setReusable(bool reusable) { reUse = reusable; }
 
    template <typename T>
    AsyncBufferView(T&& t) // universal reference, it can capture both lvalue and rvalue
    {
        type     = Type::Growable;
        offset   = 0;
        length   = 0;
        parentID = ID();
        // Here we're type-erasing T in our own inline storage provided by a slightly oversized Function<>
        // that it will be able to construct (and destruct) the right GrowableBuffer<T> from just a piece of storage
        // and return a pointer to the corresponding IGrowableBuffer* interface
        getGrowableBuffer = [t = forward<T>(t)](GrowableStorage& storage, bool construct) mutable -> IGrowableBuffer*
        {
            using Type = typename TypeTraits::RemoveReference<T>::type;
            if (construct)
            {
                placementNew(storage.reinterpret_as<GrowableBuffer<Type>>(), t);
                return &storage.reinterpret_as<GrowableBuffer<Type>>();
            }
            else
            {
                dtor(storage.reinterpret_as<GrowableBuffer<Type>>());
                return nullptr;
            }
        };
    }
 
    template <int N>
    AsyncBufferView(const char (&literal)[N])
    {
        readonlyData = {literal, N - 1};
        type         = Type::ReadOnly;
        offset       = 0;
        length       = N - 1;
    }
 
    Type getType() const { return type; }
 
  private:
#if SC_PLATFORM_64_BIT
    static constexpr int TypeErasedCaptureSize = sizeof(void*) * 3; // This is enough to hold String / Buffer by copy
#else
    static constexpr int TypeErasedCaptureSize = sizeof(void*) * 6; // This is enough to hold String / Buffer by copy
#endif
    static constexpr int TypeErasedGrowableSize = sizeof(void*) * 6;
 
    using GrowableStorage = AlignedStorage<TypeErasedGrowableSize>;
    Function<IGrowableBuffer*(GrowableStorage&, bool), TypeErasedCaptureSize> getGrowableBuffer;
 
    union
    {
        Span<char>       writableData;
        Span<const char> readonlyData;
    };
    AsyncBufferView::ID parentID;
 
    friend struct AsyncBuffersPool;
 
    size_t  offset = 0;
    size_t  length = 0;
    int32_t refs   = 0;           // Counts AsyncReadable (single) or AsyncWritable (multiple) using it
    Type    type   = Type::Empty; // If it's Empty, Writable, ReadOnly, Growable or Child
    bool    reUse  = false;       // If it can be re-used after refs == 0
};
 
struct SC_ASYNC_STREAMS_EXPORT AsyncBuffersPool
{
    void refBuffer(AsyncBufferView::ID bufferID);
 
    void unrefBuffer(AsyncBufferView::ID bufferID);
 
    Result getReadableData(AsyncBufferView::ID bufferID, Span<const char>& data);
 
    Result getWritableData(AsyncBufferView::ID bufferID, Span<char>& data);
 
    AsyncBufferView* getBuffer(AsyncBufferView::ID bufferID);
 
    Result requestNewBuffer(size_t minimumSizeInBytes, AsyncBufferView::ID& bufferID, Span<char>& data);
 
    void setNewBufferSize(AsyncBufferView::ID bufferID, size_t newSizeInBytes);
 
    Result pushBuffer(AsyncBufferView&& buffer, AsyncBufferView::ID& bufferID);
 
    static Result sliceInEqualParts(Span<AsyncBufferView> buffers, Span<char> memory, size_t numSlices);
 
    void setBuffers(Span<AsyncBufferView> newBuffers) { buffers = newBuffers; }
 
    [[nodiscard]] size_t getNumBuffers() const { return buffers.sizeInElements(); }
 
    Result createChildView(AsyncBufferView::ID parentBufferID, size_t offset, size_t length,
                           AsyncBufferView::ID& outChildBufferID);
 
  private:
    Span<AsyncBufferView> buffers;
};
 
struct SC_ASYNC_STREAMS_EXPORT AsyncReadableStream
{
    using BufferViewID = AsyncBufferView::ID;
    using BuffersPool  = AsyncBuffersPool;
 
    struct Request
    {
        AsyncBufferView::ID bufferID;
    };
 
    static constexpr int MaxListeners = 8;
 
    Event<MaxListeners, Result>              eventError; 
    Event<MaxListeners, AsyncBufferView::ID> eventData;  
    Event<MaxListeners>                      eventEnd;   
    Event<MaxListeners>                      eventClose; 
 
    Result init(AsyncBuffersPool& buffersPool);
 
    Result start();
 
    void pause();
 
    void resumeReading();
 
    void destroy();
 
    [[nodiscard]] bool isEnded() const { return state == State::Ended; }
 
    [[nodiscard]] bool hasQueuedData() const { return not readQueue.isEmpty(); }
 
    [[nodiscard]] bool hasBeenDestroyed() const { return destroyed; }
 
    [[nodiscard]] bool canStart() const { return state == State::CanRead; }
 
    AsyncBuffersPool& getBuffersPool();
 
    constexpr void setReadQueue(Span<Request> requests) { readQueue = requests; }
 
    [[nodiscard]] size_t getReadQueueSize() const { return readQueue.size(); }
 
    [[nodiscard]] bool push(AsyncBufferView::ID bufferID, size_t newSize);
 
    Result unshift(AsyncBufferView::ID bufferID);
 
    void pushEnd();
 
    void reactivate(bool doReactivate);
 
    void emitError(Result error);
 
    [[nodiscard]] bool getBufferOrPause(size_t minumumSizeInBytes, AsyncBufferView::ID& bufferID, Span<char>& data);
 
    void setAutoDestroy(bool value) { autoDestroy = value; }
 
    [[nodiscard]] bool getAutoDestroy() const { return autoDestroy; }
 
  protected:
    virtual ~AsyncReadableStream();
 
    virtual Result asyncRead() = 0;
 
    virtual Result asyncResumeReading();
 
    virtual Result asyncDestroyReadable();
 
    Result finishedDestroyingReadable();
 
  private:
    void maybeDestroyEndedReadable();
    void emitOnData();
    void executeRead();
 
    enum class State : uint8_t
    {
        Stopped,      // Stream must be inited
        CanRead,      // Stream is ready to issue a read ( AsyncReadableStream::start / AsyncReadableStream::resume)
        Reading,      // A read is being issued (may be sync or async)
        SyncPushing,  // One or multiple AsyncReadableStream::push have been received (sync)
        SyncReadMore, // SyncPushing + AsyncReadableStream::reactivate(true)
        AsyncReading, // An async read is in flight
        AsyncPushing, // AsyncReading + AsyncReadableStream::push
        Pausing,      // Pause requested while read in flight
        Paused,       // Actually paused with no read in flight
        Ended,        // Emitted all data, no more data will be emitted
        Destroying,   // Destroy has been requested and it's in progress
        Errored,      // Error occurred
    };
    State state = State::Stopped;
 
    bool destroyed   = false;
    bool autoDestroy = true;
 
    AsyncBuffersPool* buffers = nullptr;
 
    CircularQueue<Request> readQueue;
};
 
struct SC_ASYNC_STREAMS_EXPORT AsyncWritableStream
{
    using BufferViewID = AsyncBufferView::ID;
    using BuffersPool  = AsyncBuffersPool;
 
    friend struct AsyncPipeline;
 
    struct Request
    {
        AsyncBufferView::ID bufferID;
 
        Function<void(AsyncBufferView::ID)> cb;
    };
    static constexpr int MaxListeners = 8;
 
    Event<MaxListeners, Result> eventError; 
 
    Event<MaxListeners> eventDrain;  
    Event<MaxListeners> eventFinish; 
    Event<MaxListeners> eventClose;  
 
    Result init(AsyncBuffersPool& buffersPool);
 
    constexpr void setWriteQueue(Span<Request> requests) { writeQueue = requests; }
 
    [[nodiscard]] size_t getWriteQueueSize() const { return writeQueue.size(); }
 
    Result write(AsyncBufferView::ID bufferID, Function<void(AsyncBufferView::ID)> cb = {});
 
    Result write(AsyncBufferView&& bufferView, Function<void(AsyncBufferView::ID)> cb = {});
 
    void end();
 
    void destroy();
 
    AsyncBuffersPool& getBuffersPool();
 
    void finishedWriting(AsyncBufferView::ID bufferID, Function<void(AsyncBufferView::ID)>&& cb, Result res);
 
    void resumeWriting();
 
    Result unshift(AsyncBufferView::ID bufferID, Function<void(AsyncBufferView::ID)>&& cb);
 
    void emitError(Result error);
 
    void tryAsync(Result potentialError);
 
    [[nodiscard]] bool isStillWriting() const { return state == State::Writing or state == State::Ending; }
 
    [[nodiscard]] bool hasBeenDestroyed() const { return destroyed; }
 
    void setAutoDestroy(bool value) { autoDestroy = value; }
 
    [[nodiscard]] bool getAutoDestroy() const { return autoDestroy; }
 
  protected:
    virtual ~AsyncWritableStream();
 
    virtual Result asyncWrite(AsyncBufferView::ID, Function<void(AsyncBufferView::ID)> func) = 0;
 
    virtual bool canEndWritable();
 
    virtual Result asyncDestroyWritable();
 
    void finishedDestroyingWritable();
 
    void stop() { state = State::Stopped; }
 
  private:
    [[nodiscard]] bool canAcceptWrite() const
    {
        return (state == State::Stopped or state == State::Writing) and not writeQueue.isFull();
    }
 
    enum class State : uint8_t
    {
        Stopped,
        Writing,
        Ending,
        Ended,
        Destroying,
        Errored,
    };
    State state = State::Stopped;
 
    bool destroyed   = false;
    bool autoDestroy = true;
 
    AsyncBuffersPool* buffers = nullptr;
 
    CircularQueue<Request> writeQueue;
};
 
struct SC_ASYNC_STREAMS_EXPORT AsyncDuplexStream : public AsyncReadableStream, public AsyncWritableStream
{
    AsyncDuplexStream();
 
    Result init(AsyncBuffersPool& buffersPool, Span<AsyncReadableStream::Request> readableRequests,
                Span<AsyncWritableStream::Request> writableRequests);
 
    virtual Result asyncRead() override;
};
 
struct SC_ASYNC_STREAMS_EXPORT AsyncTransformStream : public AsyncDuplexStream
{
    AsyncTransformStream();
 
    void afterProcess(Span<const char> inputAfter, Span<char> outputAfter);
    void afterFinalize(Span<char> outputAfter, bool streamEnded);
 
    virtual Result onProcess(Span<const char>, Span<char>) = 0;
    virtual Result onFinalize(Span<char>)                  = 0;
 
  private:
    virtual Result asyncWrite(AsyncBufferView::ID bufferID, Function<void(AsyncBufferView::ID)> cb) override;
    virtual bool   canEndWritable() override;
 
    Function<void(AsyncBufferView::ID)> inputCallback;
 
    Span<const char> inputData;
    Span<char>       outputData;
 
    AsyncBufferView::ID inputBufferID;
    AsyncBufferView::ID outputBufferID;
 
    Result prepare(AsyncBufferView::ID bufferID, Function<void(AsyncBufferView::ID)> cb);
 
    void tryFinalize();
 
    enum class State
    {
        None,
        Paused,
        Processing,
        Finalizing,
        Finalized,
    };
    State state = State::None;
};
 
struct SC_ASYNC_STREAMS_EXPORT AsyncStreams
{
    using BufferView     = AsyncBufferView;
    using BuffersPool    = AsyncBuffersPool;
    using ReadableStream = AsyncReadableStream;
    using WritableStream = AsyncWritableStream;
    using DuplexStream   = AsyncDuplexStream;
};
 
struct SC_ASYNC_STREAMS_EXPORT AsyncPipeline
{
    static constexpr int MaxListeners  = 8;
    static constexpr int MaxTransforms = 8;
    static constexpr int MaxSinks      = 8;
 
    AsyncPipeline()                                = default;
    AsyncPipeline(const AsyncPipeline&)            = delete;
    AsyncPipeline(AsyncPipeline&&)                 = delete;
    AsyncPipeline& operator=(const AsyncPipeline&) = delete;
    AsyncPipeline& operator=(AsyncPipeline&&)      = delete;
    ~AsyncPipeline();
 
    AsyncReadableStream* source                    = nullptr;   
    AsyncDuplexStream*   transforms[MaxTransforms] = {nullptr}; 
    AsyncWritableStream* sinks[MaxSinks]           = {nullptr}; 
    Event<MaxListeners, Result> eventError         = {};        
 
    Result pipe();
 
    [[nodiscard]] bool unpipe();
 
    Result start();
 
    // Internal state used by the pipeline implementation.
    AsyncReadableStream* dispatchReadable               = nullptr;
    AsyncReadableStream* transformInputs[MaxTransforms] = {nullptr};
 
    struct PendingWrite
    {
        AsyncReadableStream* readable = nullptr;
        AsyncWritableStream* writable = nullptr;
        AsyncBufferView::ID  bufferID;
    };
    PendingWrite pendingWrites[MaxTransforms + MaxSinks] = {};
    bool         shouldEndWhenDrained                    = false;
    bool         endingPipes                             = false;
 
    // TODO: Add a pause and cancel/step
  private:
    void   emitError(Result res);
    Result checkBuffersPool();
    Result chainTransforms(AsyncReadableStream*& readable);
    Result validate();
 
    void asyncWriteWritable(AsyncBufferView::ID bufferID, AsyncReadableStream& readable, AsyncWritableStream& writable);
    void dispatchToPipes(AsyncBufferView::ID bufferID);
    void endPipes();
    void afterSinkEnd();
    void afterWrite(AsyncBufferView::ID bufferID);
    void dispatchToTransform(AsyncBufferView::ID bufferID, size_t transformIndex);
    bool retryPendingWrites();
    bool hasPendingWrites() const;
    bool hasPendingWritesForReadable(const AsyncReadableStream& readable) const;
    void releasePendingWrites();
    bool listenToEventData(AsyncReadableStream& readable, AsyncDuplexStream& transform, bool listen);
 
    PendingWrite* findPendingWrite(AsyncReadableStream& readable, AsyncWritableStream& writable);
};
} // namespace SC