Compiler Explorer

Source code

// Options -----------------------------------
// #define BEFORE // uncomment to see BEFORE
#define NDEBUG // uncomment to disable debug output

// Select barrier implementation:
    // std::barrier can be used in combination with #define BEFORE
    // #include <barrier> 
    // using std::barrier;
    
    // Reference barrier implementation
    //#include "barrier.hpp"

// Posible barrier implementation under P2588:
    // #include "barrier_p2588.hpp"

// Implementation with extensions from PXYZ
    #include "barrier_memory_order.hpp"

// Vary the number of producer and consumer threads
constexpr int nproducers = 2;
constexpr int nconsumers = 3;

// Options ----------------------------------

#include <thread>
#include <cassert>
#include <iostream>
#include <vector>
#include "utils.hpp"

int data[nproducers];
constexpr int max_count = 3;

struct F { const char* s = nullptr; void operator()() { if (s) dbg(s); } };
using bar = barrier<F>;

void tests();
void houston();

void produce(bar& producer, bar& consumer, int producer_id);
void consume(bar& producer, bar& consumer, int consumer_id);
void producer_consumer();

int main() {
    tests();
    std::cerr << "TESTS SUCCESS!" << std::endl;
    producer_consumer();
    houston();
    std::cerr << "EXAMPLE SUCCESS!" << std::endl;
    return 0;
}

void producer_consumer() {
    #ifdef BEFORE
        constexpr int expected_count = 2;
    #else // AFTER
        constexpr int expected_count = 1;
    #endif // BEFORE / AFTER
    bar producer(expected_count, F{"p: barrier flipped"});
    bar consumer(expected_count, F{"c: barrier flipped"});
    std::thread t([&] { produce(producer, consumer, 0); });
    consume(producer, consumer, 0);
    t.join();
}

#ifdef BEFORE

void produce(bar& producer, bar& consumer, int producer_id) {
    int count = 0;
    do {
        dbg("p id=", producer_id, "unblocked!, count=", count);
        data[producer_id] = count++;
        static_cast<void>(consumer.arrive());    
        if (count == max_count) {
            dbg("p returns");
            return;
        }
        dbg("p id=", producer_id, "waiting -> count=", count);
        producer.arrive_and_wait();
    } while(true);
}

void consume(bar& producer, bar& consumer, int consumer_id) {
    int count = 0;
    while(true) {
        dbg("c id=", consumer_id, "waiting -> count=", count);
        consumer.arrive_and_wait();
        dbg("c id=", consumer_id, "unblocked!, count=", count);
        if (consumer_id < nproducers) {
            assert(data[consumer_id] == count);
        }
        count++;
        static_cast<void>(producer.arrive());
        if (count == max_count) {
            dbg("c returns");
            return;
        }
    }
}

#else // AFTER

void produce(bar& producer, bar& consumer, int producer_id) {
    int count = 0;
    bool phase = false;
    do {
        dbg("p id=", producer_id, "unblocked!, count=", count);
        data[producer_id] = count++; 
        std::atomic_thread_fence(std::memory_order_release);    
        consumer.arrive_and_discard(1, std::memory_order_relaxed);     
        if (count == max_count) return;
        dbg("p id=", producer_id, "waiting -> count=", count);
        producer.wait_parity(phase, std::memory_order_relaxed);
        std::atomic_thread_fence(std::memory_order_acquire); 
        phase = !phase;
    } while(true);
}

void consume(bar& producer, bar& consumer, int consumer_id) {
    int count = 0;
    bool phase = false;
    while(true) {
        dbg("c id=", consumer_id, "waiting -> count=", count);
        consumer.wait_parity(phase, std::memory_order_relaxed);
        std::atomic_thread_fence(std::memory_order_acquire);
        dbg("c id=", consumer_id, "unblocked!, count=", count);
        phase = !phase;
        if (consumer_id < nproducers) {
            assert(data[consumer_id] == count);
        }
        count++;
        std::atomic_thread_fence(std::memory_order_release);
        producer.arrive_and_discard(1, std::memory_order_relaxed);
        if (count == max_count) return;
    }
}

#endif // BEFORE / AFTER

void houston() {
#ifdef BEFORE
    barrier barrier(2);
#else
    barrier barrier(1);
#endif
    std::thread apollo13([&barrier] {
        std::cout << "Swigert: Okay, Houston, we've had a problem here" << std::endl;
        #ifdef BEFORE
            static_cast<void>(barrier.arrive());
        #else
            barrier.arrive_and_discard();
        #endif
    });
    #ifdef BEFORE
        barrier.arrive_and_wait();
    #else
    bool phase_parity = false;
        barrier.wait_parity(phase_parity);
    #endif
    std::cout << "Lousma: This is Houston. Say again, please." << std::endl;
    apollo13.join();
}

void tests() {
#ifndef BEFORE
    {  // Initial barrier phase  
        bar barrier(1, F{"test0: barrier flipped"});
        std::thread t([&barrier] { barrier.arrive_and_discard(); });
        bool phase_parity = false;
        barrier.wait_parity(phase_parity);
        t.join();
    }
    {  // Initial barrier phase  
        bar barrier(1, F{"test1: barrier flipped"});
        std::atomic<int> wait_count = 0;
        int arrive_count = 5;

std::thread t([=,&barrier,&wait_count] { 
            for (int i = 0; i < arrive_count - 1; ++i) {
                barrier.arrive_and_discard(); 
                while(wait_count != i + 1);
            } 
            barrier.arrive_and_discard();
        });

bool phase_parity = false;
        for (int i = 0; i < arrive_count; ++i) {
            barrier.wait_parity(phase_parity);
            wait_count++;
            phase_parity = !phase_parity;
        }
        if (wait_count != arrive_count) {
            std::cerr << "wait_count=" << wait_count << " != " << arrive_count << std::endl;
            throw 42;
        }
        t.join();
    }
#endif // AFTER
    {  // Pipeline with multiple producer and consumers
    #ifdef BEFORE
        bar producer(nconsumers + nproducers, F{"test2: p barrier flipped"});
        bar consumer(nproducers + nconsumers, F{"test2: c barrier flipped"});
    #else
        bar producer(nconsumers, F{"test2: p barrier flipped"});
        bar consumer(nproducers, F{"test2: c barrier flipped"});
    #endif

std::vector<std::thread> threads;
        threads.reserve(nconsumers + nproducers);
        for (int t = 0; t < nproducers + nconsumers; ++t) {
            if (t < nproducers) threads.emplace_back([&,t] { produce(producer, consumer, t); });
            else threads.emplace_back([&,t] { consume(producer, consumer, t - nproducers); });
        }
        assert(threads.size() == nproducers + nconsumers);

for (int t = 0; t < nproducers + nconsumers; ++t) {
            threads[t].join();
        }
    }
    {  // Pipeline with multiple producer and consumers
        std::cerr << "test3 start..." << std::endl;
    #ifdef BEFORE
        bar producer(nconsumers+nproducers);
        bar consumer(nproducers+nconsumers);
    #else
        bar producer(nconsumers);
        bar consumer(nproducers);
    #endif

for (int t = 0; t < nproducers + nconsumers; ++t) {
            threads[t].join();
        }
        std::cerr << "...test3 ended!" << std::endl;
    }
}

Source code

//! This file contains a "naive" reference implementation
//! of std::barrier including the proposed APIs.
#pragma once
#include <cstddef>
#include <cassert>
#include <atomic>
#include <limits>
#include <utility>

namespace detail { 
using std::ptrdiff_t;
struct default_cf { constexpr void operator()() const noexcept { } };

template <typename CompletionF = default_cf>
struct barrier {
    struct arrival_token { int phase; };
    static constexpr ptrdiff_t max() noexcept;
    barrier(ptrdiff_t expected,
            CompletionF f = CompletionF());
    ~barrier();
    barrier(const barrier&) = delete;
    barrier& operator=(const barrier&) = delete;

[[nodiscard]] arrival_token arrive(ptrdiff_t update = 1);
    void wait(arrival_token&& arrival) const;

void arrive_and_wait();
    void arrive_and_drop();

private:
        [[no_unique_address]] CompletionF completion;
        std::atomic<int> initial;
        std::atomic<int> current;
        std::atomic<int> dropped;
        std::atomic<int> phase;
    public: // Extension proposed in PXYZW
        void wait_parity(bool phase) const;
        void arrive_and_discard(ptrdiff_t update = 1);
};

template <typename CF>
constexpr ptrdiff_t barrier<CF>::max() noexcept {
    return std::numeric_limits<int>::max() - 1;
}

template <typename CF>
barrier<CF>::barrier(ptrdiff_t e, CF f)
  : completion((assert(e >= 0 && e < max()), std::move(f)))
  , initial(e)
  , current(e)
  , dropped(0)
  , phase(0)
{}

template <typename CF>
barrier<CF>::~barrier() {}

template <typename CF>
typename barrier<CF>::arrival_token 
barrier<CF>::arrive(ptrdiff_t update) {
    // dbg("arrive", update);
    auto r = phase.load(std::memory_order_relaxed);
    if (current.fetch_add(-update, std::memory_order_release) == update) {
        std::atomic_thread_fence(std::memory_order_acquire);
        completion();
        auto d = dropped.load(std::memory_order_relaxed);
        dropped.store(0, std::memory_order_relaxed);
        auto o = initial.fetch_add(-d, std::memory_order_relaxed);
        current.store(o - d, std::memory_order_relaxed);
        phase.fetch_xor(1, std::memory_order_release);
        phase.notify_all();
    }
    return {r};
}

template <typename CF>
void barrier<CF>::wait(typename barrier<CF>::arrival_token&& t) const {
    // dbg("wait", phase.load(), t.phase);
    phase.wait(t.phase, std::memory_order_acquire);
}

template <typename CF>
void barrier<CF>::arrive_and_wait() {
    return wait(arrive(1));
}

template <typename CF>
void barrier<CF>::arrive_and_drop() {
    dropped.fetch_add(1, std::memory_order_relaxed);
    arrive(1);
}

template <typename CF>
void barrier<CF>::wait_parity(bool p) const {
    wait(barrier<CF>::arrival_token { .phase = p? 1 : 0 });
}

template <typename CF>
void barrier<CF>::arrive_and_discard(ptrdiff_t update) {
    static_cast<void>(arrive(update));
}

} // namespace detail

using detail::barrier;

Source code

//! This file contains a "naive" reference implementation
//! of std::barrier that is enabled by P2588, in which one 
//! of the threads that waits completes the barrier phase.
//!
//! If no thread waits on the barrier, the phase will 
//! never complete, and the CompletionFunction will not run.
#pragma once
#include <cstddef>
#include <cassert>
#include <atomic>
#include <limits>
#include <utility>

namespace detail { 
using std::ptrdiff_t;
struct default_cf { constexpr void operator()() const noexcept { } };

[[nodiscard]] arrival_token arrive(ptrdiff_t update = 1);
    void wait(arrival_token&& arrival) const;

void arrive_and_wait();
    void arrive_and_drop();

private:
        // Since wait APIs are "const", these fields must be "mutable":
        [[no_unique_address]] mutable CompletionF completion;
        mutable std::atomic<int> initial;
        mutable std::atomic<int> current;
        mutable std::atomic<int> dropped;
        mutable std::atomic<int> phase;
        mutable std::atomic_flag completer;
    public: // Extension proposed in PXYZW
        void wait_parity(bool phase) const;
        void arrive_and_discard(ptrdiff_t update = 1);
};

template <typename CF>
constexpr ptrdiff_t barrier<CF>::max() noexcept {
    return std::numeric_limits<int>::max() - 1;
}

template <typename CF>
barrier<CF>::barrier(ptrdiff_t e, CF f)
  : completion((assert(e >= 0 && e < max()), std::move(f)))
  , initial(e)
  , current(e)
  , dropped(0)
  , phase(false)
{}

template <typename CF>
barrier<CF>::~barrier() {}

template <typename CF>
void barrier<CF>::wait(typename barrier<CF>::arrival_token&& t) const {
    // dbg("wait", phase.load(), t.phase);
    // Exit early if the phase has flipped
    if (phase.load(std::memory_order_relaxed) != t.phase) return;

if (completer.test_and_set()) {
        // all threads but one just wait on the phase changing
        phase.wait(t.phase, std::memory_order_acquire);
        return;
    }

// One thread completes the phase when current becomes 0.
    int old = current.load(std::memory_order_relaxed);
    while (old != 0) {
        current.wait(old);
        old = current.load(std::memory_order_relaxed);
    }
        
    std::atomic_thread_fence(std::memory_order_acquire);
    completion();
    auto d = dropped.load(std::memory_order_relaxed);
    dropped.store(0, std::memory_order_relaxed);
    auto o = initial.fetch_add(-d, std::memory_order_relaxed);
    current.store(o - d, std::memory_order_relaxed);
    completer.clear();
    phase.fetch_xor(1, std::memory_order_release);
    phase.notify_all();
    return;
}

template <typename CF>
void barrier<CF>::arrive_and_wait() {
    return wait(arrive(1));
}

template <typename CF>
void barrier<CF>::arrive_and_drop() {
    dropped.fetch_add(1, std::memory_order_relaxed);
    arrive(1);
}

template <typename CF>
void barrier<CF>::wait_parity(bool p) const {
    wait(barrier<CF>::arrival_token { .phase = p });
}

template <typename CF>
void barrier<CF>::arrive_and_discard(ptrdiff_t update) {
    current.fetch_add(-update, std::memory_order_release); // FAR RMW suffices
    current.notify_one();
}

} // namespace detail

using detail::barrier;

Source code

//! This file contains an implementation of barrier
//! APIs extended with memory order
#pragma once
#include <cstddef>
#include <cassert>
#include <atomic>
#include <limits>
#include <utility>

namespace detail { 
using std::ptrdiff_t;
struct default_cf { constexpr void operator()() const noexcept { } };

[[nodiscard]] arrival_token arrive(ptrdiff_t update = 1, std::memory_order order = std::memory_order_release);
    void wait(arrival_token&& arrival, std::memory_order order = std::memory_order_acquire) const;

void arrive_and_wait(ptrdiff_t update = 1, std::memory_order order = std::memory_order_acq_rel);
    void arrive_and_drop(ptrdiff_t update = 1, std::memory_order order = std::memory_order_release);

private:
        [[no_unique_address]] CompletionF completion;
        std::atomic<int> initial;
        std::atomic<int> current;
        std::atomic<int> dropped;
        std::atomic<int> phase;
    public: // Extension proposed in PXYZW
        void wait_parity(bool phase, std::memory_order order = std::memory_order_acquire) const;
        void arrive_and_discard(ptrdiff_t update = 1, std::memory_order order = std::memory_order_release);
};

template <typename CF>
constexpr ptrdiff_t barrier<CF>::max() noexcept {
    return std::numeric_limits<int>::max() - 1;
}

template <typename CF>
barrier<CF>::barrier(ptrdiff_t e, CF f)
  : completion((assert(e >= 0 && e < max()), std::move(f)))
  , initial(e)
  , current(e)
  , dropped(0)
  , phase(0)
{}

template <typename CF>
barrier<CF>::~barrier() {}

template <typename CF>
typename barrier<CF>::arrival_token 
barrier<CF>::arrive(ptrdiff_t update, std::memory_order order) {
    assert(order == std::memory_order_relaxed || std::memory_order_release);
    // dbg("arrive", update);
    auto r = phase.load(std::memory_order_relaxed);
    if (current.fetch_add(-update, order) == update) {
        // If the ordering is relaxed, we don't stablish cumulativity between 
        // threads that arrived and thread that runs the completion function:
        if (order != std::memory_order_relaxed) {
            std::atomic_thread_fence(std::memory_order_acquire);
        }
        completion();
        auto d = dropped.load(std::memory_order_relaxed);
        dropped.store(0, std::memory_order_relaxed);
        auto o = initial.fetch_add(-d, std::memory_order_relaxed);
        current.store(o - d, std::memory_order_relaxed);
        phase.fetch_xor(1, order);
        phase.notify_all();
    }
    return {r};
}

template <typename CF>
void barrier<CF>::wait(typename barrier<CF>::arrival_token&& t, std::memory_order order) const {
    assert(order == std::memory_order_relaxed || std::memory_order_acquire);
    // dbg("wait", phase.load(), t.phase);
    phase.wait(t.phase, order);
}

template <typename CF>
void barrier<CF>::arrive_and_wait(ptrdiff_t update, std::memory_order order) {
    // If the order is acq_rel we map them to acq or rel accordingly:
    auto arrive_order = order == std::memory_order_acq_rel? std::memory_order_release : order;
    auto wait_order = order == std::memory_order_acq_rel? std::memory_order_acquire : order;
    return wait(arrive(update, arrive_order), wait_order);
}

template <typename CF>
void barrier<CF>::arrive_and_drop(ptrdiff_t update, std::memory_order order) {
    dropped.fetch_add(update, std::memory_order_relaxed);
    arrive(update, order);
}

template <typename CF>
void barrier<CF>::wait_parity(bool p, std::memory_order order) const {
    wait(barrier<CF>::arrival_token { .phase = p? 1 : 0 }, order);
}

template <typename CF>
void barrier<CF>::arrive_and_discard(ptrdiff_t update, std::memory_order order) {
    static_cast<void>(arrive(update, order));
}

} // namespace detail

using detail::barrier;