shm.hpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup ipcq_detail
 * @brief Contains core shm implementation details.
 * @copyright
 *   SPDX-FileCopyrightText: 2018-2024 European Southern Observatory (ESO)
 *
 *   SPDX-License-Identifier: LGPL-3.0-only
 */
#ifndef IPCQ_DETAIL_SHM_HPP_
#define IPCQ_DETAIL_SHM_HPP_
#include <atomic>
#include <array>
#include <chrono>
#include <functional>
#include <signal.h>  // kill
#include <system_error>
#include <thread>
#include <type_traits>
 
#include <boost/version.hpp>
// Headers for BoostIpcCondition
#include <boost/date_time/microsec_time_clock.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/interprocess/mapped_region.hpp>
#include <boost/interprocess/shared_memory_object.hpp>
#include <boost/interprocess/sync/interprocess_condition.hpp>
#include <boost/interprocess/sync/interprocess_mutex.hpp>
#include <boost/interprocess/sync/scoped_lock.hpp>
 
#include <numapp/mempolicy.hpp>
 
#include "../error.hpp"
#include "iterator.hpp"
#include "traits.hpp"
 
#if __GNUC__
#define restrict __restrict__
#else
#define restrict
#endif
 
#ifndef NDEBUG
#define IPCQ_ENABLE_ASSERT
#define IPCQ_ENABLE_LOG
#endif
 
#ifdef IPCQ_ENABLE_ASSERT
#define IPCQ_ASSERT(x) \
    do {               \
        assert(x);     \
    } while (0)
#else
#define IPCQ_ASSERT(x)
#endif
 
#ifdef IPCQ_ENABLE_LOG
#define IPCQ_LOG(x)                  \
    do {                             \
        std::cout << x << std::endl; \
    } while (0)
#else
#define IPCQ_LOG(x)
#endif
 
/** Version number */
#define IPCQ_FILE_VERSION 1
 
namespace ipcq::detail {
namespace config {
 
/**
 * File magic number that identifies ipcq shared objects.
 */
constexpr std::array<uint8_t, 7> MAGIC_NUMBER = {0x69, 0x70, 0x63, 0x71, 0xF0, 0x6A, 0x9C};
 
/**
 * Prefix used for shared object names.
 */
constexpr std::string_view SHM_OBJECT_NAME_PREFIX = "ipcq-";
 
/**
 * Number of bits making up first, last and size each for a total of 63 bits
 *
 * @ingroup ipcq_detail
 */
constexpr auto BIT_COUNT = 21u;
/**
 * Bitmask for bit count
 *
 * @ingroup ipcq_detail
 */
constexpr auto BIT_MASK = 0x1f'ffffu;
 
/**
 * Maximum number of elements in queue.
 * `(2 ^ BIT_COUNT) - 1`
 * @ingroup ipcq_detail
 */
constexpr auto MAX_VALUE = 2097151u;
 
/**
 * Bit indicating if queue is closed (highest bit).
 *
 * @ingroup ipcq_detail
 */
constexpr auto BIT_IS_CLOSED = 0x8000'0000'0000'0000u;
 
/**
 * Length of typename character sequences.
 *
 * @ingroup ipcq_detail
 */
constexpr size_t TYPENAME_SEQ_LEN = 128;
 
}  // namespace config
 
namespace ipc = boost::interprocess;
 
/**
 * Static parameters for a given shm queue.
 */
struct QueueInfo {
    static constexpr QueueInfo Make(size_t shm_size,
                                    size_t capacity,
                                    size_t first_element_offset,
                                    size_t counter_offset,
                                    size_t element_offset,
                                    size_t element_size) noexcept {
        return QueueInfo{
            shm_size, capacity, first_element_offset, counter_offset, element_offset, element_size};
    }
 
    /**
     * Number of bytes.
     */
    size_t shm_size;
    /**
     * Queue capacity in number of elements.
     */
    size_t capacity;
    size_t first_element_offset;
    /**
     * Offset to std::atomic<CounterType> from &Element<T>.
     */
    size_t counter_offset;
    /**
     * Offset to std::atomic<CounterType> from &Element<T>::element.
     */
    size_t element_offset;
    /**
     * Size of Element<T>.
     */
    size_t element_size;
};
 
static_assert(std::is_trivially_copyable_v<QueueInfo>, "must be trivial since it resides in SHM");
 
/**
 * Prefault memory by reading a byte in every page starting at @a start up to @a lenght bytes.
 *
 * To avoid optimizing this no-op the pointer is declared volatile.
 */
inline void PrefaultPages(void const* begin, void const* end) noexcept {
    volatile const uint8_t* ptr = static_cast<uint8_t const*>(begin);
    size_t pz = getpagesize();
    for (; ptr < end; ptr += pz) {
        (void)*ptr;
    }
}
 
/**
 * Control block residing in shared memory shared by readers and writers.
 * The control block only uses offsets since base address for each user is different.
 *
 * @ingroup ipcq_detail
 */
struct ShmControlBlock {
    /** Magic number that identifies ipcq files*/
    std::array<uint8_t, 7> magic;
    /** IPCQ protocol version number (not IPCQ version number)*/
    uint8_t  version;
 
    // They are frequently updated together (but not atomically).
    /** Range contains the compressed information about begin, end of queue. */
    std::atomic<unsigned long long> range;
    size_t element_type_hash;    ///< std::hash of element type T
    size_t condition_type_hash;  ///< std::hash of ConditionPolicy
    QueueInfo queue_spec;
    /**
     * Non-zero process id of owner if owner is available or
     * 0 if owner gracefully destructed itself but could
     * not free shared memory because it is still allocated by readers.
     */
    std::atomic<pid_t> owner_pid;
    /** @name Parameters used by ipcq-spy */
    // @{
    // Char sequences may or may not be terminated with \0.
    char element_type_name[config::TYPENAME_SEQ_LEN];
    char condition_type_name[config::TYPENAME_SEQ_LEN];
    // @}
};
 
static_assert(std::is_trivially_copyable<ShmControlBlock>::value,
              "ShmControlBlock must be a POD type since it located in shared memory");
 
static_assert(sizeof(ShmControlBlock) % 8 == 0);
 
/**
 * Helper that adds to posix_time::ptime a std:chrono duration.
 */
template <class Rep, class Period>
boost::posix_time::ptime AddDuration(boost::posix_time::ptime const& point,
                                     std::chrono::duration<Rep, Period> duration) noexcept {
    using namespace std::chrono;
    using usec = std::chrono::duration<int64_t, std::micro>;  // because std only specifies 55 bits
    return point + boost::posix_time::microseconds(duration_cast<usec>(duration).count());
}
 
/**
 * Creates the shared memory object name from topic_name.
 *
 * @param topic_name The ipcq shared memory topic name.
 */
inline std::string MakeShmObjectName(char const* topic_name) {
    std::string base(config::SHM_OBJECT_NAME_PREFIX);
    base += topic_name;
    return base;
}
 
/**
 * Small helper that removes shared memory, to make ShmControl exception-safe with RAII.
 *
 * @ingroup ipcq_detail
 */
template <class ShmTraits>
class ShmRemover {
public:
    using SharedMemoryObject = typename ShmTraits::SharedMemoryObject;
    using MappedRegion = typename ShmTraits::MappedRegion;
 
    enum class Policy { None = 0, Destructor = 1, ConstructorAndDestructor = 2 };
    ShmRemover(std::string name, Policy policy) : m_shm_name(std::move(name)), m_policy(policy) {
        if (m_policy == Policy::ConstructorAndDestructor) {
            SharedMemoryObject::remove(m_shm_name.c_str());
        }
    }
    ShmRemover(ShmRemover&& rhs) noexcept
        : m_shm_name(std::move(rhs.m_shm_name)), m_policy(rhs.m_policy) {
        // Moved from object should do nothing.
        rhs.m_shm_name = "";
        rhs.m_policy = Policy::None;
    }
    ShmRemover& operator=(ShmRemover&& rhs) noexcept {
        m_shm_name = std::move(rhs.m_shm_name);
        m_policy = rhs.m_policy;
        // Moved from object should do nothing.
        rhs.m_shm_name = "";
        rhs.m_policy = Policy::None;
        return *this;
    }
    ~ShmRemover() noexcept {
        if (m_policy == Policy::ConstructorAndDestructor || m_policy == Policy::Destructor) {
            (void)SharedMemoryObject::remove(m_shm_name.c_str());
        }
    }
 
private:
    std::string m_shm_name;
    Policy m_policy;
};
 
/**
 * Simple spinning condition
 *
 * @ingroup ipcq_detail
 */
class SpinConditionPolicy final {
public:
    using condition_t = unsigned int;
 
    void Signal() noexcept {
        m_condition.fetch_add(1, std::memory_order_release);
    }
 
    /**
     * @note: It is possible that between detecting that a wait is necessary and actually creating
     * the initial state to await a change on, the writer calls Signal();
     * In this scenario the reader will not be able to see the signal as a true signal.
     *
     * @todo: Mitigate this data race by reading range inside wait?
     */
    template <class Rep, class Period, class Predicate>
    [[nodiscard]] bool Wait(std::chrono::duration<Rep, Period> timeout, Predicate pred) noexcept {
        auto start = std::chrono::high_resolution_clock::now();
        while (!pred()) {
            auto initial = m_condition.load(std::memory_order_acquire);
            // Spin until condition variable changes, indicating a notication or that we time out.
            while (initial == m_condition.load(std::memory_order_acquire)) {
                if ((std::chrono::high_resolution_clock::now() - start) > timeout) {
                    return false;
                }
                std::this_thread::yield();
            }
        }
        return true;
    }
 
private:
    std::atomic<unsigned int> m_condition;
};
 
/**
 * Condition policy based on boost::interprocess::interprocess_condition.
 *
 * @ingroup ipcq_detail
 */
class BoostConditionPolicy final {
public:
    // Needs a default constructor
    BoostConditionPolicy() = default;
    /**
     * Unlike std::condition_variable in C++11, it is NOT safe to invoke the destructor if all
     * threads have been only notified. It is required that they have exited their respective wait
     * functions.
     */
    ~BoostConditionPolicy() {
        m_cond.notify_all();
        // Sleep for a while to let any reader wake up?
    }
 
    void Signal() noexcept {
        {
            // Prevent signal while readers hold lock so that the window between running predicate
            // and calling wait on m_cond is removed (which would make it possible to miss
            // a notification).
            ipc::scoped_lock<ipc::interprocess_mutex> lock(m_mtx);
        }
        m_cond.notify_all();
    }
 
    template <class Rep, class Period, class Predicate>
    [[nodiscard]] bool Wait(std::chrono::duration<Rep, Period> timeout, Predicate pred) noexcept {
        ipc::scoped_lock<ipc::interprocess_mutex> lock(m_mtx);
        return m_cond.timed_wait(
            lock,
            AddDuration(boost::posix_time::microsec_clock::universal_time(), timeout),
            std::forward<Predicate>(pred));
    }
 
private:
    ipc::interprocess_mutex m_mtx;
    ipc::interprocess_condition m_cond;
};
 
// @implements ShmTrait used by ShmControl
struct BoostInterprocessTraits {
    using SharedMemoryObject = boost::interprocess::shared_memory_object;
    using MappedRegion = boost::interprocess::mapped_region;
};
 
/**
 * Non-virtual base-class for ShmControl that does not depend on any static type.
 *
 * This is mainly a requirement to support ipcq-spy which is not templated on element type or
 * any policies.
 *
 * @tparam ShmTraits Trait type used to facilitate testing. Does not depend on user provided types.
 */
template <class ShmTraits = BoostInterprocessTraits>
class ShmControlBase {
public:
    using CounterType = size_t;
    using SharedMemoryObject = typename ShmTraits::SharedMemoryObject;
    using MappedRegion = typename ShmTraits::MappedRegion;
 
    /**
     * Owner constructor.
     *
     * @param size Data buffer in number of bytes.
     * @param data_offset Offset to first byte of the contigous queue buffer, relative m_queue
     * region.
     */
    explicit ShmControlBase(char const* name,
                            QueueInfo queue_info,
                            std::optional<numapp::MemPolicy> mem_policy = std::nullopt,
                            int mmap_flags = 0)
        : m_topic_name(name)
        , m_queue_info(queue_info)
        , m_shm_obj{ipc::create_only, MakeShmObjectName(name).c_str(), ipc::read_write}
        , m_shmem_remover{m_shm_obj.get_name(), ShmRemover<ShmTraits>::Policy::Destructor}
        , m_ctl_block{nullptr}
        , m_is_owner(true) {
        if (m_queue_info.capacity > config::MAX_VALUE) {
            throw std::system_error(make_error_code(Error::Overflow),
                                    "Requested capacity overflows storage capacity");
        }
        // Normally mbind(2) could be used to specify the memory policy when using mmap(2). But as
        // it says in mbind(2):
        //
        //    The specified policy will be ignored for any MAP_SHARED mappings in
        //    the specified memory range.  Rather the pages will be allocated
        //    according to the memory policy of the thread that caused the page to
        //    be allocated.  Again, this may not be the thread that called mbind().
        //
        // instead we would have to:
        // 1. Allocate (and not touch memory!)
        // 2. Set default mem policy (not with `mbind(2)`)
        // 3. Set mlock to avoid swapping out getting different policy next time.
        // 4. Fault all memory.
        //
        // And that is what is essentially done here by setting default memory policy for this
        // thread and then making sure that all memory is paged in (with MAP_POPULATE) but
        // with two sys-calls (`set_mempolicy(2)` and then `mmap(2)`). Three if we count the second
        // `set_mempolicy` to restore the previous policy after shared memory allocation.
        //
        std::optional<numapp::ScopedMemPolicy> scoped_policy;
        if (mem_policy) {
            scoped_policy.emplace(*mem_policy);
        }
        // Sets the size of the shared memory object to shm_size
#if BOOST_VERSION >= 108500
        m_shm_obj.truncate(m_queue_info.shm_size);
#else
        // Workaround for missing error code from `posix_fallocate()` fixed in boost 1.85
        // https://github.com/boostorg/interprocess/issues/166
        try {
            errno = 0;
            m_shm_obj.truncate(m_queue_info.shm_size);
        } catch (boost::interprocess::interprocess_exception const& ex) {
            if (ex.get_native_error() == 0) {
                throw boost::interprocess::interprocess_exception(
                    boost::interprocess::error_info(ENOSPC));
            } else {
                throw;
            }
        }
#endif
 
        m_control_region = std::move(MappedRegion(
            m_shm_obj,
            ipc::read_write,
            0,                        // control begins at 0
            sizeof(ShmControlBlock),  // size of region
            nullptr,
            (::boost::interprocess::default_map_options | MAP_LOCKED | MAP_POPULATE | mmap_flags)));
        m_queue_region =
            std::move(MappedRegion(m_shm_obj,
                                   ipc::read_write,
                                   sizeof(ShmControlBlock),  // begins where m_control_region ends
                                   0,                        // size of region is the rest (0)
                                   nullptr,
                                   ipc::default_map_options | MAP_LOCKED | MAP_POPULATE));
 
        // Initialize control block in shared memory.
        m_ctl_block = static_cast<ShmControlBlock*>(m_control_region.get_address());
        m_ctl_block->magic = config::MAGIC_NUMBER;
        m_ctl_block->version = IPCQ_FILE_VERSION;
        std::memcpy(&m_ctl_block->queue_spec,
                    &queue_info,
                    sizeof(queue_info));
        std::memcpy(&m_queue_info,
                    &queue_info,
                    sizeof(m_queue_info));
    }
 
    /**
     * Non-owner constructor.
     */
    explicit ShmControlBase(char const* name)
        : m_topic_name(name)
        , m_shm_obj{ipc::open_only,
                    (std::string(config::SHM_OBJECT_NAME_PREFIX) + name).c_str(),
                    ipc::read_write}
        , m_shmem_remover{m_shm_obj.get_name(), ShmRemover<ShmTraits>::Policy::None}
        , m_is_owner{false} {
        // We don't want readers to destruct the shared memory in constructor
        m_control_region =
            std::move(MappedRegion(m_shm_obj,
                                   ipc::read_only,
                                   0,                        // control begins at 0
                                   sizeof(ShmControlBlock),  // size
                                   nullptr,                  // addr
                                   ipc::default_map_options | MAP_LOCKED | MAP_POPULATE));
        m_queue_region = std::move(
            MappedRegion(m_shm_obj,
                         ipc::read_write,          // rw needed for condition policy
                         sizeof(ShmControlBlock),  // start at offset by size of control block
                         0,
                         nullptr,  // addr
                         ipc::default_map_options | MAP_LOCKED | MAP_POPULATE));
 
        m_ctl_block = static_cast<ShmControlBlock*>(m_control_region.get_address());
        if (this->m_ctl_block->magic != config::MAGIC_NUMBER) {
            throw std::system_error(make_error_code(Error::BadFile),
                                    m_shm_obj.get_name());
        }
        std::memcpy(&m_queue_info,
                    &m_ctl_block->queue_spec,
                    sizeof(m_queue_info));
    }
 
    ShmControlBase(ShmControlBase&& rhs) noexcept
        : m_topic_name(std::move(rhs.m_topic_name))
        , m_queue_info(rhs.m_queue_info)
        , m_shm_obj(std::move(rhs.m_shm_obj))
        , m_shmem_remover(std::move(rhs.m_shmem_remover))
        , m_control_region(std::move(rhs.m_control_region))
        , m_queue_region(std::move(rhs.m_queue_region))
        , m_ctl_block(rhs.m_ctl_block)
        , m_is_owner(rhs.m_is_owner) {
        // rhs no longer own the object.
        rhs.m_is_owner = false;
        rhs.m_queue_info.shm_size = 0u;
    }
 
    ShmControlBase& operator=(ShmControlBase&& rhs) noexcept {
        m_topic_name = std::move(rhs.m_topic_name);
        m_queue_info = rhs.m_queue_info;
        m_shm_obj = std::move(rhs.m_shm_obj);
        m_shmem_remover = std::move(rhs.m_shmem_remover);
        m_control_region = std::move(rhs.m_control_region);
        m_queue_region = std::move(rhs.m_queue_region);
        m_ctl_block = rhs.m_ctl_block;
        m_is_owner = rhs.m_is_owner;
        // rhs no longer own the object.
        rhs.m_is_owner = false;
        rhs.m_queue_info.shm_size = 0u;
        return *this;
    }
 
    [[nodiscard]] char const* TopicName() const noexcept {
        return m_topic_name.c_str();
    }
 
    [[nodiscard]] char const* ShmObjectName() const {
        return m_shm_obj.get_name();
    }
 
    [[nodiscard]] constexpr size_t Capacity() const noexcept {
        return m_queue_info.capacity;
    }
 
    /**
     * @returns ipcq shared object file/protocol version
     */
    [[nodiscard]] constexpr uint8_t FileVersion() const noexcept {
        return this->m_ctl_block->version;
    }
 
    /**
     * @returns PID of registered owner (writer) process.
     */
    [[nodiscard]] constexpr pid_t OwnerPid() const noexcept {
        return this->m_ctl_block->owner_pid;
    }
 
protected:
    /// Topic name, as provided by user.
    std::string m_topic_name;
    QueueInfo m_queue_info;
 
    /// Manages the shared memory
    SharedMemoryObject m_shm_obj;
    // Helper object that deletes the allocated shared memory
    ShmRemover<ShmTraits> m_shmem_remover;
    /// Mapped segment to the control (mapped as read only for Reader).
    MappedRegion m_control_region;
    /// Mapped region to the queue (mapped as read only for Reader).
    MappedRegion m_queue_region;
    /// @name Pointers to mapped shared memory regions
    // @{
    ShmControlBlock* m_ctl_block;
    // @}
 
    bool m_is_owner;  ///< Indicates if this object owns the shared memory or not
};
 
/**
 * ShmControl provides the basic data model for the queue and low level operations
 * on the shared memory queue, some of which are atomic.
 *
 * The shared data model for the queue includes:
 *
 * - a Range describing the valid range of the data and
 * - the capacity of the circular buffer.
 *
 * This information can be compressed into a single unsigned long long value
 * which can be atomically read or writen.
 *
 * The atomic operations are:
 *
 * - `AtomicLoadState()'
 * - `AtomicStoreState()`
 *
 *
 * ShmControl does not encode knowledge of the buffer semantics. It merely provides the shared
 * access patterns to it.
 *
 * @todo: Optimize atomic stores. Since x86 (and AMD64) do not reorder stores with other stores and
 * loads with other loads a single operation could be used to load both range and generation as well
 * as store range and generation.
 *
 * Two mappings are created to the shm segment, one for the control block and one for the queue.
 * It's not really important, but if each region is mapped individually, there's less chance that an
 * accidental overwrite of the queue region affects the control region. It also make offset
 * calculations simpler since the control block is not accounted for.
 *
 * Writer Protocol
 * ---------------
 *
 * Writers to the queue should adhere to the following protocol to ensure data consistency or detect
 * inconsistent states:
 *
 * To write one or more elements:
 *
 *   1. Reserve enough room for elements to be written by calling `Pop()` once for each element to
 *   write.
 *   2. "Publish" change of queue state by calling `AtomicStoreState()`.
 *   3. Add elements by calling `Push()`.
 *   4. "Publish" change of queue again by calling `AtomicStoreState()`.
 *
 * Reader Protocol
 * ---------------
 *
 * 1. Read current queue state with `AtomicLoadState()`
 * 2. Verify that iterator is valid.
 * 3. Copy.
 * 4. Read current queue state with `AtomicLoadState()`
 * 5. Verify that iterator is still valid.
 * 6. Verify that copied element has the expected counter value.
 *
 * Important
 * ---------
 *
 * boost interprocess uses mmap on linux, and it is possible that write combine is used
 * under unknown circumstances. If write combining is enabled then the strong
 * memory ordering is no longer true.
 *
 * Implementation Notes
 * --------------------
 *
 * boost::shared_memory_object corresponds to a file handle, opened with `open(2)`.
 * boost::shared_memory_object::truncate corresponds to `ftruncate(2)`.
 *
 * boost::mapped_region construction corresponds to a call to `mmap(2)` which in
 * the case of ShmControl also allocate, lock and populate the memory using
 * MAP_LOCKED and MAP_POPULATE.
 *
 * User control over memory policy for the allocation is made possible because
 * ShmControl in the "writer" is always the one that creates the boost::shared_memory_object
 * and can then make sure that the chosen memory policy is applied when the allocation
 * is performed in `mmap(2)`.
 *
 * @ingroup ipcq_detail
 */
template <class T, class ConditionPolicy, class ShmTraits = BoostInterprocessTraits>
class ShmControl final : public ShmControlBase<ShmTraits> {
public:
    using CounterType = typename ShmControlBase<ShmTraits>::CounterType;
    using SharedMemoryObject = typename ShmTraits::SharedMemoryObject;
    struct Element {
        std::atomic<CounterType> counter;
        T element;
    };
 
    using value_type = Element;
    using size_type = typename NonConstTraits<value_type>::size_type;
    using reference = typename NonConstTraits<value_type>::reference;
    using const_reference = typename ConstTraits<value_type>::reference;
    using difference_type = ptrdiff_t;
    using ConstIterator = detail::Iterator<ShmControl, ConstTraits<value_type>>;
    using Iterator = detail::Iterator<ShmControl, NonConstTraits<value_type>>;
 
    friend Iterator;
    friend ConstIterator;
 
    // Object in shared memory
    struct Queue {
        ConditionPolicy condition;
        value_type buffer[1];  // beginning of buffer
    };
 
    /**
     * Attempts to remove shared memory object.
     *
     * @note @a shm_object_name is not the topic name, but the raw object name.
     *
     * @param shm_object_name The raw object name to remove.
     * @returns false on error.
     */
    static std::error_code RemoveShmObject(char const* shm_object_name) noexcept {
        errno = 0;
        // Returns false on error
        if (!SharedMemoryObject::remove(shm_object_name)) {
            return std::make_error_code(std::errc(errno));
        }
        return {};
    }
 
    /**
     * Create a new (owned) named shared memory segment.
     *
     * @warning Will (try to) destroy shared memory with same name if it exists.
     *
     * @param name Full name of shared memory segment.
     * @param capacity Capacity of queue. Maximum capacity is config::MAX_CAPACITY.
     * @param mem_policy Optional memory policy that is applied when allocating the shared memory.
     * @param mmap_flags Additional flags that will be or-ed with the default flags
     *                   MAP_LOCKED and MAP_POPULATE.
     *
     * @throws boost::interprocess::interprocess_exception if creating shared memory fails.
     * @throws std::system_error with
     *  - Error::Overflow if requested capacity is greater than what can be represented by the
     *    queue.
     */
    explicit ShmControl(char const* name,
                        size_t capacity,
                        std::optional<numapp::MemPolicy> mem_policy = std::nullopt,
                        int mmap_flags = 0)
        : ShmControlBase<ShmTraits>(
              name,
              QueueInfo::Make(
                  sizeof(ShmControlBlock) + offsetof(Queue, buffer) + sizeof(value_type) * capacity,
                  capacity,
                  offsetof(Queue, buffer),
                  offsetof(Element, counter),
                  offsetof(Element, element),
                  sizeof(Element)
                  ),
              mem_policy,
              mmap_flags)
        , m_buffer{nullptr}
        , m_capacity{capacity}
        , m_first{nullptr}
        , m_last{nullptr}
        , m_size{0u}
        , m_is_closed{false}
        , m_counter{0} {
        assert(this->m_ctl_block);
 
        m_queue = new (this->m_queue_region.get_address()) Queue();
 
        m_buffer = m_queue->buffer;
        m_first = m_buffer;
        m_last = m_buffer;
 
        {
            // Possibly truncated string is expected and allowed.
            #pragma GCC diagnostic push
            #pragma GCC diagnostic ignored "-Wpragmas"
            #pragma GCC diagnostic ignored "-Wstringop-truncation"
            this->m_ctl_block->element_type_hash = std::hash<std::string>{}(typeid(T).name());
            strncpy(this->m_ctl_block->element_type_name, typeid(T).name(),
                    config::TYPENAME_SEQ_LEN);
            this->m_ctl_block->condition_type_hash =
                std::hash<std::string>{}(typeid(ConditionPolicy).name());
            strncpy(this->m_ctl_block->condition_type_name, typeid(ConditionPolicy).name(),
                    config::TYPENAME_SEQ_LEN);
            #pragma GCC diagnostic pop
            AtomicStoreState();
        }
        // Finally mark the queue available by setting the pid
        this->m_ctl_block->owner_pid.store(getpid(), std::memory_order_release);
    }
 
    /**
     * Open an existing (not owned) shared memory segment in read only mode.
     *
     * @param name Full name of shared memory segment.
     *
     * @throws std::system_error If
     *  - attaching to shared memory fail
     *  - ipcq::Error::TypeMismatch if element or condition policy types don't match
     *  - ipcq::Error::Closed if queue is closed or owner does not exist.
     */
    explicit ShmControl(char const* name)
        : ShmControlBase<ShmTraits>(name)
        , m_first{nullptr}
        , m_last{nullptr}
        , m_size{0u}
        , m_is_closed{false}
        , m_counter{0} {
        assert(this->m_ctl_block);
 
        // Check if writer exists by checking the pid, which is the last synchronization being done
        // by writer before queue is usable. Therefore this is checked first.
        //
        // Writer will set pid to zero if it gracefully exits, on crashes it may still be set
        // but process is gone, so we check that the process exist with `kill()`.
        auto owner_pid = this->m_ctl_block->owner_pid.load(std::memory_order_acquire);
        auto owner_exists = owner_pid != 0 && kill(owner_pid, 0) == 0;
 
        if (this->m_ctl_block->version != IPCQ_FILE_VERSION) {
            throw std::system_error(make_error_code(Error::BadVersion),
                                    this->ShmObjectName());
        }
 
        if (!owner_exists) {
            throw std::system_error(make_error_code(Error::Closed), "Writer does not exist");
        }
 
        if (this->m_ctl_block->element_type_hash != std::hash<std::string>{}(typeid(T).name())) {
            throw std::system_error(make_error_code(Error::TypeMismatch), "Element type mismatch");
        }
 
        if (this->m_ctl_block->condition_type_hash !=
            std::hash<std::string>{}(typeid(ConditionPolicy).name())) {
            throw std::system_error(make_error_code(Error::TypeMismatch),
                                    "ConditionPolicy type mismatch");
        }
 
        m_queue = static_cast<Queue*>(this->m_queue_region.get_address());
        m_buffer = m_queue->buffer;
        m_capacity = this->Capacity();
 
        // Check if writer exists by checking the pid.
        // writer will set it to zero if it gracefully exits, on crashes it may still be set
        // but process is gone, so we check that as well.
        AtomicLoadState();
        if (m_is_closed) {
            throw std::system_error(make_error_code(Error::Closed),
                                    "Writer exists but queue is closed");
        }
    }
 
    ~ShmControl() noexcept {
        if (this->m_is_owner) {
            Close();
            AtomicStoreState();
            this->m_ctl_block->owner_pid = 0;
            m_queue->~Queue();
        }
    }
 
    /**
     * Moves from rhs into this.
     *
     * @pre rhs is valid.
     * @post rhs is invalid.
     */
    ShmControl(ShmControl&& rhs) noexcept
        : ShmControlBase<ShmTraits>(std::move(rhs))
        , m_queue(rhs.m_queue)
        , m_buffer(rhs.m_buffer)
        , m_capacity(rhs.m_capacity)
        , m_first(rhs.m_first)
        , m_last(rhs.m_last)
        , m_size(rhs.m_size)
        , m_is_closed(rhs.m_is_closed)
        , m_counter(rhs.m_counter) {
        // rhs no longer own the object.
        rhs.m_is_owner = false;
        rhs.m_queue = nullptr;
        rhs.m_buffer = nullptr;
        rhs.m_first = nullptr;
        rhs.m_last = nullptr;
        rhs.m_size = 0u;
        rhs.m_is_closed = true;
        rhs.m_counter = 0u;
    }
 
    /**
     * Moves from rhs into this.
     *
     * @pre rhs is valid.
     * @post rhs is invalid.
     */
    ShmControl& operator=(ShmControl&& rhs) noexcept {
        ShmControlBase<ShmTraits>::operator=(std::move(rhs));
 
        m_queue = rhs.m_queue;
        m_buffer = rhs.m_buffer;
        m_capacity = rhs.m_capacity;
        m_first = rhs.m_first;
        m_last = rhs.m_last;
        m_size = rhs.m_size;
        m_is_closed = rhs.m_is_closed;
        m_counter = rhs.m_counter;
        // rhs no longer own the object.
        rhs.m_queue = nullptr;
        rhs.m_buffer = nullptr;
        rhs.m_first = nullptr;
        rhs.m_last = nullptr;
        rhs.m_size = 0u;
        rhs.m_is_closed = true;
        rhs.m_counter = 0u;
        return *this;
    }
 
    /**
     * Load queue state from shared memory
     *
     * Performs acquire-synchronization that pairs with AtomicStoreState.
     */
    constexpr void AtomicLoadState() noexcept {
        unsigned long long val = this->m_ctl_block->range.load(std::memory_order_acquire);
        DecompressState(val);
    }
 
    /**
     * Store queue range state to shared memory.
     *
     * Performs release-synchronization that pairs with AtomicLoadState.
     */
    constexpr void AtomicStoreState() noexcept {
        unsigned long long val = CompressState();
        this->m_ctl_block->range.store(val, std::memory_order_release);
    }
 
    /**
     * Signal objects blocking on AwaitSignal.
     */
    void Signal() {
        m_queue->condition.Signal();
    }
 
    /**
     * Await notification signal from writer.
     *
     * @returns false on timeout, true otherwise.
     */
    template <class Rep, class Period, class Predicate>
    [[nodiscard]] bool
    AwaitSignal(std::chrono::duration<Rep, Period> timeout, Predicate pred) noexcept {
        return m_queue->condition.Wait(timeout, std::move(pred));
    }
 
    /**
     * Pop @a first element in queue.
     *
     * @note: Changes won't be seen by others until AtomicStoreState is invoked.
     */
    constexpr void Pop() {
        IPCQ_ASSERT(!IsEmpty());
        Increment(m_first);
        --m_size;
    }
 
 
    /**
     * @returns true if queue is full, false otherwise.
     */
    [[nodiscard]] constexpr bool IsFull() const noexcept {
        return m_size == this->m_capacity;
    }
 
    /**
     * @returns true if queue is empty, false otherwise.
     */
    [[nodiscard]] constexpr bool IsEmpty() const noexcept {
        return m_size == 0;
    }
 
    /**
     * @returns true if queue is closed and will not be written to again, false otherwise.
     * @sa Close().
     */
    [[nodiscard]] constexpr bool IsClosed() const noexcept {
        return m_is_closed;
    }
 
    /**
     * Close queue for further writing. Existing elements may still be written and Pop() is allowed.
     *
     * @note Push() will fail after this.
     * @sa IsClosed()
     */
    constexpr void Close() noexcept {
        m_is_closed = true;
    }
 
    /**
     * Push element to back of queue, if `IsFull()` it will overwrite first element.
     *
     * @param element Element to push to queue.
     * @returns `Error::Closed` if queue is closed.
     * @returns `0` if element was added to queue.
     */
    [[nodiscard]] std::error_code Push(const T& element) noexcept {
        /*
         * @todo: Require that queue is not full, to enforce Popping first? Might not make a
         * difference since were not enforcing the full protocol...
         */
        return Push(&element, 1);
    }
 
    /**
     * @returns Error::Closed if queue is closed.
     */
    [[nodiscard]] std::error_code Push(const T* data, size_t n) noexcept {
        if (m_is_closed) {
            return make_error_code(Error::Closed);
        }
 
        for (auto i = 0u; i < n; ++i) {
            // Atomically store new counter value which signals that element
            // is invalidated.
            m_last->counter.store(m_counter++, std::memory_order_release);
 
            // Copy new element to last position
            std::memcpy(&(m_last->element), data++, sizeof(*data));
 
            if (IsFull()) {
                // Note: Capacity cannot be 0, IsFull() and IsEmpty() cannot both be true.
                //
                // Overwrite last element, then increment both first and last element to rotate
                // queue one
                Increment(m_last);
                m_first = m_last;
            } else {
                // Add without overwriting existing elements
                Increment(m_last);
                ++m_size;
            }
        }
        return {};
    }
 
    [[nodiscard]] reference operator[](size_type index) {
        IPCQ_ASSERT(index < Size());  // check for invalid index
        return *Add(m_first, index);
    }
 
    [[nodiscard]] const_reference operator[](size_type index) const {
        IPCQ_ASSERT(index < Size());  // check for invalid index
        return *Add(m_first, index);
    }
 
    [[nodiscard]] constexpr size_type Size() const noexcept {
        return m_size;
    }
 
    [[nodiscard]] constexpr value_type* GetBuffer() const noexcept {
        return m_buffer;
    }
 
    [[nodiscard]] constexpr value_type* GetQueuePtr() const noexcept {
        return m_buffer;
    }
 
    [[nodiscard]] constexpr ConstIterator Begin() const noexcept {
        return ConstIterator(this, IsEmpty() ? nullptr : m_first);
    }
 
    [[nodiscard]] constexpr ConstIterator End() const noexcept {
        return ConstIterator(this, nullptr);
    }
    // @}
private:
    template <class Pointer>
    [[nodiscard]] constexpr Pointer Add(Pointer p, difference_type n) const noexcept {
        return p + (n < (m_buffer + this->m_capacity - p) ? n : n - m_capacity);
    }
 
    template <class Pointer>
    constexpr void Increment(Pointer& p) const noexcept {
        if (++p == m_buffer + m_capacity) {
            p = m_buffer;
        }
    }
 
    constexpr void Decrement(value_type*& p) const noexcept {
        if (p == m_buffer) {
            p = m_buffer + m_capacity;
        }
        --p;
    }
 
    [[nodiscard]] constexpr unsigned long long CompressState() const noexcept {
        using namespace ipcq::detail::config;
        unsigned long long dest =
            (static_cast<unsigned long long>(m_first - m_buffer) << BIT_COUNT * 0) |
            (static_cast<unsigned long long>(m_last - m_buffer) << BIT_COUNT * 1) |
            (static_cast<unsigned long long>(m_size) << BIT_COUNT * 2) |
            (static_cast<unsigned long long>(m_is_closed ? BIT_IS_CLOSED : 0u));
        return dest;
    }
 
    constexpr void DecompressState(unsigned long long src) noexcept {
        using namespace ipcq::detail::config;
        m_first = m_buffer + ((src >> BIT_COUNT * 0) & BIT_MASK);
        m_last = m_buffer + ((src >> BIT_COUNT * 1) & BIT_MASK);
        m_size = (src >> BIT_COUNT * 2) & BIT_MASK;
        m_is_closed = (src & BIT_IS_CLOSED) == BIT_IS_CLOSED;
    }
 
    /// @name Pointers to mapped shared memory regions
    // @{
    Queue* m_queue;
    // @}
 
    value_type* m_buffer;  // Base address of queue for which offsets in m_ctl_block is valid
    size_t m_capacity;
    /**
     * @name Atomically updated together in the shared state with
     * AtomicStoreState and AtomicLoadstate.
     */
    // @{
    value_type* m_first;  // First element
    value_type* m_last;   // One past last element
    unsigned m_size;      // Current size of the queue
    bool m_is_closed;     //< Is the queue closed?
    // @}
    CounterType m_counter;  // Element counter
};
 
}  // namespace ipcq::detail
 
#endif