处理 SIGPIPE

　　在网络编程中经常会遇到SIGPIPE信号，默认情况下这个信号会终止整个进程，当然你并不想让进程被SIGPIPE信号杀死。我们不禁会这样思考：

• 在什么场景下会产生SIGPIPE信号？
• 要怎样处理SIGPIPE信号？

　　SIGPIPE产生的原因是这样的：如果一个 socket 在接收到了 RST packet 之后，程序仍然向这个 socket 写入数据，那么就会产生SIGPIPE信号。
　　这种现象是很常见的，譬如说，当 client 连接到 server 之后，这时候 server 准备向 client 发送多条消息，但在发送消息之前，client 进程意外奔溃了，那么接下来 server 在发送多条消息的过程中，就会出现SIGPIPE信号。下面我们看看 server 的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

#include <stdio.h>
#include <string.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <errno.h>
#include <signal.h>
#define MAXLINE 1024
void handle_client(int fd)
{
    // 假设此时 client 奔溃, 那么 server 将接收到 client 发送的 FIN
    sleep(5);
    // 写入第一条消息
    char msg1[MAXLINE] = {"first message"}; 
    ssize_t n = write(fd, msg1, strlen(msg1));
    printf("write %ld bytes\n", n);
    // 此时第一条消息发送成功，server 接收到 client 发送的 RST
    sleep(1); 
    // 写入第二条消息，出现 SIGPIPE 信号，导致 server 被杀死
    char msg2[MAXLINE] = {"second message"};
    n = write(fd, msg2, strlen(msg2));
    printf("%ld, %s\n", n, strerror(errno));
}
int main()
{
    unsigned short port = 8888;
    struct sockaddr_in server_addr;
    bzero(&server_addr, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    server_addr.sin_port = htons(port);
    int listenfd = socket(AF_INET , SOCK_STREAM , 0);
    bind(listenfd, (struct sockaddr *)&server_addr, sizeof(server_addr));
    listen(listenfd, 128);
    int fd = accept(listenfd, NULL, NULL);
    handle_client(fd);
    return 0;
}

　　我们可以使用 Linux 的 nc 工具作为 client，当 client 连接到 server 之后，就立即杀死 client (模拟 client 的意外奔溃)。这时可以观察 server 的运行情况：

1
2
3
4
5
6
7

$ gcc -o server server.c 
$ ./server &              # 后台运行 server
$ nc localhost 8888       # 运行 nc 连接到 server
^C                        # Ctrl-C 杀死 nc
write 13 bytes
[1]+  Broken pipe             ./server

　　让我们分析一下整个过程：

• client 连接到 server 之后，client 进程意外奔溃，这时它会发送一个 FIN 给 server。
• 此时 server 并不知道 client 已经奔溃了，所以它会发送第一条消息给 client。但 client 已经退出了，所以 client 的 TCP 协议栈会发送一个 RST 给 server。
• server 在接收到 RST 之后，继续写入第二条消息。往一个已经收到 RST 的 socket 继续写入数据，将导致SIGPIPE信号，从而杀死 server。

信号处理的机制

　　在 Linux 中，每个进程都拥有两个位向量，这两个位向量共同决定了进程将如何处理信号：

• 一个是pending位向量，它包含了那些内核发送给进程，但还没有被进程处理掉的信号。
• 另一个是blocked位向量，它包含了那些被进程屏蔽掉的信号。

　　当内核发送一个信号给进程时，它将会修改进程的pending位向量，譬如说，当内核发送一个SIGINT信号给进程，那么它会将进程的pending[SIGINT]的值设置成 1：

1
2
3
4
5
6
7
8
9
10
11
12
13

for (int i = 0; i < pending.size(); ++i) 
{
    if (pending[i] & (~blocked[i])) 
    {
        // 将这个信号从 pending 位向量中清除
        pending[i] = 0;
		
        // 让进程接收并处理这个信号
        // ...
        break;
    }
}

　　譬如说，下面的程序一开始就屏蔽了SIGINT信号，所以即使内核发送SIGINT信号给这个程序，这个信号也不会得到处理。而当程序解除了对SIGINT的屏蔽之后，这个SIGINT信号才会得到处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

#include <signal.h>
#include <unistd.h>
#include <string.h>
void sigint_handler(int sig)
{
    const char *message = "handle SIGINT signal\n";
    write(STDOUT_FILENO, message, strlen(message));
}
int main()
{
    signal(SIGINT, sigint_handler);
    sigset_t mask, prev_mask;
    sigemptyset(&mask);
    sigaddset(&mask, SIGINT);
    // 屏蔽掉 SIGINT 信号
    sigprocmask(SIG_BLOCK, &mask, &prev_mask);
    // 假设此时接收到 SIGINT 信号
    sleep(10);
    // 解除对 SIGINT 的屏蔽之后，进程会开始处理 SIGINT 信号
    sigprocmask(SIG_SETMASK, &prev_mask, NULL);
    return 0;
}

CPU 亲和性

　　Linux 可以运行在多处理器的机器上，为了维持多个CPU之间的负载均衡，线程可能会被OS调度到其它CPU上，这种情况下线程就无法利用原先CPU上边的缓存了，也就降低了CPU cache的命中率了。所谓的CPU亲和性，就是让线程在指定的CPU上长时间运行而不被调度到其它CPU上边，以提高CPU cache的命中率。
　　在Linux中，可以使用pthread_setaffinity_np()为线程设置CPU的亲和性：

1
2

int pthread_setaffinity_np(pthread_t thread, size_t cpusetsize,
                           const cpu_set_t *cpuset);

　　注意到，它的第一个参数类型是pthread_t，代表 Linux 线程的ID。在C++中，每个std::thread都对应着底层操作系统的一个线程，不过我们可以使用std::thread的native_handle()函数来返回它对应的 OS 线程的ID，在Linux系统中，这个函数的返回值类型是pthread_t。
　　下面的例子，我们让线程绑定到 CPU-0 上：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

#include <iostream>
#include <thread>
#include <chrono>
#include <sched.h>
int main()
{
    std::thread t([]() {
        while (true)
        {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            std::cout << "Thread #" << std::this_thread::get_id()
                      << ": on CPU" << sched_getcpu() << std::endl;
        }
    });
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    CPU_SET(0, &cpuset);
    pthread_setaffinity_np(t.native_handle(), sizeof(cpu_set_t), &cpuset);
    t.join();
    return 0;
}

改善 std::thread

　　在工程实践中，你会发现std::thread并不像想象中的那么完美。一个明显的问题是，当std::thread即将析构时，你要人为地确保实际的线程已经结束运行了或者已经分离了(可以分别调用它的join()和detach()成员函数)，否则程序将会抛出std::terminate异常。当然，由于std::thread的析构函数没有保证正确地释放资源，这也同时导致了std::thread不是异常安全的。
　　下面提供了一个ThreadRAII类，它在析构时会等待线程运行结束，或者直接分离线程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

class ThreadRAII
{
public:
    enum class DtorAction {join, detach};
    template<typename... Ts>
    ThreadRAII(DtorAction a, Ts&&... params)
        : action_(a), thread_(std::forward<Ts>(params)...)
    {
    }
    ThreadRAII(ThreadRAII &&) = default;
    ThreadRAII &operator=(ThreadRAII &&) = default;
    
    ~ThreadRAII()
    {
        if (thread_.joinable()) {
            if (action_ == DtorAction::join) {
                thread_.join();
            } else {
                thread_.detach();
            }
        }
    }
	
    std::thread &get() { return thread_; }
private:
    DtorAction action_;
    std::thread thread_;
};

　　创建ThreadRAII对象时，可以顺带指定它的析构行为，到底是等待线程运行结束呢还是直接分离线程。下面的例子中，程序退出时，ThreadRAII对象的析构函数会保证线程正常结束运行：

1
2
3
4
5
6
7

void sayHello() { cout << "Hello" << endl; }
int main()
{
    ThreadRAII t{ThreadRAII::DtorAction::join, sayHello};
    return 0;
}

管道

　　管道可以说是 Unix 系统最古老的 IPC 方式了。管道最常见的一个用法，就是将一个进程的输出导入到另一个进程的输入。譬如说，当用户执行下面的命令，ls命令的输出将会作为wc -l的输入：

1

$ ls | wc -l

　　在 Linux 系统中，管道有这些特征：

• 数据以字节流的形式在管道中传输，也就是说，数据是没有消息边界。
• 数据在管道中是单向流动的，管道的一端负责写数据，而另一端负责读数据。
• 管道的缓冲区长度是 64 KB。
• 调用write()写入数据时，如果指定写入的字节数不超过PIPE_BUF(默认值是 4KB)，那么 Linux 会保证这个写入操作是原子的。

　　下面的例子中，父进程向子进程发送多条消息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

#include <stdio.h>
#include <unistd.h>
#include <stdbool.h>
#include <errno.h>
#define MAXLINE 1024
void readAndPrint(int fd)
{
    char line[MAXLINE];
    while (true) {
        ssize_t n = read(fd, line, MAXLINE);
        if (n == 0) {
            break;
        }
        if (n == -1) {
            if (errno == EINTR) {
                continue;
            }
            perror("read");
        }
        write(STDOUT_FILENO, line, n);
    }
}
int main()
{
    // 创建管道, fd[0] 为读端，fd[1] 为写端
    int fd[2];	
    pipe(fd);
    pid_t pid = fork();
    if (pid > 0) {
        // 父进程不需要读数据，所以关闭读端
        close(fd[0]);
        write(fd[1], "first message\n", 14);
        write(fd[1], "second message\n", 15);
        // 父进程关闭写端，导致子进程 read() 返回 0
        close(fd[1]);
    } else if (pid == 0) {
        // 子进程不需要写数据，所以关闭写端
        close(fd[1]);
        readAndPrint(fd[0]);
        close(fd[0]);
    } else {
        perror("fork");
    }
    return 0;
}

select 和 poll 的缺陷

　　I/O多路复用允许进程同时监听多个文件描述符，这样进程就可以知道哪些文件描述符存在I/O就绪了。select()是最早出现的I/O多路复用方式，然而由于早期select()的API设计不当，导致了select()存在不少缺陷。

1
2

int select(int nfds, fd_set *readfds, fd_set *writefds, 
           fd_set *exceptfds, struct timeval *timeout);

　　从API层面来讲，select()使用fd_set来表示文件描述符的集合，而fd_set其实就是一个固定长度的位向量(bit vector)，在Linux上，这个固定长度是FD_SETSIZE，其数值是1024。这就带来了一个限制：凡是select()监听的文件描述符，它的大小必须小于1024。
　　poll()在某些程度上改进了select()存在的一些问题，例如poll()要求用户传递一个pollfd数组，但它不会限定这个数组的长度，所以理论上poll()可以监听任意数量的文件描述符（但实际上仍受限于进程最多能打开的文件描述符数量或系统的内存）。

1

int poll(struct pollfd *fds, nfds_t nfds, int timeout);

　　使用select()或poll()监听大量的文件描述符时，往往会遭遇到性能问题。当用户每次调用select()或poll()时，内核会对传入的所有文件描述符都检查一遍，并记录其中有哪些文件描述符存在I/O就绪，这个操作的耗时将随着文件描述符数量的增加而线性增长。
　　另一个重要因素也会影响select()和poll()的性能，例如用户每次调用poll()时，都需要传递一个pollfd数组，而poll()会将这个数组从用户空间拷贝到内核空间，当内核对这个数组作了修改之后，poll()又会将这个数组从内核空间拷贝到用户空间。随着pollfd数组长度的增加，每次拷贝的时间也会线性增长，一旦poll()需要监听大量的文件描述符，每次调用poll()时，这个拷贝操作将带来不小的开销。这个问题的根源在于select()和poll()的API设计不当，例如，对于应用程序来说，它每次调用poll()所监听的文件描述符集合应该是差不多的，所以我们不禁这样想，如果内核愿意提供一个数据结构，记录程序所要监听的文件描述符集合，这样每次用户调用poll()时，poll()就不需要将pollfd数组拷贝来拷贝去了（没错，epoll 就是这样解决的）。

epoll 的救赎

　　epoll 很好地解决了select()和poll()中存在的问题，从API层面来讲，epoll 使用 3 个调用来完成原本由select()或poll()所做的事，首先epoll_create()负责在内核中创建一个eventpoll类型的数据结构：

1
2
3
4
5
6
7
8
9

struct eventpoll 
{
    // 红黑树，用来记录进程所要监听的文件描述符（及其要监听的事件）
    struct rb_root rbr;
	
    // 双向链表，用来记录有哪些文件描述符已经就绪了
    struct list_head rdllist;
    // ...
};

　　epoll_ctl()负责增加、删除或修改红黑树上的节点，而epoll_wait()则负责返回双向链表中就绪的文件描述符(及其事件)。说到这里，我们不禁要问，epoll是怎么知道有哪些文件描述符已经就绪的呢？难道是遍历一次红黑树，逐个检查文件描述符？不可能，这效率太低了。
　　与select()或poll()不同，epoll 是事件驱动的，简单来说，当网卡收到一个 packet 的时候，会触发一个硬件中断，这导致内核调用相应的中断 handler，从网卡中读入数据放到协议栈，当数据量满足一定条件时，内核将回调ep_poll_callback()这个方法，它负责把这个就绪的文件描述符添加到双向链表中。这样当用户调用epoll_wait()时，epoll_wait()所做的就只是检查双向链表是否为空，如果不为空，就把文件描述符和数量返回给用户即可。