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 <stdexcept>
#include <atomic>
 
namespace join
{
    struct QueueSync
    {
        static constexpr uint64_t MAGIC = 0x9F7E3B2A8D5C4E1B;
 
        alignas (64) std::atomic_uint64_t _magic;
 
        alignas (64) std::atomic_uint64_t _head;
 
        alignas (64) std::atomic_uint64_t _tail;
 
        alignas (64) uint64_t _capacity;
 
        alignas (64) uint64_t _mask;
    };
 
    template <typename Type>
    struct QueueSlot
    {
        alignas (64) std::atomic_uint64_t _seq;
 
        Type data;
 
        char _padding[(64 - ((sizeof (std::atomic_uint64_t) + sizeof (Type)) % 64)) % 64];
    };
 
    template <typename Type>
    struct QueueSegment
    {
        alignas (64) QueueSync _sync;
 
        QueueSlot<Type> _elements[];
    };
 
    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;
 
        template <typename... Args>
        BasicQueue (uint64_t capacity, Args&&... args)
        : _capacity (roundPow2 (capacity))
        , _elementSize (sizeof (QueueSlot<Type>))
        , _totalSize (sizeof (QueueSync) + (_capacity * _elementSize))
        , _backend (_totalSize, std::forward<Args> (args)...)
        , _segment (static_cast<QueueSegment<Type>*> (_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._tail.store (0, std::memory_order_relaxed);
                _segment->_sync._capacity = _capacity;
                _segment->_sync._mask     = _capacity - 1;
 
                for (uint64_t i = 0; i < _capacity; ++i)
                {
                    _segment->_elements[i]._seq.store (i, std::memory_order_relaxed);
                }
 
                _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 ();
                }
            }
 
            if (_segment->_sync._capacity != _capacity)
            {
                throw std::runtime_error ("capacity mismatch");
            }
        }
 
        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 _policy.tryPush (_segment, element);
        }
 
        int push (const Type& element) noexcept
        {
            Backoff backoff;
 
            while (tryPush (element) == -1)
            {
                if (JOIN_UNLIKELY (lastError != Errc::TemporaryError))
                {
                    return -1;
                }
 
                backoff ();
            }
 
            return 0;
        }
 
        int tryPop (Type& element) noexcept
        {
            return _policy.tryPop (_segment, element);
        }
 
        int pop (Type& element) noexcept
        {
            Backoff backoff;
 
            while (tryPop (element) == -1)
            {
                if (JOIN_UNLIKELY (lastError != Errc::TemporaryError))
                {
                    return -1;
                }
 
                backoff ();
            }
 
            return 0;
        }
 
        uint64_t pending () const noexcept
        {
            if (JOIN_UNLIKELY (_segment == nullptr))
            {
                return 0;
            }
            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;
            }
            return _segment->_sync._capacity - pending ();
        }
 
        bool full () const noexcept
        {
            if (JOIN_UNLIKELY (_segment == nullptr))
            {
                return false;
            }
            return pending () == _segment->_sync._capacity;
        }
 
        bool empty () const noexcept
        {
            if (JOIN_UNLIKELY (_segment == nullptr))
            {
                return true;
            }
            return pending () == 0;
        }
 
        int mbind (int numa) const noexcept
        {
            return _backend.mbind (numa);
        }
 
        int mlock () const noexcept
        {
            return _backend.mlock ();
        }
 
    protected:
        static uint64_t roundPow2 (uint64_t v) noexcept
        {
            if (v == 0)
            {
                return 1;
            }
            v--;
            v |= v >> 1;
            v |= v >> 2;
            v |= v >> 4;
            v |= v >> 8;
            v |= v >> 16;
            v |= v >> 32;
            return v + 1;
        }
 
        uint64_t _capacity = 0;
 
        uint64_t _elementSize = 0;
 
        uint64_t _totalSize = 0;
 
        Backend _backend;
 
        SyncPolicy _policy;
 
        QueueSegment<Type>* _segment = nullptr;
    };
 
    template <typename Type, typename Backend>
    struct Spsc
    {
        using Queue = BasicQueue<Type, Backend, Spsc>;
 
        static int tryPush (QueueSegment<Type>* segment, const Type& element) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            auto& sync    = segment->_sync;
            uint64_t head = sync._head.load (std::memory_order_relaxed);
            uint64_t tail = sync._tail.load (std::memory_order_acquire);
 
            if (JOIN_UNLIKELY ((head - tail) == sync._capacity))
            {
                lastError = make_error_code (Errc::TemporaryError);
                return -1;
            }
 
            segment->_elements[head & sync._mask].data = element;
            sync._head.store (head + 1, std::memory_order_release);
 
            return 0;
        }
 
        static int tryPop (QueueSegment<Type>* segment, Type& element) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr))
            {
                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);
 
            if (JOIN_UNLIKELY (head == tail))
            {
                lastError = make_error_code (Errc::TemporaryError);
                return -1;
            }
 
            element = segment->_elements[tail & sync._mask].data;
            sync._tail.store (tail + 1, std::memory_order_release);
 
            return 0;
        }
    };
 
    template <typename Type, typename Backend>
    struct Mpsc
    {
        using Queue = BasicQueue<Type, Backend, Mpsc>;
 
        static int tryPush (QueueSegment<Type>* segment, const Type& element) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            auto& sync    = segment->_sync;
            uint64_t head = sync._head.load (std::memory_order_relaxed);
            Backoff backoff;
 
            for (;;)
            {
                auto* slot   = &segment->_elements[head & sync._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 int tryPop (QueueSegment<Type>* segment, Type& element) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            auto& sync    = segment->_sync;
            uint64_t tail = sync._tail.load (std::memory_order_relaxed);
            auto* slot    = &segment->_elements[tail & sync._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 + sync._capacity, std::memory_order_release);
            sync._tail.store (tail + 1, std::memory_order_release);
 
            return 0;
        }
    };
 
    template <typename Type, typename Backend>
    struct Mpmc
    {
        using Queue = BasicQueue<Type, Backend, Mpmc>;
 
        static int tryPush (QueueSegment<Type>* segment, const Type& element) noexcept
        {
            return Mpsc<Type, Backend>::tryPush (segment, element);
        }
 
        static int tryPop (QueueSegment<Type>* segment, Type& element) noexcept
        {
            if (JOIN_UNLIKELY (segment == nullptr))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            auto& sync    = segment->_sync;
            uint64_t tail = sync._tail.load (std::memory_order_relaxed);
            Backoff backoff;
 
            for (;;)
            {
                auto* slot   = &segment->_elements[tail & sync._mask];
                uint64_t seq = slot->_seq.load (std::memory_order_acquire);
 
                if (seq == (tail + 1))
                {
                    if (JOIN_LIKELY (sync._tail.compare_exchange_weak (tail, tail + 1, std::memory_order_acquire,
                                                                       std::memory_order_relaxed)))
                    {
                        element = slot->data;
                        slot->_seq.store (tail + sync._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);
                }
            }
        }
    };
 
    template <typename Backend, template <typename, typename> class SyncPolicy>
    struct SyncBinding
    {
        template <typename Type>
        using Queue = BasicQueue<Type, Backend, SyncPolicy<Type, Backend>>;
    };
}
 
#endif