concurrent_queue.h

/**
 * Basic concurrent queue with several synchronization helpers, 
 * Copyright (c) 2017-2018, 2023, Jorma Rebane
 * Distributed under MIT Software License
 */
#pragma once
#include "config.h"
#include "condition_variable.h" // rpp::condition_variable
#include "mutex.h" // rpp::mutex, rpp::spin_lock, rpp::yield
#include "atomic_timepoint.h" // rpp::AtomicTimeSource
#include <vector>
#include <type_traits> // std::is_trivially_destructible_v
#include <optional> // std::optional
#include <atomic> // std::atomic_bool
#include <cstring> // memmove
#include <malloc.h> // malloc, free

#if RPP_HAS_CXX20
#  include "future_types.h" // RPP_HAS_COROUTINES, rpp::coro_handle
#  if RPP_HAS_COROUTINES
#    include "delegate.h" // rpp::delegate (for coroutine awaiter)
#  endif
#endif

namespace rpp
{
    #if RPP_HAS_COROUTINES
        // forward declaration: avoids circular include with thread_pool.h
        void parallel_task_detached(rpp::delegate<void()>&& genericTask) noexcept;
    #endif

    /**
     * Provides a simple thread-safe queue with several synchronization helpers
     * as a simple way to write thread-safe code between multiple worker threads.
     *
     * @note This is not optimized for speed, but has acceptable performance
     *       and due to its simplicity it won't randomly deadlock on you.
     *
     * C++20 coroutine support (include `<rpp/future_types.h>` and `<rpp/thread_pool.h>`):
     *   - `co_await queue.await(timeout)` — suspends until items are available
     *   - `co_await queue.await_pop(item, timeout)` — suspends until an item is popped (returns bool)
     *   - `co_await queue.await_pop(timeout)` — suspends until an item is popped (returns optional<T>)
     */
    template<class T>
    class concurrent_queue
    {
        // for faster performance the Head and Tail are always within the linear range
        // of [ItemsStart, ItemsEnd) and the items are always contiguous in memory.
        T* Head = nullptr;
        T* Tail = nullptr;
        T* ItemsStart = nullptr; // storage for the items
        T* ItemsEnd = nullptr;

    public:
        using duration = rpp::Duration;
        using time_point = rpp::TimePoint;

    private:
        using mutex_t = rpp::mutex;
        using lock_t = std::unique_lock<mutex_t>;
        mutable mutex_t Mutex;
        mutable rpp::condition_variable Waiter;
        // special state flag for all waiters to immediately exit the queue
        std::atomic_bool Destroying = false;
        // state flag for pop_atomic_start()/end()
        std::atomic_bool SlowPopInProgress = false;

    public:
        concurrent_queue() = default;
        ~concurrent_queue() noexcept
        {
            clear_and_destroy();
        }

        concurrent_queue(const concurrent_queue&) = delete;
        concurrent_queue& operator=(const concurrent_queue&) = delete;

        concurrent_queue(concurrent_queue&& q) noexcept
        {
            // safely swap the states from q to default empty state
            std::scoped_lock dual_lock { mutex(), q.mutex() };
            std::swap(Head, q.Head);
            std::swap(Tail, q.Tail);
            std::swap(ItemsStart, q.ItemsStart);
            std::swap(ItemsEnd, q.ItemsEnd);
            q.Destroying.store(false, std::memory_order::release);
            q.SlowPopInProgress.store(false, std::memory_order::release);
            // notify only the old queue
            q.notify_all_unlocked();
        }

        // unsafely swaps two atomic booleans
        static void swap_atomics_unsafe(std::atomic_bool& a, std::atomic_bool& b) noexcept
        {
            bool temp = a.load(std::memory_order_acquire);
            a.store(b.load(std::memory_order_acquire), std::memory_order::release);
            b.store(temp, std::memory_order::release);
        }

        concurrent_queue& operator=(concurrent_queue&& q) noexcept
        {
            // safely swap the states of both queues
            std::scoped_lock dual_lock { mutex(), q.mutex() };
            std::swap(Head, q.Head);
            std::swap(Tail, q.Tail);
            std::swap(ItemsStart, q.ItemsStart);
            std::swap(ItemsEnd, q.ItemsEnd);
            swap_atomics_unsafe(Destroying, q.Destroying);
            swap_atomics_unsafe(SlowPopInProgress, q.SlowPopInProgress);
            // notify all waiters for both queues
            q.notify_all_unlocked();
            notify_all_unlocked();
            return *this;
        }

        /** @returns the internal mutex for this queue */
        [[nodiscard]] FINLINE mutex_t& mutex() const noexcept
        {
            return Mutex;
        }

        [[nodiscard]] FINLINE lock_t spin_lock() const noexcept
        {
            return rpp::spin_lock(Mutex);
        }

        /** @returns TRUE if this queue is empty */
        [[nodiscard]] FINLINE bool empty() const noexcept
        {
            lock_t lock_guard = spin_lock();
            return Head == Tail;
        }

        /**
         * @returns Capacity of the queue (safe)
         */
        [[nodiscard]] FINLINE size_t capacity() const noexcept
        {
            lock_t lock_guard = spin_lock();
            return capacity_unlocked();
        }

        /**
         * @returns Synchronized current size of the queue,
         *          however using this value is not atomic
         */
        [[nodiscard]] FINLINE size_t size() const noexcept
        {
            lock_t lock_guard = spin_lock();
            return size_unlocked();
        }

        /**
         * @brief Notifies all waiters that the queue has changed
         */
        void notify() noexcept
        {
            lock_t lock_guard = spin_lock();
            notify_all_unlocked();
        }

        /**
         * @brief Notifies only a single waiter that the queue has changed
         */
        void notify_one() noexcept
        {
            lock_t lock_guard = spin_lock();
            notify_one_unlocked();
        }

        /**
         * @brief Thread-safely modify wait condition in the changeWaitFlags callback
         *        and then notifies all waiters.
         *        If any other threads have locked the mutex, then modifying condition
         *        flags could be illegal. This allows synchronizing to the queue state
         *        to ensure all other threads are idle and have released the mutex.
         * @details This is meant to be used with wait_pop() callback overload
         *          where the wait_pop() is checking for a cancellation condition.
         *          This allows for safely setting that condition from another thread
         *          and notify all waiters.
         * @code
         *      queue.notify([this]{ this->atomic_cancelled = true; })
         * @endcode
         */
        template<class ChangeWaitFlags>
        void notify(const ChangeWaitFlags& changeWaitFlags) noexcept(noexcept(changeWaitFlags()))
        {
            lock_t lock_guard = spin_lock();
            changeWaitFlags();
            notify_all_unlocked();
        }

        /**
         * @brief Thread-safely clears the entire queue and notifies all waiters.
         */
        void clear() noexcept
        {
            lock_t lock_guard = spin_lock();
            clear_unlocked(); // destroy all elements
            notify_all_unlocked(); // notify all waiters that the queue was emptied
        }

        /**
         * @brief Thread-safely clears & destroys the entire queue and notifies all waiters.
         * The queue will enter destroy phase and all timed waiters will quit prematurely.
         * All pushes to the queue will fail silently after this.
         * @warning This should only be done prior to destruction, since the flag cannot be cleared.
         * @warning All waits will turn into hot loops after this !
         */
        void clear_and_destroy() noexcept
        {
            // signal all waiters to exit by setting Destroying flag and notifying all
            {
                std::lock_guard lock1 { mutex() };
                Destroying.store(true, std::memory_order_release);
                notify_all_unlocked();
            } // release lock so that blocking threads can wake up and see the Destroying flag

            // lock again before destroying items -- hopefully all waiters have exited by now
            std::lock_guard lock2 { mutex() };
            clear_unlocked(/*force_destroy*/true);
        }

        /**
         * @brief Pre-allocates memory for the queue
         * @returns TRUE if pre-allocation succeeded, FALSE if allocation failed
         */
        bool reserve(size_t newCapacity) noexcept
        {
            lock_t lock_guard = spin_lock();
            if (newCapacity > capacity_unlocked())
                return grow_to(newCapacity);
            return true;
        }

        /**
         * @returns An atomic copy of the entire queue items
         */
        [[nodiscard]] std::vector<T> atomic_copy() const
        {
            lock_t lock_guard = spin_lock();
            return std::vector<T>{Head, Tail}; // can throw
        }

        /** @brief A no-op proxy for iterating an already locked queue */
        struct iterator_proxy
        {
            concurrent_queue* Queue;
            [[nodiscard]] FINLINE T* begin() noexcept { return Queue->Head; }
            [[nodiscard]] FINLINE T* end()   noexcept { return Queue->Tail; }
            [[nodiscard]] FINLINE const T* begin() const noexcept { return Queue->Head; }
            [[nodiscard]] FINLINE const T* end()   const noexcept { return Queue->Tail; }
        };

        /** @brief A safe iterator for the queue - which holds a lock until iteration is complete */
        struct iterator_lock : public iterator_proxy
        {
            lock_t Lock;
            [[nodiscard]] FINLINE lock_t& lock() noexcept { return Lock; }
            [[nodiscard]] FINLINE T* erase(T* it) noexcept { return this->Queue->erase(Lock, it); }
        };

        /** @brief creates a safe iterator for the queue */
        FINLINE       iterator_lock iterator()       noexcept { return { {this}, spin_lock() }; }
        FINLINE const iterator_lock iterator() const noexcept { return { {this}, spin_lock() }; }

        /** @brief Creates a safe iterator for the queue, using a previously acquired lock */
        FINLINE       iterator_proxy iterator(lock_t& lock)       noexcept { if (!lock.owns_lock()) { lock.lock(); } return { this }; }
        FINLINE const iterator_proxy iterator(lock_t& lock) const noexcept { if (!lock.owns_lock()) { lock.lock(); } return { this }; }

        /**
         * @brief Erases an item from the queue and shifts all items after it
         * @param lock Mandatory queue.mutex() lock to ensure iteration is thread-safe
         * @param it The item to erase
         * @returns Iterator position of the next item after the erased item
         *          (behavior identical to std::vector::erase())
         */
        T* erase(lock_t& lock_guard, T* it) noexcept
        {
            // if the lock is not held, or the iterator is out of bounds, then return the end
            if (!lock_guard.owns_lock() || it < Head || it >= Tail)
                return Tail;

            if constexpr (!std::is_trivially_destructible_v<T>)
                it->~T();

            // shift all items after the erased item
            for (T* i = it + 1; i < Tail; ++i)
                *(i - 1) = std::move(*i);

            --Tail;
            return it;
        }

        /**
         * @brief Attempts to pop all pending items from the queue without waiting
         */
        [[nodiscard]] bool try_pop_all(std::vector<T>& out) noexcept
        {
            auto lock_guard = spin_lock();
            if (T* head = Head, *tail = Tail; head != tail)
            {
                try { out.assign(head, tail); }
                catch (...) { }
                clear_unlocked();
                return true;
            }
            return false;
        }

        /**
         * @brief Thread-safely moves an item into the queue and notifies one waiter
         */
        void push(T&& item) noexcept(std::is_nothrow_move_constructible_v<T>)
        {
            lock_t lock_guard = spin_lock();
            push_unlocked(lock_guard, std::move(item));
            notify_all_unlocked();
        }

        /**
         * @brief Thread-safely copies an item into the queue and notifies one waiter
         */
        void push(const T& item) noexcept(std::is_nothrow_copy_constructible_v<T>)
        {
            lock_t lock_guard = spin_lock();
            push_unlocked(lock_guard, item);
            notify_all_unlocked();
        }

        /**
         * @brief Thread-safely moves an item into the queue without notifying waiters
         */
        void push_no_notify(T&& item) noexcept(std::is_nothrow_move_constructible_v<T>)
        {
            lock_t lock_guard = spin_lock();
            push_unlocked(lock_guard, std::move(item));
        }

        /**
         * @brief Thread-safely copies an item into the queue without notifying waiters
         */
        void push_no_notify(const T& item) noexcept(std::is_nothrow_copy_constructible_v<T>)
        {
            lock_t lock_guard = spin_lock();
            push_unlocked(lock_guard, item);
        }

        /**
         * @returns Thread-safely pops an item from the queue and notifies other waiters
         * @throws runtime_error if the queue is empty
         */
        [[nodiscard]] T pop()
        {
            lock_t lock_guard = spin_lock();
            if (Head == Tail)
                throw std::runtime_error{"concurrent_queue<T>::pop(): Queue was empty!"};

            T item;
            pop_unlocked(item);
            return item;
        }

        /**
         * @brief Attempts to pop an item from the queue without waiting
         * @note try_pop() is excellent for polling scenarios if you don't
         *       want to wait for an item, but just check if any work could be done, 
         *       otherwise just continue
         * @returns TRUE if an item was popped, FALSE if the queue was empty
         */ 
        [[nodiscard]] bool try_pop(T& outItem) noexcept
        {
            lock_t lock_guard = spin_lock();
            if (Head == Tail)
                return false;
            pop_unlocked(outItem);
            return true;
        }

        /**
         * @brief Attempts to peek an item without popping it. This will copy the item!
         * @returns TRUE if an item was available and was copied to outItem
         */
        [[nodiscard]] bool peek(T& outItem) const noexcept
        {
            lock_t lock_guard = spin_lock();
            if (Head == Tail)
                return false;
            outItem = *Head;
            return true;
        }

        /**
         * @brief Moves the front item without popping it. This will
         *        allow for proper atomic operations where the item is not removed
         *        until it has been fully processed.
         * @warning This is only valid for single-consumer scenarios !!!
         * @code
         * T item;
         * if (queue.pop_atomic_start(item))
         * {
         *     channel.send(item); // process the item (slow)
         *     queue.pop_atomic_end(); // remove the processed item from the queue
         * }
         * @endcode
         * @param outItem [out] The front item. Only valid if return value is TRUE
         * @returns true if an item was moved to outItem
         */
        [[nodiscard]] bool pop_atomic_start(T& outItem)
        {
            lock_t lock_guard = spin_lock();
            if (Head == Tail)
                return false;
            bool wasPopInProgress = SlowPopInProgress.exchange(true, std::memory_order_acq_rel);
            if (wasPopInProgress)
                throw std::runtime_error{"concurrent_queue<T>::pop_atomic_start(): Another atomic pop was already in progress!"};
            outItem = std::move(*Head);
            return true;
        }

        /**
         * @see pop_atomic_start()
         * @warning This is only valid for single-consumer scenarios !!!
         */
        void pop_atomic_end()
        {
            lock_t lock_guard = spin_lock();
            bool wasPopInProgress = SlowPopInProgress.exchange(false, std::memory_order_acq_rel);
            if (!wasPopInProgress)
            {
                throw std::runtime_error{"concurrent_queue<T>::pop_atomic_end(): No atomic pop was in progress!"};
            }
            if (Head != Tail)
            {
                pop_unlocked();
            }
        }

        /**
         * @brief Atomically pops and processes an item within a callback.
         *        The item is removed only if the callback returns TRUE.
         * @see pop_atomic_start() and pop_atomic_end()
         * @param callback [in] The callback to process the item, taking 1 argument which Item &&
         * @returns TRUE if an item was popped and processed
         */
        template<class Lambda> FINLINE bool pop_atomic(const Lambda& callback) noexcept
        {
            T item;
            if (pop_atomic_start(item))
            {
                callback(std::move(item));
                pop_atomic_end();
                return true;
            }
            return false;
        }

        /**
         * @brief Attempts to wait until an item is available.
         * @note Can be interrupted by notify_one()/all(), in which case it can return false.
         * @returns TRUE if an item is available to peek or pop. FALSE if notified and no items.
         */
        [[nodiscard]] bool wait_available() const noexcept
        {
            lock_t lock_guard = spin_lock();
            return wait_notify(lock_guard);
        }

        /**
         * @brief Attempts to wait until an item is available.
         * @note A timed wait cannot be notified to cancel, unless queue is destroyed.
         * @param timeout Maximum time to wait before returning FALSE
         * @param time_source [Optional] Time source for virtual or simulation accelerated time. If not provided, uses real time.
         * @returns TRUE if an item is available to peek or pop. FALSE if notified and no items.
         */
        [[nodiscard]] bool wait_available(rpp::Duration timeout,
                                          const rpp::AtomicTimeSource* time_source = nullptr) const noexcept
        {
            lock_t lock_guard = spin_lock();
            return wait_notify_for(lock_guard, timeout, time_source);
        }

        /**
         * Waits until an item is available, or until this queue is NOTIFIED.
         * @note This is a convenience method for wait_pop() with no timeout
         *       and is most convenient for producer/consumer threads.
         *       @see test_concurrent_queue.cpp TestCase(basic_producer_consumer) for example usage.
         * @returns The popped item, or std::nullopt if the queue had no items when NOTIFIED
         */
        [[nodiscard]] std::optional<T> wait_pop() noexcept
        {
            lock_t lock_guard = spin_lock();
            if (!wait_notify(lock_guard))
                return std::nullopt;
            T result;
            pop_unlocked(result);
            return result;
        }

        /**
         * Waits until an item is available, or until this queue is NOTIFIED.
         * @note This variant does not use a timeout and thus has better performance.
         *       However, this means the queue needs to be notified to wake up
         */
        [[nodiscard]]
        bool wait_pop(T& outItem) noexcept
        {
            lock_t lock_guard = spin_lock();
            if (!wait_notify(lock_guard))
                return false;
            pop_unlocked(outItem);
            return true;
        }

        /**
         * Waits up to @param timeout duration until an item is ready to be popped.
         * For waiting without timeout @see wait_pop().
         * 
         * @note A timed wait cannot be notified to cancel, unless queue is destroyed.
         * @note This is best used for cases where you want to wait up to a certain time
         *       before giving up. This may return false before the timeout due to spurious wakeups.
         *       Useful for synchronization tasks that have a time limit.
         * 
         * @param out_item [out] The popped item. Only valid if return value is TRUE
         * @param time_source [Optional] Time source for virtual or simulation accelerated time. If not provided, uses real time.
         * @param timeout [required] Time to wait before returning FALSE
         * @return TRUE if an item was popped, FALSE if timed out.
         * @code
         *   string item;
         *   if (queue.wait_pop(item, rpp::Duration::from_millis(100)))
         *   {
         *       // item is valid
         *   }
         *   // else: timeout was reached
         * @endcode
         */
        [[nodiscard]]
        bool wait_pop(T& out_item, rpp::Duration timeout,
                      const rpp::AtomicTimeSource* time_source = nullptr) noexcept
        {
            lock_t lock_guard = spin_lock();
            if (!wait_notify_for(lock_guard, timeout, time_source))
                return false;
            pop_unlocked(out_item);
            return true;
        }

        /**
         * Waits up to @param timeout duration until an item is ready and peeks the value
         * without popping it.
         * 
         * @note A timed wait cannot be notified to cancel, unless queue is destroyed.
         * @param out_item [out] The popped item. Only valid if return value is TRUE
         * @param time_source [Optional] Time source for virtual or simulation accelerated time. If not provided, uses real time.
         * @return TRUE if an item was peeked successfully
         */
        [[nodiscard]]
        bool wait_peek(T& out_item, rpp::Duration timeout,
                       const rpp::AtomicTimeSource* time_source = nullptr) const noexcept
        {
            lock_t lock_guard = spin_lock();
            if (!wait_notify_for(lock_guard, timeout, time_source))
                return false;
            out_item = *Head; // copy (may throw)
            return true;
        }

        /**
         * Only returns items if `time_now(time_source) < until`. This is excellent for message handling
         * loops that have an absolute time limit for processing messages.
         * 
         * @note A timed wait cannot be notified to cancel, unless queue is destroyed.
         * @note This is best used for cases where you want to wait up to a certain time
         * @param out_item [out] The popped item. Only valid if return value is TRUE
         * @param until [required] Timepoint to wait until before returning FALSE
         * @param time_source [Optional] Time source for virtual or simulation accelerated time. If not provided, uses real time.
         * @return TRUE if an item was popped, FALSE if timed out.
         * @code
         *   auto until = time_now(time_source) + time_limit;
         *   string item;
         *   // process messages until time limit is reached
         *   while (queue.wait_pop_until(item, until))
         *   {
         *       // item is valid
         *   }
         * @endcode
         */
        [[nodiscard]]
        bool wait_pop_until(T& out_item, rpp::TimePoint until,
                            const rpp::AtomicTimeSource* time_source = nullptr) noexcept
        {
            auto now = time_now(time_source);

            // if we're already past the time limit, then don't check anything
            // this ensures while() loops don't get stuck processing items endlessly
            if (now > until)
                return false;

            lock_t lock_guard = spin_lock();
            if (!wait_notify_until(lock_guard, until, time_source))
                return false;
            pop_unlocked(out_item);
            return true;
        }

        /**
         * Waits up to @param timeout duration until an item is ready to be popped.
         * The cancel_cond is used to terminate the wait.
         * 
         * @note A timed wait cannot be notified to cancel, unless queue is destroyed. cancel_cond() is checked instead.
         * @note This is a wrapper around wait_pop_interval, with the cancellation check
         *       interval set to 1/10th of the timeout.
         * 
         * @param outItem [out] The popped item. Only valid if return value is TRUE
         * @param timeout [required] Maximum time to wait before returning FALSE
         * @param cancel_cond Cancellation condition, CANCEL if cancel_cond()==true
         * @param time_source [Optional] Time source for virtual or simulation accelerated time. If not provided, uses real time.
         * @return TRUE if an item was popped,
         *         FALSE if no item popped due to: timeout or cancellation.
         * @code
         *   string item;
         *   auto timeout = rpp::Duration::from_millis(100);
         *   if (queue.wait_pop(item, timeout, [this]{ return this->cancelled || this->finished; }))
         *   {
         *       // item is valid
         *   }
         * @endcode
         */
        template<IsPredicate WaitUntil>
        [[nodiscard]]
        bool wait_pop(T& outItem, rpp::Duration timeout, const WaitUntil& cancel_cond,
                      const rpp::AtomicTimeSource* time_source = nullptr)
            noexcept(noexcept(cancel_cond()))
        {
            rpp::Duration interval = timeout / 10;
            return wait_pop_interval<WaitUntil>(outItem, timeout, interval, cancel_cond, time_source);
        }

        /**
         * Waits until an item is ready to be popped.
         * The @param cancel_cond is used to terminate the wait.
         * And @param interval defines how often the cancel_cond is checked.
         * 
         * @note This is a superior alternative to wait_pop(), because the cancel_cond
         *       is checked multiple times, instead of only between waits if someone notifies.
         * 
         * @param outItem [out] The popped item. Only valid if return value is TRUE
         * @param timeout Total timeout for the entire wait loop, granularity
         *                is defined by @param interval
         * @param interval Interval for checking the cancellation condition
         *  NOTE: this is just a hint and there is no guarantee to make this accurate without
         *        affecting system performance. A 1ms interval could be anywhere from 1ms to 15ms.
         * @param cancel_cond Cancellation condition, CANCEL if cancel_cond()==true
         * @param time_source [Optional] Time source for virtual or simulation accelerated time. If not provided, uses real time.
         * @return TRUE if an item was popped,
         *         FALSE if no item popped due to: timeout or cancellation
         * @code
         *   string item;
         *   auto interval = rpp::Duration::from_millis(100);
         *   if (queue.wait_pop_interval(item, timeout, interval, [this]{ return this->cancelled || this->finished; })
         *   {
         *       // item is valid
         *   }
         * @endcode
         */
        template<IsPredicate WaitUntil>
        [[nodiscard]]
        bool wait_pop_interval(T& outItem, rpp::Duration timeout, rpp::Duration interval,
                               const WaitUntil& cancel_cond,
                               const rpp::AtomicTimeSource* time_source = nullptr)
            noexcept(noexcept(cancel_cond()))
        {
            lock_t lock_guard = spin_lock();
            if (is_destroying()) return false; // give up immediately
            if (Head == Tail)
            {
                rpp::Duration remaining = timeout;
                rpp::TimePoint prevTime = time_now(time_source);
                constexpr rpp::Duration zero = rpp::Duration{0};
                // When using a virtual time source, poll in short real-time intervals
                // so that time warping (fast-forward) causes the wait to exit promptly.
                constexpr rpp::Duration warp_poll = rpp::millis(1);
                do
                {
                    if (cancel_cond())
                        return false;
                    const rpp::Duration& limit = (time_source ? warp_poll : interval);
                    // make sure we don't suspend past the final waiting point
                    rpp::Duration real_wait = (remaining < limit ? remaining : limit);

                    // use wait_for(duration) instead of wait_until(absolute) to avoid
                    // NTP time jumps in the condition variable
                    (void)Waiter.wait_for(lock_guard, real_wait); // noexcept

                    if (is_destroying()) return false; // give up immediately
                    if (Head != Tail) break; // got data

                    rpp::TimePoint now = time_now(time_source);
                    remaining -= (now - prevTime);
                    if (remaining <= zero)
                        break; // timed out
                    prevTime = now;
                } while (Head == Tail);

                if (Head == Tail)
                    return false;
            }
            pop_unlocked(outItem);
            return true;
        }

    private:
        rpp::TimePoint time_now(const rpp::AtomicTimeSource* time_source) const noexcept
        {
            if (time_source)
                return time_source->time_now();
            else // use monotonic clock to avoid NTP time jumps affecting wait timeouts
                return rpp::TimePoint::monotonic_now();
        }

        [[nodiscard]] FINLINE bool is_destroying() const noexcept { return Destroying.load(std::memory_order_acquire); }

        // waits until this thread is notified and returns true if there's an item
        bool wait_notify(lock_t& lock_guard) const noexcept
        {
            if (is_destroying()) {
                rpp::yield(); // need to yield here to avoid burning a hole into the CPU
                return false; // give up immediately
            }
            if (Head != Tail) return true; // got an item

            // Wait only once, otherwise a lot of code could deadlock forever waiting for items
            // that will never arrive, because Producer has exited without destroying the queue.
            // It's an important behavioral guarantee, that a notify_one()/all()
            // from a Producer thread can wake up and cancel waiting Consumer thread(s),
            // for example when Producer thread is about to exit.
            // Since we always return false, there is no requirement for a valid item.
            (void)Waiter.wait(lock_guard); // noexcept

            if (is_destroying()) return false; // give up immediately, always signal false
            return Head != Tail;
        }
        // wait_notify with a timeout, returns true if there is an item
        bool wait_notify_for(lock_t& lock_guard, const rpp::Duration& timeout,
                             const rpp::AtomicTimeSource* time_source) const noexcept
        {
            if (is_destroying()) {
                rpp::yield(); // need to yield here to avoid burning a hole into the CPU
                return false; // give up immediately
            }
            if (Head != Tail) return true; // wait is not needed
            if (timeout <= rpp::Duration::zero()) {
                // NOTE: this makes CI fail, because a yield() on a busy CI machine can take more than 1ms
                //rpp::yield(); // need to yield here to avoid burning a hole into the CPU
                return false; // zero timeout, don't enter wait loop
            }
            auto now = time_now(time_source);
            auto end = now + timeout;
            // When using a virtual time source, we must use real system time for
            // the CV wait, and poll in short intervals so that time warping
            // (fast-forward) causes the wait to exit promptly.
            constexpr rpp::Duration warp_poll = rpp::millis(1);
            do {
                rpp::Duration remaining = end - now;
                rpp::Duration real_wait;
                if (time_source) real_wait = (remaining < warp_poll ? remaining : warp_poll);
                else             real_wait = remaining;

                // use wait_for(duration) instead of wait_until(absolute) to avoid
                // NTP time jumps in the condition variable
                (void)Waiter.wait_for(lock_guard, real_wait); // noexcept

                if (is_destroying()) return false; // give up immediately
                if (Head != Tail) return true; // got an item
                now = time_now(time_source);
            } while (now < end); // handle spurious wakeups
            return false;
        }
        // PRECONDITION: until.is_valid()
        bool wait_notify_until(lock_t& lock_guard, const rpp::TimePoint& until,
                               const rpp::AtomicTimeSource* time_source) const noexcept
        {
            if (is_destroying()) {
                rpp::yield(); // need to yield here to avoid burning a hole into the CPU
                return false; // give up immediately
            }
            if (Head != Tail) return true; // wait is not needed

            // When using a virtual time source, poll in short real-time intervals
            // so that time warping (fast-forward) causes the wait to exit promptly.
            constexpr rpp::Duration warp_poll = rpp::millis(1);
            rpp::TimePoint now = time_now(time_source);
            do {
                rpp::Duration remaining = until - now;
                rpp::Duration real_wait;
                if (time_source) { real_wait = remaining < warp_poll ? remaining : warp_poll; }
                else             { real_wait = remaining; }

                // use wait_for(duration) instead of wait_until(absolute) to avoid
                // NTP time jumps in the condition variable
                (void)Waiter.wait_for(lock_guard, real_wait); // noexcept

                if (is_destroying()) return false; // give up immediately
                if (Head != Tail) return true; // got an item
                now = time_now(time_source);
            } while (now < until); // handle spurious wakeups
            return false;
        }
        void notify_one_unlocked() noexcept
        {
            Waiter.notify_one();
        }
        void notify_all_unlocked() noexcept
        {
            Waiter.notify_all();
        }
        void push_unlocked(lock_t& lock_guard, T&& item) noexcept(std::is_nothrow_move_constructible_v<T>)
        {
            if (is_destroying()) return; // give up immediately
            if (Tail == ItemsEnd)
            {
                if (!ensure_capacity(lock_guard))
                    return; // queue is being destroyed/OOM, abandon ship
            }

            if constexpr (std::is_trivially_move_assignable_v<T>)
                *Tail++ = std::move(item);
            else
                new (Tail++) T{ std::move(item) };
        }
        void push_unlocked(lock_t& lock_guard, const T& item) noexcept(std::is_nothrow_copy_constructible_v<T>)
        {
            if (is_destroying()) return; // give up immediately
            if (Tail == ItemsEnd)
            {
                if (!ensure_capacity(lock_guard))
                    return; // queue is being destroyed/OOM, abandon ship
            }

            if constexpr (std::is_trivially_copy_assignable_v<T>)
                *Tail++ = item;
            else
                new (Tail++) T{ item };
        }
        void pop_unlocked(T& outItem) noexcept
        {
            T* head = Head++;
            outItem = std::move(*head);
            if constexpr (!std::is_trivially_destructible_v<T>)
                head->~T();

            if (Head == Tail) // clear the queue if Head catches the Tail
            {
                clear_unlocked();
            }
        }
        void pop_unlocked() noexcept
        {
            T* head = Head++;
            if constexpr (!std::is_trivially_destructible_v<T>)
                head->~T();

            if (Head == Tail) // clear the queue if Head catches the Tail
            {
                clear_unlocked();
            }
        }
        bool ensure_capacity(lock_t& lock_guard) noexcept
        {
            const size_t oldCap = capacity_unlocked();
            // we have enough capacity, just shift the items to the front
            if (oldCap > 0 && size_t(Head - ItemsStart) >= (oldCap / 2))
            {
                // unshift elements to the front of the queue
                T* newStart = ItemsStart;
                T* newTail = move_items(Head, Tail, newStart);
                Head = newStart;
                Tail = newTail;
            }
            else // grow
            {
                size_t growBy = oldCap ? oldCap : 32;
                if (growBy > (16*1024)) growBy = 16*1024; // put a hard limit to exponential growth
                const size_t newCap = oldCap + growBy;

                // if alloc fails (OOM), we can't push new items,
                // consumer should consume items to drain the queue
                if (!grow_to(newCap))
                {
                    for (int i = 0; i < 100; ++i) // wait until there is more space
                    {
                        if (is_destroying()) return false; // give up immediately
                        size_t available = capacity_unlocked() - size_unlocked();
                        if (available == 0)
                        {
                            // notify any waiters to pick up some slack
                            notify_one_unlocked();
                            rpp::unlock_guard unlocker { lock_guard };
                            rpp::sleep_ms(5); // yield heavily
                        }
                        else // available > 0
                        {
                            return true;
                        }
                    }
                    // all attempts timed out
                    return false;
                }
            }
            return true;
        }
        bool grow_to(size_t newCap) noexcept
        {
            T* oldStart = ItemsStart;
            T* newStart = (T*)malloc(newCap * sizeof(T));
            if (!newStart)
                return false;
            T* newTail = move_items(Head, Tail, newStart);
            Head = newStart;
            Tail = newTail;
            ItemsStart = newStart;
            ItemsEnd = newStart + newCap;
            free(oldStart);
            return true;
        }
        static T* move_items(T* oldHead, T* oldTail, T* newStart) noexcept
        {
            if constexpr (std::is_trivially_move_assignable_v<T>)
            {
                size_t count = (oldTail - oldHead);
                if (count)
                    ::memmove(newStart, oldHead, count * sizeof(T));
                return newStart + count;
            }
            else
            {
                T* newTail = newStart;
                for (; oldHead != oldTail; ++newTail, ++oldHead)
                {
                    new (newTail) T{ std::move(*oldHead) };
                    if constexpr (!std::is_trivially_destructible_v<T>)
                        oldHead->~T();
                }
                return newTail;
            }
        }
        void clear_unlocked(bool force_destroy = false) noexcept
        {
            for (T* head = Head, *tail = Tail; head != tail; ++head)
                head->~T();

            // if the capacity was huge, then free the entire buffer
            // to avoid massive memory usage for a small queue
            if (force_destroy || capacity_unlocked() > 8192)
            {
                free(ItemsStart);
                ItemsStart = ItemsEnd = nullptr;
            }

            Head = Tail = ItemsStart;
        }
        [[nodiscard]] FINLINE size_t capacity_unlocked() const noexcept
        {
            return (ItemsEnd - ItemsStart);
        }
        [[nodiscard]] FINLINE size_t size_unlocked() const noexcept
        {
            return (Tail - Head);
        }

    public:
#if RPP_HAS_COROUTINES
        /**
         * @brief Awaitable handle for co_await on queue availability.
         * Dispatches a blocking wait to a background thread and resumes the coroutine.
         * @code
         *     bool available = co_await queue.await(rpp::millis(100));
         * @endcode
         */
        struct RPP_CORO_RETURN_TYPE co_await_handle
        {
            concurrent_queue& queue;
            rpp::Duration timeout;
            bool available = false;

            bool await_ready() noexcept
            {
                return (available = !queue.empty());
            }
            void await_suspend(rpp::coro_handle<> cont) noexcept
            {
                rpp::parallel_task_detached(rpp::delegate<void()>{[this, cont]() mutable
                {
                    available = queue.wait_available(timeout);
                    cont.resume();
                }});
            }
            bool await_resume() noexcept { return available; }
        };
        RPP_CORO_WRAPPER co_await_handle await(rpp::Duration timeout) noexcept { return { *this, timeout }; }

        /**
         * @brief Awaitable handle for co_await on queue pop.
         * Dispatches a blocking wait_pop to a background thread and resumes the coroutine.
         * @code
         *     T item;
         *     bool got = co_await queue.await_pop(item, rpp::millis(100));
         * @endcode
         */
        struct RPP_CORO_RETURN_TYPE co_pop_handle
        {
            concurrent_queue& queue;
            T& out;
            rpp::Duration timeout;
            bool success = false;

            bool await_ready() noexcept
            {
                return (success = queue.try_pop(out));
            }
            void await_suspend(rpp::coro_handle<> cont) noexcept
            {
                rpp::parallel_task_detached(rpp::delegate<void()>{[this, cont]() mutable
                {
                    success = queue.wait_pop(out, timeout);
                    cont.resume();
                }});
            }
            bool await_resume() noexcept { return success; }
        };
        RPP_CORO_WRAPPER co_pop_handle await_pop(T& out, rpp::Duration timeout) noexcept { return { *this, out, timeout }; }

        /**
         * @brief Awaitable handle for co_await on queue pop returning std::optional<T>.
         * Dispatches a blocking wait_pop to a background thread and resumes the coroutine.
         * @code
         *     std::optional<T> item = co_await queue.await_pop(rpp::millis(100));
         *     if (item) { use(*item); }
         * @endcode
         */
        struct RPP_CORO_RETURN_TYPE co_pop_optional_handle
        {
            concurrent_queue& queue;
            rpp::Duration timeout;
            std::optional<T> result;

            bool await_ready() noexcept
            {
                T item;
                if (queue.try_pop(item)) { result.emplace(std::move(item)); return true; }
                return false;
            }
            void await_suspend(rpp::coro_handle<> cont) noexcept
            {
                rpp::parallel_task_detached(rpp::delegate<void()>{[this, cont]() mutable
                {
                    T item;
                    if (queue.wait_pop(item, timeout))
                        result.emplace(std::move(item));
                    cont.resume();
                }});
            }
            std::optional<T> await_resume() noexcept { return std::move(result); }
        };
        RPP_CORO_WRAPPER co_pop_optional_handle await_pop(rpp::Duration timeout) noexcept { return { *this, timeout, std::nullopt }; }
#endif
    };
}