AuroraRuntime/Source/Grug/AuGrug.cpp

/***
    Copyright (C) 2021 J Reece Wilson (a/k/a "Reece"). All rights reserved.

    File: AuGrug.cpp
    Date: 2022-01-03
    File: Flusher.cpp
    Date: 2021-8-27
    Author: Reece
    Note: Grug couples async telemetry, console flushing, and log dispatching
          to one lazy working thread w/ high performance thread primitives for
          when urgent telemetry events are raised.
***/
#include <Source/RuntimeInternal.hpp>
#include "AuGrug.hpp"

#include <Source/Logging/AuLogger.hpp>
#include <Source/Console/ConsoleFIO/ConsoleFIO.hpp>
#include <Source/RuntimeInternal.hpp>
#include <Source/Console/Flusher.hpp>
#include <Source/Console/Console.hpp>
#include <Source/Debug/MemoryCrunch.hpp>

#if defined(AURORA_IS_LINUX_DERIVED)
    void LinuxSuperSecretIOTick();
    void LinuxSuperSecretFuckGlibc();
#endif

namespace Aurora::Grug
{
    static const auto kGrugSleepMs = 100;
    static const auto kGrugFlushMs = 500;

    static AuThreads::ThreadUnique_t            gGrugsBigWorld;
    static AuCondMutex                          gMutex;   // ^ that
    static AuConditionVariable                  gCondVar(AuUnsafeRaiiToShared(gMutex.AsPointer())); // slow logger work queue
    static AuSemaphore                          gArrows;
    static AuMutex                              gOtherMutex;
    static AuList<AuPair<AuUInt, bool>>         gHandlesToClose;
    static AuList<AuThreadPrimitives::IEvent *> gEventsToTrigger;

    static void SlowStartupTasks()
    {
        Console::ConsoleFIO::FIOCleanup();
    }

    // grug require only 1 strand
    static void GrugWorld()
    {
    #if defined(AURORA_IS_LINUX_DERIVED)
        LinuxSuperSecretFuckGlibc();
    #endif
        
        // grug surive first night
        SlowStartupTasks();

        AU_LOCK_GUARD(gMutex);

        AuDebug::AddMemoryCrunch(); // this thread must never encounter an out of memory condition. begin out of memory mitigations.

        Utility::RateLimiter limiter;
        limiter.SetNextStep(kGrugFlushMs);

        // grug has cave
        while (AuIsThreadRunning())
        {
            // * -> grug wake up from 100ms (kGrugSleepMs)
            // grug smashy alarm
            gCondVar->WaitForSignal(kGrugFlushMs);

            // grug anoy
            if (gArrows && gArrows->TryLock())
            {
                // grug yeet
                DequeueOneArrow();
            }

            // grug pump all async log messages
            GrugFlushWrites();

            // adhd grug wonder around the log pool every 500ms (rate limited to 100ms sleeps)
            if (limiter.CheckExchangePass())
            {
                GrugFlushFlushs();
            }

            // grug give up
            // grug yield for at least 100ms -> * 
            if (gArrows)
            {
                // grug sleep for 100ms or until poked
                if (gArrows->LockMS(kGrugSleepMs))
                {
                    DequeueOneArrow();
                }
            }
            else
            {
                // grug sleep 100ms
                AuThreading::Sleep(kGrugSleepMs);
            }

        #if defined(AURORA_IS_LINUX_DERIVED)
            ::LinuxSuperSecretIOTick();
        #endif

            GrugDoIoWork();
        }
    }

    static void DestroyFlushThread()
    {
        gGrugsBigWorld.reset();
    }

    static void InitFlushThread()
    {
        // Startup a runner thread that will take care of all the stress of triggering IO every so often on a remote thread 
        gGrugsBigWorld = AuThreads::ThreadUnique(AuThreads::ThreadInfo(
            AuMakeShared<AuThreads::IThreadVectorsFunctional>(AuThreads::IThreadVectorsFunctional::OnEntry_t(std::bind(GrugWorld)),
                                                              AuThreads::IThreadVectorsFunctional::OnExit_t{}),
            "Cave Grug"
        ));

        if (!gGrugsBigWorld)
        {
            return;
        }

        // he's a lazy bastard
        gGrugsBigWorld->SetThrottle(AuThreads::EThreadThrottle::eEfficient);
        gGrugsBigWorld->SetPriority(AuThreads::EThreadPriority::ePrioLow);
        gGrugsBigWorld->Run();
    }

    void InitGrug()
    {
        InitFlushThread();
    }

    void DeinitGrug()
    {
        if (bool(gGrugsBigWorld) && gGrugsBigWorld.get() != AuThreads::GetThread())
        {
            gGrugsBigWorld.reset();
        }

        GrugFlushWrites();
        GrugFlushFlushs();
    
        DeinitArrows();
    }

    void NotifyGrugOfTelemetry()
    {
        if (gArrows)
        {
            gArrows->Unlock(1);
        }

        DrachenlordScreech();
    }

    void DrachenlordScreech()
    {
        if (!gCondVar)
        {
            return;
        }
     
        gCondVar->Signal();
    }

    bool IsGrug()
    {
        if (!gGrugsBigWorld) return false;
        return AuThreads::GetThread() == gGrugsBigWorld.get();
    }

    void NotifyGrugOfLogs()
    {
        if (Grug::IsGrug())
        {
            Grug::GrugFlushWrites();
        }
        else
        {
            Grug::DrachenlordScreech();
        }
    }

    void GrugFlushWrites()
    {
        Logging::ForceFlushLoggers();
    }

    void GrugDoIoWork()
    {
        decltype(gHandlesToClose) toClose;
        decltype(gEventsToTrigger) toTrigger;

        if (gHandlesToClose.empty() &&
            gEventsToTrigger.empty())
        {
            return;
        }

        {
            AU_LOCK_GUARD(gOtherMutex);
            toClose = AuMove(gHandlesToClose);
            toTrigger = AuMove(gEventsToTrigger);
        }

        for (const auto [uHandle, bFlush] : toClose)
        {
        #if defined(AURORA_IS_MODERNNT_DERIVED)
            auto pHandle = (void *)uHandle;
            if (bFlush && pHandle)
            {
                if (AuIO::IsHandleFile(uHandle))
                {
                    FlushFileBuffers(pHandle);
                }
            }
            AuWin32CloseHandle(pHandle);
        #elif defined(AURORA_IS_POSIX_DERIVED)
            if (bFlush)
            {
                ::fsync(uHandle);
            }
            ::close(uHandle);
        #endif
        }

        for (const auto pEvent : toTrigger)
        {
            pEvent->Set();
        }
    }

    void GrugFlushFlushs()
    {
        Logging::ForceFlushFlush();
        Console::PumpOffMain();
        Console::ForceFlush();
        GrugDoIoWork();
    }

    void CloseHandle(AuUInt64 handle, bool bFlush)
    {
        {
            AU_DEBUG_MEMCRUNCH;
            AU_LOCK_GUARD(gOtherMutex);
            gHandlesToClose.push_back(AuMakePair(AuUInt(handle), bFlush));
        }

        NotifyGrugOfTelemetry();
    }

    void WaitForGrugTick()
    {
        AuEvent event(false, true, true);

        if (gGrugsBigWorld &&
             gGrugsBigWorld.get() == AuThreads::GetThread())
        {
            return;
        }

        {
            AU_DEBUG_MEMCRUNCH;
            AU_LOCK_GUARD(gOtherMutex);
            gEventsToTrigger.push_back(event.AsPointer());
        }

        {
            NotifyGrugOfTelemetry();
        }

        event->Lock();
    }
}