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

[+] Initial attempt at an epoll backend

Browse Source 
[+] Added comments in nt opener
[*] Fixed rng close
[*] Fixed possible aarch64 crash where unix thread ep function didnt return a value
This commit is contained in:
Reece Wilson 2022-04-09 16:53:14 +01:00
parent c311df8858
commit cf219eabaa
9 changed files with 883 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
Include/Aurora/Loop/ILoopQueue.hpp
View File

@ -24,14 +24,18 @@ namespace Aurora::Loop
*
* @param source
* @return
* @note thread safe / nonblocking | can be called alongside any other function marked as such
*/
virtual bool SourceAdd(const AuSPtr<ILoopSource> &source) = 0;
/**
* @brief Same behaviour as SourceAdd
* Note: Timeout is specified in MS and polled during WaitAny/All failure or completition
* Objects awaiting timeout will not preempt the loop
* @param source
* @param timeoutMS
* @return
* @note thread safe / nonblocking | can be called alongside any other function marked as such
*/
virtual bool SourceAddWithTimeout(const AuSPtr<ILoopSource> &source, AuUInt32 timeoutMS) = 0;
@ -40,18 +44,21 @@ namespace Aurora::Loop
* Sources are defacto poped unless subscriber returns false indicating repeated lock attempts are wanted.
* Should no subscriber be registered, the loop source will not be automatically removed
* @param source
* @note thread safe / nonblocking | can be called alongside any other function marked as such
*/
virtual bool SourceRemove(const AuSPtr<ILoopSource> &source) = 0;
/**
* @brief Updates the OS watchdog list cache concept after Source[Remove/Add[WithTimeout]]
* @return
* @note thread safe
*/
virtual bool Commit() = 0;
/**
* @brief
* @return the amount of loop sources added to the queue
* @note thread safe | can be called alongside any other function marked as such
*/
virtual AuUInt32 GetSourceCount() = 0;
@ -71,6 +78,7 @@ namespace Aurora::Loop
* @param source
* @param subscriber
* @return
* @note thread safe | can be called alongside any other function marked as such
*/
virtual bool AddCallback(const AuSPtr<ILoopSource> &source, const AuSPtr<ILoopSourceSubscriber> &subscriber) = 0;
@ -80,6 +88,7 @@ namespace Aurora::Loop
* @param source
* @param subscriber
* @return
* @note thread safe | can be called alongside any other function marked as such
*/
virtual bool AddCallbackEx(const AuSPtr<ILoopSource> &source, const AuSPtr<ILoopSourceSubscriberEx> &subscriber) = 0;
@ -88,6 +97,7 @@ namespace Aurora::Loop
* Registers a callback to handle all loop source signaled events.
* @param subscriber
* @return
* @note thread safe | can be called alongside any other function marked as such
*/
virtual bool AddCallback(const AuSPtr<ILoopSourceSubscriber> &subscriber) = 0;
@ -96,6 +106,7 @@ namespace Aurora::Loop
* @brief Nonblocking wait-any for all objects in the loop queue
* @warning (may yield to ILoopSourceSubscriber delegate on the current context)
* @return
* @note thread safe / nonblocking | can be called alongside any other function marked as such
*/
virtual bool IsSignaled() = 0;
@ -104,14 +115,17 @@ namespace Aurora::Loop
* Note: the completion of another Wait[All/Any[Ex]] call may result in a
* @return
* @warning (may yield to ILoopSourceSubscriber delegate on the current context)
* @warning (thread safety is limited blocking callers of the object)
*/
virtual bool WaitAll (AuUInt32 timeout = 0) = 0;
/**
* @brief Waits on all the loop sources until at least one is signaled
* @brief Waits on all the loop sources until at least one is signaled.
* Additional work may be scheduled on other threads.
* @param timeout
* @return
* @warning (may yield to ILoopSourceSubscriber delegate on the current context)
* @note thread safe | can be called alongside any other function marked as such
*/
virtual AuUInt32 WaitAny (AuUInt32 timeout = 0) = 0;
@ -120,6 +134,7 @@ namespace Aurora::Loop
* @param timeout
* @return
* @warning (may yield to ILoopSourceSubscriber delegate on the current context)
* @note thread safe | can be called alongside any other function marked as such
*/
virtual AuList<AuSPtr<ILoopSource>> WaitAnyEx(AuUInt32 timeout = 0) = 0;
@ -172,6 +187,8 @@ namespace Aurora::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;
// TODO: once Win32 is finished and linux is looking alright, readd the removed functions
};
AUKN_SYM AuSPtr<ILoopQueue> NewLoopQueue();

775
Source/Loop/LoopQueue.Linux.cpp
View File

@ -10,6 +10,7 @@
#include "ILoopSourceEx.hpp"
#include "LoopQueue.Linux.hpp"
#include <sys/epoll.h>
#include <Source/Time/Time.hpp>
namespace Aurora::Loop
{
@ -30,7 +31,7 @@ namespace Aurora::Loop
// ...it wouldn't make sense create another loop queue per thread concept
// outside of the async subsystem (not counting TLS overlapped io)
LoopQueue::LoopQueue()
LoopQueue::LoopQueue() : lockStealer_(false, false, true)
{
}
@ -42,31 +43,184 @@ namespace Aurora::Loop
bool LoopQueue::Init()
{
this->epollFd_ = epoll_create1(-1);
if (this->epollFd_ == -1) return false;
this->epollFd_ = epoll_create1(0);
if (this->epollFd_ == -1)
{
return false;
}
return true;
this->sourceMutex_ = AuThreadPrimitives::RWLockUnique();
if (!this->sourceMutex_)
{
return false;
}
this->polledItemsMutex_ = AuThreadPrimitives::RWLockUnique();
if (!this->polledItemsMutex_)
{
return false;
}
this->globalEpoll_.parent = this;
return AuTryInsert(this->alternativeEpolls_, &this->globalEpoll_);
}
void LoopQueue::Deinit()
{
auto handle = AuExchange(this->epollFd_, -1);
if (handle != -1) close(handle);
int fd;
if ((fd = AuExchange(this->epollFd_, -1)) != -1)
{
::close(fd);
}
}
void LoopQueue::AnEpoll::Add(SourceExtended *source)
{
epoll_event event;
auto ex = source->sourceExtended;
if (!ex)
{
return;
}
event.data.ptr = source;
if (ex->Singular())
{
bool bDouble {};
int oldReadRef {};
int oldWriteRef {};
auto read = ex->GetHandle();
if (read != -1)
{
oldReadRef = startingWorkRead[read]++;
bDouble |= startingWorkWrite.find(read) != startingWorkWrite.end();
}
auto write = ex->GetWriteHandle();
if (write != -1)
{
oldWriteRef = startingWorkWrite[write]++;
bDouble |= startingWorkRead.find(write) != startingWorkRead.end();
}
if (bDouble)
{
epoll_event event;
event.events = EPOLLOUT | EPOLLIN;
event.data.ptr = source;
if ((oldReadRef == 0) && (oldWriteRef == 0))
{
epoll_ctl(this->parent->epollFd_, EPOLL_CTL_ADD, write, &event);
}
else
{
epoll_ctl(this->parent->epollFd_, EPOLL_CTL_MOD, write, &event);
}
}
if ((write != -1) && (!oldWriteRef))
{
event.events = EPOLLOUT;
epoll_ctl(this->parent->epollFd_, EPOLL_CTL_ADD, write, &event);
}
if ((read != -1) && (!oldReadRef))
{
event.events = EPOLLIN;
epoll_ctl(this->parent->epollFd_, EPOLL_CTL_ADD, read, &event);
}
}
else
{
auto read = ex->GetHandles();
auto write = ex->GetWriteHandles();
for (auto readHandle : read)
{
auto count = startingWorkRead[readHandle]++;
if (count)
{
continue;
}
if (AuExists(write, readHandle))
{
continue;
}
event.events = EPOLLIN;
epoll_ctl(this->parent->epollFd_, EPOLL_CTL_ADD, readHandle, &event);
}
for (auto writeHandle : write)
{
auto count = startingWorkWrite[writeHandle]++;
if (count)
{
if (AuExists(read, writeHandle))
{
event.events = EPOLLOUT | EPOLLIN;
epoll_ctl(this->parent->epollFd_, EPOLL_CTL_MOD, writeHandle, &event);
}
continue;
}
if (AuExists(read, writeHandle))
{
event.events = EPOLLOUT | EPOLLIN;
epoll_ctl(this->parent->epollFd_, EPOLL_CTL_ADD, writeHandle, &event);
}
else
{
event.events = EPOLLOUT;
epoll_ctl(this->parent->epollFd_, EPOLL_CTL_ADD, writeHandle, &event);
}
}
}
}
bool LoopQueue::SourceAdd(const AuSPtr<ILoopSource> &source)
{
return {};
return SourceAddWithTimeout(source, 0);
}
bool LoopQueue::SourceAddWithTimeout(const AuSPtr<ILoopSource> &source, AuUInt32 ms)
{
return {};
this->lockStealer_.Set();
AU_LOCK_GUARD(this->sourceMutex_->AsWritable());
this->lockStealer_.Reset();
auto src = AuMakeShared<SourceExtended>(this, source);
if (!src)
{
return false;
}
if (ms)
{
src->timeoutAbs = (AuUInt64)ms + AuTime::CurrentClockMS();
}
if (!AuTryInsert(this->sources_, src))
{
return false;
}
this->globalEpoll_.Add(src.get());
return true;
}
bool LoopQueue::SourceRemove(const AuSPtr<ILoopSource> &source)
{
return {};
AU_LOCK_GUARD(this->commitQueueMutex_);
return AuTryInsert(this->decommitQueue_, source);
}
AuUInt32 LoopQueue::GetSourceCount()
@ -77,18 +231,13 @@ namespace Aurora::Loop
bool LoopQueue::AddCallback(const AuSPtr<ILoopSource> &source, const AuSPtr<ILoopSourceSubscriber> &subscriber)
{
AU_LOCK_GUARD(this->commitQueueMutex_);
bool bAdded {};
return bAdded;
return AuTryInsert(this->commitPending_, AuMakeTuple(source, subscriber, AuSPtr<ILoopSourceSubscriberEx>{}));
}
bool LoopQueue::AddCallbackEx(const AuSPtr<ILoopSource> &source, const AuSPtr<ILoopSourceSubscriberEx> &subscriber)
{
AU_LOCK_GUARD(this->commitQueueMutex_);
bool bAdded {};
return bAdded;
return AuTryInsert(this->commitPending_, AuMakeTuple(source, AuSPtr<ILoopSourceSubscriber>{}, subscriber));
}
bool LoopQueue::AddCallback(const AuSPtr<ILoopSourceSubscriber> &subscriber)
@ -107,38 +256,614 @@ namespace Aurora::Loop
// Intentionally NO-OP under Linux
}
bool LoopQueue::CommitDecommit()
{
AuUInt32 dwSuccess {};
if (this->decommitQueue_.empty())
{
return true;
}
auto decommitQueue = AuExchange(this->decommitQueue_, {});
for (auto sourceExtended : sources_)
{
bool bFound {};
for (auto decommit : decommitQueue)
{
if (decommit == sourceExtended->source)
{
bFound = true;
break;
}
}
if (!bFound)
{
continue;
}
AU_LOCK_GUARD(this->polledItemsMutex_->AsReadable());
for (auto epoll : this->alternativeEpolls_)
{
epoll->Remove(sourceExtended.get(), true, true);
}
dwSuccess++;
}
SysAssertDbg(dwSuccess == decommitQueue.size(), "caught SourceRemove on invalid");
return dwSuccess;
}
bool LoopQueue::Commit()
{
AU_LOCK_GUARD(this->commitQueueMutex_);
this->lockStealer_.Set();
AU_LOCK_GUARD(this->sourceMutex_->AsWritable());
this->lockStealer_.Reset();
if (!CommitDecommit())
{
return false;
}
auto pending = AuExchange(this->commitPending_, {});
for (auto &source : this->sources_)
{
for (auto itr = pending.begin(); itr != pending.end(); )
{
if (source->source != AuGet<0>(*itr))
{
itr ++;
continue;
}
auto a = AuGet<1>(*itr);
if (a)
{
if (!AuTryInsert(source->subscribers, a))
{
this->commitPending_ = AuMove(this->commitPending_);
return false;
}
}
auto b = AuGet<2>(*itr);
if (b)
{
if (!AuTryInsert(source->subscriberExs, b))
{
// 1 and 2 are mutually exclusive, dont worry about clean up
this->commitPending_ = AuMove(this->commitPending_);
return false;
}
}
itr = pending.erase(itr);
}
source->Commit(source);
}
return true;
}
bool LoopQueue::IsSignaled()
{
return {};
fd_set readSet;
struct timeval tv {};
FD_ZERO(&readSet);
FD_SET(this->epollFd_, &readSet);
auto active = select(this->epollFd_ + 1, &readSet, NULL, NULL, &tv);
if (active == -1)
{
// todo push error
return false;
}
return active == 1;
}
bool LoopQueue::WaitAll(AuUInt32 timeout)
bool LoopQueue::WaitAll(AuUInt32 timeoutIn)
{
return {};
AnEpoll epollReference;
{
AU_LOCK_GUARD(this->globalEpoll_.lock);
epollReference = this->globalEpoll_;
}
epollReference.lock = {};
AuUInt64 timeout {timeoutIn};
if (timeout)
{
timeout += AuTime::CurrentClockMS();
}
{
AU_LOCK_GUARD(this->polledItemsMutex_->AsWritable());
AuTryInsert(this->alternativeEpolls_, &epollReference);
}
bool anythingLeft {};
bool bTimeout {};
do
{
anythingLeft = epollReference.startingWorkRead.size() || epollReference.startingWorkWrite.size();
if (!anythingLeft) return true;
//WaitAny(0);
// [==========] 1 test from 1 test suite ran. (11100 ms total)
// ...and a turbojet
//bool bTryAgain {};
//DoTick(timeout, {}, &bTryAgain);
// ...and + ~10ms latency
//bool bTryAgain {};
//DoTick(AuMin(AuUInt64(AuTime::CurrentClockMS() + 4), timeout), {}, &bTryAgain);
// [----------] 1 test from Loop (11101 ms total)
// ...and no jet engine (+ lower latency than windows)
bool bTryAgain {};
DoTick(timeout, {}, &bTryAgain);
// but this hack should apply to wait any as well, so i'm moving it to the DoTick function
anythingLeft = epollReference.startingWorkRead.size() || epollReference.startingWorkWrite.size();
bTimeout = AuTime::CurrentClockMS() >= timeout;
} while (anythingLeft && !bTimeout);
{
AU_LOCK_GUARD(this->polledItemsMutex_->AsWritable());
SysAssert(AuTryRemove(this->alternativeEpolls_, &epollReference));
}
return !anythingLeft;
}
AuUInt32 LoopQueue::WaitAny(AuUInt32 timeout)
AuUInt32 LoopQueue::WaitAny(AuUInt32 timeoutIn)
{
return {};
AuUInt64 timeout = timeoutIn;
if (timeout)
{
timeout += AuTime::CurrentClockMS();
}
AuUInt32 cTicked {};
bool bTryAgain {};
do
{
bTryAgain = false;
AuUInt32 ticked = DoTick(timeout, {}, &bTryAgain);
cTicked += ticked;
} while (bTryAgain);
return cTicked;
}
AuList<AuSPtr<ILoopSource>> LoopQueue::WaitAnyEx(AuUInt32 timeout)
AuList<AuSPtr<ILoopSource>> LoopQueue::WaitAnyEx(AuUInt32 timeoutIn)
{
return {};
AuList<AuSPtr<ILoopSource>> ret;
AuUInt64 timeout = timeoutIn;
if (timeout)
{
timeout += AuTime::CurrentClockMS();
}
bool bTryAgain {};
AuUInt32 cTicked {};
do
{
bTryAgain = false;
AuUInt32 ticked = DoTick(timeout, &ret, &bTryAgain);
cTicked += ticked;
} while (bTryAgain);
return ret;
}
void LoopQueue::DoTick()
void LoopQueue::AnEpoll::Remove(SourceExtended *source, bool readData, bool writeData)
{
if (!source->sourceExtended)
{
return;
}
auto ex = source->sourceExtended;
AU_LOCK_GUARD(this->lock);
bool bIsRoot = this == &this->parent->globalEpoll_;
if (readData)
{
for (auto i = startingWorkRead.begin(); i != startingWorkRead.end(); )
{
bool doesntMatch {};
auto &fd = i->first;
auto &usage = i->second;
if (ex->Singular())
{
doesntMatch = fd != ex->GetHandle();
}
else
{
doesntMatch = !AuExists(ex->GetHandles(), fd);
}
if (doesntMatch)
{
i++;
continue;
}
if ((--(usage)) != 0)
{
i++;
continue;
}
if (bIsRoot)
{
if (startingWorkWrite.find(fd) == startingWorkWrite.end())
{
epoll_ctl(this->parent->epollFd_, EPOLL_CTL_DEL, fd, nullptr);
}
else
{
epoll_event event;
event.events = EPOLLOUT;
event.data.ptr = source;
epoll_ctl(this->parent->epollFd_, EPOLL_CTL_MOD, fd, &event);
}
}
i = startingWorkRead.erase(i);
}
}
if (writeData)
{
for (auto i = startingWorkWrite.begin(); i != startingWorkWrite.end(); )
{
bool doesntMatch {};
auto &fd = i->first;
auto &usage = i->second;
if (ex->Singular())
{
doesntMatch = fd != ex->GetWriteHandle();
}
else
{
doesntMatch = !AuExists(ex->GetWriteHandles(), fd);
}
if (doesntMatch )
{
i++;
continue;
}
if ((--(usage)) != 0)
{
i++;
continue;
}
if (bIsRoot)
{
if (startingWorkRead.find(fd) == startingWorkRead.end())
{
epoll_ctl(this->parent->epollFd_, EPOLL_CTL_DEL, fd, nullptr);
}
else
{
epoll_event event;
event.events = EPOLLIN;
event.data.ptr = source;
epoll_ctl(this->parent->epollFd_, EPOLL_CTL_MOD, fd, &event);
}
}
i = startingWorkWrite.erase(i);
}
}
}
AuUInt32 LoopQueue::DoTick(AuUInt64 time, AuList<AuSPtr<ILoopSource>> *optOut, bool *tryAgain)
{
AuUInt32 bTicked {};
AuUInt64 now {};
epoll_event events[128];
AU_LOCK_GUARD(this->sourceMutex_->AsReadable());
for (const auto & source : this->sources_)
{
if (source->sourceExtended)
{
source->sourceExtended->OnPresleep();
}
}
// epoll_pwait2 is fucking broken and the dipshits who wrote the test used relative values
//
// Nothing I tried worked.
//
// Am I stupid? Probably, but...
// (1) no one as far as i can tell has ever written anything using this api, per a github search
// (2) i found one reference that the that are the linux kernel developers used MONO time for this
// one timespec API unlike everything else, using an abs value rel to that clock didn't change
// anything.
// (3) i found a test that would indicate its relative despite the fact UNIX/Linux sync APIs
// tend to use abs time
//
// What does my experience working on xenus tell me?
// Because the GOOOOGLERs in the form of linux kernel developers were faced with an issue that
// couldn't be solved by involve copy/pasting memory map code, making a mess of public headers,
// or taking credit for third party driver code as their own kernel code, indeed are to blame
// for making my life miserable once again.
auto deltaMS = time ? AuMin(AuInt64(4), (AuInt64)time - (AuInt64)AuTime::CurrentClockMS()) : 0;
if (deltaMS < 0) deltaMS = 0;
int iEvents = epoll_wait(this->epollFd_, events, AuArraySize(events), deltaMS);
if (iEvents == -1)
{
goto out;
}
for (int i = 0; i < iEvents; i++)
{
bool readData = events[i].events & EPOLLIN;
bool writeData = events[i].events & EPOLLOUT;
auto handle = events[i].data.ptr;
if (!handle)
{
continue;
}
auto source = AuReinterpretCast<SourceExtended *>(handle)->pin.lock();
auto [ticked, remove] = source->DoWork(readData, writeData);
bTicked += ticked;
if (ticked)
{
if (optOut)
{
optOut->push_back(source->source);
}
}
if (remove)
{
this->sourceMutex_->UpgradeReadToWrite(0);
AuTryRemove(this->sources_, source);
this->sourceMutex_->DowngradeWriteToRead();
}
if (remove)
{
AU_LOCK_GUARD(this->polledItemsMutex_->AsReadable());
for (auto epoll : this->alternativeEpolls_)
{
epoll->Remove(source.get(), readData, writeData);
}
}
}
now = AuTime::CurrentClockMS();
if (!bTicked)
{
if (tryAgain)
{
*tryAgain = ((this->lockStealer_.IsSignaled()) ||
(now < time));
}
}
out:
if (!now)
{
now = AuTime::CurrentClockMS();
}
for (auto itr = this->sources_.begin(); itr != this->sources_.end(); )
{
AuSPtr<SourceExtended> source = *itr;
bool remove {};
if (!remove)
{
remove = source->ConsiderTimeout(now);
}
if (remove)
{
this->sourceMutex_->UpgradeReadToWrite(0);
itr = this->sources_.erase(itr);
this->sourceMutex_->DowngradeWriteToRead();
}
if (remove)
{
AU_LOCK_GUARD(this->polledItemsMutex_->AsReadable());
for (auto epoll : this->alternativeEpolls_)
{
epoll->Remove(source.get(), true, true);
}
}
if (source->sourceExtended)
{
source->sourceExtended->OnFinishSleep();
}
if (!remove)
{
itr ++;
}
}
return bTicked;
}
LoopQueue::SourceExtended::SourceExtended(LoopQueue *parent, const AuSPtr<ILoopSource> &source) :
parent(parent),
source(source)
{
this->sourceExtended = AuDynamicCast<ILoopSourceEx>(source.get());
}
LoopQueue::SourceExtended::~SourceExtended()
{
Deinit();
}
void LoopQueue::SourceExtended::Deinit()
{
this->pin.reset();
}
void LoopQueue::SourceExtended::Commit(const AuSPtr<SourceExtended> &self)
{
this->pin = self;
this->bHasCommited = true;
}
AuPair<bool, bool> LoopQueue::SourceExtended::DoWork(bool read, bool write)
{
if (!this->sourceExtended)
{
return DoWork(-1);
}
if (this->sourceExtended->Singular())
{
AuPair<bool, bool> ret;
if (read)
{
auto [a, b] = DoWork(this->sourceExtended->GetHandle());
ret.first |= a;
ret.second |= b;
}
if (write)
{
auto [a, b] = DoWork(this->sourceExtended->GetWriteHandle());
ret.first |= a;
ret.second |= b;
}
return ret;
}
else
{
// Whatever, I doubt implementing this is worth the perf hit
return DoWork(-1);
}
}
AuPair<bool, bool> LoopQueue::SourceExtended::DoWork(int fd)
{
bool bShouldRemove {true};
if (!this->bHasCommited)
{
return {};
}
if (this->sourceExtended)
{
if (!this->sourceExtended->OnTrigger(fd))
{
return {};
}
}
for (const auto &handler : this->subscribers)
{
try
{
bShouldRemove &= handler->OnFinished(this->source);
}
catch (...)
{
SysPushErrorCatch();
}
}
if (bShouldRemove)
{
for (const auto &handler : this->subscriberExs)
{
try
{
bShouldRemove &= handler->OnFinished(this->source);
}
catch (...)
{
SysPushErrorCatch();
}
}
}
if (bShouldRemove)
{
AU_LOCK_GUARD(this->parent->globalLockMutex_);
for (const auto &handler : this->parent->allSubscribers_)
{
try
{
bShouldRemove &= handler->OnFinished(this->source);
}
catch (...)
{
SysPushErrorCatch();
}
}
}
return AuMakePair(true, bShouldRemove);
}
AUKN_SYM AuSPtr<ILoopQueue> NewLoopQueue()
{
return AuMakeShared<LoopQueue>();
auto queue = AuMakeShared<LoopQueue>();
if (!queue)
{
return {};
}
if (!queue->Init())
{
return {};
}
return queue;
}
}

81
Source/Loop/LoopQueue.Linux.hpp
View File

@ -8,6 +8,7 @@
#pragma once
#include "ILoopSourceEx.hpp"
#include "LSEvent.hpp"
namespace Aurora::Loop
{
@ -40,39 +41,87 @@ namespace Aurora::Loop
AuUInt32 WaitAny(AuUInt32 timeout) override;
AuList<AuSPtr<ILoopSource>> WaitAnyEx(AuUInt32 timeout) override;
void DoTick();
AuUInt32 DoTick(AuUInt64, AuList<AuSPtr<ILoopSource>> *optOut = nullptr, bool *tryAgain = nullptr);
private:
struct SpecialHandle
{
int fd;
void *priv;
};
bool CommitDecommit();
struct SourceExtended
{
SpecialHandle internal;
AuSPtr<void> pin;
SourceExtended(LoopQueue *parent, const AuSPtr<ILoopSource> &source);
~SourceExtended();
AuList<AuSPtr<ILoopSourceSubscriber>> subscribers;
void Deinit();
void Commit(const AuSPtr<SourceExtended> &self);
AuSPtr<ILoopSource> source;
ILoopSourceEx *sourceExtended;
LoopQueue *parent;
AuWPtr<SourceExtended> pin;
AuUInt64 timeoutAbs;
bool ConsiderTimeout(AuUInt64 time) const
{
if ((timeoutAbs) && (time >= timeoutAbs))
{
for (const auto &handler : subscriberExs)
{
try
{
handler->OnTimeout(source);
}
catch (...)
{
SysPushErrorCatch();
}
}
return true;
}
return false;
}
AuList<AuSPtr<ILoopSourceSubscriber>> subscribers;
AuList<AuSPtr<ILoopSourceSubscriberEx>> subscriberExs;
bool bHasCommited {};
// ticked, should remove
AuPair<bool, bool> DoWork(int fd);
AuPair<bool, bool> DoWork(bool read, bool write);
};
struct AnEpoll
{
LoopQueue *parent;
AuThreadPrimitives::SpinLock commitQueueMutex_;
AuList<AuSPtr<SourceExtended>> commitQueue_;
AuThreadPrimitives::SpinLock lock;
AuBST<int, int> startingWorkRead;
AuBST<int, int> startingWorkWrite;
void Add(SourceExtended *source);
void Remove(SourceExtended *source, bool readData, bool writeData);
};
AuThreadPrimitives::SpinLock globalLockMutex_;
int epollFd_{ -1 };
LSEvent lockStealer_;
AuThreadPrimitives::SpinLock commitQueueMutex_;
AuList<AuTuple<AuSPtr<ILoopSource>, AuSPtr<ILoopSourceSubscriber>, AuSPtr<ILoopSourceSubscriberEx>>> commitPending_;
AuList<AuSPtr<ILoopSource>> decommitQueue_;
AuThreadPrimitives::SpinLock globalLockMutex_;
AuList<AuSPtr<ILoopSourceSubscriber>> allSubscribers_;
AuThreadPrimitives::RWLockUnique_t sourceMutex_;
AuList<AuSPtr<SourceExtended>> sources_;
AuThreadPrimitives::RWLockUnique_t polledItemsMutex_;
AuList<AnEpoll *> alternativeEpolls_;
int epollFd_ {-1};
bool hasActiveBeforeCommit_ {};
AnEpoll globalEpoll_;
};
}

6
Source/Loop/LoopQueue.NT.cpp
View File

@ -56,7 +56,7 @@ namespace Aurora::Loop
void LoopQueue::Sync()
{
if (this->hEvent_ != INVALID_HANDLE_VALUE)
if (this->hEvent_ == INVALID_HANDLE_VALUE)
{
this->rwMutex_->AsWritable()->Lock();
return;
@ -418,7 +418,6 @@ namespace Aurora::Loop
bool LoopQueue::WaitAll(AuUInt32 timeout)
{
// TODO:
AU_LOCK_GUARD(this->rwMutex_->AsReadable());
bool bReturnStatus {true};
@ -441,8 +440,6 @@ namespace Aurora::Loop
source.source->OnPresleep();
}
bool active = this->hEvent_ == INVALID_HANDLE_VALUE;
while (count != index)
{
auto next = AuMin(count - index, AuUInt32(MAXIMUM_WAIT_OBJECTS));
@ -460,7 +457,6 @@ namespace Aurora::Loop
auto timeDelta = endTime - startTime; // TODO: cap to last obj
DWORD status {};
// TODO: queue apc
if (this->bIsWinLoop_)
{
status = ::MsgWaitForMultipleObjectsEx(next, this->handleArrayAnd_.data() + index, timeDelta, QS_ALLPOSTMESSAGE | QS_ALLINPUT | QS_ALLEVENTS, MWMO_INPUTAVAILABLE | MWMO_ALERTABLE | MWMO_WAITALL);

3
Source/Loop/LoopQueue.NT.hpp
View File

@ -60,7 +60,7 @@ namespace Aurora::Loop
AuSPtr<ILoopSourceEx> source;
AuSPtr<ILoopSource> sourceBase;
AuThreadPrimitives::SpinLock lock; // im too brain dead to think of a solution to the AddCallback thread safety issue, so i'm going to optimize Wait[...]s subscriber lookup with filtered ccaches in here, protected by this spinlock
AuUInt32 timeoutAbs;
AuUInt64 timeoutAbs;
SourceCallbacks callbacks;
bool ConsiderTimeout(AuUInt64 time) const
@ -82,7 +82,6 @@ namespace Aurora::Loop
return true;
}
return false;
}
};

38
Source/Processes/Open.Win32.cpp
View File

@ -33,7 +33,34 @@ namespace Aurora::Processes
{
for (const auto &open : gOpenItems)
{
ShellExecuteW(NULL, AuIOFS::DirExists(open) ? L"explore" : NULL, Locale::ConvertFromUTF8(open).c_str(), NULL, NULL, SW_SHOWNORMAL);
if (open.empty())
{
// We probably ran out of memory.
// AuProcess/Open can safely drop as we expect shells to be kinda fucky and async
//
// Case in point: Minecraft on Linux (would?) blocks when you click a link in chat
//
// Fuck tons of applications support clicking of links, in the case of TS and others, allowing for RCE.
// In the case of MC and others, they don't even know if the operation blocks until the process closes.
// Assuming non-blocking, the API returns false on failure; but if it's blocking, who knows what that
// means... Nonzero exit code? Not enough resources? No error?
//
// Websites, programs, and scripts wouldn't know how to process "missing protocol handler,"
// "not enough resources," "process crashed before pump," "shell busy." For the most part, we don't
// expect expect the developer to be aware of what happens after a request to open a resource is
// requested. It's a lot of engineering effort for what should be fork, exec("start", ...)
//
// Dropping invalid paths, out of memory during UTF8 conversion, and other IO issues is probably fine.
// Use an actual IProcess object, if you care about spawning and monitoring executables.
continue;
}
ShellExecuteW(nullptr,
AuIOFS::DirExists(open) ? L"explore" : L"open",
Locale::ConvertFromUTF8(open).c_str(),
nullptr,
nullptr,
SW_SHOWNORMAL);
}
gOpenItems.clear();
gCondVariable->WaitForSignal();
@ -48,7 +75,7 @@ namespace Aurora::Processes
static void OpenerThread()
{
CoInitializeEx(NULL, COINIT_APARTMENTTHREADED | COINIT_DISABLE_OLE1DDE);
CoInitializeEx(nullptr, COINIT_APARTMENTTHREADED | COINIT_DISABLE_OLE1DDE);
RunTasks();
CoUninitialize();
}
@ -83,7 +110,7 @@ namespace Aurora::Processes
AUKN_SYM void OpenUri(const AuString &uri)
{
AU_LOCK_GUARD(gCondMutex);
gOpenItems.push_back(uri);
AuTryInsert(gOpenItems, uri);
gCondVariable->Broadcast();
}
@ -91,4 +118,9 @@ namespace Aurora::Processes
{
OpenUri(AuIOFS::NormalizePathRet(file));
}
// TODO: Consider creating blocking apis whose return value is an IProcess (construct from ShellExecuteExW -> in.hProcess, or ("xdg-start", ...))
// For the most part, blocking for a specific application in the context of a protocol or file open request is a dated computing construct.
// Nowdays, opening an editor, mail client, or such like means poking a single executable that'll spawn a fuck ton of background workers, io threads,
// and other resources, to manage multiple instances of whatever the application deals with (think: editor tabs; browser windows; sendto: isnt a modal)
}

2
Source/RNG/RandomDevice.cpp
View File

@ -171,6 +171,6 @@ namespace Aurora::RNG
AUKN_SYM void RandomRelease(IRandomDevice *stream)
{
AuSafeDelete<IRandomDevice*>(stream);
AuSafeDelete<RandomDevice*>(stream);
}
}

1
Source/Threading/Threads/SpawnThread.Unix.cpp
View File

@ -38,6 +38,7 @@ namespace Aurora::Threading::Threads
auto callMe = *handle;
delete handle;
callMe();
return nullptr;
};
auto ret = pthread_attr_init(&tattr);

11
Source/Time/Time.hpp
View File

@ -48,5 +48,16 @@ namespace Aurora::Time
ts->tv_nsec = remainderNS;
}
static void ms2tsabsmono(struct timespec *ts, unsigned long ms)
{
clock_gettime(CLOCK_MONOTONIC, ts);
auto baseNS = ((AuUInt64)ms * (AuUInt64)1'000'000) + (AuUInt64)ts->tv_nsec;
auto remainderNS = (AuUInt64)baseNS % (AuUInt64)1'000'000'000;
ts->tv_sec += baseNS / 1'000'000'000ull;
ts->tv_nsec = remainderNS;
}
#endif
}