Async.h

// Copyright (c) Stefano Cristiano
// SPDX-License-Identifier: MIT
#pragma once
 
#include "../Common/CompilerMacrosExport.h"
#ifndef SC_EXPORT_LIBRARY_ASYNC
#define SC_EXPORT_LIBRARY_ASYNC 0
#endif
#define SC_ASYNC_EXPORT SC_COMPILER_LIBRARY_EXPORT(SC_EXPORT_LIBRARY_ASYNC)
 
#include "../Async/Internal/IntrusiveDoubleLinkedList.h"
#include "../Common/Assert.h"
#include "../Common/Function.h"
#include "../Common/OpaqueObject.h"
#include "../File/File.h"
#include "../FileSystem/FileSystem.h"
#include "../Socket/Socket.h"
#include "../Threading/Atomic.h"
#include "../Threading/ThreadPool.h"
 
namespace SC
{
SC_DECLARE_ASSERT_PROVIDER(AsyncAssert, SC_ASYNC_EXPORT);
 
#define SC_ASYNC_ASSERT_RELEASE(e)        SC_ASSERT_PROVIDER_RELEASE(SC::AsyncAssert, e)
#define SC_ASYNC_ASSERT_DEBUG(e)          SC_ASSERT_PROVIDER_DEBUG(SC::AsyncAssert, e)
#define SC_ASYNC_TRUST_RESULT(expression) SC_ASYNC_ASSERT_RELEASE(expression)
 
struct ThreadPool;
struct ThreadPoolTask;
struct EventObject;
} // namespace SC
 
namespace SC
{
struct AsyncEventLoop;
struct AsyncResult;
struct AsyncSequence;
struct AsyncTaskSequence;
 
enum class AsyncThreadPoolMode : uint8_t
{
    NativePreferred, 
    ForceThreadPool, 
};
 
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 SC_ASYNC_EXPORT WinWaitHandle : public UniqueHandle<AsyncWinWaitDefinition>
{
};
} // namespace detail
 
struct SC_ASYNC_EXPORT AsyncRequest
{
    AsyncRequest* next = nullptr;
    AsyncRequest* prev = nullptr;
 
    void setDebugName(const char* newDebugName);
 
    void executeOn(AsyncSequence& sequence);
 
    Result executeOn(AsyncTaskSequence& task, ThreadPool& pool,
                     AsyncThreadPoolMode mode = AsyncThreadPoolMode::NativePreferred);
 
    void disableThreadPool();
 
    enum class Type : uint8_t
    {
        LoopTimeout,         
        LoopWakeUp,          
        LoopWork,            
        ProcessExit,         
        Signal,              
        SocketAccept,        
        SocketConnect,       
        SocketSend,          
        SocketSendTo,        
        SocketReceive,       
        SocketReceiveFrom,   
        FileRead,            
        FileWrite,           
        FileSend,            
        FileReadiness,       
        ExternalCompletion,  
        FileSystemOperation, 
    };
 
    AsyncRequest(Type type) : state(State::Free), type(type), flags(0), unused(0), userFlags(0) {}
 
    Result stop(AsyncEventLoop& eventLoop, Function<void(AsyncResult&)>* afterStopped = nullptr);
 
    [[nodiscard]] bool isFree() const;
 
    [[nodiscard]] bool isCancelling() const;
 
    [[nodiscard]] bool isActive() const;
 
    [[nodiscard]] Type getType() const { return type; }
 
    Result start(AsyncEventLoop& eventLoop);
 
    void setUserFlags(uint16_t externalFlags) { userFlags = externalFlags; }
 
    uint16_t getUserFlags() const { return userFlags; }
 
    [[nodiscard]] Function<void(AsyncResult&)>* getCloseCallback() { return closeCallback; }
 
    [[nodiscard]] const Function<void(AsyncResult&)>* getCloseCallback() const { return closeCallback; }
 
  protected:
    Result checkState();
 
    void queueSubmission(AsyncEventLoop& eventLoop);
 
    AsyncSequence* sequence = nullptr;
 
    AsyncTaskSequence* getTask();
 
    [[nodiscard]] bool isThreadPoolForced() const;
 
  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_ASYNC_ENABLE_LOG
    const char* debugName = "None";
#endif
    State   state; // 1 byte
    Type    type;  // 1 byte
    int16_t flags; // 2 bytes
 
    uint16_t unused;    // 2 bytes
    uint16_t userFlags; // 2 bytes
};
 
struct SC_ASYNC_EXPORT AsyncSequence
{
    AsyncSequence* next = nullptr;
    AsyncSequence* prev = nullptr;
 
    bool clearSequenceOnCancel = true; 
    bool clearSequenceOnError  = true; 
  private:
    friend struct AsyncEventLoop;
    bool runningAsync = false; // true if an async from this sequence is being run
    bool tracked      = false;
 
    AsyncRequest* runningRequest = nullptr;
 
    IntrusiveDoubleLinkedList<AsyncRequest> submissions;
};
 
struct SC_ASYNC_EXPORT AsyncCompletionData
{
};
 
struct SC_ASYNC_EXPORT AsyncResult
{
    AsyncResult(AsyncEventLoop& eventLoop, AsyncRequest& request, SC::Result& res, bool* hasBeenReactivated = nullptr)
        : eventLoop(eventLoop), async(request), hasBeenReactivated(hasBeenReactivated), returnCode(res)
    {}
 
    void reactivateRequest(bool shouldBeReactivated);
 
    [[nodiscard]] const SC::Result& isValid() const { return returnCode; }
 
    AsyncEventLoop& eventLoop;
    AsyncRequest&   async;
 
  protected:
    friend struct AsyncEventLoop;
 
    bool  shouldCallCallback = true;
    bool* hasBeenReactivated = nullptr;
 
    SC::Result& returnCode;
};
 
template <typename T, typename C>
struct 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;
    int32_t eventIndex = 0;
};
 
struct SC_ASYNC_EXPORT AsyncLoopTimeout : public AsyncRequest
{
    AsyncLoopTimeout() : AsyncRequest(Type::LoopTimeout) {}
 
    using CompletionData = AsyncCompletionData;
    using Result         = AsyncResultOf<AsyncLoopTimeout, CompletionData>;
    using AsyncRequest::start;
 
    SC::Result start(AsyncEventLoop& eventLoop, TimeMs relativeTimeout);
 
    Function<void(Result&)> callback; 
 
    TimeMs relativeTimeout; 
 
    TimeMs getExpirationTime() const { return expirationTime; }
 
  private:
    SC::Result validate(AsyncEventLoop&);
    friend struct AsyncEventLoop;
    TimeMs expirationTime;
};
 
struct AsyncLoopWakeUpOptions
{
    bool coalesce = true;
};
 
struct SC_ASYNC_EXPORT AsyncLoopWakeUp : public AsyncRequest
{
    AsyncLoopWakeUp() : AsyncRequest(Type::LoopWakeUp) {}
 
    struct CompletionData : public AsyncCompletionData
    {
        uint32_t deliveryCount = 1; 
    };
 
    using Result = AsyncResultOf<AsyncLoopWakeUp, CompletionData>;
 
    SC::Result start(AsyncEventLoop& eventLoop, AsyncLoopWakeUpOptions options = {});
 
    SC::Result start(AsyncEventLoop& eventLoop, EventObject& eventObject, AsyncLoopWakeUpOptions options = {});
 
    SC::Result wakeUp(AsyncEventLoop& eventLoop);
 
    Function<void(Result&)> callback; 
    EventObject* eventObject = nullptr; 
 
  private:
    friend struct AsyncEventLoop;
    SC::Result validate(AsyncEventLoop&);
    int32_t    consumePendingWakeUps();
    int32_t    getPendingWakeUps() const;
 
    AsyncLoopWakeUpOptions wakeUpOptions;
    Atomic<int32_t>        pendingWakeUps = 0;
};
 
struct SC_ASYNC_EXPORT AsyncProcessExit : public AsyncRequest
{
    AsyncProcessExit() : AsyncRequest(Type::ProcessExit) {}
 
    struct CompletionData : public AsyncCompletionData
    {
        int exitStatus;
    };
 
    struct Result : public AsyncResultOf<AsyncProcessExit, CompletionData>
    {
        using AsyncResultOf<AsyncProcessExit, CompletionData>::AsyncResultOf;
 
        SC::Result get(int& status)
        {
            status = completionData.exitStatus;
            return returnCode;
        }
    };
    using AsyncRequest::start;
 
    SC::Result start(AsyncEventLoop& eventLoop, FileDescriptor::Handle process);
 
    Function<void(Result&)> callback; 
 
  private:
    friend struct AsyncEventLoop;
    SC::Result validate(AsyncEventLoop&);
 
    FileDescriptor::Handle handle = FileDescriptor::Invalid;
#if SC_PLATFORM_WINDOWS
    detail::WinOverlappedOpaque overlapped;
    detail::WinWaitHandle       waitHandle;
    AsyncEventLoop*             eventLoop = nullptr;
#elif SC_PLATFORM_LINUX
    FileDescriptor pidFd;
#endif
};
 
struct AsyncSignalOptions
{
    enum class Mode : uint8_t
    {
        Persistent, 
        OneShot     
    };
    Mode mode     = Mode::Persistent; 
    bool coalesce = true;             
};
 
struct SC_ASYNC_EXPORT AsyncSignal : public AsyncRequest
{
    AsyncSignal() : AsyncRequest(Type::Signal) {}
 
    struct CompletionData : public AsyncCompletionData
    {
        int      signalNumber  = 0; 
        uint32_t deliveryCount = 1; 
    };
 
    struct Result : public AsyncResultOf<AsyncSignal, CompletionData>
    {
        using AsyncResultOf<AsyncSignal, CompletionData>::AsyncResultOf;
    };
    using AsyncRequest::start;
 
    SC::Result start(AsyncEventLoop& eventLoop, int num, AsyncSignalOptions options = {});
 
    Function<void(Result&)> callback; 
 
  private:
    friend struct AsyncEventLoop;
    SC::Result validate(AsyncEventLoop&);
 
    int                signalNumber = 0;
    AsyncSignalOptions signalOptions;
#if SC_PLATFORM_WINDOWS
    detail::WinOverlappedOpaque overlapped;
    AsyncEventLoop*             eventLoop = nullptr;
#elif SC_PLATFORM_LINUX
    FileDescriptor         signalFd;
    FileDescriptor::Handle signalFdHandle = FileDescriptor::Invalid;
#endif
};
 
struct AsyncSocketAccept;
namespace detail
{
struct SC_ASYNC_EXPORT AsyncSocketAcceptData
{
#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 SC_ASYNC_EXPORT AsyncSocketAcceptBase : public AsyncRequest
{
    AsyncSocketAcceptBase() : 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 SC::Result(client.assign(move(completionData.acceptedClient)));
        }
    };
    using AsyncRequest::start;
 
    SC::Result start(AsyncEventLoop& eventLoop, const SocketDescriptor& socketDescriptor, AsyncSocketAcceptData& data);
    SC::Result validate(AsyncEventLoop&);
 
    Function<void(Result&)>    callback; 
    SocketDescriptor::Handle   handle        = SocketDescriptor::Invalid;
    SocketFlags::AddressFamily addressFamily = SocketFlags::AddressFamilyIPV4;
    AsyncSocketAcceptData*     acceptData    = nullptr;
};
 
} // namespace detail
 
struct SC_ASYNC_EXPORT AsyncSocketAccept : public detail::AsyncSocketAcceptBase
{
    AsyncSocketAccept() { AsyncSocketAcceptBase::acceptData = &data; }
    using AsyncSocketAcceptBase::start;
 
    SC::Result start(AsyncEventLoop& eventLoop, const SocketDescriptor& socketDescriptor);
 
  private:
    detail::AsyncSocketAcceptData data;
};
 
struct SC_ASYNC_EXPORT AsyncSocketConnect : public AsyncRequest
{
    AsyncSocketConnect() : AsyncRequest(Type::SocketConnect) {}
 
    using CompletionData = AsyncCompletionData;
    using Result         = AsyncResultOf<AsyncSocketConnect, CompletionData>;
    using AsyncRequest::start;
 
    SC::Result start(AsyncEventLoop& eventLoop, const SocketDescriptor& descriptor, SocketIPAddress address);
 
    Function<void(Result&)> callback; 
 
    SocketDescriptor::Handle handle = SocketDescriptor::Invalid;
    SocketIPAddress          ipAddress;
 
  private:
    friend struct AsyncEventLoop;
    SC::Result validate(AsyncEventLoop&);
 
#if SC_PLATFORM_WINDOWS
    void (*pConnectEx)() = nullptr;
    detail::WinOverlappedOpaque overlapped;
#endif
};
 
struct SC_ASYNC_EXPORT AsyncSocketSend : public AsyncRequest
{
    AsyncSocketSend() : AsyncRequest(Type::SocketSend) {}
    struct CompletionData : public AsyncCompletionData
    {
        size_t numBytes = 0;
    };
    using Result = AsyncResultOf<AsyncSocketSend, CompletionData>;
    using AsyncRequest::start;
 
    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);
 
    Function<void(Result&)> callback; 
 
    SocketDescriptor::Handle handle = SocketDescriptor::Invalid; 
 
    SC::Result closeHandle() { return detail::SocketDescriptorDefinition::releaseHandle(handle); }
 
    Span<const char>       buffer;              
    Span<Span<const char>> buffers;             
    bool                   singleBuffer = true; 
 
  protected:
    AsyncSocketSend(Type type) : AsyncRequest(type) {}
    friend struct AsyncEventLoop;
    SC::Result validate(AsyncEventLoop&);
 
    size_t totalBytesWritten = 0;
#if SC_PLATFORM_WINDOWS
    detail::WinOverlappedOpaque overlapped;
#endif
};
 
struct SC_ASYNC_EXPORT AsyncSocketSendTo : public AsyncSocketSend
{
    AsyncSocketSendTo() : AsyncSocketSend(Type::SocketSendTo) {}
 
    SocketIPAddress address;
 
    SC::Result start(AsyncEventLoop& eventLoop, const SocketDescriptor& descriptor, SocketIPAddress ipAddress,
                     Span<const char> data);
 
    SC::Result start(AsyncEventLoop& eventLoop, const SocketDescriptor& descriptor, SocketIPAddress ipAddress,
                     Span<Span<const char>> data);
 
  private:
    using AsyncSocketSend::start;
    friend struct AsyncEventLoop;
    SC::Result validate(AsyncEventLoop&);
#if SC_PLATFORM_LINUX
    AlignedStorage<56> typeErasedMsgHdr;
#endif
};
 
struct SC_ASYNC_EXPORT 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;
 
        bool isEnded() const { return completionData.disconnected; }
 
        SC::Result get(Span<char>& outData)
        {
            SC_TRY(getAsync().buffer.sliceStartLength(0, completionData.numBytes, outData));
            return returnCode;
        }
 
        SocketIPAddress getSourceAddress() const;
    };
    using AsyncRequest::start;
 
    SC::Result start(AsyncEventLoop& eventLoop, const SocketDescriptor& descriptor, Span<char> data);
 
    Function<void(Result&)> callback; 
 
    Span<char>               buffer; 
    SocketDescriptor::Handle handle = SocketDescriptor::Invalid; 
 
    SC::Result closeHandle() { return detail::SocketDescriptorDefinition::releaseHandle(handle); }
 
  protected:
    AsyncSocketReceive(Type type) : AsyncRequest(type) {}
    friend struct AsyncEventLoop;
    SC::Result validate(AsyncEventLoop&);
#if SC_PLATFORM_WINDOWS
    detail::WinOverlappedOpaque overlapped;
#endif
};
 
struct SC_ASYNC_EXPORT AsyncSocketReceiveFrom : public AsyncSocketReceive
{
    AsyncSocketReceiveFrom() : AsyncSocketReceive(Type::SocketReceiveFrom) {}
    using AsyncSocketReceive::start;
 
  private:
    SocketIPAddress address;
    friend struct AsyncSocketReceive;
    friend struct AsyncEventLoop;
#if SC_PLATFORM_LINUX
    AlignedStorage<56> typeErasedMsgHdr;
#endif
};
 
struct SC_ASYNC_EXPORT 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;
 
        bool isEnded() const { return completionData.endOfFile; }
 
        SC::Result get(Span<char>& data)
        {
            SC_TRY(getAsync().buffer.sliceStartLength(0, completionData.numBytes, data));
            return returnCode;
        }
    };
    using AsyncRequest::start;
 
    SC::Result start(AsyncEventLoop& eventLoop, const FileDescriptor& descriptor, Span<char> data);
 
    Function<void(Result&)> callback; 
    Span<char>              buffer;   
    FileDescriptor::Handle  handle;   
 
    SC::Result closeHandle() { return detail::FileDescriptorDefinition::releaseHandle(handle); }
 
    uint64_t getOffset() const { return offset; }
 
    void setOffset(uint64_t fileOffset)
    {
        useOffset = true;
        offset    = fileOffset;
    }
 
  private:
    friend struct AsyncEventLoop;
    SC::Result validate(AsyncEventLoop&);
 
    bool useOffset = false;
    bool endedSync = false;
 
    uint64_t offset = 0; 
#if SC_PLATFORM_WINDOWS
    uint64_t                    readCursor = 0;
    detail::WinOverlappedOpaque overlapped;
#endif
};
 
struct SC_ASYNC_EXPORT 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 AsyncRequest::start;
 
    SC::Result start(AsyncEventLoop& eventLoop, const FileDescriptor& descriptor, Span<Span<const char>> data);
 
    SC::Result start(AsyncEventLoop& eventLoop, Span<Span<const char>> data);
 
    SC::Result start(AsyncEventLoop& eventLoop, const FileDescriptor& descriptor, Span<const char> data);
 
    SC::Result start(AsyncEventLoop& eventLoop, Span<const char> data);
 
    Function<void(Result&)> callback; 
 
    FileDescriptor::Handle handle; 
 
    SC::Result closeHandle() { return detail::FileDescriptorDefinition::releaseHandle(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;
    SC::Result validate(AsyncEventLoop&);
 
#if SC_PLATFORM_WINDOWS
    bool endedSync = false;
#else
    bool isWatchable = false;
#endif
    bool     useOffset = false;
    uint64_t offset    = 0xffffffffffffffff; 
 
    size_t totalBytesWritten = 0;
#if SC_PLATFORM_WINDOWS
    detail::WinOverlappedOpaque overlapped;
#endif
};
 
struct SC_ASYNC_EXPORT AsyncFileReadiness : public AsyncRequest
{
    AsyncFileReadiness() : AsyncRequest(Type::FileReadiness) {}
 
    using CompletionData = AsyncCompletionData;
    using Result         = AsyncResultOf<AsyncFileReadiness, CompletionData>;
 
    SC::Result start(AsyncEventLoop& eventLoop, FileDescriptor::Handle fileDescriptor);
 
    Function<void(Result&)> callback;
 
  private:
    friend struct AsyncEventLoop;
    SC::Result validate(AsyncEventLoop&);
 
    FileDescriptor::Handle handle = FileDescriptor::Invalid;
};
 
struct SC_ASYNC_EXPORT AsyncExternalCompletion : public AsyncRequest
{
    struct CompletionData
    {
        size_t bytesTransferred = 0;
    };
    using Result = AsyncResultOf<AsyncExternalCompletion, CompletionData>;
 
    AsyncExternalCompletion() : AsyncRequest(Type::ExternalCompletion) {}
 
    SC::Result start(AsyncEventLoop& eventLoop);
 
#if SC_PLATFORM_WINDOWS
    SC::Result start(AsyncEventLoop& eventLoop, FileDescriptor::Handle fileDescriptor);
 
    [[nodiscard]] void* getWindowsOverlapped();
#endif
 
    SC::Result markSubmissionPending();
 
    SC::Result clearSubmissionPending();
 
    [[nodiscard]] bool hasSubmissionPending() const { return submissionPending; }
 
    Function<void(Result&)> callback;
 
  private:
    friend struct AsyncEventLoop;
    SC::Result validate(AsyncEventLoop&);
 
    FileDescriptor::Handle handle            = FileDescriptor::Invalid;
    size_t                 bytesTransferred  = 0;
    bool                   manualMode        = true;
    bool                   submissionPending = false;
    bool                   completionPosted  = false;
#if SC_PLATFORM_WINDOWS
    detail::WinOverlappedOpaque overlapped;
#endif
};
 
struct SC_ASYNC_EXPORT AsyncFileSend : public AsyncRequest
{
    AsyncFileSend() : AsyncRequest(Type::FileSend) {}
 
    struct CompletionData : public AsyncCompletionData
    {
        size_t bytesTransferred = 0;     
        bool   usedZeroCopy     = false; 
    };
 
    struct Result : public AsyncResultOf<AsyncFileSend, CompletionData>
    {
        using AsyncResultOf<AsyncFileSend, CompletionData>::AsyncResultOf;
 
        [[nodiscard]] size_t getBytesTransferred() const { return completionData.bytesTransferred; }
 
        [[nodiscard]] bool usedZeroCopy() const { return completionData.usedZeroCopy; }
 
        [[nodiscard]] bool isComplete() const
        {
            return returnCode && completionData.bytesTransferred == getAsync().length;
        }
    };
 
    using AsyncRequest::start;
 
    SC::Result start(AsyncEventLoop& eventLoop, const FileDescriptor& file, const SocketDescriptor& socket,
                     int64_t offset = 0, size_t length = 0, size_t pipeSize = 0);
 
    Function<void(Result&)> callback; 
 
    // Internal handles (set by start())
    FileDescriptor::Handle   fileHandle   = FileDescriptor::Invalid;
    SocketDescriptor::Handle socketHandle = SocketDescriptor::Invalid;
 
    int64_t offset    = 0;
    size_t  length    = 0;
    size_t  bytesSent = 0; 
  private:
    friend struct AsyncEventLoop;
    SC::Result validate(AsyncEventLoop&);
 
#if SC_PLATFORM_WINDOWS
    detail::WinOverlappedOpaque overlapped;
#elif SC_PLATFORM_LINUX
    size_t         pipeBufferSize = 0; 
    PipeDescriptor splicePipe;         
#endif
};
 
// forward declared because it must be defined after AsyncTaskSequence
struct AsyncLoopWork;
struct AsyncFileSystemOperation;
 
struct SC_ASYNC_EXPORT AsyncFileSystemOperationCompletionData : public AsyncCompletionData
{
    FileDescriptor::Handle handle = FileDescriptor::Invalid; // for open
 
    int    code     = 0; // for open/close
    size_t numBytes = 0; // for read
};
 
namespace detail
{
// A simple hand-made variant of all completion types
struct SC_ASYNC_EXPORT AsyncCompletionVariant
{
    AsyncCompletionVariant() {}
    ~AsyncCompletionVariant() { destroy(); }
 
    AsyncCompletionVariant(const AsyncCompletionVariant&)            = delete;
    AsyncCompletionVariant(AsyncCompletionVariant&&)                 = delete;
    AsyncCompletionVariant& operator=(const AsyncCompletionVariant&) = delete;
    AsyncCompletionVariant& operator=(AsyncCompletionVariant&&)      = delete;
 
    bool inited = false;
 
    AsyncRequest::Type type;
    union
    {
        AsyncCompletionData completionDataLoopWork; // Defined after AsyncCompletionVariant / AsyncTaskSequence
        AsyncLoopTimeout::CompletionData        completionDataLoopTimeout;
        AsyncLoopWakeUp::CompletionData         completionDataLoopWakeUp;
        AsyncProcessExit::CompletionData        completionDataProcessExit;
        AsyncSignal::CompletionData             completionDataSignal;
        AsyncSocketAccept::CompletionData       completionDataSocketAccept;
        AsyncSocketConnect::CompletionData      completionDataSocketConnect;
        AsyncSocketSend::CompletionData         completionDataSocketSend;
        AsyncSocketSendTo::CompletionData       completionDataSocketSendTo;
        AsyncSocketReceive::CompletionData      completionDataSocketReceive;
        AsyncSocketReceiveFrom::CompletionData  completionDataSocketReceiveFrom;
        AsyncFileRead::CompletionData           completionDataFileRead;
        AsyncFileWrite::CompletionData          completionDataFileWrite;
        AsyncFileSend::CompletionData           completionDataFileSend;
        AsyncFileReadiness::CompletionData      completionDataFileReadiness;
        AsyncExternalCompletion::CompletionData completionDataExternalCompletion;
 
        AsyncFileSystemOperationCompletionData completionDataFileSystemOperation;
    };
 
    auto& getCompletion(AsyncLoopWork&) { return completionDataLoopWork; }
    auto& getCompletion(AsyncLoopTimeout&) { return completionDataLoopTimeout; }
    auto& getCompletion(AsyncLoopWakeUp&) { return completionDataLoopWakeUp; }
    auto& getCompletion(AsyncProcessExit&) { return completionDataProcessExit; }
    auto& getCompletion(AsyncSignal&) { return completionDataSignal; }
    auto& getCompletion(AsyncSocketAccept&) { return completionDataSocketAccept; }
    auto& getCompletion(AsyncSocketConnect&) { return completionDataSocketConnect; }
    auto& getCompletion(AsyncSocketSend&) { return completionDataSocketSend; }
    auto& getCompletion(AsyncSocketReceive&) { return completionDataSocketReceive; }
    auto& getCompletion(AsyncFileRead&) { return completionDataFileRead; }
    auto& getCompletion(AsyncFileWrite&) { return completionDataFileWrite; }
    auto& getCompletion(AsyncFileSend&) { return completionDataFileSend; }
    auto& getCompletion(AsyncFileReadiness&) { return completionDataFileReadiness; }
    auto& getCompletion(AsyncExternalCompletion&) { return completionDataExternalCompletion; }
    auto& getCompletion(AsyncFileSystemOperation&) { return completionDataFileSystemOperation; }
 
    template <typename T>
    auto& construct(T& t)
    {
        destroy();
        placementNew(getCompletion(t));
        inited = true;
        type   = t.getType();
        return getCompletion(t);
    }
    void destroy();
};
} // namespace detail
 
struct SC_ASYNC_EXPORT AsyncTaskSequence : public AsyncSequence
{
  protected:
    ThreadPoolTask task;
    ThreadPool*    threadPool = nullptr;
 
    friend struct AsyncEventLoop;
    friend struct AsyncRequest;
 
    detail::AsyncCompletionVariant completion;
 
    SC::Result returnCode = SC::Result(true);
};
 
struct SC_ASYNC_EXPORT AsyncLoopWork : public AsyncRequest
{
    AsyncLoopWork() : AsyncRequest(Type::LoopWork) {}
 
    using CompletionData = AsyncCompletionData;
    using Result         = AsyncResultOf<AsyncLoopWork, CompletionData>;
 
    SC::Result setThreadPool(ThreadPool& threadPool, AsyncThreadPoolMode mode = AsyncThreadPoolMode::NativePreferred);
 
    Function<SC::Result()>  work;     
    Function<void(Result&)> callback; 
 
  private:
    friend struct AsyncEventLoop;
    SC::Result        validate(AsyncEventLoop&);
    AsyncTaskSequence task;
};
 
struct SC_ASYNC_EXPORT AsyncFileSystemOperation : public AsyncRequest
{
    AsyncFileSystemOperation() : AsyncRequest(Type::FileSystemOperation) {}
    ~AsyncFileSystemOperation() { destroy(); }
#ifdef CopyFile
#undef CopyFile
#endif
#ifdef RemoveDirectory
#undef RemoveDirectory
#endif
    enum class Operation
    {
        None = 0,
        Open,
        Close,
        Read,
        Write,
        CopyFile,
        CopyDirectory,
        Rename,
        RemoveDirectory,
        RemoveFile,
    };
 
    using CompletionData = AsyncFileSystemOperationCompletionData;
    using Result         = AsyncResultOf<AsyncFileSystemOperation, CompletionData>;
 
    SC::Result setThreadPool(ThreadPool& threadPool, AsyncThreadPoolMode mode = AsyncThreadPoolMode::NativePreferred);
 
    SC::Result stop(AsyncEventLoop& eventLoop, Function<void(AsyncResult&)>* afterStopped = nullptr);
 
    Function<void(Result&)> callback; 
 
    SC::Result open(AsyncEventLoop& eventLoop, StringSpan path, FileOpen mode);
 
    SC::Result close(AsyncEventLoop& eventLoop, FileDescriptor::Handle handle);
 
    SC::Result read(AsyncEventLoop& eventLoop, FileDescriptor::Handle handle, Span<char> buffer, uint64_t offset);
 
    SC::Result write(AsyncEventLoop& eventLoop, FileDescriptor::Handle handle, Span<const char> buffer,
                     uint64_t offset);
 
    SC::Result copyFile(AsyncEventLoop& eventLoop, StringSpan path, StringSpan destinationPath,
                        FileSystemCopyFlags copyFlags = FileSystemCopyFlags());
 
    SC::Result copyDirectory(AsyncEventLoop& eventLoop, StringSpan path, StringSpan destinationPath,
                             FileSystemCopyFlags copyFlags = FileSystemCopyFlags());
 
    SC::Result rename(AsyncEventLoop& eventLoop, StringSpan path, StringSpan newPath);
 
    SC::Result removeEmptyDirectory(AsyncEventLoop& eventLoop, StringSpan path);
 
    SC::Result removeFile(AsyncEventLoop& eventLoop, StringSpan path);
 
  private:
    friend struct AsyncEventLoop;
    Operation           operation      = Operation::None;
    AsyncThreadPoolMode threadPoolMode = AsyncThreadPoolMode::NativePreferred;
    AsyncLoopWork       loopWork;
    CompletionData      completionData;
 
    void onOperationCompleted(AsyncLoopWork::Result& res);
 
    struct FileDescriptorData
    {
        FileDescriptor::Handle handle;
    };
 
    struct OpenData
    {
        StringSpan path;
        FileOpen   mode;
    };
 
    struct ReadData
    {
        FileDescriptor::Handle handle;
        Span<char>             buffer;
        uint64_t               offset;
    };
 
    struct WriteData
    {
        FileDescriptor::Handle handle;
        Span<const char>       buffer;
        uint64_t               offset;
    };
 
    struct CopyFileData
    {
        StringSpan          path;
        StringSpan          destinationPath;
        FileSystemCopyFlags copyFlags;
    };
 
    using CopyDirectoryData = CopyFileData;
 
    using CloseData = FileDescriptorData;
 
    struct RenameData
    {
        StringSpan path;
        StringSpan newPath;
    };
 
    struct RemoveData
    {
        StringSpan path;
    };
 
    union
    {
        OpenData          openData;
        CloseData         closeData;
        ReadData          readData;
        WriteData         writeData;
        CopyFileData      copyFileData;
        CopyDirectoryData copyDirectoryData;
        RenameData        renameData;
        RemoveData        removeData;
    };
 
    void destroy();
 
    SC::Result start(AsyncEventLoop& eventLoop, FileDescriptor::Handle fileDescriptor);
    SC::Result validate(AsyncEventLoop&);
};
 
struct SC_ASYNC_EXPORT AsyncKernelEvents
{
    Span<uint8_t> eventsMemory; 
 
  private:
    int numberOfEvents = 0;
    friend struct AsyncEventLoop;
    friend struct AsyncEventLoopMonitor;
};
 
struct SC_ASYNC_EXPORT AsyncEventLoopListeners
{
    Function<void(AsyncEventLoop&)> beforeBlockingPoll;
    Function<void(AsyncEventLoop&)> afterBlockingPoll;
};
 
struct SC_ASYNC_EXPORT AsyncEventLoop
{
    struct Options
    {
        enum class ApiType : uint8_t
        {
            Automatic = 0,   
            ForceUseIoUring, 
            ForceUseEpoll,   
        };
        ApiType apiType; 
 
        Options() { apiType = ApiType::Automatic; }
    };
 
    AsyncEventLoop();
 
    AsyncEventLoop(const AsyncEventLoop&)            = delete;
    AsyncEventLoop(AsyncEventLoop&&)                 = delete;
    AsyncEventLoop& operator=(AsyncEventLoop&&)      = delete;
    AsyncEventLoop& operator=(const AsyncEventLoop&) = delete;
 
    Result create(Options options = Options());
 
    Result close();
 
    Result start(AsyncRequest& async);
 
    Result postExternalCompletion(AsyncExternalCompletion& async, size_t bytesTransferred = 0);
 
    void interrupt();
 
    [[nodiscard]] bool isInitialized() const;
 
    [[nodiscard]] bool needsThreadPoolForFileOperations() const;
 
    Result run();
 
    Result runOnce();
 
    Result runNoWait();
 
    Result submitRequests(AsyncKernelEvents& kernelEvents);
 
    Result blockingPoll(AsyncKernelEvents& kernelEvents);
 
    Result dispatchCompletions(AsyncKernelEvents& kernelEvents);
 
    Result wakeUpFromExternalThread(AsyncLoopWakeUp& wakeUp);
 
    Result wakeUpFromExternalThread();
 
    Result createAsyncTCPSocket(SocketFlags::AddressFamily family, SocketDescriptor& outDescriptor);
 
    Result createAsyncUDPSocket(SocketFlags::AddressFamily family, SocketDescriptor& outDescriptor);
 
    Result associateExternallyCreatedSocket(SocketDescriptor& outDescriptor);
 
    Result associateExternallyCreatedSocketHandle(SocketDescriptor::Handle handle);
 
    Result associateExternallyCreatedFileDescriptor(FileDescriptor& outDescriptor);
 
    Result associateExternallyCreatedFileDescriptorHandle(FileDescriptor::Handle handle);
 
    static Result removeAllAssociationsFor(SocketDescriptor& outDescriptor);
 
    static Result removeAllAssociationsForSocketHandle(SocketDescriptor::Handle handle);
 
    static Result removeAllAssociationsFor(FileDescriptor& outDescriptor);
 
    static Result removeAllAssociationsForFileDescriptorHandle(FileDescriptor::Handle handle);
 
    void updateTime();
 
    [[nodiscard]] TimeMs 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 tryProbingIOUring();
 
    void clearSequence(AsyncSequence& sequence);
 
    struct Internal;
 
    using LoopWakeUp          = AsyncLoopWakeUp;
    using LoopWork            = AsyncLoopWork;
    using LoopTimeout         = AsyncLoopTimeout;
    using ProcessExit         = AsyncProcessExit;
    using Signal              = AsyncSignal;
    using SocketAccept        = AsyncSocketAccept;
    using SocketConnect       = AsyncSocketConnect;
    using SocketSend          = AsyncSocketSend;
    using SocketSendTo        = AsyncSocketSendTo;
    using SocketReceive       = AsyncSocketReceive;
    using SocketReceiveFrom   = AsyncSocketReceiveFrom;
    using FileRead            = AsyncFileRead;
    using FileWrite           = AsyncFileWrite;
    using FileSend            = AsyncFileSend;
    using FileReadiness       = AsyncFileReadiness;
    using ExternalCompletion  = AsyncExternalCompletion;
    using FileSystemOperation = AsyncFileSystemOperation;
    using EventObjectType     = EventObject;
    using ResultType          = AsyncResult;
 
  public:
    struct SC_ASYNC_EXPORT InternalDefinition
    {
        static constexpr int Windows = 576;
        static constexpr int Apple   = 552;
        static constexpr int Linux   = 816;
        static constexpr int Default = Linux;
 
        static constexpr size_t Alignment = 8;
 
        using Object = Internal;
    };
 
    using InternalOpaque = OpaqueObject<InternalDefinition>;
 
  private:
    InternalOpaque internalOpaque;
    Internal&      internal;
 
    friend struct AsyncRequest;
    friend struct AsyncFileWrite;
    friend struct AsyncFileRead;
    friend struct AsyncFileSystemOperation;
    friend struct AsyncResult;
};
 
struct SC_ASYNC_EXPORT AsyncEventLoopMonitor
{
    Function<void(void)> onNewEventsAvailable; 
 
    Result create(AsyncEventLoop& eventLoop);
 
    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);
};
 
} // namespace SC