barrier.hpp

Go to the documentation of this file.

/**
 * @defgroup ion_barriers Barriers
 * Provides ion::Barrier and ion::Latch
 *
 * @ingroup ion
 *
 * @file
 * @ingroup ion_barriers
 *
 * @brief ion::Barrier
 *
 * @copyright
 *   SPDX-FileCopyrightText: 2022-2024 European Southern Observatory (ESO)
 
 *   SPDX-License-Identifier: LGPL-3.0-only
 */
#ifndef ION_BARRIER_HPP
#define ION_BARRIER_HPP
#include <atomic>
#include <cassert>
#include <cstddef>
#include <cstdint>
#include <limits>
 
#include <ion/detail/barrier.hpp>
#include <ion/detail/macros.hpp>
 
namespace ion {
 
/**
 * Token from `ion::Barrier::Wait()`
 *
 * @see ion::Barrier
 *
 * @thread_compatible
 * @ingroup ion_barriers
 */
class ArrivalToken {
public:
    ArrivalToken(ArrivalToken&&) noexcept = default;
    ArrivalToken& operator=(ArrivalToken&&) noexcept = default;
 
protected:
    enum Expired : std::uint8_t {
        NotExpired,
        AlreadyExpired,
    };
    /**
     * Initalize with known last signal value and whether token has already expired.
     */
    inline explicit ArrivalToken(std::uint64_t last, Expired expired);
    template <class>
    friend class Barrier;
    std::uint64_t m_last;
    Expired m_expired;
};
 
/**
 * Reusable thread-coordination mechanism modelled after `std::barrier`.
 *
 * Modelled after C++20 std::barrier but does not perform system calls or yields and as such is more
 * suitable in a real-time environment. Notable differences:
 * - Methods are noexcept as no mutexes are involved.
 *
 * A Barrier phase consist of two states:
 * 1. *arrival*, when threads fetch and decrement the expected count
 * 2. *departure*, when all threads have arrived and will depart. The barrier is automatically
 *    reset to the next arrival phase.
 *
 * @tparam CompletionFunction no-except invocable function, invoked by *one* of the threads that
 * arrive at the barrier via `Arrive()`, `ArriveAndWait()` or `ArriveAndDrop()`.
 *
 * @memory_order{
 *   @synchronizes_with{All calls to `ArriveAndWait()` or `ArriveAndDrop()`,
 *   any calls to `ArriveAndWait()` or `Wait()`.}
 *
 *   The *CompletionFunction* @synchronizes_with calls to `ArriveAndWait()` or `Wait()` after
 *   it is executed. This means that no additional synchronization is necessary in the
 *   *CompletionFunction* w.r.t. to the threads that have arrived and is waiting for departure.
 * }
 * @thread_safe
 * @ingroup ion_barriers
 * @headerfile <> <ion/barrier.hpp>
 */
#if !defined(DOXYGEN)
template <class CompletionFunction = detail::NoOp>
class Barrier : detail::BarrierBase<CompletionFunction> {
#else
template <class CompletionFunction = *unspecified*>
class Barrier {
#endif
public:
    /** Construct new barrier with an expected count of @a expected.
     *
     * @pre @a expected <= Barrier::Max()
     *
     * @param expected Expected number of arrivals.
     * @param func Completion function invoked once for every barrier phase.
     * phases *arrival* and *departure*.
     */
    constexpr explicit Barrier(std::ptrdiff_t expected,
                               CompletionFunction func = CompletionFunction()) noexcept;
 
    /**
     * @name Barrier requires stable address and is neither copyable or movable.
     */
    /// @{
    Barrier(Barrier const&) = delete;
    Barrier& operator=(Barrier const&) = delete;
    /// @}
 
    /**
     * Arrives at barrier, possibly executing completion function and returns token.
     *
     * @pre @a n must be > 0 and <= the expected count for the current barrier phase.
     *
     * Constructs an ArrivalToken object associated with the phase synchronization point for the
     * current phase. Then, decrements the expected count by n.
     *
     * This function executes atomically. The call to this function strongly happens-before the
     * start of the phase completion step for the current phase.
     *
     * The behavior is undefined if @a n is less than or equal to @a 0 or greater than the expected
     * count for the current barrier phase.
     *
     * @param n the value by which the expected count will be decreased.
     * @return ArrivalToken associated with current phase.
     * @sa Wait()
     * @sa ArriveAndWait()
     */
    [[nodiscard]] auto Arrive(std::ptrdiff_t n = 1) noexcept -> ArrivalToken;
 
    /**
     * Block at the barrier synchronization point associated with @a token.
     *
     * If arrival is associated with the phase synchronization point for the current phase of *this,
     * blocks at the synchronization point associated with arrival until the phase completion step
     * of the synchronization point's phase is run.
     *
     * Otherwise, if arrival is associated with the phase synchronization point for the immediately
     * preceding phase of *this, returns immediately. Otherwise, i.e. if arrival is associated with
     * the phase synchronization point for an earlier phase of *this or any phase of a barrier
     * object other than *this, the behavior is undefined.
     *
     * @param token Arrival token from previous call to `Arrive()`.
     * @sa Arrive()
     */
    void Wait(ArrivalToken&& token) const noexcept;
 
    /**
     * Equivalent to `barrier.Wait(barrier.Arrive(n))`
     *
     * @param n the value by which the expected count will be decreased.
     */
    void ArriveAndWait(std::ptrdiff_t n = 1) noexcept;
 
    /**
     * Decrements both the initial expected count for subsequent phases and the expected count for
     * current phase by one.
     *
     * @pre Expected count for current barrier phase must be >= 1.
     *
     * Unlike ArriveAndWait it will not wait.
     */
    void ArriveAndDrop() noexcept;
 
    /**
     * @return Maximum possible value for expected count.
     */
    static constexpr auto Max() noexcept -> std::ptrdiff_t;
 
private:
    ION_FORCE_INLINE inline void Signal() noexcept;
 
    /**
     * Barrier expected number of arrivals.
     * Must be `acquired` when resetting m_remaining.
     */
    std::atomic<std::ptrdiff_t> m_expected;
 
    /**
     * Remaining number of arrivals in current iteration.
     */
    std::atomic<std::ptrdiff_t> m_remaining;
 
    /**
     * Signal to wait on.
     * @note will eventually overflow and reset to 0.
     */
    alignas(ION_DESTRUCTIVE_INTERFERENCE_SIZE) std::atomic<std::uint64_t> m_signal;
};
 
/**
 * Controls barrier completion function invocation policy.
 *
 * @ingroup ion_barriers
 */
enum class InvokePolicy {
    /**
     * Completion function invocation is allowed.
     */
    Allowed,
 
    /**
     * Completion function invocation is not allowed.
     */
    Disallowed,
};
 
/**
 * Reusable thread-coordination mechanism similar to ion::Barrier but allows control where the
 * completion function is invoked.
 *
 * @tparam CompletionFunction no-except invocable function, invoked by *one* of the threads that
 * arrive at the barrier via `ArriveAndWait()` and specified policy InvokePolicy::Allowed.
 *
 * @memory_order{
 *   @synchronizes_with{All calls to `ArriveAndWait()`, any calls to `ArriveAndWait()`.}
 *
 *   The *CompletionFunction* @synchronizes_with calls to `ArriveAndWait()` after
 *   it is executed. This means that no additional synchronization is necessary in the
 *   *CompletionFunction* w.r.t. to the threads that have arrived and is waiting for departure.
 * }
 * @thread_safe
 * @ingroup ion_barriers
 * @headerfile <> <ion/barrier.hpp>
 */
template <class CompletionFunction = detail::NoOp>
class FlexBarrier : detail::BarrierBase<CompletionFunction> {
public:
    /** Construct new barrier with an expected count of @a expected.
     *
     * @pre @a expected <= Barrier::Max()
     *
     * @param expected Expected number of arrivals.
     * @param func Completion function invoked at most once for every barrier phase.
     * phases *arrival* and *departure*.
     */
    constexpr explicit FlexBarrier(std::uint32_t expected,
                                   CompletionFunction func = CompletionFunction()) noexcept;
 
    /**
     * @name FlexBarrier requires stable address and is neither copyable or movable.
     */
    /// @{
    FlexBarrier(FlexBarrier const&) = delete;
    FlexBarrier& operator=(FlexBarrier const&) = delete;
    /// @}
 
    /**
     * Arrives at barrier, waits until number of arrivals reaches the expected number and depending
     * on @a policy it may execute the completion function.
     *
     * @pre @a n must be > 0 and <= the expected count for the current barrier phase.
     *
     * Decrements the expected count by @a n and iff policy == InvokePolicy::Allowed then it may
     * also invoke the completion function.
     *
     * This function executes atomically. The call to this function strongly happens-before the
     * start of the phase completion step for the current phase.
     *
     * The behavior is undefined if @a n is less than or equal to @a 0 or greater than the expected
     * count for the current barrier phase.
     *
     * @param policy Policy indicating whether the completion function is allowed to be invoked.
     * @param n the value by which the expected count will be decreased.
     * @param wait function invoked repeatedly while waiting for barrier to be lifted. The
     * wait function is invoked without any synchronization guaranteees and may or may not be
     * invoked at all if calling thread does not have to wait. If it is desired that all threads
     * leave barrier at the same time, user should take special care to not introduce any latency in
     * `WaitFunction` such that the thread may lag. Generally this feature should be avoided if
     * possible.
     *
     * @tparam WaitFunction no-except invocable function with signature `void() noexcept`. Default
     * type is a no-op. See @a wait argument for details.
     */
    template <class WaitFunction = detail::NoOp>
    void ArriveAndWait(InvokePolicy policy,
                       std::uint32_t n = 1u,
                       WaitFunction&& wait = WaitFunction()) noexcept;
 
    /**
     * @return Maximum possible value for expected count.
     */
    static constexpr auto Max() noexcept -> std::uint32_t;
 
private:
    static ION_FORCE_INLINE constexpr auto
    MakeRemaining(std::uint64_t expected) noexcept -> std::uint64_t;
    template <class WaitFunction>
    void Wait(std::uint64_t last, WaitFunction&& wait) const noexcept;
    inline void Signal(bool invoke) noexcept;
    template <class WaitFunction>
    inline void AwaitArrivals(WaitFunction&& wait) noexcept;
 
    /**
     * Remaining number of arrivals in current iteration.
     *
     * @note Low 32 bits are used to indicate remaining barrier arrival count.
     * The high 32 bits are initially set to all ones and are subtracted once per call to Arrive*
     * iff callbacks are allowed.
     */
    std::atomic<std::uint64_t> m_remaining;
 
    /**
     * Barrier expected number of arrivals.
     * Must be `acquired` when resetting m_remaining.
     */
    std::uint32_t const m_expected;
 
    /**
     * Signal to wait on.
     * @note will eventually overflow and reset to 0.
     */
    alignas(ION_DESTRUCTIVE_INTERFERENCE_SIZE) std::atomic<std::uint64_t> m_signal;
};
 
template <class CompletionFunction>
constexpr Barrier<CompletionFunction>::Barrier(std::ptrdiff_t expected,
                                               CompletionFunction func) noexcept
    : detail::BarrierBase<CompletionFunction>(std::move(func))
    , m_expected(expected)
    , m_remaining(expected)
    , m_signal(0) {
    assert(expected >= 0 && expected <= Max());
}
 
template <class CompletionFunction>
[[nodiscard]] auto Barrier<CompletionFunction>::Arrive(std::ptrdiff_t n) noexcept -> ArrivalToken {
    // Subtract current slot with `n`
    // Iff this thread is the one that triggers the barrier signal condition the following applies:
    // 1. Completion function is invoked.
    // 2. Signal is emitted (to unblock other threads).
 
    // We must read the current signal state *before* we read m_remaining, to be able to guarantee
    // that the ArrivalToken contains an accurate value (the following write-release prevents
    // reordering).
    auto signal = m_signal.load(std::memory_order_relaxed);
 
    // We use release as it needs release-acquire synchronization with the thread that will end up
    // triggering the barrier phase transition.
    auto remaining_pre_sub = m_remaining.fetch_sub(n, std::memory_order_release);
    assert(remaining_pre_sub >= n);
    if (remaining_pre_sub == n) {
        // This thread was the last to arrive and must trigger the transition from *arrival* to
        // *departure* phase. This includes invoking completion function and signalling the other
        // threads. This is done in Signal().
        //
        // note: synchronize release from waiting threads using acquire semantics
        std::atomic_thread_fence(std::memory_order_acquire);
        ION_ANNOTATE_ACQUIRE(&m_remaining);
 
        Signal();
        // value of signal doesn't matter in this case.
        return ArrivalToken(signal, ArrivalToken::AlreadyExpired);
    } else {
        // This wasn't the thread that triggered the barrier
        // note: We cannot read *m_signal* here as it might have already been signalled.
        return ArrivalToken(signal, ArrivalToken::NotExpired);
    }
}
 
template <class CompletionFunction>
void Barrier<CompletionFunction>::Wait(ArrivalToken&& token) const noexcept {
    if (token.m_expired == ArrivalToken::AlreadyExpired) {
        // ArrivalToken was already recorded as expired when created in previous call to arrive.
        return;
    }
    // Spinlock waiting for signal to change from current value
    while (true) {
        auto sig = m_signal.load(std::memory_order_relaxed);
        if (sig != token.m_last) {
            // Signal changed -> synchronize with completion handler and return
            std::atomic_thread_fence(std::memory_order_acquire);
            ION_ANNOTATE_ACQUIRE(const_cast<void*>(static_cast<void const*>(&m_signal)));
            return;
        }
        ION_PAUSE();
    }
}
 
template <class CompletionFunction>
constexpr auto Barrier<CompletionFunction>::Max() noexcept -> std::ptrdiff_t {
    return std::numeric_limits<std::ptrdiff_t>::max();
}
 
ArrivalToken::ArrivalToken(std::uint64_t last, Expired expired) : m_last(last), m_expired(expired) {
}
 
template <class CompletionFunction>
void Barrier<CompletionFunction>::ArriveAndWait(std::ptrdiff_t n) noexcept {
    Wait(Arrive(n));
}
 
template <class CompletionFunction>
void Barrier<CompletionFunction>::ArriveAndDrop() noexcept {
    // Subtract expected *and* remaining with `1`
    // Iff this thread is the one that triggers the barrier signal condition the following applied:
    // 1. Completion function is invoked.
    // 2. Signal is emitted.
 
    (void)m_expected.fetch_sub(1, std::memory_order_relaxed);
    // We use release as it needs release-acquire synchronization with the thread that will end up
    // triggering the barrier phase transition.
    auto remaining_pre_sub = m_remaining.fetch_sub(1, std::memory_order_release);
    assert(remaining_pre_sub >= 1);
    if (remaining_pre_sub == 1) {
        // This thread was the last to arrive and must trigger the transition from *arrival* to
        // *departure* phase. This includes invoking completion function and signalling the other
        // threads. This is done in Signal().
        //
        // note: synchronize release from waiting threads using acquire semantics
        std::atomic_thread_fence(std::memory_order_acquire);
        ION_ANNOTATE_ACQUIRE(&m_remaining);
        Signal();
    }
}
 
template <class CompletionFunction>
void Barrier<CompletionFunction>::Signal() noexcept {
    // *this* thread triggered condition so we must execute handler
    if constexpr (!detail::IsNoOp<CompletionFunction>) {
        this->Complete();
        // note: synchronize-with waiting threads using release semantics happens with the
        // signalling as the last thing in this function.
    }
 
    constexpr auto acquire_order = []() -> std::memory_order {
        if constexpr (!detail::IsNoOp<CompletionFunction>) {
            // As we just acquired above m_expected does not need any memory barrier at all.
            return std::memory_order_relaxed;
        } else {
            return std::memory_order_consume;
        }
    }();
 
    // Update value of m_remaining from m_expected using minimum synchronization
    m_remaining.store(m_expected.load(acquire_order), std::memory_order_relaxed);
 
    // Signal awaiters with memory release barrier for the ordering constraint
    (void)m_signal.fetch_add(1, std::memory_order_release);
}
 
template <class CompletionFunction>
constexpr FlexBarrier<CompletionFunction>::FlexBarrier(std::uint32_t expected,
                                                       CompletionFunction func) noexcept
    : detail::BarrierBase<CompletionFunction>(std::move(func))
    , m_remaining(MakeRemaining(expected))
    , m_expected(expected)
    , m_signal(0u) {
    assert(expected <= Max());
}
 
template <class CompletionFunction>
constexpr auto FlexBarrier<CompletionFunction>::Max() noexcept -> std::uint32_t {
    return std::numeric_limits<std::uint32_t>::max();
}
 
template <class CompletionFunction>
constexpr auto
FlexBarrier<CompletionFunction>::MakeRemaining(std::uint64_t expected) noexcept -> std::uint64_t {
    return 0xffff'ffff'0000'0000 | expected;
}
 
template <class CompletionFunction>
template <class WaitFunction>
void FlexBarrier<CompletionFunction>::ArriveAndWait(InvokePolicy mode,
                                                    std::uint32_t n,
                                                    WaitFunction&& wait) noexcept {
    // Subtract current slot with `n`
    // Iff this thread is the one that triggers the barrier signal condition the following applies:
    // 1. Completion function is invoked.
    // 2. Signal is emitted (to unblock other threads).
 
    // We must read the current signal state *before* we read m_remaining, to be able to guarantee
    // that the ArrivalToken contains an accurate value (the following write-release prevents
    // reordering).
    auto signal = m_signal.load(std::memory_order_relaxed);
 
    std::uint64_t const value = mode == InvokePolicy::Allowed
                                    ? static_cast<std::uint64_t>(n) + 0x1'0000'0000
                                    : static_cast<std::uint64_t>(n);
    // We use release as it needs release-acquire synchronization with the thread that will end up
    // triggering the barrier phase transition.
    auto const value_pre_sub = m_remaining.fetch_sub(value, std::memory_order_release);
    auto const remaining = (value_pre_sub & 0xffff'ffff) - n;
    auto const any_cf = value_pre_sub >> 32 != 0xffff'ffff;
    auto const first_cf = mode == InvokePolicy::Allowed && !any_cf;
 
    if (remaining == 0u) {
        // This thread was the last to arrive, the following possibilities exist:
        //
        // 1. This arrival was the first to allow CF to be invoked.
        // 2. Another arrival allowed CF to be invoked.
        // 3. No arrivals allowed CF to be invoked.
        //
        // Yields:
        // 1 => barrier is expiring now => signal with CF
        // 2 => wait for another arrival to signal.
        // 3 => barrier is expiring now => signal without CF.
        if (first_cf) {
            // 1:
            // note: synchronize release from waiting threads using acquire semantics
            std::atomic_thread_fence(std::memory_order_acquire);
            ION_ANNOTATE_ACQUIRE(&m_remaining);
            Signal(true);
        } else if (any_cf) {
            // 2:
            Wait(signal, std::forward<WaitFunction>(wait));
        } else {
            // 3:
            Signal(false);
        }
    } else {
        // This thread was not last to arrive, but if it was the first to allow CF to be invoked it
        // will have to wait to do so.
        //
        // The possibilities are:
        //
        // 1. Arrival does not allow CF to be invoked.
        // 2. Arrival allowes CF to be invoked but was not first.
        // 3. Arrival allowes CF to be invoked and was first.
        //
        // Yields:
        // 1 & 2 => barrier has not expired
        // 3 => barrier is expiring *later* but signalling will be done by this thread => wait and
        //   signal.
 
        // note: We cannot read *m_signal* here as it might have already been signalled.
        if (!first_cf) {
            // 1 & 2:
            Wait(signal, std::forward<WaitFunction>(wait));
        } else {
            // 3: Wait for barrier arrivals to reach zero.
            AwaitArrivals(std::forward<WaitFunction>(wait));
            // note: synchronize release from waiting threads using acquire semantics
            std::atomic_thread_fence(std::memory_order_acquire);
            ION_ANNOTATE_ACQUIRE(&m_remaining);
            Signal(true);
        }
    }
}
 
template <class CompletionFunction>
template <class WaitFunction>
void FlexBarrier<CompletionFunction>::Wait(std::uint64_t last,
                                           WaitFunction&& wait) const noexcept {
    // Spin waiting for signal to change from current value
    while (true) {
        auto sig = m_signal.load(std::memory_order_relaxed);
        if (ION_UNLIKELY(sig != last)) {
            // Signal changed -> synchronize with completion handler and return
            std::atomic_thread_fence(std::memory_order_acquire);
            ION_ANNOTATE_ACQUIRE(const_cast<void*>(static_cast<void const*>(&m_signal)));
            return;
        }
        detail::Invoke(wait);
        ION_PAUSE();
    }
}
 
template <class CompletionFunction>
void FlexBarrier<CompletionFunction>::Signal(bool invoke) noexcept {
    // *this* thread triggered condition so we must execute handler
    if constexpr (!detail::IsNoOp<CompletionFunction>) {
        if (invoke) {
            this->Complete();
        }
        // note: synchronize-with waiting threads using release semantics happens with the
        // signalling as the last thing in this function.
    }
 
    // Update value of m_remaining from m_expected using minimum synchronization
    m_remaining.store(MakeRemaining(m_expected), std::memory_order_relaxed);
 
    // Signal awaiters with memory release barrier for the ordering constraint
    (void)m_signal.fetch_add(1, std::memory_order_release);
}
 
template <class CompletionFunction>
template <class WaitFunction>
void FlexBarrier<CompletionFunction>::AwaitArrivals(WaitFunction&& wait) noexcept {
    for (auto r = m_remaining.load(std::memory_order_relaxed); ION_LIKELY((r & 0xffff'ffff) != 0);
        r = m_remaining.load(std::memory_order_relaxed)) {
        detail::Invoke(wait);
        ION_PAUSE();
    }
}
 
}  // namespace ion
#endif  // ION_BARRIER_HPP