AuroraRuntime/Source/Async/Schedular.cpp

221 lines
5.7 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"
#include <Console/Commands/Commands.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 AuWorkerPId_t gMainThread;
static bool gBOriginal;
static AuList<SchedEntry> gEntries;
void StartSched2();
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)
{
if (!AuTryInsert(pending, AuMove(*itr)))
{
break;
}
itr = gEntries.erase(itr);
}
else
{
itr ++;
}
}
}
static bool gLockedPump = false;
static void PumpSysThread()
{
try
{
RuntimeSysPump();
}
catch (...)
{
SysPushErrorCatch("SysPump failed");
}
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())
{
AuLogWarn("Dropped scheduled task! Expect a leaky counter!");
AuLogWarn("Would you rather `Why u no exit?!` or a spurious crash in production?");
}
Debug::PrintError();
}
}
counter++;
if (gRuntimeConfig.async.enableSysPumpFreqnecy)
{
if ((!gRuntimeConfig.async.sysPumpFrequency) || ((gRuntimeConfig.async.sysPumpFrequency) && (counter % gRuntimeConfig.async.sysPumpFrequency) == 0))
{
try
{
if (!AuExchange(gLockedPump, true))
{
NewWorkItem(gMainThread, AuMakeShared<BasicWorkStdFunc>(PumpSysThread))->Dispatch();
}
}
catch (...)
{
AuLogWarn("Dropped SysRuntimePump");
Debug::PrintError();
}
}
}
}
}
void InitSched()
{
gSchedLock = AuThreadPrimitives::MutexUnique();
}
void DeinitSched()
{
gThread.reset();
gSchedLock.reset();
}
AUKN_SYM void SetMainThreadForSysPumpScheduling(AuWorkerPId_t pid)
{
if (!pid)
{
gRuntimeConfig.async.enableSysPumpFreqnecy = gBOriginal;
gMainThread = AuWorkerPId_t(AuAsync::GetSharedAsyncApp(), AuWorkerId_t {0, 0});
Console::Commands::UpdateDispatcher(gMainThread);
return;
}
gMainThread = pid;
Console::Commands::UpdateDispatcher(gMainThread);
gRuntimeConfig.async.enableSysPumpFreqnecy = true;
StartSched2();
}
void StartSched()
{
if (!gRuntimeConfig.async.enableSchedularThread)
{
return;
}
else
{
if (gRuntimeConfig.async.sysPumpFrequency && !gMainThread)
{
gMainThread = AuWorkerPId_t(AuAsync::GetSharedAsyncApp(), AuWorkerId_t {0, 0});
}
}
StartSched2();
}
void StartSched2()
{
AU_LOCK_GUARD(gSchedLock);
if (gThread) return;
gBOriginal = gRuntimeConfig.async.enableSysPumpFreqnecy;
gThread = AuThreads::ThreadUnique(AuThreads::ThreadInfo(
AuMakeShared<AuThreads::IThreadVectorsFunctional>(AuThreads::IThreadVectorsFunctional::OnEntry_t(std::bind(SchedThread)),
AuThreads::IThreadVectorsFunctional::OnExit_t {})
));
gThread->Run();
}
void StopSched()
{
gThread.reset();
}
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 ++;
}
}
}
}