AuroraRuntime/Source/Async/Schedular.cpp
Reece 99c5e1fa65 A pretty large patch not worth breaking up into separate commits
[*] Split up Aurora Async
[*] Split Async app into seperate ThreadPool concept
[*] Fix various OSThread bugs and tls transfer issues
[*] Set default affinity to 0xFFFFFFFF
[*] Update Build script
[+] Add AuTuplePopFront
[+] New Network Interface (unimplemented)
[*] Stub out the interfaces required for a better logger
[*] Fix Win32 ShellExecute bug; windows 11 struggles without explicit com init per the docs - now deferring to thread pool
[*] Update gitignore
[*] Follow XDG home standard
[*] Refactor some namespaces to use the shorthand aliases
[*] Various stability fixes
2021-11-05 17:34:23 +00:00

167 lines
4.4 KiB
C++

/***
Copyright (C) 2021 J Reece Wilson (a/k/a "Reece"). All rights reserved.
File: Schedular.cpp
Date: 2021-6-26
Author: Reece
***/
#include <Source/RuntimeInternal.hpp>
#include "Async.hpp"
#include "Schedular.hpp"
//#include "AsyncApp.hpp"
#include "ThreadPool.hpp"
namespace Aurora::Async
{
struct SchedEntry
{
AuUInt64 ns;
WorkerId_t target;
AuSPtr<IAsyncRunnable> runnable;
IThreadPoolInternal *pool;
};
static AuThreads::ThreadUnique_t gThread;
static AuThreadPrimitives::MutexUnique_t gSchedLock;
static AuList<SchedEntry> gEntries;
static void GetDispatchableTasks(AuList<SchedEntry> &pending)
{
AU_LOCK_GUARD(gSchedLock);
auto time = Time::CurrentClockNS();
for (auto itr = gEntries.begin(); itr != gEntries.end(); )
{
if (itr->ns <= time)
{
pending.push_back(std::move(*itr));
itr = gEntries.erase(itr);
}
else
{
itr ++;
}
}
}
static bool gLockedPump = false;
static void PumpSysThread()
{
RuntimeSysPump();
gLockedPump = false;
}
static void SchedThread()
{
AuUInt32 counter {};
AuList<SchedEntry> pending;
auto thread = AuThreads::GetThread();
while (!thread->Exiting())
{
AuList<SchedEntry> pending;
Threading::SleepNs(1000000 / 2 * gRuntimeConfig.async.schedularFrequency);
GetDispatchableTasks(pending);
for (auto &entry : pending)
{
try
{
entry.pool->Run(entry.target, entry.runnable);
entry.pool->DecrementTasksRunning();
}
catch (...)
{
if (entry.pool->ToThreadPool()->InRunnerMode())
{
LogWarn("Dropped scheduled task! Expect a leaky counter!");
LogWarn("Would you rather `Why u no exit?!` or `WHY DID U JUST CRASH REEEE` in production?");
}
Debug::PrintError();
}
}
counter++;
if ((!gRuntimeConfig.async.sysPumpFrequency) || ((gRuntimeConfig.async.sysPumpFrequency) && (counter % gRuntimeConfig.async.sysPumpFrequency) == 0))
{
try
{
if (!std::exchange(gLockedPump, true))
{
NewWorkItem(AuWorkerId_t{0, 0}, AuMakeShared<BasicWorkStdFunc>(PumpSysThread))->Dispatch();
}
}
catch (...)
{
LogWarn("Dropped SysRuntimePump");
Debug::PrintError();
}
}
}
}
void InitSched()
{
gSchedLock = AuThreadPrimitives::MutexUnique();
}
void DeinitSched()
{
gThread.reset();
gSchedLock.reset();
}
void StartSched()
{
AU_LOCK_GUARD(gSchedLock);
if (gThread) return;
gThread = AuThreads::ThreadUnique(AuThreads::ThreadInfo(
AuMakeShared<AuThreads::IThreadVectorsFunctional>(AuThreads::IThreadVectorsFunctional::OnEntry_t(std::bind(SchedThread)),
AuThreads::IThreadVectorsFunctional::OnExit_t{})
));
gThread->Run();
}
/// @deprecated
void StopSched()
{
// intentionally no-op
#if 0
gThread.reset();
#endif
}
void Schedule(AuUInt64 ns, IThreadPoolInternal *pool, WorkerId_t target, AuSPtr<IAsyncRunnable> runnable)
{
AU_LOCK_GUARD(gSchedLock);
pool->IncrementTasksRunning();
gEntries.push_back({ns, target, runnable, pool});
}
void TerminateSceduledTasks(IThreadPoolInternal *pool, WorkerId_t target)
{
AU_LOCK_GUARD(gSchedLock);
for (auto itr = gEntries.begin(); itr != gEntries.end(); )
{
if ((itr->pool == pool) &&
((itr->target == target) || (target.second == Async::kThreadIdAny && target.first == itr->target.first)))
{
itr->runnable->CancelAsync();
itr = gEntries.erase(itr);
}
else
{
itr ++;
}
}
}
}