C++20并发编程之线程闩(std::latch)和线程卡(std::barrier)

std::latch

std::latch类是一种基于std::ptrdiff_t类型的倒计数器，可用于同步线程。计数器的值在创建时进行初始化。线程可以在 latch 上阻塞，直到计数器减少到零为止。无法增加或重置计数器，这使得 latch 成为一次性的屏障。
std::latch的成员函数的并发调用（除了析构函数）不会引入数据竞争。
与std::barrier不同，std::latch可以被参与的线程多次递减。

主要成员函数

count_down：这是一个非阻塞的成员函数，用于递减内部计数器。当一个线程完成了它在同步点的工作时，可以调用此函数来告诉std::barrier，一个线程已经到达。这个操作不会阻塞调用线程，因此线程可以继续执行其他任务。

try_wait：这个成员函数用于测试内部计数器是否已经达到零。如果计数器为零，表示所有线程都已经到达同步点，可以执行后续操作。这是一个非阻塞操作，因为它只是测试计数器的状态而不阻塞调用线程。

wait：这是一个阻塞的成员函数，用于使调用线程等待，直到内部计数器达到零。当所有线程都已经到达同步点，调用wait的线程将被解除阻塞，可以继续执行后续操作。

arrive_and_wait：这是一个组合操作，它先递减计数器，然后等待直到计数器达到零。调用线程会递减计数器，然后被阻塞，直到所有线程都到达同步点。这是一个常见的用法，用于确保所有线程在继续执行之前都已经完成了特定的工作。

代码示例

#include <functional>
#include <iostream>
#include <latch>
#include <string>
#include <thread>
 
struct Job
{
    const std::string name;
    std::string product{"not worked"};
    std::thread action{};
};
 
int main()
{
    Job jobs[]{{"Annika"}, {"Buru"}, {"Chuck"}};
 
    std::latch work_done{std::size(jobs)};
    std::latch start_clean_up{1};
 
    auto work = [&](Job& my_job)
    {
        my_job.product = my_job.name + " worked";
        work_done.count_down();
        start_clean_up.wait();
        my_job.product = my_job.name + " cleaned";
    };
 
    std::cout << "Work is starting... ";
    for (auto& job : jobs)
        job.action = std::thread{work, std::ref(job)};
 
    work_done.wait();
    std::cout << "done:\n";
    for (auto const& job : jobs)
        std::cout << "  " << job.product << '\n';
 
    std::cout << "Workers are cleaning up... ";
    start_clean_up.count_down();
    for (auto& job : jobs)
        job.action.join();
 
    std::cout << "done:\n";
    for (auto const& job : jobs)
        std::cout << "  " << job.product << '\n';
}

std::latch 的特点：

一次性的：std::latch 是一次性的，一旦计数器减至零，无法重新使用。如果需要可以多次使用的机制，可以考虑 std::barrier。

线程安全：std::latch 是线程安全的，多个线程可以同时调用 count_down 和 wait。

无超时等待：std::latch 的 wait 函数没有提供超时参数，如果需要超时等待，可以使用 std::barrier 或其他机制。

std::barrier

std::barrier类模板提供了一种线程协调机制，它阻塞了一个已知大小的线程组，直到该组中的所有线程都到达屏障为止。与std::latch不同，屏障是可重用的：一旦一组到达的线程被解除阻塞，就可以重新使用该屏障。与std::latch不同，屏障在解除线程阻塞之前执行一个可能为空的可调用对象。

屏障对象的生命周期由一个或多个阶段组成。每个阶段定义了一个阶段同步点，在该点等待的线程将被阻塞。线程可以到达屏障，但通过调用arrive可以延迟在阶段同步点上等待。这样的线程稍后可以通过调用wait在阶段同步点上阻塞。

主要成员函数

arrive: 这是 std::barrier 的一个成员函数，用于将线程到达栅栏并减少预期计数。当线程调用 arrive 时，预期计数会减少。一旦所有线程都到达这个栅栏，预期计数归零，栅栏会打开，允许所有线程继续执行。

wait: 这个成员函数使线程在阶段同步点阻塞，直到当前阶段的完成步骤运行。一旦线程调用 wait 并在同步点阻塞，它将一直等待直到当前阶段完成，然后才能继续执行。

arrive_and_wait: 这个成员函数结合了前两者的功能。它使线程到达栅栏并将预期计数减少一，然后在同步点阻塞，直到当前阶段完成。与简单的 arrive 不同，arrive_and_wait 在到达后会立即阻塞，直到当前阶段完成为止。

arrive_and_drop: 这个成员函数不仅减少当前阶段的预期计数，还会减少后续阶段的初始预期计数。这意味着它不仅影响当前阶段，还会影响未来阶段。这对于在某些情况下提前放弃等待是有用的，因为它不仅影响当前阶段的计数，还影响到后续的阶段。

代码示例

#include <barrier>
#include <iostream>
#include <string>
#include <syncstream>
#include <thread>
#include <vector>
 
int main()
{
    const auto workers = {"Anil", "Busara", "Carl"};
 
    auto on_completion = []() noexcept
    {
        // locking not needed here
        static auto phase =
            "... done\n"
            "Cleaning up...\n";
        std::cout << phase;
        phase = "... done\n";
    };
 
    std::barrier sync_point(std::ssize(workers), on_completion);
 
    auto work = [&](std::string name)
    {
        std::string product = "  " + name + " worked\n";
        std::osyncstream(std::cout) << product;  // ok, op<< call is atomic
        sync_point.arrive_and_wait();
 
        product = "  " + name + " cleaned\n";
        std::osyncstream(std::cout) << product;
        sync_point.arrive_and_wait();
    };
 
    std::cout << "Starting...\n";
    std::vector<std::jthread> threads;
    threads.reserve(std::size(workers));
    for (auto const& worker : workers)
        threads.emplace_back(work, worker);
}