AuroraRuntime/Source/Loop/LoopQueue.Linux.hpp
J Reece Wilson f1ef6ac43c [+] Async/overlapped UNIX pipe IO
[*] Further work on Linux LoopQueue for WaitAll
2022-04-17 23:46:05 +01:00

129 lines
4.0 KiB
C++

/***
Copyright (C) 2022 J Reece Wilson (a/k/a "Reece"). All rights reserved.
File: LoopQueue.Linux.hpp
Date: 2022-4-4
Author: Reece
***/
#pragma once
#include "ILoopSourceEx.hpp"
#include "LSEvent.hpp"
namespace Aurora::Loop
{
struct LoopQueue : ILoopQueue
{
LoopQueue();
~LoopQueue();
bool Init();
void Deinit();
bool SourceAdd(const AuSPtr<ILoopSource> &source) override;
bool SourceAddWithTimeout(const AuSPtr<ILoopSource> &source, AuUInt32 ms) override;
bool SourceRemove(const AuSPtr<ILoopSource> &source) override;
AuUInt32 GetSourceCount() override;
virtual bool AddCallback(const AuSPtr<ILoopSource> &source, const AuSPtr<ILoopSourceSubscriber> &subscriber) override;
virtual bool AddCallbackEx(const AuSPtr<ILoopSource> &source, const AuSPtr<ILoopSourceSubscriberEx> &subscriber) override;
virtual bool AddCallback(const AuSPtr<ILoopSourceSubscriber> &subscriber) override;
void ChugPathConfigure(AuUInt32 sectionTickTime, AuSInt sectionDequeCount) override;
void ChugHint(bool value) override;
virtual bool Commit() override;
bool IsSignaledPeek() override;
AuUInt32 PumpNonblocking() override;
AuList<AuSPtr<ILoopSource>> PumpNonblockingEx() override;
bool WaitAll(AuUInt32 timeout) override;
AuUInt32 WaitAny(AuUInt32 timeout) override;
AuList<AuSPtr<ILoopSource>> WaitAnyEx(AuUInt32 timeout) override;
AuUInt32 DoTick(AuUInt64, AuList<AuSPtr<ILoopSource>> *optOut = nullptr, bool *tryAgain = nullptr, bool async = false);
private:
bool CommitDecommit();
struct SourceExtended
{
SourceExtended(LoopQueue *parent, const AuSPtr<ILoopSource> &source);
~SourceExtended();
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
AuTuple<bool, bool, bool> DoWork(int fd);
AuTuple<bool, bool, bool> DoWork(bool read, bool write);
};
struct AnEpoll
{
LoopQueue *parent;
AuThreadPrimitives::SpinLock lock;
AuBST<int, int> startingWorkRead;
AuBST<int, int> startingWorkWrite;
void Add(SourceExtended *source);
void Remove(SourceExtended *source, bool readData, bool writeData);
};
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_;
AnEpoll globalEpoll_;
};
}