Async.h

// Copyright (c) Stefano Cristiano
// SPDX-License-Identifier: MIT
#pragma once
 
#include "../Foundation/Function.h"
#include "../Foundation/OpaqueObject.h"
#include "../Foundation/Span.h"
#include "../Threading/Atomic.h"
#include "../Threading/ThreadPool.h"
#include "../Time/Time.h"
 
// Descriptors
#include "../File/FileDescriptor.h"
#include "../Process/ProcessDescriptor.h"
#include "../Socket/SocketDescriptor.h"
 
namespace SC
{
// Forward Declarations
struct ThreadPool;
struct ThreadPoolTask;
 
struct EventObject;
struct AsyncKernelEvents;
struct AsyncEventLoopListeners;
struct AsyncEventLoop;
struct AsyncEventLoopMonitor;
 
struct AsyncRequest;
struct AsyncResult;
template <typename T, typename C>
struct AsyncResultOf;
struct AsyncCompletionData;
 
struct AsyncTask;
template <typename AsyncType>
struct AsyncTaskOf;
} // namespace SC
 
namespace SC
{
namespace detail
{
struct AsyncWinOverlapped;
struct AsyncWinOverlappedDefinition
{
    static constexpr int    Windows   = sizeof(void*) * 4 + sizeof(uint64_t);
    static constexpr size_t Alignment = alignof(void*);
 
    using Object = AsyncWinOverlapped;
};
using WinOverlappedOpaque = OpaqueObject<AsyncWinOverlappedDefinition>;
 
struct AsyncWinWaitDefinition
{
    using Handle                    = FileDescriptor::Handle;  // fd
    static constexpr Handle Invalid = FileDescriptor::Invalid; // invalid fd
 
    static Result releaseHandle(Handle& waitHandle);
};
struct WinWaitHandle : public UniqueHandle<AsyncWinWaitDefinition>
{
};
} // namespace detail
} // namespace SC
 
 
struct SC::AsyncRequest
{
    AsyncRequest* next = nullptr;
    AsyncRequest* prev = nullptr;
 
    void setDebugName(const char* newDebugName);
 
    [[nodiscard]] AsyncEventLoop* getEventLoop() const { return eventLoop; }
 
    void cacheInternalEventLoop(AsyncEventLoop& loop) { eventLoop = &loop; }
 
    Result setThreadPoolAndTask(ThreadPool& pool, AsyncTask& task);
 
    void resetThreadPoolAndTask();
 
    enum class Type : uint8_t
    {
        LoopTimeout,   
        LoopWakeUp,    
        LoopWork,      
        ProcessExit,   
        SocketAccept,  
        SocketConnect, 
        SocketSend,    
        SocketReceive, 
        SocketClose,   
        FileRead,      
        FileWrite,     
        FileClose,     
        FilePoll,      
    };
 
    AsyncRequest(Type type) : state(State::Free), type(type), flags(0), eventIndex(-1) {}
 
    [[nodiscard]] Result stop(Function<void(AsyncResult&)>* afterStopped = nullptr);
 
    [[nodiscard]] bool isFree() const;
 
    [[nodiscard]] bool isCancelling() const;
 
    [[nodiscard]] bool isActive() const;
 
    [[nodiscard]] Type getType() const { return type; }
 
  protected:
    [[nodiscard]] Result validateAsync();
 
    void queueSubmission(AsyncEventLoop& eventLoop);
 
    AsyncEventLoop* eventLoop = nullptr;
    AsyncTask*      asyncTask = nullptr;
 
  private:
    Function<void(AsyncResult&)>* closeCallback = nullptr;
 
    friend struct AsyncEventLoop;
    friend struct AsyncResult;
 
    void markAsFree();
 
    [[nodiscard]] static const char* TypeToString(Type type);
    enum class State : uint8_t
    {
        Free,       // not in any queue, this can be started with an async.start(...)
        Setup,      // when in submission queue waiting to be setup (after an async.start(...))
        Submitting, // when in submission queue waiting to be activated or re-activated
        Active,     // when monitored by OS syscall or in activeLoopWakeUps / activeTimeouts queues
        Reactivate, // when flagged for reactivation inside the callback (after a result.reactivateRequest(true))
        Cancelling, // when in cancellation queue waiting for a cancelAsync (on active async)
    };
 
#if SC_CONFIGURATION_DEBUG
    const char* debugName = "None";
#endif
    State   state;      // 1 byte
    Type    type;       // 1 byte
    int16_t flags;      // 2 bytes
    int32_t eventIndex; // 4 bytes
};
 
struct SC::AsyncCompletionData
{
};
 
struct SC::AsyncResult
{
    AsyncResult(AsyncRequest& request, SC::Result&& res) : async(request), returnCode(move(res)) {}
 
    AsyncResult(AsyncRequest& request) : async(request) {}
 
    void reactivateRequest(bool value);
 
    [[nodiscard]] const SC::Result& isValid() const { return returnCode; }
 
    AsyncRequest& async;
 
  protected:
    friend struct AsyncEventLoop;
 
    bool shouldBeReactivated = false;
    bool shouldCallCallback  = true;
 
    SC::Result returnCode = SC::Result(true);
};
 
template <typename T, typename C>
struct SC::AsyncResultOf : public AsyncResult
{
    T&       getAsync() { return static_cast<T&>(AsyncResult::async); }
    const T& getAsync() const { return static_cast<const T&>(AsyncResult::async); }
 
    using AsyncResult::AsyncResult;
 
    C completionData;
};
 
struct SC::AsyncTask
{
    AsyncTask(AsyncCompletionData& asyncCompletionData) : completionData(asyncCompletionData) {}
 
  protected:
    ThreadPoolTask task;
    ThreadPool*    threadPool = nullptr;
 
    void freeTask() { async = nullptr; }
    bool isFree() const { return async == nullptr; }
 
    friend struct AsyncEventLoop;
    friend struct AsyncRequest;
 
    AsyncCompletionData& completionData;
 
    SC::Result    returnCode = SC::Result(true);
    AsyncRequest* async      = nullptr;
};
 
template <typename AsyncType>
struct SC::AsyncTaskOf : public AsyncTask
{
    typename AsyncType::CompletionData asyncCompletionData;
    AsyncTaskOf() : AsyncTask(asyncCompletionData) {}
};
 
namespace SC
{
 
struct AsyncLoopTimeout : public AsyncRequest
{
    AsyncLoopTimeout() : AsyncRequest(Type::LoopTimeout) {}
 
    using CompletionData = AsyncCompletionData;
 
    using Result = AsyncResultOf<AsyncLoopTimeout, CompletionData>;
 
    SC::Result start(AsyncEventLoop& eventLoop, Time::Milliseconds relativeTimeout);
 
    SC::Result start(AsyncEventLoop& eventLoop);
 
    Function<void(Result&)> callback; 
 
    Time::Milliseconds relativeTimeout; 
 
    Time::Absolute getExpirationTime() const { return expirationTime; }
 
  private:
    friend struct AsyncEventLoop;
    Time::Absolute expirationTime;
};
 
struct AsyncLoopWakeUp : public AsyncRequest
{
    AsyncLoopWakeUp() : AsyncRequest(Type::LoopWakeUp) {}
 
    using CompletionData = AsyncCompletionData;
 
    using Result = AsyncResultOf<AsyncLoopWakeUp, CompletionData>;
 
    SC::Result start(AsyncEventLoop& eventLoop, EventObject* eventObject = nullptr);
 
    SC::Result wakeUp();
 
    Function<void(Result&)> callback; 
 
  private:
    friend struct AsyncEventLoop;
 
    EventObject* eventObject = nullptr;
    Atomic<bool> pending     = false;
};
 
struct AsyncLoopWork : public AsyncRequest
{
    AsyncLoopWork() : AsyncRequest(Type::LoopWork) {}
 
    using CompletionData = AsyncCompletionData;
 
    using Result = AsyncResultOf<AsyncLoopWork, CompletionData>;
 
    SC::Result setThreadPool(ThreadPool& threadPool);
 
    SC::Result start(AsyncEventLoop& eventLoop);
 
    Function<SC::Result()>  work;     
    Function<void(Result&)> callback; 
 
  private:
    AsyncTaskOf<AsyncLoopWork> task;
};
 
struct AsyncProcessExit : public AsyncRequest
{
    AsyncProcessExit() : AsyncRequest(Type::ProcessExit) {}
 
    struct CompletionData : public AsyncCompletionData
    {
        ProcessDescriptor::ExitStatus exitStatus;
    };
 
    struct Result : public AsyncResultOf<AsyncProcessExit, CompletionData>
    {
        using AsyncResultOf<AsyncProcessExit, CompletionData>::AsyncResultOf;
 
        SC::Result get(ProcessDescriptor::ExitStatus& status)
        {
            status = completionData.exitStatus;
            return returnCode;
        }
    };
 
    SC::Result start(AsyncEventLoop& eventLoop, ProcessDescriptor::Handle process);
 
    Function<void(Result&)> callback; 
 
  private:
    friend struct AsyncEventLoop;
    ProcessDescriptor::Handle handle = ProcessDescriptor::Invalid;
#if SC_PLATFORM_WINDOWS
    detail::WinOverlappedOpaque overlapped;
    detail::WinWaitHandle       waitHandle;
#elif SC_PLATFORM_LINUX
    FileDescriptor     pidFd;
#endif
};
 
struct AsyncSocketAccept : public AsyncRequest
{
    AsyncSocketAccept() : AsyncRequest(Type::SocketAccept) {}
 
    struct CompletionData : public AsyncCompletionData
    {
        SocketDescriptor acceptedClient;
    };
 
    struct Result : public AsyncResultOf<AsyncSocketAccept, CompletionData>
    {
        using AsyncResultOf<AsyncSocketAccept, CompletionData>::AsyncResultOf;
 
        SC::Result moveTo(SocketDescriptor& client)
        {
            SC_TRY(returnCode);
            return client.assign(move(completionData.acceptedClient));
        }
    };
 
    SC::Result start(AsyncEventLoop& eventLoop, const SocketDescriptor& socketDescriptor);
 
    Function<void(Result&)> callback; 
 
  private:
    friend struct AsyncEventLoop;
    SocketDescriptor::Handle   handle        = SocketDescriptor::Invalid;
    SocketFlags::AddressFamily addressFamily = SocketFlags::AddressFamilyIPV4;
#if SC_PLATFORM_WINDOWS
    void (*pAcceptEx)() = nullptr;
    detail::WinOverlappedOpaque overlapped;
    SocketDescriptor            clientSocket;
    uint8_t                     acceptBuffer[288] = {0};
#elif SC_PLATFORM_LINUX
    AlignedStorage<28> sockAddrHandle;
    uint32_t           sockAddrLen;
#endif
};
 
struct AsyncSocketConnect : public AsyncRequest
{
    AsyncSocketConnect() : AsyncRequest(Type::SocketConnect) {}
 
    using CompletionData = AsyncCompletionData;
 
    using Result = AsyncResultOf<AsyncSocketConnect, CompletionData>;
 
    SC::Result start(AsyncEventLoop& eventLoop, const SocketDescriptor& descriptor, SocketIPAddress address);
 
    Function<void(Result&)> callback; 
 
  private:
    friend struct AsyncEventLoop;
    SocketDescriptor::Handle handle = SocketDescriptor::Invalid;
    SocketIPAddress          ipAddress;
#if SC_PLATFORM_WINDOWS
    void (*pConnectEx)() = nullptr;
    detail::WinOverlappedOpaque overlapped;
#endif
};
 
struct AsyncSocketSend : public AsyncRequest
{
    AsyncSocketSend() : AsyncRequest(Type::SocketSend) {}
 
    struct CompletionData : public AsyncCompletionData
    {
        size_t numBytes = 0;
    };
 
    using Result = AsyncResultOf<AsyncSocketSend, CompletionData>;
 
    SC::Result start(AsyncEventLoop& eventLoop, const SocketDescriptor& descriptor, Span<const char> data);
 
    SC::Result start(AsyncEventLoop& eventLoop, const SocketDescriptor& descriptor, Span<Span<const char>> data);
 
    SC::Result start(AsyncEventLoop& eventLoop);
 
    Function<void(Result&)> callback; 
 
    SocketDescriptor::Handle handle = SocketDescriptor::Invalid; 
 
    Span<const char>       buffer;              
    Span<Span<const char>> buffers;             
    bool                   singleBuffer = true; 
 
  private:
    friend struct AsyncEventLoop;
 
    size_t totalBytesWritten = 0;
#if SC_PLATFORM_WINDOWS
    detail::WinOverlappedOpaque overlapped;
#endif
};
 
struct AsyncSocketReceive : public AsyncRequest
{
    AsyncSocketReceive() : AsyncRequest(Type::SocketReceive) {}
 
    struct CompletionData : public AsyncCompletionData
    {
        size_t numBytes     = 0;
        bool   disconnected = false;
    };
 
    struct Result : public AsyncResultOf<AsyncSocketReceive, CompletionData>
    {
        using AsyncResultOf<AsyncSocketReceive, CompletionData>::AsyncResultOf;
 
        SC::Result get(Span<char>& outData)
        {
            SC_TRY(getAsync().buffer.sliceStartLength(0, completionData.numBytes, outData));
            return returnCode;
        }
    };
 
    SC::Result start(AsyncEventLoop& eventLoop, const SocketDescriptor& descriptor, Span<char> data);
 
    SC::Result start(AsyncEventLoop& eventLoop);
 
    Function<void(Result&)> callback; 
 
    Span<char>               buffer; 
    SocketDescriptor::Handle handle = SocketDescriptor::Invalid; 
 
  private:
#if SC_PLATFORM_WINDOWS
    friend struct AsyncEventLoop;
    detail::WinOverlappedOpaque overlapped;
#endif
};
 
struct AsyncSocketClose : public AsyncRequest
{
    AsyncSocketClose() : AsyncRequest(Type::SocketClose) {}
 
    using CompletionData = AsyncCompletionData;
 
    using Result = AsyncResultOf<AsyncSocketClose, CompletionData>;
 
    SC::Result start(AsyncEventLoop& eventLoop, const SocketDescriptor& descriptor);
 
    // TODO: Move code to CompletionData
    int code = 0; 
 
    Function<void(Result&)> callback; 
 
  private:
    friend struct AsyncEventLoop;
 
    SocketDescriptor::Handle handle = SocketDescriptor::Invalid;
};
 
struct AsyncFileRead : public AsyncRequest
{
    AsyncFileRead() : AsyncRequest(Type::FileRead) { handle = FileDescriptor::Invalid; }
 
    struct CompletionData : public AsyncCompletionData
    {
        size_t numBytes  = 0;
        bool   endOfFile = false;
    };
 
    struct Result : public AsyncResultOf<AsyncFileRead, CompletionData>
    {
        using AsyncResultOf<AsyncFileRead, CompletionData>::AsyncResultOf;
 
        SC::Result get(Span<char>& data)
        {
            SC_TRY(getAsync().buffer.sliceStartLength(0, completionData.numBytes, data));
            return returnCode;
        }
    };
 
    using Task = AsyncTaskOf<AsyncFileRead>;
 
    [[nodiscard]] SC::Result start(AsyncEventLoop& eventLoop);
 
    Function<void(Result&)> callback; 
 
    Span<char>             buffer; 
    FileDescriptor::Handle handle; 
 
    uint64_t getOffset() const { return offset; }
 
    void setOffset(uint64_t fileOffset)
    {
        useOffset = true;
        offset    = fileOffset;
    }
 
  private:
    friend struct AsyncEventLoop;
    bool     useOffset = false;
    uint64_t offset    = 0; 
#if SC_PLATFORM_WINDOWS
    uint64_t                    readCursor = 0;
    detail::WinOverlappedOpaque overlapped;
#endif
};
 
struct AsyncFileWrite : public AsyncRequest
{
    AsyncFileWrite() : AsyncRequest(Type::FileWrite) { handle = FileDescriptor::Invalid; }
 
    struct CompletionData : public AsyncCompletionData
    {
        size_t numBytes = 0;
    };
 
    struct Result : public AsyncResultOf<AsyncFileWrite, CompletionData>
    {
        using AsyncResultOf<AsyncFileWrite, CompletionData>::AsyncResultOf;
 
        SC::Result get(size_t& writtenSizeInBytes)
        {
            writtenSizeInBytes = completionData.numBytes;
            return returnCode;
        }
    };
 
    using Task = AsyncTaskOf<AsyncFileWrite>;
 
    SC::Result start(AsyncEventLoop& eventLoop);
 
    SC::Result start(AsyncEventLoop& eventLoop, Span<Span<const char>> data);
 
    SC::Result start(AsyncEventLoop& eventLoop, Span<const char> data);
 
    Function<void(Result&)> callback; 
 
    FileDescriptor::Handle handle; 
 
    Span<const char>       buffer;              
    Span<Span<const char>> buffers;             
    bool                   singleBuffer = true; 
 
    uint64_t getOffset() const { return offset; }
 
    void setOffset(uint64_t fileOffset)
    {
        useOffset = true;
        offset    = fileOffset;
    }
 
  private:
    friend struct AsyncEventLoop;
    bool     isWatchable = false;
    bool     useOffset   = false;
    uint64_t offset      = 0xffffffffffffffff; 
 
    size_t totalBytesWritten = 0;
#if SC_PLATFORM_WINDOWS
    detail::WinOverlappedOpaque overlapped;
#endif
};
 
struct AsyncFileClose : public AsyncRequest
{
    AsyncFileClose() : AsyncRequest(Type::FileClose) {}
 
    using CompletionData = AsyncCompletionData;
 
    using Result = AsyncResultOf<AsyncFileClose, CompletionData>;
 
    SC::Result start(AsyncEventLoop& eventLoop, FileDescriptor::Handle fileDescriptor);
 
    // TODO: Move code to CompletionData
    int code = 0; 
 
    Function<void(Result&)> callback; 
 
  private:
    friend struct AsyncEventLoop;
    FileDescriptor::Handle handle = FileDescriptor::Invalid;
};
 
struct AsyncFilePoll : public AsyncRequest
{
    AsyncFilePoll() : AsyncRequest(Type::FilePoll) {}
 
    using CompletionData = AsyncCompletionData;
 
    using Result = AsyncResultOf<AsyncFilePoll, CompletionData>;
 
    SC::Result start(AsyncEventLoop& loop, FileDescriptor::Handle fileDescriptor);
 
#if SC_PLATFORM_WINDOWS
    [[nodiscard]] auto& getOverlappedOpaque() { return overlapped; }
#endif
 
    Function<void(Result&)> callback;
 
  private:
    friend struct AsyncEventLoop;
 
    FileDescriptor::Handle handle = FileDescriptor::Invalid;
#if SC_PLATFORM_WINDOWS
    detail::WinOverlappedOpaque overlapped;
#endif
};
 
 
} // namespace SC
 
struct SC::AsyncKernelEvents
{
    Span<uint8_t> eventsMemory; 
 
  private:
    int numberOfEvents = 0;
    friend struct AsyncEventLoop;
};
 
struct SC::AsyncEventLoopListeners
{
    Function<void(AsyncEventLoop&)> beforeBlockingPoll;
    Function<void(AsyncEventLoop&)> afterBlockingPoll;
};
 
struct SC::AsyncEventLoop
{
    struct Options
    {
        enum class ApiType : uint8_t
        {
            Automatic = 0,   
            ForceUseIOURing, 
            ForceUseEpoll,   
        };
        ApiType apiType; 
 
        Options() { apiType = ApiType::Automatic; }
    };
 
    AsyncEventLoop();
 
    [[nodiscard]] Result create(Options options = Options());
 
    [[nodiscard]] Result close();
 
    void interrupt();
 
    [[nodiscard]] bool isInitialized() const;
 
    [[nodiscard]] Result run();
 
    [[nodiscard]] Result runOnce();
 
    [[nodiscard]] Result runNoWait();
 
    [[nodiscard]] Result submitRequests(AsyncKernelEvents& kernelEvents);
 
    [[nodiscard]] Result blockingPoll(AsyncKernelEvents& kernelEvents);
 
    [[nodiscard]] Result dispatchCompletions(AsyncKernelEvents& kernelEvents);
 
    [[nodiscard]] Result wakeUpFromExternalThread(AsyncLoopWakeUp& wakeUp);
 
    [[nodiscard]] Result wakeUpFromExternalThread();
 
    [[nodiscard]] Result createAsyncTCPSocket(SocketFlags::AddressFamily family, SocketDescriptor& outDescriptor);
 
    [[nodiscard]] Result associateExternallyCreatedTCPSocket(SocketDescriptor& outDescriptor);
 
    [[nodiscard]] Result associateExternallyCreatedFileDescriptor(FileDescriptor& outDescriptor);
 
    void updateTime();
 
    [[nodiscard]] Time::Monotonic getLoopTime() const;
 
    [[nodiscard]] int getNumberOfActiveRequests() const;
 
    [[nodiscard]] int getNumberOfSubmittedRequests() const;
 
    [[nodiscard]] AsyncLoopTimeout* findEarliestLoopTimeout() const;
 
    void excludeFromActiveCount(AsyncRequest& async);
 
    void includeInActiveCount(AsyncRequest& async);
 
    void enumerateRequests(Function<void(AsyncRequest&)> enumerationCallback);
 
    void setListeners(AsyncEventLoopListeners* listeners);
 
    [[nodiscard]] static bool isExcludedFromActiveCount(const AsyncRequest& async);
 
    [[nodiscard]] static bool tryLoadingLiburing();
 
    struct Internal;
 
  private:
    struct InternalDefinition
    {
        static constexpr int Windows = 488;
        static constexpr int Apple   = 480;
        static constexpr int Linux   = 688;
        static constexpr int Default = Linux;
 
        static constexpr size_t Alignment = 8;
 
        using Object = Internal;
    };
 
  public:
    using InternalOpaque = OpaqueObject<InternalDefinition>;
 
  private:
    InternalOpaque internalOpaque;
    Internal&      internal;
 
    friend struct AsyncRequest;
    friend struct AsyncFileWrite;
    friend struct AsyncFileRead;
};
 
struct SC::AsyncEventLoopMonitor
{
    Function<void(void)> onNewEventsAvailable; 
 
    Result create(AsyncEventLoop& loop);
 
    Result close();
 
    Result startMonitoring();
 
    Result stopMonitoringAndDispatchCompletions();
 
  private:
#if SC_COMPILER_MSVC
#pragma warning(push)
#pragma warning(disable : 4324) // useless warning on 32 bit... (structure was padded due to __declspec(align()))
#endif
    alignas(uint64_t) uint8_t eventsMemory[8 * 1024]; // 8 Kb of kernel events
#if SC_COMPILER_MSVC
#pragma warning(pop)
#endif
 
    AsyncKernelEvents asyncKernelEvents;
    AsyncEventLoop*   eventLoop = nullptr;
    AsyncLoopWakeUp   eventLoopWakeUp;
 
    Thread      eventLoopThread;
    EventObject eventObjectEnterBlockingMode;
    EventObject eventObjectExitBlockingMode;
 
    Atomic<bool> finished    = false;
    Atomic<bool> needsWakeUp = true;
 
    bool wakeUpHasBeenCalled = false;
 
    Result monitoringLoopThread(Thread& thread);
};