谈谈线程的中断退出

线程启动后可以一直运行，直到线程函数运行完毕，即RTC（Run To Complete：一站到底？）模式。这种模式非常容易实现，启动线程后调用join()等待线程执行完成就行了，或者调用detach()让线程在后台运行，线程运行完之后就悄无声息的退出了，不带走一片云彩（这种模式有一种形象的名称是：Fire And Forget）。

线程在运行过程中也可以被中断，提前结束线程运行，即可中断模式。相比RTC模式，该模式就不容易实现了，因为需要多线程之间的通信，涉及到了线程安全问题。如何让线程能够安全地中断退出呢？

如何实现

肯定不能像杀死一个进程那样，比如类似通过kill -9命令直接去杀死进程。因为进程被杀死之后还有操作系统来收拾残局，也就是进程杀死后没有释放的资源，最后都会被操作系统统一回收，而线程如果也像杀死进程那样无条件的退出，不做必要的退出准备，可能会造成资源泄露，比如一个线程分配了内存、申请了互斥锁，它们都还没有释放线程就被杀死了，那么这些资源谁来回收呢？显然这样中断退出是行不通的。

大多数线程中断退出时，都是采用协作式的方式，简单的说就是先给需要退出的线程发送一个中断请求，通知线程做好退出前的准备工作，然后让线程函数提前返回。通常需要线程函数的配合，在一些程序关键点的位置编写逻辑来检查是否需要退出。常用的方法是使用一个共享变量标记，比如atomic_flag或者atomic类型的共享变量，当需要线程退出时，就设置这个共享变量为一个约定好的值，比如true，线程函数在运行过程中发现这个变量是true之后，就中断后面的执行流程，做完收尾工作后就从函数中返回。

中断线程操作是一个比较重要、经常使用的操作。因此，有的线程库就提供了相关操作接口，以方便用户使用，比如C++20的std::jthread就提供了stop_token机制来通知线程退出，使用时程序员要在线程函数中调用stop_token::stop_requested()接口来检查是否需要退出线程；而pthread库使用了另一种中断机制，它提供了线程取消接口，如pthread_cancel()、pthread_testcancel()等函数，在中断线程时，通过pthread_cancel()来通知线程退出，程序员不需要编写相关的代码来检查是否退出线程，而是由一些取消点函数自己响应中断请求。

C++11中的线程库并没有提供中断线程的功能，不过我们可以自己实现。

例子

下面看一个最简单的实现方案：

// 线程函数
void foo(bool &stop_token, std::string name) {int count = 0;while (!stop_token) {cout << name << ": " << count++ << endl;this_thread::sleep_for(chrono::seconds(1));}
}int main() {bool stop_token = false; thread thr(foo, ref(stop_token), "test");this_thread::sleep_for(chrono::seconds(5));stop_token = true;thr.join();
}

stop_token是线程中断退出的共享标识变量，它被目标线程和管理线程之间共享，通常由管理线程来设置标识变量，而目标线程检查来标识变量。目标线程检查到stop_token的值是true之后，说明不需要它继续运行了，就从循环中退出，结束线程函数的执行。

这个例子还是有不安全因素的，中断标识变量stop_token只是一个普通的bool型共享变量，它没有使用volatile修饰，而它在线程函数中又是作为循环变量使用的，编译器有可能在对它读取访问时进行优化，把它缓存在寄存器中。这样，当管理线程设置stop_token = true时，只是把在内存中的stop_token值设置为true了，而目标线程thr中缓存在寄存器中的值并没有被修改。因此，线程函数可能会一直处于死循环中，导致线程无法退出。

安全的例子

因此，为了安全，需要使用volatile修饰线程函数的参数stop_token：

void foo(volatile bool &stop_token, std::string name);

这样，在foo函数中每次访问stop_token时，都必须从内存中加载，当管理线程修改stop_token为true时，因为处理器缓存一致性协议的保证，thr线程读取stop_token时，最终肯定能得到它被管理线程修改后的值，此时为true，然后中断循环，从foo()函数中返回。

当然，就这个例子而言，这样实现线程中断是没有问题，因为stop_token只是一个独立读写变量（IRIW），仅仅是对它独立设置和读取值的操作，处理器的缓存一致性就能保证它在线程间的一致性。但是，如果线程在中断退出时又涉及到了其它共享变量的读写，stop_token此时就是一个起同步作用的共享变量了，它的读写访问就不是独立的了，也就不得不考虑它和其它共享变量之间读写操作的顺序性，此时就得需要内存一致性的保证了。

看下面在目标线程中断退出前和管理线程通信的例子：

int shared = 0;
// 线程函数
void foo(volatile bool &stop_token, string name) {int count = 0;while (!stop_token) {cout << name << ": " << count++ << endl;this_thread::sleep_for(chrono::milliseconds(200));}// 返回前读sharedcout << shared << endl;
}int main() {bool stop_token = false;thread thr(foo, ref(stop_token), "test");this_thread::sleep_for(chrono::seconds(1));shared = 42; // 中断前写sharedstop_token = true;thr.join();
}

添加了一个共享变量shared，当管理线程在请求目标线程中断退出前，对shared进行了写入操作，当目标线程收到中断请求后，在退出前，先打印shared的值，即对shared进行了读操作。那么目标线程肯定能看到管理线程写入的42吗？

不一定。

首先，编译器在编译时，可能会乱序，把stop_token = true;和shared = 42;这两个内存写操作乱序生产指令，有可能让stop_token先赋值为true，再让shared 赋值为42。显然目标线程在发现stop_token为true时，shared可能还是旧值0。即使编译器没有乱序，如果是运行在弱序的处理器中，因为内存“写-写”操作之间也有可能乱序执行，在执行时，还是有可能让stop_token先赋值为true，再让shared 赋值为42。同样在目标线程中，stop_token的读操作和shared的读操作之间也有可能在编译或者执行时乱序，导致读取到shared的旧值0。

因此，在此情况下还必须保证内存顺序一致性，通常的方法是使用原子量来定义stop_token ，即：

int shared = 0;
// 线程函数
void foo(atomic<bool> &stop_token, string name) {int count = 0;while (!stop_token) {cout << name << ": " << count++ << endl;this_thread::sleep_for(chrono::milliseconds(200));}// 返回前读sharedcout << shared << endl;
}int main() {atomic<bool> stop_token{false};thread thr(foo, ref(stop_token), "test");this_thread::sleep_for(chrono::seconds(1));shared = 42; // 中断前写sharedstop_token = true;thr.join();
}

当然，为了优化性能，可以在管理线程中写stop_token时，使用release内存序语义，在目标线程中读stop_token时，使用acquire内存序语义，让它们之间保证了acquire-release内存序同步语义。
即：

stop_token = true;   改为 stop_token.store(true, std::memory_order_release);
while (!stop_token) 改为：while (!stop_token.load(std::memory_order_acquire))

协作式机制

前面说过，线程中断退出时不是简单的杀死线程就可以了，一般都是协作式机制，即管理线程发送请求中断的信号，目标线程检查中断信号，发现需要中断时就准备退出。因为线程是在运行过程中被提前中断运行，不能无条件强行退出，退出之前也要做好善后工作，不能留下后遗症。

下面看一个线程函数，它的功能是对一批文件进行读取操作，在读文件过程中是可以随时中断退出的。

void read_file(atomic<bool> &stop_token) {char *buf = new char[4096];int fd = open(...);int r;do {r = read(fd, buf, 4096);for (int i=0; i<r; i++) {...对文件数据进行处理的逻辑}} while (r != -1);close(fd);delete[] buf;
}

那么应该怎么修改代码，以配合线程退出呢？

首先可以在每次外围循环前检查是否退出，即：

	do {if (stop_token) break; // Ar = read(fd, buf, 1024);...} while (r != -1);

从循环退出后，调用close()函数关闭文件和delete释放内存，然后退出函数，没有造成资源泄露，是安全的。

在对文件内容数据进行处理时，是CPU密集型的模式，有可能长时间运行。如果想要线程能够及时的退出，也得要在关键点检查中断标记。因此，可以在B处检查是否需要中断退出：

do {if (stop_token) break; // Ar = read(fd, buf, 4096);for (int i=0; i<r; i++) {if (stop_token) break; // B...对文件数据进行处理的逻辑}
} while (r != -1);

线程收到中断请求从B处退出内层循环后，又在A处退出外层循环，调用close和delete，释放资源，退出前的收尾工作也是安全的。

我们知道，IO操作和内存分配都是非常耗时的操作，如果希望线程能够尽快的中断退出，也应该在new和open操作之前进行检查。

void read_file(atomic<bool> &stop_token) {if (stop_token) return;char *buf = new char[4096];if (stop_token) goto DELETE;int fd = open(...);...close(fd);
DELETE:delete[] buf;
}

在new操作之前，函数还没有分配任何资源，可以直接返回，而在open之前检查stop_token，需要退出时，就得要先释放已经分配的内存了，因此使用goto语句跳转到delete[]语句处。

当然，也可以把打开文件和关闭文件都封装成RAII类，同时把分配的内存资源使用智能指针管理，这样可以在检查到需要中断退出时，通过throw异常的方式来退出函数，由C++运行时来保证RAII对象所持有资源的回收。

通过共享变量协作式线程中断退出，虽然实现简单，但也有缺点，那就是它的中断响应的及时性。如果目标线程被阻塞了，此时管理线程发出了中断请求，因为目标线程正处于阻塞中，无法检查中断标识，只能等待一段时间，从阻塞中唤醒后才有机会去检查中断标识。如果在请求中断的同时，也能够有办法让目标线程能从阻塞中唤醒是比较好的方式，比如，在Linux环境下可以考虑给目标线程发送SIGINT信号，如果目标线程处于阻塞中，可以唤醒它们来响应中断请求。

这些都是在编写线程中断退出时需要注意的地方，需要根据线程函数实现的具体业务场景，编写中断退出的收尾工作逻辑，确保线程安全退出。

C++20提供的线程中断退出机制，也是协作式的，需要靠目标线程配合来实现安全的退出流程。同时它也提供了有限的唤醒阻塞线程的方法，即等待条件变量时可以使用带有stop_token参数的wait()函数，它可以通过stop_token中断机制来唤醒等待中的条件变量，不过对于目标线程处于mutex互斥锁和IO操作的阻塞时，仍然需要程序员自己想办法。

而pthread线程库提供的线程取消机制就比上述方式要好用的多，对目标线程进行取消请求时，如果线程处于阻塞中，可以从中唤醒。因为按照要求，这些会导致线程阻塞的函数
，它们一般都是取消点函数，自己会响应中断请求。在中断线程时，通过调用pthread_cancel()来通知线程退出，程序员一般也不需要编写相关的代码来判断是否退出线程。当线程函数调用到某个取消点时，这个函数会对取消请求做出响应，并中断程序的执行，取消点是一组函数，通常都是引起线程阻塞的函数，比如open()、read()、write()、close()、usleep()、pause()、wait()、socket操作接口函数…等，遇到取消请求时，它们可以从阻塞中唤醒。而对于CPU密集型的场景，pthread库的线程取消机制就无能为力了，就得需要程序员参与了，通常是在某些安全关键点，调用pthread_testcancel()来响应pthread_cancel()的线程取消请求。

毒丸模式

前面说的是通用的线程中断方式，在一些特殊的场合也有专用的中断推出模式，就以前面文章的例子2来说，线程拥有一个消息循环，其它线程可以通过handler接口给它提交消息，后台looper线程通过消息循环队列来接收并处理这个消息。如果要中断这个线程，一般不使用前面介绍的方法，更常用的方法是通过发送一个表示退出的消息给这个线程。如果此时线程还有其它消息没有处理完，这个退出消息就放入消息队列中排队，因为消息队列是FIFO机制，也就是说目标线程在遇到这个退出消息之前，需要把已有的消息全部处理完，显然这应该是更合理的中断线程方式。这种方式也有一种形象的名称：毒丸（Poison Pill）退出模式，“毒丸”是指一种放在消息队列上的对象，它的含义是：当得到这个对象时，线程立即停止。

这也是线程之间的一种协作式退出机制：looper线程在收到毒丸对象之后，知道这是最后一个消息，以后不会再收到消息，它可以退出了。而且在提交“毒丸”对象之前提交的所有消息都已经被处理了，此时消息队列中没有消息了，显然这是由FIFO的消息队列保证的。而生产者线程在提交了“毒丸”对象之后，也将不会再提交任何消息对象。

这样，请求中断线程退出时，生产者线程提交”毒丸“消息：

message msg(message.POISON); // 创建“毒丸”对象
handler->put(msg); // 发送“毒丸”消息

looper线程的消息处理流程：

	virtual bool handle(const message &msg) override {if (msg.what == message.POISON) { // 收到“毒丸”// 返回false，表示线程不会收到后续的消息return false;} ...// 其它类型的消息处理逻辑return true;}

显然，“毒丸”退出模式是由生产者-消费者线程的工作模式决定的，即生产者线程提交的消息不应该被丢弃，looper线程保证把所有收到的消息都要进行处理。显然，这种模式不要求looper线程立即响应中断，允许线程的退出可以有很长的延迟。

当然，也可以要求尽可能的及时退出，这时就可以使用前面介绍的共享中断标识的机制了，即looper线程在每次从消息队列获取消息时，先检查这个中断标识，如果它被设置为true，就不再处理剩余的消息了。那么剩余的消息怎么处理呢？通常情况下，当looper线程发现需要中断退出时，就不再处理消息队列中剩余的消息了，而是在退出时把这些把剩余的消息返回给生产者（比如通过future-promise机制），让提交“毒丸”的线程自己决定如何处理，looper线程不决定这些剩余消息的命运，不会处理它们，更不能把它丢弃，万一中断线程的程序需要把这些消息发送给性能更好的机器节点去处理呢？这是在此中断模式下生产者和消费者线程的协作机制。