AuroraSupport/AuroraRuntime
2
1
Fork 0
Code Issues 27 Pull Requests Releases Wiki Activity

[+] AuLoop::kWaitMultipleFlagNoIOCallbacks

Browse Source 
[+] AuLoop::kWaitMultipleFlagBreakAfterAPC
[+] Alternative Wait AND implementations for NT, POSIX, and generic
[+] IOConfig::...
[*] LoopQueue improvements
[+] ILoopQueue::ConfigureDoIOApcCallbacks
[+] ILoopQueue::ConfigureDoIOApcCallbacks
This commit is contained in:
Reece Wilson 2024-10-10 11:03:26 +01:00
parent 82ed6e5617
commit 02826d2365
18 changed files with 1380 additions and 261 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
Include/Aurora/IO/Loop/ILoopQueue.hpp
View File

@ -221,6 +221,18 @@ namespace Aurora::IO::Loop
* @brief Hints that the calling program understands the kernel shouldnt schedule tne entire source list, and instead, we should long poll
*/
virtual void ChugHint(bool value) = 0;
/*
@brief Do io completion callbacks?
Default: true
*/
virtual void ConfigureDoIOApcCallbacks(bool bDoIOAPCs = true) = 0;
/*
@brief Do "any" paths give up upon APC-like callback? Or should the thread go back to sleep until a provided SourceAdd has awoken?
Default: false
*/
virtual void ConfigureBreakAnyAfterAPC(bool bAPCBreaks = false) = 0;
};
AUKN_SYM AuSPtr<ILoopQueue> NewLoopQueue();

66
Include/Aurora/IO/Loop/Loop.hpp
View File

@ -19,25 +19,41 @@
namespace Aurora::IO::Loop
{
// TODO: We cannot reach true parity with NTs WaitForMultipleObjectsEx in this current state ( TODO )
// [1] There is no flag for enter/do-not-enter alertable state (processes file transactions and their subscribers, network transactions and their subscribers, APCs, etc)
// (also see: ./../IOSleep.hpp)
// [2] There is no explicit flag for awoken by APC ( could use true && signaled.empty() ).
// [3] And we can't do timer callbacks in-thread with the state of our current APC implementation ( TODO ).
// optTimeoutMS = {} | indefinite
// optTimeoutMS = 0 | poll
// optTimeoutMS = 1 | 1ms
AUKN_SYM bool WaitMultipleLoopSources(const AuList<AuSPtr<Loop::ILoopSource>> &lsList,
AuList<AuSPtr<Loop::ILoopSource>> &signaled,
bool bAny = { true },
AuOptional<AuUInt32> optTimeoutMS = {});
static const AuUInt64 kWaitMultipleFlagAny = 1 << 0;
static const AuUInt64 kWaitMultipleFlagAny = 1 << 0; // ~= NT: ~MWMO_WAITALL, !bWaitAll
static const AuUInt64 kWaitMultipleFlagNoSpin = 1 << 1;
static const AuUInt64 kWaitMultipleFlagAvoidKern = 1 << 2;
static const AuUInt64 kWaitMultipleFlagBreakAfterOne = 1 << 3;
static const AuUInt64 kWaitMultipleFlagBreakAfterOne = 1 << 3; // ~= NT: WaitForMultipleObjects return value
static const AuUInt64 kWaitMultipleFlagBreakAfterAPC = 1 << 4; // ~= NT: WAIT_IO_COMPLETION return value
static const AuUInt64 kWaitMultipleFlagNoIOCallbacks = 1 << 5; // ~= NT: !bAlert, !fAlert, ~MWMO_ALERTABLE
static const AuUInt64 kWaitMultipleFlagPragAnyAPC = 1 << 6; // See notes
/**
* Parameter "signaled" indicates the exact loop sources that were signaled.
* If this "signaled" array is empty, and WaitMultipleLoopSourcesEx returned true, then an APC was served.
* In addition, win32 waitalls may return an empty signaled[] if lsList contained a win32 message queue loop source.
*
* If kWaitMultipleFlagAny is specified, signaled may contain any amonut of lsList entries
* If kWaitMultipleFlagAny is not specified, signaled must be an array of the same elements, or empty if some other reason caused the wait to beak.
*
* If kWaitMultipleFlagBreakAfterOne and kWaitMultipleFlagAny are specified, signaled will only contain no more than one entry.
*
* @returns false: timeout
* true: APC dispatched, io callback dispatched, all of (~uFlags) & kWaitMultipleFlagAny woke up, or at least one of (uFlags) & kWaitMultipleFlagAny woke up.
*
* @warning depending on the platform, and provided parameters, wait all may consume semaphore and event objects in order, followed by any mutexes in order.
* @warning depending on the platform, kWaitMultipleFlagPragAnyAPC may be required to recieve IAsyncTransaction::SetCallback callbacks of transaction objects not covered by lsList sources.
* @warning IAsyncTransaction::SetCallback does not respect kWaitMultipleFlagNoIOCallbacks.
* @warning kWaitMultipleFlagNoIOCallbacks only filters user queued APCs and implicit io callbacks.
*/
AUKN_SYM bool WaitMultipleLoopSourcesEx(const AuList<AuSPtr<Loop::ILoopSource>> &lsList,
AuList<AuSPtr<Loop::ILoopSource>> &signaled,
AuUInt64 uFlags = { kWaitMultipleFlagAny },
@ -141,4 +157,38 @@ namespace Aurora::IO::Loop
#if defined(AURORA_IS_LINUX_DERIVED)
AUKN_SYM AuSPtr<ILSSignalCatcher> NewLSSignalCatcher(const AuList<int> &signals);
#endif
// Note for Win32-like emulators (Xbox, Xbox 360, GameOS sandboxes, etc)
// Aurora does not parrot exact win32 smenatics.
// On OVERLAPPED COMPLETIONs, you should implement an APC queue yourself, and dequeue notificaitons once WaitForMultipleLoopSources completes.
// On SetTimer callback, aurora does not * yet * implement this, even tho we could soon and other oses do.
// Attach all thread local timers to the input loopsource list and dispatch the cb in your emulated APC queue.
// Usage of kWaitMultipleFlagBreakAfterAPC may be used by some users as a replacement for WAIT_IO_COMPLETION; to others, it returns spuriously far too many times.
// Async file transaction callbacks will occur even in non-alert mode, if you explicitly specify an IAsyncTransactions' loop source.
// Additional ICompletionGroup callbacks may occur even in non-alert mode, if you explicitly sleep on its' loop source.
// WaitMultipleLoopSourcesEx makes for a really good stateless yield function for these users; however, a naive shim cannot be used to recreate WaitForMultipleObjectsEx.
// ...likewise, network loop source quirks means you might need to alert sleep under some oses.
// ...likewise, async transaction loop source quirks means the b/fAlert state of the emulated application
// Note on limits:
// lsList should be kept under NTs limit of 64; however, you can go above this with poor long polling.
// a user facing application on NT will not simply throw its' hands into the air the moment it exceeds Microsofts limit.
// Both Linux and Win32 have a fallback for their fast path (POSIX.1-2008 poll, WaitForMultipeObjects) using an ILoopQueue (epoll, also WaitForMultipeObjects).
// POSIX and NT platforms should use stateless kernel resources on WaitMultipleLoopSources[Ex] calls.
// NT: lsList.size() <= 64 generally does not allocate so much (also noting: signaled return array values always alloc with reserved memory active).
// Linux: lsList.size() <= 128 generally does not allocate so much (also noting: signaled return array values always alloc with reserved memory active).
// Notes on callback lists
// All allocations done under WaitMultipleLoopSources are performed under a AU_DEBUG_MEMCRUNCH scope.
// Notes on IO callbacks
// If no io callback suppression flag is provided, the application always:
// * enters an NT bAlert=true/MWMO_ALERTABLE sleep,
// * dispatches completed io transactions underneath sleep routines (ILoopSource::WaitOn(...), AuIO::IOSleep(...), AuIO::IOYield(), etc)
// Notes all around
// If you squint your eyes, or realize all of the edge cases in emulating Win32 do not apply to your high level use cases,
// WaitMultipleLoopSources[Ex] APIs are great for porting Win32 applications to POSIX. The primitives defined in Loop.hpp
// should match the expected behaviour of Win9x / NT sync primitives, except in some cases where they can utilize fast paths
// to bypass the kernel all together.
}

31
Include/Aurora/RuntimeConfig.hpp
View File

@ -189,6 +189,8 @@ namespace Aurora
bool bWin32VerifyTrustFailMissingAPI { true }; // just to not be annoying in the future
};
// Checkout some of the members towards the top.
// Do not adjust the bitfield entries towards the bottom.
struct ThreadingConfig
{
// WARN: these values are not final
@ -205,10 +207,10 @@ namespace Aurora
// This is comparable to Win32's SetCriticalSectionSpinCount applied across every single AuThreadPrimitives try-lock and lock.
// Adjust this value to compensate for longer critical sections when context switching isn't preferrable.
// Using 128 as a default (bouncing around 64 and 512)
// Facebook says half this (cant find src), I used to say about 82 to 512, Windows 7s implementation of CRITICAL_SECTION and SRWLOCK says double that (256), for aggressive (and now incorrect) spin mutex examples ive seen around 2k or less, for not so aggressive pause loops ive seen people use 32-128-ish pauses (also incorrect), dumb shits parroting Win9x documentation and SetCriticalSectionSpinCount example value think you need above >= 4k (stackexchange man strike again).
// Facebook used to say half this (cant find src), I used to say about 82 to 512, Windows 7s implementation of CRITICAL_SECTION and SRWLOCK says double that (256), for aggressive (and now incorrect) spin mutex examples ive seen around 2k or less, some intel reference material uses 64 as a demo max spin value, for not so aggressive pause loops ive seen people use 32-128-ish pauses (also incorrect), dumb shits parroting Win9x documentation and SetCriticalSectionSpinCount's example value think you need above >= 4k (stackexchange man strike again).
// Personally, I've seen this tested on 5-12th gen intel, Windows 7 through 11, Linux, and various other configurations.
// Personally, I've seen this run Qt with less CPU resources than every other Qt process on Win7. I've seen this run JavaScript programs dead last on the taskmanagers detail panel, on both 10 and 7.
// 128 to 512 is fine, unless you need to start asserting you are a real time application aware of your hardware requirements / have properly matched task affinity / etc, and don't mind shredding old processor power efficiency while chewing thru nop cycles
// 128 to 512 is fine. on the upper end you, the developer, need to start asserting you are a real time application aware of your hardware requirements / have properly matched task affinity / etc, and don't mind shredding old processor power efficiency while chewing thru nop cycles
// <<<<<<<<<<<<<<< (QA:) Each applications will probably need its own nudge value
AuUInt64 bEnableAggressiveScheduling : 1 AU_BIT_FIELD_INIT_AFTER_20( false ); // <<<<<<<<<<<<<<< (SHIP:) ENABLE ME FOR AROUND 1MS OR LESS SCHED RESOLUTION
AuUInt64 bEnableAgrSchedulingRatelimit : 1 AU_BIT_FIELD_INIT_AFTER_20( true );
@ -303,6 +305,20 @@ namespace Aurora
//
bool bAPCUseCoroutineStack { true };
// Decreases syscall and global lock contension if false.
// Enable me if youre emulating Win32 or NT as a platform.
// This helps mitigate waitables returning false during multiple AND-mode acquisition.
bool bAimCloserForNTParityWaitALL { false };
bool bUseOldIOWaitAllAlg { false };
bool bUseSelectWaitAllStrat { true };
// Enable me if youre emulating Win32 or NT as a platform, leave configurable.
// If the emulated target can run without this, which is very very likely, this will save a ass-ton of syscalls.
// Affected platforms: NT-likes only. Win32-emu-on-POSIX can use in-process primitives all they like.
bool bINeedPerfectAnds { false };
};
struct Win32Config
@ -310,6 +326,17 @@ namespace Aurora
bool bProcessCheckWinLdrForModNameFirst { true };
};
// ADHD/shipping check-list version:
// Go check:
// ioConfig.bIsVeryLargeIOApplication | for increased memory overhead, allows servers to open more io completion contexts. looks bad on linux client applications.
// fio.optDefaultBrand | for application branding. change this to your publishers name. used for configuration and ~home isolation.
// threadingConfig.bEnableAggressiveScheduling | for real time applications. required for retarded timing resolution coalescence. 0.0Xms to 0.3MS tier resolution is viable on modern PC platforms (say <250,000NS).
// threadingConfig.bNoThreadNames | disable vendor libraries from specifying their thread name to an attached debugger, if not stripped from the application.
// threadingConfig.uSpinLoopPowerA | increase me if the *global* context switch rate is too high. use AuThreading:: APIs is it's a per-thread issue.
// async.dwSchedulerRateLimitNS | for sub 2MS AuAsync timers and Windows 7 timers.
// | Real-time applications should set this to 0.
// | Interactive applications should lower this to 500'000 nanoseconds (.5MS) to 1'000'000ns (1MS).
// | GUI applications should keep this value high to prevent high idle CPU usage.
struct RuntimeStartInfo
{
ConsoleConfig console;

11
Source/IO/Loop/LSFromFdNonblocking.hpp
View File

@ -18,6 +18,8 @@ namespace Aurora::IO::UNIX
namespace Aurora::IO::Loop
{
extern AuRWRenterableLock gWaitForMultipleALLLock;
template<typename T>
inline bool IsSignaledFromNonblockingImpl(ILoopSourceEx *source, T * that, bool(T::*IsSignaledNonblocking)(), bool bIOTick)
{
@ -25,6 +27,15 @@ namespace Aurora::IO::Loop
source->OnPresleep();
val = ((that)->*(IsSignaledNonblocking))();
source->OnFinishSleep();
#if !defined(AU_NO_WAITMULTIPLELS_ALL_MS_PARITY)
if (!val && gRuntimeConfig.ioConfig.bAimCloserForNTParityWaitALL)
{
AU_LOCK_GLOBAL_GUARD(gWaitForMultipleALLLock->AsReadable());
source->OnPresleep();
val = ((that)->*(IsSignaledNonblocking))();
source->OnFinishSleep();
}
#endif
// Required for precise Windows on Linux OVERLAPPED semantics for emulators.
// IAsyncTransaction::Complete()/Wait() will always enter an alertable state and dispatch io completion callbacks.
// In theory, we don't need this.

21
Source/IO/Loop/LSLocalEvent.cpp
View File

@ -11,6 +11,8 @@
namespace Aurora::IO::Loop
{
extern AuRWRenterableLock gWaitForMultipleALLLock;
LSLocalEvent::LSLocalEvent()
{
@ -213,10 +215,20 @@ namespace Aurora::IO::Loop
bool LSLocalEvent::TryTakeSpin()
{
return Threading::Primitives::DoTryIfAlderLake([&]
bool bRet = Threading::Primitives::DoTryIfAlderLake([&]
{
return this->TryTakeNoSpin();
}, &this->state_);
#if !defined(AU_NO_WAITMULTIPLELS_ALL_MS_PARITY)
if (!bRet && gRuntimeConfig.ioConfig.bAimCloserForNTParityWaitALL)
{
AU_LOCK_GLOBAL_GUARD(gWaitForMultipleALLLock->AsReadable());
bRet = this->TryTakeNoSpin();
}
#endif
return bRet;
}
bool LSLocalEvent::IsSignaledNoSpinIfUserland()
@ -274,6 +286,13 @@ namespace Aurora::IO::Loop
AUKN_SYM AuSPtr<ILSEvent> NewLSEvent(bool bTriggered, bool bAtomicRelease, bool bPermitMultipleTriggers)
{
#if defined(AURORA_IS_MODERNNT_DERIVED)
if (gRuntimeConfig.ioConfig.bINeedPerfectAnds)
{
return NewLSEventSlow(bTriggered, bAtomicRelease, bPermitMultipleTriggers);
}
#endif
auto pMutex = AuMakeShared<LSLocalEvent>();
if (!pMutex)
{

20
Source/IO/Loop/LSLocalMutex.cpp
View File

@ -12,6 +12,7 @@
namespace Aurora::IO::Loop
{
static auto const kFutexBitWake = 256u;
extern AuRWRenterableLock gWaitForMultipleALLLock;
LSLocalMutex::LSLocalMutex()
{
@ -103,10 +104,20 @@ namespace Aurora::IO::Loop
bool LSLocalMutex::TryTakeSpin()
{
return Threading::Primitives::DoTryIfAlderLake([&]
bool bRet = Threading::Primitives::DoTryIfAlderLake([&]
{
return this->TryTakeNoSpin();
}, &this->uAtomicWord);
#if !defined(AU_NO_WAITMULTIPLELS_ALL_MS_PARITY)
if (!bRet && gRuntimeConfig.ioConfig.bAimCloserForNTParityWaitALL)
{
AU_LOCK_GLOBAL_GUARD(gWaitForMultipleALLLock->AsReadable());
bRet = this->TryTakeNoSpin();
}
#endif
return bRet;
}
bool LSLocalMutex::TryTake()
@ -159,6 +170,13 @@ namespace Aurora::IO::Loop
AUKN_SYM AuSPtr<ILSMutex> NewLSMutex()
{
#if defined(AURORA_IS_MODERNNT_DERIVED)
if (gRuntimeConfig.ioConfig.bINeedPerfectAnds)
{
return NewLSMutexSlow();
}
#endif
auto pMutex = AuMakeShared<LSLocalMutex>();
if (!pMutex)
{

1
Source/IO/Loop/LSLocalMutex.hpp
View File

@ -26,6 +26,7 @@ namespace Aurora::IO::Loop
virtual bool OnTrigger(AuUInt handle) override;
bool TryTakeNoSpin_();
bool TryTakeNoSpin();
bool TryTakeSpin();
bool TryTake();

21
Source/IO/Loop/LSLocalSemaphore.cpp
View File

@ -11,6 +11,8 @@
namespace Aurora::IO::Loop
{
extern AuRWRenterableLock gWaitForMultipleALLLock;
LSLocalSemaphore::LSLocalSemaphore()
{
@ -273,10 +275,20 @@ namespace Aurora::IO::Loop
bool LSLocalSemaphore::TryTakeSpin()
{
return Threading::Primitives::DoTryIfAlderLake([&]
bool bRet = Threading::Primitives::DoTryIfAlderLake([&]
{
return this->TryTakeNoSpin();
}, &this->uAtomicSemaphore);
#if !defined(AU_NO_WAITMULTIPLELS_ALL_MS_PARITY)
if (!bRet && gRuntimeConfig.ioConfig.bAimCloserForNTParityWaitALL)
{
AU_LOCK_GLOBAL_GUARD(gWaitForMultipleALLLock->AsReadable());
bRet = this->TryTakeNoSpin();
}
#endif
return bRet;
}
bool LSLocalSemaphore::TryTake()
@ -339,6 +351,13 @@ namespace Aurora::IO::Loop
AUKN_SYM AuSPtr<ILSSemaphore> NewLSSemaphore(AuUInt32 uInitialCount)
{
#if defined(AURORA_IS_MODERNNT_DERIVED)
if (gRuntimeConfig.ioConfig.bINeedPerfectAnds)
{
return NewLSSemaphoreSlow(uInitialCount);
}
#endif
auto pMutex = AuMakeShared<LSLocalSemaphore>();
if (!pMutex)
{

204
Source/IO/Loop/Loop.NT.cpp
View File

@ -10,13 +10,19 @@
#include "ILoopSourceEx.hpp"
#include "LSWin32.NT.hpp"
#define MAXIMUM_WAIT_OBJECTS_ (MAXIMUM_WAIT_OBJECTS - 1)
namespace Aurora::IO::Loop
{
AuList<AuSPtr<ILoopSource>> WaitMultipleOrObjects(const AuList<AuSPtr<ILoopSource>> &objects, bool bZeroTick, AuUInt32 dwTimeoutReq, bool bAllowOthers, bool &bTooMany)
extern AuRWRenterableLock gWaitForMultipleALLLock;
void ResetLoopSourceFalseAlarm(const AuSPtr<Loop::ILoopSource> &pLoopSource);
AuList<AuSPtr<ILoopSource>> WaitMultipleOrObjects(const AuList<AuSPtr<ILoopSource>> &objects, bool bAlert, bool bBreakAPCs, bool bZeroTick, AuUInt32 dwTimeoutReq, bool bAllowOthers, bool &bTooMany)
{
bool bIsWinLoop;
AuSPtr<ILoopSourceEx> loopSourceExs[MAXIMUM_WAIT_OBJECTS];
HANDLE handleArray[MAXIMUM_WAIT_OBJECTS];
AuSPtr<ILoopSourceEx> loopSourceExs[MAXIMUM_WAIT_OBJECTS_];
HANDLE handleArray[MAXIMUM_WAIT_OBJECTS_];
AuUInt32 uCounterHandle, uCounterFD;
AuSPtr<ILoopSource> pMsgSource;
AuList<AuSPtr<ILoopSource>> triggered;
@ -27,7 +33,7 @@ namespace Aurora::IO::Loop
try
{
triggered.reserve(triggered.size());
triggered.reserve(objects.size());
}
catch (...)
{
@ -52,7 +58,7 @@ namespace Aurora::IO::Loop
{
{
auto uNewIndex = uCounterHandle++;
if (uNewIndex == MAXIMUM_WAIT_OBJECTS)
if (uNewIndex == MAXIMUM_WAIT_OBJECTS_)
{
bTooMany = true;
return {};
@ -63,7 +69,7 @@ namespace Aurora::IO::Loop
if (extended->Singular())
{
auto uNewIndex = uCounterFD++;
if (uNewIndex == MAXIMUM_WAIT_OBJECTS)
if (uNewIndex == MAXIMUM_WAIT_OBJECTS_)
{
bTooMany = true;
return {};
@ -75,7 +81,7 @@ namespace Aurora::IO::Loop
for (const auto &handle : extended->GetHandles())
{
auto uNewIndex = uCounterFD++;
if (uNewIndex == MAXIMUM_WAIT_OBJECTS)
if (uNewIndex == MAXIMUM_WAIT_OBJECTS_)
{
bTooMany = true;
return {};
@ -103,11 +109,11 @@ namespace Aurora::IO::Loop
if (bIsWinLoop &&
pMsgWaitForMultipleObjectsEx)
{
ret = pMsgWaitForMultipleObjectsEx(uCounterFD, handleArray, uTimeout, QS_ALLPOSTMESSAGE | QS_ALLINPUT, MWMO_INPUTAVAILABLE | MWMO_ALERTABLE);
ret = pMsgWaitForMultipleObjectsEx(uCounterFD, handleArray, uTimeout, QS_ALLPOSTMESSAGE | QS_ALLINPUT, MWMO_INPUTAVAILABLE | (bAlert ? MWMO_ALERTABLE : 0));
}
else
{
ret = ::WaitForMultipleObjectsEx(uCounterFD, handleArray, false, uTimeout, true);
ret = ::WaitForMultipleObjectsEx(uCounterFD, handleArray, false, uTimeout, bAlert);
}
bool error = ((ret == WAIT_TIMEOUT) ||
@ -182,4 +188,184 @@ namespace Aurora::IO::Loop
return AuMove(triggered);
}
bool WaitMultipleAndObjects_(const AuList<AuSPtr<ILoopSource>> &objects, bool bAlert, bool bBreakAPCs, bool bZeroTick, AuUInt64 qwTimeoutAbs, bool &bTooMany, bool &bTimeout)
{
bool bIsWinLoop;
bool bRet;
AuSPtr<ILoopSourceEx> loopSourceExs[MAXIMUM_WAIT_OBJECTS_];
HANDLE handleArray[MAXIMUM_WAIT_OBJECTS_];
AuUInt32 uCounterHandle, uCounterFD;
AuSPtr<ILoopSource> pMsgSource;
bIsWinLoop = false;
uCounterHandle = 0;
uCounterFD = 0;
bRet = true;
for (const auto &source : objects)
{
if (!source)
{
continue;
}
if (source->GetType() == ELoopSource::eSourceWin32)
{
bIsWinLoop = true;
pMsgSource = source;
continue;
}
if (auto extended = AuDynamicCast<ILoopSourceEx>(source))
{
{
auto uNewIndex = uCounterHandle++;
if (uNewIndex == MAXIMUM_WAIT_OBJECTS_)
{
bTooMany = true;
return {};
}
loopSourceExs[uNewIndex] = extended;
}
if (extended->Singular())
{
auto uNewIndex = uCounterFD++;
if (uNewIndex == MAXIMUM_WAIT_OBJECTS_)
{
bTooMany = true;
return {};
}
handleArray[uNewIndex] = reinterpret_cast<HANDLE>(extended->GetHandle());
}
else
{
for (const auto &handle : extended->GetHandles())
{
auto uNewIndex = uCounterFD++;
if (uNewIndex == MAXIMUM_WAIT_OBJECTS_)
{
bTooMany = true;
return {};
}
handleArray[uNewIndex] = reinterpret_cast<HANDLE>(extended->GetHandle());
}
}
}
}
for (AU_ITERATE_N(i, uCounterHandle))
{
loopSourceExs[i]->OnPresleep();
}
DWORD ret;
do
{
AuUInt32 uTimeout;
if (bZeroTick)
{
uTimeout = 0;
}
else if (qwTimeoutAbs)
{
auto uNow = AuTime::SteadyClockNS();
if (uNow >= qwTimeoutAbs)
{
uTimeout = 0;
}
else
{
uTimeout = AuNSToMS<AuUInt32>(qwTimeoutAbs - uNow);
}
}
else
{
uTimeout = -1;
}
if (bIsWinLoop &&
pMsgWaitForMultipleObjectsEx)
{
ret = pMsgWaitForMultipleObjectsEx(uCounterFD, handleArray, uTimeout, QS_ALLPOSTMESSAGE | QS_ALLINPUT, MWMO_INPUTAVAILABLE | MWMO_WAITALL | (bAlert ? MWMO_ALERTABLE : 0));
}
else
{
ret = ::WaitForMultipleObjectsEx(uCounterFD, handleArray, TRUE, uTimeout, bAlert);
}
}
while (!bBreakAPCs && ret == WAIT_IO_COMPLETION);
bool error = ((ret == WAIT_TIMEOUT) ||
(ret == WAIT_IO_COMPLETION) ||
(ret == WAIT_FAILED));
if (WAIT_OBJECT_0 + uCounterFD == ret)
{
bTimeout = false;
bRet = false;
}
else if (error)
{
bTimeout = ret == WAIT_TIMEOUT;
bRet = false;
}
{
AU_LOCK_GLOBAL_GUARD(gWaitForMultipleALLLock->AsWritable());
for (AU_ITERATE_N(i, uCounterHandle))
{
auto &pSource = loopSourceExs[i];
if (bRet)
{
AuUInt lastHandle {};
bool wasTriggered {};
if (pSource->Singular())
{
auto handle = pSource->GetHandle();
lastHandle = handle;
}
else
{
for (const auto &handle : pSource->GetHandles())
{
lastHandle = handle;
break;
}
}
if (!pSource->OnTrigger(lastHandle))
{
for (AU_ITERATE_N(z, i))
{
ResetLoopSourceFalseAlarm(loopSourceExs[z]);
}
bRet = false;
break;
}
}
pSource->OnFinishSleep();
}
}
return bRet;
}
bool WaitMultipleAndObjects(const AuList<AuSPtr<ILoopSource>> &objects, bool bAlert, bool bBreakAPCs, bool bZeroTick, AuUInt64 qwTimeoutAbs, bool &bTooMany, bool &bTimeout)
{
bool bRet {};
do
{
bRet = WaitMultipleAndObjects_(objects, bAlert, bBreakAPCs, bZeroTick, qwTimeoutAbs, bTooMany, bTimeout);
}
while (!bRet && !bTimeout && !bTooMany);
return bRet;
}
}

308
Source/IO/Loop/Loop.Unix.cpp
View File

@ -22,7 +22,11 @@ namespace Aurora::IO::UNIX
namespace Aurora::IO::Loop
{
AuList<AuSPtr<ILoopSource>> WaitMultipleOrObjects(const AuList<AuSPtr<ILoopSource>> &objects, bool bZeroTick, AuUInt32 dwTimeoutReq, bool bAllowOthers, bool &bTooMany)
extern AuRWRenterableLock gWaitForMultipleALLLock;
void ResetLoopSourceFalseAlarm(const AuSPtr<Loop::ILoopSource> &pLoopSource);
AuList<AuSPtr<ILoopSource>> WaitMultipleOrObjects(const AuList<AuSPtr<ILoopSource>> &objects, bool bAlert, bool bBreakAPCs, bool bZeroTick, AuUInt32 dwTimeoutReq, bool bAllowOthers, bool &bTooMany)
{
#if defined(AURORA_HAS_NO_POSIX_POLL)
// do the WinNT falllback into loop queue.
@ -230,11 +234,15 @@ namespace Aurora::IO::Loop
int ret;
do
{
// TODO: bAlert awareness (cannot enter alert sleep yet. no such impl on linux)
// (we need an io_submit posix poll shim + apc event + queue that we dont have yet)
ret = poll(pHandleArray, uFDOffset, uTimeout);
uTimeout = 0;
if (ret > 0)
{
AU_LOCK_GLOBAL_GUARD(gWaitForMultipleALLLock->AsReadable());
for (AU_ITERATE_N(i, uFDOffset))
{
if (!pHandleArray[i].revents)
@ -275,7 +283,7 @@ namespace Aurora::IO::Loop
pLoopSourceExs2[i]->OnFinishSleep();
}
// TODO: ugly workaround (see: LSFromHdNonblocking rationale) for an ugly TODO issue implicating all targets (see public Loop.hpp)
if (bAlert)
{
#if defined(AURORA_IS_LINUX_DERIVED)
// Do not syscall after read or select again under Linux
@ -288,4 +296,300 @@ namespace Aurora::IO::Loop
return AuMove(triggered);
#endif
}
bool WaitMultipleAndObjects_(const AuList<AuSPtr<ILoopSource>> &objects, bool bAlert, bool bBreakAPCs, bool bZeroTick, AuUInt64 qwTimeoutAbs, bool &bTooMany, bool &bTimeout)
{
#if defined(AURORA_HAS_NO_POSIX_POLL)
// do the WinNT falllback into loop queue.
//
// loop queues should have long term persistency in a user-space ABI (kqueue) or in the kernel (epoll)
// they may take longer to setup, may seem wasteful to create and throw away, but it's probably the best we can do under BSD kernels.
// why? kevents exist for posix aio objects. they dont use fds.
//
// BSDs with posix AIO based async file transactions, >2.2 linux kernels, macos with corefoundation interop with kqueues, and other posix kernels with alternative aio will probably need
// need to use the win32 64 <= handles path.
bTooMany = true;
return false;
#else
AuList<AuSPtr<ILoopSource>> triggered;
// fallback
AuList<AuSPtr<ILoopSourceEx>> loopSourceExs1;
AuList<AuSPtr<ILoopSourceEx>> loopSourceExs21;
AuList<pollfd> handleArray1;
AuList<pollfd> handleArrayPersistent1;
AuList<AuPair<AuUInt, bool>> handleIndicies1;
// 4k~ of stack
AuSPtr<ILoopSourceEx> loopSourceExs12[128];
AuSPtr<ILoopSourceEx> loopSourceExs22[128];
pollfd handleArray2[128];
pollfd handleArrayPersistent2[128];
AuPair<AuUInt, bool> handleIndicies2[128];
// phead
AuSPtr<ILoopSourceEx> * pLoopSourceExs1;
AuSPtr<ILoopSourceEx> * pLoopSourceExs2;
pollfd * pHandleArray;
pollfd * pHandleArrayPersistent;
AuPair<AuUInt, bool> * pHandleIndicies;
AuUInt32 uHandleOffset, uFDOffset, uHandleOffset2;
// length / size
uHandleOffset = 0;
uHandleOffset2 = 0;
uFDOffset = 0;
// default heads
pLoopSourceExs1 = loopSourceExs12;
pLoopSourceExs2 = loopSourceExs22;
pHandleArray = handleArray2;
pHandleIndicies = handleIndicies2;
pHandleArrayPersistent = handleArrayPersistent2;
AuUInt32 uTotal { 0 };
#define HANDLE_PUSH_CHILD3(value) \
{ \
HANDLE_PUSH_CHILD_Z(value, loopSourceExs22, pHandleArrayPersistent, handleArrayPersistent1, handleArrayPersistent2, (uFDOffset - 1)) \
}
try
{
if (AuArraySize(loopSourceExs22) < objects.size())
{
loopSourceExs1.reserve(objects.size());
loopSourceExs21.reserve(objects.size());
handleArray1.reserve(objects.size());
handleIndicies1.reserve(triggered.size());
}
triggered.reserve(triggered.size());
for (const auto &source : objects)
{
if (!source)
{
continue;
}
if (auto extended = AuDynamicCast<ILoopSourceEx>(source))
{
auto uCurHandle = uHandleOffset++;
HANDLE_PUSH_MAIN2(extended)
if (extended->Singular())
{
auto handle = extended->GetHandle();
auto handleWrite = extended->GetWriteHandle();
auto i = uHandleOffset2++;
HANDLE_PUSH_MAIN(extended, i);
if (handle != -1)
{
pollfd poll;
poll.fd = handle;
poll.events = POLLIN;
poll.revents = 0;
HANDLE_PUSH_CHILD(poll);
HANDLE_PUSH_CHILD2(AuMakePair(i, false));
HANDLE_PUSH_CHILD3(poll)
}
if (handleWrite != -1)
{
pollfd poll;
poll.fd = handleWrite;
poll.events = POLLOUT;
poll.revents = 0;
HANDLE_PUSH_CHILD(poll);
HANDLE_PUSH_CHILD2(AuMakePair(i, false));
HANDLE_PUSH_CHILD3(poll)
}
}
else
{
auto handles = extended->GetHandles();
auto handlesWrite = extended->GetWriteHandles();
auto i = uHandleOffset2++;
HANDLE_PUSH_MAIN(extended, i);
for (const auto &handle : handles)
{
pollfd poll;
if (handle == -1)
{
continue;
}
poll.fd = handle;
poll.events = POLLIN;
poll.revents = 0;
HANDLE_PUSH_CHILD(poll);
HANDLE_PUSH_CHILD2(AuMakePair(i, false));
HANDLE_PUSH_CHILD3(poll)
}
for (const auto &handle : handlesWrite)
{
pollfd poll;
if (handle == -1)
{
continue;
}
poll.fd = handle;
poll.events = POLLOUT;
poll.revents = 0;
HANDLE_PUSH_CHILD(poll);
HANDLE_PUSH_CHILD2(AuMakePair(i, false));
HANDLE_PUSH_CHILD3(poll)
}
}
}
}
}
catch (...)
{
return {};
}
// must be 1:1 - do not mess with
for (AU_ITERATE_N(i, uHandleOffset))
{
pLoopSourceExs2[i]->OnPresleep();
}
int ret;
do
{
AuUInt32 uTimeout;
if (bZeroTick)
{
uTimeout = 0;
}
else if (qwTimeoutAbs)
{
auto uNow = AuTime::SteadyClockNS();
if (uNow >= qwTimeoutAbs)
{
uTimeout = 0;
}
else
{
uTimeout = AuNSToMS<AuUInt32>(qwTimeoutAbs - uNow);
}
}
else
{
uTimeout = -1;
}
ret = poll(pHandleArray, uFDOffset, uTimeout);
uTimeout = 0;
if (ret > 0)
{
for (AU_ITERATE_N(i, uFDOffset))
{
if (!pHandleArray[i].revents)
{
continue;
}
uTotal += 1;
pHandleArrayPersistent[i] = pHandleArray[i];
pHandleArray[i].fd = -1;
}
}
if (qwTimeoutAbs)
{
auto uNow = AuTime::SteadyClockNS();
if (uNow >= qwTimeoutAbs)
{
bTimeout = true;
break;
}
}
}
while ((ret == -1 &&
errno == EINTR) || (uTotal != uFDOffset));
bool bRet = uTotal == uFDOffset;
if (bRet)
{
AU_LOCK_GLOBAL_GUARD(gWaitForMultipleALLLock->AsWritable());
for (AU_ITERATE_N(i, uFDOffset))
{
auto uIndex = AuGet<0>(pHandleIndicies[i]);
auto bRead = AuGet<1>(pHandleIndicies[i]);
auto pLoopSource = AuExchange(pLoopSourceExs1[uIndex], nullptr);
if (!pLoopSource)
{
continue;
}
if (!pLoopSource->OnTrigger(pHandleArrayPersistent[i].fd))
{
for (AU_ITERATE_N(z, i))
{
ResetLoopSourceFalseAlarm(pLoopSourceExs1[z]);
}
bRet = false;
break;
}
}
}
// must be 1:1 - do not mess with
for (AU_ITERATE_N(i, uHandleOffset))
{
pLoopSourceExs2[i]->OnFinishSleep();
}
return bRet;
#endif
}
bool WaitMultipleAndObjects(const AuList<AuSPtr<ILoopSource>> &objects, bool bAlert, bool bBreakAPCs, bool bZeroTick, AuUInt64 qwTimeoutAbs, bool &bTooMany, bool &bTimeout)
{
bool bRet {};
do
{
bool bAPCDispatch {};
bRet = WaitMultipleAndObjects_(objects, bAlert, bBreakAPCs, bZeroTick, qwTimeoutAbs, bTooMany, bTimeout);
if (bAlert)
{
#if defined(AURORA_IS_LINUX_DERIVED)
// Do not syscall after read or select again under Linux
bAPCDispatch = UNIX::LinuxOverlappedYield(true);
#else
bAPCDispatch = IOYield();
#endif
}
if (bBreakAPCs && bAPCDispatch)
{
break;
}
}
while (!bRet && !bTimeout);
return bRet;
}
}

709
Source/IO/Loop/Loop.cpp
View File

@ -14,6 +14,38 @@
namespace Aurora::IO::Loop
{
// Write guards the transition period just after ::select-like and during { {onTriggered(...); } { reset spurious } }
// By read-locking, depending on the platform, you provide some insurance that WaitMultipleLoopSources didn't
// temporarily lock the resource. Under POSIX, this RWLock is fully respected. Under the platform agnostic path,
// it is only partially respected. Under Win32, it's not really respected, but these issues can be worked around,
// by using the interop-ready loop sources; under win32s fast path, interop-ready loop sources can be atomically
// AND locked via the native nt mechanism.
AuRWRenterableLock gWaitForMultipleALLLock;
bool WaitMultipleAndOldImpl(const AuList<AuSPtr<Loop::ILoopSource>> &lsList,
AuList<AuSPtr<Loop::ILoopSource>> &signaled,
AuUInt64 uFlags,
AuOptional<AuUInt32> optTimeoutMS,
bool bSpin,
bool bHasTimeOut,
bool bSleepForever,
bool bZeroTick,
AuUInt64 uTimeoutEnd,
bool bAvoidKrn,
AuUInt32 uBaseFlags);
bool WaitMultipleAndNewImpl(const AuList<AuSPtr<Loop::ILoopSource>> &lsList,
AuList<AuSPtr<Loop::ILoopSource>> &signaled,
AuUInt64 uFlags,
AuOptional<AuUInt32> optTimeoutMS,
bool bSpin,
bool bHasTimeOut,
bool bSleepForever,
bool bZeroTick,
AuUInt64 uTimeoutEnd,
bool bAlert,
bool bBreakAPCs);
#if !defined(AURORA_IS_MODERNNT_DERIVED)
AUKN_SYM AuSPtr<ILoopSource> NewLSWin32Source(bool)
{
@ -55,9 +87,10 @@ namespace Aurora::IO::Loop
}
#if defined(AURORA_IS_MODERNNT_DERIVED) || defined(AURORA_IS_POSIX_DERIVED)
AuList<AuSPtr<ILoopSource>> WaitMultipleOrObjects(const AuList<AuSPtr<ILoopSource>> &objects, bool bZeroTick, AuUInt32 timeout, bool bAllowOthers, bool &bTooMany);
AuList<AuSPtr<ILoopSource>> WaitMultipleOrObjects(const AuList<AuSPtr<ILoopSource>> &objects, bool bAlert, bool bBreakAPCs, bool bZeroTick, AuUInt32 timeout, bool bAllowOthers, bool &bTooMany);
bool WaitMultipleAndObjects(const AuList<AuSPtr<ILoopSource>> &objects, bool bAlert, bool bBreakAPCs, bool bZeroTick, AuUInt64 qwTimeoutAbs, bool &bTooMany, bool &bTimeout);
#endif
AuList<AuSPtr<ILoopSource>> WaitMultipleOrObjectsFallback(const AuList<AuSPtr<ILoopSource>> &objects, AuUInt32 timeout, bool bZeroTick, bool bAllowOthers, bool &bTimeout);
AuList<AuSPtr<ILoopSource>> WaitMultipleOrObjectsFallback(const AuList<AuSPtr<ILoopSource>> &objects, AuUInt32 timeout, bool bAlert, bool bBreak, bool bZeroTick, bool bAllowOthers, bool &bTimeout);
void ResetLoopSourceFalseAlarm(const AuSPtr<Loop::ILoopSource> &pLoopSource)
{
@ -118,19 +151,45 @@ namespace Aurora::IO::Loop
AuOptional<AuUInt32> optTimeoutMS)
{
signaled.clear();
AuList<AuUInt32> reverseList;
if (lsList.empty())
{
return true;
}
bool bAny { bool(uFlags & kWaitMultipleFlagAny) };
bool bSpin { !(uFlags & kWaitMultipleFlagNoSpin) };
bool bAvoidKrn { bool(uFlags & kWaitMultipleFlagAvoidKern) };
bool bZeroTick { optTimeoutMS && optTimeoutMS.value() == 0 };
bool bHasTimeOut { optTimeoutMS && optTimeoutMS.value() };
bool bZeroTick { optTimeoutMS && optTimeoutMS.Value() == 0 };
bool bHasTimeOut { optTimeoutMS && optTimeoutMS.Value() };
bool bBreakAPCs { bool(uFlags & kWaitMultipleFlagBreakAfterAPC) };
bool bNoAlert { bool(uFlags & kWaitMultipleFlagNoIOCallbacks) };
bool bAlert { !bNoAlert };
bool bSleepForever { !optTimeoutMS };
AuUInt32 uBaseFlags {};
if (lsList.empty())
{
if (bNoAlert)
{
return true;
}
else
{
if (bZeroTick)
{
return IOYield();
}
else if (bHasTimeOut)
{
return IOYieldFor(optTimeoutMS.Value());
}
else
{
return IOYieldFor(0);
}
}
}
if (bNoAlert)
{
uBaseFlags |= kFlagLSTryNoIOAlerts;
}
if (lsList.size() == 1)
{
@ -147,6 +206,8 @@ namespace Aurora::IO::Loop
uFlags = kFlagLSTryNoSpin;
}
uFlags |= uBaseFlags;
bool bStatus {};
if (bSleepForever)
{
@ -179,169 +240,55 @@ namespace Aurora::IO::Loop
if (!bAny)
{
AuUInt32 uStartingOffset { 1 };
auto &entryZero = lsList[0];
if (!entryZero)
// try to sleep somewhat atomically.
// under windows, this isn't quite possible with in process primitives. we just have to hope for interop-ready + kernel fast path.
// under linux, it's fine, because we can select or poll, and then { lock { ... } }
if (gRuntimeConfig.ioConfig.bUseSelectWaitAllStrat)
{
signaled.push_back({});
bool bTooMany {};
bool bTimeout {};
bool bRet =
WaitMultipleAndObjects(lsList, bAlert, bBreakAPCs, false, uTimeoutEnd, bTooMany, bTimeout);
if (!bTooMany)
{
if (bRet)
{
// explicit everyone became alert via select-like and said OK to our vtable query
signaled = lsList;
return true;
}
if (entryZero)
else if (bTimeout)
{
bool bStatus {};
auto eType = entryZero->GetType();
AuUInt8 uFlags {};
if (!bSpin)
{
uFlags = kFlagLSTryNoSpin;
}
if (eType == ELoopSource::eSourceMutex ||
eType == ELoopSource::eSourceFastMutex)
{
bStatus = false;
uStartingOffset = 0;
goto mainAllSleep;
}
else if (bSleepForever)
{
bStatus = entryZero->WaitOnExt(uFlags, 0);
}
else if (bHasTimeOut)
{
bStatus = entryZero->WaitOnExt(uFlags, optTimeoutMS.value());
// explicit timeout timestamp exceeded
return false;
}
else
{
bStatus = entryZero->IsSignaledExt(uFlags);
// Win32 message loop needs tending to or APC dispatch
return true;
}
}
}
if (!bStatus)
// try to sleep on one item in the list, and then atomically acquire the rest if that one succeeded.
// it can be observed that a single item in this lsList dips to non-signaled momentarily.
// the same may also happen with local (not native) win32 loop sources under WaitMultipleAndObjects.
// (interop ready/win32/win32-fast path is safe)
// (local or interop/linux/posix-fast path is safe)
// (local/win32 technically breaks WaitForMultipleObject docs)
// (any/win32/>64 handles generic path technically breaks WaitForMultipleObject docs)
// (any/*/generic path technically breaks WaitForMultipleObject docs)
// our implementation is good enough to work on real world code, but it technically can break some multi-mutex wait and multi-consumer event-happened queries code.
if (gRuntimeConfig.ioConfig.bUseOldIOWaitAllAlg)
{
goto next;
return WaitMultipleAndOldImpl(lsList, signaled, uFlags, optTimeoutMS, bSpin, bHasTimeOut, bSleepForever, bZeroTick, uTimeoutEnd, bAvoidKrn, uBaseFlags);
}
else
{
reverseList.push_back(0);
signaled.push_back(entryZero);
return WaitMultipleAndNewImpl(lsList, signaled, uFlags, optTimeoutMS, bSpin, bHasTimeOut, bSleepForever, bZeroTick, uTimeoutEnd, bAlert, bBreakAPCs);
}
}
mainAllSleep:
if (lsList.size() > uStartingOffset &&
(!bAvoidKrn || signaled.empty()))
{
for (AU_ITERATE_N(a, 2))
{
bool dBreak {};
for (AU_ITERATE_N_TO_X(i, uStartingOffset, lsList.size()))
{
AuUInt32 uTimeoutMS {};
bool bIsMutex {};
ELoopSource eType;
auto &pCurrent = lsList[i];
eType = pCurrent->GetType();
bIsMutex = eType == ELoopSource::eSourceMutex ||
eType == ELoopSource::eSourceFastMutex;
if (bIsMutex ^ bool(a))
{
continue;
}
if (uTimeoutEnd && !bZeroTick)
{
auto uStartTime = Time::SteadyClockNS();
if (uStartTime >= uTimeoutEnd)
{
bZeroTick = true;
}
uTimeoutMS = AuNSToMS<AuInt64>(uTimeoutEnd - uStartTime);
if (!uTimeoutMS)
{
bZeroTick = true;
}
}
if (bZeroTick)
{
if (bSpin)
{
if (!pCurrent->IsSignaled())
{
dBreak = true;
break;
}
bSpin = false;
}
else
{
if (!pCurrent->IsSignaledExt(kFlagLSTryNoSpin))
{
dBreak = true;
break;
}
}
}
else
{
if (bSpin)
{
if (!pCurrent->WaitOnAbs(uTimeoutEnd))
{
dBreak = true;
break;
}
// TBD
bSpin = false;
}
else
{
if (!pCurrent->WaitOnAbsExt(kFlagLSTryNoSpin, uTimeoutEnd))
{
dBreak = true;
break;
}
}
}
reverseList.push_back(i);
signaled.push_back(pCurrent);
}
if (dBreak || signaled.size() != lsList.size())
{
break;
}
}
}
next:
bool bReturnStatus { true };
if (signaled.size() != lsList.size())
{
bReturnStatus = false;
signaled.clear();
for (const auto &uIndex : reverseList)
{
ResetLoopSourceFalseAlarm(lsList[uIndex]);
}
}
return bReturnStatus;
}
else
{
bool bTimedout {};
@ -378,7 +325,7 @@ namespace Aurora::IO::Loop
{
bAnyFound = true;
if (pSource->IsSignaledExt(kFlagLSTryNoSpin))
if (pSource->IsSignaledExt(uBaseFlags | kFlagLSTryNoSpin))
{
signalTemp.push_back(pSource);
itr = lsList2.erase(itr);
@ -458,7 +405,7 @@ namespace Aurora::IO::Loop
if (AuBuild::kCurrentVendor == AuBuild::EVendor::eGenericMicrosoft &&
lsList2.size() < MAXIMUM_WAIT_OBJECTS)
{
signaled = WaitMultipleOrObjects(lsList2, bZeroTick, uTimeoutMS, bAllowOthers, bTooMany);
signaled = WaitMultipleOrObjects(lsList2, bAlert, bBreakAPCs, bZeroTick, uTimeoutMS, bAllowOthers, bTooMany);
bTimedout = uTimeoutEnd && uTimeoutMS && !bZeroTick ?
Time::SteadyClockNS() >= uTimeoutEnd :
false;
@ -467,7 +414,7 @@ namespace Aurora::IO::Loop
#elif defined(AURORA_IS_POSIX_DERIVED)
if (true)
{
signaled = WaitMultipleOrObjects(lsList2, bZeroTick, uTimeoutMS, bAllowOthers, bTooMany);
signaled = WaitMultipleOrObjects(lsList2, bAlert, bBreakAPCs,bZeroTick, uTimeoutMS, bAllowOthers, bTooMany);
bTimedout = uTimeoutEnd && uTimeoutMS && !bZeroTick ?
Time::SteadyClockNS() >= uTimeoutEnd :
false;
@ -478,11 +425,11 @@ namespace Aurora::IO::Loop
bTooMany = true;
}
}
while (!bTimedout && !bTooMany && !bZeroTick && signaled.empty());
while (!bTimedout && !bTooMany && !bZeroTick && signaled.empty() && (!bBreakAPCs || bNoAlert));
if (bTooMany)
{
signaled = WaitMultipleOrObjectsFallback(lsList2, uTimeoutMS, bZeroTick, bAllowOthers, bTimedout);
signaled = WaitMultipleOrObjectsFallback(lsList2, uTimeoutMS, bAlert, bBreakAPCs, bZeroTick, bAllowOthers, bTimedout);
bTimedout &= !bZeroTick;
}
}
@ -493,14 +440,404 @@ namespace Aurora::IO::Loop
{
return false;
}
else if (bBreakAPCs && bAlert)
{
return true;
}
else
{
return signaled.size();
}
}
return false;
}
AuList<AuSPtr<ILoopSource>> WaitMultipleOrObjectsFallback(const AuList<AuSPtr<ILoopSource>> &objects, AuUInt32 timeout, bool bZeroTick, bool bAllowOthers, bool &bTimeout)
bool WaitMultipleAndNewImpl(const AuList<AuSPtr<Loop::ILoopSource>> &lsList,
AuList<AuSPtr<Loop::ILoopSource>> &signaled,
AuUInt64 uBaseFlags,
AuOptional<AuUInt32> optTimeoutMS,
bool bSpin,
bool bHasTimeOut,
bool bSleepForever,
bool bZeroTick,
AuUInt64 uTimeoutEnd,
bool bAlert,
bool bBreakAPCs)
{
AuUInt32 uTick {};
while (true)
{
AuUInt32 uChecked {};
bool bStatus {};
// Sleep on one (warn: this one primitive will appear to be non-signaled until we can write-acquire gWaitForMultipleALLLock)
// (... : select, poll, etc based paths do not have this issue!)
{
bool bIsMutex {};
ELoopSource eType;
AuUInt8 uFlags {};
if (!bSpin)
{
uFlags |= kFlagLSTryNoSpin;
}
if (!bAlert)
{
uFlags |= kFlagLSTryNoIOAlerts;
}
auto a = (uTick / lsList.size()) % 2;
uChecked = uTick++ % lsList.size();
auto pCurrent = lsList[uChecked];
AuUInt32 uLocalCounter {};
while (!pCurrent)
{
uChecked = uTick++ % lsList.size();
pCurrent = lsList[pCurrent];
if ((uLocalCounter ++) > lsList.size())
{
return false;
}
}
eType = pCurrent->GetType();
bIsMutex = eType == ELoopSource::eSourceMutex ||
eType == ELoopSource::eSourceFastMutex;
if (bIsMutex ^ bool(a))
{
continue;
}
if (bHasTimeOut || bSleepForever)
{
bStatus = pCurrent->WaitOnAbsExt(uFlags, uTimeoutEnd);
if (!bStatus)
{
if (uTimeoutEnd && AuNSToMS<AuUInt64>(uTimeoutEnd) <= AuTime::SteadyClockMS())
{
return false;
}
else if (bAlert && bBreakAPCs)
{
return true;
}
}
}
else
{
bStatus = pCurrent->IsSignaledExt(uFlags | kFlagLSTryNoIOAlerts);
if (!bStatus && bAlert)
{
if (IOYield())
{
return true;
}
}
}
if (bIsMutex)
{
//uTick = 0;
}
}
if (bStatus)
{
// Warning: this does not protect the first acquisition !
AU_LOCK_GLOBAL_GUARD(gWaitForMultipleALLLock->AsWritable());
AuUInt32 uBreakA {};
AuUInt32 uBreakIndex {};
{
for (AU_ITERATE_N(a, 2))
{
bool dBreak {};
for (AU_ITERATE_N(i, lsList.size()))
{
bool bIsMutex {};
ELoopSource eType;
if (i == uChecked)
{
continue;
}
auto &pCurrent = lsList[i];
if (!pCurrent)
{
continue;
}
eType = pCurrent->GetType();
bIsMutex = eType == ELoopSource::eSourceMutex ||
eType == ELoopSource::eSourceFastMutex;
if (bIsMutex ^ bool(a))
{
continue;
}
if (!pCurrent->IsSignaledExt(kFlagLSTryNoSpin | kFlagLSTryNoIOAlerts))
{
dBreak = true;
uBreakIndex = i;
break;
}
}
if (dBreak)
{
uBreakA = a + 1;
bStatus = false;
break;
}
}
}
if (!bStatus)
{
for (AU_ITERATE_N(a, uBreakA))
{
for (AU_ITERATE_N(i, uBreakIndex))
{
auto &pCurrent = lsList[i];
bool bIsMutex {};
ELoopSource eType;
if (!pCurrent)
{
continue;
}
if (i == uChecked)
{
continue;
}
eType = pCurrent->GetType();
bIsMutex = eType == ELoopSource::eSourceMutex ||
eType == ELoopSource::eSourceFastMutex;
if (bIsMutex ^ bool(a))
{
continue;
}
ResetLoopSourceFalseAlarm(pCurrent);
}
}
if (auto pSource = lsList[uChecked])
{
ResetLoopSourceFalseAlarm(pSource);
}
if (uTimeoutEnd && uTimeoutEnd <= AuTime::SteadyClockNS())
{
return false;
}
}
else
{
signaled = lsList;
return true;
}
}
}
}
bool WaitMultipleAndOldImpl(const AuList<AuSPtr<Loop::ILoopSource>> &lsList,
AuList<AuSPtr<Loop::ILoopSource>> &signaled,
AuUInt64 uFlags,
AuOptional<AuUInt32> optTimeoutMS,
bool bSpin,
bool bHasTimeOut,
bool bSleepForever,
bool bZeroTick,
AuUInt64 uTimeoutEnd,
bool bAvoidKrn,
AuUInt32 uBaseFlags)
{
AuList<AuUInt32> reverseList;
{
AuUInt32 uStartingOffset { 1 };
auto &entryZero = lsList[0];
if (!entryZero)
{
signaled.push_back({});
}
if (entryZero)
{
bool bStatus {};
auto eType = entryZero->GetType();
AuUInt8 uFlags {};
if (!bSpin)
{
uFlags = kFlagLSTryNoSpin;
}
uFlags |= uBaseFlags;
if (eType == ELoopSource::eSourceMutex ||
eType == ELoopSource::eSourceFastMutex)
{
bStatus = false;
uStartingOffset = 0;
goto mainAllSleep;
}
else if (bSleepForever)
{
bStatus = entryZero->WaitOnExt(uFlags, 0);
}
else if (bHasTimeOut)
{
bStatus = entryZero->WaitOnExt(uFlags, optTimeoutMS.value());
}
else
{
bStatus = entryZero->IsSignaledExt(uFlags);
}
if (!bStatus)
{
goto next;
}
else
{
reverseList.push_back(0);
signaled.push_back(entryZero);
bZeroTick = true;
}
}
mainAllSleep:
if (lsList.size() > uStartingOffset &&
(!bAvoidKrn || signaled.empty()))
{
for (AU_ITERATE_N(a, 2))
{
bool dBreak {};
for (AU_ITERATE_N_TO_X(i, uStartingOffset, lsList.size()))
{
AuUInt32 uTimeoutMS {};
bool bIsMutex {};
ELoopSource eType;
auto &pCurrent = lsList[i];
eType = pCurrent->GetType();
bIsMutex = eType == ELoopSource::eSourceMutex ||
eType == ELoopSource::eSourceFastMutex;
if (bIsMutex ^ bool(a))
{
continue;
}
if (uTimeoutEnd && !bZeroTick)
{
auto uStartTime = Time::SteadyClockNS();
if (uStartTime >= uTimeoutEnd)
{
bZeroTick = true;
}
uTimeoutMS = AuNSToMS<AuInt64>(uTimeoutEnd - uStartTime);
if (!uTimeoutMS)
{
bZeroTick = true;
}
}
if (bZeroTick)
{
if (bSpin)
{
if (!pCurrent->IsSignaledExt(uBaseFlags))
{
dBreak = true;
break;
}
bSpin = false;
}
else
{
if (!pCurrent->IsSignaledExt(uBaseFlags | kFlagLSTryNoSpin))
{
dBreak = true;
break;
}
}
}
else
{
if (bSpin)
{
if (!pCurrent->WaitOnAbsExt(uBaseFlags, uTimeoutEnd))
{
dBreak = true;
break;
}
// TBD
bSpin = false;
}
else
{
if (!pCurrent->WaitOnAbsExt(kFlagLSTryNoSpin | uBaseFlags, uTimeoutEnd))
{
dBreak = true;
break;
}
}
}
reverseList.push_back(i);
signaled.push_back(pCurrent);
}
if (dBreak || signaled.size() != lsList.size())
{
break;
}
}
}
next:
bool bReturnStatus { true };
if (signaled.size() != lsList.size())
{
bReturnStatus = false;
signaled.clear();
for (const auto &uIndex : reverseList)
{
ResetLoopSourceFalseAlarm(lsList[uIndex]);
}
}
return bReturnStatus;
}
}
AuList<AuSPtr<ILoopSource>> WaitMultipleOrObjectsFallback(const AuList<AuSPtr<ILoopSource>> &objects, AuUInt32 timeout, bool bAlert, bool bBreak, bool bZeroTick, bool bAllowOthers, bool &bTimeout)
{
AuList<AuSPtr<ILoopSourceEx>> loopSourceExs;
AuList<AuSPtr<ILoopSource>> triggered;
@ -512,12 +849,54 @@ namespace Aurora::IO::Loop
bTimeout = false;
#if defined(AURORA_IS_MODERNNT_DERIVED)
// Optimization / can be ignored / based impl details of NTs LoopQueue
{
AuUInt32 uBaseFlags { 0 };
if (!bAlert)
{
uBaseFlags |= kFlagLSTryNoIOAlerts;
}
// Check past the first loopqueues chug batch
for (AU_ITERATE_N_TO_X(i, MAXIMUM_WAIT_OBJECTS - 1, objects.size()))
{
if (objects[i]->IsSignaledExt(uBaseFlags))
{
triggered.push_back(objects[i]);
}
}
// If we got something
if (triggered.size())
{
// Check everyone
for (AU_ITERATE_N_TO_X(i, 0, MAXIMUM_WAIT_OBJECTS - 1))
{
if (objects[i]->IsSignaledExt(uBaseFlags))
{
triggered.push_back(objects[i]);
}
}
// and then return
return triggered;
}
// otherwise, we just need to gross chug poll.
}
#endif
auto pQueue = AuLoop::NewLoopQueue();
if (!pQueue)
{
return {};
}
pQueue->ConfigureDoIOApcCallbacks(bAlert);
pQueue->ConfigureBreakAnyAfterAPC(bBreak);
try
{
loopSourceExs.reserve(objects.size());

97
Source/IO/Loop/LoopQueue.Linux.cpp
View File

@ -197,56 +197,52 @@ namespace Aurora::IO::Loop
bool LoopQueue::SourceAdd(const AuSPtr<ILoopSource> &source)
{
return SourceAddWithTimeout(source, 0);
return LoopQueue::SourceAddWithTimeout(source, 0);
}
bool LoopQueue::SourceAddWithTimeout(const AuSPtr<ILoopSource> &source, AuUInt32 ms)
{
AU_LOCK_GUARD(this->commitQueueMutex_);
return this->SourceAddWithTimeoutEx(source, ms);
return LoopQueue::SourceAddWithTimeoutEx(source, ms);
}
bool LoopQueue::SourceAddWithTimeoutEx(const AuSPtr<ILoopSource> &source, AuUInt32 ms)
{
this->lockStealer_.Set();
#if 0
AU_LOCK_GUARD(this->sourceMutex_->AsWritable());
#else
auto pWaitable = this->sourceMutex_->AsWritable();
auto pLocked = pWaitable->TryLock();
if (pLocked)
#endif
auto pSrc = AuMakeShared<SourceExtended>(this, source);
if (!pSrc)
{
this->lockStealer_.Reset();
auto src = AuMakeShared<SourceExtended>(this, source);
if (!src)
{
pWaitable->Unlock();
return false;
}
if (ms)
{
src->timeoutAbs = (AuUInt64)ms + AuTime::SteadyClockMS();
pSrc->timeoutAbs = (AuUInt64)ms + AuTime::SteadyClockMS();
}
else
{
pSrc->timeoutAbs = 0;
}
if (!AuTryInsert(this->sources_, AuConstReference(src)))
auto pWaitable = this->sourceMutex_->AsWritable();
auto pLocked = pWaitable->TryLock();
if (pLocked)
{
if (!AuTryInsert(this->pendingBlocking_, AuMakePair(pSrc, ms)))
{
pWaitable->Unlock();
return false;
}
this->globalEpoll_.Add(src.get());
else
{
pWaitable->Unlock();
return true;
}
#if 1
}
else
{
return AuTryInsert(this->pendingBlocking_, AuMakePair(source, ms));
AU_LOCK_GUARD(this->commitQueueMutex_);
this->lockStealer_.Set();
return AuTryInsert(this->pendingBlocking_, AuMakePair(pSrc, ms));
}
#endif
}
bool LoopQueue::SourceRemove(const AuSPtr<ILoopSource> &source)
@ -288,6 +284,16 @@ namespace Aurora::IO::Loop
// Intentionally NO-OP under Linux
}
void LoopQueue::ConfigureDoIOApcCallbacks(bool bDoIOAPCs)
{
this->bDoIOAPCs_ = bDoIOAPCs;
}
void LoopQueue::ConfigureBreakAnyAfterAPC(bool bAPCBreaks)
{
this->bAPCBreaks_ = bAPCBreaks;
}
bool LoopQueue::CommitDecommit()
{
AuUInt32 dwSuccess {};
@ -335,23 +341,28 @@ namespace Aurora::IO::Loop
bool LoopQueue::Commit()
{
AU_DEBUG_MEMCRUNCH;
AU_LOCK_GUARD(this->commitQueueMutex_);
AU_DEBUG_MEMCRUNCH;
for (const auto & [pSource, ms] : AuExchange(this->pendingBlocking_, {}))
{
this->SourceAddWithTimeoutEx(pSource, ms);
}
this->lockStealer_.Set();
auto pWritable = this->sourceMutex_->AsWritable();
if (pWritable->TryLock())
{
this->lockStealer_.Reset();
doUnderLock:
for (const auto & [pSource, ms] : AuExchange(this->pendingBlocking_, {}))
{
if (!AuTryInsert(this->sources_, AuConstReference(pSource)))
{
SysPanic("OOM under memcrunch. this shouldn't be possible?");
}
this->globalEpoll_.Add(pSource.get());
}
if (!CommitDecommit())
{
// TODO: this was commented out years ago
// fix bug?
//pWritable->Unlock();
//return false;
}
@ -401,6 +412,13 @@ namespace Aurora::IO::Loop
else
{
this->bRecommitLater = true;
this->lockStealer_.Set();
// try lock doesn't block future reads. this might.
if (pWritable->LockNS(250'000))
{
goto doUnderLock;
}
}
return true;
@ -719,6 +737,8 @@ namespace Aurora::IO::Loop
AuUInt64 now {};
epoll_event events[128];
this->Commit();
AU_LOCK_GUARD(this->sourceMutex_->AsReadable());
for (const auto & source : this->sources_)
@ -743,7 +763,16 @@ namespace Aurora::IO::Loop
deltaMS = nonblock ? 0 : -1;
}
int iEvents = IO::UNIX::LinuxOverlappedEpollShim(this->epollFd_, events, AuArraySize(events), deltaMS);
int iEvents;
if (this->bDoIOAPCs_)
{
iEvents = IO::UNIX::LinuxOverlappedEpollShim(this->epollFd_, events, AuArraySize(events), deltaMS);
}
else
{
iEvents = epoll_wait(this->epollFd_, events, AuArraySize(events), deltaMS);
}
if (iEvents == -1)
{
goto out;

7
Source/IO/Loop/LoopQueue.Linux.hpp
View File

@ -36,6 +36,9 @@ namespace Aurora::IO::Loop
void ChugPathConfigure(AuUInt32 sectionTickTime, AuSInt sectionDequeCount) override;
void ChugHint(bool value) override;
void ConfigureDoIOApcCallbacks(bool bDoIOAPCs) override;
void ConfigureBreakAnyAfterAPC(bool bAPCBreaks) override;
virtual bool Commit() override;
bool IsSignaledPeek() override;
@ -139,9 +142,11 @@ namespace Aurora::IO::Loop
AuList<AuFunction<void()>> epilogueHooks_;
AuList<AuPair<AuSPtr<ILoopSource>, AuUInt32>> pendingBlocking_;
AuList<AuPair<AuSPtr<SourceExtended>, AuUInt32>> pendingBlocking_;
bool bRecommitLater {};
AnEpoll globalEpoll_;
bool bDoIOAPCs_ { true };
bool bAPCBreaks_ { false };
};
}

74
Source/IO/Loop/LoopQueue.NT.cpp
View File

@ -312,7 +312,7 @@ namespace Aurora::IO::Loop
{
auto count = queueIterator.itr->source->Singular() ? 1 : queueIterator.itr->source->GetHandles().size();
if (this->handleArrayOr_.size() <= MAXIMUM_WAIT_OBJECTS)
if (this->handleArrayOr_.size() <= MAXIMUM_WAIT_OBJECTS_)
{
AuRemoveRange(this->handleArrayOr_, queueIterator.startingIndexOr, count);
AuRemoveRange(this->handleArrayAnd_, queueIterator.startingIndexAnd, count);
@ -361,7 +361,7 @@ namespace Aurora::IO::Loop
for (int i = 0; i < this->handleArrayAnd_.size(); i++)
{
if (((this->handleArrayOr_.size() % MAXIMUM_WAIT_OBJECTS)) == 0)
if (((this->handleArrayOr_.size() % MAXIMUM_WAIT_OBJECTS_)) == 0)
{
this->handleArrayOr_.push_back(this->hEvent_);
}
@ -464,7 +464,7 @@ namespace Aurora::IO::Loop
void LoopQueue::ChugPathConfigure(AuUInt32 sectionTickTime, AuSInt sectionDequeCount)
{
this->slowTickRate_ = AuMin(sectionDequeCount, AuSInt(MAXIMUM_WAIT_OBJECTS));
this->slowTickRate_ = AuMin(sectionDequeCount, AuSInt(MAXIMUM_WAIT_OBJECTS_));
this->slowTickMs_ = sectionTickTime ? sectionTickTime : 1;
}
@ -473,6 +473,16 @@ namespace Aurora::IO::Loop
this->forceChug_ = value;
}
void LoopQueue::ConfigureDoIOApcCallbacks(bool bDoIOAPCs)
{
this->bDoIOAPCs_ = bDoIOAPCs;
}
void LoopQueue::ConfigureBreakAnyAfterAPC(bool bAPCBreaks)
{
this->bAPCBreaks_ = bAPCBreaks;
}
bool LoopQueue::AddCallback(const AuSPtr<ILoopSource> &source, const AuSPtr<ILoopSourceSubscriber> &subscriber)
{
AU_LOCK_GUARD(this->rwMutex_->AsReadable());
@ -658,7 +668,7 @@ namespace Aurora::IO::Loop
while (count != index)
{
auto next = AuMin(count - index, AuUInt32(MAXIMUM_WAIT_OBJECTS));
auto next = AuMin(count - index, AuUInt32(MAXIMUM_WAIT_OBJECTS_));
startTime = AuTime::SteadyClockNS();
@ -718,11 +728,11 @@ namespace Aurora::IO::Loop
if (this->bIsWinLoop_ &&
pMsgWaitForMultipleObjectsEx)
{
status = pMsgWaitForMultipleObjectsEx(next, this->handleArrayAnd_.data() + index, timeDelta, QS_ALLPOSTMESSAGE | QS_ALLINPUT | QS_ALLEVENTS, MWMO_INPUTAVAILABLE | MWMO_ALERTABLE | MWMO_WAITALL);
status = pMsgWaitForMultipleObjectsEx(next, this->handleArrayAnd_.data() + index, timeDelta, QS_ALLPOSTMESSAGE | QS_ALLINPUT | QS_ALLEVENTS, MWMO_INPUTAVAILABLE | (this->bDoIOAPCs_ ? MWMO_ALERTABLE : 0) | MWMO_WAITALL);
}
else
{
status = ::WaitForMultipleObjectsEx(next, this->handleArrayAnd_.data() + index, true, timeDelta, true);
status = ::WaitForMultipleObjectsEx(next, this->handleArrayAnd_.data() + index, true, timeDelta, (this->bDoIOAPCs_ ? MWMO_ALERTABLE : 0));
}
}
while (status == WAIT_IO_COMPLETION);
@ -790,11 +800,11 @@ namespace Aurora::IO::Loop
{
if (!source.timeoutAbs)
{
bCompletedInTimeout = source.source->WaitOnAbsExt(0, endTime);
bCompletedInTimeout = source.source->WaitOnAbsExt(kFlagLSTryNoIOAlerts, endTime);
}
else
{
bCompletedInTimeout = source.source->WaitOnAbsExt(0, !endTime ? AuMSToNS<AuUInt64>(source.timeoutAbs) : AuMin(endTime, AuMSToNS<AuUInt64>(source.timeoutAbs)));
bCompletedInTimeout = source.source->WaitOnAbsExt(kFlagLSTryNoIOAlerts, !endTime ? AuMSToNS<AuUInt64>(source.timeoutAbs) : AuMin(endTime, AuMSToNS<AuUInt64>(source.timeoutAbs)));
}
if (bCompletedInTimeout)
@ -859,7 +869,7 @@ namespace Aurora::IO::Loop
{
AuUInt uHandles = (source.source->Singular() ? 1 : source.source->GetHandles().size());
int sOffset = -((int)uHandles);
if (this->handleArrayOr_.size() > MAXIMUM_WAIT_OBJECTS)
if (this->handleArrayOr_.size() > MAXIMUM_WAIT_OBJECTS_)
{
this->RemoveSourceNB(source.source);
this->willCommitInFuture_ = true;
@ -993,7 +1003,7 @@ namespace Aurora::IO::Loop
while (count != index)
{
auto next = AuMin(count - index, AuUInt32(MAXIMUM_WAIT_OBJECTS));
auto next = AuMin(count - index, AuUInt32(MAXIMUM_WAIT_OBJECTS_));
auto sleepMS = this->slowTickMs_;
if (internalEndTime && internalEndTime != AuUInt64(-1))
@ -1015,7 +1025,7 @@ namespace Aurora::IO::Loop
{
do
{
status = pMsgWaitForMultipleObjectsEx(next, this->handleArrayOr_.data() + index, sleepMS, QS_ALLINPUT | QS_ALLPOSTMESSAGE, MWMO_ALERTABLE);
status = pMsgWaitForMultipleObjectsEx(next, this->handleArrayOr_.data() + index, sleepMS, QS_ALLINPUT | QS_ALLPOSTMESSAGE, this->bDoIOAPCs_ ? MWMO_ALERTABLE : 0);
{
auto temp2 = status;
@ -1025,7 +1035,7 @@ namespace Aurora::IO::Loop
}
}
}
while (status == WAIT_IO_COMPLETION);
while (!this->bAPCBreaks_ && status == WAIT_IO_COMPLETION);
if (status == next)
{
@ -1044,14 +1054,14 @@ namespace Aurora::IO::Loop
{
uTimeout /= 2ull;
}
status = pNtWaitForMultipleObjects(next, this->handleArrayOr_.data() + index, (_OBJECT_WAIT_TYPE)1, TRUE, &uTimeout);
status = pNtWaitForMultipleObjects(next, this->handleArrayOr_.data() + index, (_OBJECT_WAIT_TYPE)1, this->bDoIOAPCs_ ? TRUE : FALSE, &uTimeout);
}
else
{
status = ::WaitForMultipleObjectsEx(next, this->handleArrayOr_.data() + index, false, sleepMS, true);
status = ::WaitForMultipleObjectsEx(next, this->handleArrayOr_.data() + index, false, sleepMS, this->bDoIOAPCs_);
}
}
while (status == WAIT_IO_COMPLETION);
while (!this->bAPCBreaks_ && status == WAIT_IO_COMPLETION);
}
if (WaitStatusFromAligned(status, active))
@ -1071,6 +1081,12 @@ namespace Aurora::IO::Loop
return false;
}
if (status == WAIT_IO_COMPLETION && this->bAPCBreaks_)
{
indexOfTriggered = -1;
return true;
}
index += count;
}
@ -1108,6 +1124,7 @@ namespace Aurora::IO::Loop
DWORD temp;
AuUInt32 indexOfTriggered {};
AuUInt32 triggeredCount {};
bool bAPCHit {};
if (trigger)
{
@ -1118,9 +1135,13 @@ namespace Aurora::IO::Loop
{
status = this->handleArrayOr_.size();
}
else if ((this->handleArrayOr_.size() > MAXIMUM_WAIT_OBJECTS) || (this->forceChug_))
else if ((this->handleArrayOr_.size() > MAXIMUM_WAIT_OBJECTS_) || (this->forceChug_))
{
status = ChugWaitAny(internalEndTime, chuggerIndex, indexOfTriggered);
if (indexOfTriggered == -1 && this->bAPCBreaks_)
{
return true;
}
}
else
{
@ -1167,28 +1188,29 @@ namespace Aurora::IO::Loop
if (!DoTryIf())
#endif
{
temp = pMsgWaitForMultipleObjectsEx(this->handleArrayOr_.size(), this->handleArrayOr_.data(), sleepDelta, QS_ALLPOSTMESSAGE | QS_ALLINPUT | QS_ALLEVENTS, MWMO_INPUTAVAILABLE | MWMO_ALERTABLE);
temp = pMsgWaitForMultipleObjectsEx(this->handleArrayOr_.size(), this->handleArrayOr_.data(), sleepDelta, QS_ALLPOSTMESSAGE | QS_ALLINPUT | QS_ALLEVENTS, MWMO_INPUTAVAILABLE | (this->bDoIOAPCs_ ? MWMO_ALERTABLE : 0));
}
}
else
{
temp = pMsgWaitForMultipleObjectsEx(this->handleArrayOr_.size(), this->handleArrayOr_.data(), sleepDelta, QS_ALLPOSTMESSAGE | QS_ALLINPUT | QS_ALLEVENTS, MWMO_INPUTAVAILABLE | MWMO_ALERTABLE);
temp = pMsgWaitForMultipleObjectsEx(this->handleArrayOr_.size(), this->handleArrayOr_.data(), sleepDelta, QS_ALLPOSTMESSAGE | QS_ALLINPUT | QS_ALLEVENTS, MWMO_INPUTAVAILABLE | (this->bDoIOAPCs_ ? MWMO_ALERTABLE : 0));
}
}
else
{
temp = ::WaitForMultipleObjectsEx(this->handleArrayOr_.size(), this->handleArrayOr_.data(), false, sleepDelta, true);
temp = ::WaitForMultipleObjectsEx(this->handleArrayOr_.size(), this->handleArrayOr_.data(), false, sleepDelta, this->bDoIOAPCs_);
}
{
auto temp2 = temp;
while (temp2 == WAIT_IO_COMPLETION)
{
bAPCHit = true;
temp2 = SleepEx(0, true);
}
}
}
while (temp == WAIT_IO_COMPLETION);
while (!this->bAPCBreaks_ && temp == WAIT_IO_COMPLETION);
status = WaitToRetStatus(temp);
@ -1224,8 +1246,16 @@ namespace Aurora::IO::Loop
ConsiderEvicitingTimeoutsAll();
bFinished = true;
StartUserAndTakeOwn_Release();
if (this->bAPCBreaks_)
{
return bAPCHit;
}
else
{
return false;
}
}
else
{
static Aurora::Utility::RateLimiter limit(AuMSToNS<AuUInt64>(24));
@ -1383,7 +1413,7 @@ namespace Aurora::IO::Loop
{
AuUInt uHandles = (source.source->Singular() ? 1 : source.source->GetHandles().size());
int sOffset = -(uHandles);
if (this->handleArrayOr_.size() > MAXIMUM_WAIT_OBJECTS)
if (this->handleArrayOr_.size() > MAXIMUM_WAIT_OBJECTS_)
{
this->RemoveSourceNB(source.source);
this->willCommitInFuture_ = true;
@ -1575,7 +1605,7 @@ namespace Aurora::IO::Loop
{
auto count = source.source->Singular() ? 1 : source.source->GetHandles().size();
if (this->handleArrayOr_.size() <= MAXIMUM_WAIT_OBJECTS)
if (this->handleArrayOr_.size() <= MAXIMUM_WAIT_OBJECTS_)
{
AuRemoveRange(this->handleArrayOr_, queueIterator.startingIndexOr, count);
AuRemoveRange(this->handleArrayAnd_, queueIterator.startingIndexAnd, count);

8
Source/IO/Loop/LoopQueue.NT.hpp
View File

@ -10,6 +10,8 @@
#include "ILoopSourceEx.hpp"
#include "LoopQueue.hpp"
#define MAXIMUM_WAIT_OBJECTS_ (MAXIMUM_WAIT_OBJECTS - 1)
namespace Aurora::IO::Loop
{
struct LoopQueue : ILoopQueue//, ILoopEpilogueHook
@ -35,6 +37,9 @@ namespace Aurora::IO::Loop
void ChugPathConfigure(AuUInt32 sectionTickTime, AuSInt sectionDequeCount) override;
void ChugHint(bool value) override;
void ConfigureDoIOApcCallbacks(bool bDoIOAPCs) override;
void ConfigureBreakAnyAfterAPC(bool bAPCBreaks) override;
virtual bool Commit() override;
bool IsSignaledPeek() override;
@ -91,6 +96,9 @@ namespace Aurora::IO::Loop
bool TryPumpWin32();
bool bDoIOAPCs_ { true };
bool bAPCBreaks_ { false };
//
SourceCallbacks msgCallbacks_;
AuList<AuSPtr<ILoopSource>> removedSources_;

29
Source/IO/Loop/WaitSingleBase.cpp
View File

@ -10,6 +10,8 @@
namespace Aurora::IO::Loop
{
extern AuRWRenterableLock gWaitForMultipleALLLock;
bool WaitSingleBase::IsSignaledNoSpinIfUserland()
{
return this->IsSignaled();
@ -22,12 +24,7 @@ namespace Aurora::IO::Loop
bool WaitSingleBase::IsSignaledExt(AuUInt8 uFlags)
{
bool val {};
AuUInt one {AuNumericLimits<AuUInt>::max()};
AuUInt two {AuNumericLimits<AuUInt>::max()};
bool bFlagUser = uFlags & kFlagLSTryNoSpin;
bool bFlagNoAlert = uFlags & kFlagLSTryNoIOAlerts;
if (bFlagUser)
{
@ -36,6 +33,27 @@ namespace Aurora::IO::Loop
this->OnPresleep();
bool bRet = this->IsSignaledExt_(uFlags);
if (!bRet)
{
AU_LOCK_GUARD(gWaitForMultipleALLLock->AsReadable());
bRet = this->IsSignaledExt_(uFlags);
}
this->OnFinishSleep();
return bRet;
}
bool WaitSingleBase::IsSignaledExt_(AuUInt8 uFlags)
{
bool val {};
AuUInt one {AuNumericLimits<AuUInt>::max()};
AuUInt two {AuNumericLimits<AuUInt>::max()};
bool bFlagUser = uFlags & kFlagLSTryNoSpin;
bool bFlagNoAlert = uFlags & kFlagLSTryNoIOAlerts;
if (this->Singular())
{
#if defined(AURORA_IS_POSIX_DERIVED)
@ -69,7 +87,6 @@ namespace Aurora::IO::Loop
}
}
this->OnFinishSleep();
return val;
}

2
Source/IO/Loop/WaitSingleBase.hpp
View File

@ -38,5 +38,7 @@ namespace Aurora::IO::Loop
virtual bool WaitOnAbsExt(AuUInt8 uFlags, AuUInt64 uTimeoutAbs) override;
virtual bool WaitForAtleastOne(AuUInt32 timeout, const AuList<AuUInt> &handles _OPT_WRITE_ARRAY, AuUInt &one _OPT_WRITE_REF, bool bNoAlert) = 0;
virtual bool WaitForOne(AuUInt32 timeout, AuUInt handle _OPT_WRITE, bool bNoAlert) = 0;
private:
bool IsSignaledExt_(AuUInt8 uFlags);
};
}

2
Source/IO/UNIX/IOSubmit.Linux.cpp
View File

@ -74,6 +74,8 @@ namespace Aurora::IO::UNIX
{
SysPushErrorCatch("IO Callback threw an exception");
}
co_return;
}
#endif