AuroraRuntime/Source/IO/AuIOProcessor.cpp

1020 lines
25 KiB
C++
Raw Normal View History

/***
Copyright (C) 2022 J Reece Wilson (a/k/a "Reece"). All rights reserved.
File: AuIOProcessor.cpp
Date: 2022-6-6
Author: Reece
***/
#include <Source/RuntimeInternal.hpp>
#include <Aurora/IO/IOExperimental.hpp>
#include "AuIOProcessorItem.hpp"
#include "AuIOProcessorItems.hpp"
#include "AuIOProcessor.hpp"
#include "AuIOPipeProcessor.hpp"
#include "Loop/Loop.hpp"
#include "Loop/LoopQueue.hpp"
#include <Async/ThreadPool.hpp>
namespace Aurora::IO
{
2022-08-29 15:46:46 +00:00
IOProcessor::IOProcessor(AuUInt threadId,
bool bTickOnly,
AuOptionalEx<AuAsync::WorkerPId_t> worker,
const AuSPtr<AuLoop::ILoopQueue> &pLoopQueue,
bool bIsNoQueue) :
2022-08-29 15:46:46 +00:00
mutliplexIOAndTimer(!bTickOnly),
pLoopQueue(pLoopQueue),
asyncWorker(worker),
threadId(threadId),
streamProcessors(this),
bIsNoQueue(bIsNoQueue)
{
}
IOProcessor::~IOProcessor()
{
this->ReleaseAllWatches();
this->items.Deinit();
}
bool IOProcessor::Init()
{
auto pQueue = this->ToQueue();
if (!pQueue)
2022-06-14 01:14:51 +00:00
{
return {};
}
if (!this->items.Init())
{
SysPushErrorNested();
return false;
}
auto pTimer = AuLoop::NewLSTimer(0);
if (!pTimer)
{
SysPushErrorNested();
return false;
}
if (!this->timers.Init(this, pTimer))
{
SysPushErrorNested();
return false;
}
if (!pQueue->SourceAdd(this->items.cvEvent))
{
SysPushErrorNested();
return false;
}
if (!this->bIsNoQueue)
{
auto pAS = AuMakeShared<AuLoop::ILoopSourceSubscriberFunctional>([pThat = AuSharedFromThis()](const AuSPtr<AuLoop::ILoopSource> &pSource) -> bool
{
pThat->ManualTick();
return false;
});
if (!pQueue->AddCallback(this->items.cvEvent, pAS))
{
return false;
}
}
(void)pQueue->Commit();
2022-06-14 01:14:51 +00:00
if (asyncWorker)
{
AuAtomicAdd(&AuStaticCast<AuAsync::ThreadPool>(asyncWorker.value().GetPool())->uAtomicIOProcessors, 1u);
}
return true;
}
bool IOProcessor::QueueIOEvent(const AuSPtr<IIOProcessorItem> &ioEvent)
{
return this->items.AddFrameOrFallback(AuStaticCast<IOProcessorItem>(ioEvent));
}
AuUInt32 IOProcessor::TryTick()
{
if (!CheckThread())
{
SysPushErrorGeneric("Wrong Thread");
return 0;
}
AU_LOCK_GUARD(this->items.mutex);
FrameStart();
FrameRunThreadIO();
FrameRunCheckLSes();
bool bHasAlertedItems = this->items.workSignaled.size() ||
this->timers.nbTicker.HasPassed();
if (!bHasAlertedItems)
{
ReportState(EIOProcessorEventStage::eFrameEndOfFrame);
return 0;
}
return FrameRunEpilogue();
}
bool IOProcessor::TickForRegister(const AuSPtr<IIOProcessorItem> &ioEvent)
{
return this->items.AddFrameOrFallback(AuStaticCast<IOProcessorItem>(ioEvent));
}
void IOProcessor::TickForHack(const AuSPtr<IIOProcessorItem> &ioEvent)
{
SysAssert(this->TickForRegister(ioEvent));
if (AuExchange(this->bScheduled_, true))
{
return;
}
2022-08-29 15:46:46 +00:00
AuStaticCast<Loop::LoopQueue>(this->pLoopQueue)->AddHook([that = AuSharedFromThis()]()
{
that->RunTick();
});
}
AuUInt32 IOProcessor::TickFor(const AuSPtr<IIOProcessorItem> &ioEvent)
{
SysAssert(TickForRegister(ioEvent));
if (IsTickOnly())
{
return 0;
}
if (this->bFrameStart &&
!this->bNoDeferredTick)
{
return 0;
}
if (!CheckThread())
{
SysPushErrorGeneric("Wrong Thread");
return 0;
}
AU_LOCK_GUARD(this->items.mutex);
FrameStart();
FrameRunThreadIO();
FrameRunCheckLSes();
return FrameRunEpilogue();
}
AuUInt32 IOProcessor::RunTick()
{
if (!CheckThread())
{
SysPushErrorGeneric("Wrong Thread");
AuIO::IOSleep(1);
return 0;
}
this->bScheduled_ = false;
this->bFrameStart = false;
AU_LOCK_GUARD(this->items.mutex);
FrameStart();
FrameWaitForAny(0);
2022-06-14 01:14:51 +00:00
FramePumpWaitingBlocked();
FrameRunThreadIO();
FrameRunCheckLSes();
return FrameRunEpilogue();
}
AuUInt32 IOProcessor::RunTickEx(AuUInt32 dwTimeout)
{
if (!CheckThread())
{
SysPushErrorGeneric("Wrong Thread");
AuIO::IOSleep(1);
return 0;
}
this->bScheduled_ = false;
this->bFrameStart = false;
AU_LOCK_GUARD(this->items.mutex);
FrameStart();
FrameWaitForAny(dwTimeout);
FramePumpWaitingBlocked();
FrameRunThreadIO();
FrameRunCheckLSes();
return FrameRunEpilogue();
}
AuUInt32 IOProcessor::ManualTick()
{
if (!CheckThread())
{
SysPushErrorGeneric("Wrong Thread");
return 0;
}
this->bFrameStart = false;
AU_LOCK_GUARD(this->items.mutex);
FrameStart();
FrameRunThreadIO();
FrameRunCheckLSes();
return FrameRunEpilogue();
}
void IOProcessor::FrameRunIOWorkUnits()
{
AU_LOCK_GUARD(this->items.mutex);
while (this->workUnits.size())
{
for (auto workItems : AuExchange(this->workUnits, {}))
{
workItems->OnRun();
}
}
}
AuUInt IOProcessor::FrameRunEpilogue()
{
FrameRunAlerted();
FrameRunAlertedSniffers();
FrameRunIOWorkUnits();
IO::SendBatched();
FrameEndOfFrameEvents();
return FrameFinalize();
}
void IOProcessor::FrameRunCheckLSes()
{
if (!this->IsTickOnly())
{
return;
}
for (auto &item : this->items.allItems)
{
2022-08-29 15:46:46 +00:00
if (item->pItem->IsRunOnSelfIO() && item->pItem->IsRunOnSelfIOCheckedOnTimerTick())
{
2022-08-29 15:46:46 +00:00
auto ls = item->pItem->GetSelfIOSource();
if (ls && ls->IsSignaled())
{
// Should deadlock without critical sections
this->QueueIOEvent(item);
}
}
}
}
void IOProcessor::FrameRunAlerted()
{
2024-03-13 05:37:51 +00:00
for (const auto &a : this->items.onTickReceivers)
{
2022-08-29 15:46:46 +00:00
if (a->pListener)
{
try
{
2022-08-29 15:46:46 +00:00
a->pListener->Tick_RunOnTick();
if (!AuExists(this->items.workSignaled, a) &&
!AuExists(this->items.onTickReceivers, a))
{
2022-08-29 15:46:46 +00:00
a->pListener->Tick_Any();
}
}
catch (...)
{
SysPushErrorCatch();
}
}
2024-03-13 05:37:51 +00:00
SysAssert(AuTryInsert(this->items.finalizeQueue, AuConstReference(a)));
}
2024-03-13 05:37:51 +00:00
for (const auto &a : this->items.workSignaled)
{
2022-08-29 15:46:46 +00:00
if (a->pListener)
{
try
{
2022-08-29 15:46:46 +00:00
a->pListener->Tick_SelfIOEvent();
a->pListener->Tick_Any();
}
catch (...)
{
SysPushErrorCatch();
}
}
2024-03-13 05:37:51 +00:00
SysAssert(AuTryInsert(this->items.finalizeQueue, AuConstReference(a)));
}
}
void IOProcessor::FrameRunAlertedSniffers()
{
if (this->items.workSignaled.empty())
{
return;
}
for (auto &a : this->items.onOtherReceivers)
{
if (AuExists(this->items.workSignaled, a))
{
continue;
}
2022-08-29 15:46:46 +00:00
if (a->pListener)
{
try
{
2022-08-29 15:46:46 +00:00
a->pListener->Tick_OtherIOEvent();
a->pListener->Tick_Any();
}
catch (...)
{
SysPushErrorCatch();
SysAssert(a->FailWatch());
}
}
this->items.finalizeQueue.push_back(a);
}
}
void IOProcessor::FrameEndOfFrameEvents()
{
ReportState(EIOProcessorEventStage::eFrameWorkDone);
for (auto &pumped : this->items.finalizeQueue)
{
2022-08-29 15:46:46 +00:00
if (!pumped->pListener)
{
continue;
}
try
{
2022-08-29 15:46:46 +00:00
pumped->pListener->Tick_FrameEpilogue();
}
catch (...)
{
SysPushErrorCatch();
pumped->FailWatch();
}
}
auto hasRemoved = this->items.crossThreadAbort.size();
for (auto itr = this->items.crossThreadAbort.begin(); itr != this->items.crossThreadAbort.end(); )
{
bool fatal = itr->second;
auto item = itr->first;
this->ClearProcessor(item, fatal);
itr = this->items.crossThreadAbort.erase(itr);
}
if (hasRemoved)
{
ToQueue()->Commit();
}
}
void IOProcessor::ClearProcessor(const AuSPtr<IOProcessorItem> &processor, bool fatal)
{
if (!AuTryRemove(this->items.allItems, processor))
{
return;
}
try
{
2022-08-29 15:46:46 +00:00
if (processor->pListener)
{
[+] Network + Protocol + TLS - Initial Commit ============================================================================= Network ]==================================================================== ============================================================================= [+] Added (very) early Aurora::IO::Net implementation [+] AuNet::EHostnameType [+] AuNet::EIPProtocol [+] AuNet::ENetworkError [+] AuNet::ETransportProtocol [+] AuNet::INetInterface [+] AuNet::INetSrvDatagram [+] AuNet::INetSrvResolve [+] AuNet::INetSrvSockets [+] AuNet::INetSrvWorkers [+] AuNet::INetWorker [+] AuNet::IPAddress [+] AuNet::IResolver [+] AuNet::ISocket [+] AuNet::IResolver [+] AuNet::ISocketBase [+] AuNet::ISocketChannel [+] AuNet::ISocketDriver [+] AuNet::ISocketDriverFactory [+] AuNet::ISocketServer [+] AuNet::ISocketServerDriver [+] AuNet::NetEndpoint [+] AuNet::NetError [+] AuNet::NetHostname (+implementation) ============================================================================= Protocol ]=================================================================== ============================================================================= [+] IProtocolInterceptor [+] IProtocolInterceptorEx [+] IProtocolStack (+implementation) ============================================================================= TLS ]======================================================================== ============================================================================= [+] ITLSContext [+] TLSProtocolRecv [+] TLSProtocolSend (+implementation) ============================================================================= IO Bug Fixes ]=============================================================== ============================================================================= [*] IOProcessor::SubmitIOWorkItem should signal the CvEvent, forcing at least once future tick (wont optimize with if in tick & not yet dispatched work items) [*] Split IOPipeWork in into IOPipeProcessor header [+] IOPipeWork::GetBuffer (internal reallocation) [*] Harden against IAsyncTransactions without a loop source [*] Missing null `if (processor->listener)` in IOProcessor [*] Solved some soft-lock conditions under Linux's LoopQueue (added deferred commits) [*] Quick hack: IOProcessor::HasItems() should OR the early can-tick check function. ============================================================================= Other ]====================================================================== ============================================================================= [+] Linux: LSSignalCatcher [+] `static void AuResetMember(Aurora::Memory::ByteBuffer &ref)` for AuROXTL [*] Attempt to enforce a normalization and don't overwrite-readptr-under-istreamwriters policy in ByteBuffer_ReadWrite (circular buffers) [*] Bad ECC ctors ============================================================================= Known issues ]=============================================================== ============================================================================= > Linux net is nowhere near done > UDP socket emulation layer isn't implemented > Ciphersuite API is a stub > Private key API is a stub > ...therefore no TLS servers > Missing thread safety precautions under net > Net implementation is still beri early
2022-08-28 19:02:06 +00:00
if (fatal)
{
2022-08-29 15:46:46 +00:00
processor->pListener->OnFailureCompletion();
[+] Network + Protocol + TLS - Initial Commit ============================================================================= Network ]==================================================================== ============================================================================= [+] Added (very) early Aurora::IO::Net implementation [+] AuNet::EHostnameType [+] AuNet::EIPProtocol [+] AuNet::ENetworkError [+] AuNet::ETransportProtocol [+] AuNet::INetInterface [+] AuNet::INetSrvDatagram [+] AuNet::INetSrvResolve [+] AuNet::INetSrvSockets [+] AuNet::INetSrvWorkers [+] AuNet::INetWorker [+] AuNet::IPAddress [+] AuNet::IResolver [+] AuNet::ISocket [+] AuNet::IResolver [+] AuNet::ISocketBase [+] AuNet::ISocketChannel [+] AuNet::ISocketDriver [+] AuNet::ISocketDriverFactory [+] AuNet::ISocketServer [+] AuNet::ISocketServerDriver [+] AuNet::NetEndpoint [+] AuNet::NetError [+] AuNet::NetHostname (+implementation) ============================================================================= Protocol ]=================================================================== ============================================================================= [+] IProtocolInterceptor [+] IProtocolInterceptorEx [+] IProtocolStack (+implementation) ============================================================================= TLS ]======================================================================== ============================================================================= [+] ITLSContext [+] TLSProtocolRecv [+] TLSProtocolSend (+implementation) ============================================================================= IO Bug Fixes ]=============================================================== ============================================================================= [*] IOProcessor::SubmitIOWorkItem should signal the CvEvent, forcing at least once future tick (wont optimize with if in tick & not yet dispatched work items) [*] Split IOPipeWork in into IOPipeProcessor header [+] IOPipeWork::GetBuffer (internal reallocation) [*] Harden against IAsyncTransactions without a loop source [*] Missing null `if (processor->listener)` in IOProcessor [*] Solved some soft-lock conditions under Linux's LoopQueue (added deferred commits) [*] Quick hack: IOProcessor::HasItems() should OR the early can-tick check function. ============================================================================= Other ]====================================================================== ============================================================================= [+] Linux: LSSignalCatcher [+] `static void AuResetMember(Aurora::Memory::ByteBuffer &ref)` for AuROXTL [*] Attempt to enforce a normalization and don't overwrite-readptr-under-istreamwriters policy in ByteBuffer_ReadWrite (circular buffers) [*] Bad ECC ctors ============================================================================= Known issues ]=============================================================== ============================================================================= > Linux net is nowhere near done > UDP socket emulation layer isn't implemented > Ciphersuite API is a stub > Private key API is a stub > ...therefore no TLS servers > Missing thread safety precautions under net > Net implementation is still beri early
2022-08-28 19:02:06 +00:00
}
else
{
2022-08-29 15:46:46 +00:00
processor->pListener->OnNominalCompletion();
[+] Network + Protocol + TLS - Initial Commit ============================================================================= Network ]==================================================================== ============================================================================= [+] Added (very) early Aurora::IO::Net implementation [+] AuNet::EHostnameType [+] AuNet::EIPProtocol [+] AuNet::ENetworkError [+] AuNet::ETransportProtocol [+] AuNet::INetInterface [+] AuNet::INetSrvDatagram [+] AuNet::INetSrvResolve [+] AuNet::INetSrvSockets [+] AuNet::INetSrvWorkers [+] AuNet::INetWorker [+] AuNet::IPAddress [+] AuNet::IResolver [+] AuNet::ISocket [+] AuNet::IResolver [+] AuNet::ISocketBase [+] AuNet::ISocketChannel [+] AuNet::ISocketDriver [+] AuNet::ISocketDriverFactory [+] AuNet::ISocketServer [+] AuNet::ISocketServerDriver [+] AuNet::NetEndpoint [+] AuNet::NetError [+] AuNet::NetHostname (+implementation) ============================================================================= Protocol ]=================================================================== ============================================================================= [+] IProtocolInterceptor [+] IProtocolInterceptorEx [+] IProtocolStack (+implementation) ============================================================================= TLS ]======================================================================== ============================================================================= [+] ITLSContext [+] TLSProtocolRecv [+] TLSProtocolSend (+implementation) ============================================================================= IO Bug Fixes ]=============================================================== ============================================================================= [*] IOProcessor::SubmitIOWorkItem should signal the CvEvent, forcing at least once future tick (wont optimize with if in tick & not yet dispatched work items) [*] Split IOPipeWork in into IOPipeProcessor header [+] IOPipeWork::GetBuffer (internal reallocation) [*] Harden against IAsyncTransactions without a loop source [*] Missing null `if (processor->listener)` in IOProcessor [*] Solved some soft-lock conditions under Linux's LoopQueue (added deferred commits) [*] Quick hack: IOProcessor::HasItems() should OR the early can-tick check function. ============================================================================= Other ]====================================================================== ============================================================================= [+] Linux: LSSignalCatcher [+] `static void AuResetMember(Aurora::Memory::ByteBuffer &ref)` for AuROXTL [*] Attempt to enforce a normalization and don't overwrite-readptr-under-istreamwriters policy in ByteBuffer_ReadWrite (circular buffers) [*] Bad ECC ctors ============================================================================= Known issues ]=============================================================== ============================================================================= > Linux net is nowhere near done > UDP socket emulation layer isn't implemented > Ciphersuite API is a stub > Private key API is a stub > ...therefore no TLS servers > Missing thread safety precautions under net > Net implementation is still beri early
2022-08-28 19:02:06 +00:00
}
}
}
catch (...)
{
SysPushErrorCatch();
}
2022-08-29 15:46:46 +00:00
auto item = processor->pItem;
if (item->IsRunOnTick())
{
if (!AuTryRemove(this->items.onTickReceivers, processor))
{
SysPushErrorMem();
}
}
if (item->IsRunOnOtherTick())
{
if (!AuTryRemove(this->items.onOtherReceivers, processor))
{
SysPushErrorMem();
}
}
if (item->IsRunOnSelfIO())
{
if (!AuTryRemove(this->items.registeredIO, processor))
{
SysPushErrorMem();
}
auto src = item->GetSelfIOSource();
if (!src)
{
SysPushErrorGeneric();
return;
}
if (!this->ToQueue()->SourceRemove(src))
{
SysPushErrorNested("IO Remove Error [!!!]");
}
}
processor->Finalize();
}
AuUInt IOProcessor::FrameFinalize()
{
auto C = this->items.finalizeQueue.size();
this->items.workSignaled.clear();
this->items.finalizeQueue.clear();
ReportState(EIOProcessorEventStage::eFrameEndOfFrame);
this->bFrameStart = false;
return C;
}
bool IOProcessor::InternalIsTickReady()
{
// TODO: tickless io ad-hoc event-as-condvar?
return this->items.workSignaled.size() ||
[+] Network + Protocol + TLS - Initial Commit ============================================================================= Network ]==================================================================== ============================================================================= [+] Added (very) early Aurora::IO::Net implementation [+] AuNet::EHostnameType [+] AuNet::EIPProtocol [+] AuNet::ENetworkError [+] AuNet::ETransportProtocol [+] AuNet::INetInterface [+] AuNet::INetSrvDatagram [+] AuNet::INetSrvResolve [+] AuNet::INetSrvSockets [+] AuNet::INetSrvWorkers [+] AuNet::INetWorker [+] AuNet::IPAddress [+] AuNet::IResolver [+] AuNet::ISocket [+] AuNet::IResolver [+] AuNet::ISocketBase [+] AuNet::ISocketChannel [+] AuNet::ISocketDriver [+] AuNet::ISocketDriverFactory [+] AuNet::ISocketServer [+] AuNet::ISocketServerDriver [+] AuNet::NetEndpoint [+] AuNet::NetError [+] AuNet::NetHostname (+implementation) ============================================================================= Protocol ]=================================================================== ============================================================================= [+] IProtocolInterceptor [+] IProtocolInterceptorEx [+] IProtocolStack (+implementation) ============================================================================= TLS ]======================================================================== ============================================================================= [+] ITLSContext [+] TLSProtocolRecv [+] TLSProtocolSend (+implementation) ============================================================================= IO Bug Fixes ]=============================================================== ============================================================================= [*] IOProcessor::SubmitIOWorkItem should signal the CvEvent, forcing at least once future tick (wont optimize with if in tick & not yet dispatched work items) [*] Split IOPipeWork in into IOPipeProcessor header [+] IOPipeWork::GetBuffer (internal reallocation) [*] Harden against IAsyncTransactions without a loop source [*] Missing null `if (processor->listener)` in IOProcessor [*] Solved some soft-lock conditions under Linux's LoopQueue (added deferred commits) [*] Quick hack: IOProcessor::HasItems() should OR the early can-tick check function. ============================================================================= Other ]====================================================================== ============================================================================= [+] Linux: LSSignalCatcher [+] `static void AuResetMember(Aurora::Memory::ByteBuffer &ref)` for AuROXTL [*] Attempt to enforce a normalization and don't overwrite-readptr-under-istreamwriters policy in ByteBuffer_ReadWrite (circular buffers) [*] Bad ECC ctors ============================================================================= Known issues ]=============================================================== ============================================================================= > Linux net is nowhere near done > UDP socket emulation layer isn't implemented > Ciphersuite API is a stub > Private key API is a stub > ...therefore no TLS servers > Missing thread safety precautions under net > Net implementation is still beri early
2022-08-28 19:02:06 +00:00
this->items.crossThreadAbort.size() ||
this->workUnits.size();
}
bool IOProcessor::FrameWaitForAny(AuUInt32 msMax)
{
if (InternalIsTickReady())
{
return true;
}
if (this->IsTickOnly())
{
auto next = this->timers.nbTicker.nextTriggerTime;
auto now = AuTime::SteadyClockNS();
if (now >= next)
{
return true;
}
auto delay = AuNSToMS<AuUInt>(next - now);
#if 0
if (delay == 1)
{
bool ret {};
while (IO::IOYield());
}
else
#endif
{
return IO::WaitFor(msMax ? AuMin<AuUInt32>(msMax, delay) : delay, true);
}
}
else
{
2022-08-29 15:46:46 +00:00
return this->pLoopQueue->WaitAny(msMax);
}
}
bool IOProcessor::SubmitIOWorkItem(const AuSPtr<IIOProcessorWorkUnit> &work)
{
AU_LOCK_GUARD(this->items.mutex); // < critical section / reentrant | can nest submission
[+] Network + Protocol + TLS - Initial Commit ============================================================================= Network ]==================================================================== ============================================================================= [+] Added (very) early Aurora::IO::Net implementation [+] AuNet::EHostnameType [+] AuNet::EIPProtocol [+] AuNet::ENetworkError [+] AuNet::ETransportProtocol [+] AuNet::INetInterface [+] AuNet::INetSrvDatagram [+] AuNet::INetSrvResolve [+] AuNet::INetSrvSockets [+] AuNet::INetSrvWorkers [+] AuNet::INetWorker [+] AuNet::IPAddress [+] AuNet::IResolver [+] AuNet::ISocket [+] AuNet::IResolver [+] AuNet::ISocketBase [+] AuNet::ISocketChannel [+] AuNet::ISocketDriver [+] AuNet::ISocketDriverFactory [+] AuNet::ISocketServer [+] AuNet::ISocketServerDriver [+] AuNet::NetEndpoint [+] AuNet::NetError [+] AuNet::NetHostname (+implementation) ============================================================================= Protocol ]=================================================================== ============================================================================= [+] IProtocolInterceptor [+] IProtocolInterceptorEx [+] IProtocolStack (+implementation) ============================================================================= TLS ]======================================================================== ============================================================================= [+] ITLSContext [+] TLSProtocolRecv [+] TLSProtocolSend (+implementation) ============================================================================= IO Bug Fixes ]=============================================================== ============================================================================= [*] IOProcessor::SubmitIOWorkItem should signal the CvEvent, forcing at least once future tick (wont optimize with if in tick & not yet dispatched work items) [*] Split IOPipeWork in into IOPipeProcessor header [+] IOPipeWork::GetBuffer (internal reallocation) [*] Harden against IAsyncTransactions without a loop source [*] Missing null `if (processor->listener)` in IOProcessor [*] Solved some soft-lock conditions under Linux's LoopQueue (added deferred commits) [*] Quick hack: IOProcessor::HasItems() should OR the early can-tick check function. ============================================================================= Other ]====================================================================== ============================================================================= [+] Linux: LSSignalCatcher [+] `static void AuResetMember(Aurora::Memory::ByteBuffer &ref)` for AuROXTL [*] Attempt to enforce a normalization and don't overwrite-readptr-under-istreamwriters policy in ByteBuffer_ReadWrite (circular buffers) [*] Bad ECC ctors ============================================================================= Known issues ]=============================================================== ============================================================================= > Linux net is nowhere near done > UDP socket emulation layer isn't implemented > Ciphersuite API is a stub > Private key API is a stub > ...therefore no TLS servers > Missing thread safety precautions under net > Net implementation is still beri early
2022-08-28 19:02:06 +00:00
this->items.cvEvent->Set();
return AuTryInsert(this->workUnits, work);
}
void IOProcessor::WakeupThread()
{
this->items.cvEvent->Set();
}
bool IOProcessor::AddEventListener(const AuSPtr<IIOProcessorEventListener> &eventListener)
{
AU_LOCK_GUARD(this->listenersSpinLock->AsWritable());
return AuTryInsert(this->listeners, eventListener);
}
void IOProcessor::RemoveEventListener(const AuSPtr<IIOProcessorEventListener> &eventListener)
{
AU_LOCK_GUARD(this->listenersSpinLock->AsWritable());
AuTryRemove(this->listeners, eventListener);
}
void IOProcessor::DispatchFrame(ProcessInfo &info)
{
this->ManualTick();
auto ns = this->refreshRateNs;
if (ns)
{
info = AuAsync::ETickType::eSchedule;
info.reschedNs = ns;
}
}
2022-06-14 01:14:51 +00:00
void IOProcessor::FramePumpWaitingBlocked()
{
auto blocked = this->items.GetBlockedSignals();
if (blocked.size())
{
this->items.workSignaled.insert(this->items.workSignaled.end(), blocked.begin(), blocked.end());
}
}
2022-06-14 01:14:51 +00:00
void IOProcessor::FrameStart()
{
ReportState(EIOProcessorEventStage::eFrameStartOfFrame);
this->bFrameStart = true;
FramePumpWaitingBlocked();
}
void IOProcessor::FrameRunThreadIO()
{
ReportState(EIOProcessorEventStage::eFrameYieldIO);
while (IO::IOYield());
ReportState(EIOProcessorEventStage::eFrameIODone);
}
AuUInt64 IOProcessor::SetRefreshRate(AuUInt64 ns)
{
if (!CheckThread())
{
AU_THROW_STRING("Wrong Thread");
}
bool bBinaryCanged = bool(ns) != bool(this->refreshRateNs);
auto old = AuExchange(this->refreshRateNs, ns);
UpdateTimers();
if (bBinaryCanged)
{
if (ns)
{
AddTimer();
}
else
{
RemoveTimer();
}
}
return old;
}
bool IOProcessor::HasRefreshRate()
{
return bool(this->refreshRateNs);
}
bool IOProcessor::MultiplexRefreshRateWithIOEvents()
{
return this->mutliplexIOAndTimer;
}
AuUInt64 IOProcessor::GetOwnedThreadId()
{
return this->threadId;
}
void IOProcessor::UpdateTimers()
{
2022-08-29 15:46:46 +00:00
this->timers.pLsTicker->UpdateTickRateIfAnyNs(this->refreshRateNs);
this->timers.nbTicker.nsTimeStep = this->refreshRateNs;
if (!this->timers.nbTicker.nextTriggerTime)
{
this->timers.nbTicker.nextTriggerTime = AuTime::SteadyClockNS() + this->timers.nbTicker.nsTimeStep;
}
}
void IOProcessor::AddTimer()
{
if (IsAsync())
{
StartAsyncTimerIfAny();
}
else
{
AddTimerLS();
}
}
void IOProcessor::AddTimerLS()
{
if (auto optWorker = this->asyncWorker)
{
AuAtomicAdd(&AuStaticCast<AuAsync::ThreadPool>(optWorker.value().GetPool())->uAtomicIOProcessorsWorthlessSources, 1u);
}
2022-08-29 15:46:46 +00:00
this->ToQueue()->SourceAdd(this->timers.pLsTicker);
this->ToQueue()->AddCallback(this->timers.pLsTicker, AuSPtr<AuLoop::ILoopSourceSubscriber>(AuSharedFromThis(), &this->timers));
this->ToQueue()->Commit();
}
void IOProcessor::StartAsyncTimerIfAny()
{
2022-08-29 15:46:46 +00:00
if (!this->pWorkItem)
{
this->pWorkItem = this->asyncWorker.value().GetPool()->NewWorkItem(this->asyncWorker.value(), AuSharedFromThis());
}
2022-08-29 15:46:46 +00:00
this->pWorkItem->SetSchedTimeNs(this->refreshRateNs);
2022-08-29 15:46:46 +00:00
this->pWorkItem->Dispatch();
}
void IOProcessor::RemoveTimer()
{
if (IsAsync())
{
CancelWorkItem();
}
else
{
RemoveLSTimer();
}
}
void IOProcessor::CancelWorkItem()
{
2022-08-29 15:46:46 +00:00
if (!this->pWorkItem)
2022-06-14 01:14:51 +00:00
{
return;
}
2022-08-29 15:46:46 +00:00
this->pWorkItem->Cancel();
this->pWorkItem.reset();
}
void IOProcessor::RemoveLSTimer()
{
2022-06-14 01:14:51 +00:00
auto queue = this->ToQueue();
if (!queue)
{
return;
}
2022-08-29 15:46:46 +00:00
queue->SourceRemove(this->timers.pLsTicker);
if (auto optWorker = this->asyncWorker)
{
AuAtomicSub(&AuStaticCast<AuAsync::ThreadPool>(optWorker.value().GetPool())->uAtomicIOProcessorsWorthlessSources, 1u);
}
}
bool IOProcessor::IsAsync()
{
return bool(this->asyncWorker);
}
bool IOProcessor::IsTickOnly()
{
return !this->mutliplexIOAndTimer &&
this->timers.nbTicker.nextTriggerTime;
}
2022-08-29 15:46:46 +00:00
bool IOProcessor::RequestRemovalForItemFromAnyThread(const AuSPtr<IIOProcessorItem> &pProcessor)
{
return {};
}
2022-08-29 15:46:46 +00:00
AuSPtr<IIOProcessorItem> IOProcessor::StartIOWatchEx(const AuSPtr<IIOWaitableItem> &pItem,
const AuSPtr<IIOEventListener> &pListener,
bool bSingleshot)
{
auto item = AuMakeShared<IOProcessorItem>();
if (!item)
{
SysPushErrorMemory();
return {};
}
2022-08-29 15:46:46 +00:00
item->pParent = this;
item->pListener = pListener;
item->pItem = pItem;
item->bSingleshot = bSingleshot;
AU_LOCK_GUARD(this->items.mutex);
2024-03-13 05:37:51 +00:00
if (!AuTryInsert(this->items.allItems, AuConstReference(item)))
{
return {};
}
2022-08-29 15:46:46 +00:00
if (pItem->IsRunOnTick())
{
2024-03-03 22:09:19 +00:00
if (!AuTryInsert(this->items.onTickReceivers, AuConstReference(item)))
{
2024-03-13 05:37:51 +00:00
SysPushErrorMemory();
return {};
}
}
2022-08-29 15:46:46 +00:00
if (pItem->IsRunOnOtherTick())
{
2024-03-03 22:09:19 +00:00
if (!AuTryInsert(this->items.onOtherReceivers, AuConstReference(item)))
{
2024-03-13 05:37:51 +00:00
SysPushErrorMemory();
return {};
}
}
2022-08-29 15:46:46 +00:00
if (pItem->IsRunOnSelfIO())
{
2022-08-29 15:46:46 +00:00
auto src = pItem->GetSelfIOSource();
if (!src)
{
SysPushErrorGeneric();
return {};
}
2022-08-29 15:46:46 +00:00
auto timeout = pItem->IOTimeoutInMS();
if (timeout)
{
if (!this->ToQueue()->SourceAddWithTimeout(src, timeout))
{
SysPushErrorNested("IO Error");
return {};
}
}
else
{
if (!this->ToQueue()->SourceAdd(src))
{
SysPushErrorNested("IO Error");
return {};
}
}
if (!this->ToQueue()->AddCallbackEx(src, item))
{
SysPushErrorNested("IO Error");
SysAssert(this->ToQueue()->SourceRemove(src));
return {};
}
if (!this->ToQueue()->Commit())
{
SysPushErrorIO("midframe recommit");
}
2024-03-03 22:09:19 +00:00
if (!AuTryInsert(this->items.registeredIO, AuConstReference(item)))
{
SysPushErrorMem();
SysAssert(this->ToQueue()->SourceRemove(src));
return {};
}
}
if (pItem->CanRequestTick())
{
pItem->OnReportPumper(item);
}
return item;
}
AuSPtr<IIOProcessorItem> IOProcessor::StartSimpleIOWatchEx(const AuSPtr<IIOWaitableItem>& object, const AuSPtr<IIOSimpleEventListener>& listener, bool singleshot)
{
[+] Network + Protocol + TLS - Initial Commit ============================================================================= Network ]==================================================================== ============================================================================= [+] Added (very) early Aurora::IO::Net implementation [+] AuNet::EHostnameType [+] AuNet::EIPProtocol [+] AuNet::ENetworkError [+] AuNet::ETransportProtocol [+] AuNet::INetInterface [+] AuNet::INetSrvDatagram [+] AuNet::INetSrvResolve [+] AuNet::INetSrvSockets [+] AuNet::INetSrvWorkers [+] AuNet::INetWorker [+] AuNet::IPAddress [+] AuNet::IResolver [+] AuNet::ISocket [+] AuNet::IResolver [+] AuNet::ISocketBase [+] AuNet::ISocketChannel [+] AuNet::ISocketDriver [+] AuNet::ISocketDriverFactory [+] AuNet::ISocketServer [+] AuNet::ISocketServerDriver [+] AuNet::NetEndpoint [+] AuNet::NetError [+] AuNet::NetHostname (+implementation) ============================================================================= Protocol ]=================================================================== ============================================================================= [+] IProtocolInterceptor [+] IProtocolInterceptorEx [+] IProtocolStack (+implementation) ============================================================================= TLS ]======================================================================== ============================================================================= [+] ITLSContext [+] TLSProtocolRecv [+] TLSProtocolSend (+implementation) ============================================================================= IO Bug Fixes ]=============================================================== ============================================================================= [*] IOProcessor::SubmitIOWorkItem should signal the CvEvent, forcing at least once future tick (wont optimize with if in tick & not yet dispatched work items) [*] Split IOPipeWork in into IOPipeProcessor header [+] IOPipeWork::GetBuffer (internal reallocation) [*] Harden against IAsyncTransactions without a loop source [*] Missing null `if (processor->listener)` in IOProcessor [*] Solved some soft-lock conditions under Linux's LoopQueue (added deferred commits) [*] Quick hack: IOProcessor::HasItems() should OR the early can-tick check function. ============================================================================= Other ]====================================================================== ============================================================================= [+] Linux: LSSignalCatcher [+] `static void AuResetMember(Aurora::Memory::ByteBuffer &ref)` for AuROXTL [*] Attempt to enforce a normalization and don't overwrite-readptr-under-istreamwriters policy in ByteBuffer_ReadWrite (circular buffers) [*] Bad ECC ctors ============================================================================= Known issues ]=============================================================== ============================================================================= > Linux net is nowhere near done > UDP socket emulation layer isn't implemented > Ciphersuite API is a stub > Private key API is a stub > ...therefore no TLS servers > Missing thread safety precautions under net > Net implementation is still beri early
2022-08-28 19:02:06 +00:00
if (!listener)
{
return this->StartIOWatchEx(object, {}, singleshot);
}
2022-06-29 13:56:59 +00:00
auto adapter = DesimplifyIOEventListenerAdapter(listener);
if (!adapter)
{
SysPushErrorMem();
return {};
}
2022-06-29 13:56:59 +00:00
return this->StartIOWatchEx(object, adapter, singleshot);
}
AuSPtr<IIOProcessorItem> IOProcessor::StartSimpleLSWatchEx(const AuSPtr<Loop::ILoopSource>& source, const AuSPtr<IIOSimpleEventListener>& listener, bool singleshot, AuUInt32 msTimeout)
{
auto sourceAdapter = NewWaitableLoopSourceEx(source, msTimeout);
if (!sourceAdapter)
{
SysPushErrorMem();
return {};
}
return this->StartSimpleIOWatchEx(sourceAdapter, listener, singleshot);
}
AuSPtr<IIOProcessorItem> IOProcessor::StartIOWatch(const AuSPtr<IIOWaitableItem> &object, const AuSPtr<IIOEventListener> &listener)
{
return StartIOWatchEx(object, listener, false);
}
AuSPtr<IIOProcessorItem> IOProcessor::StartSimpleIOWatch(const AuSPtr<IIOWaitableItem> &object, const AuSPtr<IIOSimpleEventListener> &listener)
{
return this->StartSimpleIOWatchEx(object, listener, false);
}
AuSPtr<IIOProcessorItem> IOProcessor::StartSimpleLSWatch(const AuSPtr<Loop::ILoopSource> &source, const AuSPtr<IIOSimpleEventListener> &listener)
{
return this->StartSimpleLSWatchEx(source, listener, false, 0);
}
AuSPtr<IIOPipeProcessor> IOProcessor::ToPipeProcessor()
{
if (!CheckThread())
{
AU_THROW_STRING("Wrong Thread");
}
return AuSPtr<IIOPipeProcessor>(AuSharedFromThis(), &this->streamProcessors);
}
AuSPtr<AuLoop::ILoopQueue> IOProcessor::ToQueue()
{
2022-08-29 15:46:46 +00:00
return this->pLoopQueue;
}
bool IOProcessor::ConfigureNoDeferredTicks(bool bOption)
{
return AuExchange(this->bNoDeferredTick, bOption);
}
void IOProcessor::ReleaseAllWatches()
{
if (auto optWorker = this->asyncWorker)
{
AuAtomicSub(&AuStaticCast<AuAsync::ThreadPool>(optWorker.value().GetPool())->uAtomicIOProcessors, 1u);
}
2022-06-14 01:14:51 +00:00
RemoveTimer();
2022-06-14 01:14:51 +00:00
auto queue = ToQueue();
2022-06-14 01:14:51 +00:00
if (queue)
{
if (this->items.cvEvent)
{
queue->SourceRemove(this->items.cvEvent);
}
for (auto &io : this->items.registeredIO)
{
2022-08-29 15:46:46 +00:00
queue->SourceRemove(io->pItem->GetSelfIOSource());
}
2022-06-14 01:14:51 +00:00
queue->Commit();
}
}
bool IOProcessor::HasItems()
{
[+] Network + Protocol + TLS - Initial Commit ============================================================================= Network ]==================================================================== ============================================================================= [+] Added (very) early Aurora::IO::Net implementation [+] AuNet::EHostnameType [+] AuNet::EIPProtocol [+] AuNet::ENetworkError [+] AuNet::ETransportProtocol [+] AuNet::INetInterface [+] AuNet::INetSrvDatagram [+] AuNet::INetSrvResolve [+] AuNet::INetSrvSockets [+] AuNet::INetSrvWorkers [+] AuNet::INetWorker [+] AuNet::IPAddress [+] AuNet::IResolver [+] AuNet::ISocket [+] AuNet::IResolver [+] AuNet::ISocketBase [+] AuNet::ISocketChannel [+] AuNet::ISocketDriver [+] AuNet::ISocketDriverFactory [+] AuNet::ISocketServer [+] AuNet::ISocketServerDriver [+] AuNet::NetEndpoint [+] AuNet::NetError [+] AuNet::NetHostname (+implementation) ============================================================================= Protocol ]=================================================================== ============================================================================= [+] IProtocolInterceptor [+] IProtocolInterceptorEx [+] IProtocolStack (+implementation) ============================================================================= TLS ]======================================================================== ============================================================================= [+] ITLSContext [+] TLSProtocolRecv [+] TLSProtocolSend (+implementation) ============================================================================= IO Bug Fixes ]=============================================================== ============================================================================= [*] IOProcessor::SubmitIOWorkItem should signal the CvEvent, forcing at least once future tick (wont optimize with if in tick & not yet dispatched work items) [*] Split IOPipeWork in into IOPipeProcessor header [+] IOPipeWork::GetBuffer (internal reallocation) [*] Harden against IAsyncTransactions without a loop source [*] Missing null `if (processor->listener)` in IOProcessor [*] Solved some soft-lock conditions under Linux's LoopQueue (added deferred commits) [*] Quick hack: IOProcessor::HasItems() should OR the early can-tick check function. ============================================================================= Other ]====================================================================== ============================================================================= [+] Linux: LSSignalCatcher [+] `static void AuResetMember(Aurora::Memory::ByteBuffer &ref)` for AuROXTL [*] Attempt to enforce a normalization and don't overwrite-readptr-under-istreamwriters policy in ByteBuffer_ReadWrite (circular buffers) [*] Bad ECC ctors ============================================================================= Known issues ]=============================================================== ============================================================================= > Linux net is nowhere near done > UDP socket emulation layer isn't implemented > Ciphersuite API is a stub > Private key API is a stub > ...therefore no TLS servers > Missing thread safety precautions under net > Net implementation is still beri early
2022-08-28 19:02:06 +00:00
bool bVal = bool(this->items.allItems.size()) ||
bool(this->workUnits.size()) || this->InternalIsTickReady();
return bVal;
}
void IOProcessor::ReportState(EIOProcessorEventStage stage)
{
AU_LOCK_GUARD(this->listenersSpinLock->AsReadable());
for (auto &listeners : this->listeners)
{
try
{
listeners->OnIOProcessorStateEvent(stage);
}
catch (...)
{
SysPushErrorCatch();
}
}
}
bool IOProcessor::CheckThread()
{
if (this->threadId == 0)
{
this->threadId = AuThreads::GetThreadId();
return true;
}
return this->threadId == AuThreads::GetThreadId();
}
static AuSPtr<IIOProcessor> NewIOProcessorEx(bool tickOnly, const AuSPtr<AuLoop::ILoopQueue> &queue, bool bIsNoQueue)
{
auto processor = AuMakeShared<IOProcessor>(0, tickOnly, AuOptionalEx<AuAsync::WorkerPId_t> {}, queue, bIsNoQueue);
SysCheckNotNullMemory(processor, {});
if (!processor->Init())
{
SysPushErrorNested();
return {};
}
return processor;
}
AUKN_SYM AuSPtr<IIOProcessor> NewIOProcessor(bool tickOnly, const AuSPtr<AuLoop::ILoopQueue> &queue)
{
auto processor = AuMakeShared<IOProcessor>(0, tickOnly, AuOptionalEx<AuAsync::WorkerPId_t> {}, queue, false);
SysCheckNotNullMemory(processor, {});
if (!processor->Init())
{
SysPushErrorNested();
return {};
}
return processor;
}
AUKN_SYM AuSPtr<IIOProcessor> NewIOProcessorOnThread(bool tickOnly, AuAsync::WorkerPId_t id)
{
if (!id.GetPool())
{
SysPushErrorArg();
return {};
}
auto thread = id.GetPool()->ResolveHandle(id);
if (!thread)
{
SysPushErrorGeneric("Worker PID failed to resolve to a thread object");
return {};
}
auto queue = id.GetPool()->ToKernelWorkQueue(id);
if (!queue)
{
SysPushErrorGeneric("Worker PID has no kernel work queue");
return {};
}
auto processor = AuMakeShared<IOProcessor>(AuUInt(thread.get()), tickOnly, id, queue, false);
SysCheckNotNullMemory(processor, {});
if (!processor->Init())
{
SysPushErrorNested();
return {};
}
return processor;
}
AUKN_SYM AuSPtr<IIOProcessor> NewIOProcessorNoQueue(bool tickOnly)
{
auto loop = AuLoop::NewLoopQueue();
SysCheckNotNull(loop, {});
return NewIOProcessorEx(tickOnly, loop, true);
}
}