memory.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_MEMORY_HPP
#define JOIN_CORE_MEMORY_HPP
 
// libjoin.
#include <join/error.hpp>
#include <join/utils.hpp>
 
// C++.
#include <system_error>
#include <stdexcept>
#include <limits>
 
// C.
#include <sys/types.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
#ifdef JOIN_HAS_NUMA
#include <numaif.h>
#include <numa.h>
#endif
#include <fcntl.h>
#include <cstdint>
 
namespace join
{
    template <typename Backend, size_t Count, size_t... Sizes>
    class BasicArena;
 
    template <typename Backend, template <typename, typename> class SyncPolicy>
    struct SyncBinding;
 
    template <typename, typename>
    struct Spsc;
 
    template <typename, typename>
    struct Mpsc;
 
    template <typename, typename>
    struct Mpmc;
 
#ifdef JOIN_HAS_NUMA
    inline int mbind (void* ptr, size_t size, int numa) noexcept
    {
        if (ptr == nullptr || numa < 0 || numa > (static_cast<int> (sizeof (unsigned long) * 8) - 1))
        {
            lastError = make_error_code (Errc::InvalidParam);
            return -1;
        }
 
        unsigned long mask = (1UL << numa);
        if (::mbind (ptr, size, MPOL_BIND, &mask, sizeof (mask) * 8, MPOL_MF_STRICT) == -1)
        {
            lastError = std::error_code (errno, std::generic_category ());
            return -1;
        }
 
        return 0;
    }
#endif
 
    inline int mlock (void* ptr, size_t size) noexcept
    {
        if (ptr == nullptr)
        {
            lastError = make_error_code (Errc::InvalidParam);
            return -1;
        }
 
        if (::mlock (ptr, size) == -1)
        {
            lastError = std::error_code (errno, std::generic_category ());
            return -1;
        }
 
        return 0;
    }
 
    class LocalMem
    {
    public:
        template <size_t Count, size_t... Sizes>
        using Allocator = BasicArena<LocalMem, Count, Sizes...>;
 
        using Spsc = SyncBinding<LocalMem, ::join::Spsc>;
        using Mpsc = SyncBinding<LocalMem, ::join::Mpsc>;
        using Mpmc = SyncBinding<LocalMem, ::join::Mpmc>;
 
        explicit LocalMem (uint64_t size)
        {
            long sc = sysconf (_SC_PAGESIZE);
            uint64_t pageSize = (sc > 0) ? static_cast<uint64_t> (sc) : _defaultPageSize;
            _size = (size + pageSize - 1) & ~(pageSize - 1);
 
            create ();
        }
 
        LocalMem (const LocalMem& other) = delete;
 
        LocalMem& operator= (const LocalMem& other) = delete;
 
        LocalMem (LocalMem&& other) noexcept
        : _size (other._size)
        , _ptr (other._ptr)
        {
            other._size = 0;
            other._ptr = nullptr;
        }
 
        LocalMem& operator= (LocalMem&& other) noexcept
        {
            cleanup ();
 
            _size = other._size;
            _ptr = other._ptr;
 
            other._size = 0;
            other._ptr = nullptr;
 
            return *this;
        }
 
        ~LocalMem () noexcept
        {
            cleanup ();
        }
 
        const void* get (uint64_t offset = 0) const
        {
            if (JOIN_UNLIKELY (_ptr == nullptr))
            {
                throw std::runtime_error ("memory not mapped");
            }
 
            if (JOIN_UNLIKELY (offset >= _size))
            {
                throw std::out_of_range ("offset out of bounds");
            }
 
            return static_cast<const char*> (_ptr) + offset;
        }
 
        void* get (uint64_t offset = 0)
        {
            if (JOIN_UNLIKELY (_ptr == nullptr))
            {
                throw std::runtime_error ("memory not mapped");
            }
 
            if (JOIN_UNLIKELY (offset >= _size))
            {
                throw std::out_of_range ("offset out of bounds");
            }
 
            return static_cast<char*> (_ptr) + offset;
        }
 
#ifdef JOIN_HAS_NUMA
        int mbind (int numa) const noexcept
        {
            return join::mbind (_ptr, _size, numa);
        }
#endif
 
        int mlock () const noexcept
        {
            return join::mlock (_ptr, _size);
        }
 
    private:
        void create ()
        {
            _ptr = ::mmap (nullptr, _size, PROT_READ | PROT_WRITE,
                           MAP_PRIVATE | MAP_ANONYMOUS | MAP_POPULATE | MAP_HUGETLB, -1, 0);
            if ((_ptr == MAP_FAILED) && ((errno == ENOMEM) || (errno == EINVAL)))
            {
                // no hugepages available or no support.
                _ptr =
                    ::mmap (nullptr, _size, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS | MAP_POPULATE, -1, 0);
            }
 
            if (_ptr == MAP_FAILED)
            {
                throw std::system_error (errno, std::generic_category (), "mmap failed");
            }
        }
 
        void cleanup () noexcept
        {
            if ((_ptr != nullptr) && (_ptr != MAP_FAILED))
            {
                ::munlock (_ptr, _size);
                ::munmap (_ptr, _size);
                _ptr = nullptr;
            }
 
            _size = 0;
        }
 
        static constexpr uint64_t _defaultPageSize = 4096;
 
        uint64_t _size = 0;
 
        void* _ptr = nullptr;
    };
 
    class ShmMem
    {
    public:
        template <size_t Count, size_t... Sizes>
        using Allocator = BasicArena<ShmMem, Count, Sizes...>;
 
        using Spsc = SyncBinding<ShmMem, ::join::Spsc>;
        using Mpsc = SyncBinding<ShmMem, ::join::Mpsc>;
        using Mpmc = SyncBinding<ShmMem, ::join::Mpmc>;
 
        explicit ShmMem (uint64_t size, const std::string& name)
        : _name (name)
        {
            long sc = sysconf (_SC_PAGESIZE);
            uint64_t pageSize = (sc > 0) ? static_cast<uint64_t> (sc) : _defaultPageSize;
            _size = (size + pageSize - 1) & ~(pageSize - 1);
 
            if (_size > static_cast<uint64_t> (std::numeric_limits<off_t>::max ()))
            {
                throw std::overflow_error ("size will overflow");
            }
 
            create ();
        }
 
        ShmMem (const ShmMem& other) = delete;
 
        ShmMem& operator= (const ShmMem& other) = delete;
 
        ShmMem (ShmMem&& other) noexcept
        : _size (other._size)
        , _name (std::move (other._name))
        , _ptr (other._ptr)
        , _fd (other._fd)
        {
            other._size = 0;
            other._ptr = nullptr;
            other._fd = -1;
        }
 
        ShmMem& operator= (ShmMem&& other) noexcept
        {
            cleanup ();
 
            _size = other._size;
            _name = std::move (other._name);
            _ptr = other._ptr;
            _fd = other._fd;
 
            other._size = 0;
            other._ptr = nullptr;
            other._fd = -1;
 
            return *this;
        }
 
        ~ShmMem () noexcept
        {
            cleanup ();
        }
 
        const void* get (uint64_t offset = 0) const
        {
            if (JOIN_UNLIKELY (_ptr == nullptr))
            {
                throw std::runtime_error ("memory not mapped");
            }
 
            if (JOIN_UNLIKELY (offset >= _size))
            {
                throw std::out_of_range ("offset out of bounds");
            }
 
            return static_cast<const char*> (_ptr) + offset;
        }
 
        void* get (uint64_t offset = 0)
        {
            if (JOIN_UNLIKELY (_ptr == nullptr))
            {
                throw std::runtime_error ("memory not mapped");
            }
 
            if (JOIN_UNLIKELY (offset >= _size))
            {
                throw std::out_of_range ("offset out of bounds");
            }
 
            return static_cast<char*> (_ptr) + offset;
        }
 
#ifdef JOIN_HAS_NUMA
        int mbind (int numa) const noexcept
        {
            return join::mbind (_ptr, _size, numa);
        }
#endif
 
        int mlock () const noexcept
        {
            return join::mlock (_ptr, _size);
        }
 
        static int unlink (const std::string& name) noexcept
        {
            if (::shm_unlink (name.c_str ()) == -1)
            {
                if (errno == ENOENT)
                {
                    return 0;
                }
                lastError = std::error_code (errno, std::generic_category ());
                return -1;
            }
 
            return 0;
        }
 
    private:
        void create ()
        {
            bool created = true;
 
            _fd = ::shm_open (_name.c_str (), O_CREAT | O_RDWR | O_EXCL | O_CLOEXEC, 0644);
            if ((_fd == -1) && (errno == EEXIST))
            {
                created = false;
                _fd = ::shm_open (_name.c_str (), O_RDWR | O_CLOEXEC, 0644);
            }
 
            if (_fd == -1)
            {
                throw std::system_error (errno, std::generic_category (), "shm_open failed");
            }
 
            if (!created)
            {
                struct stat st;
 
                if (fstat (_fd, &st) == -1)
                {
                    int err = errno;
                    ::close (_fd);
                    throw std::system_error (err, std::generic_category (), "fstat failed");
                }
 
                if (static_cast<uint64_t> (st.st_size) != _size)
                {
                    ::close (_fd);
                    throw std::runtime_error ("shared memory size mismatch");
                }
            }
            else
            {
                if (::ftruncate (_fd, _size) == -1)
                {
                    int err = errno;
                    ::close (_fd);
                    throw std::system_error (err, std::generic_category (), "ftruncate failed");
                }
            }
 
            _ptr = ::mmap (nullptr, _size, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_HUGETLB, _fd, 0);
            if ((_ptr == MAP_FAILED) && ((errno == ENOMEM) || (errno == EINVAL)))
            {
                // no hugepages available or no support.
                _ptr = ::mmap (nullptr, _size, PROT_READ | PROT_WRITE, MAP_SHARED, _fd, 0);
            }
 
            if (_ptr == MAP_FAILED)
            {
                int err = errno;
                ::close (_fd);
                throw std::system_error (err, std::generic_category (), "mmap failed");
            }
        }
 
        void cleanup () noexcept
        {
            if ((_ptr != nullptr) && (_ptr != MAP_FAILED))
            {
                ::munlock (_ptr, _size);
                ::munmap (_ptr, _size);
                _ptr = nullptr;
            }
 
            if (_fd != -1)
            {
                ::close (_fd);
                _fd = -1;
            }
 
            _name.clear ();
            _size = 0;
        }
 
        static constexpr uint64_t _defaultPageSize = 4096;
 
        uint64_t _size = 0;
 
        std::string _name;
 
        void* _ptr = nullptr;
 
        int _fd = -1;
    };
}
 
#endif