C++11实现一个cyclic barrier

在上文中使用计数器作为同步事件实现了latch，其实在多线程并发编程实践中，还有一种使用计数器作为同步事件的机制：Cyclic-Barrier，即循环屏障的意思。它指定了参与执行线程的数量，当某个线程运行到指定位置时就处于等待状态，只有当这些线程全部都到达该位置时，它们才能一起从等待状态中退出。它的实现原理也是以一个计数器作为同步变量，当某个线程到达指定的位置时，就让计数器的值减少1，如果结果不为0时，线程就进入等待状态，只有当计数器减到0时，即指定数量的线程全部到达了指定位置，此时所有在该位置等待的线程全部被唤醒，然后执行各自后续的流程。

下面是cyclic_barrier类的定义：

class cyclic_barrier {mutable std::mutex m;std::condition_variable cv;const int parties;volatile int count;volatile bool broken;volatile int cycle; // 第几轮循环 std::function<void(void)> callback;public:cyclic_barrier(int n);cyclic_barrier(int n, std::function<void(void)> f);~cyclic_barrier() {}cyclic_barrier(const cyclic_barrier&) = delete; cyclic_barrier& operator=(const cyclic_barrier&) = delete;void wait();int get_waitings() const;int get_parties() const;void cancel();
};

既然线程有需要等待唤醒、通知的机制，使用互斥量m和条件变量cv就是不二之选了，使用它们进行线程的等待和唤醒操作。
为了进行计数，还需要一个整型的数据成员count来存放计数器，每当某个线程进入同步点时，就让count减1，直到为0；因为要循环计数，还需要保存计数器的初始值parties，它是参与线程的个数，初始化后就不会修改了，使用const修饰。
回调函数callback用于当最后一个线程到达后唤醒其它线程前，所执行的操作。
此外还有用于标记中断的broken变量，线程被唤醒后会检查该值，判断是否是被提前中断了。

看一下各个成员函数的实现。
1、构造函数

cyclic_barrier::cyclic_barrier(int n) : parties(n), cycle(1), callback([](){}) {if (n <= 0) {throw std::string("invalid parameter!");}count = n;broken = false;
}cyclic_barrier::cyclic_barrier (int n, std::function<void(void)> f) : cyclic_barrier(n) {callback = move(f);
}

构造函数有两个，如果不需要回调函数的话，可以使用第一个构造函数来创建对象，它指定了一个缺省实现的回调函数。如果有自定义的回调函数，可以使用第二个构造函数来创建对象。参数n来指定参与者线程的数量，该参数必须大于0，否则直接抛出异常，构造失败。

2、wait函数

void cyclic_barrier::wait() {std::unique_lock<std::mutex> ul(m);int n = cycle;if (--count == 0) {callback(); // callback是在临界区内执行cv.notify_all();count = parties; // 复位计数器cycle++; // 下一轮开始 return; // 如果是最后一个到达的线程，显然条件满足了，就直接返回}// 新一轮开始或者被打断了，才会被唤醒while (n == cycle && !broken) {cv.wait(ul);}if (broken) {throw broken_exception();}
}

每当有一个线程调用了wait()，就让count减1，并检查count是否为0，如果不为0，说明还有别的线程没有到达，就进入等待状态；如果为0，说明这些线程全部到达，就先调用回调函数，然后唤醒其它所有等待中的线程。因为是循环barrier，在count为0时，重新设置为parties，等待下一轮的线程。

回调函数执行完之后再唤醒其它线程，这样，如果在回调函数中修改了共享变量，可以保证在其它线程唤醒之前修改完成，以保证共享变量的访问安全。

wait()函数不妨看作是一个屏障（barrier），线程到达此位置时，都不能越过它，只能处于等待状态，直到最后一个线程到达。既然是屏障，那么线程在wait()之前所做的操作，如果涉及到共享变量的话，它们对共享变量所作的修改，也就都happen-before这个屏障之前，从而当所有线程从wait()中唤醒后，进行后续操作时，如果访问这些共享变量，都是发生在这个屏障之后，从而保证了这些共享变量的整体happen-before语义。同样，当最后一个线程到达，调用回调函数也不会越过wait()，如果在回调函数中也有修改共享变量的操作，这个操作同样也happen-before于线程唤醒后的后续操作。

线程在等待过程中也可以被打断，也就是说，即使还没有规定数量的线程调用wait()，也可以强制中断它。程序中使用broken作为中断标记，中断后它被设置为true，线程被唤醒后，如果发现broken=true，则说明是被中断唤醒的，此时线程就抛出broken异常，通知调用者。

4、中断函数

void cyclic_barrier::cancel() {std::lock_guard<std::mutex> lg(m);broken = true;cv.notify_all();
}

如果想要提前中断barrier，通过设置中断标记broken为true，并唤醒全部处于等待中的线程即可，此时线程从wait()中被唤醒后会抛出broken_exception异常。

class broken_exception {};

3、其它函数

int cyclic_barrier::get_waitings() const {std::lock_guard<std::mutex> lg(m);return parties - count;
}int cyclic_barrier::get_parties() const {return parties;
}

get_waitings()用来查询处于等待中的线程个数，get_parties()用来查询参与者的线程数量，它们都是const成员函数。

与latch相比，它们有如下特点：
1、cyclic_barrier的计数器统计的是到达屏障点的线程数量，即计数器的初始值是参与者线程的个数；而latch的计数器是根据所要满足的条件而设置的的数量，和线程数量不一定相关。

2、cyclic_barrier的计数器减少到0后，它会被复位，重新设置为初始化时的值，可以被循环使用；而latch是一个一次性的事件，当计数器变为0之后，就不再使用了。

3、使用cyclic_barrier的线程进入等待和唤醒都是同一个函数：wait()，线程在执行过程中调用它时，相当于在该函数的调用位置处设置了一个屏障点，先到达的线程会在屏障点等待后到达的，只有在所有线程全部到达这个屏障点之后，它们才能同时越过这个屏障点；而latch一般是由两种不同性质的线程相互协作，使用了等待和通知机制，一种统一在某个位置等待（调用wait()），另一种通过检查计数器为0时来通知唤醒（调用countdown()）。

4、cyclic_barrier有回调函数，在唤醒所有线程之前，可以执行这个回调函数。

示例：
1、多个线程从同一个位置同时运行。

void test(cyclic_barrier & barrier) {std::cout << "ready:" << std::this_thread::get_id() << std::endl;try {// .. 线程运行的准备工作，准备完毕之后，等待同时启动！ barrier.wait(); // 等待同时启动 } catch (broken_exception &ex) {std::cout << "barrier broken!" << std::endl;}std::cout << "startup:" << std::this_thread::get_id() << std::endl;std::this_thread::sleep_for(std::chrono::seconds(1));std::cout << "finish:" << std::this_thread::get_id() << std::endl;
}int main() {cyclic_barrier barrier(11, []{std::cout << "reach point" << std::endl;});std::vector<std::thread> vec;for (int i=0; i<10; i++) {vec.emplace_back(test, std::ref(barrier));}std::this_thread::sleep_for(std::chrono::seconds(2));std::cout << "waiting thread:" << barrier.get_waitings() << std::endl;barrier.wait(); // 主线程最后到达，通知所有线程一起运行 for (auto &t : vec) {t.join();}
}

2、模拟一个团购场景，任意三个成员就可以组成一个团购机会，有10个成员可以组成3个团，剩余1个无法组团。当活动结束后，取消团购活动。

void test(cyclic_barrier& barrier, int time) {std::this_thread::sleep_for(time*std::chrono::seconds(1));try {barrier.wait(); // 等待组团 } catch (broken_exception &ex) {std::cout << "sorry! game over" << std::endl;return;}std::cout << "join member:" << std::this_thread::get_id() << std::endl; // 哪个成员加入组团
}int main() {int n = 0; // 第几次组团成功cyclic_barrier barrier(3, [&n]() { //  每有三个成员就可以进行团购 std::cout << "reach group: " << ++n << std::endl; });srand(time(nullptr));std::vector<std::thread> vec;for (int i=0; i<10; i++) {int times = rand() % 10; // 组员随机到达 std::cout << times << std::endl;vec.emplace_back(test, std::ref(barrier), times);}std::this_thread::sleep_for(std::chrono::seconds(10));std::cout << "waiting member:" << barrier.get_waitings() << std::endl;barrier.cancel(); // 活动结束，取消团购，没有组团成功的被中断 for (std::thread &t : vec) {t.join();}
}

barrier和latch都使用了计数器倒计数进行统计，它们的应用场景不太一样，为了便于区分它们，举两个生活中例子，可以体会一下它们的区别。
假如有5个好基友（即线程），商量好周末一块去风景区爬山，他们约定好周日上午在山脚下集合，不见不散。周日那天，如果第一个先到了(调用wait)，发现没有其他人到达（count>0），就只好等着(cv.wait)，第二个人到了之后，发现人还没有到齐，也只好等待，直到第5个人到达后，即所有5个人全部到齐了(count=0)，做完准备工作之后(调用回调函数)，就一起出发开始爬山（线程全部唤醒）。我们可以想象在山脚下立着一个栅栏（Barrier），如果它不放倒，人们无法翻越过去，而它被放倒的条件是，5个基友全部到达。如果这5个基友哪怕仅有一个还没有到达，它也不会被放倒，基友就被拦在外面，只有当5个基友全部到达之后，这个栅栏才会倒下，他们就可以进入山门了。这描述了应用barrier的场景。

如果换成另一种场景，这几个基友约定谁先来谁先爬，假设风景区大门的门闩（Latch）上共有2把锁（latch的数量），钥匙分别由门卫和值班经理保管，第一个基友来的比较早，此时大门还没有开，那么他就只能等这2个掌管钥匙的人员来开门：门卫来上班了，把他掌管的那把锁打开（倒计数减1），基友继续等待；当值班经理上班后打开了第二把锁之后（倒计数为0），这个基友进入大门开始爬山（从等待中被唤醒）。此后，其它基友陆续到达后，就无需等待了，因为所有的锁都已经打开了，他们可以直接进入景区爬山。这是应用latch的场景。

C++20中提供了类似的功能，详见barrier类。

C++11实现一个cyclic barrier相关推荐

11. 搭建一个完整的K8S集群
11. 搭建一个完整的Kubernetes集群 1. kubectl的命令遵循分类的原则(重点) 语法1: kubectl 动作类具体的对象例如: """ kube ...
趣味三角——第11章——一个著名的公式
目录 1. 著名无限积公式简述及证明 2. Jules Lissajous 和他的图形(Jules Lissajous and His Figures) 11章一个著名的公式 The prototy ...
c++11实现一个自动注册的工厂模式
实现动机最近项目中需要用到工厂模式,但是普通的工厂模式面临一个问题,每新增一个派生类,都需要在工厂中加一个case分支,这样就会频繁地修改工厂的代码,而且随着派生类越来越多,case分支也逐渐增多, ...
C++11实现一个自动注册的工厂
转自:https://www.cnblogs.com/qicosmos/p/5090159.html 实现动机工厂方法是最简单地创建派生类对象的方法,也是很常用的,工厂方法内部使用switch-ca ...
如果不大于指定整数n的3个素数之和仍为素数，则把这3个素数称为一个基于n的全素组。例如对于n=15,素数3,5,11之和3+5+11=17为素数，则3,5,11 称为一个基于15的全素组。定义所有基于
全素组题目 :如果不大于指定整数n的3个素数之和仍为素数,则把这3个素数称为一个基于n的全素组.例如对于n=15,素数3,5,11之和3+5+11=17为素数,则3,5,11 称为一个基于15的全素 ...
Windows 11，一个新功能，一场新屠杀！
6月24日,微软正式公布了新一代操作系统:Windows 11.这次的更新距离上一代操作系统Windows 10的发布,隔了有6年之久. 在新一代的操作系统中,包含了这些亮点: 采用了全新的UI设计. ...
JDK 11 还有一个处于计划阶段的 JEP：让其支持 TLS 1.3
开发四年只会写业务代码,分布式高并发都不会还做程序员? >>> JDK 11 最近有什么消息?我们不妨来看一下它的进展情况,包括最新的 JEP 提案. Java 的新版本发布计划 ...
浮沉11年 | 一个互联网老兵的自白书
前天应邀给dbaplus社区做了一次关于技术人职场管理路线的线上分享.整个分享约定是30分钟,结果讲了1小时. 不知不觉入行已经11年,借着这次整理的思路,今天也给大家分享一下我的经历: 01 毕业的 ...
[2021.7.9][11 使用C++11开发一个轻量级的IoC容器(工厂模式的应用及优化)] 11.4 通过Any和闭包来擦除类型和创建依赖的对象
11.2节的对象工厂只能创建指定接口类型的对象,原因是它依赖了一个类型固定的对象构造器std::function<T*()>,这个function作为对象构造器只能创建T类型的对象,不能创 ...

C++11实现一个cyclic barrier

C++11实现一个cyclic barrier相关推荐

最新文章

热门文章