/*** Copyright (C) 2021 J Reece Wilson (a/k/a "Reece"). All rights reserved. File: Async.hpp Date: 2021-7-14 Author: Reece ***/ #pragma once namespace Aurora::Loop { class ILoopSource; } namespace Aurora::Async { class IWorkItem; class IAsyncApp; using AVoid = AuUInt8; AUKN_SYM IAsyncApp *GetAsyncApp(); using ThreadGroup_t = AuUInt8; using ThreadId_t = AuUInt16; /// ThreadGroup_t: /// 0 = system main thread /// 1+ = user defined /// /// ThreadId_t: /// -1 = invalid /// index = tid/runner id /// using WorkerId_t = AuPair; static const ThreadId_t kThreadIdAny = -1; struct WorkPriv { AuUInt32 magic; }; struct IWorkItemHandler { enum class EProcessNext { eInvalid = -1, eFinished = 0, eRerun, eSchedule, eFailed }; struct ProcessInfo { ProcessInfo(bool finished) : type(finished ? EProcessNext::eFinished : EProcessNext::eFailed) {} ProcessInfo(EProcessNext type) : type(type) {} ProcessInfo(const AuList> &blockedBy) : type(EProcessNext::eSchedule), waitFor(blockedBy) {} // ... EProcessNext type; AuList> waitFor; AuUInt32 reschedMs; AuUInt64 reschedNs; }; virtual void DispatchFrame(ProcessInfo &info) = 0; virtual void Shutdown() = 0; virtual void *GetPrivateData() { return nullptr; } }; template struct FJob { std::function onSuccess = 0; std::function onFailure = 0; }; template static inline FJob JobFromConsumer(const AuConsumer &onSuccess) { FJob ret; ret.onSuccess = [=](const Info_t &in, const Result_t &a) { onSuccess(in, a); }; return ret; } template static inline FJob JobFromConsumer(const AuConsumer &onSuccess, const AuConsumer &onFailure) { FJob ret; ret.onSuccess = [=](const Info_t &in, const Result_t &a) { onSuccess(in, a); }; ret.onFailure = [=](const Info_t &a, bool neverDispatched) { onFailure(a, neverDispatched); }; return ret; } template static inline FJob JobFromConsumer(const AuConsumer &onSuccess, const AuConsumer &onFailure) { FJob ret; ret.onSuccess = [=](const Info_t &in, const Result_t &a) { onSuccess(in, a); }; ret.onFailure = [=](const Info_t &a, bool neverDispatched) { onFailure(a); }; return ret; } using FVoidJob = FJob; template struct CJob { void(* onSuccess)(const Info_t &, const Result_t &); // void(* onFailure)(const Info_t &, bool taskNeverDispatched); // called from caller thread if taskNeverDispatched }; template struct FTask { std::function(const Info_t &)> onFrame = 0; }; using FVoidTask = FTask; template static inline FTask TaskFromConsumerRefT(const AuConsumer &func) { FTask ret; ret.onFrame = [=](const In_t &a) -> AuOptional { func(a); return AVoid{}; }; return ret; } template static inline FTask TaskFromConsumerRefT(const AuSupplierConsumer, const In_t &> &func) { FTask ret; ret.onFrame = [=](const In_t &a) -> AuOptional { return func(a); }; return ret; } static inline FTask TaskFromFunctional(const AuVoidFunc &func) { FTask ret; ret.onFrame = [=](const AVoid &a) -> AuOptional { func(); return AVoid{}; }; return ret; } template struct CTask { AuOptional(* onFrame)(const Info_t &); }; class IWorkItem { public: virtual AuSPtr WaitFor(const AuSPtr &workItem) = 0; virtual AuSPtr WaitFor(const AuList> &workItem) = 0; // ms = time relative to the current time virtual AuSPtr SetSchedTime(AuUInt32 ms) = 0; // ns = time relative to the current time virtual AuSPtr SetSchedTimeNs(AuUInt64 ns) = 0; // ms = time relative to the time at which the work item would otherwise dispatch virtual AuSPtr AddDelayTime(AuUInt32 ms) = 0; // ns = time relative to the time at which the work item would otherwise dispatch virtual AuSPtr AddDelayTimeNs(AuUInt64 ns) = 0; virtual AuSPtr Then(const AuSPtr &next) = 0; virtual AuSPtr Dispatch() = 0; virtual bool BlockUntilComplete() = 0; virtual bool HasFinished() = 0; virtual bool HasFailed() = 0; virtual void Cancel() = 0; virtual void *GetPrivateData() = 0; virtual AuOptional ToWorkResultT() = 0; }; AUKN_SYM AuSPtr NewWorkItem(const WorkerId_t &worker, const AuSPtr &task, bool supportsBlocking = false); #pragma region EASE_OF_READING struct BasicWorkStdFunc : IWorkItemHandler { std::function callback; std::function shutdown; BasicWorkStdFunc(std::function &&callback, std::function &&error) : callback(std::move(callback)), shutdown(std::move(shutdown)) {} BasicWorkStdFunc(std::function &&callback) : callback(std::move(callback)) {} BasicWorkStdFunc(const std::function &callback) : callback(callback) {} BasicWorkStdFunc(const std::function &callback, const std::function &shutdown) : callback(callback), shutdown(shutdown) {} private: #if !defined(_CPPSHARP) void DispatchFrame(ProcessInfo &info) override { try { callback(); } catch (...) { Debug::PrintError(); } } void Shutdown() override { try { if (shutdown) { shutdown(); } } catch (...) { Debug::PrintError(); } } #endif }; #if !defined(_CPPSHARP) /// @hideinitializer struct BasicWorkCtx : WorkPriv { BasicWorkCtx() { magic = AuConvertMagicTag32("BWOT"); opt = nullptr; } void *opt; }; /// @hideinitializer template, typename Job_t = FJob> struct BasicWorkCallback : IWorkItemHandler, std::enable_shared_from_this { BasicWorkCallback() { caller = GetAsyncApp()->GetCurrentThread(); } BasicWorkCallback(Task_t &&task) : task(std::move(task)) { caller = GetAsyncApp()->GetCurrentThread(); } BasicWorkCallback(Task_t &&task, Job_t &&callback) : task(std::move(task)), callback(std::move(callback)) { caller = GetAsyncApp()->GetCurrentThread(); } BasicWorkCallback(const Task_t &task) : task(task) { caller = GetAsyncApp()->GetCurrentThread(); } BasicWorkCallback(const Task_t &task, const Job_t &callback) : task(task), callback(callback) { caller = GetAsyncApp()->GetCurrentThread(); } BasicWorkCallback(const Task_t &task, const Job_t &callback, const Info_t &info) : task(task), callback(callback), input(info) { caller = GetAsyncApp()->GetCurrentThread(); } BasicWorkCallback(Task_t &&task, Job_t &&callback, Info_t &&info) : task(std::move(task)), callback(std::move(callback)), input(std::move(info)) { caller = GetAsyncApp()->GetCurrentThread(); } BasicWorkCallback(const Task_t &task, const Job_t &callback, Info_t &&info) : task(task), callback(callback), input(std::move(info)) { caller = GetAsyncApp()->GetCurrentThread(); } Info_t input; Task_t task; Job_t callback; BasicWorkCallback &SetTask(const Task_t &task) { this->task = task; return *this; } BasicWorkCallback &SetTask(const Job_t &callback) { this->callback = callback; return *this; } private: static constexpr bool IsCallbackPtr = std::is_pointer_v || AuIsBaseOfTemplate::value; static constexpr bool IsTaskPtr = std::is_pointer_v || AuIsBaseOfTemplate::value; WorkerId_t caller; BasicWorkCtx secretContext_; AuOptional resultValue_; virtual void *GetPrivateData() override { return &secretContext_; } void DispatchFrame(ProcessInfo &info) override { try { if constexpr (IsTaskPtr) { resultValue_ = task->onFrame(input); } else { resultValue_ = task.onFrame(input); } } catch (...) { Debug::PrintError(); Shutdown(); } auto pin = std::static_pointer_cast>(this->shared_from_this()); std::function func = [pin]() { try { if (pin->resultValue_.has_value()) { pin->secretContext_.opt = &pin->resultValue_.value(); if constexpr (IsCallbackPtr) { pin->callback->onSuccess(pin->input, *pin->resultValue_); } else { pin->callback.onSuccess(pin->input, *pin->resultValue_); } } else { pin->CallOnFailure(false); } } catch (...) { Debug::PrintError(); } }; try { if (caller == GetAsyncApp()->GetCurrentThread()) { func(); } else { std::function err = [pin]() { pin->CallOnFailure(false); }; // TODO: this is somewhat evil. double alloc when we could reuse this NewWorkItem(caller, AuMakeShared(func, err))->Dispatch(); } } catch (...) { Debug::PrintError(); Shutdown(); } } void Shutdown() override { try { CallOnFailure(true); } catch (...) { Debug::PrintError(); } } void CallOnFailure(bool fail) { if constexpr (IsCallbackPtr) { if constexpr (AuIsBaseOfTemplateonFailure)>::value) { if (!callback->onFailure) { return; } } callback->onFailure(input, fail); } else { if constexpr (AuIsBaseOfTemplate::value) { if (!callback.onFailure) { return; } } callback.onFailure(input, fail); } } }; /// @hideinitializer template, typename Cleanup_t = std::function> struct WorkItemCallable : IWorkItemHandler { Frame_t frame; Cleanup_t cleanup; private: void DispatchFrame(ProcessInfo &info) override { frame(); info.type = IWorkItemHandler::EProcessNext::eFinished; } void Shutdown() override { cleanup(); } }; #define ASYNC_ERROR(exp) if constexpr (std::is_same_v) { SysPushErrorGen(exp); return {}; } else { throw std::exception(exp); } #define ASYNC_FINISH if constexpr (std::is_same_v) { return true; } template || std::is_void::value)> static std::function TranslateAsyncFunctionToDispatcherWithThread(WorkerId_t id, std::function func) { return [=](Args&&... in) -> T { auto work = AuMakeShared([=]() -> void { func(in...); }); if (!work) ASYNC_ERROR("can't dispatch async call; out of memory"); auto workItem = NewWorkItem(id, work); if (!workItem) ASYNC_ERROR("can't dispatch async call; out of memory"); workItem->Dispatch(); ASYNC_FINISH; }; } template || std::is_void::value)> static std::function TranslateAsyncFunctionToDispatcher(std::function func) { return TranslateAsyncFunctionToDispatcherWithThread(GetAsyncApp()->GetCurrentThread(), func); } template || std::is_void::value)> static std::function, Args...)> TranslateAsyncReturnableFunctionToDispatcherWithThread(WorkerId_t id, std::function(Args...)> func) { return [=](std::function callback, Args... in) -> T { auto work = AuMakeShared>(); if (!work) ASYNC_ERROR("can't dispatch async call; out of memory"); work.task.onProcess = [](const AVoid &) -> AuOptional { return func(in...); }; work.callback.onSuccess = [=](const AVoid &, const B &ret) { callback(ret); }; auto workItem = NewWorkItem(id, work); if (!workItem) ASYNC_ERROR("can't dispatch async call; out of memory"); workItem->Dispatch(); ASYNC_FINISH; }; } template || std::is_void::value)> static std::function, Args...)> TranslateAsyncReturnableFunctionToDispatcher(std::function(Args...)> func) { return TranslateAsyncReturnableFunctionToDispatcherWithThread(GetAsyncApp()->GetCurrentThread(), func); } #undef ASYNC_ERROR #undef ASYNC_FINISH #endif #pragma endregion EASE_OF_READING template, typename Job_t = FJob> static AuSPtr DispatchBasicWorkCallback(const WorkerId_t &worker, const Task_t &task, const Job_t &job, bool enableWait = false) { return Async::NewWorkItem(worker, AuMakeShared>(task, job), enableWait)->Dispatch(); } template, typename Job_t = FJob> static AuSPtr DispatchBasicWorkCallback(const WorkerId_t &worker, const Task_t &task, const Job_t &job, const Info_t &inputParameters, bool enableWait = false) { return Async::NewWorkItem(worker, AuMakeShared>(task, job, inputParameters), enableWait)->Dispatch(); } class IAsyncApp { public: // Main thread logic virtual void Start() = 0; virtual void Main() = 0; virtual void Shutdown() = 0; virtual bool Exiting() = 0; virtual void SetConsoleCommandDispatcher(WorkerId_t id) = 0; // Spawning virtual bool Spawn(WorkerId_t) = 0; virtual Threading::Threads::ThreadShared_t ResolveHandle(WorkerId_t) = 0; virtual AuBST> GetThreads() = 0; virtual WorkerId_t GetCurrentThread() = 0; // Synchronization // Note: syncing to yourself will nullify requireSignal to prevent deadlock virtual bool Sync(WorkerId_t group, AuUInt32 timeoutMs = 0, bool requireSignal = false) = 0; virtual void Signal(WorkerId_t group) = 0; virtual void SyncAllSafe() = 0; // Features virtual void AddFeature(WorkerId_t id, AuSPtr feature, bool async = false) = 0; // Debug virtual void AssertInThreadGroup(ThreadGroup_t thread) = 0; virtual void AssertWorker(WorkerId_t id) = 0; virtual bool Poll(bool block) = 0; virtual void AddLoopSource(const AuSPtr &loopSource, WorkerId_t workerId, AuUInt32 timeout, const AuConsumer, bool> &callback) = 0; }; }