AuroraRuntime/Source/IO/Loop/Loop.cpp

/***
    Copyright (C) 2021 J Reece Wilson (a/k/a "Reece"). All rights reserved.

    File: Loop.cpp
    Date: 2021-9-21
    Author: Reece
***/
#include <Source/RuntimeInternal.hpp>
#include "Loop.hpp"
#include "ILoopSourceEx.hpp"

namespace Aurora::IO::Loop
{
#if !defined(AURORA_IS_MODERNNT_DERIVED)
    AUKN_SYM AuSPtr<ILoopSource>    NewLSWin32Source(bool)
    {
        return {};
    }
#endif

#if !defined(AURORA_IS_XNU_DERIVED)
    AUKN_SYM AuSPtr<ILoopSource>    NewLSAppleSource()
    {
        return {};
    }
#endif

    AUKN_SYM AuSPtr<ILoopSource>    NewLSFile(const AuSPtr<AuIO::IAsyncTransaction> &pFileTransaction)
    {
        if (!pFileTransaction)
        {
            SysPushErrorArg();
            return {};
        }

        return pFileTransaction->NewLoopSource();
    }

    AUKN_SYM AuSPtr<ILoopSource>    NewStdIn()
    {
        return AuConsole::StdInBufferLoopSource();
    }

    AUKN_SYM AuSPtr<ILoopSource>    NewLSAsync(Async::WorkerPId_t workerPid)
    {
        if (!workerPid.pool)
        {
            return Async::GetAsyncApp()->WorkerToLoopSource(workerPid);
        }

        return workerPid.pool->WorkerToLoopSource(workerPid);
    }

#if defined(AURORA_PLATFORM_WIN32)
    AuList<AuSPtr<ILoopSource>> WaitMultipleOrObjects(const AuList<AuSPtr<ILoopSource>> &objects, AuUInt32 timeout);
#endif
    AuList<AuSPtr<ILoopSource>> WaitMultipleOrObjectsFallback(const AuList<AuSPtr<ILoopSource>> &objects, AuUInt32 timeout, bool &bTimeout);

    void ResetLoopSourceFalseAlarm(const AuSPtr<Loop::ILoopSource> &pLoopSource)
    {
        if (!pLoopSource)
        {
            return;
        }

        if (pLoopSource->GetType() == ELoopSource::eSourceWin32)
        {
            return;
        }

        if (auto pMutex = AuDynamicCast<Loop::ILSMutex>(pLoopSource))
        {
            pMutex->Unlock();
            return;
        }

        if (auto pSemaphore = AuDynamicCast<Loop::ILSSemaphore>(pLoopSource))
        {
            pSemaphore->AddOne();
            return;
        }

        if (auto pEvent = AuDynamicCast<Loop::ILSEvent>(pLoopSource))
        {
            pEvent->Set();
            return;
        }
    }

    AUKN_SYM bool WaitMultipleLoopSources(const AuList<AuSPtr<Loop::ILoopSource>> &lsList,
                                          AuList<AuSPtr<Loop::ILoopSource>> &signaled,
                                          bool bAny,
                                          AuOptionalEx<AuUInt32> uTimeoutMS)
    {
        AU_DEBUG_MEMCRUNCH;

        signaled.clear();
        AuList<AuUInt32> reverseList;

        if (lsList.empty())
        {
            return true;
        }

        auto uTimeoutEnd = uTimeoutMS ?
            AuTime::SteadyClockNS() + AuMSToNS<AuUInt64>(uTimeoutMS) :
            0;
        
        signaled.reserve(lsList.size());

        if (!bAny)
        {
            AuSPtr<Loop::ILoopQueue> pQueue;
            auto &entryZero = lsList[0];

            if (!entryZero)
            {
                signaled.push_back({});
            }

            if (entryZero)
            {
                if (!entryZero->WaitOn(uTimeoutMS))
                {
                    goto next;
                }
                else
                {
                    reverseList.push_back(0);
                    signaled.push_back(entryZero);
                }
            }

            if (lsList.size() > 1)
            {
                pQueue = AuLoop::NewLoopQueue();
                if (!pQueue)
                {
                    goto next;
                }

                for (AU_ITERATE_N_TO_X(i, 1, lsList.size()))
                {
                    AuUInt32 uTimeoutMS {};

                    if (uTimeoutEnd)
                    {
                        auto uStartTime = Time::SteadyClockNS();
                        if (uStartTime >= uTimeoutEnd)
                        {
                            break;
                        }

                        uTimeoutMS = AuNSToMS<AuInt64>(uTimeoutEnd - uStartTime);
                        if (!uTimeoutMS)
                        {
                            break;
                        }
                    }

                    if (!lsList[i]->WaitOn(uTimeoutMS))
                    {
                        break;
                    }

                    reverseList.push_back(i);
                    signaled.push_back(lsList[i]);
                }
            }

        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 bTimeout {};
            signaled = WaitMultipleOrObjectsFallback(lsList, uTimeoutMS, bTimeout);

            if (bTimeout)
            {
                return false;
            }

            return true;
        }
    }

    AuList<AuSPtr<ILoopSource>> WaitMultipleOrObjectsFallback(const AuList<AuSPtr<ILoopSource>> &objects, AuUInt32 timeout, bool &bTimeout)
    {
        AuList<AuSPtr<ILoopSourceEx>> loopSourceExs;
        AuList<AuSPtr<ILoopSource>> triggered;

        bTimeout  = false;

        auto pQueue = AuLoop::NewLoopQueue();
        if (!pQueue)
        {
            return {};
        }

        try
        {
            loopSourceExs.reserve(objects.size());
            triggered.reserve(triggered.size());

            for (const auto &source : objects)
            {
                if (!source)
                {
                    continue;
                }

                if (!pQueue->SourceAdd(source))
                {
                    return {};
                }

                if (source->GetType() == ELoopSource::eSourceWin32)
                {
                    continue;
                }

                if (auto extended = AuDynamicCast<ILoopSourceEx>(source))
                {
                    if (!AuTryInsert(loopSourceExs, extended))
                    {
                        return {};
                    }
                }
            }
        }
        catch (...)
        {
            return {};
        }

        auto pListener = AuMakeSharedThrow<AuLoop::ILoopSourceSubscriberFunctional>([&](const AuSPtr<ILoopSource> &source)
        {
            AU_DEBUG_MEMCRUNCH;
            triggered.push_back(source);
            return false;
        });

        if (!pQueue->AddCallback(pListener))
        {
            return {};
        }

        for (const auto &source : loopSourceExs)
        {
            source->OnPresleep();
        }
       
        bTimeout = !pQueue->WaitAny(timeout);

        for (AU_ITERATE_N(i, objects.size()))
        {
            auto source = loopSourceExs[i];
         
            if (std::find(triggered.begin(), triggered.end(), source) == triggered.end())
            {
                if (source->IsSignaled())
                {
                    triggered.push_back(source);
                }
            }

            source->OnFinishSleep();
        }

        return triggered;
    }
}