/*** Copyright (C) 2021 J Reece Wilson (a/k/a "Reece"). All rights reserved. File: WorkItem.cpp Date: 2021-6-26 Author: Reece ***/ #include #include "Async.hpp" #include "WorkItem.hpp" #include "AsyncApp.hpp" #include "Schedular.hpp" namespace Aurora::Async { WorkItem::WorkItem(IThreadPoolInternal *owner, const WorkerPId_t &worker, const AuSPtr &task, bool bSupportsBlocking) : worker_(worker), task_(task), owner_(owner) { if (bSupportsBlocking) { this->finishedEvent_ = AuThreadPrimitives::EventUnique(false, true, true); SysAssert(this->finishedEvent_); } } WorkItem::~WorkItem() { //Fail(); } AuSPtr WorkItem::WaitFor(const AuSPtr &workItem) { bool status {}; { auto dependency = AuReinterpretCast(workItem); AU_LOCK_GUARD(this->lock); AU_LOCK_GUARD(dependency->lock); if (dependency->HasFailed()) { Fail(); return AU_SHARED_FROM_THIS; } if (!AuTryInsert(dependency->waiters_, AuSharedFromThis())) { Fail(); return AU_SHARED_FROM_THIS; } if (!AuTryInsert(this->waitOn_, workItem)) { AuTryRemove(dependency->waiters_, AuSharedFromThis()); Fail(); return AU_SHARED_FROM_THIS; } } return AU_SHARED_FROM_THIS; } bool WorkItem::WaitForLocked(const AuList> &workItems) { for (auto &workItem : workItems) { auto dependency = AuReinterpretCast(workItem); AU_LOCK_GUARD(dependency->lock); if (dependency->HasFailed()) { return false; } if (!AuTryInsert(dependency->waiters_, AuSharedFromThis())) { return false; } if (!AuTryInsert(this->waitOn_, workItem)) { AuTryRemove(dependency->waiters_, AuSharedFromThis()); return false; } } return true; } AuSPtr WorkItem::WaitFor(const AuList> &workItems) { bool status {}; { AU_LOCK_GUARD(this->lock); status = WaitForLocked(workItems); } if (!status) { Fail(); } return AU_SHARED_FROM_THIS; } AuSPtr WorkItem::Then(const AuSPtr &next) { auto that = AU_SHARED_FROM_THIS; next->WaitFor(that); next->Dispatch(); return that; } AuSPtr WorkItem::SetSchedTimeNs(AuUInt64 ns) { this->dispatchTimeNs_ = Time::CurrentClockNS() + ns; return AU_SHARED_FROM_THIS; } AuSPtr WorkItem::SetSchedTimeAbs(AuUInt32 ms) { this->dispatchTimeNs_ = AuUInt64(ms) * AuMSToNS(ms); return AU_SHARED_FROM_THIS; } AuSPtr WorkItem::SetSchedTimeNsAbs(AuUInt64 ns) { this->dispatchTimeNs_ = ns; return AU_SHARED_FROM_THIS; } AuSPtr WorkItem::SetSchedTime(AuUInt32 ms) { this->dispatchTimeNs_ = Time::CurrentClockNS() + AuMSToNS(ms); return AU_SHARED_FROM_THIS; } AuSPtr WorkItem::AddDelayTime(AuUInt32 ms) { this->delayTimeNs_ += AuMSToNS(ms); return AU_SHARED_FROM_THIS; } AuSPtr WorkItem::AddDelayTimeNs(AuUInt64 ns) { this->delayTimeNs_ += ns; return AU_SHARED_FROM_THIS; } AuSPtr WorkItem::Dispatch() { DispatchEx(false); return AU_SHARED_FROM_THIS; } void WorkItem::DispatchEx(bool check) { AU_LOCK_GUARD(lock); DispatchExLocked(check); } void WorkItem::DispatchExLocked(bool check) { if (check) { if (this->dispatchPending_) { return; } } if (HasFailed()) { return; } for (auto itr = waitOn_.begin(); itr != waitOn_.end(); ) { auto &waitable = *itr; if (!waitable->HasFinished()) { return; } itr = waitOn_.erase(itr); } this->dispatchPending_ = true; if (Time::CurrentClockNS() < this->dispatchTimeNs_) { if (!Schedule()) { this->Fail(); } return; } if (auto delay = AuExchange(delayTimeNs_, {})) { this->dispatchTimeNs_ = delay + Time::CurrentClockNS(); if (!Schedule()) { this->Fail(); } return; } SendOff(); } float WorkItem::GetPrio() { return prio_; } void WorkItem::SetPrio(float val) { prio_ = val; } void WorkItem::CancelAsync() { AU_TRY_LOCK_GUARD_NAMED(this->lock, asd); Fail(); } void WorkItem::RunAsync() { AU_LOCK_GUARD(this->lock); RunAsyncLocked(); } void WorkItem::RunAsyncLocked() { IWorkItemHandler::ProcessInfo info(true); info.pool = this->owner_->ToThreadPool(); if (this->task_) { try { this->task_->DispatchFrame(info); } catch (...) { // TODO: runtime config for root level exception caught behaviour SysPushErrorCatch(); Fail(); return; } } switch (info.type) { case ETickType::eFinished: { // do nothing break; } case ETickType::eEnumInvalid: { SysPanic("Handle Invalid"); break; } case ETickType::eSchedule: { if (info.reschedMs) { SetSchedTime(info.reschedMs); } else if (info.reschedNs) { SetSchedTimeNs(info.reschedNs); } else if (info.reschedClockAbsMs) { SetSchedTimeAbs(info.reschedMs); } else if (info.reschedClockAbsNs) { SetSchedTimeNsAbs(info.reschedNs); } if (!WaitForLocked(info.waitFor)) { Fail(); } } [[fallthrough]]; case ETickType::eRerun: { DispatchExLocked(false); return; } case ETickType::eFailed: { Fail(); return; } } this->finished = true; if (this->finishedEvent_) { this->finishedEvent_->Set(); } for (auto &waiter : this->waiters_) { AuReinterpretCast(waiter)->DispatchEx(true); } } void WorkItem::Fail() { failed = true; if (auto task_ = AuExchange(this->task_, {})) { task_->OnFailure(); } for (auto &waiter : this->waiters_) { AuReinterpretCast(waiter)->Fail(); } this->waiters_.clear(); this->waitOn_.clear(); if (this->finishedEvent_) { this->finishedEvent_->Set(); } } bool WorkItem::BlockUntilComplete() { if (!this->finishedEvent_) return false; return this->owner_->WaitFor(this->worker_, AuUnsafeRaiiToShared(this->finishedEvent_), 0); } bool WorkItem::HasFinished() { return this->finished; } void WorkItem::Cancel() { AU_LOCK_GUARD(this->lock); Fail(); } bool WorkItem::HasFailed() { return this->failed; } bool WorkItem::Schedule() { return Async::Schedule(this->dispatchTimeNs_, this->owner_, this->worker_, AuSharedFromThis()); } void WorkItem::SendOff() { if (!this->task_) { // If we aren't actually calling a task interface, we may as well just dispatch objects waiting on us from here RunAsyncLocked(); } else { this->owner_->Run(this->worker_, AuSharedFromThis()); } } static auto GetWorkerInternal() { return static_cast(GetAsyncApp()); } inline auto GetWorkerInternal(const AuSPtr &pool) { if (pool.get() == AuStaticCast(gAsyncApp)) { return AuUnsafeRaiiToShared(AuStaticCast(gAsyncApp)); } return AuStaticPointerCast(pool); } AUKN_SYM AuSPtr NewWorkItem(const WorkerId_t &worker, const AuSPtr &task, bool supportsBlocking) { if (!task) { SysPushErrorArg("WorkItem has null task. Running out of memory?"); return {}; } return AuMakeShared(GetWorkerInternal(), WorkerPId_t { AuAsync::GetCurrentWorkerPId().pool, worker }, task, supportsBlocking); } AUKN_SYM AuSPtr NewWorkItem(const WorkerPId_t &worker, const AuSPtr &task, bool supportsBlocking) { if (!task) { SysPushErrorArg("WorkItem has null task. Running out of memory?"); return {}; } if (!worker.pool) { SysPushErrorArg("WorkItem has null pool"); return {}; } return AuMakeShared(GetWorkerInternal(worker.pool).get(), worker, task, supportsBlocking); } AUKN_SYM AuSPtr NewFence() { return AuMakeShared(GetWorkerInternal(), AuAsync::GetCurrentWorkerPId(), AuSPtr{}, true); } void *WorkItem::GetPrivateData() { if (!this->task_) { return nullptr; } return this->task_->GetPrivateData(); } AuOptional WorkItem::ToWorkResultT() { if (!this->task_) { return nullptr; } auto priv = reinterpret_cast(this->task_->GetPrivateData()); if (!priv) { return nullptr; } if (priv->magic == AuConvertMagicTag32("BWOT")) { return reinterpret_cast(priv)->opt; } return {}; } }