writer.hpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup ipcq
 * @brief Contains declarations for @c ipcq::BasicWriter.
 * @copyright
 *   SPDX-FileCopyrightText: 2018-2023 European Southern Observatory (ESO)
 *
 *   SPDX-License-Identifier: LGPL-3.0-only
 */
#ifndef IPCQ_WRITER_HPP_
#define IPCQ_WRITER_HPP_
 
#include "policies.hpp"
#include "detail/shm.hpp"
 
namespace ipcq {
 
/**
 * Notification policy.
 *
 * @ingroup ipcq
 */
enum class Notify {
    /**
     * Don't notify.
     */
    None,
    /**
     * Notify all waiting readers.
     */
    All
};
 
/**
 * Queue writer class.
 *
 * @tparam T Type of elements in queue.
 * @tparam ConditionPolicy is the policy to use to signal readers when
 *         new elements have been appended to queue. ipcq provides @a ipcq::BoostConditionPolicy
 *         and @a ipcq::SpinConditionPolicy.
 * @tparam ShmTraits Implementation details. Leave default.
 *
 * Basic Usage:
 *
 *     struct MyTopicType {
 *        ...
 *     };
 *
 *     // Create writer for the topic "topicname" with type MyTopicType.
 *     auto writer = ipcq::Writer<MyTopicType>("topicname", 100);
 *
 *     while (!shouldStop) {
 *         MyTopicType sample;
 *         // Prepare sample
 *         ...
 *         auto err = writer.Write(sample, ipcq::Notify::NotifyAll);
 *         if (err) {
 *             // Handle error
 *             break;
 *         }
 *     }
 *     // Signal to readers that the queue is now closed.
 *     writer.Close();
 *
 * @sa ipcq::BoostConditionPolicy ipcq::SpinConditionPolicy.
 * @ingroup ipcq
 */
template <class T, class ConditionPolicy, class ShmTraits=detail::BoostInterprocessTraits>
class BasicWriter {
    using ShmControl = typename detail::ShmControl<T, ConditionPolicy, ShmTraits>;
    ShmControl m_ctl;
 
public:
    /**
     * Remove shared memory object from the system.
     *
     * @note Internally this unlinks the file.
     *
     * @param topic_name Topic name, as provided to BasicWriter constructor.
     * @return 0 on success and contents of `errno` on failure.
     */
    static std::error_code RemoveShmObject(char const* topic_name) noexcept {
        return detail::ShmControl<T, ConditionPolicy, ShmTraits>::RemoveShmObject(
            detail::MakeShmObjectName(topic_name).c_str());
    }
 
    /**
     * Creates a writer and allocates shared memory to have room for capacity Ts.
     *
     * @param topic_name Name of the topic to create, together with the template parameter @c T this
     * forms the full topic for the queue.
     * @param capacity The queue capacity in number of elements.
     * @param mempolicy An optional memory policy that is used for the shared memory allocation.
     */
    explicit BasicWriter(char const* topic_name,
                         size_t capacity,
                         std::optional<numapp::MemPolicy> const& mempolicy = std::nullopt)
        : m_ctl{topic_name, capacity, mempolicy} {
    }
 
    /**
     * Get queue topic name.
     *
     * @returns topic name in queue.
     */
    [[nodiscard]] char const* TopicName() const noexcept {
        return m_ctl.TopicName();
    }
 
    /**
     * Query queue capacity, which is the number of elements the queue can hold.
     *
     * @return queue capacity in number of elements.
     */
    [[nodiscard]] constexpr size_t Capacity() const noexcept {
        return m_ctl.Capacity();
    }
 
    /**
     * Query number of elements in the queue.
     *
     * @return queue size in number of elements.
     */
    [[nodiscard]] constexpr size_t Size() noexcept {
        return m_ctl.Size();
    }
 
    /**
     * Query whether queue is closed or not.
     *
     * @returns true if queue is closed, false otherwise.
     */
    [[nodiscard]] constexpr bool IsClosed() const noexcept {
        return m_ctl.IsClosed();
    }
 
    /**
     * Close queue for further writing.
     *
     * Any waiting readers will be notified. Once existing data is read out subsequent reads will
     * fail with `ipcq::Error::Closed`.
     *
     * @post Shared memory remains valid but new reader instances will not attach.
     *
     * @releases
     *
     * @sa IsClosed()
     */
    constexpr void Close() noexcept {
        m_ctl.Close();
        m_ctl.AtomicStoreState();
        m_ctl.Signal();
    }
 
    /**
     * Append single element to queue, possibly overwriting the front element.
     *
     * @releases
     *
     * @param data Element to copy into queue
     * @param notify_readers Determines if readers will be notified (woken up if blocked) that new
     * element is available.
     * @return error value is queue is closed.
     */
    [[nodiscard]] std::error_code Write(T const& data, Notify notify_readers) noexcept {
        return Write(&data, 1, notify_readers);
    }
 
 
    /**
     * Remove single element from queue.
     *
     * @releases
     *
     * @pre @code Size() > 0 @endcode
     */
    constexpr void Pop() noexcept {
        m_ctl.Pop();
        // Make this change visible to all readers
        m_ctl.AtomicStoreState();
    }
 
    /**
     * Write @c count elements to queue.
     *
     * @pre @code count <= Capacity() @endcode
     *
     * @note Function will make room as necessary to write @c count elements.
     *
     * @releases
     *
     * @return Error::Closed if queue is closed.
     */
    [[nodiscard]] std::error_code
    Write(const T* data, size_t count, Notify notify_readers) noexcept {
        IPCQ_ASSERT(count <= m_ctl.Capacity());
 
        // See if we need to pop elements from queue
        size_t num_free = static_cast<size_t>(m_ctl.Capacity() - m_ctl.Size());
        if (count > num_free) {
            int num_to_pop = count - num_free;
            while (num_to_pop--) {
                m_ctl.Pop();
            }
            // Make this change visible to all readers to invalidate that will be written to
            m_ctl.AtomicStoreState();
        }
 
        if (auto err = m_ctl.Push(data, count); err) {
            return err;
        }
 
        // Make this change visible to all reader by performing release synchronization.
        m_ctl.AtomicStoreState();
 
        if (notify_readers == Notify::All) {
            m_ctl.Signal();
        }
 
        return {};
    }
};
 
/**
 * Convenience alias with reasonable default condition policy.
 *
 * @ingroup ipcq
 */
template <class T>
using Writer = BasicWriter<T, BoostConditionPolicy>;
 
}  // namespace ipcq
#endif