AuroraRuntime/Source/Threading/Primitives/AuConditionVariable.Linux.cpp

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

    File: AuConditionVariable.Linux.cpp
    Date: 2023-8-11
    Author: Reece
***/
#include <Source/RuntimeInternal.hpp>
#include "AuConditionVariable.Generic.hpp"
#include <Source/Time/Time.hpp>
#include "SMTYield.hpp"

#if !defined(_AURUNTIME_GENERICCV)

namespace Aurora::Threading::Primitives
{
    ConditionVariableImpl::ConditionVariableImpl(const AuSPtr<IConditionMutex> &pMutex) : 
            mutex_(AuStaticCast<LinuxConditionMutex>(pMutex))
    {

    }
    
    ConditionVariableImpl::~ConditionVariableImpl()
    {

    }

    AuSPtr<IConditionMutex> ConditionVariableImpl::GetMutex()
    {
        return this->mutex_;
    }

    bool ConditionVariableImpl::WaitForSignal(AuUInt32 qwTimeoutRelative)
    {
        return WaitForSignalNS(AuMSToNS<AuUInt64>(qwTimeoutRelative));
    }

    bool ConditionVariableImpl::WaitForSignalNS(AuUInt64 qwTimeoutRelative)
    {
        return WaitForSignalNsEx(this->mutex_, qwTimeoutRelative);
    }

    bool ConditionVariableImpl::WaitOne(AuUInt64 qwTimeoutRelative)
    {
        AuUInt64 uStart {};
        AuUInt64 uEnd {};
        struct timespec tspec;

        if (qwTimeoutRelative != 0)
        {
            uStart = AuTime::SteadyClockNS();
            uEnd   = uStart + qwTimeoutRelative;
            
            Time::monoabsns2ts(&tspec, uEnd);
        }

        if (DoTryIf([=]()
        {
            auto old = this->uState_;
            return (old != 0 && AuAtomicCompareExchange(&this->uState_, old - 1, old) == old);
        }))
        {
            return true;
        }

        auto old = this->uState_;
        while (!((old != 0) &&
                 (AuAtomicCompareExchange(&this->uState_, old - 1, old) == old)))
        {
            if (qwTimeoutRelative != 0)
            {
                if (Time::SteadyClockNS() >= uEnd)
                {
                    return false;
                }

                int ret {};
                do
                {
                    ret = futex_wait(&this->uState_, 0, &tspec);
                }
                while (ret == EINTR);
            }
            else
            {
                int ret {};
                bool bStatus {};

                do
                {
                    if ((ret = futex_wait(&this->uState_, 0)) == 0)
                    {
                        bStatus = true;
                        continue;
                    }

                    if (ret == EAGAIN || errno == EAGAIN)
                    {
                        bStatus = true;
                        continue;
                    }
                    
                }
                while (ret == EINTR);

                RUNTIME_ASSERT_SHUTDOWN_SAFE(bStatus, "semaphore wait failed: {}", ret)
            }

            old = this->uState_;
        }

        return true;
    }

    bool ConditionVariableImpl::WaitForSignalNsEx(const std::shared_ptr<LinuxConditionMutex> &pMutex, 
                                                  AuUInt64 qwTimeoutRelative)
    {
    
        AuAtomicAdd(&this->uSleeping_, 1u);

        if (pMutex)
        {
            pMutex->Unlock();
        }

        auto bSuccess = this->WaitOne(qwTimeoutRelative);
        if (!bSuccess)
        {
            auto uWaiters = this->uSleeping_;
            auto uWaitCount = 1;

            while (AuAtomicCompareExchange(&this->uSleeping_, uWaiters - uWaitCount, uWaiters) != uWaiters)
            {
                uWaiters = this->uSleeping_;

                if (uWaiters == 0)
                {
                    break;
                }
            }
        }

        if (pMutex)
        {
            pMutex->Lock();
        }

        return bSuccess;
    }

    void ConditionVariableImpl::Signal()
    {
        AuUInt32 uWaitCount {};
        AuUInt32 uWaiters {};

        uWaiters = this->uSleeping_;
        if (uWaiters > 0)
        {
            AuAtomicAdd(&this->uState_, 1u);
            futex_wake(&this->uState_, 1u);
            uWaitCount = 1;
        }

        while (AuAtomicCompareExchange(&this->uSleeping_, uWaiters - uWaitCount, uWaiters) != uWaiters)
        {
            uWaiters = this->uSleeping_;

            if (uWaiters == 0)
            {
                return;
            }
        }
    }

    void ConditionVariableImpl::Broadcast()
    {
        AuUInt32 uWaitCount {};
        AuUInt32 uWaiters {};

        uWaiters = this->uSleeping_;
        if (uWaiters > 0)
        {
            AuAtomicAdd(&this->uState_, uWaiters);
            futex_wake(&this->uState_, uWaiters);
            uWaitCount = uWaiters;
        }

        while (AuAtomicCompareExchange(&this->uSleeping_, uWaiters - uWaitCount, uWaiters) != uWaiters)
        {
            uWaiters = this->uSleeping_;

            if (uWaiters <= uWaitCount)
            {
                uWaitCount = uWaiters;
            }
        }
    }

    AUKN_SYM IConditionVariable *ConditionVariableNew(const AuSPtr<IConditionMutex> &mutex)
    {
        return _new ConditionVariableImpl(mutex);
    }

    AUKN_SYM void ConditionVariableRelease(IConditionVariable *mutex)
    {
        AuSafeDelete<ConditionVariableImpl *>(mutex);
    }

    AUROXTL_INTERFACE_SOO_SRC(ConditionVariable, ConditionVariableImpl, (const AuSPtr<IConditionMutex> &, pMutex))
}

#endif