reader.hpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup ipcq
 * @brief Contains definition of ipcq::BasicReader.
 * @copyright
 *   SPDX-FileCopyrightText: 2018-2023 European Southern Observatory (ESO)
 *
 *   SPDX-License-Identifier: LGPL-3.0-only
 */
#ifndef IPCQ_READER_HPP_
#define IPCQ_READER_HPP_
#include <chrono>
#include <optional>
#include <thread>
 
#include "detail/shm.hpp"
#include "error.hpp"
#include "policies.hpp"
 
namespace ipcq {
namespace detail {
 
template <class>
inline constexpr bool always_false_v = false;  // NOLINT
 
template <bool nothrow, class State>
struct ReaderRollback;
template <class State>
struct ReaderRollback<true, State> {};
template <class State>
struct ReaderRollback<false, State> {
    State state;
};
 
}  // namespace detail
 
template <class T, class ConditionPolicy, class ShmTraits>
class BasicReader;
 
/**
 * Queue BasicReader class.
 *
 * @tparam T Type of elements in queue.
 * @tparam ConditionPolicy Policy used to determine how readers are notified when
 *         queue has new elements appeneded to it.
 * @tparam ShmTraits Implementation details. Leave default.
 *
 * Basic Usage:
 *
 *     struct MyTopicType {
 *        ...
 *     };
 *     // Create reader for the topic "topicname" with type MyTopicType.
 *     // If the Writer is not yet created, this function will retry until it gives up
 *     // after the given timeout.
 *     auto reader = ipcq::Reader<MyTopicType>::MakeReader("topicname", 30s);
 *
 *     while (true) {
 *         MyTopicType sample;
 *         // Read sample or timeout after 2s
 *         auto err = reader.Read(ipcq::OutputIteratorAdapter(&sample), 2s);
 *         if (err) {
 *             // Handle error
 *             break;
 *         }
 *         // Process sample
 *         ...
 *     }
 *
 * @ingroup ipcq
 */
template <class T, class ConditionPolicy, class ShmTraits = detail::BoostInterprocessTraits>
class BasicReader {
    using ShmControl = typename detail::ShmControl<T, ConditionPolicy, ShmTraits>;
    using ConstIterator = typename ShmControl::ConstIterator;
    using CounterType = typename ShmControl::CounterType;
 
    /**
     * Local state of what has been copied.
     */
    ShmControl m_ctl;
    struct State {
        std::optional<ConstIterator> last_element;  //< Iterator to next element to read from queue
        CounterType next_counter;
    };
    State m_state;
 
    /**
     * Wait for iterator to next element to become valid
     *
     * Performs acquire-synchronization.
     *
     * @returns Error::Closed if queue is closed and element to read is not valid.
     * @returns Error::Timeout if writer did not signal us before `timeout` time elapsed and
     *          element iterator become valid.
     */
    template <class Rep, class Period>
    [[nodiscard]] std::pair<std::error_code, ConstIterator>
    AwaitNextIterator(std::chrono::duration<Rep, Period> timeout) noexcept {
        m_ctl.AtomicLoadState();
 
        ConstIterator it{};
        if (m_state.last_element) {
            it = *m_state.last_element;
            ++it;  // This might advance to End()
        } else {
            it = m_ctl.Begin();
        }
 
        if (it == m_ctl.End()) {
            // This may also happen if queue has 1 element or more generally if writer catches up to
            // reader such that when m_last_element was incremented, it compares equally to "one
            // past" last valid element.
            if (m_ctl.IsClosed()) {
                return {make_error_code(Error::Closed), {}};
            }
 
            // Predicate returns false if the waiting should be continued, true if waiting should be
            // stopped.
            //
            // @todo: Optimize by only using a predicate if spurious wakeups are possible?
            // @todo: Add prefetch hints?
            if (!m_ctl.AwaitSignal(timeout, [this, &it]() mutable {
                    // Guard against spurious wakeups while waiting for condition
                    m_ctl.AtomicLoadState();
                    auto no_wait_more = m_ctl.IsClosed();
 
                    if (m_ctl.IsEmpty()) {
                        return no_wait_more;
                    }
                    // if the queue has a single element then this doesnt work
                    if (m_state.last_element) {
                        auto test_it = *m_state.last_element;
                        ++test_it;  // note: may iterate to m_ctl.End()
                        if (test_it < m_ctl.Begin() || test_it >= m_ctl.End()) {
                            // invalid iterator
                            return no_wait_more;
                        }
                        it = test_it;
                        return true;
                    } else {
                        it = m_ctl.Begin();
                        return true;
                    }
                })) {
                // Timeout occured
                return {make_error_code(Error::Timeout), {}};
            } else {
                // Not timeout, but could also be because Writer woke us up due to queue closing.
                // If queue is closed, only report it as an error if iterator is invalid
                if (it == m_ctl.End()) {
                    return {make_error_code(Error::Closed), it};
                }
            }
        } else if (it < m_ctl.Begin() || it >= m_ctl.End()) {
            return {make_error_code(Error::InconsistentState), {}};
        }
 
        // The predicate checks everything, so it should be good now!
        IPCQ_ASSERT(it >= m_ctl.Begin() && it < m_ctl.End());
        return {{}, it};
    }
 
public:
    /**
     * @name Construction
     * @{
     */
    /**
     * Create and connect reader to existing queue.
     *
     * @param topic_name Name of topic to associate reader with.
     *
     * @throws boost::interprocess::interprocess_exception if attaching to shared memory fails.
     * @throws std::system_error with
     *     ipcq::Error::TypeMismatch if T or ConditionPolicy does not match types in queue.
     *     ipcq::Error::Closed if queue is closed, not yet ready or writer process does not exist.
     *
     * @sa BasicReader::MakeReader
     */
    explicit BasicReader(char const* topic_name) : m_ctl(topic_name), m_state{{}, {0}} {
        // @todo: Decide if Reset() should be called automatically and immediately.
    }
 
    /**
     * Factory function that retries to create a reader until success or timeout.
     */
    template <class Rep, class Period>
    [[nodiscard]] static BasicReader
    MakeReader(char const* topic_name, std::chrono::duration<Rep, Period> timeout) {
        using namespace std::chrono;
        auto expiry = high_resolution_clock::now() + timeout;
 
        do {
            try {
                return BasicReader(topic_name);
            } catch (boost::interprocess::interprocess_exception const& e) {
                std::this_thread::sleep_for(500ms);
            }
        } while (high_resolution_clock::now() < expiry);
        throw std::system_error(make_error_code(Error::Timeout),
                                "Timeout waiting for ipcq shared memory queue to be created");
    }
    /** @} */
    /**
     * @name Accessors
     * @{
     */
    /**
     * Get queue topic name.
     *
     * @returns topic name in queue.
     */
    [[nodiscard]] char const* TopicName() const noexcept {
        return m_ctl.TopicName();
    }
 
    /**
     * 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();
    }
 
    /**
     * 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.
     *
     * @sa BasicReader::Synchronize
     */
    [[nodiscard]] constexpr size_t Size() const noexcept {
        return m_ctl.Size();
    }
 
    /**
     * Query number of elements that is currently available for reading. In other words it's the
     * number of unread elements.
     *
     * @note This does not synchronize with shared memory.
     *
     * This can e.g. be used to calculate how near the reader is to be overwritten as the following
     * approaches zero:
     * @code
     * reader.Size() - reader.NumAvailable()
     * @endcode
     *
     * @return number of currently available elements.
     */
    [[nodiscard]] constexpr size_t NumAvailable() const noexcept {
        if (m_state.last_element) {
            return std::distance(*m_state.last_element, m_ctl.End()) -
                   1;  // - 1 since End is one-past
        } else {
            return m_ctl.Size();
        }
    }
    /** @} */
 
    /**
     * @name Read operations
     * @{
     */
 
    /**
     * Reads available data if any, or waits for notification from writer, and then reads available
     * data from queue.
     *
     * @note Method is not guaranteed to read all the requested @c count elements.
     *
     * Reads a maximum of @a count elements from the queue synchronously by invoking the provided
     * @c Operation @c op once with each element. The reader advances the next element to read
     * automatically.
     *
     * If @a any data is already available then no waiting will occur.
     * If no data is available then it will wait for writer notification and then
     * read a maximum of `count` elements. If no notification occurs before specified @c timeout
     * duration no reads will occur and @c ipcq::Error::Timeout is returned.
     *
     * @warning Data read from the queue must be considered invalid until the return value error
     * code is verified to be 0.
     *
     * @acquires
     *
     * @param op Operation function used to operate on read element. Is invoked once for each
     * element. Adapters to standard containers can be used as operation c.f. @c
     * ipcq::OutputIteratorAdapter and @c ipcq::BackInserter.
     * @param count Maximum number of elements to read from queue.
     * @param timeout Maximum time to wait for any data, if no data is not already available to
     * read. Value range and precision is guaranteed to 1 microsecond in range 1 microsecond to 256
     * hours.
     *
     * @tparam Operation callable with requirements: @c `std::is_invocable_v<Operation, T
     * const&>`, e.g. a callable with signature @c `void Operation(T const& element)`.
     * or @c `void Operation(T const& element) noexcept`.
     *
     * @returns a pair with error code and number of read elements.
     * @returns Error::InconsistentState if memory consistency errors occurred (see Reset() for how
     * to recover).
     * @returns Error::Timeout if no data became available before timing out.
     *
     * @throw Exceptions originating from @a op. Function provides strong exception guarantee for
     * side-effects in the class itself but not in provided read-operation @c `op`. User may attempt
     * to read the same elements again by issuing a new call to @a Read..
     *
     * @sa ipcq::OutputIteratorAdapter ipcq::BackInserter ipcq::BasicReader::Reset
     */
    template <class Operation, class Rep, class Period>
    [[nodiscard]] std::pair<std::error_code, size_t>
    Read(Operation&& op, size_t count, std::chrono::duration<Rep, Period> timeout) noexcept(
        std::is_nothrow_invocable_v<Operation, T const&>) {
        return ReadExt(std::forward<Operation>(op), count, timeout);
    }
 
    /**
     * Skip up to @c count elements from queue.
     *
     * @note It is functionally equivalent as @c Read with a no-op @c op.
     *
     * @acquires
     *
     * @returns error, count pair where count is the number of elements skipped.
     * @sa BasicReader::Read
     */
    template <class Rep, class Period>
    [[nodiscard]] std::pair<std::error_code, size_t>
    Skip(size_t count, std::chrono::duration<Rep, Period> timeout) noexcept {
        return ReadExt([](T const&) noexcept {}, count, timeout);
    }
 
    /** @} */
 
    /**
     * @name Modifiers
     * @{
     */
    /**
     * Synchronizes internal state from shared memory.
     *
     * This is mainly useful to explicitly synchronize state to update non-synchronizing methods
     * like:
     * - BasicReader::Size()
     * - BasicReader::NumAvailable()
     *
     * @note Synchronize should not be invoked for BasicReader::Read as it performs synchronization
     * as a side-effect of reading.
     *
     * @acquires
     */
    constexpr void Synchronize() noexcept {
        m_ctl.AtomicLoadState();
    }
 
    /**
     * Reset internal reader state to synchronize with writer and recover from
     * ipcq::Error::InconsistentState.
     *
     * To reset successfully the queue cannot be empty, as the reader must get state information
     * from the queue to know what to expect next. In this case it returns ipcq::error::WouldBlock
     * instead of waiting for data. Caller can then make the choice to assume the queue is in its
     * initial state. If it was not the next read operation will fail with
     * ipcq::Error::InconsistentState.
     *
     * @post On success the next element to read will be the next element to be written.
     *
     * @acquires
     *
     * @param keep Number of samples to keep as *unread*. Default is to keep 0 which is to say the
     * next element to read is the next written. @c keep is automatically truncated to number of
     * available elements.
     *
     * @returns 0 On success.
     * @returns ipcq::Error::Closed if queue is closed and @c keep is 0 (i.e. reading will not be
     * possible after Reset()).
     * @returns ipcq::Error::WouldBlock if queue is empty.
     */
    [[nodiscard]] std::error_code Reset(std::size_t keep = 0u) noexcept {
        m_ctl.AtomicLoadState();
 
        if (keep == 0u && m_ctl.IsClosed()) {
            return make_error_code(Error::Closed);
        }
 
        if (m_ctl.IsEmpty()) {
            // Note that empty does not mean that next element is 0. It could have been filled,
            // and then emptied.
            return make_error_code(Error::WouldBlock);
        }
        auto const distance_end = m_ctl.Size();
        auto const distance_newest = distance_end - 1u;
        keep = keep > distance_end ? distance_end : keep;
 
        // Update m_last_element to the currently last element - keep.
        auto it = m_ctl.Begin();
        // The expected counter is always derived from last element to reduce likelihood of reading
        // the counter value mid-update.
        m_state.next_counter =
            (it + distance_newest)->counter.load(std::memory_order_relaxed) - keep + 1u;
        if (keep > distance_newest) {
            // Set no last element so when waiting for next value Begin() is used.
            m_state.last_element.reset();
        } else {
            it += distance_newest - keep;
            m_state.last_element = it;
        }
 
        return {};
    }
    /** @} */
 
private:
    template <class Operation, class Rep, class Period>
    [[nodiscard]] std::pair<std::error_code, size_t>
    ReadExt(Operation op, size_t count, std::chrono::duration<Rep, Period> timeout) noexcept(
        std::is_nothrow_invocable_v<Operation, T const&>) {
        constexpr auto nothrow = std::is_nothrow_invocable_v<Operation, T const&>;
        // Rollback is only used when operation may throw.
        [[maybe_unused]] detail::ReaderRollback<nothrow, State> rollback;
        if constexpr (!nothrow) {
            rollback.state = m_state;
        }
 
        auto num_elements_copied{0u};
 
        auto [err, it] = AwaitNextIterator(timeout);
        if (err) {
            return {err, num_elements_copied};
        }
        IPCQ_ASSERT(&*it != nullptr);
 
        for (auto count_it = 0u; count_it < count; ++count_it) {
            if (it == m_ctl.End()) {
                return {{}, num_elements_copied};
            }
            if constexpr (nothrow) {
                op(it->element);
            } else if constexpr (std::is_invocable_v<Operation, T const&>) {
                try {
                    op(it->element);
                } catch (...) {
                    // Roll back and rethrow
                    m_state = rollback.state;
                    throw;
                }
            } else {
                static_assert(detail::always_false_v<Operation>,
                              "Operation must be invocable with `T const&`");
            }
 
            // Load counter value atomically
            auto counter = it->counter.load(std::memory_order_acquire);
            if (counter != m_state.next_counter) {
                // This may occur if reader missed an entire cycle of the queue such that the
                // iterator position is valid, but only because it was completely missed, for
                // example.
                return {make_error_code(Error::InconsistentState), num_elements_copied};
            }
            // Success, iterate to next element (but only if queue has > 1 elements
            if (Capacity() > 1) {
                m_state.last_element = it;
            }
 
            ++num_elements_copied;
            ++m_state.next_counter;
            ++it;
        }  // end for
 
        return {{}, num_elements_copied};
    }
};
 
/**
 * Convenience alias with reasonable default condition policy.
 *
 * @ingroup ipcq
 */
template <class T>
using Reader = BasicReader<T, detail::BoostConditionPolicy>;
 
}  // namespace ipcq
 
#endif