C++11实现一个countdown latch

C++11实现一个countdown latch

考虑这样一个线程应用场景，有两个变量，它们初始化时较为耗时，一般需要两个线程分别进行初始化，然后主线程等它们初始化完之后，再对它们进行后续处理。由于工作线程是异步执行的，无法知道它们什么时候初始化完，通常情况下，只能等待它们运行结束后才能知道。

下面是示意性的方案代码：

  thread t1(...); // 初始化一个共享变量
  thread t2(...); // 初始化另一个共享变量
  ... // 主线程的其它业务逻辑
  // 等待t1、t2结束
  t1.join();
  t2.join();
  ... // 两个共享变量初始化完毕，主线程使用它们进行后续的业务逻辑

主线程需要等待工作线程 t1 和 t2 执行完，才能使用它们的执行结果进行后面的 *** 作，为此通过调用 t1.join() 和 t2.join() 来实现等待，当 t1 和 t2 线程退出时，调用 t1.join() 和 t2.join() 的主线程就会从阻塞态被唤醒，从而执行之后的业务流程处理。

这种方案有不便之处：
1、如果需要等待的事件很多，主线程需要调用很多个thread::join()，主线程需要关注到具体的工作线程，过于具体化，导致代码不易扩展。
2、只能通过工作线程运行结束，来通知主线程事件发生了，如果在事件发生之后，工作线程还要处理其它业务，主线程还要无条件地等待其它业务处理完后线程退出，无形中增加了事件通知的时延。
3、不能使用线程池，因为线程池中的线程一般是不会退出的，主线程也就无法调用join()来等待线程结束。

那么，有没有更好的解决方案呢？我们不妨看一个现实生活中的例子。
在日常生活中，我们经常会遇到这样场景，当要做一件事情时，需要先等待几个固定数目的事情完成了之后才能进行，如果这些事情还没有完成，只能等待，差一件也不行。比如，工厂中有存放原材料的仓库，仓库的大门共有三把锁，假设钥匙分别由仓管员、主管部门经理和值班经理保管，当一个（或几个）领料员去仓库领材料，发现仓库大门锁着，那么他们就只能等这三个掌管钥匙的人员来开门：如果主管部门经理来上班了，就把他负责的那把锁打开，然后去工作了，此时领料员只能继续等待，当值班经理来上班之后，把他负责的那把锁也打开，然后也去工作了，最后当仓管员上班后打开了最后一把锁之后，领料员才能进入仓库零材料。对领料员而言，他不关心有几位人员、都是谁来开锁，只要门最后被打开了就行。此后，如果再有别的工作人员来领材料时，就无需等待了，因为所有的锁都已经打开了，可以直接进入仓库。

在多线程并发编程中，也有类似的需求场景和解决方案。如果把门锁看作是门闩（Latch）的话，也就是并发编程中的一种常用的同步机制：Countdown Latch，即倒计数门闩的意思。它的原理是以一个计数器作为同步变量，线程可以减少计数器的值，直到变为0为止，线程可以把计数器为0作为一个事件来等待：如果计数器不为0时，需要这个事件的线程就只能处于等待状态中（正如领料员在等待相关人员来开锁），当其它线程在条件就绪时（正如仓管员等来上班了），就让计数器减1，当计数器减到0之后，等待中的线程就被唤醒。显然，计数器为0这个事件是个一次性事件，一旦计数器为0了，其它后来再需要这个事件的线程就不用等待了，可以继续运行（正如仓库门打开之后，后来的领料员就不用等待了）。

下面看一下countdown latch类的实现。
先分析一下类的组成结构，既然线程间需要等待、通知唤醒的机制，使用互斥量mutex和条件变量condition_variable就当仁不让了。为了进行计数，还需要一个整型数据成员来存放计数器，这些都可以作为数据成员封装在类中。此外，它还要提供相应的成员函数来完成相关功能，一般有wait()，用于等待计数为0时的事件，countdown()用于进行倒计数，此外还可以提供了查询当前计数器数值的函数get_count()和强制中断的函数interrupt()。

下面是类的定义：

class countdown_latch {
	volatile int count;
	std::mutex mut;
	std::condition_variable cv;

public: 
	countdown_latch(int count);
	~countdown_latch();
	countdown_latch(const countdown_latch &) = delete; 
	countdown_latch& operator=(const countdown_latch &) = delete;
	
	int get_count() {
		return count;
	}
	void wait();
	void countdown();
	void interrupt();
};

其中count是同步变量，作为计数器，是整数类型，因为要在线程间共享，所以要使用volatile修饰。在构造时要设置一个大于0的初始值，如果它的值为0，说明倒计数事件已完成，如果它的值为-1，则说明这个latch还没有倒计时到0，就被意外中断了。
互斥量mut和条件变量cv用来进行条件等待和唤醒用，因为countdown_latch不提供拷贝和移动语义，它们都是以值类型方式定义的。

下面看看各个成员函数:
1、构造、析构函数
构造函数初始化计数器，显然，计数器初始值不能是小于0的值，如果小于0，则构造时直接抛出异常。在析构时，如果还有正在等待的线程，抛出异常，这样可以直截了当地把问题暴露出来，便于查找定位。

countdown_latch::countdown_latch(int cnt) : count(cnt) {
	if (count < 0) {
		throw std::string("parameter error!");
	}
}

countdown_latch::~countdown_latch() {
	std::lock_guard lg(mut); 
	if (count > 0) { // 如果counte>0，说明还有正在等待的线程，抛出异常
		throw std::string("state error!");
	}
}

2、wait()和coutdown()函数，这两个函数一个用于等待，一个用于通知，二者通过条件变量cv组成了线程间的条件通知场景。在调用wait()时，如果同步变量count不等于0，则阻塞住，直到count等于0时被唤醒；而coutdown()用于倒计数，在调用它时，会让count减1，直到变为0时，就通过条件变量向wait()发送唤醒通知，实现方式都是condition_variable常见的典型用法。

void countdown_latch::wait() { 
	if (count > 0) { // 双检查法，如果count等于0，说明已经倒计数到0，不用等待了
		std::unique_lock ul(mut);
		cv.wait(ul, [this]{return count<= 0;});
		if (count == -1) { // 如果是被中断的，抛出异常 
			throw latch_broken_exception();
		}
	} 
}

void countdown_latch::countdown() {
	if (count > 0) { // 双检查法，第一次检查
		std::lock_guard lg(mut); 
		if (count > 0) { // 第二次检查在mutex保护范围内 
			--count;
			if (count == 0) { // 倒计数为0，唤醒等待中的线程
				cv.notify_all();
			}
		}
	}
}

因为当遇到count为0时，说明已经发送过通知了，此时，如果再调用coutdown()或wait()时，也不会有发送通知或等待的 *** 作了，因此使用了双检查锁的实现机制，以免每次调用都无条件的调用互斥锁的 *** 作。wait()被唤醒的条件有两种，一是count等于0了，二是收到了中断通知，它们都是一次性的事件，唤醒后检查count的值，如果等于-1，说明是被中断唤醒的，抛出异常。

这两个函数提供了release-acquire语义，如同mutex的lock *** 作和unlock *** 作那样。其实，在这里count是同步变量，要通过它建立起共享资源和它之间的内存 *** 作顺序，以保证线程之间共享变量的可见性。调用wait()函数相当于加锁 *** 作，当它从阻塞中唤醒后就如同获得了锁，后面对共享对象资源的读写访问，都不应该越过count *** 作的前面去，因此wait()要保证acquire语义。同样，调用countdown()函数相当于一个解锁 *** 作，当它调用完毕返回后，要保证在调用它之前对共享对象资源的修改都有没有越过count *** 作，跑到它的后面去，因此countdown()要保证release语义。因为mutex的加锁、解锁 *** 作本身就拥有这些语义，代码中无需专门编写指定相关语义代码。

当然，如果线程之间有共享变量要 *** 作，一般在编写应用代码时，都会使用同步锁进行保护，这种情况也保证了共享变量在读写线程之间的release-acquire语义。不过有时候，共享变量分别只在独立的写线程（调用countdown()的线程）中 *** 作，各个写线程之间没有共享这些变量，它们仅在写线程和读线程（调用wait()的线程）之间共享访问，可能不会使用同步锁进行保护，countdown()和wait()之间的release-acquire语义，保证了写和读 *** 作之间的可见性，下面测试用例1就是这样的一个例子。

3、interrupt()中断函数。下面考虑一下异常情况，拿文中开头的例子来说，如果有一天，仓库的门锁其中有一把坏了，使用钥匙无法打开，或者仓管员失踪了，永远等不到他那把钥匙了，那么可以使用暴力破坏掉它：把锁砸了。类似场景也能反映到线程的应用场景中，如果调用countdown()的其中一个线程崩溃了，导致count无法倒计数到0了；或者不想让线程继续等待count变为0了，怎样唤醒因调用wait()而处于阻塞中的线程？下面提供了一种中断机制来满足这个场景需要。

void countdown_latch::interrupt() { // 当中断时，如果还有正在waiting的线程，唤醒它们 
	if (count <= 0) { // 此处无需使用release语义，如果 count <= 0，说明已经被别的线程唤醒了，在那里会有一个release语义，如果count > 0，由下面的mutex在unlock时来保证release 
		return;
	}

	std::lock_guard lg(mut); // mutex本身就提供了release语义，此函数无需提供专门的release语义 
	if (count > 0) { // 当count = 0时，则说明已经倒计数到0了，不用发送中断通知了 
		count = -1;
	}
	cv.notify_all();
}

此外，为了说明wait()是被异常中断了，可以在被中断的同时抛出一个异常，下面是异常类的定义。

class latch_broken_exception {
};

interrupt()函数很简单，就是无条件的使用cv.notify_all()，让所有处于wait状态中的线程被唤醒，同时让同步变量count为-1，标志它是因为异常情况下产生的事件。当wait()函数从cv.wait()中返回时，如果发现count=-1，则抛出异常，通知调用方，发生了异常事件，调用方可以根据此情况做一些特殊的处理。

测试示例
1、一个工作线程初始化完共享变量后通知主线程

volatile int shared_var1;
volatile int shared_var2;
void foo1(countdown_latch& latch) {
	std::this_thread::sleep_for(1100ms); // 模拟初始化的耗时时间
	shared_var1 = 42;
	latch.countdown();
}

void bar1(countdown_latch& latch) {
	std::this_thread::sleep_for(1300ms); // 模拟初始化的耗时时间
	shared_var2 = 51;
	latch.countdown();
} 

// 两个线程，每个线程使用一个相互独立的共享变量，latch保证了写和读之间的happen-before关系 
int test0() {
	countdown_latch latch(2);
	cout << "ready for init:" << latch.get_count() << endl;
	thread t1(foo1, std::ref(latch)); // 让工作线程初始化共享变量，自己做别的工作
	thread t2(bar1, std::ref(latch));
	this_thread::sleep_for(1000ms); // 模拟主线程的处理其它业务的耗时时间
	latch.wait();
	cout << "result:" << shared_var1 << endl;
	cout << "result:" << shared_var2 << endl;
	t1.join();
	t2.join();
}

注意，示例中的共享变量shared_var1和shared_var2是整型类型，对它的读和写都是原子指令，它们分别由两个不同的工作线程分别 *** 作，没有写写之间的数据竞争，程序并没有使用互斥锁进行保护，由latch的release-acquire语义保证了线程对共享变量写读之间的happen-before关系，这样在写和读线程在运行过程中，访问共享变量时就不会有时间上的重叠，也就不存在数据竞争，保证了共享数据的安全。

2、让多个工作线程同时启动，它们处理完业务之后，分别调用countdown()，最终在主线程中等待所有工作线程完成。主线程启动了10个工作线程，只要有5个线程完成了工作，就算是任务完成，即只要处理速度最快的top5结果。

void foo(countdown_latch& start_latch, countdown_latch& finish_latch) {
	cout << "ready:" << this_thread::get_id() << endl;
	start_latch.wait();
	cout << "startup:" << this_thread::get_id() << endl;
	this_thread::sleep_for(2000ms); // 模拟处理业务花费的时间 
	finish_latch.countdown();
	cout << "finish:" << this_thread::get_id() << endl;
}

int test1() {
	countdown_latch start_latch(1);
	countdown_latch finish_latch(5);

	vector vec;
	for (int i=0; i<10; i++) {// worker线程
		vec.emplace_back(foo, std::ref(start_latch), std::ref(finish_latch));
	}

	this_thread::sleep_for(1000ms); // 模拟准备工作花费的时间
	cout << "startup all worker" << endl;
	start_latch.countdown();
	thread t6(foo, std::ref(start_latch), std::ref(finish_latch)); // t6不会阻塞，因为start_latch在前面已经启动了 

	finish_latch.wait();
	// ... 所有工作线程结束，汇总运行结果
	for (thread &t : vec) {
		t.join();
	}
	t6.join();
	cout << finish_latch.get_count() << endl;
}

3、测试异常中断

void bar(countdown_latch& latch) {
	cout << "ready:" << this_thread::get_id() << endl;
	try {
		latch.wait();
	} catch (latch_broken_exception exception) {
		cout << "broken!" << endl;
	}
}

int test2() {
	countdown_latch latch(5);

	vector vec;
	for (int i=0; i<4; i++) { // count初始值为5，但是只倒计数了4次。
		vec.emplace_back(bar, std::ref(latch));
	}

	this_thread::sleep_for(2000ms);
	latch.interrupt(); //  强制终止
	for (thread &t : vec) {
		t.join();
	}
}

countdown_latch类既不允许复制，也不允许移动。这和它的使用场景有关，countdown_latch既然用于线程间的同步 *** 作，它肯定要被多个线程同时共享使用，即每个线程使用的都是同一个对象实例，一般都是通过引用&或者指针*形式进行传参，被多个线程按引用形式使用，从上面的例子中也能看到这种方式。设想一下，如果countdown_latch提供了移动语义，一个latch正在被多个线程使用时，当其中一个线程把它移动到别的地方去了，那么，其它正在使用该latch的线程，用到的都是无效的latch，所包含的资源已经被转移了，再使用时就会发生错误，导致程序异常。所以，不允许countdown_latch拥有移动语义，是为程序安全性考虑的。

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