AuroraRuntime/Source/Async/Schedular.cpp

/***
    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"

namespace Aurora::Async
{
    struct SchedEntry
    {
        AuUInt64 ns;
        WorkerId_t target;
        AuSPtr<IAsyncRunnable> runnable;
    };

    static Threading::Threads::ThreadUnique_t gThread;
    static Threading::Primitives::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 = Threading::Threads::GetThread();

        while (!thread->Exiting())
        {
            AuList<SchedEntry> pending;

            Threading::SleepNs(1000000 / 2 * gRuntimeConfig.async.schedularFrequency);

            GetDispatchableTasks(pending);

            for (auto &entry : pending)
            {
                try
                {
                    static_cast<AsyncApp *>(GetAsyncApp())->Run(entry.target, entry.runnable);
                    DecRunningTasks();
                }
                catch (...)
                {
                    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({0, 0}, AuMakeShared<BasicWorkStdFunc>(PumpSysThread))->Dispatch();
                    }
                }
                catch (...)
                {
                    LogWarn("Dropped SysRuntimePump");
                    Debug::PrintError();
                }
            }
        }
        
    }

    void InitSched()
    {
        gSchedLock = Threading::Primitives::MutexUnique();
    }

    void DeinitSched()
    {
        gThread.reset();
        gSchedLock.reset();
    }

    void StartSched()
    {
        Threading::Threads::AbstractThreadVectors handler;
        handler.DoRun = [=](const Threading::Threads::IAuroraThread *thread)
        {
            SchedThread();
        };
        gThread = Threading::Threads::ThreadUnique(handler);
        gThread->Run();
    }

    void StopSched()
    {
        gThread.reset();
    }

    void Schedule(AuUInt64 ns, WorkerId_t target, AuSPtr<IAsyncRunnable> runnable)
    {
        AU_LOCK_GUARD(gSchedLock);
        IncRunningTasks();
        gEntries.push_back({ns, target, runnable});
    }

    void TerminateSceduledTasks(WorkerId_t target)
    {
        AU_LOCK_GUARD(gSchedLock);

        for (auto itr = gEntries.begin(); itr != gEntries.end(); )
        {
            if (itr->target <= target)
            {
                itr->runnable->CancelAsync();
                itr = gEntries.erase(itr);
            }
            else
            {
                itr ++;
            }
        }
    }
}