AuroraRuntime/Source/IO/Loop/LoopQueue.Linux.hpp

/***
    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::IO::Loop
{
    struct LoopQueue : ILoopQueue
    {
        LoopQueue();
        ~LoopQueue();

        bool AddHook(const AuFunction<void()> &func);

        bool Init();
        void Deinit();

        bool SourceAdd(const AuSPtr<ILoopSource> &source) override;
        bool SourceAddWithTimeout(const AuSPtr<ILoopSource> &source, AuUInt32 ms) override;
        bool SourceAddWithTimeoutEx(const AuSPtr<ILoopSource> &source, AuUInt32 ms) ;
        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;

        void ConfigureDoIOApcCallbacks(bool bDoIOAPCs) override;
        void ConfigureBreakAnyAfterAPC(bool bAPCBreaks) 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;

    private:
        struct AnEpoll;

    public:
        AuUInt32 DoTick(AuUInt64, AuList<AuSPtr<ILoopSource>> *optOut = nullptr, bool *tryAgain = nullptr, bool async = false, bool all = false, AnEpoll *pAllEpoll = nullptr);

    private:

        void PumpHooks();

        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);
            AuTuple<bool, bool, bool> DoWorkLate();
            bool CanDispatchLayer(int fd);
            bool CanDispatchLayer(bool read, bool write);
        };

        struct AnEpoll
        {
            LoopQueue *parent {};

            bool bAll {};
            AuThreadPrimitives::SpinLock lock;
            AuBST<int, int> startingWorkRead;
            AuBST<int, int> startingWorkWrite;
            AuList<AuSPtr<SourceExtended>> done;

            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_;

        AuList<AuFunction<void()>>            epilogueHooks_;
        
        AuList<AuPair<AuSPtr<SourceExtended>, AuUInt32>> pendingBlocking_;
        bool bRecommitLater {};

        AnEpoll globalEpoll_;
        bool bDoIOAPCs_ { true };
        bool bAPCBreaks_ { false };
    };
}