Daily bit(e) C++. std::barrier

Daily bit(e) C++ 24. Механизм синхронизации для введения фаз выполнения в нескольких потоках: std::barrier.

std::barrier – это примитив синхронизации из C++20, позволяющий создавать синхронизированные фазы выполнения в нескольких потоках.

std::barrier инициализируется счетчиком (количеством потоков). Когда поток достигает барьера, он может блокироваться до тех пор, пока не достигнут барьера или не отключатся все остальные потоки, уменьшая значение счетчика.

В отличие от std::latch (который предоставляет аналогичную функциональность), std::barrier является многоразовым и вызовет необязательную функцию завершения перед разблокировкой ожидающих потоков.

#include <barrier>
#include <vector>
#include <thread>
#include <algorithm>
#include <random>
#include <print>

// Барьер с функцией завершения, которая
// печатает информацию о фазе:
std::barrier phase(4,[id = 1] mutable {
    std::println("Phase {} complete.", id);
    id++;
});

std::vector<std::jthread> runners;
// Запускаем 4 потока:
std::generate_n(std::back_inserter(runners), 4, [&phase]{
    return std::jthread([&phase]{
        std::println("Running phase 1 for thread {}",
                     std::this_thread::get_id());

        std::this_thread::yield();
        // блокируем, пока все потоки не достигнут барьера
        phase.arrive_and_wait();

        std::println("Running phase 2 for thread {}",
                     std::this_thread::get_id());

        std::this_thread::yield();
        // блокируем, пока все потоки не достигнут барьера
        phase.arrive_and_wait();
    });
});
runners.clear(); // присоединяем потоки

// Это гарантированно напечатает:
// Running phase 1 for thread xxxxx (для каждого из четырех потоков)
// Phase 1 complete.
// Running phase 2 for thread xxxxx (для каждого из четырех потоков)
// Phase 2 complete.


// Барьер без пользовательской функции завершения:
std::barrier other(5);
std::vector<std::jthread> runners2;
// Запускаем 5 потоков:
std::generate_n(std::back_inserter(runners2), 5, [&other]{
    return std::jthread([&other]{
        // Используем хеш id потока в качестве начального значения,
        // чтобы получить различное поведения для каждого потока
        size_t seed = 
            std::hash<std::thread::id>{}(std::this_thread::get_id());
        std::mt19937 gen(seed);

        // 30% шанс получить true, 70% - false
        std::bernoulli_distribution done(0.3); 

        int id = 1;
        while (true) 
        {
            std::println("Running phase {} for thread {}", 
                         id, std::this_thread::get_id());

            std::this_thread::yield();
            if (done(gen)) // 30% шанс остановки потока
            {
                // уменьшаем значение счетчика барьера,
                // чтобы он ожидал n-1 потоков
                other.arrive_and_drop();
                return;
            }

            // Иначе блокируем, пока все (еще запущенные)
            // потоки не достигнут барьера
            other.arrive_and_wait();
            ++id;
        }
    });
});
// Это напечатает последовательность фаз, при этом количество
// потоков в каждой фазе будет случайным образом уменьшаться.

Пример на Compiler Explorer

Теги