// Copyright (c) 2020-2023, NVIDIA CORPORATION. All rights reserved.
//
// NVIDIA CORPORATION and its licensors retain all intellectual property
// and proprietary rights in and to this software, related documentation
// and any modifications thereto. Any use, reproduction, disclosure or
// distribution of this software and related documentation without an express
// license agreement from NVIDIA CORPORATION is strictly prohibited.
//
#pragma once

#include "../Defines.h"
#include "../cpp/Bit.h"
#include "../math/Util.h"
#include "../thread/Util.h"

#include <atomic>

#if CARB_PLATFORM_WINDOWS
#    pragma comment(lib, "synchronization.lib") // must link with synchronization.lib
#    include "../CarbWindows.h"
#elif CARB_PLATFORM_LINUX
#    include <linux/futex.h>
#    include <sys/syscall.h>
#    include <sys/time.h>

#    include <unistd.h>
#elif CARB_PLATFORM_MACOS
/* nothing for now */
#else
CARB_UNSUPPORTED_PLATFORM();
#endif

namespace carb
{
namespace thread
{
namespace detail
{

template <class T, size_t S = sizeof(T)>
struct to_integral
{
};

template <class T>
struct to_integral<T, 1>
{
    using type = int8_t;
};

template <class T>
struct to_integral<T, 2>
{
    using type = int16_t;
};

template <class T>
struct to_integral<T, 4>
{
    using type = int32_t;
};

template <class T>
struct to_integral<T, 8>
{
    using type = int64_t;
};

template <class T>
using to_integral_t = typename to_integral<T>::type;

template <class As, class T>
CARB_NODISCARD std::enable_if_t<std::is_integral<T>::value && sizeof(As) == sizeof(T), As> reinterpret_as(const T& in) noexcept
{
    static_assert(std::is_integral<As>::value, "Must be integral type");
    return static_cast<As>(in);
}

template <class As, class T>
CARB_NODISCARD std::enable_if_t<std::is_pointer<T>::value && sizeof(As) == sizeof(T), As> reinterpret_as(const T& in) noexcept
{
    static_assert(std::is_integral<As>::value, "Must be integral type");
    return reinterpret_cast<As>(in);
}

template <class As, class T>
CARB_NODISCARD std::enable_if_t<(!std::is_pointer<T>::value && !std::is_integral<T>::value) || sizeof(As) != sizeof(T), As> reinterpret_as(
    const T& in) noexcept
{
    static_assert(std::is_integral<As>::value, "Must be integral type");
    As out{}; // Init to zero
    memcpy(&out, std::addressof(in), sizeof(in));
    return out;
}

template <class Duration>
Duration clampDuration(Duration offset)
{
    using namespace std::chrono;
    constexpr static Duration Max = duration_cast<Duration>(milliseconds(0x7fffffff));
    return ::carb_max(Duration(0), ::carb_min(Max, offset));
}


#if CARB_PLATFORM_WINDOWS
// Windows WaitOnAddress() supports 1, 2, 4 or 8 bytes, so it doesn't need to use ParkingLot. For testing ParkingLot
// or for specific modules this can be enabled though.
#    ifndef CARB_USE_PARKINGLOT
#        define CARB_USE_PARKINGLOT 0
#    endif

using hundrednanos = std::chrono::duration<int64_t, std::ratio<1, 10'000'000>>;

template <class T>
inline bool WaitOnAddress(const std::atomic<T>& val, T compare, int64_t* timeout) noexcept
{
    static_assert(sizeof(val) == sizeof(compare), "Invalid assumption about atomic");
    // Use the NTDLL version of this function since we can give it relative or absolute times in 100ns units
    using RtlWaitOnAddressFn = DWORD(__stdcall*)(volatile const void*, void*, size_t, int64_t*);
    static RtlWaitOnAddressFn RtlWaitOnAddress =
        (RtlWaitOnAddressFn)GetProcAddress(GetModuleHandleW(L"ntdll.dll"), "RtlWaitOnAddress");

    volatile const T* addr = reinterpret_cast<volatile const T*>(std::addressof(val));
    switch (DWORD ret = RtlWaitOnAddress(addr, &compare, sizeof(compare), timeout))
    {
        case CARBWIN_STATUS_SUCCESS:
            return true;

        default:
            CARB_FATAL_UNLESS(0, "Unexpected result from RtlWaitOnAddress: %u, GetLastError=%u", ret, ::GetLastError());
            CARB_FALLTHROUGH; // (not really, but the compiler doesn't know that CARB_FATAL_UNLESS doesn't return)
        case CARBWIN_STATUS_TIMEOUT:
            return false;
    }
}

template <class T>
inline void futex_wait(const std::atomic<T>& val, T compare) noexcept
{
    WaitOnAddress(val, compare, nullptr);
}

template <class T, class Rep, class Period>
inline bool futex_wait_for(const std::atomic<T>& val, T compare, std::chrono::duration<Rep, Period> duration)
{
    // RtlWaitOnAddress treats negative timeouts as positive relative time
    int64_t timeout = -std::chrono::duration_cast<hundrednanos>(clampDuration(duration)).count();
    if (timeout >= 0)
    {
        // duration to wait is negative
        return false;
    }

    CARB_ASSERT(timeout < 0);
    return WaitOnAddress(val, compare, &timeout);
}

template <class T, class Clock, class Duration>
inline bool futex_wait_until(const std::atomic<T>& val, T compare, std::chrono::time_point<Clock, Duration> time_point)
{
    int64_t absTime;
    auto now = Clock::now();

    // RtlWaitOnAddress is quite slow to return if the time has already elapsed. It's much faster for us to check first.
    if (time_point <= now)
    {
        return false;
    }

    // Constrain the time to something that is well before the heat death of the universe
    auto tp = now + clampDuration(time_point - now);

    // if ((void*)std::addressof(Clock::now) != (void*)std::addressof(std::chrono::system_clock::now))
    if (!std::is_same<Clock, std::chrono::system_clock>::value)
    {
        // If we're not using the system clock, then we need to convert to the system clock
        absTime = std::chrono::duration_cast<detail::hundrednanos>(
                      (tp - now + std::chrono::system_clock::now()).time_since_epoch())
                      .count();
    }
    else
    {
        // Already in terms of system clock
        // According to https://github.com/microsoft/STL/blob/master/stl/inc/chrono, the system_clock appears to
        // use GetSystemTimePreciseAsFileTime minus an epoch value so that it lines up with 1/1/1970 midnight GMT.
        // Unfortunately there's not an easy way to check for it here, but we have a unittest in TestSemaphore.cpp.
        absTime = std::chrono::duration_cast<detail::hundrednanos>(tp.time_since_epoch()).count();
    }

    // Epoch value from https://github.com/microsoft/STL/blob/master/stl/src/xtime.cpp
    // This is the number of 100ns units between 1 January 1601 00:00 GMT and 1 January 1970 00:00 GMT
    constexpr int64_t kFiletimeEpochToUnixEpochIn100nsUnits = 0x19DB1DED53E8000LL;
    absTime += kFiletimeEpochToUnixEpochIn100nsUnits;

    CARB_ASSERT(absTime >= 0);
    return detail::WaitOnAddress(val, compare, &absTime);
}

template <class T>
inline void futex_wake_one(std::atomic<T>& val) noexcept
{
    WakeByAddressSingle(std::addressof(val));
}

template <class T>
inline void futex_wake_n(std::atomic<T>& val, size_t n) noexcept
{
    while (n--)
        futex_wake_one(val);
}

template <class T>
inline void futex_wake_all(std::atomic<T>& val) noexcept
{
    WakeByAddressAll(std::addressof(val));
}

#    if !CARB_USE_PARKINGLOT
template <class T, size_t S = sizeof(T)>
class Futex
{
    static_assert(S == 1 || S == 2 || S == 4 || S == 8, "Unsupported size");

public:
    using AtomicType = typename std::atomic<T>;
    using Type = T;
    static inline void wait(const AtomicType& val, Type compare) noexcept
    {
        futex_wait(val, compare);
    }
    template <class Rep, class Period>
    static inline bool wait_for(const AtomicType& val, Type compare, std::chrono::duration<Rep, Period> duration)
    {
        return futex_wait_for(val, compare, duration);
    }
    template <class Clock, class Duration>
    static inline bool wait_until(const AtomicType& val, Type compare, std::chrono::time_point<Clock, Duration> time_point)
    {
        return futex_wait_until(val, compare, time_point);
    }
    static inline void notify_one(AtomicType& a) noexcept
    {
        futex_wake_one(a);
    }
    static inline void notify_n(AtomicType& a, size_t n) noexcept
    {
        futex_wake_n(a, n);
    }
    static inline void notify_all(AtomicType& a) noexcept
    {
        futex_wake_all(a);
    }
};
#    endif

#elif CARB_PLATFORM_LINUX
#    define CARB_USE_PARKINGLOT 1 // Linux only supports 4 byte futex so it must use the ParkingLot

constexpr int64_t kNsPerSec = 1'000'000'000;

inline int futex(const std::atomic_uint32_t& aval,
                 int futex_op,
                 uint32_t val,
                 const struct timespec* timeout,
                 uint32_t* uaddr2,
                 int val3) noexcept
{
    static_assert(sizeof(aval) == sizeof(uint32_t), "Invalid assumption about atomic");
    int ret = syscall(SYS_futex, std::addressof(aval), futex_op, val, timeout, uaddr2, val3);
    return ret >= 0 ? ret : -errno;
}

inline void futex_wait(const std::atomic_uint32_t& val, uint32_t compare) noexcept
{
    for (;;)
    {
        int ret = futex(val, FUTEX_WAIT_BITSET_PRIVATE, compare, nullptr, nullptr, FUTEX_BITSET_MATCH_ANY);
        switch (ret)
        {
            case 0:
            case -EAGAIN: // Valid or spurious wakeup
                return;

            case -ETIMEDOUT:
                // Apparently on Windows Subsystem for Linux, calls to the kernel can timeout even when a timeout value
                // was not specified. Fall through.
            case -EINTR: // Interrupted by signal; loop again
                break;

            default:
                CARB_FATAL_UNLESS(0, "Unexpected result from futex(): %d/%s", -ret, strerror(-ret));
        }
    }
}

template <class Rep, class Period>
inline bool futex_wait_for(const std::atomic_uint32_t& val, uint32_t compare, std::chrono::duration<Rep, Period> duration)
{
    // Relative time
    int64_t ns = std::chrono::duration_cast<std::chrono::nanoseconds>(clampDuration(duration)).count();
    if (ns <= 0)
    {
        return false;
    }

    struct timespec ts;
    ts.tv_sec = time_t(ns / detail::kNsPerSec);
    ts.tv_nsec = long(ns % detail::kNsPerSec);

    // Since we're using relative time here, we can use FUTEX_WAIT_PRIVATE (see futex() man page)
    int ret = futex(val, FUTEX_WAIT_PRIVATE, compare, &ts, nullptr, 0);
    switch (ret)
    {
        case 0: // Valid wakeup
        case -EAGAIN: // Valid or spurious wakeup
        case -EINTR: // Interrupted by signal; treat as a spurious wakeup
            return true;

        default:
            CARB_FATAL_UNLESS(0, "Unexpected result from futex(): %d/%s", -ret, strerror(-ret));
            CARB_FALLTHROUGH; // (not really but the compiler doesn't know that the above won't return)
        case -ETIMEDOUT:
            return false;
    }
}

template <class Clock, class Duration>
inline bool futex_wait_until(const std::atomic_uint32_t& val,
                             uint32_t compare,
                             std::chrono::time_point<Clock, Duration> time_point)
{
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);

    // Constrain the time to something that is well before the heat death of the universe
    auto now = Clock::now();
    auto tp = now + clampDuration(time_point - now);

    // Get the number of nanoseconds to go
    int64_t ns = std::chrono::duration_cast<std::chrono::nanoseconds>(tp - now).count();
    if (ns <= 0)
    {
        return false;
    }

    ts.tv_sec += time_t(ns / kNsPerSec);
    ts.tv_nsec += long(ns % kNsPerSec);

    // Handle rollover
    if (ts.tv_nsec >= kNsPerSec)
    {
        ++ts.tv_sec;
        ts.tv_nsec -= kNsPerSec;
    }

    for (;;)
    {
        // Since we're using absolute monotonic time, we use FUTEX_WAIT_BITSET_PRIVATE. See the man page for futex for
        // more info.
        int ret = futex(val, FUTEX_WAIT_BITSET_PRIVATE, compare, &ts, nullptr, FUTEX_BITSET_MATCH_ANY);
        switch (ret)
        {
            case 0: // Valid wakeup
            case -EAGAIN: // Valid or spurious wakeup
                return true;

            case -EINTR: // Interrupted by signal; loop again
                break;

            default:
                CARB_FATAL_UNLESS(0, "Unexpected result from futex(): %d/%s", -ret, strerror(-ret));
                CARB_FALLTHROUGH; // (not really but the compiler doesn't know that the above won't return)
            case -ETIMEDOUT:
                return false;
        }
    }
}

inline void futex_wake_n(std::atomic_uint32_t& val, unsigned count) noexcept
{
    int ret = futex(val, FUTEX_WAKE_BITSET_PRIVATE, count, nullptr, nullptr, FUTEX_BITSET_MATCH_ANY);
    CARB_ASSERT(ret >= 0, "futex(FUTEX_WAKE) failed with errno=%d/%s", -ret, strerror(-ret));
    CARB_UNUSED(ret);
}

inline void futex_wake_one(std::atomic_uint32_t& val) noexcept
{
    futex_wake_n(val, 1);
}

inline void futex_wake_all(std::atomic_uint32_t& val) noexcept
{
    futex_wake_n(val, INT_MAX);
}
#elif CARB_PLATFORM_MACOS
#    define CARB_USE_PARKINGLOT 1

#    define UL_COMPARE_AND_WAIT 1
#    define UL_UNFAIR_LOCK 2
#    define UL_COMPARE_AND_WAIT_SHARED 3
#    define UL_UNFAIR_LOCK64_SHARED 4
#    define UL_COMPARE_AND_WAIT64 5
#    define UL_COMPARE_AND_WAIT64_SHARED 6

#    define ULF_WAKE_ALL 0x00000100
#    define ULF_WAKE_THREAD 0x00000200
#    define ULF_NO_ERRNO 0x01000000

/** Undocumented OSX futex-like call.
 *  @param     operation A combination of the UL_* and ULF_* flags.
 *  @param[in] addr      The address to wait on.
 *                       This is a 32 bit value unless UL_COMPARE_AND_WAIT64 is passed.
 *  @param     value     The address's previous value.
 *  @param     timeout   Timeout in microseconds.
 *  @returns    0 or a positive value (representing additional waiters) on success, or a negative value on error. If a
 *              negative value is returned, `errno` will be set, unless ULF_NO_ERRNO is provided, in which case the
 *              return value is the negated error value.
 */
extern "C" int __ulock_wait(uint32_t operation, void* addr, uint64_t value, uint32_t timeout);

/** Undocumented OSX futex-like call.
 *  @param      operation A combination of the UL_* and ULF_* flags.
 *  @param[in]  addr The address to wake, passed to __ulock_wait by other threads.
 *  @param      wake_value An extra value to be interpreted based on \p operation. If `ULF_WAKE_THREAD` is provided,
 *              then this is the mach_port of the specific thread to wake.
 *  @returns    0 or a positive value (representing additional waiters) on success, or a negative value on error. If a
 *              negative value is returned, `errno` will be set, unless ULF_NO_ERRNO is provided, in which case the
 *              return value is the negated error value.
 */
extern "C" int __ulock_wake(uint32_t operation, void* addr, uint64_t wake_value);


inline void futex_wait(const std::atomic_uint32_t& val, uint32_t compare) noexcept
{
    for (;;)
    {
        int rc = __ulock_wait(UL_COMPARE_AND_WAIT | ULF_NO_ERRNO,
                              const_cast<uint32_t*>(reinterpret_cast<const uint32_t*>(std::addressof(val))), compare, 0);
        if (rc >= 0)
        {
            // According to XNU source, the non-negative return value is the number of remaining waiters.
            // See ulock_wait_cleanup in sys_ulock.c
            return;
        }
        switch (-rc)
        {
            case EINTR: // According to XNU source, EINTR can be returned.
                continue;
            case ETIMEDOUT:
                CARB_FALLTHROUGH;
            case EFAULT:
                CARB_FALLTHROUGH;
            default:
                CARB_FATAL_UNLESS(0, "Unexpected result from __ulock_wait: %d/%s", -rc, strerror(-rc));
        }
    }
}

template <class Rep, class Period>
inline bool futex_wait_for(const std::atomic_uint32_t& val, uint32_t compare, std::chrono::duration<Rep, Period> duration)
{
    // Relative time
    int64_t usec = std::chrono::duration_cast<std::chrono::microseconds>(clampDuration(duration)).count();
    if (usec <= 0)
    {
        return false;
    }
    if (usec > UINT32_MAX)
    {
        usec = UINT32_MAX;
    }

    int rc = __ulock_wait(UL_COMPARE_AND_WAIT | ULF_NO_ERRNO,
                          const_cast<uint32_t*>(reinterpret_cast<const uint32_t*>(std::addressof(val))), compare, usec);
    if (rc >= 0)
    {
        // According to XNU source, the non-negative return value is the number of remaining waiters.
        // See ulock_wait_cleanup in sys_ulock.c
        return true;
    }
    switch (-rc)
    {
        case EINTR: // Treat signal interrupt as a spurious wakeup
            return true;
        case ETIMEDOUT:
            return false;
        default:
            CARB_FATAL_UNLESS(0, "Unexpected result from __ulock_wait: %d/%s", -rc, strerror(-rc));
    }
}

template <class Clock, class Duration>
inline bool futex_wait_until(const std::atomic_uint32_t& val,
                             uint32_t compare,
                             std::chrono::time_point<Clock, Duration> time_point)
{
    // Constrain the time to something that is well before the heat death of the universe
    auto now = Clock::now();
    auto tp = now + clampDuration(time_point - now);

    // Convert to number of microseconds from now
    int64_t usec = std::chrono::duration_cast<std::chrono::microseconds>(tp - now).count();
    if (usec <= 0)
    {
        return false;
    }
    if (usec > UINT32_MAX)
    {
        usec = UINT32_MAX;
    }

    int rc = __ulock_wait(UL_COMPARE_AND_WAIT | ULF_NO_ERRNO,
                          const_cast<uint32_t*>(reinterpret_cast<const uint32_t*>(std::addressof(val))), compare, usec);
    if (rc >= 0)
    {
        // According to XNU source, the non-negative return value is the number of remaining waiters.
        // See ulock_wait_cleanup in sys_ulock.c
        return true;
    }
    switch (-rc)
    {
        case EINTR: // Treat signal interrupt as a spurious wakeup
            return true;
        case ETIMEDOUT:
            return false;
        default:
            CARB_FATAL_UNLESS(0, "Unexpected result from __ulock_wait: %d/%s", -rc, strerror(-rc));
    }
}

inline void futex_wake_n(std::atomic_uint32_t& val, unsigned count) noexcept
{
    for (unsigned i = 0; i < count; i++)
    {
        __ulock_wake(UL_COMPARE_AND_WAIT, std::addressof(val), 0);
    }
}

inline void futex_wake_one(std::atomic_uint32_t& val) noexcept
{
    __ulock_wake(UL_COMPARE_AND_WAIT, std::addressof(val), 0);
}

inline void futex_wake_all(std::atomic_uint32_t& val) noexcept
{
    __ulock_wake(UL_COMPARE_AND_WAIT | ULF_WAKE_ALL, std::addressof(val), 0);
}
#endif

class NativeFutex
{
public:
    using AtomicType = std::atomic_uint32_t;
    using Type = uint32_t;
    static inline void wait(const AtomicType& val, Type compare) noexcept
    {
        futex_wait(val, compare);
    }
    template <class Rep, class Period>
    static inline bool wait_for(const AtomicType& val, Type compare, std::chrono::duration<Rep, Period> duration)
    {
        return futex_wait_for(val, compare, duration);
    }
    template <class Clock, class Duration>
    static inline bool wait_until(const AtomicType& val, Type compare, std::chrono::time_point<Clock, Duration> time_point)
    {
        return futex_wait_until(val, compare, time_point);
    }
    static inline void notify_one(AtomicType& a) noexcept
    {
        futex_wake_one(a);
    }
    static inline void notify_n(AtomicType& a, size_t n) noexcept
    {
        futex_wake_n(a, n);
    }
    static inline void notify_all(AtomicType& a) noexcept
    {
        futex_wake_all(a);
    }
};

#if CARB_USE_PARKINGLOT
struct ParkingLot
{

    struct WaitEntry
    {
        const void* addr;
        WaitEntry* next{ nullptr };
        WaitEntry* prev{ nullptr };
        uint32_t changeId{ 0 };
        NativeFutex::AtomicType wakeup{ 0 };

        enum Bits : uint32_t
        {
            kNoBits = 0,
            kNotifyBit = 1,
            kWaitBit = 2,
        };
    };

    class WaitBucket
    {
        // In an effort to conserve size and implement appearsEmpty(), we use the least significant bit as a lock bit.
        constexpr static size_t kUnlocked = 0;
        constexpr static size_t kLock = 1;

        union
        {
            std::atomic<WaitEntry*> m_head{ nullptr };
            std::atomic_size_t m_lock;
            // The native futex is only 32 bits, so m_lock must coincide with the LSBs of m_lock, hence we assert
            // little-endian
            NativeFutex::AtomicType m_futex;
            static_assert(carb::cpp::endian::native == carb::cpp::endian::little, "Requires little endian");
        };
        WaitEntry* m_tail;
        std::atomic_uint32_t m_waiters{ 0 };
        std::atomic_uint32_t m_changeTracker{ 0 };

        void setHead(WaitEntry* newHead) noexcept
        {
            assertLockState();
            assertNoLockBits(newHead);
            size_t val = size_t(newHead);
            // Maintain bits but set to new value
            // Relaxed semantics because we own the lock so shouldn't need to synchronize-with any other threads
            m_lock.store(val | kLock, std::memory_order_relaxed);
        }

        void setHeadAndUnlock(WaitEntry* newHead) noexcept
        {
            assertLockState();
            assertNoLockBits(newHead);
            size_t val = size_t(newHead);
            m_lock.store(val, std::memory_order_seq_cst);
            if (m_waiters.load(std::memory_order_relaxed))
                NativeFutex::notify_one(m_futex);
        }

        void assertLockState() const noexcept
        {
            // Relaxed because this should only be done within the lock
            CARB_ASSERT(m_lock.load(std::memory_order_relaxed) & kLock);
        }

        static void assertNoLockBits(WaitEntry* e) noexcept
        {
            CARB_UNUSED(e);
            CARB_ASSERT(!(size_t(e) & kLock));
        }

        void assertHead(WaitEntry* e) const noexcept
        {
            // Relaxed because this should only be done within the lock
            CARB_UNUSED(e);
            CARB_ASSERT((m_lock.load(std::memory_order_relaxed) & ~kLock) == size_t(e));
        }

    public:
        constexpr WaitBucket() noexcept : m_head{ nullptr }, m_tail{ nullptr }
        {
        }

        void incChangeTracker() noexcept
        {
            assertLockState(); // under lock
            m_changeTracker.store(m_changeTracker.load(std::memory_order_relaxed) + 1, std::memory_order_relaxed);
        }

        bool appearsEmpty() const noexcept
        {
            // fence-fence synchronization with wait functions
            this_thread::atomic_fence_seq_cst();
            return m_lock.load(std::memory_order_relaxed) == 0;
        }

        std::atomic_uint32_t& changeTracker() noexcept
        {
            return m_changeTracker;
        }

        WaitEntry* lock() noexcept
        {
            size_t val = m_lock.load(std::memory_order_relaxed);
            for (;;)
            {
                if (!(val & kLock))
                {
                    if (m_lock.compare_exchange_strong(val, val | kLock))
                        return reinterpret_cast<WaitEntry*>(val);
                    continue;
                }
                if (!this_thread::spinTryWait([&] { return !((val = m_lock.load(std::memory_order_relaxed)) & kLock); }))
                {
                    ++m_waiters;
                    while ((val = m_lock.load(std::memory_order_relaxed)) & kLock)
                        carb::thread::detail::NativeFutex::wait(m_futex, uint32_t(val));
                    --m_waiters;
                }
            }
        }

        void unlockHint(WaitEntry* head) noexcept
        {
            assertLockState();
            assertNoLockBits(head);
            assertHead(head);
            m_lock.store(size_t(head), std::memory_order_seq_cst);
            if (m_waiters.load(std::memory_order_relaxed))
                NativeFutex::notify_one(m_futex);
        }

        void unlock() noexcept
        {
            unlockHint(reinterpret_cast<WaitEntry*>(m_lock.load(std::memory_order_relaxed) & ~kLock));
        }

        void appendAndUnlock(WaitEntry* e) noexcept
        {
            assertLockState();
            assertNoLockBits(e);
            e->prev = m_tail;
            e->next = nullptr;
            if (e->prev)
            {
                m_tail = e->prev->next = e;
                unlock();
            }
            else
            {
                m_tail = e;
                setHeadAndUnlock(e);
            }
        }

        void remove(WaitEntry* e) noexcept
        {
            assertLockState();
            if (e->next)
                e->next->prev = e->prev;
            else
            {
                CARB_ASSERT(m_tail == e);
                m_tail = e->prev;
            }
            if (e->prev)
                e->prev->next = e->next;
            else
            {
                assertHead(e);
                setHead(e->next);
            }
        }

        void removeAndUnlock(WaitEntry* e) noexcept
        {
            assertLockState();
            assertNoLockBits(e);
            if (e->next)
                e->next->prev = e->prev;
            else
            {
                CARB_ASSERT(m_tail == e);
                m_tail = e->prev;
            }
            if (e->prev)
            {
                e->prev->next = e->next;
                unlock();
            }
            else
            {
                assertHead(e);
                setHeadAndUnlock(e->next);
            }
        }

        constexpr static size_t kNumWaitBuckets = 2048; // Must be a power of two
        static_assert(carb::cpp::has_single_bit(kNumWaitBuckets), "Invalid assumption");

        static inline WaitBucket& bucket(const void* addr) noexcept
        {
            static WaitBucket waitBuckets[kNumWaitBuckets];

#    if 1
            // FNV-1a hash is fast with really good distribution
            // In "futex buckets" test, about ~70% on Windows and ~80% on Linux
            auto hash = carb::hashBuffer(&addr, sizeof(addr));
            return waitBuckets[hash & (kNumWaitBuckets - 1)];
#    else
            // Simple bitshift method
            // In "futex buckets" test:
            //  >> 4 bits: ~71% on Windows, ~72% on Linux
            //  >> 5 bits: ~42% on Windows, ~71% on Linux
            return waitBuckets[(size_t(addr) >> 4) & (kNumWaitBuckets - 1)];
#    endif
        }
    };

    template <typename T>
    static void wait(const std::atomic<T>& val, T compare) noexcept
    {
        WaitEntry entry{ std::addressof(val) };

        using I = to_integral_t<T>;

        // Check before waiting
        if (reinterpret_as<I>(val.load(std::memory_order_acquire)) != reinterpret_as<I>(compare))
        {
            return;
        }

        WaitBucket& b = WaitBucket::bucket(std::addressof(val));
        auto hint = b.lock();

        // Check inside the lock to reduce spurious wakeups
        if (CARB_UNLIKELY(reinterpret_as<I>(val.load(std::memory_order_acquire)) != reinterpret_as<I>(compare)))
        {
            b.unlockHint(hint);
            return;
        }

        entry.changeId = b.changeTracker().load(std::memory_order_relaxed);
        b.appendAndUnlock(&entry);

        // fence-fence synchronization with appearsEmpty()
        this_thread::atomic_fence_seq_cst();

        // Do the wait if everything is consistent
        if (CARB_LIKELY(reinterpret_as<I>(val.load(std::memory_order_relaxed)) == reinterpret_as<I>(compare) &&
                        b.changeTracker().load(std::memory_order_relaxed) == entry.changeId))
        {
            NativeFutex::wait(entry.wakeup, uint32_t(WaitEntry::kNoBits));
        }

        // Speculatively see if we've been removed
        uint32_t v = entry.wakeup.load(std::memory_order_acquire);
        if (CARB_UNLIKELY(!v))
        {
            // Need to remove
            hint = b.lock();
            // Check again under the lock (relaxed because we're under the lock)
            v = entry.wakeup.load(std::memory_order_relaxed);
            if (!v)
            {
                b.removeAndUnlock(&entry);
                return;
            }
            else
            {
                // Already removed
                b.unlockHint(hint);
            }
        }

        // Spin briefly while the wait bit is set, though this should be rare
        this_thread::spinWait([&] { return !(entry.wakeup.load(std::memory_order_acquire) & WaitEntry::kWaitBit); });
    }

    template <typename T, typename Rep, typename Period>
    static bool wait_for(const std::atomic<T>& val, T compare, std::chrono::duration<Rep, Period> duration)
    {
        WaitEntry entry{ std::addressof(val) };

        using I = to_integral_t<T>;

        // Check before waiting
        if (reinterpret_as<I>(val.load(std::memory_order_acquire)) != reinterpret_as<I>(compare))
        {
            return true;
        }

        WaitBucket& b = WaitBucket::bucket(std::addressof(val));
        auto hint = b.lock();

        // Check inside the lock to reduce spurious wakeups
        if (CARB_UNLIKELY(reinterpret_as<I>(val.load(std::memory_order_acquire)) != reinterpret_as<I>(compare)))
        {
            b.unlockHint(hint);
            return true;
        }

        entry.changeId = b.changeTracker().load(std::memory_order_relaxed);
        b.appendAndUnlock(&entry);

        // fence-fence synchronization with appearsEmpty()
        this_thread::atomic_fence_seq_cst();

        // Do the wait if everything is consistent
        bool finished = true;
        if (CARB_LIKELY(reinterpret_as<I>(val.load(std::memory_order_relaxed)) == reinterpret_as<I>(compare) &&
                        b.changeTracker().load(std::memory_order_relaxed) == entry.changeId))
        {
            finished = NativeFutex::wait_for(entry.wakeup, uint32_t(WaitEntry::kNoBits), duration);
        }

        // Speculatively see if we've been removed
        uint32_t v = entry.wakeup.load(std::memory_order_acquire);
        if (CARB_UNLIKELY(!v))
        {
            // Need to remove
            hint = b.lock();
            // Check again under the lock (relaxed because we're under the lock)
            v = entry.wakeup.load(std::memory_order_relaxed);
            if (!v)
            {
                b.removeAndUnlock(&entry);
                return finished;
            }
            else
            {
                // Already removed
                b.unlockHint(hint);
                finished = true;
            }
        }

        // Spin briefly while the wait bit is set, though this should be rare
        this_thread::spinWait([&] { return !(entry.wakeup.load(std::memory_order_acquire) & WaitEntry::kWaitBit); });

        return finished;
    }

    template <typename T, typename Clock, typename Duration>
    static bool wait_until(const std::atomic<T>& val, T compare, std::chrono::time_point<Clock, Duration> time_point)
    {
        WaitEntry entry{ std::addressof(val) };

        using I = to_integral_t<T>;

        // Check before waiting
        if (reinterpret_as<I>(val.load(std::memory_order_acquire)) != reinterpret_as<I>(compare))
        {
            return true;
        }

        WaitBucket& b = WaitBucket::bucket(std::addressof(val));
        auto hint = b.lock();

        // Check inside the lock to reduce spurious wakeups
        if (CARB_UNLIKELY(reinterpret_as<I>(val.load(std::memory_order_acquire)) != reinterpret_as<I>(compare)))
        {
            b.unlockHint(hint);
            return true;
        }

        entry.changeId = b.changeTracker().load(std::memory_order_relaxed);
        b.appendAndUnlock(&entry);

        // fence-fence synchronization with appearsEmpty()
        this_thread::atomic_fence_seq_cst();

        // Do the wait if everything is consistent
        bool finished = true;
        if (CARB_LIKELY(reinterpret_as<I>(val.load(std::memory_order_relaxed)) == reinterpret_as<I>(compare) &&
                        b.changeTracker().load(std::memory_order_relaxed) == entry.changeId))
        {
            finished = NativeFutex::wait_until(entry.wakeup, uint32_t(WaitEntry::kNoBits), time_point);
        }

        // Speculatively see if we've been removed
        uint32_t v = entry.wakeup.load(std::memory_order_acquire);
        if (CARB_UNLIKELY(!v))
        {
            // Need to remove
            hint = b.lock();
            // Check again under the lock (relaxed because we're under the lock)
            v = entry.wakeup.load(std::memory_order_relaxed);
            if (!v)
            {
                b.removeAndUnlock(&entry);
                return finished;
            }
            else
            {
                // Already removed
                b.unlockHint(hint);
                finished = true;
            }
        }

        // Spin briefly while the wait bit is set, though this should be rare
        this_thread::spinWait([&] { return !(entry.wakeup.load(std::memory_order_acquire) & WaitEntry::kWaitBit); });

        return finished;
    }

    static void notify_one(void* addr) noexcept
    {
        WaitBucket& b = WaitBucket::bucket(addr);

        // Read empty state with full fence to avoid locking
        if (b.appearsEmpty())
            return;

        WaitEntry* head = b.lock();
        b.incChangeTracker();
        for (WaitEntry* e = head; e; e = e->next)
        {
            if (e->addr == addr)
            {
                // Remove before setting the wakeup flag
                b.remove(e);

                // Even though we're under the lock, we need release semantics to synchronize-with the wait not under
                // the lock.
                e->wakeup.store(WaitEntry::kNotifyBit, std::memory_order_release);

                b.unlock();

                // Wake the waiter
                NativeFutex::notify_one(e->wakeup);
                return;
            }
        }
        b.unlockHint(head);
    }

    constexpr static size_t kWakeCacheSize = 128;

private:
    static CARB_NOINLINE void notify_n_slow(
        void* addr, size_t n, WaitBucket& b, WaitEntry* e, std::atomic_uint32_t** wakeCache) noexcept
    {
        // We are waking a lot of things and the wakeCache is completely full. Now we need to start building a list
        WaitEntry *wake = nullptr, *end = nullptr;

        for (WaitEntry* next; e; e = next)
        {
            next = e->next;
            if (e->addr == addr)
            {
                b.remove(e);
                e->next = nullptr;
                // Don't bother with prev pointers
                if (end)
                    end->next = e;
                else
                    wake = e;
                end = e;
                // Even though we're under the lock we use release semantics here to synchronize-with the waking thread
                // possibly outside the lock.
                // Need to set the wait bit since we're still reading/writing to the WaitEntry
                e->wakeup.store(WaitEntry::kWaitBit | WaitEntry::kNotifyBit, std::memory_order_release);

                if (!--n)
                    break;
            }
        }

        b.unlock();

        // Wake the entire cache since we know it's full
        auto wakeCacheEnd = (wakeCache + kWakeCacheSize);
        do
        {
            NativeFutex::notify_one(**(wakeCache++));
        } while (wakeCache != wakeCacheEnd);

        for (WaitEntry* next; wake; wake = next)
        {
            next = wake->next;
            // Clear the wait bit so that only the wake bit is set
            wake->wakeup.store(WaitEntry::kNotifyBit, std::memory_order_release);
            NativeFutex::notify_one(wake->wakeup);
        }
    }

    static CARB_NOINLINE void notify_all_slow(void* addr, WaitBucket& b, WaitEntry* e, std::atomic_uint32_t** wakeCache) noexcept
    {
        // We are waking a lot of things and the wakeCache is completely full. Now we need to start building a list
        WaitEntry *wake = nullptr, *end = nullptr;

        for (WaitEntry* next; e; e = next)
        {
            next = e->next;
            if (e->addr == addr)
            {
                b.remove(e);
                e->next = nullptr;
                // Don't bother with prev pointers
                if (end)
                    end->next = e;
                else
                    wake = e;
                end = e;
                // Even though we're under the lock we use release semantics here to synchronize-with the waking thread
                // possibly outside the lock.
                // Need to set the wait bit since we're still reading/writing to the WaitEntry
                e->wakeup.store(WaitEntry::kWaitBit | WaitEntry::kNotifyBit, std::memory_order_release);
            }
        }

        b.unlock();

        // Wake the entire cache since we know it's full
        auto wakeCacheEnd = (wakeCache + kWakeCacheSize);
        do
        {
            NativeFutex::notify_one(**(wakeCache++));
        } while (wakeCache != wakeCacheEnd);

        for (WaitEntry* next; wake; wake = next)
        {
            next = wake->next;
            // Clear the wait bit so that only the wake bit is set
            wake->wakeup.store(WaitEntry::kNotifyBit, std::memory_order_release);
            NativeFutex::notify_one(wake->wakeup);
        }
    }

public:
    static void notify_n(void* addr, size_t n) noexcept
    {
        if (CARB_UNLIKELY(n == 0))
        {
            return;
        }

        // It is much faster overall to not set kWaitBit and force the woken threads to wait until we're clear of their
        // WaitEntry, so keep a local cache of addresses to wake here that don't require a WaitEntry.
        // Note that we do retain a pointer to the std::atomic_uint32_t *contained in* the WaitEntry and tell the
        // underlying OS system to wake by that address, but this is safe as it does not read that memory. The address
        // is used as a lookup to find any waiters that registered with that address. If the memory was quickly reused
        // for a different wait operation, this will cause a spurious wakeup which is allowed, but this should be
        // exceedingly rare.
        size_t wakeCount = 0;
        std::atomic_uint32_t* wakeCache[kWakeCacheSize];

        WaitBucket& b = WaitBucket::bucket(addr);

        // Read empty state with full fence to avoid locking
        if (b.appearsEmpty())
            return;

        WaitEntry* e = b.lock();
        b.incChangeTracker();
        for (WaitEntry* next; e; e = next)
        {
            next = e->next;
            if (e->addr == addr)
            {
                b.remove(e);
                // Even though we're under the lock we use release semantics here to synchronize-with the waking thread
                // possibly outside the lock.
                wakeCache[wakeCount++] = &e->wakeup;
                e->wakeup.store(WaitEntry::kNotifyBit, std::memory_order_release);

                if (!--n)
                    break;

                if (CARB_UNLIKELY(wakeCount == kWakeCacheSize))
                {
                    // Cache is full => transition to a list for the rest
                    notify_n_slow(addr, n, b, next, wakeCache);
                    return;
                }
            }
        }
        b.unlock();

        auto start = wakeCache;
        auto const end = wakeCache + wakeCount;
        while (start != end)
            NativeFutex::notify_one(**(start++));
    }

    static void notify_all(void* addr) noexcept
    {
        // It is much faster overall to not set kWaitBit and force the woken threads to wait until we're clear of their
        // WaitEntry, so keep a local cache of addresses to wake here that don't require a WaitEntry.
        // Note that we do retain a pointer to the std::atomic_uint32_t *contained in* the WaitEntry and tell the
        // underlying OS system to wake by that address, but this is safe as it does not read that memory. The address
        // is used as a lookup to find any waiters that registered with that address. If the memory was quickly reused
        // for a different wait operation, this will cause a spurious wakeup which is allowed, but this should be
        // exceedingly rare.
        size_t wakeCount = 0;
        std::atomic_uint32_t* wakeCache[kWakeCacheSize];

        WaitBucket& b = WaitBucket::bucket(addr);

        // Read empty state with full fence to avoid locking
        if (b.appearsEmpty())
            return;

        WaitEntry* e = b.lock();
        b.incChangeTracker();
        for (WaitEntry* next; e; e = next)
        {
            next = e->next;
            if (e->addr == addr)
            {
                b.remove(e);
                // Even though we're under the lock we use release semantics here to synchronize-with the waking thread
                // possibly outside the lock.
                wakeCache[wakeCount++] = &e->wakeup;
                e->wakeup.store(WaitEntry::kNotifyBit, std::memory_order_release);
                if (CARB_UNLIKELY(wakeCount == kWakeCacheSize))
                {
                    // Full => transition to a list for the rest of the items
                    notify_all_slow(addr, b, next, wakeCache);
                    return;
                }
            }
        }
        b.unlock();

        auto start = wakeCache;
        const auto end = wakeCache + wakeCount;
        while (start != end)
            NativeFutex::notify_one(**(start++));
    }

}; // struct ParkingLot

// Futex types that must use the ParkingLot
template <class T, size_t S = sizeof(T)>
class Futex
{
    static_assert(S == 1 || S == 2 || S == 4 || S == 8, "Unsupported size");

public:
    using AtomicType = typename std::atomic<T>;
    using Type = T;
    static void wait(const AtomicType& val, T compare) noexcept
    {
        ParkingLot::wait(val, compare);
    }
    template <class Rep, class Period>
    static bool wait_for(const AtomicType& val, T compare, std::chrono::duration<Rep, Period> duration)
    {
        return ParkingLot::wait_for(val, compare, duration);
    }
    template <class Clock, class Duration>
    static bool wait_until(const AtomicType& val, T compare, std::chrono::time_point<Clock, Duration> time_point)
    {
        return ParkingLot::wait_until(val, compare, time_point);
    }
    static void notify_one(AtomicType& val) noexcept
    {
        ParkingLot::notify_one(std::addressof(val));
    }
    static void notify_n(AtomicType& val, size_t count) noexcept
    {
        ParkingLot::notify_n(std::addressof(val), count);
    }
    static void notify_all(AtomicType& val) noexcept
    {
        ParkingLot::notify_all(std::addressof(val));
    }
};

#endif

} // namespace detail
} // namespace thread
} // namespace carb
