queue.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_QUEUE_HPP
#define JOIN_CORE_QUEUE_HPP
 
// libjoin.
#include <join/backoff.hpp>
#include <join/memory.hpp>
#include <join/utils.hpp>
 
// C++.
#include <type_traits>
#include <algorithm>
#include <atomic>
 
// C.
#include <sys/types.h>
#include <cstring>
 
namespace join
{
    struct QueueSync
    {
        static constexpr uint64_t MAGIC = 0x9F7E3B2A8D5C4E1B;
 
        alignas (64) std::atomic_uint64_t _magic;
 
        alignas (64) std::atomic_uint64_t _head;
 
        uint64_t _cachedTail;
 
        alignas (64) std::atomic_uint64_t _tail;
 
        uint64_t _cachedHead;
    };
 
    template <typename Type>
    struct QueueSlotLight
    {
        Type data;
    };
 
    template <typename Type>
    struct QueueSlotFull
    {
        alignas (64) std::atomic_uint64_t _seq;
 
        Type data;
 
        char _padding[(64 - ((sizeof (std::atomic_uint64_t) + sizeof (Type)) % 64)) % 64];
    };
 
    template <typename Type, typename Slot>
    struct QueueSegment
    {
        alignas (64) QueueSync _sync;
 
        Slot _elements[];
    };
 
    // forward declarations.
    template <typename Type, typename Backend>
    struct Spsc;
 
    template <typename SyncPolicy>
    struct needs_seq : std::true_type
    {
    };
 
    template <typename Type, typename Backend>
    struct needs_seq<Spsc<Type, Backend>> : std::false_type
    {
    };
 
    template <typename Type, typename Backend, typename SyncPolicy>
    class BasicQueue
    {
        static_assert (std::is_trivially_copyable<Type>::value, "type must be trivially copyable");
        static_assert (std::is_trivially_destructible<Type>::value, "type must be trivially destructible");
 
    public:
        using ValueType = Type;
        using Slot =
            typename std::conditional<needs_seq<SyncPolicy>::value, QueueSlotFull<Type>, QueueSlotLight<Type>>::type;
        using Segment = QueueSegment<Type, Slot>;
 
        template <typename... Args>
        explicit BasicQueue (uint64_t capacity, Args&&... args)
        : _capacity (nextPow2 (capacity))
        , _mask (_capacity - 1)
        , _elementSize (sizeof (Slot))
        , _totalSize (sizeof (QueueSync) + (_capacity * _elementSize))
        , _backend (_totalSize, std::forward<Args> (args)...)
        , _segment (static_cast<Segment*> (_backend.get ()))
        {
            uint64_t expected = 0;
 
            if (_segment->_sync._magic.compare_exchange_strong (expected, 0xFFFFFFFFFFFFFFFF,
                                                                std::memory_order_acq_rel))
            {
                _segment->_sync._head.store (0, std::memory_order_relaxed);
                _segment->_sync._cachedTail = 0;
                _segment->_sync._tail.store (0, std::memory_order_relaxed);
                _segment->_sync._cachedHead = 0;
 
                initSlots<needs_seq<SyncPolicy>::value> ();
 
                _segment->_sync._magic.store (QueueSync::MAGIC, std::memory_order_release);
            }
            else
            {
                Backoff backoff;
                while (_segment->_sync._magic.load (std::memory_order_acquire) != QueueSync::MAGIC)
                {
                    backoff ();  // LCOV_EXCL_LINE
                }
            }
        }
 
        BasicQueue (const BasicQueue& other) = delete;
 
        BasicQueue& operator= (const BasicQueue& other) = delete;
 
        BasicQueue (BasicQueue&& other) = delete;
 
        BasicQueue& operator= (BasicQueue&& other) = delete;
 
        ~BasicQueue () noexcept = default;
 
        int tryPush (const Type& element) noexcept
        {
            return SyncPolicy::tryPush (_segment, element, _capacity, _mask);
        }
 
        ssize_t tryPush (const Type* elements, size_t size) noexcept
        {
            return SyncPolicy::tryPush (_segment, elements, size, _capacity, _mask);
        }
 
        int push (const Type& element) noexcept
        {
            Backoff backoff;
 
            while (tryPush (element) == -1)
            {
                if (JOIN_UNLIKELY (lastError != Errc::TemporaryError))
                {
                    return -1;  // LCOV_EXCL_LINE
                }
 
                backoff ();
            }
 
            return 0;
        }
 
        int push (const Type* elements, size_t size) noexcept
        {
            Backoff backoff;
            uint64_t pushed = 0;
 
            while (pushed < size)
            {
                ssize_t n = tryPush (elements + pushed, size - pushed);
                if (n == -1)
                {
                    if (JOIN_UNLIKELY (lastError != Errc::TemporaryError))
                    {
                        return -1;
                    }
 
                    backoff ();  // LCOV_EXCL_LINE
                }
                else
                {
                    pushed += static_cast<uint64_t> (n);
                }
            }
 
            return 0;
        }
 
        int tryPop (Type& element) noexcept
        {
            return SyncPolicy::tryPop (_segment, element, _capacity, _mask);
        }
 
        ssize_t tryPop (Type* elements, size_t size) noexcept
        {
            return SyncPolicy::tryPop (_segment, elements, size, _capacity, _mask);
        }
 
        int pop (Type& element) noexcept
        {
            Backoff backoff;
 
            while (tryPop (element) == -1)
            {
                if (JOIN_UNLIKELY (lastError != Errc::TemporaryError))
                {
                    return -1;  // LCOV_EXCL_LINE
                }
 
                backoff ();
            }
 
            return 0;
        }
 
        int pop (Type* elements, size_t size) noexcept
        {
            Backoff backoff;
            uint64_t popped = 0;
 
            while (popped < size)
            {
                ssize_t n = tryPop (elements + popped, size - popped);
                if (n == -1)
                {
                    if (JOIN_UNLIKELY (lastError != Errc::TemporaryError))
                    {
                        return -1;
                    }
 
                    backoff ();  // LCOV_EXCL_LINE
                }
                else
                {
                    popped += static_cast<uint64_t> (n);
                }
            }
 
            return 0;
        }
 
        uint64_t pending () const noexcept
        {
            if (JOIN_UNLIKELY (_segment == nullptr))
            {
                return 0;  // LCOV_EXCL_LINE
            }
            auto head = _segment->_sync._head.load (std::memory_order_acquire);
            auto tail = _segment->_sync._tail.load (std::memory_order_relaxed);
            return head - tail;
        }
 
        uint64_t available () const noexcept
        {
            if (JOIN_UNLIKELY (_segment == nullptr))
            {
                return 0;  // LCOV_EXCL_LINE
            }
            return _capacity - pending ();
        }
 
        bool full () const noexcept
        {
            if (JOIN_UNLIKELY (_segment == nullptr))
            {
                return false;  // LCOV_EXCL_LINE
            }
            return pending () == _capacity;
        }
 
        bool empty () const noexcept
        {
            if (JOIN_UNLIKELY (_segment == nullptr))
            {
                return true;  // LCOV_EXCL_LINE
            }
            return pending () == 0;
        }
 
#ifdef JOIN_HAS_NUMA
        int mbind (int numa) const noexcept
        {
            return _backend.mbind (numa);
        }
#endif
 
        int mlock () const noexcept
        {
            return _backend.mlock ();
        }
 
    private:
        template <bool NeedsSeq, typename std::enable_if<!NeedsSeq>::type* = nullptr>
        void initSlots () noexcept
        {
            // nothing to initialize.
        }
 
        template <bool NeedsSeq, typename std::enable_if<NeedsSeq>::type* = nullptr>
        void initSlots () noexcept
        {
            for (uint64_t i = 0; i < _capacity; ++i)
            {
                _segment->_elements[i]._seq.store (i, std::memory_order_relaxed);
            }
        }
 
        alignas (64) const uint64_t _capacity = 0;
 
        const uint64_t _mask = 0;
 
        const uint64_t _elementSize = 0;
 
        const uint64_t _totalSize = 0;
 
        Backend _backend;
 
        Segment* _segment = nullptr;
    };
 
    template <typename Type, typename Backend>
    struct Spsc
    {
        using Queue = BasicQueue<Type, Backend, Spsc>;
        using Segment = QueueSegment<Type, QueueSlotLight<Type>>;
 
        static int tryPush (Segment* segment, const Type& element, uint64_t capacity, uint64_t mask) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr))
            {
                // LCOV_EXCL_START
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            auto& sync = segment->_sync;
            uint64_t head = sync._head.load (std::memory_order_relaxed);
 
            if (JOIN_UNLIKELY ((head - sync._cachedTail) == capacity))
            {
                sync._cachedTail = sync._tail.load (std::memory_order_acquire);
                if ((head - sync._cachedTail) == capacity)
                {
                    lastError = make_error_code (Errc::TemporaryError);
                    return -1;
                }
            }
 
            segment->_elements[head & mask].data = element;
            sync._head.store (head + 1, std::memory_order_release);
 
            return 0;
        }
 
        static ssize_t tryPush (Segment* segment, const Type* elements, size_t size, uint64_t capacity,
                                uint64_t mask) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr || elements == nullptr || size == 0))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            auto& sync = segment->_sync;
            uint64_t head = sync._head.load (std::memory_order_relaxed);
            uint64_t avail = capacity - (head - sync._cachedTail);
 
            if (JOIN_UNLIKELY (avail == 0))
            {
                sync._cachedTail = sync._tail.load (std::memory_order_acquire);
                avail = capacity - (head - sync._cachedTail);
                if (avail == 0)
                {
                    lastError = make_error_code (Errc::TemporaryError);
                    return -1;
                }
            }
 
            uint64_t toWrite = std::min (static_cast<uint64_t> (size), avail);
            uint64_t offset = head & mask;
            uint64_t first = std::min (toWrite, capacity - offset);
 
            std::memcpy (&segment->_elements[offset], elements, first * sizeof (Type));
 
            if (toWrite > first)
            {
                std::memcpy (&segment->_elements[0], elements + first, (toWrite - first) * sizeof (Type));
            }
 
            sync._head.store (head + toWrite, std::memory_order_release);
 
            return static_cast<ssize_t> (toWrite);
        }
 
        static int tryPop (Segment* segment, Type& element, uint64_t /*capacity*/, uint64_t mask) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr))
            {
                // LCOV_EXCL_START
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            auto& sync = segment->_sync;
            uint64_t tail = sync._tail.load (std::memory_order_relaxed);
 
            if (sync._cachedHead == tail)
            {
                sync._cachedHead = sync._head.load (std::memory_order_acquire);
                if (sync._cachedHead == tail)
                {
                    lastError = make_error_code (Errc::TemporaryError);
                    return -1;
                }
            }
 
            element = segment->_elements[tail & mask].data;
            sync._tail.store (tail + 1, std::memory_order_release);
 
            return 0;
        }
 
        static ssize_t tryPop (Segment* segment, Type* elements, size_t size, uint64_t capacity, uint64_t mask) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr || elements == nullptr || size == 0))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            auto& sync = segment->_sync;
            uint64_t tail = sync._tail.load (std::memory_order_relaxed);
            uint64_t pending = sync._cachedHead - tail;
 
            if (pending == 0)
            {
                sync._cachedHead = sync._head.load (std::memory_order_acquire);
                pending = sync._cachedHead - tail;
                if (pending == 0)
                {
                    lastError = make_error_code (Errc::TemporaryError);
                    return -1;
                }
            }
 
            uint64_t toRead = std::min (static_cast<uint64_t> (size), pending);
            uint64_t offset = tail & mask;
            uint64_t first = std::min (toRead, capacity - offset);
 
            std::memcpy (elements, &segment->_elements[offset], first * sizeof (Type));
 
            if (toRead > first)
            {
                std::memcpy (elements + first, &segment->_elements[0], (toRead - first) * sizeof (Type));
            }
 
            sync._tail.store (tail + toRead, std::memory_order_release);
 
            return static_cast<ssize_t> (toRead);
        }
    };
 
    template <typename Type, typename Backend>
    struct Mpsc
    {
        using Queue = BasicQueue<Type, Backend, Mpsc>;
        using Segment = QueueSegment<Type, QueueSlotFull<Type>>;
 
        static int tryPush (Segment* segment, const Type& element, uint64_t /*capacity*/, uint64_t mask) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr))
            {
                // LCOV_EXCL_START
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            auto& sync = segment->_sync;
            uint64_t head = sync._head.load (std::memory_order_relaxed);
            Backoff backoff;
 
            for (;;)
            {
                auto* slot = &segment->_elements[head & mask];
                uint64_t seq = slot->_seq.load (std::memory_order_acquire);
 
                if (seq == head)
                {
                    if (JOIN_LIKELY (sync._head.compare_exchange_weak (head, head + 1, std::memory_order_acquire,
                                                                       std::memory_order_relaxed)))
                    {
                        slot->data = element;
                        slot->_seq.store (head + 1, std::memory_order_release);
                        return 0;
                    }
                }
                else if (JOIN_UNLIKELY (seq < head))
                {
                    lastError = make_error_code (Errc::TemporaryError);
                    return -1;
                }
                else
                {
                    backoff ();
                    head = sync._head.load (std::memory_order_relaxed);
                }
            }
        }
 
        static ssize_t tryPush (Segment* segment, const Type* elements, size_t size, uint64_t capacity,
                                uint64_t mask) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr || elements == nullptr || size == 0))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            Backoff backoff;
            auto& sync = segment->_sync;
            uint64_t head = sync._head.load (std::memory_order_relaxed);
 
            for (;;)
            {
                uint64_t tail = sync._tail.load (std::memory_order_acquire);
                uint64_t toWrite = std::min (static_cast<uint64_t> (size), capacity - (head - tail));
 
                if (JOIN_UNLIKELY (toWrite == 0))
                {
                    lastError = make_error_code (Errc::TemporaryError);
                    return -1;
                }
 
                if (JOIN_LIKELY (sync._head.compare_exchange_weak (head, head + toWrite, std::memory_order_acquire,
                                                                   std::memory_order_relaxed)))
                {
                    for (uint64_t i = 0; i < toWrite; ++i)
                    {
                        auto* slot = &segment->_elements[(head + i) & mask];
                        slot->data = elements[i];
                        slot->_seq.store (head + i + 1, std::memory_order_release);
                    }
 
                    return static_cast<ssize_t> (toWrite);
                }
 
                backoff ();  // LCOV_EXCL_LINE
            }
        }
 
        static int tryPop (Segment* segment, Type& element, uint64_t capacity, uint64_t mask) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr))
            {
                // LCOV_EXCL_START
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            auto& sync = segment->_sync;
            uint64_t tail = sync._tail.load (std::memory_order_relaxed);
            auto* slot = &segment->_elements[tail & mask];
            uint64_t seq = slot->_seq.load (std::memory_order_acquire);
 
            if (JOIN_UNLIKELY (seq != tail + 1))
            {
                lastError = make_error_code (Errc::TemporaryError);
                return -1;
            }
 
            element = slot->data;
            slot->_seq.store (tail + capacity, std::memory_order_release);
            sync._tail.store (tail + 1, std::memory_order_release);
 
            return 0;
        }
 
        static ssize_t tryPop (Segment* segment, Type* elements, size_t size, uint64_t capacity, uint64_t mask) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr || elements == nullptr || size == 0))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            auto& sync = segment->_sync;
            uint64_t tail = sync._tail.load (std::memory_order_relaxed);
            uint64_t head = sync._head.load (std::memory_order_acquire);
            uint64_t toRead = std::min (static_cast<uint64_t> (size), head - tail);
            uint64_t popped = 0;
 
            for (uint64_t i = 0; i < toRead; ++i)
            {
                auto* slot = &segment->_elements[(tail + i) & mask];
 
                if (JOIN_UNLIKELY (slot->_seq.load (std::memory_order_acquire) != tail + i + 1))
                {
                    break;
                }
 
                elements[i] = slot->data;
                slot->_seq.store (tail + i + capacity, std::memory_order_release);
                ++popped;
            }
 
            if (JOIN_LIKELY (popped > 0))
            {
                sync._tail.store (tail + popped, std::memory_order_release);
                return static_cast<ssize_t> (popped);
            }
 
            lastError = make_error_code (Errc::TemporaryError);
            return -1;
        }
    };
 
    template <typename Type, typename Backend>
    struct Mpmc
    {
        using Queue = BasicQueue<Type, Backend, Mpmc>;
        using Segment = QueueSegment<Type, QueueSlotFull<Type>>;
 
        static int tryPush (Segment* segment, const Type& element, uint64_t capacity, uint64_t mask) noexcept
        {
            return Mpsc<Type, Backend>::tryPush (segment, element, capacity, mask);
        }
 
        static ssize_t tryPush (Segment* segment, const Type* elements, size_t size, uint64_t capacity,
                                uint64_t mask) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr || elements == nullptr || size == 0))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            Backoff backoff;
            auto& sync = segment->_sync;
            uint64_t head = sync._head.load (std::memory_order_relaxed);
 
            for (;;)
            {
                uint64_t tail = sync._tail.load (std::memory_order_acquire);
                uint64_t toWrite = std::min (static_cast<uint64_t> (size), capacity - (head - tail));
 
                if (JOIN_UNLIKELY (toWrite == 0))
                {
                    lastError = make_error_code (Errc::TemporaryError);
                    return -1;
                }
 
                if (JOIN_LIKELY (sync._head.compare_exchange_weak (head, head + toWrite, std::memory_order_acquire,
                                                                   std::memory_order_relaxed)))
                {
                    for (uint64_t i = 0; i < toWrite; ++i)
                    {
                        auto* slot = &segment->_elements[(head + i) & mask];
                        Backoff slotBackoff;
                        while (slot->_seq.load (std::memory_order_acquire) != head + i)
                        {
                            slotBackoff ();  // LCOV_EXCL_LINE
                        }
                        slot->data = elements[i];
                        slot->_seq.store (head + i + 1, std::memory_order_release);
                    }
 
                    return static_cast<ssize_t> (toWrite);
                }
 
                backoff ();  // LCOV_EXCL_LINE
            }
        }
 
        static int tryPop (Segment* segment, Type& element, uint64_t capacity, uint64_t mask) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr))
            {
                // LCOV_EXCL_START
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            auto& sync = segment->_sync;
            uint64_t tail = sync._tail.load (std::memory_order_relaxed);
            Backoff backoff;
 
            for (;;)
            {
                auto* slot = &segment->_elements[tail & mask];
                uint64_t seq = slot->_seq.load (std::memory_order_acquire);
 
                if (seq == (tail + 1))
                {
                    Type local = slot->data;
                    if (JOIN_LIKELY (sync._tail.compare_exchange_weak (tail, tail + 1, std::memory_order_acquire,
                                                                       std::memory_order_relaxed)))
                    {
                        element = local;
                        slot->_seq.store (tail + capacity, std::memory_order_release);
                        return 0;
                    }
                }
                else if (JOIN_UNLIKELY (seq < (tail + 1)))
                {
                    lastError = make_error_code (Errc::TemporaryError);
                    return -1;
                }
                else
                {
                    backoff ();
                    tail = sync._tail.load (std::memory_order_relaxed);
                }
            }
        }
 
        static ssize_t tryPop (Segment* segment, Type* elements, size_t size, uint64_t capacity, uint64_t mask) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr || elements == nullptr || size == 0))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            Backoff backoff;
            auto& sync = segment->_sync;
            uint64_t tail = sync._tail.load (std::memory_order_relaxed);
 
            for (;;)
            {
                uint64_t head = sync._head.load (std::memory_order_acquire);
                uint64_t toRead = std::min (static_cast<uint64_t> (size), head - tail);
 
                if (JOIN_UNLIKELY (toRead == 0))
                {
                    lastError = make_error_code (Errc::TemporaryError);
                    return -1;
                }
 
                uint64_t ready = 0;
                for (; ready < toRead; ++ready)
                {
                    auto* slot = &segment->_elements[(tail + ready) & mask];
                    if (JOIN_UNLIKELY (slot->_seq.load (std::memory_order_acquire) != tail + ready + 1))
                    {
                        break;
                    }
                    elements[ready] = slot->data;
                }
 
                if (JOIN_UNLIKELY (ready == 0))
                {
                    lastError = make_error_code (Errc::TemporaryError);
                    return -1;
                }
 
                if (JOIN_LIKELY (sync._tail.compare_exchange_weak (tail, tail + ready, std::memory_order_acquire,
                                                                   std::memory_order_relaxed)))
                {
                    for (uint64_t i = 0; i < ready; ++i)
                    {
                        segment->_elements[(tail + i) & mask]._seq.store (tail + i + capacity,
                                                                          std::memory_order_release);
                    }
 
                    return static_cast<ssize_t> (ready);
                }
 
                backoff ();  // LCOV_EXCL_LINE
            }
        }
    };
 
    template <typename Backend, template <typename, typename> class SyncPolicy>
    struct SyncBinding
    {
        template <typename Type>
        using Queue = BasicQueue<Type, Backend, SyncPolicy<Type, Backend>>;
    };
}
 
#endif