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 
#include 
#include 
#include 
#include 
 
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';
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

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 
#include 
#include 
#include 
#include 
#include 
 
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);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

主要特点

线程同步： 提供了一种机制，可以阻塞一组线程，直到所有线程都达到了某个同步点。这有助于协调并发执行的线程，确保它们在特定点同步执行或释放。

可重用： 与 std::latch 不同，std::barrier 是可重用的。一旦一组线程被释放，栅栏可以被重新使用，允许线程再次聚集在同一点。

阶段性同步： std::barrier 可以分为多个阶段，每个阶段定义了一个同步点。线程可以在同步点前到达，并在需要时等待。这使得可以在多个同步点执行特定的操作。

可定制的同步操作： 提供了一个回调函数，被称为 CompletionFunction，可以在所有线程到达同步点后执行。这个函数可以用来执行一些特定的操作，例如修改共享数据结构或执行其他同步操作。

多种成员函数： std::barrier 提供了不同的成员函数，如 arrive、wait、arrive_and_wait 和 arrive_and_drop，使得线程可以以不同的方式到达同步点并执行不同的同步操作。

线程安全： 除了析构函数外，std::barrier 的成员函数的并发调用不会引入数据竞争。这意味着多个线程可以安全地调用 std::barrier 的成员函数，而不需要额外的同步措施。