atomic_wait.hpp

浏览该文件的文档.

#ifndef NEFORCE_CORE_ASYNC_ATOMIC_WAIT_HPP__
#define NEFORCE_CORE_ASYNC_ATOMIC_WAIT_HPP__

 
#include "NeForce/core/async/futex.hpp"
#include "NeForce/core/async/this_thread.hpp"
#include "NeForce/core/memory/memory.hpp"
NEFORCE_BEGIN_NAMESPACE__



struct default_spin_policy {
    bool operator()() const noexcept { return false; }
};

template <typename Pred, typename Spin = default_spin_policy>
bool atomic_spin(Pred& pred, Spin spin = Spin{}) noexcept {
    constexpr auto atomic_spin_count = 16;
    constexpr auto atomic_spin_count_relax = 12;
    for (auto idx = 0; idx < atomic_spin_count; ++idx) {
        if (pred()) {
            return true;
        }
        if (idx < atomic_spin_count_relax) {
            this_thread::relax();
        } else {
            this_thread::yield();
        }
    }
 
    while (spin()) {
        if (pred()) {
            return true;
        }
    }
    return false;
}

NEFORCE_BEGIN_INNER__

struct waiter_pool_base {
    static constexpr auto align_inner = 64; 
 
    alignas(align_inner) platform_wait_t wait = 0;  
    alignas(align_inner) platform_wait_t value = 0; 
 
    waiter_pool_base() = default;

    void waiter_enter_wait() noexcept {
#ifdef NEFORCE_PLATFORM_WINDOWS
        ::_InterlockedIncrement(&wait);
#else
        __atomic_fetch_add(&wait, 1, __ATOMIC_SEQ_CST);
#endif
    }

    void waiter_leave_wait() noexcept {
#ifdef NEFORCE_PLATFORM_WINDOWS
        ::_InterlockedDecrement(&wait);
#else
        __atomic_fetch_sub(&wait, 1, __ATOMIC_RELEASE);
#endif
    }

    bool waiter_waiting() const noexcept {
#ifdef NEFORCE_PLATFORM_WINDOWS
        platform_wait_t res = ::_InterlockedExchangeAdd(const_cast<volatile platform_wait_t*>(&wait), 0);
#else
        platform_wait_t res;
        __atomic_load(&wait, &res, __ATOMIC_SEQ_CST);
#endif
        return res != 0;
    }

    void waiter_notify(platform_wait_t* addr, bool all, const bool bare) const noexcept {
        if (addr == &value) {
#ifdef NEFORCE_PLATFORM_WINDOWS
            ::_InterlockedIncrement(addr);
#else
            __atomic_fetch_add(addr, 1, __ATOMIC_SEQ_CST);
#endif
            all = true;
        }
        if (bare || waiter_waiting()) {
            futex_notify(addr, all);
        }
    }

    static waiter_pool_base& waiter_for(const void* addr) noexcept {
        constexpr uintptr_t pool_size = 16;
        static waiter_pool_base waiter[pool_size];
        const auto key = (reinterpret_cast<uintptr_t>(addr) >> 2) % pool_size;
        return waiter[key];
    }
};

struct waiter_pool : waiter_pool_base {
    NEFORCE_ALWAYS_INLINE void waiter_do_wait(platform_wait_t* addr, const platform_wait_t old) const noexcept {
        _NEFORCE futex_wait(addr, old);
    }
};
 

template <typename T>
struct waiter_base {
private:
    template <typename U>
    static constexpr bool platform_wait_valid_v =
            is_scalar_v<U> && sizeof(U) == sizeof(platform_wait_t) && alignof(U*) >= alignof(platform_wait_t);
 
    template <typename U, enable_if_t<platform_wait_valid_v<U>, int> = 0>
    NEFORCE_ALWAYS_INLINE static void waiter_do_spin_v_impl(platform_wait_t*, const U& old, platform_wait_t& value) {
        _NEFORCE memory_copy(&value, &old, sizeof(value));
    }
    template <typename U, enable_if_t<!platform_wait_valid_v<U>, int> = 0>
    NEFORCE_ALWAYS_INLINE static void waiter_do_spin_v_impl(platform_wait_t* addr, const U&, platform_wait_t& value) {
#ifdef NEFORCE_PLATFORM_WINDOWS
        value = ::_InterlockedExchangeAdd(addr, 0);
#else
        __atomic_load(addr, &value, __ATOMIC_ACQUIRE);
#endif
    }
 
public:
    using waiter_type = T; 
 
    waiter_type& waiter_;   
    platform_wait_t* addr_; 

    template <typename U>
    static enable_if_t<platform_wait_valid_v<U>, platform_wait_t*> waiter_wait_addr(const U* addr, platform_wait_t*) {
        return reinterpret_cast<platform_wait_t*>(const_cast<U*>(addr));
    }

    template <typename U>
    static enable_if_t<!platform_wait_valid_v<U>, platform_wait_t*> waiter_wait_addr(const U*, platform_wait_t* wait) {
        return wait;
    }

    static waiter_type& waiter_for(const void* addr) noexcept {
        static_assert(sizeof(waiter_type) == sizeof(waiter_pool_base),
                      "waiter_for should be same size with waiter_pool_base");
        auto& res = waiter_pool_base::waiter_for(addr);
        return reinterpret_cast<waiter_type&>(res);
    }

    template <typename U>
    explicit waiter_base(const U* addr) noexcept :
    waiter_(waiter_base::waiter_for(addr)),
    addr_(waiter_base::waiter_wait_addr(addr, &waiter_.value)) {}

    void waiter_notify(bool all, bool bare = false) noexcept { waiter_.waiter_notify(addr_, all, bare); }

    template <typename U, typename Func, typename Spin = default_spin_policy>
    static bool waiter_do_spin_v(platform_wait_t* addr, const U& old, Func f, platform_wait_t& value,
                                 Spin spin = Spin{}) {
        auto const pred = [=] { return _NEFORCE memory_compare<U>(old, f()) != 0; };
        waiter_base::waiter_do_spin_v_impl(addr, old, value);
        return _NEFORCE atomic_spin(pred, spin);
    }

    template <typename U, typename Func, typename Spin = default_spin_policy>
    bool waiter_do_spin_v(const U& old, Func f, platform_wait_t& value, Spin spin = Spin{}) {
        return waiter_base::waiter_do_spin_v(addr_, old, f, value, spin);
    }

    template <typename Pred, typename Spin = default_spin_policy>
    static bool waiter_do_spin(const platform_wait_t* addr, Pred pred, platform_wait_t& value, Spin spin = Spin{}) {
#ifdef NEFORCE_PLATFORM_WINDOWS
        value = ::_InterlockedExchangeAdd(const_cast<volatile LONG*>(addr), 0);
#else
        __atomic_load(addr, &value, __ATOMIC_ACQUIRE);
#endif
        return _NEFORCE atomic_spin(pred, spin);
    }

    template <typename Pred, typename Spin = default_spin_policy>
    bool waiter_do_spin(Pred pred, platform_wait_t& value, Spin spin = Spin{}) {
        return waiter_base::waiter_do_spin(addr_, pred, value, spin);
    }
};
 

template <typename EntersWait>
struct waiter : waiter_base<waiter_pool> {
public:
    using base_type = waiter_base<waiter_pool>;
 
private:
    template <bool Wait = EntersWait::value, enable_if_t<Wait, int> = 0>
    NEFORCE_ALWAYS_INLINE void enter() const noexcept {
        waiter_.waiter_enter_wait();
    }
    template <bool Wait = EntersWait::value, enable_if_t<!Wait, int> = 0>
    NEFORCE_ALWAYS_INLINE void enter() const noexcept {}
 
    template <bool Wait = EntersWait::value, enable_if_t<Wait, int> = 0>
    NEFORCE_ALWAYS_INLINE void leave() const noexcept {
        waiter_.waiter_leave_wait();
    }
    template <bool Wait = EntersWait::value, enable_if_t<!Wait, int> = 0>
    NEFORCE_ALWAYS_INLINE void leave() const noexcept {}
 
public:
    template <typename T>
    explicit waiter(const T* addr) noexcept :
    base_type(addr) {
        enter();
    }

    ~waiter() { leave(); }

    template <typename T, typename Func>
    void waiter_do_wait_v(T old, Func f) {
        do {
            platform_wait_t value;
            if (base_type::waiter_do_spin_v(old, f, value)) {
                return;
            }
            waiter_.waiter_do_wait(base_type::addr_, value);
        } while (_NEFORCE memory_compare<T>(old, f()) == 0);
    }

    template <typename Pred>
    void waiter_do_wait(Pred pred) noexcept {
        do {
            platform_wait_t value;
            if (base_type::waiter_do_spin(pred, value)) {
                return;
            }
            waiter_.waiter_do_wait(base_type::addr_, value);
        } while (!pred());
    }
};

using enters_wait = waiter<true_type>;

using bare_wait = waiter<false_type>;
 
NEFORCE_END_INNER__

template <typename T, typename Func>
void atomic_wait_address_v(const T* addr, T old, Func f) noexcept {
    inner::enters_wait waiter(addr);
    waiter.waiter_do_wait_v(old, f);
}

template <typename T, typename Pred>
void atomic_wait_address(const T* addr, Pred pred) noexcept {
    inner::enters_wait waiter(addr);
    waiter.waiter_do_wait(pred);
}

template <typename T>
void atomic_notify_address(const T* addr, const bool all) noexcept {
    inner::bare_wait waiter(addr);
    waiter.waiter_notify(all);
}
 // AtomicOperations
 // AsyncComponents
 
NEFORCE_END_NAMESPACE__
#endif // NEFORCE_CORE_ASYNC_ATOMIC_WAIT_HPP__