第十二章：并发编程

《CSAPP》第十二章：并发编程

---

更详细的参考《Java并发编程实战》

并发不仅仅是操作系统内核用来运行多个应用程序的机制，在应用程序中也扮演很重要的角色，应用级并发的作用

• 访问慢速I/O设备：内核遇到应用在等到I/O数据时，会运行其他进程，使CPU保持繁忙，应用程序也可以模仿类似思路，交替执行I/O请求和其他操作
• 与人交互：现代视窗系统通过并发实现用户可以在打印文档的同时，调整窗口大小，实现更好的人机交互
• 通过推迟工作以降低延迟：应用程序通过推迟其他操作和并发地执行它们，利用并发来降低某些操作的延迟
• 服务多个网络客户端：这允许服务器同时服务多个客户端，避免慢速客户端独占服务器
• 在多核计算机上进行并行计算

使用应用级并发的程序成为并发程序，现代操作系统提供三种基本的构造并发程序的方法

• 进程：每个逻辑控制流都是一个进程，由内核来调用和维护，因为进程由独立的虚拟地址空间，控制流需要显式的进程间通信机制来和其他逻辑控制流通信
• I/O多路复用：应用程序在一个进程的上下文中显式地调度它自己的逻辑流？？？
• 线程：线程是运行在单一进程上下文中的逻辑流，由内核调度，可以把线程看做上述两种方法的混合，像进程流一样由内核调度，又像I/O多路复用一样享受同一个虚拟地址空间

基于进程

构造并发服务器最自然的方法就是在父进程接受客户端连接请求，然后创建一个新的子进程来为每个新客户端提供服务。使用熟悉的函数，如fork exec waitpid

进程有独立的地址空间既是优点又是缺点，优点是进程不可能覆盖另一个进程的虚拟内存，缺点则是使得进程共享状态信息变得困难。为了共享信息，必须使用显式的IPC机制，除此之外进程控制和IPC（进程间通信）开销也大

#include "csapp.h"

void sigchld_handler(int sig)
{
        while(waitpid(-1,0,WNOHANG) > 0)
                ;
        return;
}

int main(int argc,char **argv)
{
       
        ...

        Signal(SIGCHLD,sigchld_handler);
        listenfd = Open_lisentfd(argv[1]);
        while(1){
                clientlen = sizeof(struct sockaddr_storage);
                connfd = Accept(listenfd,(SA *) &clientaddr,&clientlen);
                if(Fork() == 0){
                        Close(listenfd);
                        echo(connfd);
                        Close(connfd);
                        exit(0);
                }
                Close(connfd);
        }
}

void echo(int connfd)
{	...	}

父进程也就是服务器派生一个子进程，这个子进程获得服务器描述符表的完整副本，子进程要关闭监听描述符，父进程要关闭已连接描述符，如果不及时关闭已连接描述符，将导致内存泄漏，是系统崩溃

服务器会运行很长时间，因此需要一个SIGCHLD处理程序，来回收僵死子进程的资源，因为SIGCHLD处理程序执行时，SIGCHLD信号是阻塞的，而Linux信号是不排队的，所有该处理程序必须准备好回收多个僵死子进程的资源、

UNIX IPC机制包括信号和waitpid()、套接字接口、管道、先进先出、系统V共享内存、系统V信号量等

基于I/O多路复用

当服务器同时遇到用户从标准输入键入的交互命令和客户端发起的连接命令是，该如何选择呢？

基本思路就是使用select函数，要求内核挂起进程，只有在一个或多个I/O事件发生后，才将控制返回给应用程序

#include "csapp.h"

void echo(int connfd){...}
void command(void){...}

int main(int argc,char **argv)
{
        ...

        listenfd = Open_lisentfd(port);

        FD_ZERO(&read_set);             // clear read set
        FD_SET(STDIN_FILENO,&read_set); // add stdin to read set
        FD_SET(listenfd,&read_set);     // add listenfd to read set

        while(1){
                ready_set = read_set; 
                Select(listenfd+1,&ready_set,NULL,NULL,NULL);

                if(FD_ISSET(STDIN_FILENO,&ready_set))
                        command();      // read command line from stdin
                if(FD_ISSET(listenfd,&ready_set)){
                        clientlen = sizeof(struct sockaddr_storage);
                        connfd = Accept(listenfd,(SA *)&clientaddr,&clientlen);
                        echo(connfd);   // echo client input util EOF
                        Close(connfd);
                }
        }
}

select函数会处理类型为fd_set的集合，称为描述符集合，函数会一直阻塞，直到该集合至少有一个描述符准备好可以读

上面实现了基于I/O多路复用的迭代服务器，下面则是基于I/O多路复用的并发服务器

基于I/O多路复用的并发事件驱动服务器

I/O多路复用可以作为并发事件驱动程序的基础，在事件驱动程序中，某些事件会导致流向前推进。事件驱动模型本质上就是状态机。对于一个新的客户端$k$，此服务器会创建一个新的状态机$s_k$，并将其与已连接描述符$d_k$联系起来

服务器使用I/O多路复用，借助select函数检测输入事件的发生，当每个已连接描述符准备好可读时，服务器就为相应的状态机执行转移，在这里就是从描述符读写一个文本行

#include "csapp.h"

typedef struct{ /* represents a pool of connected descriptors */
        int maxfd; // largest descriptor in read_set
        fd_set read_set;// set of all active descriptors
        fd_set ready_set;
        int nready;
        int maxi;
        int clientfd[FD_SETSIZE];
        rio_t clientrio[FD_SETSIZE];
}pool;

int byte_cnt = 0; // counts total bytes received by server 

int main(int agrc,char **argv)
{
        int listenfd,connfd;
        int post;
        socklen_t clientlen;
        struct sockaddr_storage clientaddr;

        static pool pool;

        port = atoi(argv[1]);

        if(argc != 2){
                fprintf(stderr,"usage:%s <port>\n",argv[0]);
                exit(0);
        }

        listenfd = Open_listenfd(port);
        init_pool(listenfd,&pool);

        while(1){
                /* wait for listening/connected descriptor to become ready */
                pool.ready_set = pool.read_set;
                pool.nready = Select(pool.maxfd+1,&pool.ready_set,NULL,NULL,NULL);

                /* if listening descriptor ready,add new client to pool */
                if(FD_ISSET(listenfd,&pool.ready_set)){
                        clientlen = sizeof(struct sockaddr_storage);
                        connfd = Accept(listenfd,(SA *)&clientaddr,&clientlen);
                        add_client(connfd,&pool);
                }

                /* echo a text line from each ready connected descriptor */
                check_clients(&pool);
        }

}

在上述代码中，pool结构里维护着活动客户端的集合，在调用init_pool之后，服务器进入无限循环。在每次循环中，服务器调用select函数来检测两种不同类型的输入事件

• 来自一个新客户端的连接请求到达
• 一个已存在的客户端的已连接描述符准备好可以读

当连接请求到达时，服务器打开连接，并调用add_client函数，将该客户端加入池里，最后调用check_clients，把来自每个准备好的已连接描述符的一个文本行送回去

void init_pool(int listenfd,pool *p)
{
        int i;
        p->maxi = -1;
        for(i = 0;i < FD_SETSIZE;i++)
                p->clientfd[i] = -1;

        p->maxfd = listenfd;
        FD_ZERO(&p->read_set);
        FD_SET(listenfd,&p->read_set);
}

上述函数初始化客户端池，clientfd数组表示已连接描述符集合，-1表示可用槽位，此时监听描述符是最大且唯一的描述符

void add_client(int connfd,pool *p)
{
        int i;
        p->nready--;
        for(i = 0;i < FD_SETSIZE;i++)
                if(p->clientfd[i] < 0){
                        p->clientfd[i] = connfd;
                        Rio_readinitb(&p->clientrio[i],connfd);

                        FD_SET(connfd,&p->read_set);

                        if(connfd > p->maxfd)
                                p->maxfd = connfd;
                        if(i > p->maxi)
                                p->maxi = i;
                        break;
                }
        if(i == FD_SETSIZE)
                app_error("add_client error:Too many clients");
}

上述函数添加一个新的客户端到活动客户端中，在clientfd数组中找到空的槽位，就将已连接描述符添加到数组中，并初始化RIO读缓冲区，然后将已连接描述符添加到读集合中，并更新全局数值

void check_clients(pool *p)
{
        int i,connfd,n;
        char buf[MAXLINE];
        rio_t rio;

        for(i = 0;(i <= p->maxi) && (p->nready > 0);i++){
                connfd = p->clientfd[i];
                rio = p->clientrio[i];

                if((connfd > 0) && (FD_ISSET(connfd,&p->ready_set))){
                        p->nready--;
                        if((n = Rio_readlineb(&rio,buf,MAXLINE)) != 0){
                                byte_cnt += n;
                                printf("Server received %d (%d total) bytes on fd %d\n",
                                                n,byte_cnt,connfd);
                                Rio_writen(connfd,buf,n);
                        }

                        else{
                                Close(connfd);
                                FD_CLR(connfd,&p->read_set);
                                p->clientfd[i] = -1;
                        }
                }
        }
}

上述函数回送来自每个准备好的已连接描述符的一个文本行，还维护了一个从所有客户端接受全部字节的累计值

上面说到该服务器是事件驱动的，也就是状态机，对应的，select函数检测到输入事件，add_client创建新的逻辑流（状态机），check_clients回送输入行，执行状态转移，并在完成文本行发送后，删除该状态机

现代高性能服务器（如Node.js、Nginx）使用的都是基于I/O多路复用的事件驱动的编程方式，虽然有其缺点，但是相比于进程和线程，它有明显的性能优势，因为不需要进程上下文切换来调度新的流

使用事件驱动设计的一个好处在于比基于进程设计给了程序员更多的对程序行为的控制，另一个好处在于只运行单一进程的上下文中，每个逻辑流都能访问该进程的全部地址空间，这就可以使用GDB来调试并发服务器。缺点则是编码复杂和不能充分使用多核处理器

参见APUE的

基于线程

是上述两种方法的混合，线程就是运行在进程上下文中的逻辑流。线程上下文的切换比进程切换快很多，其次线程不按照进程一样有严格的父子层次，和一个进程有关的线程组成一个线程池，线程池中的线程可以杀死他的任何对等线程，或者等待他的任意线程终止

Posix线程（Pthreads）是在C语言中处理线程的一个标准接口，定义了约60个函数，运行程序创建、杀死和回收线程，与对等线程安全的共享数据，还可以通知对等线系统状态的变化。下面是最简单的线程程序

#include "csapp.h"

void *thread(void *vargp);

int main()
{
        pthread_t tid;
        Pthread_create(&tid,NULL,thread,NULL);
        Pthread_join(tid,NULL);
        exit(0);
}

void *thread(void *vargp)
{
        printf("Hello world!\n");
        return NULL;
}

创建线程：pthread_create(pthread_t *tid,pthread_attr_t *attr,func *f,void *arg)，arg是输入变量，f是线程例程，attr可以修改新创建的线程的默认属性

终止线程有多种方式

• 顶层线程例程返回，线程隐式终止
• pthread_exit(void *thread_return)：显式终止，主线程调用该函数，他会等待所有其他对等线程终止，然后在终止主线程和整个进程
• 对等线程调用exit函数，该函数终止进程以及所有与进程相关的线程
• pthread_cancel(pthread_t tid)：对等线程调用该函数，终止线程ID为tid的线程

回收资源：pthread_join(pthread_t tid,void **thread_return)，该函数会阻塞，直到线程tid终止，将线程例程返回的通用指针赋值给thread_return指向的位置，然后回收已终止线程占用的所有内存资源

分离线程：pthread_detach(pthread_t tid)，一个可结合的线程能够被其他线程回收和杀死，再被其他线程回收之前，它的内存资源是不释放的，一个分离的线程不能被其他线程回收或杀死，内存资源由系统自动释放，该函数分离可结合线程tid。为了防止内存泄漏，线程要么被其他线程显式回收，要么通过调用该函数被分离，由系统来回收。在Web服务器中通常采取这种方式回收线程，因为没有必要显式地等待每个对等线程终止

初始线程：pthread_once(pthread_once_t *once_control,void (*init_routine)(void))，初始化与线程例程相关的状态

基于线程的并发服务器

#include "csapp.h"

void echo(int connfd);
void *thread(void *vargp);

int main(int argc,char **argv)
{    
        int listenfd,*connfdp;
        socklen_t clientlen;
        struct sockaddr_storage clientaddr;
        pthread_t tid;
     
        if(argc != 2){
                fprintf(stderr,"usage:%s <port>\n",argv[0]);
                exit(0);
        }

        listenfd = Open_listenfd(atoi(argv[1]));

        while(1){
                clientlen = sizeof(struct sockaddr_storage);
                connfdp = Malloc(sizeof(int));
                *connfdp = Accept(listenfd,(SA *) &clientaddr,&clientlen);
                Pthread_create(&tid,NULL,thread,connfdp);
        }
}

void *thread(void *vargp)
{
        int connfd = *((int*)vargp);
        Pthread_detach(pthread_self());
        Free(vargp);
        echo(connfd);
        Close(connfd);
        return NULL;
}

在调用Pthread_create，通过传递一个指向已连接描述符的指针*connfdp，将描述符传递给对等线程

但是这样做却会引入竞争，试想一下，如果对象线程中的赋值在下一个Accept之前完成，那局部变量将得到正确的值，但是在之后完成，就会得到下一个连接的描述符，这样两个线程就会使用同一个描述符，为此，我们将accept返回的每个已连接描述符分配到它自己的动态分配的内存块？？？

另一个要注意的地方就是分离线程，避免内存泄漏，还要释放主线程分配的内存块

参见APUE的

多线程的共享变量

线程的吸引力就在于多个线程能共享程序变量，但是为了编写正确的多线程程序，就需要对所谓的共享如何工作有清楚地了解

一组并发线程运行在一个进程的上下文中，每个线程都有自己独立的线程上下文，包括线程ID、栈、栈指针、程序计数器、条件码和寄存器值，并共享其余部分，包括只读文本、读写数据、堆、共享库代码和打开文件

线程内存模型参见《程序员的自我修养》

多线程的C程序中变量根据存储类型被映射到虚拟内存

• 全局变量：虚拟内存的读/写区域只包含每个全局变量的一个实例，任何线程都可以引用
• 本地自动变量：定义在函数内部但是没有static属性的变量，每个线程的栈包含自己的所有本地自动变量实例
• 本地静态变量：定义在函数内部且有static属性的变量，和全部变量一样，虚拟内存的读/写区域只包含一个实例

共享变量就是当它的实例被一个以上的线程引用，例如在多线程程序下的全局变量

信号量

互斥

共享变量十分方便，却引入了同步错误的可能性。一般而言，不能预测OS是否将为线程选择一个正确的顺序

为此引入了叫做信号量（$s$，semaphore）的特殊类型变量，是具有非负整数的全局变量，只能有两种特殊操作处理

• $P(s)$：如果$s$是非零的，将其减一，并立即返回，如果为零，就挂起这个线程，直到$s$变为非零，
• $V(s)$：将$s$加一，如果任何线程阻塞在$P$等待$s$变为非零，就重启这些线程中的某一个且是不可预测的

$P$和$V$操作都是不可分割的，这使得信号量$s$绝不可能变为负值，这种属性称为信号量不变性，$P$和$V$来自荷兰语测试和增加

信号量提供一种方便的方法来确保对共享变量的互斥访问，基本思想就是将每个共享变量和一个信号量联系起来，然后使用$P(s)$和$V(s)$将相应的临界区包围起来

这种信号量又称为二元信号量，因为它的值总是0或1，也常叫做互斥锁，相应的$P$和$V$操作就叫做加锁和解锁

调度共享资源

信号量的另一个作用是调度对共享资源的访问，在这种场景中，一个线程用信号量操作来通知另一个线程，程序状态的某个条件已经为真。经典例子就是生产者-消费者和读者-写者问题

生产者-消费者

生产者和消费者线程共享一个有$n$个槽的有限缓冲区，生产者线程反复生成新的项目，并插入缓冲区，消费者线程不断从缓冲区取出这些项目

为此，我们要保证对缓冲区的访问是互斥的，还有调度对缓冲区的访问，例如如果缓冲区是满的，就要挂起生产者线程

生产者-消费者模型很普遍，例如生产者编码视频帧，消费者解码并在屏幕上呈现出来，缓冲区的目的就是为了减少视频流的抖动

typedef struct{
    int *buf;   /* buffer array */
    int n;      /* maximun number of slots */
    int front;  /* buf[(front+1)%n] is first item */
    int rear;   /* buf[rear%n] is last item */
    sem_t mutex;/* protects accesses to buf */
    sem_t slots;/* counts available slots */
    sem_t items;/* counts available items */
}sbuf_t;

上面定义了sbuf_t类型的缓冲区，项目存放在动态分配的整数数组buf中，front和rear记录数组的第一项和最后一项，三个信号量分别提供对缓冲区、空槽位和已用项目数量的互斥访问

读者-写者

该问题是互斥问题的一个概括。一组并发的线程要访问一个共享对象，有些线程只读对象，叫做读者；有些线程只修改对象，叫做写者。写者必须拥有对对象的独占的访问，读者可以和无限个读者共享对象

该模型也很常见，例如航空预定系统中，允许无限个用户查看座位，但是正在预定的写者必须拥有对数据库的独占的访问，在多线程缓存web代理中也是相同的模型，无限个线程从共享页面缓存中取出已有的页面，但向缓存写入新页面的线程必须独占的访问

该线程也有多个变种，分别区别于读者和写者的优先级。如果读者优先，要求不要让读者等待，读者不会因为有一个写者在等待而等待，第二类，则是写者优先，要求一旦一个写者准备好可以写，就会尽可能完成写操作，如果有读者到来，就必须等待，即使写者也是等待的

Java线程用Java Monitor的机制来同步，它是对信号量互斥和调度能力的更高级别的抽象，具体参见第21章 并发

基于预线程化的并发服务器

服务器由一个主线程和一组工作者线程构成，主线程不断接受来自客户端的连接请求，并将得到的连接描述符放在一个有限缓冲区，每个工作线程反复从共享缓冲区取出描述符，为客户端服务

编译的时候一直显示重复定义，都要烦死了，最后发现自己好像修改过csapp.h文件，自己加了#include "csapp.c"进去作死，而且应该是和csapp.c文件一起编译链接，笨死了！

int main(int argc, char **argv)
{
    int i, listenfd, connfd, port;
    socklen_t clientlen=sizeof(struct sockaddr_in);
    struct sockaddr_in clientaddr;
    pthread_t tid;

    if (argc != 2) {
        fprintf(stderr, "usage: %s <port>\n", argv[0]);
        exit(0);
    }
    port = atoi(argv[1]);
    sbuf_init(&sbuf, SBUFSIZE); //line:conc:pre:initsbuf
    listenfd = Open_listenfd(port);

    for (i = 0; i < NTHREADS; i++)  /* Create worker threads */ //line:conc:pre:begincreate
        Pthread_create(&tid, NULL, thread, NULL);               //line:conc:pre:endcreate

    while (1) {
        connfd = Accept(listenfd, (SA *) &clientaddr, &clientlen);
        sbuf_insert(&sbuf, connfd); /* Insert connfd in buffer */
    }
}

在调用sbuf_init初始化缓冲区之后，创建一组工作线程，然后进入无限循环

static int byte_cnt;
static sem_t mutex;

static void init_echo_cnt(void)
{
        Sem_init(&mutex,0,1);
        byte_cnt = 0;
}

void echo_cnt(int connfd)
{
        int n;
        char buf[MAXLINE];
        rio_t rio;
        static pthread_once_t once = PTHREAD_ONCE_INIT;

        Pthread_once(&once,init_echo_cnt);
        Rio_readinitb(&rio,connfd);
        while((n = Rio_readlineb(&rio,buf,MAXLINE)) != 0){
                P(&mutex);
                byte_cnt += n;
                printf("server received %d (%d total) bytes on fd %d\n",
                                n,byte_cnt,connfd);
                V(&mutex);
                Rio_writen(connfd,buf,n);
        }
}

真正有趣的是上面这段代码，全局变量byte_cnt记录从所有客户端接受的所有累计字节数，先调用Pthread_once(&once,init_echo_cnt);，要求主线程显式地调用一个初始化函数，byte_cnt作为共享变量，是被$P$和$V$保护着的

并行性

在多核处理器中，OS在多个核中并行的调度并发程序，并行程序是运行在多个处理器的并发程序，因此并行程序是并发程序集合的真子集

作为示例，为$0,…,n-1$个数求和，每个线程负责一个区域

最为简单的就是把线程的和放在一个共享全局变量gsum中，用互斥锁保护

void *sum_mutex(void *vargp)
{
        long myid = *((long *)vargp);
        long start = myid * nelems_per_thread;
        long end = start + nelems_per_thread;
        long i;

        for(i = start;i < end;i++){
                P(&mutex);
                gsum += i;
                V(&mutex);
        }

        return NULL;
}

这种方式速度很慢，而且核数越多，性能越差，因为同步操作（$P$和$V$）代价太大，甚至出现了错误，原因是因为要假设整数$n$假设为线程数的倍数

另一种方式就是每个对等线程在一个私有变量计算它的部分和，不和其他线程共享

void *sum_array(void *vargp)
{
        long myid = *((long *)vargp);
        long start = myid * nelems_per_thread;
        long end = start + nelems_per_thread;
        long i;

        for(i = start;i < end;i++){
                psum[myid] += i;
        }

        return NULL;
}

虽然只改动了很小的一部分，但性能的提升是数量级的

写并发程序并不是轻而易举的事情，要避免错误，还要兼顾效率

其他

同步从根本上来说是很难的问题，这一节论述在编写并发程序中会遇到的典型问题，如线程安全、可重入性、竞争、死锁等，更详细的同步可参见《Java并发编程实战》

线程安全

一个函数是线程安全的，当且仅当被多个并发线程反复地调用时，一直产生正确的答案

下面是几类线程不安全的函数

• 不保护共享变量的函数。这类函数只要引入$P$和$V$，就能变成线程安全，缺点就是同步操作造成的代价
• 保持跨越多个调用的状态的函数。？？
• 返回指向静态变量的指针的函数。如果并发线程调用这些函数，一个线程使用的结果可能会被另一个线程悄悄覆盖
• 调用线程不安全函数的函数

可重入

当它们被多个线程调用时，不会引用任何共享数据。可重入函数是线程安全函数的子集

如果所有函数参数都是传值传递的（即没有指针），且所有数据引用都是本地自动栈变量，那么该函数就是显式可重入的。如果调用线程小心传递指向非共享数据的指针，那就是隐式可重入的

可重入性即是调用者属性也是被调用者的属性？？？

大部分Linux函数都是线程安全的，如果要调用线程不安全的函数，可以使用以_r结尾的安全版本，最好的办法就是加锁-复制？？？

竞争

通常发生竞争都是因为程序员假定线程将按照某种特殊的轨迹穿过执行状态空间，但是多线程的程序必须对任何可行的轨迹线都安全

#include "csapp.h"
#define N 100

void *thread(void *vargp);

int main()
{
        pthread_t tid[N];
        int i;

        for(i = 0;i < N;i++)
                Pthread_create(&tid[i],NULL,thread,&i);

        for(i = 0;i < N;i++)
                Pthread_join(tid[i],NULL);

        exit(0);
}

void *thread(void *vargp)
{
        int myid = *((int *) vargp);
        printf("hello from thread %d\n",myid);
        return NULL;
}

主线程在创建对等线程的时候，传入了一个指向本地栈变量i的指针，问题就出在线程如果在i++之前就引用了该值，就会获得其他线程的ID

上面是N=4，下面是N=100，每次运行结果都不尽相同

死锁

信号量引入了潜在的运行时错误死锁，它指的是一组线程被阻塞了，等待一个永远不会为真的条件。进程图对于理解死锁是无价的工具

死锁是不可避免，且很难复现的，下面的规则能有效避免死锁

• 互斥锁加锁顺序规则：给定所有互斥操作的一个全序，如果每个线程都以一种顺序获得互斥锁并以相反的顺序释放，那就是无死锁的？？