AuroraRuntime/Source/Grug/AuGrug.cpp
Jamie Reece Wilson 63050b2262 [+] IFileStream::IsFlushOnClose
[+] IFileStream::SetFlushOnClose
[+] IFileStream::IsWriteEoSOnClose
[+] IFileStream::SetWriteEoSOnClose
[+] IIOHandle::IsFlushOnClose
[+] IIOHandle::SetFlushOnClose
2024-01-02 00:00:47 +00:00

281 lines
7.0 KiB
C++

/***
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();
}
}