C现代方法（第28章）笔记——C1X新增的多线程和原子操作支持

2023-12-17 03:47:10

文章目录

第28章 C1X新增的多线程和原子操作支持

在现代计算机中，我们可以同时执行多个程序。比如，可以在Windows系统里打开多个应用程序，这样就可以一边听歌，一边处理电子表格。当然，能够这样做的前提是处理器越来越快、越来越强。

在计算机领域，进程（process）是一个常见的术语。一个程序被加载到计算机内部执行时，就成为一个进程。为了充分利用处理器的计算能力，一台计算机上通常会有多个进程同时运行。我们运行的每个应用程序都对应着一个进程。本书中的所有程序，运行时都是一个进程，或者说运行时都会被操作系统创建为一个独立的进程。

早期的进程在结构上比较简单，它们都按照单一流程执行。以C程序为例，当它作为一个进程被创建和启动后，先完成一些初始化工作，比如创建和初始化全局变量，然后调用它的第一个函数（通常是main函数）。如果在main函数内又调用了其他函数，则转入其他函数执行，并返回到调用点继续执行。从main函数返回后，整个进程就完成了一次执行，随后被撤销。进程的执行可能不是连续的，操作系统可能会在多个进程之间调度，让它们轮流执行，但这种调度和轮转对计算机用户来说可能是无法察觉的。

为了加快进程的执行速度，并缩短进程的执行时间，可以把进程进一步划分为若干可并行执行的部分，并把每一个可独立执行的部分叫作一个线程（thread）。实际上，在引入线程这个概念之后，即使一个进程还是按单一流程执行，其中不存在并行处理的部分，它也将包含一个线程。原先的进程变成一个容器，实际的工作交给线程来完成。换句话说，每个进程至少包含一个线程。如此一来，从某种意义上来说，原先的进程调度更像是在一大堆线程中间进行调度。

将进程划分为一个或者多个线程之后，如果处理器的利用率原本就不高，这将可以提高整个进程的执行速度。在一个多处理器（核）的系统中，甚至可以将线程指派给不同的处理器（核）来达到负载均衡的效果。

C语言一开始并不支持多线程，但这不能怪它，毕竟在那个时代，多线程并没有像今天这样流行。但是，后来的C89、C99也没有提供语言层面上的支持，这多少让人觉得有点奇怪。这并不是说C语言不能用来编写多线程应用程序，而是指C语言在语法和标准库的层面缺乏对多线程的支持。随着在C语言中加入线程支持的呼声越来越高，多线程支持最终被添加到2011年的C语言版本（C11）中，作为一个可选的特性。

多线程提高了程序的性能，但如果线程之间需要通信和同步，这也带来了问题，而且这些问题随着多处理器（核）技术的应用而越发严重。所以，线程库中也包括了互斥锁和条件变量。同时，在C11中也引入了原子类型和原子操作以支持锁无关的编程。

在本章中，我们先学习如何创建线程（28.1.2节），然后讨论因多线程而引发的数据竞争问题（28.1.3节），并由此引出原子类型和原子操作（28.2节）。即使是引入了原子操作，也不能完全解决数据访问的同步问题，为此还将学习如何用内存屏障技术来同步线程间的内存访问（28.2.8节）。

28.1 <threads.h>: 多线程执行支持(C1X)

从C11开始，C语言提供了头<threads.h>。这个头包含了若干宏定义、类型定义、枚举常量以及函数声明，用来对多线程应用程序的编写和执行提供支持。

但是，以上特性是可选的，而不是强制性的，如果C实现不打算支持以上特性，则它必须定义一个宏__STDC_NO_THREADS__，表明它未提供头<threads.h>，从而也不包括上述宏定义、类型定义和函数。因此，在决定使用C语言的多线程特性前，程序员应当先用预处理指令判断这个宏是否存在：

# ifdef __STDC_NO_THREADS__ 
# error "Not support multi-theads. " 
# endif

请注意!!可能和你预期的相反：如果这个宏存在，实际上表明头<threads.h>是不存在的且不可用的。不存在这个宏，才是好消息。

实际上，在C标准引入多线性的特性之前，很多C实现本身就已经支持多线程了。比如GCC，它早就支持多线程了，只不过它有自己的线程库，提供的头是<pthreads.h>，里面的函数也和C标准不一样。

本章中主要介绍C标准的多线程特性，为了编译本章中的代码，可以使用pelles C编译器，这个编译器已经完整地实现了C标准的多线程特性。（这里推荐C/C++在线编译器&调试器，如果本地编译器不支持多线程或者原子操作，可以方便地在上面运行调试本章的代码)

28.1.1 线程启动函数

线程本质上是一个程序的组成部分。这个程序被创建为一个进程，而程序的特定部分则被创建为线程。这些特定的部分，也就是作为线程来执行的部分，形式上和普通的函数没有任何区别，但它们具有特定的参数和返回类型。

在程序中，作为线程来执行的函数叫作线程函数，或者线程启动函数，因为线程是从这个函数开始执行的。进一步来讲，这也暗示着在线程启动函数内可以调用其他函数。按照标准的要求，线程启动函数的原型必须是下面这样的（函数的名字无关紧要，可以用你自己喜欢的名字替换这里的thread_function）：

int thread_function (void *);

在创建一个线程时，可以给它传递一个指向void类型的指针作为参数。这个参数是一个指针，可以指向任何类型的变量。比如，如果要传递的内容很多，它可以是一个指向结构变量的指针，转换为指向void类型的指针后再传递。当线程开始执行后，可以通过这个指针取得这些参数。

线程启动函数的返回类型是int，通常用于在线程结束时返回一个状态码。状态码反映了线程结束时的状态，可以自行定义，比如用0表示线程正常结束。

用C标准库函数创建线程时，需要提供一个指向线程启动函数的指针。上面已经给出了线程启动函数的原型，那么类型int (*) (void *)就是指向线程启动函数的指针。为了方便起见，<threads.h>头定义了thrd_start_t类型，它是int (*) (void *)类型的别名。

28.1.2 线程的创建和管理函数

int thrd_create(thrd_t * thr, thrd_start_t func, void * arg); 
thrd_t thrd_current(void); 
int thrd_detach(thrd_t thr); 
int thrd_equal(thrd_t thr0, thrd_t thr1); 
_Noreturn void thrd_exit (int res); 
int thrd_join(thrd_t thr, int * res); 
int thrd_sleep(const struct timespec * duration, struct timespqc * remaining); 
void thrd_yield(void);

  • thrd_create函数创建一个新的线程。线程创建后，用参数thr返回一个标识，用来标记一个线程，方便后续的管理。线程标识的类型是thrd_t，它是在头<threads.h>中定义的。

    创建线程时，要传入指向线程启动函数的指针，这是通过参数func传入的。参数arg是传递给线程启动函数的参数。从实际效果来看，thrd_create函数的工作是执行函数调用func(arg)。

    如果线程成功创建，thrd_create函数返回thrd_success（包括下面的thrd_nomem和thrd_error都是在<threads.h>中定义的宏）。如果无法为线程的请求分配内存空间，则返回thrd_nomem。如果线程创建失败，则返回thrd_error。

  • thrd_current函数返回它当前所在线程的标识。它的工作很简单，在哪个线程内调用了这个函数，这个函数就返回哪个线程的标识。

  • thrd_detach函数将指定的线程设置为分离线程。一个具有分离属性的线程是由操作系统来管理的，当它结束后，由操作系统负责回收它所占用的资源。thrd_detach函数执行成功时返回thrd_success，失败则返回thrd_error。如果以前曾经将线程设置为结合线程或者分离线程，则不可再次设置，否则会返回thrd_error。

  • thrd_equal函数对两个线程标识进行比较，看它们是否为同一线程，参数thr0和thr1的内容是两个线程的标识。如果两个线程标识不同（即不同的线程），则函数返回0；否则返回非零值。

  • thrd_exit函数终止执行它所在的当前线程（即调用这个函数的线程），并将线程的返回码设置为参数res的值。注意，这个函数是不返回的。

  • thrd_join函数将指定的线程同当前线程（即调用此函数的线程）相结合。这将阻塞当前线程的执行，直到指定的线程终止执行。如果参数res不是空指针，则它将返回指定线程的返回码。注意，参数thr指定的线程不能是此前曾设置为分离或者结合的线程。在执行成功时该函数返回thrd_success，否则返回thrd_error。

  • thrd_sleep函数将当前线程（即调用此函数的线程）挂起，经过由参数duration指定的时间间隔之后，或者某个信号产生时，才恢复执行。如果恢复执行是因信号而引起，而且第二个参数remaining不为空指针，则剩余的时间（指定的间隔减去实际休眠的时间）将存储到第一个参数duration中。实际上，duration和remaining可以指向同一个对象。

    此函数的两个参数都是指向结构类型struct timespec的指针。结构类型struct timespec是在<time.h>头中定义的（26.3节），<threads.h>头中包含<time.h>，所以可在包含<threads.h>的情况下直接使用。结构类型struct timespec用来保存一个用秒和纳秒来指定的时间间隔，因此从理论上来说，它的时间精度还是非常高的。

    线程挂起的时间可能会长于要求的时间间隔，毕竟时间间隔会根据实际的系统定时器精度进行舍入，还要加上系统调度的时间开销。如果thrd_sleep函数返回0，表明指定的时间间隔已经到达；如果返回-1，表明遇到了中断信号；如果返回任何负值，表明执行失败。

  • thrd_yield函数试图主动将执行的机会让渡给其他线程，即使当前线程（即执行此函数的线程）其实可以继续正常执行。

程序——多线程同时运行的实例。

/*
下面通过示例程序mthrs.c来演示如何在一个程序中创建多个线程。
程序本身并不复杂，但它可以使我们了解多线程编程的基本要素、
线程函数的用法，以及多线程的工作特点。
--mthrs.c
*/
# include <stdio.h> 
# include <threads.h> 

int thread_proc(void * arg) 
{ 
    unsigned cnt = 5; 
    struct timespec interv = {1, 0}; 
    
    while (cnt --)  
    { 
        printf("%s\t", (char *) arg); 
        thrd_sleep (& interv, 0); 
    } 
    
    return 0; 
} 

int main(void) 
{ 
    thrd_t t0, t1; 
    
    thrd_create(& t0, thread_proc, "A"); 
    thrd_create(& t1, thread_proc, "b"); 
    
    thrd_detach(t0); 
    thrd_detach(t1); 
    
    printf("+\t"); 
    thrd_sleep(& (struct timespec) {1, 500000000}, 0); 
    printf("+\t"); 
    
    thrd_exit(0); 
}
/*
如果本地gcc不支持threads，使用在线C编译器https://www.onlinegdb.com/
output：
+       A       b       A       b       +       A       b       b       A       b       A（第一次）
A       +       b       A       b       +       A       b       A       b       b       A（第二次）
*/

在这个程序中可以看到，main函数首先开始执行。在main函数内调用线程创建函数thrd_create创建了两个线程，并用变量t0和t1保存这两个线程的标识。这两个线程来自同一个线程启动函数thread_proc，这是允许的。这样做的结果是用同一个线程启动函数创建了两个独立执行的线程，线程的代码一模一样，完成的工作也相同。虽然在程序中只有一份线程代码，但是两个线程有自己完全独立的执行环境以及执行流程，而且互不影响。

在创建线程时，我们还给它传递了参数。我们给第一个线程传递的参数是指向字符串"A"的指针，给第二个线程传送的参数是指向字符串"b"的指针。接下来看看线程启动函数thread_proc会完成什么工作。

线程启动函数的主体是用变量cnt和while语句构造一个循环。在循环体内，先打印输出传入的字符串，然后输出一个制表符。输出之后，再用thrd_sleep挂起当前线程，时间间隔为1秒。其实线程很简单，这样做是为了延长线程的执行时间，方便我们观察线程之间的同时执行以及交替输出过程。

传递给thrd_sleep的第一个参数是指向struct timespec结构的指针，之前讲过这个结构。该结构的第一个成员用来指定秒，第二个成员用来指定纳秒。因为是传入指针，而且需要在循环中反复使用，所以我们声明了一个结构变量interv。从它的初始化器可以知道，我们指定的时间间隔是1秒（即1秒0纳秒）。

在线程启动函数的最后，我们用return语句返回，这将结束线程的执行。当一个线程结束后，如果它是分离的，则由系统负责清理；如果它是与其他线程相结合的，则由生成结合关系的那个thrd_join函数负责清理。

再回到main函数，两个线程创建之后，紧接着用thrd_detach函数将它们设置为分离线程。这意味着，两个线程运行结束后，将由系统负责清理。

注意!我们创建的两个线程和main函数是同时运行的，而且实际上，main函数也代表了一个线程，它是整个进程的主线程。

现在的主流操作系统都支持多线程，你用C语言编写一个带有main函数的程序，在运行时会创建一个进程，你已经知道了。进一步来讲，这个进程会用main函数作为启动函数创建一个主线程。从这个意义上来说，你编写的每一个C语言程序，在运行时至少会创建一个线程。

就当前程序来说，它会用main函数创建一个主线程，这个线程又创建两个线程，所以现在是有3个线程在同时运行。当两个新线程打印字符串时，主线程先打印一个字符串"+"以及一个制表符，然后休眠1.5秒。我们在函数中指定的是1秒零500000000纳秒，折合1.5秒。休眠之后，再接着打印一个字符串"+"以及一个制表符，最后执行thrd_exit函数终止主线程。

在这个程序中，用main函数创建的主线程是非常特殊的，因为main函数返回时，会隐式地调用exit函数结束整个进程并返回到操作系统。所以，为了让其他线程正常结束，应当阻止main函数过早地返回。

为此，需要在main函数中调用thrd_exit来结束当前线程，而不是任由它返回。该函数仅仅结束当前主线程，但整个进程不会结束，进程内的其他线程也不会结束。只有进程内的所有线程都结束后，整个进程才会结束。

根据以上叙述也可以知道，在任何一个线程中调用exit函数都会导致整个进程结束，而进程内的所有线程也将结束。

另外，在这个main函数内没有return语句，这是允许的。从C99开始，如果main函数内没有return语句，则它执行到组成函数体的右花括号“}”时，相当于执行

return 0;

本程序执行时，3个线程都在屏幕上输出，但底层的输入/输出系统每次只能输出一个线程的内容，其结果就是3个线程的输出呈现交错的状态。

28.1.3 数据竞争

引入多线程，可以加快程序的执行速度，并提高程序的执行效率。如果每个线程的工作是独立的，线程之间彼此无关，也不需要访问同一个变量，那将是最理想的。但现实中，多个线程之间分工协作的情况很常见，而且必要时还必须进行同步，一个线程需要另一个线程完成特定的工作之后才能继续执行。

多线程给编程工作带来了挑战。如果多个线程访问同一个变量，那么线程之间的同步就变得尤其重要。一个线程在修改一个变量时，如果其他线程也同时在读取或者修改这个变量，它们之间就会产生冲突，这也叫作数据竞争（data race）。此时，必须有一种机制来保证多个线程对同一变量的读取和更新按有序的方式进行。否则，我们将得不到正确的结果。

下面通过示例程序datarace.c来演示多线程的数据竞争，以及它所造成的后果，并解释这种竞争是如何发生的。

/*
datarace.c
*/
# include <stdio.h> 
# include <threads.h> 

long long counter = 0; 

int thrd_proc1(void * arg) 
{ 
    struct timespec interv = {0, 20}; 
    
    for (size_t x = 0; x < 5000; x ++) 
    { 
        counter += 1; 
        thrd_sleep(& interv, 0); 
    } 
    
    return 0; 
} 

int thrd_proc2(void * arg) 
{ 
    struct timespec interv = {0, 30}; 
    
    for (size_t x = 0; x < 5000; x ++) 
    { 
        counter -= 1; 
        thrd_sleep(& interv, 0); 
    } 
    
    return 0; 
} 

int main(void) 
{ 
    thrd_t t0, t1; 
    
    thrd_create(& t0, thrd_proc1, 0); 
    thrd_create(& t1, thrd_proc2, 0); 
    
    thrd_join(t0, & (int){0}); 
    thrd_join(t1, & (int){0}); 
    
    printf("%lld\n", counter); 
}
/*
可能的输出：-1、0、1、-2....
*/

在这个程序中，有一个自动创建的主线程，它对应于函数main。主线程中创建了两个新的线程，它们的线程启动函数分别是thrd_proc1和thrd_proc2。

这个程序的另一个特点是，它有一个静态存储期的变量counter，而且这个变量在所有线程内都可见，也都可以访问。线程thrd_proc1的任务是对变量counter执行5000次加1操作；线程thrd_proc2的任务正好相反，是对变量counter执行5000次减1操作。

表面上看，这两个线程的工作是相反的，因其效果互相抵消。在main函数里，thrd_join函数等待线程结束并释放它所占用的资源，如果线程已经结束则直接释放它占用的资源并立即返回。两个线程结束之后，我们打印了变量counter的值，它应当是初始的数值0。但实际上，打印的结果是不确定的，有时是正值，有时是负值，当然也可能偶尔会是0，而且每次打印的结果都不一样。

那么，为什么会出现这种情况呢？这其实和C语言无关。在C语言的层面上，程序的结构和功能是清晰的、完整的，语法和程序逻辑也没有任何问题。实际上，问题出在C语言之下的层面和工作机制上。C语言是高级语言，最终要转换成更低层次的机器指令。举个例子来说，语句

counter += 1;

会被编译成3条机器指令，这3条机器指令对应于完成上述语句的功能所需要的3个步骤：

  • 读变量counter的值；
  • 将读来的值加1；
  • 将加1后的值更新（写入）到变量counter。

同样，下面的语句

counter -= 1;

会被编译成3条机器指令，这3条机器指令对应于完成上述语句的功能所需要的3个步骤：

  • 读变量counter的值；
  • 将读来的值减1；
  • 将减1后的值更新（写入）到变量counter。

在多线程环境下，属于不同线程的机器指令可能会同时执行（多处理器或者多核的情况下），也可能会交错执行（单处理器或者单核的情况下）。因此，在这里，前3条机器指令（来自线程thrd_proc1）和后3条机器指令（来自线程thrd_proc2）可能会交错执行。并且由于它们访问的是同一个变量，因此必然会发生数据竞争问题。

如表28-1所示，假定这两个线程只循环1次，而不是5000次。即使是这样，第一个线程将counter加1，第二个线程将counter减1，变量counter的值应该依然是原先的0。但是，由于数据竞争的缘故，在最坏的情况下，两个线程执行结束后，变量counter的值是-1而不是0。分析如下。

表28-1 单次循环时线程间数据竞争的过程分析

时间点线程thrd_proc1的操作线程thrd_proc2的操作counter的值
0读counter的值（得到0）0
1将读取的值加1（得到1）读counter的值（得到0）0
2将加1后的值写回counter将读取的值减1（得到-1）1
3将减1后的值写回counter-1
  • 在时间点0上，线程thrd_proc1读变量counter的值，读到的值是0。在这个时间点上，线程thrd_proc2不能读变量counter，因为已经有一个线程在执行读操作，它只能等待。
  • 在时间点1上，线程thrd_proc1将刚刚读到的值（0）加1，结果是1；与此同时，线程thrd_proc2开始读变量counter，读到的值是0。
  • 在时间点2上，线程thrd_proc1将加1后的新值1写回到变量counter，变量counter的值是1；与此同时，线程thrd_proc2将读来的值（0）减1，结果是-1。
  • 在时间点3上，线程thrd_proc1已经完成了对变量counter的操作，线程thrd_proc2开始将减1后的新值-1写回到变量counter，变量counter的值是-1。

显然，如果两个线程都只循环一次，在以上最坏的情况下，变量counter的值并不会因为减1和加1互相抵消而保持不变（0），而是-1。机器指令交错执行的概率取决于程序的复杂程度，为了提高这种概率，我们在线程中使用了循环，循环的次数是5000次。非但如此，我们还用thrd_sleep函数延迟线程的执行时间。如果你的计算机速度很快，那么可以适当增加循环的次数，或者增加延迟的时间。

通过以上分析可以看出，尽管在C语言的语法层面上，语句

counter += 1;

描述了一个完整的变量更新动作，但它在底层被分割为3个动作，而且这些动作之间可以被打断。在多线程环境中，要想解决数据竞争的问题，必须使这3个动作不能被别的线程打断。要做到这一点，可以使用锁，比较典型的锁就是互斥锁（mutex lock）。

对互斥锁的操作是通过互斥函数进行的，这些互斥函数在<threads.h>头中声明。除此之外，<threads.h>中还定义了与互斥锁相关的类型。

28.1.4 互斥函数

void mix_destroy(mtx_t * mtx)
int mtx_init(mtx_t * mtx, int type); 
int mtx_lock(mtx_t * mtx); 
int mtx_timedlock(mtx_t * restrict mtx, const struct timespec * restrict ts); 
int mtx_trylock(mtx_t * mtx); 
int mtx_unlock(mtx_t * mtx);

mtx_destroy函数释放参数mtx指向的互斥锁所占用的任何资源。释放之后，任何等待该锁的线程都不再被阻塞。参数mtx是一个指针，指向mtx_t类型的对象，这种类型是在<threads.h>头中定义的，这种类型的值用于标识互斥对象。mtx_destroy函数不返回任何值。

mtx_init函数创建一个互斥对象，互斥对象的属性（类型）由参数type指定，一共有6种，可以用下面的宏来指定，这些宏是在<threads.h>头中定义的。

  • mtx_plain：简单的非递归互斥锁。
  • mtx_timed：支持超时设定的非递归互斥锁。
  • mtx_plain|mtx_recursive：简单的递归互斥锁。
  • mtx_timed|mtx_recursive：支持超时设定的递归互斥锁。

有关这些属性值的含义，本节的后面会介绍。如果mtx_init函数执行成功，则将为mtx指向的互斥对象设置一个新值，这个值唯一地标识新创建的互斥锁，然后该函数返回thrd_success；如果创建失败，则返回thrd_error。

mtx_lock函数阻塞当前线程的执行，直到它锁定由参数mtx指向的互斥锁。如果互斥锁是非递归的，在调用此函数之前，它不能已经被当前线程锁定。如果此函数执行成功，则返回thrd_success；否则返回thrd_error。

mtx_timedlock函数试图阻塞当前线程的执行，直到它锁定由参数mtx指向的互斥锁，或者在经过了由参数ts所指向的日历时间之后（这个日历时间是基于TIME-UTC的，26.3.1节）返回。参数mtx指向的互斥锁必须支持超时设置。如果此函数执行成功，则返回thrd_success；如果指定的时间已到，但未成功加锁，则返回thrd_timedout（这是一个在<threads.h>头中定义的宏）；在其他情况下返回thrd_error表示执行失败。

mtx_trylock函数尝试锁定由参数mtx指向的互斥锁。如果它原先已经被锁定，则此函数无阻塞地立即返回。在成功锁定时返回thrd_success；返回thrd_busy意味着互斥锁已经处于锁定状态；返回thrd_error表示执行失败。

mtx_unlock函数释放（解锁）由参数mtx指向的互斥锁。参数mtx指向的互斥锁必须在此之前已经被当前线程锁定。若解锁成功，此函数返回thrd_success；否则返回thrd_error。

下面的程序mtxlock.c是datarace.c的修改版本，它通过在访问变量前加锁、访问变量后解锁的方式，来协调线程间的数据访问竞争。

/*
mtxlock.c
*/
# include <stdio.h> 
# include <threads.h> 

long long counter = 0; 
mtx_t mtx; 

int thrd_proc1(void * arg) 
{ 
    struct timespec interv = {0, 20}; 
    
    for (size_t x = 0; x < 5000; x++) 
    { 
        mtx_lock(& mtx); 
        counter += 1; 
        mtx_unlock(& mtx); 
        thrd_sleep(& interv, 0); 
    } 
    
    return 0; 
} 

int thrd_proc2(void * arg) 
{ 
    struct timespec interv = {0, 30}; 
    
    for (size_t x = 0; x < 5000; x++) 
    { 
        mtx_lock(& mtx); 
        counter -= 1; 
        mtx_unlock(& mtx); 
        thrd_sleep(& interv, 0); 
    } 
    return 0; 
}

int main(void) 
{ 
    thrd_t t0, t1; 
    mtx_init(& mtx, mtx_plain); 
    
    thrd_create(& t0, thrd_proc1, 0); 
    thrd_create(& t1, thrd_proc2, 0); 
    
    thrd_join(t0, & (int){0}); 
    thrd_join(t1, & (int){0}); 
    
    printf("%lld\n", counter); 
}
/*
output:
0
*/

在这个示例程序中，我们声明了一个mtx_t类型的互斥对象，它代表一个互斥锁，但并非一开始就有效，而是直到在main函数内用mtx_init创建一个真正的互斥对象时才开始工作。注意，锁的类型是mtx_plain，这是一个非递归的互斥锁。

紧接着，我们创建两个线程thrd_proc1和thrd_proc2，这和前面是一样的。这两个线程的内容变化不大，只是各自增加了两条语句，但是很关键。在线程thrd_proc1中，执行语句

counter += 1;

之前，先调用函数mtx_lock执行一个加锁的动作，之后再调用函数mtx_unlock解锁；在线程thrd_proc2中，执行语句

counter -= 1;

之前，先调用函数mtx_lock执行一个加锁的动作，之后再调用函数mtx_unlock解锁。

我们知道，这两条语句在编译后，都对应着3条机器指令，分别用于读变量counter、执行计算，以及更新变量counter。如果能够保证这3条指令不和其他线程交错执行，就能避免数据竞争，所以要在这两条语句的前面执行加锁操作。

不管两个线程中的哪一个先执行到mtx_lock函数，在同一时刻，只能有一个加锁成功，
然后继续往下执行。另一个线程加锁不成功，只能原地等待。也就是说，mtx_lock函数不会返回，从而导致线程被阻塞，直到它等待的那个锁被另一个线程用mtx_unlock函数释放。

利用这种加锁和解锁的方法，可以保证每个线程都能完整地执行读和更新操作，而不被其他线程打断，这从程序mtxlock.c的运行结果始终为0就可以看出。

28.1.5 条件变量

到目前为止，我们已经学习了如何用互斥锁来协调多个线程访问同一个变量时的数据竞争问题。互斥锁是排他性的，多个线程之间通过抢占互斥锁来完成数据访问，只有抢到锁的线程可以访问数据，没有抢到锁的线程只能阻塞等待。

互斥锁是排他性的，因为多个线程之间没有通信渠道，它们之间是单纯的竞争关系，往往缺乏沟通。但是线程之间也未必总是竞争关系，也可能需要同步和协作。比如说，一个线程需要等待另一个线程完成了某个工作之后才能继续执行。

由于这个原因，C语言的线程库引入了条件变量。条件变量用于在两个线程之间形成同步关系，类似于在线程之间传递消息。在这种情况下，一个线程可以处于阻塞或者休眠状态以等待某个事件的发生。如果条件满足，它可以立即获得通知，并重新开始运行。

下面的程序thrdcoop.c用来说明单纯使用互斥锁的局限性。在这个例子中，我们需要汇总两个数，但这两个数是需要通过计算才能得到的。为了加快计算速度，我们创建了两个线程，让它们各自计算两个数中的一个。

除了并行计算，这两个线程还需要在执行的过程中进行同步：如果一个线程完成了自己的计算，它需要看一下另一个线程是否也完成了计算。如果另一个线程还没有完成计算，它需要等待；如果另一个线程也完成了计算，它就把两个线程的结果相加，并打印这个结果，然后结束线程。

/*
thrdcoop.c
*/
# include <stdio.h> 
# include <threads.h> 

int data1 = -1, data2 = -1; 
mtx_t mtx; 

int thrd_task1(void * arg) 
{ 
    thrd_sleep(& (struct timespec){3, 0}, 0); 
    data1 = 12000; 
    
    again: 
    mtx_lock(& mtx); 
    if (data2 != -1) 
    { 
        printf("%d\n", data1 + data2); 
        mtx_unlock(& mtx); 
        return 0; 
    } 
    mtx_unlock(& mtx); 
    goto again; 
} 

int thrd_task2(void * arg) 
{ 
    thrd_sleep(& (struct timespec){2, 0}, 0); 
    data2 = 306; 
    
    again: 
    mtx_lock(& mtx); 
    if (data1 != -1) 
    { 
        printf("%d\n", data1 + data2); 
        mtx_unlock(& mtx); 
        return 0; 
    } 
    mtx_unlock(& mtx); 
    goto again; 
} 

int main(void) 
{ 
    thrd_t t0, t1; 
    mtx_init(& mtx, mtx_plain);
    thrd_create(& t0, thrd_task1, 0); 
    thrd_create(& t1, thrd_task2, 0); 
    
    thrd_join(t0, & (int){0}); 
    thrd_join(t1, & (int){0}); 
}
/*
output:
12306
12306
*/

在这个示例程序中，两个线程有点相似。在线程thrd_task1中，语句

thrd_sleep(& (struct timespec){3, 0}, 0);
data1 = 12000;

用延时3秒的方式来表示（模拟）一个复杂的数据计算过程，而且假定最终的计算结果是12000，它保存在全局变量data1中；在线程thrd_task2中，语句

thrd_sleep(& (struct timespec){2, 0}, 0);
data2 = 306;

也用延时（2秒）的方式来表示（模拟）一个复杂的数据计算过程，而且假定最终的计算结果是306，它保存在全局变量data2中。

在两个线程中，接下来的代码都是用goto语句组成的循环，用来等待另一个线程的计算结果。如果看到对方的结果不是初始值-1，就意味着对方也算出了结果，就可以直接打印这两个结果的相加和；否则就继续判断和等待。

在这个过程中，对全局变量data1和data2的读写操作是用互斥锁保护的，在同一时间只能由一个线程访问它们，这是最基本的要求。

需要说明的是，这两个线程中的一个会在等待另一个线程的过程中占用和消耗大量的CPU时间。等待的时间越长，占用和消耗的CPU时间就越多，程序的效率就越低。原因很简单，只要获得互斥锁，线程就开始用if语句判断对方线程的数值是否为-1，然后做相应的处理。

实际上最好的方法是，每个线程在完成自己的计算之后，如果对方线程还没有计算完成，它可以进入阻塞和休眠状态，等待对方线程在完成之后给自己发一个信号，这样就不会无谓地浪费CPU时间。

一个线程可以给其他线程发送通知信号，也可以等待其他线程的信号。为此，需要定义条件变量，并使用条件变量函数来发送和等待信号。条件变量是一个特殊的变量，它的类型是cnd_t，这种类型是在<threads.h>头中定义的。在使用条件变量前，必须先声明一个这种类型的变量。

28.1.6 条件变量函数

int cnd_broadcast(cnd_t * cond); 
void cnd_destroy(cnd_t * cond); 
int cnd_init(cnd_t * cond); 
int cnd_signal(cnd_t * cond); 
int cnd_timedwait(cnd_t * restrict cond, mtx_t * restrict mtx, 
                  const struct timespec * restrict ts); 
int cnd_wait(cnd_t * cond, mtx_t * mtx);

cnd_broadcast用来解除被参数cond指向的条件变量阻塞的所有线程。调用此函数时，如果此前有任何线程被参数cond指向的条件变量阻塞，则此函数解除它们的阻塞状态。如果没有任何线程被参数cond指向的条件变量阻塞，则此函数什么也不做。此函数执行成功则返回thrd_success，否则返回thrd_error。

cnd_destroy函数注销参数cond指向的条件变量，释放它所使用的所有资源。调用此函数的前提是没有任何线程正在被参数cond指向的条件变量阻塞。

cnd_init函数用于创建一个条件变量。如果创建成功，则参数cond指向的变量会有一个合法的值，这个值用于标识新创建的条件变量。在任何线程内，使用这个新条件变量来调用cnd_wait函数时都会被阻塞。cnd_init函数执行成功返回thrd_success；如果不能为新条件变量分配足够的内存，则返回thrd_nomem；因其他原因不能完成条件变量的创建操作，则返回thrd_error。

cnd_signal函数解除被参数cond指向的条件变量阻塞的某一个线程。注意，此函数只是解除某一个线程，而不是所有被此条件变量阻塞的线程。调用此函数时，如果没有任何线程被参数cond指向的条件变量阻塞，则此函数什么也不做，并返回表示成功的状态值。此函数执行成功，则返回thrd_success；执行失败，则返回thrd_error。

cnd_timedwait函数自动解除参数mtx指向的互斥锁并尝试阻塞（当前线程），直到参数cond指向的条件变量被（其他线程的）cnd_signal或者cnd_broadcast函数调用触发，或者在经过了由参数ts所指向的日历时间之后（这个日历时间是基于TIME-UTC的，26.3.1节）才返回。

在调用此函数之前，要求参数mtx指向的互斥锁已经被调用它的线程锁定。正常情况下，当此函数返回时，它会重新锁定由参数mtx所指向的互斥锁。

此函数成功执行时返回thrd_success。若因超时而返回，返回值是thrd_timedout，其他情况下返回thrd_error。

cnd_wait函数自动解除参数mtx指向的互斥锁并尝试阻塞（当前线程），直到参数cond指向的条件变量被（其他线程的）cnd_signal或者cnd_broadcast函数调用触发。在调用此函数之前，要求参数mtx指向的互斥锁已经被调用它的线程锁定。正常情况下，当此函数返回时，它会重新锁定由参数mtx所指向的互斥锁。此函数执行成功时将返回thrd_success，否则返回thrd_error。

下面的程序condcoop.c是在上一个程序thrdcoop.c的基础上修改来的，主要的变化是使用条件变量在两个线程间形成一种等待和通知的关系，从而使它们的执行在某个特定的时间点上保持同步。

/*
condcoop.c
*/
# include <stdio.h> 
# include <threads.h> 

int data1 = -1, data2 = -1; 
mtx_t mtx; 
cnd_t cnd; 

int thrd_task1(void * arg) 
{ 
    thrd_sleep (& (struct timespec){3, 0}, 0); 
    data1 = 12000; 
    
    mtx_lock(& mtx); 
    if (data2 != -1) cnd_signal (& cnd); 
    else cnd_wait (& cnd, & mtx);
    printf("%d\n", data1 + data2); 
    mtx_unlock(& mtx); 
    
    return 0; 
} 

int thrd_task2(void * arg) 
{ 
    thrd_sleep(& (struct timespec){2, 0}, 0); 
    data2 = 306; 
    
    mtx_lock(& mtx); 
    if (data1 != -1) cnd_signal (& cnd); 
    else cnd_wait (& cnd, & mtx); 
    printf("%d\n", data1 + data2); 
    mtx_unlock(& mtx); 
        
    return 0; 
} 

int main(void) 
{ 
    thrd_t t0, t1; 
    mtx_init(& mtx, mtx_plain); 
    cnd_init(& cnd); 
    
    thrd_create(& t0, thrd_task1, 0); 
    thrd_create(& t1, thrd_task2, 0); 
    
    thrd_join(t0, & (int){0}); 
    thrd_join(t1, & (int){0}); 
}
/*
output:
12306
12306
*/

在这个程序中，变量cnd的值用来标识一个条件变量。当然，这个值在程序刚启动时不代表有效的条件变量，所以要在main函数里调用cnd_init来创建一个条件变量。

条件变量不能解决数据竞争，它只是用来同步线程的执行。也就是说，如果仅仅是为了在多个线程之间共享变量，可以只使用锁。但是，要想让多个线程的执行在某个时间点会合或者说同步，为了不把时间浪费在无谓的、无休止的反复查询上，条件变量用来和代表运行状态的全局变量一起，使线程进入休眠状态或者发送通知。因为用到了全局变量，而且可能产生数据竞争，所以条件变量必须和锁一起使用。

因为线程thrd_task1和thrd_task2的工作流程相同，所以我们以thrd_task1为例描述一下它们的执行过程。当线程thrd_task1完成数据计算后，它会尝试调用mtx_lock函数加锁。如果加锁不成功则将阻塞，同时也说明另一个线程抢先一步完成数据计算并加锁成功。如果加锁成功则将执行语句

if (data2 != -1) cnd_signal (& cnd); 
else cnd_wait (& cnd, & mtx);

这条if语句用来看一下线程thrd_task2的数据计算过程是否已经完成，这是通过判断全局变量data2的原始值-1是否被改写来进行的。

如果线程thrd_task2也已经完成数据计算，即变量data2的值不等于原始数值-1，那么线程thrd_task2将处于以下3种可能的状态之一:

  • 已经完成数据计算，但没有抢到锁，正阻塞于函数mtx_lock。
  • 早就完成数据计算，也抢到了锁，但发现线程thrd_task1还没有完成数据计算，于是调用cnd_wait函数阻塞自己并释放锁。这也是线程thrd_task1能够加锁并正在执行的原因。
  • 已经完成数据计算，而且先一步抢到了锁，同时发现线程thrd_task1也完成了数据计算（但没有抢到锁），于是调用cnd_signal发送一个信号后打印相加结果并释放锁，目前已经结束执行，或者正在执行mtx_unlock后面的指令。而这也是线程thrd_task1能够加锁并正在执行的原因。

不管线程thrd_task2处于什么状态，因为它已经完成了数据计算，所以线程thrd_task1只需要用cnd_signal发送一个信号。

如果线程thrd_task2处在状态(1)或(3)，这个信号将消失而不起任何作用，线程thrd_task1继续往下执行，打印相加的结果并释放锁，然后结束线程。如果线程thrd_task2处在状态(2)，则函数cnd_signal会把线程thrd_task2从阻塞状态中唤醒，于是线程thrd_task2的cnd_wait函数开始尝试加锁。不过这个时候互斥锁仍在被线程thrd_task1占有。好在函数cnd_signal也返回到线程thrd_task1继续往下执行，而且迟早会执行到mtx_unlock函数。此时，线程thrd_task1释放锁，而线程thrd_task2中的cnd_wait函数立即执行加锁操作。

一旦线程thrd_task2因函数cnd_signal而获得了锁，它就可以继续往下执行，打印相加的结果并释放锁，然后结束线程。与此同时，线程thrd_task1也因执行return语句而早就结束了。

回到上面的if语句，如果线程thrd_task1发现线程thrd_task2尚未完成数据计算，那么它将执行cnd_wait来阻塞自己，并释放互斥锁，这是等待对方计算完成的意思。当线程thrd_task2完成数据计算后，就可以获得锁并继续执行。在线程thrd_task2里，语句

if (data1 != -1) cnd_signal (& cnd); 
else cnd_wait (& cnd, & mtx);

用来看一下线程thrd_task1的数据计算过程是否已经完成。基于以上流程，它会发现线程thrd_task1已经完成数据计算，于是调用cnd_signal给它发送一个信号。这个函数将线程thrd_task1从阻塞状态唤醒（解除阻塞），于是cnd_wait函数开始尝试加锁。不过这个时候互斥锁仍被线程thrd_task2占有。好在函数cnd_signal也返回到线程thrd_task2继续往下执行，而且迟早会执行到mtx_unlock函数。此时，线程thrd_task2释放锁，而线程thrd_task1中的cnd_wait函数立即执行加锁操作。

一旦线程thrd_task1因函数cnd_signal而获得了锁，它就可以继续往下执行，打印相加的结果并释放锁，然后结束线程。与此同时，线程thrd_task2也因执行return语句而早就结束了。

综上，因为两个线程是以通知和等待的方式进行同步的，所以无论谁先完成计算，其中一方都可以进入阻塞状态等待对方完成并发送通知，也就避免了用反复查询的方式来获取对方的状态，从而节省了CPU时间。程序执行后，将打印

12306
12306

28.1.7 递归锁和非递归锁

前面提到了递归锁和非递归锁。那么，什么是递归锁，什么是非递归锁？来看一个函数：

void func(int x) 
{ 
    mtx_lock(& mtx); 
    if (x) func(-- x); 
    mtx_unlock(& mtx); 
}

函数func是递归调用的，如果参数x的值不为0，递归调用func，否则返回上一层调用。但是我们也要注意到，在进入函数后调用了mtx_lock加锁，在函数返回前也要调用mtx_unlock解锁。显然，每递归调用一次，都会重复加一次锁。

如果锁是非递归的，那么在尝试加锁时，如果锁本身已经处于锁定状态，本次加锁操作将被阻塞，并等待锁被释放。就这个对函数func的递归调用而言，第一次调用这个函数时，加锁可以成功。第二次递归调用时，加锁操作被阻塞，因为它需要先将第一次调用时加的锁释放，但这是不可能的，于是就形成了死锁。

为了解决锁在函数递归调用时的问题，可以允许互斥锁在同一个线程中被重复锁定，这就形成了所谓的递归锁。如果一个锁是递归锁，则上面的递归调用就不会存在死锁。任何时候，锁在线程之间都是竞争关系，如果线程A获得了锁，它可以递归加锁。但是，只有在线程A释放了这个锁的时候，其他线程才能解除阻塞并获得这个锁。

28.1.8 初始化函数

void call_once(once_flag * flag, void (* func) (void));

来看一个问题。在前面的例子中，线程thrd_task1和thrd_task2都需要使用全局变量cnd、mtx、data1和data2，也都会打印data1+data2的结果。当然，这些全局变量在使用前需要初始化，而且我们是在main函数内初始化的。

如果因为某种原因，只能将这些变量的初始化工作放在线程内进行，而且要求打印工作只能进行一次（毕竟两个线程打印的结果是一样的），该怎么办呢？

变量的初始化只需要进行一次，而且应该由那个最先开始执行的线程来完成。如果不是这样的话，当某个线程使用这些变量时，它们可能还没有初始化。问题在于，在多线程环境中，哪个线程先开始执行是不确定的。因此，我们需要使用一些技巧，比如将初始化工作放在一个函数里，所有线程都可以调用这个函数，但需要做一些判断，看别的线程是否已经调用了这个函数。

为了简化这样的问题，C标准库提供了“只执行一次”的解决方案，或者叫“单次调用”，在多线程环境下，如果多个线程都调用某个函数，则它可以保证只调用一次。但是，这个函数的原型必须是

void func(void);

也就是说，这个函数必须无参数，而且不返回任何值。函数的名字是无所谓的，这里的func可以用你喜欢的名字来代替。同时，这个函数也不是在线程中直接调用的，它需要借助于C标准库函数call_once来进行。

函数call_once的功能是间接调用func所指向的函数，而且要使用参数flag指向的once_flag对象来确保这个函数只被调用一次。当然，这个保证的前提是，参数flag的值在call_once函数的第一次调用和后续调用中都保持不变，即始终指向同一个once_flag类型的对象。

类型once_flag是在<threads.h>头中定义的，用来保存一个供call_once函数使用的标志。这种类型的变量可以用一个宏ONCE_FLAG_INIT来初始化，这个宏也是在<threads.h>头中定义的。

从现实角度来说，单次调用通常用来做一些初始化工作，所以call_once函数被归类为初始化函数。

下面的程序callonce.c是在上一个程序condcoop.c的基础上修改来的，它增加了两个函数do_init和do_print，这两个函数都只会被调用一次。

/*
callonce.c
*/
# include <stdio.h> 
# include <threads.h> 

int data1 = -1, data2 = -1; 
mtx_t mtx; 
cnd_t cnd; 
once_flag flag1 = ONCE_FLAG_INIT, flag2 = ONCE_FLAG_INIT; 

void do_init(void) 
{ 
    mtx_init(& mtx, mtx_plain); 
    cnd_init(& cnd); 
    printf("Mutex and condition variable is created.\n"); 
} 

void do_print(void) 
{ 
    printf("The combined result is %d.\n", data1 + data2); 
} 

int thrd_task1(void * arg) 
{ 
    call_once(& flag1, do_init); 
    
    thrd_sleep(& (struct timespec){3, 0}, 0); 
    data1 = 12000; 
    
    mtx_lock(& mtx); 
    if (data2 != -1) cnd_signal(& cnd); 
    else cnd_wait(& cnd, & mtx); 
    call_once(& flag2, do_print); 
    mtx_unlock(& mtx); 
    
    return 0; 
} 

int thrd_task2(void * arg) 
{ 
    call_once(& flag1, do_init); 
    
    thrd_sleep(& (struct timespec){2, 0}, 0); 
    data2 = 306; 
    
    mtx_lock(& mtx); 
    if (data1 != -1) cnd_signal(& cnd); 
    else cnd_wait(& cnd, & mtx); 
    call_once(& flag2, do_print); 
    mtx_unlock(& mtx); 
    
    return 0; 
} 

int main(void) 
{ 
    thrd_t t0, t1; 
    
    thrd_create(& t0, thrd_task1, 0); 
    thrd_create(& t1, thrd_task2, 0); 
    
    thrd_join(t0, & (int){0}); 
    thrd_join(t1, & (int){0}); 
}
/*
output:
Mutex and condition variable is created.
The combined result is 12306.
*/

从程序中可以看出，函数do_init用来初始化变量mtx和cnd，这些工作原先是在函数main中进行的。为了能够看出do_init是否真的只调用了一次，初始化之后，我们特意打印了一条信息“Mutex and condition variable is created.”。

函数do_print也很简单，它只是打印变量data1的值和变量data2的值相加的结果。

程序的其他部分变化不大。首先，在线程thrd_task1和thrd_task2中，都是先用函数call_once来调用do_init，然后再用call_once来调用do_print。为了保证函数do_init只会被调用一次，使用的标志变量是flag1；为了保证函数do_print只会被调用一次，使用的标志变量是flag2，这两个标志变量是在程序的起始处声明的，并用宏ONCE_FLAG_INIT做了初始化。

尽管两个线程都调用了do_init和do_print，但实际上它们各自只被调用了一次。因此，程序的输出如下：

Mutex and condition variable is created. 
The combined result is 12306.

28.1.9 _Thread_local存储类和线程存储期(C1X)

通常来说，一个具有外部链接的全局变量具有静态存储期，而且整个程序的所有部分都可以共享它。即使是在多线程环境下，所有线程也共享这个变量。但是从C11开始，C语言引入了一个新的存储类指定符_Thread_local，这也是一个新的关键字，可用在变量的声明中，指定该变量具有线程存储期。

具有线程存储期的变量，其生存期贯穿于线程的执行过程，而且为线程所私有。当线程启动时创建和初始化这个变量，当线程结束时就将其销毁。

下面通过一个示例程序thrdloca.c来说明具有线程存储期的变量的特点。在这个程序中声明一个全局变量g，且在它的声明中使用了存储类指定符_Thread_local，所以这个变量为每个线程所私有，在每个线程中访问的g是该线程私有的变量g。

/*
thrdloca.c
*/
# include <stdio.h> 
# include <threads.h> 

_Thread_local int g; 

void do_print(void * arg) 
{ 
    printf("%s:%s\t%p.\n", (char *) arg, __func__,  & g); 
} 

void do_calc(void * arg) 
{ 
    printf("%s:%s\t%p.\n", (char *) arg, __func__,  & g); 
    do_print(arg); 
} 

int thrd_proc(void * arg) 
{ 
    printf("%s:%s\t%p.\n", (char *) arg, __func__,  & g); 
    do_calc(arg); 
    return 0; 
} 

int main(void) 
{ 
    thrd_t t0, t1; 
    
    thrd_create(& t0, thrd_proc, "A"); 
    thrd_create(& t1, thrd_proc, "B"); 
    
    printf("%s:%s\t%p.\n", "main", __func__, & g); 
    do_calc("main"); 
    
    thrd_join(t0, & (int){0}); 
    thrd_join(t1, & (int){0}); 
}

要证明每个线程中的变量g不同于其他线程中的变量g，只需要打印变量g在每个线程中的地址即可。这个程序里一共有3个线程：自动创建的默认主线程，它的线程启动函数是main；在主线程内又创建了两个线程，它们的启动函数都是thrd_proc。为了区分后两个线程，我们在创建它们的时候分别传入字符串“A”和“B”。

在线程启动函数thrd_proc内部，先打印变量g的地址。为了区分是哪个线程，打印的内容包括线程创建时传入的字符串，以及当前函数的名字。当前函数的名字来自一个预定义的宏__func__（14.3.12节）。

对于每一个线程，函数thrd_proc还要调用do_calc。函数do_calc以同样的方法打印变量g的地址，然后调用do_print，函数do_print也以同样的方法打印变量g的地址，这样就形成一个调用链。

在程序的主线程内，除了创建两个线程，也打印变量g的地址，打印的内容还包括字符串“main”以及当前函数的名字，它们也来自__func__。然后，主线程调用do_calc并且也将形成一个调用链。

最后，程序的打印输出可能是下面这样的。每次的输出也不一定完全相同。

main:main       0x7f0dc401373c.
main:do_calc    0x7f0dc401373c.
main:do_print   0x7f0dc401373c.
B:thrd_proc     0x7f0dc381163c.
B:do_calc       0x7f0dc381163c.
B:do_print      0x7f0dc381163c.
A:thrd_proc     0x7f0dc401263c.
A:do_calc       0x7f0dc401263c.
A:do_print      0x7f0dc401263c.

以上打印输出中，每一行的左侧是线程的标记，以及它正在执行哪个函数；右侧的十六进
制数是变量g的地址。显然，不同线程（A、B或者main）内的变量g也各不相同。尽管每个线程都调用了thrd_proc、do_calc和do_print，但在这些函数内部所访问的变量g是每个线程自己独立的变量g。

定义一个全局变量，但是让它为每个线程私有，乍一看有点滑稽，但在现实中确实会遇到不得不这样做的情况。如果一个线程很复杂，由很多函数组成，那么在函数之间共享变量可能会比互相传递参数更方便。进一步地，如果有多个这样的线程，但不需要在线程之间共享变量，那么让共享的变量具有线程存储期是一个好主意。

除了现实因素之外，存储类指定符_Thread_local的引入也和历史有关。从传统的单线程（进程）进入多线程之后，很多公司和个人面临的问题之一是，如何将以前的程序迁移到多线程环境中。这个迁移的过程面临很多问题，解决起来可能很简单，也可能非常麻烦。

从传统的单线程迁移到多线程，问题之一如何处理共享变量。在单线程时代，多个函数可能需要访问同一个变量。为了方便，我们将它定义在所有函数之外。在这个时期，因为程序的执行是单一线条的，函数的调用也是依次进行的，所以对共享变量的访问也是有序的。

这里有一个非常典型的例子。C标准库里有很多函数在调用失败后却不能给出一个明确的原因。比如，我们熟悉的函数fopen在调用成功后返回一个指向FILE类型的指针，但如果失败则返回NULL值。失败的原因很多，诸如访问权限不够、给出的文件名不合法、文件不存在等，但该函数没办法告诉你确切的原因。

为了解决这个问题，C标准库的做法是在<errno.h>头中声明一个变量errno，整个C标准库共享同一个变量errno（24.2节）。对于很多库函数来说，它可以在执行失败的时候把一个代表错误原因的整数值写入变量errno。当然，你的程序也可以访问这个变量，并从中取得错误号，但必须在调用一个库函数之后立即使用它。这是不言自明的，如果你连续调用了好几个库函数，后面的库函数将有可能覆盖先前的错误号。

首次引入errno的时候，多线程还没有流行，所以这种做法很奏效。就像我们刚才所说的，因为程序的执行是单一线条的，函数的调用也是依次进行的，所以对共享变量的访问也是有序的。

后来，多线程开始流行了。在这个时候，因为errno是全局变量，所以它对所有线程都是可见的、可用的。如果在线程A中调用库函数失败了，它会用errno变量的值来观察错误原因。然而，还没等它开始读errno变量，线程B也调用某个库函数，并且因失败而覆盖了变量errno的值。如此一来，线程A将读到一个不正确的错误代码。显然，要解决这个问题，最好的办法就是让每个线程都有自己独立的errno变量。因此，C标准库的做法是将errno定义为具有线程存储期的变量。

28.1.10 线程专属存储

用存储类指定符_Thread_local来创建线程存储期的变量，这种方式不够灵活，也不具有弹性。为此，C标准库提供了线程专属存储（thread-specific storage），它支持动态分配内存以应对复杂的数据类型和数据处理；如果需要，还可以提供专门的析构函数，在线程退出的时候释放所分配的内存。

和_Thread_local不同，线程专属存储的方案是所有线程共享同一个键（key），但是这个键在不同的线程内指示不同的存储区。

键的类型是tss_t，它是在头<threads.h>中定义的。在所有线程使用它们的专属存储区之前，必须先创建这个公有的键。为了创建一个键，并且用键来访问线程专属存储，标准库提供了线程专属存储函数。

28.1.11 线程专属存储函数

int tss_create(tss_t * key, tss_dtor_t dtor); 
void tss_delete(tss_t key); 
void * tss_get(tss_t key); 
int tss_set(tss_t key, void * val);

tss_create函数创建一个线程专属存储的指针（或者叫键）并将它存放在参数key指向的变量中。创建键的同时，还可以注册一个清理函数，用于在线程返回时做一些清理工作。如果线程是用return语句或者函数调用thrd_exit返回的，那么将自动调用这个函数做清理工作。被注册的函数必须具有如下原型：

void func(void *);

在这里，函数的名字是无所谓的，可以将func替换为任何你喜欢的名字。这个函数必须由你自己编写，然后将它的指针传递给tss_create函数的dtor参数。为了方便，在<threads.h>头中定义了指向这种函数的指针类型tss_dtor_t。所以，正如你现在看到的，参数dtor的类型是tss_dtor_t。

用thrd_create函数创建的键应当存放在一个所有线程可见的全局变量中，这个键对每个线程来说都完全相同，但它会在每个线程内部关联一个指针类型的值。每创建一个新键的时候，在所有已经存在的线程内，这个键所关联的值都被设置为空指针。对于后续创建的线程，这个键所关联的值也被初始化为空指针。

如果tss_create函数成功执行，它将生成一个唯一的键并返回thrd_success，否则返回thrd_error，并且键在数值上是不确定的。

函数tss_delete释放由参数key标识的线程专属存储。函数tss_get返回参数key所指定的键在当前线程中所关联的值。如果函数执行不成功，返回值是0（空指针）。函数tss_set将参数key所指定的键在当前线程中所关联的值设置为参数val的值。这个替换动作不会触发对清理函数的调用。该函数执行成功时返回thrd_success，执行失败则返回thrd_error。

下面通过示例程序tssdemo.c来演示如何创建和使用每个线程专属的存储区，以及在这个过程中需要注意的要点。

/*
tssdemo.c
*/
# include <stdio.h> 
# include <stdlib.h> 
# include <string.h> 
# include <threads.h> 

tss_t key; 

void destructor(void * data) 
{ 
    free (data); 
    printf("freed.\n"); 
} 

void do_print(void) 
{ 
    printf("%s.\n", (char *) tss_get (key)); 
    thrd_sleep(& (struct timespec){2, 0}, 0); 
} 

int thrd_proc(void * arg) 
{ 
    tss_set(key, malloc (strlen ((char *) arg) + 10)); 
    
    strcpy((char *) tss_get (key), "hello,"); 
    strcat((char *) tss_get (key), (char *) arg); 
    
    do_print(); 
    
    return 0; 
} 

int main(void) 
{ 
    tss_create(& key, destructor); 
    
    thrd_t t0, t1; 
    
    thrd_create(& t0, thrd_proc, "world"); 
    thrd_create(& t1, thrd_proc, "kitty"); 
    
    thrd_join(t0, & (int){0}); 
    thrd_join(t1, & (int){0}); 
    
    tss_delete(key); 
}

为了给每个线程分配专属的存储区，需要声明一个全局共享的键。在程序中，这个键保存在全局变量key中，它的类型是tss_t。这个变量的内容一开始是无效的，所以接下来必须调用tss_create函数来创建一个有效的键，并将它的标识保存到变量key中。

在本程序中，键的创建工作是在主线程的启动函数main内完成的，这将在当前线程内为这个键关联一个值，而且被初始化为0（空指针）。在创建这个键的时候我们还注册了一个清理函数destructor用来做最后的清理工作。当然，很多时候这是不必要的，所以函数tss_create的第二个参数可以是空指针。

紧接着，主线程创建了两个线程，并分别传入两个字符串“world”和“kitty”。这两个线程创建后，将各自用key关联一个值，而且初始化为0（空指针）。这两个线程用的是同一个线程启动函数thrd_proc，所以功能是一样的，执行流程也是一样的。

尽管key在每个线程内都有一个关联的值，但它只是一个指针类型的值，通常来说用处不大。但因为它是一个指针，因此可以指向其他变量，或者指向一个动态分配的内存区域，这就带来了应用上的灵活性。

在这两个线程的启动函数内部，我们用tss_set函数设置key在当前线程内所关联的值，这个值是一个指针，来自malloc函数的返回值。换句话说，修改之后，键key在当前线程内所关联的值指向一个动态分配的内存块。内存分配的长度是这样计算的：先调用函数strlen来计算传入的字符串的长度，然后在这个基础上加10。

接下来用字符串复制函数strcpy将字符串“hello,”复制到我们刚才分配的内存区中。目标位置的地址来自函数tss_get的返回值，这个函数返回key在当前线程内对应的值。返回值的类型是void*，但strcpy要求第一个参数的类型是char*，因此要做类型转换。

紧接着，我们再调用字符串拼接函数strcat将刚才的“hello,”和传入的字符串连接到一起。这一步的工作和上一步在形式上相似，不用多说。唯一要注意的是，必须根据函数原型的要求对传入的实际参数做类型转换。

将两个字符串连接到一起后，接下来调用do_print函数。这个函数调用printf函数打印刚才拼接好的字符串，这个字符串的首地址是key在当前线程内所关联的值，所以要用tss_get函数取得，并将它的类型从void*转换为char*。打印完成后，再调用thrd_sleep函数阻塞当前线程2秒。

在线程返回之前，应当做一些清理工作，特别是要释放动态分配的内存。如果已经注册了清理函数，则这些工作可以放到清理函数内进行。示例程序中已经注册了这样一个函数destructor，它就用来做最后的清理工作。

在调用这个函数之前，key在当前线程内所关联的值会被设置为0（空指针），然后调用清理函数，并把设置为0前的原值传递给它。从清理函数返回后，系统将检查这个值是否依然为0，如果不为0则再次调用清理函数。如果这个值一直不为0，则重复调用的次数取决于一个宏TSS_DTOR_ITERATIONS，它是在<threads.h>头中定义的，是一个整型常量表达式，用来指定清理函数被重复调用的最大次数。在本示例程序中，如果你将清理函数改成这样：

void destructor(void * data) 
{ 
    free(data); 
    tss_set(key, malloc (1)); 
    printf("freed.\n"); 
}

那么你会发现，这个函数会被调用很多次（可以从字符串“freed.”被重复打印很多次上看出来），因为我们总是在清理函数destructor返回前重新将key在当前线程内所关联的值修改为非零值。

在程序中，key在当前线程内所关联的值被设置为0，设置为0前的原值被传递给清理函数destructor的参数data。于是，我们就可以在函数内调用free函数来释放先前分配的内存，释放之后打印字符串“freed.”来表明清理工作已经结束。

本程序执行后将打印4行文本，由于是在多线程环境下，所以这4行文本的顺序可能并不固定:

hello,world.
hello,kitty.
freed.
freed.

28.2 _Atomic、<stdatomic.h>：原子类型和原子操作支持(C1X)

早期的计算环境比较简单，这体现在两个方面：首先，只有一个处理器，而且不存在多核的概念；其次，指令的执行是按照它们在内存中的自然顺序进行的，处理器取一条指令，译码然后加以执行，接着再取下一条指令，译码并加以执行。

随着时间推移，处理器变得越来越快，内存的速度却没有跟上，还是一如既往地缓慢。忘了是谁说过，“相比于处理器，内存慢得出奇”。为此，在处理器内部开始出现了高速缓存（cache），用来缓存最常用的数据。当处理器需要从内存中读取或者写入数据时，它会先检查它是否在高速缓存中。如果它在高速缓存中，则可以直接操作高速缓存，而不用等待较慢的内存。

在这个阶段，多线程开始流行并工作得很好。所谓的多线程，实际上表现为所有线程在一个处理器上交错执行，这也叫作并发（concurrency）执行。为了解决线程间因共享数据而引发的数据竞争，各种类型的锁——比如前面介绍的互斥锁——被发明出来。不过这个时候锁的实现并不复杂，只需要用一个数据结构控制好线程间交替执行的节奏就可以了。

然而，使用锁的弊端是增加了系统开销，因为有很多时间花在线程的状态切换上。为此，人们希望能够在避免使用锁的情况下编写程序，从而降低系统开销并提高软件的性能，这就是锁无关（lock-free）的程序设计。锁无关的程序设计本身就有难度，再加上我们已经进入多处理器（核）的时代，这个任务就更具挑战性了。

现代计算环境不同于以往，已经完成了从单处理器到多处理器（核）的过渡。在多处理器（核）的环境下，可以把线程分发给不同的处理器（核），让它们各自独立地同时执行，或者说并行（parallelism）执行。在这个时代，每个线程的行为从它们内部来看没有任何变化，这一点可以保证。如果线程间没有任何依赖，那么它们都可以开足马力，全速地并行执行，我们就可以享受多处理器（核）带来的好处。

但是，如果线程之间要共享变量并进行同步操作，事情就开始变得古怪而不可捉摸：一个线程给共享变量赋值，其他线程也许能看到新值，也许根本看不到；线程间的同步关系也变得混乱和不可预测。这也进一步增加了锁无关程序的设计难度，因为它不单单涉及锁无关如何实现，也涉及并行执行的效率。

几十年来，C语言社区一直通过线程、互斥锁和条件变量来探索并发这一课题，并描述各种粒度的并行性，这些成果最终出现在C11（ISO/IEC9899:2011）及其之后的标准中。新标准加入了原子类型，以及一个用来执行原子操作的标准库。对原子类型的支持是通过新增一个关键字_Atomic来实现的，它被用作类型指定符和类型限定符。对原子操作的支持则是通过引入一个新的头<stdatomic.h>来实现的。

注意，和多线程一样，原子特性是可选的，而不是强制性的，如果C实现并未支持这个特性，则它必须定义一个宏__STDC_NO_ATOMICS__，表明它不支持关键字_Atomic，也未提供<stdatomic.h>头。因此，在决定使用C语言的原子特性前，程序员应当先用预处理指令判断这个宏是否存在：

# ifdef __STDC_NO_ATOMICS__ 
# error"Not support atomic facilities." 
# endif

28.2.1 _Atomic：类型指定符/类型限定符(C1X)

从C11开始，C语言引入了关键字_Atomic，用来指定原子类型。这个关键字既用作类型指定符，也用作类型限定符。用作类型指定符的时候，它的格式为

_Atomic(类型名)

这里的“类型名”所指代的类型不能是数组、函数、原子或者限定的类型。无论什么时候，如果关键字_Atomic后面跟着一个左圆括号，则它被解释为类型指定符。当它作为类型限定符的时候，和其他类型限定符const、volatile及restrict相比在语法上没有什么区别，但是含义不同。

  • 用来指示一个原子类型。用_Atomic限定的类型在大小（数据长度）、数据的内部表示方法以及内存对齐方面可以和限定前的类型不同。
  • 不能用来修饰数组和函数类型。

基于以上所述可知，要声明一个原子类型的变量，可以采用如下两种方式，它们是等效的：

const _Atomic (int) atom_i;

或者

const _Atomic int atom_i;

28.2.2 标准库定义的原子类型

为了方便，<stdatomic.h>头里定义了很多原子类型。表28-2的第一列给出了这些原子类型的类型名，可直接用于声明原子变量；第二列是直接用传统方式声明的等价形式，所以叫直接类型。

表28-2 标准库定义的原子类型与其直接类型对照表

原子类型名直接类型
atomic_bool_Atomic(_Bool)
atomic_char_Atomic(char)
atomic_schar_Atomic(signed char)
atomic_uchar_Atomic(unsigned char)
atomic_short_Atomic(short)
atomic_ushort_Atomic(unsigned short)
atomic_int_Atomic(int)
atomic_uint_Atomic(unsigned int)
atomic_long_Atomic(long)
atomic_ulong_Atomic(unsigned long)
atomic_llong_Atomic(long long)
atomic_ullong_Atomic(unsigned long long)
atomic_char16_t_Atomic(char16_t)
atomic_char32_t_Atomic(char32_t)
atomic_wchar_t_Atomic(wchar_t)
atomic_int_least8_t_Atomic(int_least8_t)
atomic_uint_least8_t_Atomic(uint_least8_t)
atomic_int_least16_t_Atomic(int_least16_t)
atomic_uint_least16_t_Atomic(uint_least16_t)
atomic_int_least32_t_Atomic(int_least32_t)
atomic_uint_least32_t_Atomic(uint_least32_t)
atomic_int_least64_t_Atomic(int_least64_t)
atomic_uint_least64_t_Atomic(uint_least64_t)
atomic_int_fast8_t_Atomic(int_fast8_t)
atomic_uint_fast8_t_Atomic(uint_fast8_t)
atomic_int_fast16_t_Atomic(int_fast16_t)
atomic_uint_fast16_t_Atomic(uint_fast16_t)
atomic_int_fast32_t_Atomic(int_fast32_t)
atomic_uint_fast32_t_Atomic(uint_fast32_t)
atomic_int_fast64_t_Atomic(int_fast64_t)
atomic_uint_fast64_t_Atomic(uint_fast64_t)
atomic_intptr_t_Atomic(intptr_t)
atomic_uintptr_t_Atomic(uintptr_t)
atomic_size_t_Atomic(size_t)
atomic_ptrdiff_t_Atomic(ptrdiff_t)
atomic_intmax_t_Atomic(intmax_t)
atomic_uintmax_t_Atomic(uintmax_t)

28.2.3 初始化原子变量

原子变量可能有特殊的实现方式，以及相关的状态。依原子变量的声明而定，如果它是具有静态存储期（18.2.1节）或者线程存储期（28.1.9节）的变量且没有初始化，则将自动初始化为0值，并保证处于有效状态；相反，如果它是具有自动存储期（18.2.1节）的变量且没有初始化，则将处于不确定的状态。

因此，那些具有自动存储期的变量必须进行初始化，使其处于有效的初始状态。从C11开始提供了一个宏ATOMIC_VAR_INIT，它是在<stdatomic.h>头中定义的，可以用来初始化原子变量。例如：

_Atomic int atom_i = ATOMIC_VAR_INIT(77);

不建议使用这个宏来初始化原子变量，因为按照C18的说法，下一个版本的C标准将会废除它。因此，应该使用下一节将要介绍的原子变量初始化函数atomic_init。除此之外，在本章接下来的内容中还要介绍更多函数，它们都是泛型函数，参数的类型不固定，所以要使用字母来表示。因此，在继续下面的内容之前，我们有必要先达成以下共识：

  • 用A指代某种原子类型；
  • 用C指代与A相对应的非原子类型；
  • 用M指代执行算术操作的其他参数的类型。对于原子的整数类型，M是C；对于原子的指针类型，M是ptrdiff_t。

28.2.4 原子变量的初始化函数

void atomic_init(volatile A * obj, C value);

atomic_init是一个泛型函数，它用参数value的值初始化参数obj指向的原子对象，同时为该对象设置一些由实现定义的附加状态。尽管这个函数初始化的是一个原子对象，但它并不能防止数据竞争。在初始化期间，对当前对象的其他并发访问都将造成数据竞争，即使它们也是原子操作。

正如上面所说的，对于具有自动存储期的原子变量，应当显式地初始化。下面是一个用函数atomic_init来初始化原子变量的示例：

atomic_int atom_i; 
atomic_init(& atom_i, 77);

28.2.5 原子操作

和原子类型指定符/限定符_Atomic紧密关联的概念是原子操作。“原子操作”是一个名词而不是一个动词，它指一个完整的操作，比如一个读操作、一个写操作，或者一个完整的读-改-写操作。这个概念的引入是因为存在数据竞争——如果两个线程同时发起这样一个操作，在这个操作执行期间，其他处理器、进程或者线程不能访问相同的内存位置。从程序员的直觉来看，原子操作起码应该具有以下特征。

  • 一个读对象M的操作在执行期间，其他线程不能访问对象M。
  • 一个写对象M的操作在执行期间，其他线程不能访问对象M。
  • 任何一个对象M的读-改-写操作在执行期间，其他线程不能访问对象M；或者，至少在写入新值的时候，必须确保对象的值没有因其他线程的写操作而发生变化，否则应该重新执行读-改-写过程。

在C语言里，原子操作是对原子对象（变量）施加的操作。在处理器的层面上，原子操作和机器指令紧密相关，比如一个对齐于对象自然边界的内存操作指令，一个会导致总线锁定的内存操作指令，或者一个带有总线锁定前缀的内存操作指令。

首先，存（写）取（读）原子变量的操作是原子操作。进一步来讲，在C语言里，所有复合赋值运算符，诸如+=、/=、*=等，以及所有形式的++和--运算符，在用于修改原子变量时都执行原子操作。这些操作的原子性可能是借助于处理器的硬件指令实现的，也可能是通过内联（后面将要讲的）原子操作函数来实现的。当然，内联的代码也可能直接来自编译器，而不是标准库。

其次，在<stdatomic.h>头里定义了很多库函数（28.2.6节），它们用来存取原子变量，这些库函数也都执行原子操作。

在本章开始（28.1.3节），我们写了一个存在数据竞争的程序datarace.c。为了协调数据竞争，传统的解决方案是采用互斥锁，因此我们曾经编写过一个采用互斥锁的版本（28.1.4节）。

引入原子类型和原子操作的目的是支持锁无关（lock-free）的程序设计，从而降低系统开销，并提高程序的执行效率。因此，所幸我们只需要将datarace.c做一处小小的修改，就可以达到和采用互斥锁一样的目的，也就是解决数据竞争的问题。

那么，这一处神奇的、小小的修改在哪里呢？答案是将全局变量counter声明为原子类型。尽管这是一处不起眼的修改，文件的内容变化并不大，但为了方便讨论，这里还是给出了修改后的文件内容，新文件的名字是atomic.c。

/*
atomic.c
*/
# include <stdio.h> 
# include <threads.h> 

_Atomic long long counter = 0; 

int thrd_proc1(void * arg) 
{ 
    struct timespec interv = {0, 20}; 
 
    for (size_t x = 0; x < 5000; x ++) 
    { 
        counter += 1; 
        thrd_sleep(& interv, 0); 
    } 
    
    return 0; 
} 

int thrd_proc2(void * arg) 
{ 
    struct timespec interv = {0, 30}; 
    
    for (size_t x = 0; x < 5000; x ++) 
    { 
        counter -= 1; 
        thrd_sleep(& interv, 0); 
    } 
    
    return 0; 
} 

int main(void) 
{ 
    thrd_t t0, t1; 
    
    thrd_create(& t0, thrd_proc1, 0); 
    thrd_create(& t1, thrd_proc2, 0); 
    
    thrd_join(t0, & (int){0}); 
    thrd_join(t1, & (int){0}); 
    
    printf("%lld\n", counter); 
}

对比一下旧文件datarace.c和新文件atomic.c就能发现，我们只是把旧文件中的这一行：long long counter = 0;在新文件中改成了_Atomic long long counter = 0。

除此之外，其他内容没有任何变化。但是无论什么时候运行这个新程序，它打印输出的数字永远是0，这表明修改起了作用，数据竞争被消除了。

但是，如果我们将线程thrd_proc1中的counter += 1;改成counter = counter + 1;，将线程thrd_proc2中的counter -= 1;改成counter = counter - 1;。然后重新编译和运行程序，会发现数据竞争又出现了，每次运行程序都可能得到和上一次不同的结果，只有在极其偶然的情况下才能出现正确的结果0。我们说过，给一个原子变量赋值，这个操作是原子性的，但为什么改了一下赋值的形式，就不灵了呢？

用复合赋值运算符给原子变量赋值，或者用任何形式的++、--运算符来操作原子变量，都是“读-改-写”的原子操作。这种原子性保证“读-改-写”是个单一操作，不管它是如何实现的，其实际效果等同于在这个读-改-写期间，其他线程不能访问同一个变量。

但是，语句counter = counter + 1;就不同了，它是按顺序执行以下几个操作：

  1. 读变量counter的值；
  2. 将读来的值加1；
  3. 将加1后的新值写回变量counter。

因为运算符=的原子性仅仅体现在赋值动作本身，所以(3)是原子性的，只能保证在执行(3)期间，其他线程不会访问变量counter。按照程序的逻辑和我们的意图，从(1)开始一直到(3)，其他线程都不应该访问counter，但这是无法保证的。

在实际的编程工作中类似这样的情形还是比较常见的，区别仅仅在于(2)通常不会是加1或者减1这么简单，而更可能是对读来的值做一些复杂的运算，再将运算后的新值更新（写入）到原来变量。但你也看到了，如果多个线程都对同一个变量执行上述过程，数据竞争依然存在，即使这个共享变量是一个原子变量。

那么，难不成我们还要退回去使用互斥锁和条件变量吗？这倒不必。要知道，引入原子类型和<stdatomic.h>头的动机之一就是尽量减少对锁的依赖。<stdatomic.h>头中声明了一些函数，我们可以用它们来解决上述问题。

28.2.6 原子操作函数

void atomic_store(volatile A * object, C desired); 
void atomic_store_explicit(volatile A * object, C desired, 
                        memory_order order order); 
C atomic_load(const volatile A * object); 
C atomic_load_explicit(const volatile A * object, 
                        memory_order order); 
C atomic_exchange(volatile A * object, C desired); 
C atomic_exchange_explicit(volatile A * object, C desired, 
                        memory_order order); 
_Bool atomic_compare_exchange_strong(volatile A * object, 
                        C * expected, C desired); 
_Bool atomic_compare_exchange_strong_explicit( 
                        volatile A * object, C * expected, 
                        C desired, memory_order success, 
                        memory_order failure); 
_Bool atomic_compare_exchange_weak(volatile A * object, 
                        C * expected, C desired); 
_Bool atomic_compare_exchange_weak_explicit( 
                        volatile A * object, C * expected, 
                        C desired, memory_order success, 
                        memory_order failure); 
C atomic_fetch_add(volatile A * object, M operand); 
C atomic_fetch_add_explicit(volatile A * object, M operand, 
                        memory_order order); 
C atomic_fetch_sub(volatile A * object, M operand); 
C atomic_fetch_sub_explicit(volatile A * object, M operand, 
                        memory_order order); 
C atomic_fetch_or(volatile A * object, M operand); 
C atomic_fetch_or_explicit(volatile A * object, M operand, 
                        memory_order order); 
C atomic_fetch_xor(volatile A * object, M operand); 
C atomic_fetch_xor_explicit(volatile A * object, M operand, 
                        memory_order order); 
C atomic_fetch_and(volatile A * object, M operand); 
C atomic_fetch_and_explicit(volatile A * object, M operand, 
                        memory_order order);

以上我们给出了10种函数，每一种函数都包括两个版本：不带_explicit后缀的版本和带有_explicit后缀的版本。这两个版本完成相同的功能，区别在于，不带_explicit后缀的版本使用默认的memory_order_seq_cst内存顺序，而带有_explicit后缀的版本要求显式地指定内存顺序。

原子操作函数所使用的内存顺序对编译器和处理器都有影响。首先，它决定了编译器能够在多大程度上对当前函数前后的指令进行优化和重排；其次，在函数执行时，它也会对处理器的乱序执行进行干预，这主要是解决线程之间的同步问题（28.2.7节）。

这些函数都是泛型函数，所以参数类型和返回类型只能按照前面28.2.3节的约定，用字母A和C泛指。注意，A是原子类型。

泛型函数atomic_store原子地用参数desired的值替换参数object所指向的原子对象的值。在带有_explicit后缀的版本中，参数order用来显式地指定内存顺序，但指定的内存顺序不能是memory_order_acquire、memory_order_consume和memory_order_acq_rel。函数执行时，将按照默认或者指定的内存顺序对处理器进行干预。

泛型函数atomic_load用来原子地读取并返回参数object所指向的对象的值。在带有_explicit后缀的版本中，指定的内存顺序不能是memory_order_release和memory_order_acq_rel。函数执行时，将按照默认或者指定的内存顺序对处理器进行干预。

泛型函数atomic_exchange及其带有_explicit后缀的版本执行交换操作：原子地用参数value的值替换参数object所指向的原子对象的值，并返回这个对象被替换前的原值。这两个函数执行的是“读-改-写”操作。函数执行时，将按照默认或者指定的内存顺序对处理器进行干预。

泛型函数atomic_compare_exchange_strong及其带有_explicit后缀的版本执行比较-交换操作。假定参数object指向原子对象ao，参数expected指向对象e，那么此函数执行以下原子操作：比较ao和e的值，如果相等则用参数desired的值替换ao的值，如果不相等则用ao的值去更新e的值。函数的返回值是比较的结果。

需要补充说明的是，如果比较的结果是相等，将按照参数success指定的内存顺序对处理器进行干预；如果不相等，则按照参数failure指定的内存顺序对处理器进行干预。

泛型函数atomic_compare_exchange_weak及其带有_explicit后缀的版本同样执行比较-交换操作。不同的是，比较的结果有可能是伪不相等。出现这种情况的原因是，举例来说，+0.0和-0.0在逻辑上是相等的，但它们具有不同的内部表示，所以这个函数将其按照不相等来处理。

在以atomic_compare_exchange_开头的原子操作函数中，为参数failure指定的内存顺序不能是memory_order_release和memory_order_acq_rel。这些函数执行的是“读-改-写”操作。

以atomic_fetch_开头的泛型函数及其带有_explicit后缀的版本原子地执行以下“读-改-写”操作：读参数object所指向的原子对象；用读来的值和参数operand的值一起做算术运算或者位运算（add是做加法，sub是做减法，or是逐位或，xor是逐位异或，and是逐位与）；运算的结果再写回object所指向的原子对象。

以上所有这些操作都适用于任何原子整数类型的对象，除了atomic_bool。有符号整数类型使用2的补码来表示其数据，运算溢出后将自动回绕。带后缀版本和不带后缀版本的区别仅在于使用的内存顺序是明确指定的还是默认的。这些函数统一返回修改之后的值，同时按照默认或者指定的内存顺序对处理器进行干预。对于以上函数，在本节中只需要关注其功能即可，不必纠结于它们的内存顺序。

下面这个程序atomoprs.c是atomic.c的改进版本（28.2.5节），同样是多个线程访问同一个共享变量，但这次使用了刚才介绍的原子操作函数。

/*
atomoprs.c
*/
# include <stdio.h> 
# include <threads.h> 
# include <stdatomic.h> 

_Atomic long long counter = 0; 

int thrd_proc1(void * arg) 
{ 
    struct timespec interv = {0, 20}; 
    for (size_t x = 0; x < 5000; x++) 
    { 
        long long old = atomic_load (& counter), new; 
        do 
        { 
            new = old + 1; 
        } while (! atomic_compare_exchange_weak (& counter, & old, new)); 
        thrd_sleep(& interv, 0); 
    } 
    return 0; 
} 

int thrd_proc2(void * arg) 
{ 
    struct timespec interv = {0, 30}; 
    for (size_t x = 0; x < 5000; x++) 
    { 
        long long old = atomic_load (& counter), new; 
        do 
        { 
            new = old - 1; 
        } while (! atomic_compare_exchange_weak (& counter, & old, new)); 
        thrd_sleep(& interv, 0); 
    } 
    return 0; 
} 

int main(void) 
{ 
    thrd_t t0, t1; 
    
    thrd_create(& t0, thrd_proc1, 0); 
    thrd_create(& t1, thrd_proc2, 0); 
    
    thrd_join(t0, & (int){0}); 
    thrd_join(t1, & (int){0}); 
    
    printf("%lld\n", counter); 
}

在这个程序中，和往常一样，线程thrd_proc1和thrd_proc2功能相反，只要搞懂了一个，另一个也就懂了。所以，接下来我们以线程thrd_proc1为例进行说明。需要注意的是，在程序中用到了原子操作函数，因此要包含<stdatomic.h>头。

线程thrd_proc1的工作是将共享变量counter反复加1。这件事本来很简单，因为counter是一个原子变量，使用复合赋值运算符+=就能轻松完成。但是我们不想用复合赋值运算符，而只想用简单赋值运算符=（好在不是所有程序员都这么任性），于是这就麻烦了。这是因为，在28.2.5节中已经说过，如果使用简单赋值运算符的话，需要三步：读、加1和赋值。而且只有读和赋值是原子的，在这两个原子操作中间，别的线程也可以访问counter并改变它。

不过没有关系，我们可以换一种思路，而且不在乎别的线程是否会来横插一杠子。如程序中所示，先用atomic_load函数原子地读变量counter的值到临时变量old，然后用一个do循环执行如下操作。

  1. 计算old加1的值。
  2. 将计算结果赋给临时变量new。
  3. 用atomic_compare_exchange_weak函数原子地执行比较-交换操作。它先看变量old和counter是否具有相同的值。如果相同的话，将变量new的值写入变量counter；如果不相同，说明在步骤(1)或者步骤(1)和(3)之间，别的线程改变了变量counter，此时只能放弃修改，并将counter的当前值读到old，然后返回到步骤(1)从新的起点上重新开始。

在以上过程中，第(3)步的atomic_compare_exchange_weak函数起了最重要也是最关键的作用，因为它将“比较-改/回读”封装为原子操作。在它执行期间，其他线程无法对counter进行操作。

和互斥锁相比，采用原子操作的优势是明显的：所有线程都可以全速开动，一个线程的执行不是以其他线程的阻塞为代价。使用互斥锁，如果持有锁的线程出了问题，其他线程也会因为得不到锁而一直阻塞；但是在这里不会，即使线程thrd_proc1出了问题，另一个线程thrd_proc2也会继续执行，并率先完成自己的工作。实际上，这也是锁无关程序设计的主要特征。

当然，正如我们已经看到的，锁无关的执行机制有可能是以增加线程内部的执行时间为代价的，因为这种工作方式的特点是重复尝试，大不了前面的工作白做，从头再来。

28.2.7 内存顺序

在前面我们已经谈到了高速缓存和多处理器（核）技术。从现在开始，你需要换一换脑子，摒弃一些传统的想法，比如“线程一直是交错执行的”，以及“处理器将顺序执行指令并按顺序得到结果”。来看一段代码：

a = 1;       //1
b = a + 3;   //2 
c = 7;       //3 
d = 9;       //4 
e = c + d;   //5

按照语法规则，先执行①，然后是②…最后是⑤。处理器遵循这个顺序，但只是按这个顺序将它们拆分成微操作，然后送入流水线。这样做是为了尽量填满流水线，从而提高工作效率，在单位时间里执行完更多的指令。

流水线中指令的执行是并行的，一旦进入了流水线，所有指令就具有了同时执行的特征：前面的指令还没有执行完毕，后面的指令就开始执行，这就造成了所谓的乱序执行现象。但是，指令的执行可能会卡在某些环节上。比如，要访问的数据不在高速缓存中，需要执行一个慢速的内存访问及高速缓存填充操作；再比如，当前指令的执行依赖于其他指令的执行结果。不管是什么原因，都可能导致前面的指令后执行完毕，后面的指令反而先执行完毕。

就上面这个例子来说，它们可能是按①、③、②、④、⑤的顺序依次执行完毕。注意，由于指令间的依赖关系，②不会在①之前执行完毕，⑤不会在③和④之前执行完毕。但是无论如何，从指令执行完毕的顺序来看，它们就像被处理器按下面的顺序重排过一样：

a = 1;       //① 
c = 7;       //③ 
b = a + 3;   //② 
d = 9;       //④ 
e = c + d;   //⑤

相比于处理器，编译器的指令重排才是真正的重排。如果在编译程序时启用了特定的优化，则编译器可能会在物理上调整指令间的顺序。

不管指令的重排是由于乱序执行引起的，还是由编译器安排的，对当前线程的执行结果都不会有任何影响。进一步地，在线程内部修改一个变量的值，不管它是当前线程内部私有的变量，还是在线程之间共享的变量，这个修改对当前线程的其他部分来说始终可见。但如果是一个共享变量，而且线程都位于不同的处理器（核）上，那么，在一个线程内的修改未必立即对另一个线程可见。原因很简单，直接写高速缓存和内存会导致一个处理器（核）间的同步（扩散）过程，这会影响到处理器的执行速度。为此，有些处理器会延缓向高速缓存和内存发布这些写入。相应地，向其他处理器（核）间的数据更新与同步也被延迟，并导致它们读到旧数据。

这里有一个例子：线程1和线程2共享变量x和a且它们的初始值为0，同时假定这两个线程运行在不同的处理器（核）上。

线程1             线程2
x = 1;           a = 1;
y = a;           b = x;

如果线程1和线程2同时运行，即使它们按照程序中的顺序执行，执行的结果也可能是违反直觉的：y和b的值有可能同时为0！在这里，线程1给变量x写1，但这个写入并未通知线程2所在的处理器，所以线程2读到了x的旧值0并写入变量b；线程2给变量a写1，这个写入也未及时通知线程1所在的处理器，所以线程1读到了a的旧值0并写入变量y。

除此之外，还有另一种可能的情况：即使变量值被改变的消息已经到达每一个处理器（核），它们也可能不会立即处理它。因此，不同的处理器（核）也不会在同一时间完成变量值的更新。

乱序执行、写操作的延迟，再加上变量值同步的延迟，将导致不同的处理器（核）在关于变量的值是否已经改变以及改变的顺序上有不同表现。

为了加深你的印象，让你进一步认识到这种混乱会给日常的程序设计带来什么麻烦，再来看一个例子。假定变量data和ready的初始值都为0，线程1给data和ready赋值，然后线程2等待ready的值变为非零后执行断言。这个断言会触发吗？

线程1             线程2
data = 68;       while(!ready);
ready = 1;       assert(data == 68);

答案是可能会，也可能不会。通常，如果这两个线程是在不同的处理器（核）上执行的，且实际执行的效果如同指令被重排成如下的顺序：

线程1             线程2
ready = 1;       while(!ready);
data = 68;       assert(data == 68);

那么，当线程2中的断言执行时，对变量data的修改可能还没有更新到线程2所在的处理器（核），或者data = 68还没有开始执行。

从表面上看，为了追求性能，处理器已经疯了，它不可能为线程之间共享变量并保持同步关系提供任何保障。但是请你把心放到肚子里，在把自己变成野马之前，它也为你提供了缰绳和鞭子。也就是说，你可以根据自己的需要和实际情况来控制这种混乱的程度，从而在性能和需求之间维持适当的平衡。这就产生了所谓的内存顺序。

内存顺序共有6种，它们在原子操作函数中用来执行内存的同步化操作。为了方便，
<stdatomic.h>头里将这6种内存顺序定义为枚举常量，并将它们指定为枚举类型memory_order的成员。表28-3中给出了这些枚举常量。

表28-3 内存顺序一览表

内存顺序语义
memory_order_relaxed具有“松散”的语义，没有同步效果
memory_order_release具有“发布”的语义
memory_order_acquire具有“获取”的语义
memory_order_consume具有“消费”的语义
memory_order_acq_rel兼具“发布”和“获取”的语义
memory_order_seq_cst具有顺序一致性语义

表中这6种内存顺序通常用于原子操作函数中（28.2.6节），使之除了完成数据的读写操作外，还按照“松散”“发布”和“获取”的语义对乱序执行进行干预。这里有一个例子或者说场景可以帮你快速入门：

假设在某个线程中，原子操作函数F用来写原子变量x且指定了具有发布语义的内存顺序，例如：

atomic_store_explicit (& x, 7, memory_order_release);  //F

再假设，另一个线程中，原子操作函数H用来读原子变量x且指定了具有获取语义的内存顺序，例如：

y = atomic_load_explicit (& x, memory_order_acquire);  // H

现在可以保证，如果H能够读到F写入x的值，那么，F之前的所有写操作，不管它们是原子的还是非原子的，都对H之后的读操作可见。

来看一个具体的例子。如下面的程序所示，线程1给data和ready赋值，然后线程2等待ready的值变为非零之后执行断言，这个断言会触发吗？

atomic_int ready = 0;
int data = 0;

// 线程1 
data = 68; 
atomic_store_explicit(& ready, 1, memory_order_release);

// 线程2 
while (! atomic_load_explicit(& ready, memory_order_acquire));
assert(data == 68);

注意!!对变量ready的读写都是用原子操作函数进行的，所以它必须是原子类型。在线程2中，while语句等待ready由0变成非0。一旦退出while循环，就可以断定我们也能读到data的新值68，因此断言一定不会触发。

很显然，为了保证线程2在看到ready改变时也能够得到data的新值，语句data=68;不允许重排到语句atomic_store_explicit(&ready,1,memory_order_release);之后。这种约束来自我们指定的memory_order_release。同理，为了保证一定能够用上（获取到）线程1发布的新值，语句assert(data==68)不允许重排到语句while(!atomic_load_explicit(&ready,memory_order_acquire));之前。这种约束来自我们指定的memory_order_acquire。

这6种内存顺序的作用大体上就是这样，它们之间的区别仅在于强度不同，下面我们将逐一进行介绍:

  • memory_order_release具有“发布”语义，而且只能用在执行“写”或者“读-改-写”功能的原子操作函数中，可认为它是用来“发布”前面写入的内容。这意味着，函数前的读写操作，不管它们是原子的还是非原子的，都不能重排到函数之后，但函数之后的可以重排到函数之前。也就是说，可以提前“发布”但不能推迟。如果其他线程能够看见此函数的写入，则它们也能看见函数前的写入。
  • memory_order_acquire具有“获取”语义，而且只能用在执行“读”或者“读-改-写”功能的原子操作函数中，可认为它之后的内容用来获取其他线程发布的结果。这意味着，函数前的读写操作，不管它们是原子的还是非原子的，都可以重排到函数之后，但函数之后的不允许重排到函数之前。也就是说，可以推迟“获取”但不允许提前。通常来说，在其他线程中应该有一个具有发布语义的原子写操作与之相对应。如果此函数能够看到那个写操作的结果，则之前的写操作也对当前线程可见。
  • memory_order_consume具有“消费”语义，和memory_order_acquire基本相同，都是“获取”性质的，但memory_order_consume稍微宽松一点：函数之后的读写操作也可以重排到函数之前，但前提是它们不依赖当前函数的返回值。通常来说，在其他线程中应该有一个具有发布语义的原子写操作与之相对应。如果此函数能够看到那个写操作的结果，则之前的写操作也对当前线程内与返回值有依赖关系的读写操作可见。

在下面的例子中，①和②都不会触发，因为它们都用到了变量d的值，而变量d的值来自具有获取语义的原子操作函数，所以形成了依赖关系。③可能会触发，因为它与d没有依赖关系，所以有可能被重排到while语句之前。

atomic_int ready = 0; 
int a = 0, b = 0, c = 0; 

// 线程1 
a = 1; 
b = 1; 
c = 1; 
atomic_store_explicit(& ready, 1, memory_order_release); 

// 线程2 
int d; 
while (! (d = atomic_load_explicit(& ready, memory_order_consume))); 
assert(d == a);  //① 
assert(d == b);  //② 
assert(c == 1);  //③
  • memory_order_acq_rel兼具发布和获取的语义，只能用于具有“读-改-写”性质的原子操作函数中。因为它既发布又获取，所以函数前的读写操作，不管是原子的还是非原子的，都不能重排到函数之后。函数后的读写操作，不管是原子的还是非原子的，也不能重排到函数之前。从另一方面来说，此函数之前的写操作对获取同一个原子变量的其他线程可见；同时，如果其他线程修改了同一个原子变量，则修改之前的其他写操作也对当前线程可见。
  • memory_order_seq_cst具有顺序一致性的语义，可用于任何原子操作函数。如果用在具有“写（存）”性质的原子操作函数中，则具有发布语义；如果用在具有“读（取）”性质的原子操作函数中，则具有获取语义；如果用在具有“读-改-写”操作的原子操作函数中，则兼具发布和获取语义。除此之外最重要的是，在这种内存顺序上还施加了一个单一全序（single total order）。那么，单一全序意味着什么呢？

首先，在一个线程中，所有使用了memory_order_seq_cst内存顺序的写操作，都会按它们在程序中的顺序执行，并按这个顺序同步（扩散）到其他处理器。因此，在所有处理器看来，这些写操作在这个线程内的顺序都是一致的。

其次，如果多个线程内都有使用了memory_order_seq_cst内存顺序的写操作，那么，不管哪个线程先执行，也不管线程之间如何交错执行，所有线程就都会观察到相同的执行顺序。来看一个例子：

atomic_int x = 0, y = 0; 

// 线程1 
atomic_store_explicit(& x, 1, memory_order_seq_cst); 

// 线程2  
atomic_store_explicit(& y, 1, memory_order_seq_cst); 

// 线程3 
assert(atomic_load_explicit(& x, memory_order_seq_cst) == 1 &&  
atomic_load_explicit(& y, memory_order_seq_cst) == 0); 

// 线程4 
assert(atomic_load_explicit(& x, memory_order_seq_cst) == 1 &&  
atomic_load_explicit(& y, memory_order_seq_cst) == 0);

在以上示例中，不管线程1和线程2按什么顺序执行，所有线程都能看到x和y的变化顺序，而且看到的顺序是一样的。假定这4个线程的执行顺序是１、３、２、４，那么，线程3将看到x为1、y为0，因此断言为真；顺序一致性保证线程4看到的顺序和线程3一样，所以线程3将看到x为1、y也是1，因此断言为假。

显然，获取操作能够看到的写操作依赖于发布操作，而且发布操作和获取操作的顺序并没
有保证，但顺序一致性操作的顺序则是可以保证的。但是，顺序一致性是以严重损失性能为代价的，因为它禁止任何潜在的硬件或软件优化。

在28.2.6节中，每个原子操作函数都有一个不带_explicit的版本，这个版本都默认使用memory_order_seq_cst的内存顺序。

同时，在C语言里，对原子变量的读操作和写操作（简单赋值、复合赋值、前后缀形式的递增和递减），都具有memory_order_seq_cst语义。所以，上面的例子其实还可以改成以下简单的形式：

atomic_int x = 0, y = 0; 

// 线程1 
x = 1; 

// 线程2  
y = 1; 

// 线程3 
assert(x == 1 && y == 0); 

// 线程4 
assert(x == 1 && y == 0);

这里，线程1和线程2中的赋值操作具有顺序一致性语义；线程3和线程4中，对原子变量x和y的读操作具有顺序一致性语义。

  • memory_order_relaxed具有“松散”的语义，可用在任何原子操作函数中。但是它不具有同步效果，所有读写操作的重排可以不受约束地随意进行。

在进入后面的内容之前，我们约定：如果原子操作函数在执行时具有“发布”语义，则它
执行的是“发布性的原子操作”；如果原子操作函数在执行时具有“获取”语义，则它执行的是“获取性的原子操作”。

28.2.8 围栏函数

void atomic_thread_fence(memory_order order); 
void atomic_signal_fence(memory_order order);

在前面的内容中，我们是基于原子操作函数来强化对乱序执行和指令重排的约束。但是，我们也可以在不修改任何数据的情况下，即不需要用上面的原子操作函数读写原子变量，就可以实现这种约束。为此，需要引入一个叫作“围栏”或者“内存屏障”的同步原语，它是通过围栏函数atomic_thread_fence和atomic_signal_fence来构建的。

围栏可以具有“发布”的语义，叫作发布围栏；也可以具有“获取”的语义，叫作获取围栏；或者兼而有之。具体是哪种语义，可以在构建围栏时，通过参数order来指定，如表28-4所示。

表28-4 内存围栏的类型

指定的内存顺序设置的围栏类型
memory_order_relaxed无同步效果
memory_order_release发布围栏
memory_order_acquire获取围栏
memory_order_consume获取围栏
memory_order_acq_rel获取和发布围栏
memory_order_seq_cst顺序一致性的获取和发布围栏

函数atomic_thread_fence用于线程间的同步；atomic_signal_fence用于线程内的同步（信号处理函数内外的同步）。先来看atomic_thread_fence函数，与它有关的同步包括以下几种情况。

  1. 发布围栏和获取性原子操作的同步

    如下例所示，假设线程1中有发布围栏F和原子操作X，且F前序于X；线程2中有获取性原子操作Y，且Y看到了X所写的值。在这种情况下，F和Y同步，所有前序于F的写操作，无论是原子的还是非原子的，都在Y之后可见。

    atomic_int ready = 0; 
int d = 0; 

// 线程1 
d = 1; 
atomic_thread_fence(memory_order_release);        // F 
atomic_store_explicit(& ready, 1, memory_order_relaxed);    // X 

// 线程2 
while (! atomic_load_explicit(& ready, memory_order_acquire));  // Y 
assert(d == 1); // 不会触发

  2. 发布性原子操作和获取围栏的同步

    如下例所示，假设线程1中有一个发布性原子操作X；线程2中有一个原子操作Y和一个获取围栏F，且Y前序于F。如果Y读到了X写入的值，则X和F同步。所有前序于X的写操作，无论是原子的还是非原子的，都在Y之后可见。

    atomic_int ready = 0; 
int d = 0; 

// 线程1 
d = 1; 
atomic_store_explicit(& ready, 1, memory_order_release);    // X 

// 线程2 
while (! atomic_load_explicit(& ready, memory_order_relaxed));  // Y 
atomic_thread_fence(memory_order_acquire);        // F 
assert(d == 1); // 不会触发

  3. 发布围栏和获取围栏的同步

    如下例所示，假设线程1中有一个原子操作X和一个发布围栏FR，且FR前序于X；在线程2中有一个原子操作Y和一个获取围栏FA，且Y前序于FA。如果Y读到了X所写的值，则所有前序于FR的写操作，无论是原子的还是非原子的，都在Y之后可见。

    atomic_int ready = 0; 
int d = 0; 

// 线程1 
d = 1; 
atomic_thread_fence(memory_order_release);        // FR 
atomic_store_explicit(& ready, 1, memory_order_relaxed);    // X 

// 线程2 
while (! atomic_load_explicit(& ready, memory_order_relaxed));  // Y 
atomic_thread_fence(memory_order_acquire);        // FA 
assert(d == 1); //不会触发

函数atomic_thread_fence用于达成线程之间的同步，通常这些线程在不同的处理器（核）上并行执行。相比之下，另一个函数atomic_signal_fence用于在线程内部构造同步条件。

如下例所示，thrd_proc是线程启动函数，用来创建一个线程。在线程的一开始，我们用signal函数将信号处理过程sig_handler与当前线程绑定，指向旧处理过程的指针被保存在oldsig中以便将来恢复。

接下来的工作是初始化全局变量w，如果w是一个很大的数组或者一个包含了很多成员的结构变量，那么对它的初始化可能要花一点时间。在此期间，可能会发生信号。我们知道信号是异步发生的，即你不知道它什么时候发生，所以它也可能在初始化变量w的过程中发生。为此，在信号处理函数内部可能需要知道变量w是否已经完成了初始化，并据此做适当的处理。

按照传统的方法，我们通常会用另外一个变量，比如这里的r作为标志。变量r的初始值为0，但我们会在完成变量w的初始化后将r置1。于是，在信号处理函数内部就可以根据r的值是0还是非0来判断变量w是否已经完成了初始化。但是，由于设置变量r的代码和初始化w的代码没有依赖关系，它可能被重排到前面。因此，在信号处理函数的内部和外部需要一个同步关系。

# include <stdio.h> 
# include <assert.h> 
# include <signal.h> 
# include <threads.h> 
# include <stdatomic.h> 

struct {int a; int b; int c;} w; 
atomic_int r = 0; 

void sig_handler(int arg) 
{ 
    if (atomic_load_explicit(& r, memory_order_relaxed)) 
    { 
        atomic_signal_fence(memory_order_acquire); 
        // 此处可以保证w 已经完整初始化 
    } 
} 

int thrd_proc(void * arg) 
{ 
    void (* oldsig) (int) = signal(SIGINT, sig_handler); 
    
    w.a = 3; 
    w.b = 7; 
    w.c = 9; 
    atomic_signal_fence(memory_order_release); 
    atomic_store_explicit(& r, 1, memory_order_relaxed); 
    
    // 处理其他事务 
    
    signal(SIGINT, oldsig); 
    return 0; 
}

如程序中所示，为了阻止这种可能的重排，初始化变量w的代码被放在一个发布围栏之前，对变量r的原子写操作被放在发布围栏之后。在信号处理函数内部，获取围栏用于确保在看到r的新值后，也能保证变量w已经初始化完成。

与atomic_thread_fence不同，atomic_signal_fence只在线程和线程内的信号处理函数之间建立同步，而且不使用处理器硬件指令构建同步围栏，它只是指令编译器不要将写操作移到发布围栏之后，或者将读操作移到获取围栏之前。除此之外，它们在别的方面是等效的，比如，都禁止编译期间的优化和代码重排。

28.2.9 锁无关判断函数

_Bool atomic_is_lock_free(const volatile A * obj);

锁无关的程序设计依赖于原子操作，原子操作需要处理器硬件指令的支持。但是处理器千差万别，在这方面的能力有强有弱，不是所有原子类型的原子操作都能够得到来自底层的支持，所以到头来有可能还必须借助于互斥锁。

泛型函数atomic_is_lock_free用来确定某个原子类型上的原子操作是不是锁无关的，这个类型取自参数obj所指向的对象。如果确定是锁无关的，则函数返回非零值。

为了方便，<stdatomic.h>头里还定义了一组宏，用来确定一些预定义原子类型的锁无关性，列举如下：

  • ATOMIC_BOOL_LOCK_FREE
  • ATOMIC_CHAR_LOCK_FREE
  • ATOMIC_CHAR16_T_LOCK_FREE
  • ATOMIC_CHAR32_T_LOCK_FREE
  • ATOMIC_WCHAR_T_LOCK_FREE
  • ATOMIC_SHORT_LOCK_FREE
  • ATOMIC_INT_LOCK_FREE
  • ATOMIC_LONG_LOCK_FREE
  • ATOMIC_LLONG_LOCK_FREE
  • ATOMIC_POINTER_LOCK_FREE

以上多数宏从名字上就可以看出它们的作用，比如ATOMIC_BOOL_LOCK_FREE表示原子_Bool类型的锁无关性。唯一需要说明的是ATOMIC_POINTER_LOCK_FREE，它代表任意类型指针的锁无关性。

这些宏都被定义为整数值，或者说它们都代表一个整数值。这个值如果是0，表明这种类型根本不是锁无关的；如果是1，表明它在某些时候是锁无关的；如果是2，表明它总是锁无关的。下面用一个例子来演示这些宏以及atomic_is_lock_free的用法。

void flockfree(void) 
{ 
    _Atomic struct {int x; float y;} a; 
    _Atomic struct {int x; float y; char c;} b; 
    _Atomic int c; 
    
    printf("%d.\n", ATOMIC_BOOL_LOCK_FREE); 
    printf("%s.\n", atomic_is_lock_free (& a) ? "yes" : "no"); 
    printf("%s.\n", atomic_is_lock_free (& b) ? "yes" : "no"); 
    printf("%s.\n", atomic_is_lock_free (& c) ? "yes" : "no"); 
    printf("%s.\n", atomic_is_lock_free ((_Atomic void *) 0) ? "yes" : "no"); 
}

在作者的机器上，输出是

2.
yes.
no.
yes.
no.

在你的机器上可能输出不一样的结果，但没有关系，那也是正确的。注意最后一行，我们传入的是空指针(_Atomic void *) 0，这是允许的。本质上，重点在于指针所指向的类型。

28.2.10 原子标志类型及其操作函数

_Bool atomic_flag_test_and_set(volatile atomic_flag * object); 
_Bool atomic_flag_test_and_set_explicit( 
                volatile atomic_flag * object, memory_order order); 
void atomic_flag_clear(volatile atomic_flag * object); 
void atomic_flag_clear_explicit(volatile atomic_flag * object, 
                memory_order order);

<stdatomic.h>头提供了一种原子类型atomic_flag，叫作原子标志类型。在所有原子类型中，只有它可以保证一定是锁无关的。原子标志类型用于提供经典的“测试和设置”功能。

函数atomic_flag_test_and_set及其带有_explicit后缀的版本用于原子地置位参数object所指向的原子标志并返回置位前的状态。因为这个过程是原子的，所以没有数据竞争。

这两个函数不会阻塞当前程序的执行，因此，对它们的使用完全依靠自律。通常在多线程的环境下，如果函数的返回值为真，说明原子标志原先就是置位的，可认为其他线程正拥有这个标志；如果返回值为假，说明原子标志原先是清零的，可认为本次操作使当前线程拥有了这个标志。而在此之前，没有线程拥有这个标志。

函数atomic_flag_clear及其带有_explicit后缀的版本用于原子地清零参数object所指向的原子标志。参数order不能是memory_order_acquire以及memory_order_acq_rel。

以上函数中，不带_explicit后缀的版本默认使用memory_order_seq_cst内存顺序。函数执行时，将依照默认或者指定的内存顺序对处理器进行干预。

下面这个程序spinlock.c是atomic.c的改进版本（28.2.5节），同样是多个线程访问同一个共享变量，但这次使用了原子标志及原子标志函数。从下面的程序可以看出，原子标志的使用很像互斥锁，但它不会阻塞线程，也没有线程调度方面的开销。原子标志虽然不是锁，但可以用来完成锁的功能，本程序就是用它实现了一个自旋锁。

/*
spinlock.c
*/
# include <stdio.h> 
# include <threads.h> 
# include <stdatomic.h> 

atomic_llong counter = 0; 
atomic_flag aflag = ATOMIC_FLAG_INIT; 

int thrd_proc1(void * arg) 
{ 
    struct timespec interv = {0, 20}; 
    
    for (size_t x = 0; x < 5000; x++) 
    { 
        while (atomic_flag_test_and_set(& aflag)); 
        counter = counter + 1; 
        atomic_flag_clear (& aflag); 
    
        thrd_sleep(& interv, 0); 
    } 
    
    return 0; 
} 

int thrd_proc2(void * arg) 
{ 
    struct timespec interv = {0, 30}; 
    for (size_t x = 0; x < 5000; x++) 
    { 
        while (atomic_flag_test_and_set(& aflag)); 
        counter = counter - 1; 
        atomic_flag_clear(& aflag); 
    
        thrd_sleep(& interv, 0); 
    } 
    
    return 0; 
} 

int main(void) 
{ 
    thrd_t t0, t1; 
    
    thrd_create(& t0, thrd_proc1, 0); 
    thrd_create(& t1, thrd_proc2, 0); 
    
    thrd_join(t0, & (int){0}); 
    thrd_join(t1, & (int){0}); 
    
    printf("%lld\n", counter); 
} 
/*
output:
0
*/

由于你对这个程序已经熟得不能再熟，所以我们直接进入重点。在程序一开始，我们声明了原子标志类型的变量aflag并用ATOMIC_FLAG_INIT进行初始化。ATOMIC_FLAG_INIT是一个宏，是在头<stdatomic.h>里定义的，它用于初始化原子标志对象并使之处于清零状态。这是很重要的，如果原子标志对象没有明确地用ATOMIC_FLAG_INIT初始化，那么它将处于不确定的状态。

因为表达式counter = counter + 1和counter = counter – 1的求值过程都不是原子操作，所以必须避免它们并行执行。与以往不同，这次我们用一个原子标志来实施警戒。

函数atomic_flag_test_and_set是非阻塞的，但它会告诉我们aflag原先是什么状态。如果它返回“真”，说明原子标志原先就是置位的，可认为其他线程正拥有这个标志，所以不能做会导致数据竞争的事。如果没有什么别的事可做，那么，就像程序中所做的那样，我们可以用while循环反复执行这个函数，直到返回值为“假”。注意，尽管这是一个循环，而且这个函数执行原子操作，但在这个函数的某一次执行和它的下一次执行期间，其他线程可能已经改变了原子标志的状态，也就是已经将它清零。

在while循环中，如果函数atomic_flag_test_and_set的返回值为假，说明原子标志原先是清零的（其他线程刚刚将这个标志清零）。在这种情况下，当前线程便拥有了这个标志（因为已经将它置位）。只要其他线程也使用同样的策略工作，就可以保证现在能够安全地对变量counter进行操作而不会引发数据竞争。当然，在完成工作之后，还必须用函数atomic_flag_clear将原子标志清零，这样其他处于监视状态的线程就有机会将它置位并自然地获得执行机会。

函数atomic_flag_test_and_set及其带_explicit后缀的版本是非阻塞的，所以，如果它的返回值是“真”，当前线程可以选择先去做别的事，或者在原地循环等待。这很像是在原地打转消磨时间，故称之为“自旋”。用这种方法来实现互斥锁的功能，叫作自旋锁。具体如何选择，要视情况而定。就像去坐火车，如果还有半个小时开车，你可以在候车室转转消磨时间；如果还有5个小时才开车，你完全可以出去逛逛商场，或者离家不远的话，回去睡一觉。

不管如何选择，当前线程都不会被阻塞，所以不会有线程调度的开销，而只会增加处理器
的执行时间。这也意味着，为了避免让其他线程因等待而浪费过多的处理器时间，与数据竞争有关的事务处理应尽量简短。

问与答

问1：除了没有return语句，本章中的程序既没有判断编译器是否支持多线程和原子操作，也没有在调用函数时检测其返回值。这是为什么？

答：除了main函数返回时没有return语句，我们也没有用预处理指令判断编译器是否定义了宏__STDC_NO_THREADS__以及__STDC_NO_ATOMICS__。非但如此，我们在调用库函数时都没有根据返回值做对应的处理工作。首先，支持C99的编译器允许main函数没有return语句；同时，如果编译器支持C11的多线程和原子特性，说明它是必然支持C99的。对于所有已知的C编译器来说，如果它不支持多线程和原子操作，则它也不会提供<threads.h>和<stdatomic.h>头，而且在编译程序时会报告错误，提示我们它找不到这两个头文件。如果能够在程序开头用预处理器指令检测这两个宏是否定义，并在编译器不支持多线程和原子操作时报告错误，那当然更好。至于没有检测库函数的返回值，这样做是为了让程序简单明了，让读者更快地理解程序的功能和意图。当然，我在此要求读者在验证这些程序时，一定要加上这些内容。

问2：我可以把++、--和复合赋值运算符叫作原子操作运算符吗？

答：在不那么严格的场合，这样叫也没什么问题，但并不是所有人都认同。有些人觉得，这些运算符并不是原子操作运算符，但它用于原子变量时，具有单一操作的性质。事实上，就倾向性而言，C11的原子操作是指原子操作库函数。至于这些运算符，只是表明它们具有memory_order_seq_cst语义，但没有明确它们归类于原子操作。除了界定不清而引发的一致性问题，实际上，在C11和C18中还存在其他一些与原子操作有关的问题。ISO的C语言工作组（WG14）在他们的工作文档N2389中做了必要的澄清和修正，2019年的伦敦会议同意了这些非规范性变化，并可能出现在下一个标准中。

写在最后

本文是博主阅读《C语言程序设计：现代方法（第2版·修订版）》时所作笔记的最后一篇。在此感谢各位长久以来的支持，希望大家能够从我的文章中受益，解答疑惑，扫清盲点，博主日后会继续努力，坚持创作，把所学的微薄知识，写成文章分享给大家，Thank you very much!

文章来源:https://blog.csdn.net/New_Teen/article/details/134909408
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。