shared.hpp

Go to the documentation of this file.

 
#ifndef __JOIN_SHARED_HPP__
#define __JOIN_SHARED_HPP__
 
// libjoin.
#include <join/error.hpp>
 
// C++.
#include <utility>
#include <thread>
#include <atomic>
#include <chrono>
#include <memory>
#include <string>
 
// C.
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
#include <fcntl.h>
#include <cstring>
 
namespace join
{
    class SharedMemory
    {
    public:
        SharedMemory (const std::string& name, uint64_t size)
        : _name (name)
        , _size (size)
        {
            if (_size > static_cast <uint64_t> (std::numeric_limits <off_t>::max ()))
            {
                throw std::overflow_error ("size will overflow");
            }
        }
 
        virtual ~SharedMemory ()
        {
            this->close ();
        }
 
        virtual int open ()
        {
            if (this->opened ())
            {
                lastError = make_error_code (Errc::InUse);
                return -1;
            }
 
            bool created = true;
            this->_fd = ::shm_open (this->_name.c_str (), O_CREAT | O_RDWR | O_EXCL | O_CLOEXEC, 0644);
            if ((this->_fd == -1) && (errno == EEXIST))
            {
                created = false;
                this->_fd = ::shm_open(this->_name.c_str(), O_RDWR | O_CLOEXEC, 0644);
            }
 
            if (this->_fd == -1)
            {
                lastError = std::make_error_code (static_cast <std::errc> (errno));
                this->close ();
                return -1;
            }
 
            if (created && (::ftruncate (this->_fd, this->_size) == -1))
            {
                lastError = std::make_error_code (static_cast <std::errc> (errno));
                this->close ();
                return -1;
            }
 
            this->_ptr = ::mmap (nullptr, this->_size, PROT_READ | PROT_WRITE, MAP_SHARED, this->_fd, 0);
            if (this->_ptr == MAP_FAILED)
            {
                lastError = std::make_error_code (static_cast <std::errc> (errno));
                this->close ();
                return -1;
            }
 
            return 0;
        }
 
        bool opened () const noexcept
        {
            return (this->_ptr != nullptr) && (this->_ptr != MAP_FAILED);
        }
 
        virtual void close () noexcept 
        {
            if (this->opened ())
            {
                ::munmap (this->_ptr, this->_size);
            }
 
            if (this->_fd != -1)
            {
                ::close (this->_fd);
                this->_fd = -1;
            }
 
            this->_ptr = nullptr;
        }
 
        virtual const void* get (uint64_t offset = 0) const
        {
            if (offset >= this->_size)
            {
                throw std::out_of_range ("offset out of bounds");
            }
 
            if (this->_ptr == nullptr)
            {
                return nullptr;
            }
 
            return static_cast <const char*> (this->_ptr) + offset;
        }
 
        virtual void* get (uint64_t offset = 0)
        {
            if (offset >= this->_size)
            {
                throw std::out_of_range ("offset out of bounds");
            }
 
            if (this->_ptr == nullptr)
            {
                return nullptr;
            }
 
            return static_cast <char*> (this->_ptr) + offset;
        }
 
        virtual uint64_t size () const noexcept
        {
            return this->_size;
        }
 
        static int unlink (const std::string& name) noexcept
        {
            if (::shm_unlink (name.c_str ()) == -1)
            {
                if (errno == ENOENT)
                {
                    return 0;
                }
                lastError = std::make_error_code (static_cast <std::errc> (errno));
                return -1;
            }
 
            return 0;
        }
 
    protected:
        std::string _name;
 
        uint64_t _size = 0;
 
        void* _ptr = nullptr;
 
        int _fd = -1;
    };
 
    struct SharedSync
    {
        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 _elementSize;
        alignas (64) uint64_t _capacity;
    };
 
    struct SharedSegment
    {
        alignas (64) SharedSync _sync;
        alignas (max_align_t) uint8_t _data[];
    };
 
    template <typename Policy>
    class BasicQueue : public SharedMemory
    {
    public:
        BasicQueue (const std::string& name, uint64_t elementSize = 1472, uint64_t capacity = 144)
        : SharedMemory (name, (elementSize * capacity) + sizeof (SharedSync))
        , _elementSize (elementSize)
        , _capacity (capacity)
        {
            if ((_capacity != 0) && (_elementSize > (std::numeric_limits <uint64_t>::max () / _capacity)))
            {
                throw std::overflow_error ("size will overflow");
            }
 
            _userSize = _elementSize * _capacity;
            _totalSize = _userSize + sizeof (SharedSync);
        }
 
        BasicQueue (const BasicQueue& other) = delete;
 
        BasicQueue& operator= (const BasicQueue& other) = delete;
 
        virtual ~BasicQueue ()
        {
            this->close ();
        }
 
        int open () override
        {
            if (SharedMemory::open () != 0)
            {
                return -1;
            }
 
            this->_segment = static_cast <SharedSegment*> (this->_ptr);
 
            uint64_t expected = 0;
            if (this->_segment->_sync._magic.compare_exchange_strong (expected, 0xFFFFFFFFFFFFFFFF, std::memory_order_acquire, std::memory_order_acquire))
            {
                new (&this->_segment->_sync._head) std::atomic_uint64_t (0);
                new (&this->_segment->_sync._tail) std::atomic_uint64_t (0);
                this->_segment->_sync._elementSize = this->_elementSize;
                this->_segment->_sync._capacity = this->_capacity;
 
                this->_segment->_sync._magic.store (SharedSync::MAGIC, std::memory_order_release);
            }
            else
            {
                while (this->_segment->_sync._magic.load (std::memory_order_acquire) != SharedSync::MAGIC)
                {
                    std::this_thread::yield ();
                }
            }
 
            if ((this->_segment->_sync._elementSize != this->_elementSize) || (this->_segment->_sync._capacity != this->_capacity))
            {
                lastError = make_error_code (Errc::InvalidParam);
                this->close ();
                return -1;
            }
 
            return 0;
        }
 
        void close () noexcept override
        {
            this->_segment = nullptr;
            SharedMemory::close ();
        }
 
        const void* get (uint64_t offset = 0) const override
        {
            if (offset > this->_userSize)
            {
                throw std::out_of_range ("offset out of bounds");
            }
 
            return SharedMemory::get (sizeof (SharedSync) + offset);
        }
 
        void* get (uint64_t offset = 0) override
        {
            if (offset > this->_userSize)
            {
                throw std::out_of_range ("offset out of bounds");
            }
 
            return SharedMemory::get (sizeof (SharedSync) + offset);
        }
 
        uint64_t elementSize () const noexcept
        {
            return this->_elementSize;
        }
 
        uint64_t capacity () const noexcept
        {
            return this->_capacity;
        }
 
        uint64_t size () const noexcept override
        {
            return this->_userSize;
        }
 
    protected:
        Policy _policy;
 
        uint64_t _elementSize = 0;
 
        uint64_t _capacity = 0;
 
        uint64_t _totalSize = 0;
 
        uint64_t _userSize = 0;
 
        SharedSegment* _segment = nullptr;
    };
 
    template <typename Policy>
    class BasicProducer : public BasicQueue <Policy>
    {
    public:
        BasicProducer (const std::string& name, uint64_t elementSize = 1472, uint64_t capacity = 144)
        : BasicQueue <Policy> (name, elementSize, capacity)
        {
        }
 
        BasicProducer (const BasicProducer& other) = delete;
 
        BasicProducer& operator= (const BasicProducer& other) = delete;
 
        ~BasicProducer () = default;
 
        int tryPush (const void* element) noexcept
        {
            return this->_policy.tryPush (this->_segment, element);
        }
 
        int push (const void* element) noexcept
        {
            return this->_policy.push (this->_segment, element);
        }
 
        template <class Rep, class Period>
        int timedPush (const void* element, std::chrono::duration <Rep, Period> timeout) noexcept
        {
            return this->_policy.timedPush (this->_segment, element, timeout);
        }
 
        uint64_t available () const noexcept
        {
            return this->_policy.available (this->_segment);
        }
 
        bool full () const noexcept
        {
            return this->_policy.full (this->_segment);
        }
    };
 
    template <typename Policy>
    class BasicConsumer : public BasicQueue <Policy>
    {
    public:
        BasicConsumer (const std::string& name, uint64_t elementSize = 1472, uint64_t capacity = 144)
        : BasicQueue <Policy> (name, elementSize, capacity)
        {
        }
 
        BasicConsumer (const BasicConsumer& other) = delete;
 
        BasicConsumer& operator= (const BasicConsumer& other) = delete;
 
        ~BasicConsumer () = default;
 
        int tryPop (void* element) noexcept
        {
            return this->_policy.tryPop (this->_segment, element);
        }
 
        int pop (void* element) noexcept
        {
            return this->_policy.pop (this->_segment, element);
        }
 
        template <class Rep, class Period>
        int timedPop (void* element, std::chrono::duration <Rep, Period> timeout) noexcept
        {
            return this->_policy.timedPop (this->_segment, element, timeout);
        }
 
        uint64_t pending () const noexcept
        {
            return this->_policy.pending (this->_segment);
        }
 
        bool empty () const noexcept
        {
            return this->_policy.empty (this->_segment);
        }
    };
 
    template <typename OutboundPolicy, typename InboundPolicy>
    class BasicEndpoint
    {
    public:
        using Outbound = typename OutboundPolicy::Producer;
        using Inbound  = typename InboundPolicy::Consumer;
 
        enum Side
        {
            A,      
            B,      
        };
 
        BasicEndpoint (Side side, const std::string& name, uint64_t elementSize = 1472, uint64_t capacity = 144)
        : _side (side)
        , _name (name)
        {
            if (this->_side == Side::A)
            {
                this->_out = std::make_unique <Outbound> (this->_name + "_AB", elementSize, capacity);
                this->_in  = std::make_unique <Inbound>  (this->_name + "_BA", elementSize, capacity);
            }
            else
            {
                this->_out = std::make_unique <Outbound> (this->_name + "_BA", elementSize, capacity);
                this->_in  = std::make_unique <Inbound>  (this->_name + "_AB", elementSize, capacity);
            }
        }
 
        BasicEndpoint (const BasicEndpoint& other) = delete;
 
        BasicEndpoint& operator= (const BasicEndpoint& other) = delete;
 
        ~BasicEndpoint ()
        {
            this->close();
        }
 
        int open ()
        {
            if (this->opened ())
            {
                lastError = make_error_code (Errc::InUse);
                return -1;
            }
 
            if ((this->_out->open () == -1) || (this->_in->open () == -1))
            {
                this->close ();
                return -1;
            }
 
            return 0;
        }
 
        void close ()
        {
            if (this->_out)
            {
                this->_out->close ();
            }
 
            if (this->_in)
            {
                this->_in->close ();
            }
        }
 
        bool opened () const
        {
            return this->_out && this->_out->opened () && this->_in && this->_in->opened ();
        }
 
        int trySend (const void* element)
        {
            return this->_out->tryPush (element);
        }
 
        int send (const void* element)
        {
            return this->_out->push (element);
        }
 
        template <class Rep, class Period>
        int timedSend (const void* element, std::chrono::duration <Rep, Period> timeout)
        {
            return this->_out->timedPush (element, timeout);
        }
 
        int tryReceive (void* element)
        {
            return this->_in->tryPop (element);
        }
 
        int receive (void* element)
        {
            return this->_in->pop (element);
        }
 
        template <class Rep, class Period>
        int timedReceive (void* element, std::chrono::duration <Rep, Period> timeout)
        {
            return this->_in->timedPop (element, timeout);
        }
 
        uint64_t available () const
        {
            return this->_out->available ();
        }
 
        bool full () const
        {
            return this->_out->full ();
        }
 
        uint64_t pending () const
        {
            return this->_in->pending ();
        }
 
        bool empty () const
        {
            return this->_in->empty ();
        }
 
        Side side () const
        {
            return this->_side;
        }
 
        const std::string& name () const
        {
            return this->_name;
        }
 
        uint64_t elementSize () const
        {
            return this->_in->elementSize ();
        }
 
        uint64_t capacity () const
        {
            return this->_in->capacity ();
        }
 
    private:
        Side _side;
 
        std::string _name;
 
        std::unique_ptr <Outbound> _out;
 
        std::unique_ptr <Inbound> _in;
    };
 
    class Spsc
    {
    public:
        using Producer = BasicProducer <Spsc>;
        using Consumer = BasicConsumer <Spsc>;
        using Endpoint = BasicEndpoint <Spsc, Spsc>;
 
        constexpr Spsc () noexcept = default;
 
        virtual int tryPush (SharedSegment* segment, const void* element) const noexcept
        {
            if ((segment == nullptr) || (element == nullptr))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
            uint64_t tail = segment->_sync._tail.load (std::memory_order_acquire);
            uint64_t head = segment->_sync._head.load (std::memory_order_relaxed);
            if ((head - tail) == segment->_sync._capacity)
            {
                lastError = make_error_code (Errc::TemporaryError);
                return -1;
            }
            auto next = head % segment->_sync._capacity;
            std::memcpy (segment->_data + (next * segment->_sync._elementSize), element, segment->_sync._elementSize);
            segment->_sync._head.store (head + 1, std::memory_order_release);
            return 0;
        }
 
        int push (SharedSegment* segment, const void* element) const noexcept
        {
            while (tryPush (segment, element) == -1)
            {
                if (lastError != Errc::TemporaryError)
                {
                    return -1;
                }
                std::this_thread::yield ();
            }
            return 0;
        }
 
        template <class Rep, class Period>
        int timedPush (SharedSegment* segment, const void* element, std::chrono::duration <Rep, Period> timeout) const noexcept
        {
            auto const deadline = std::chrono::steady_clock::now () + timeout;
            while (tryPush (segment, element) == -1)
            {
                if (lastError != Errc::TemporaryError)
                {
                    return -1;
                }
                if (std::chrono::steady_clock::now () >= deadline)
                {
                    lastError = make_error_code (Errc::TimedOut);
                    return -1;
                }
                std::this_thread::yield ();
            }
            return 0;
        }
 
        virtual int tryPop (SharedSegment* segment, void* element) const noexcept
        {
            if ((segment == nullptr) || (element == nullptr))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
            uint64_t head = segment->_sync._head.load (std::memory_order_acquire);
            uint64_t tail = segment->_sync._tail.load (std::memory_order_relaxed);
            if ((head - tail) == 0)
            {
                lastError = make_error_code (Errc::TemporaryError);
                return -1;
            }
            auto next = tail % segment->_sync._capacity;
            std::memcpy (element, segment->_data + (next * segment->_sync._elementSize), segment->_sync._elementSize);
            segment->_sync._tail.store (tail + 1, std::memory_order_release);
            return 0;
        }
 
        int pop (SharedSegment* segment, void* element) const noexcept
        {
            while (tryPop (segment, element) == -1)
            {
                if (lastError != Errc::TemporaryError)
                {
                    return -1;
                }
                std::this_thread::yield ();
            }
            return 0;
        }
 
        template <class Rep, class Period>
        int timedPop (SharedSegment* segment, void* element, std::chrono::duration <Rep, Period> timeout) const noexcept
        {
            auto const deadline = std::chrono::steady_clock::now () + timeout;
            while (tryPop (segment, element) == -1)
            {
                if (lastError != Errc::TemporaryError)
                {
                    return -1;
                }
                if (std::chrono::steady_clock::now () >= deadline)
                {
                    lastError = make_error_code (Errc::TimedOut);
                    return -1;
                }
                std::this_thread::yield ();
            }
            return 0;
        }
 
        uint64_t pending (SharedSegment* segment) const noexcept
        {
            auto head = segment->_sync._head.load (std::memory_order_acquire);
            auto tail = segment->_sync._tail.load (std::memory_order_acquire);
            return head - tail;
        }
 
        uint64_t available (SharedSegment* segment) const noexcept
        {
            return segment->_sync._capacity - pending (segment);
        }
 
        bool full (SharedSegment* segment) const noexcept
        {
            return pending (segment) == segment->_sync._capacity;
        }
 
        bool empty (SharedSegment* segment) const noexcept
        {
            return pending (segment) == 0;
        }
    };
 
    class Mpsc : public Spsc
    {
    public:
        using Producer = BasicProducer <Mpsc>;
        using Consumer = BasicConsumer <Mpsc>;
        using Endpoint = BasicEndpoint <Mpsc, Mpsc>;
 
        constexpr Mpsc () noexcept = default;
 
        int tryPush (SharedSegment* segment, const void* element) const noexcept override
        {
            if ((segment == nullptr) || (element == nullptr))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
            uint64_t head, tail;
            do
            {
                tail = segment->_sync._tail.load (std::memory_order_acquire);
                head = segment->_sync._head.load (std::memory_order_relaxed);
                if ((head - tail) == segment->_sync._capacity)
                {
                    lastError = make_error_code (Errc::TemporaryError);
                    return -1;
                }
            }
            while (!segment->_sync._head.compare_exchange_weak (head, head + 1, std::memory_order_acquire, std::memory_order_relaxed));
            auto slot = head % segment->_sync._capacity;
            std::memcpy (segment->_data + (slot * segment->_sync._elementSize), element, segment->_sync._elementSize);
            std::atomic_thread_fence (std::memory_order_release);
            return 0;
        }
    };
 
    class Mpmc : public Mpsc
    {
    public:
        using Producer = BasicProducer <Mpmc>;
        using Consumer = BasicConsumer <Mpmc>;
        using Endpoint = BasicEndpoint <Mpmc, Mpmc>;
 
        constexpr Mpmc () noexcept = default;
 
        int tryPop (SharedSegment* segment, void* element) const noexcept override
        {
            if ((segment == nullptr) || (element == nullptr))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
            uint64_t head, tail;
            do
            {
                head = segment->_sync._head.load (std::memory_order_acquire);
                tail = segment->_sync._tail.load (std::memory_order_relaxed);
                if ((head - tail) == 0)
                {
                    lastError = make_error_code (Errc::TemporaryError);
                    return -1;
                }
            }
            while (!segment->_sync._tail.compare_exchange_weak (tail, tail + 1, std::memory_order_acquire, std::memory_order_relaxed));
            auto slot = tail % segment->_sync._capacity;
            std::memcpy (element, segment->_data + (slot * segment->_sync._elementSize), segment->_sync._elementSize);
            std::atomic_thread_fence (std::memory_order_release);
            return 0;
        }
    };
}
 
#endif