AuroraRuntime/Source/Threading/Primitives/AuConditionEx.cpp

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

    File: AuConditionEx.cpp
    Date: 2021-6-12
    Author: Reece
***/
#include <Source/RuntimeInternal.hpp>
#include "AuConditionEx.hpp"
#include "AuSemaphore.Generic.hpp"

namespace Aurora::Threading::Primitives
{
    struct SemaphoreConditionVariableImpl : ConditionEx
    {
        SemaphoreConditionVariableImpl();

        void WaitForSignal(IWaitable *pWaitable) override;
        void WaitForSignal(const AuSPtr<IWaitable> &pWaitable) override;
        void WaitForSignal() override;
        
        bool WaitForSignalNS(AuUInt64 uRelativeNanoseconds) override;
        bool WaitForSignalNS(IWaitable *waitable, AuUInt64 uRelativeNanoseconds) override;
        bool WaitForSignalNS(const AuSPtr<IWaitable> &waitable, AuUInt64 uRelativeNanoseconds) override;

        void Signal() override;
        void Broadcast() override;

    private:
        SemaphoreImpl s_;
        AuAUInt32 uWaiters_;
    };

    SemaphoreConditionVariableImpl::SemaphoreConditionVariableImpl() :
        s_(0),
        uWaiters_(0)
    {

    }

    bool SemaphoreConditionVariableImpl::WaitForSignalNS(AuUInt64 uRelativeNanoseconds)
    {
        return WaitForSignalNS(nullptr, uRelativeNanoseconds);
    }

    bool SemaphoreConditionVariableImpl::WaitForSignalNS(IWaitable *pWaitable, AuUInt64 uRelativeNanoseconds)
    {
        return WaitForSignalNS(pWaitable ? AuUnsafeRaiiToShared(pWaitable) : AuSPtr<IWaitable> {},
                               uRelativeNanoseconds);
    }

    bool SemaphoreConditionVariableImpl::WaitForSignalNS(const AuSPtr<IWaitable> &pWaitable,
                                                         AuUInt64 uRelativeNanoseconds)
    {
        AuAtomicAdd(&this->uWaiters_, 1u);

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

        auto bSuccess = this->s_.LockNS(uRelativeNanoseconds);
        if (!bSuccess)
        {
            auto uWaiters = this->uWaiters_;
            auto uWaitCount = 1;

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

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

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

        return bSuccess;
    }

    void SemaphoreConditionVariableImpl::WaitForSignal(IWaitable *pWaitable)
    {
        return WaitForSignal(AuUnsafeRaiiToShared(pWaitable));
    }

    void SemaphoreConditionVariableImpl::WaitForSignal(const AuSPtr<IWaitable> &pWaitable)
    {
        AuAtomicAdd(&this->uWaiters_, 1u);

        if (pWaitable)
        {
            pWaitable->Unlock();
        }
        
        this->s_.Lock();

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

    void SemaphoreConditionVariableImpl::WaitForSignal() 
    {
        WaitForSignal(nullptr);
    }

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

        uWaiters = AuAtomicLoad(&this->uWaiters_);
        if (uWaiters == 0)
        {
            return;
        }

        this->s_.Unlock();

        while (uWaiters &&
               AuAtomicCompareExchange(&this->uWaiters_, uWaiters - 1u, uWaiters) != uWaiters)
        {
            uWaiters = this->uWaiters_;

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

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

        while ((uWaiters = AuAtomicLoad(&this->uWaiters_)))
        {
            this->s_.Unlock(uWaiters);
            uWaitCount = uWaiters;

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

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

    AUKN_SYM ConditionEx *FlexibleConditionVariableNew()
    {
        return _new SemaphoreConditionVariableImpl();
    }

    AUKN_SYM void FlexibleConditionVariableRelease(ConditionEx *pCVEx)
    {
        AuSafeDelete<SemaphoreConditionVariableImpl *>(pCVEx);
    }

    AUROXTL_INTERFACE_SOO_SRC_EX(AURORA_SYMBOL_EXPORT, FlexibleConditionVariable, SemaphoreConditionVariableImpl)
}