进程池

发表于

在并发服务器调用fork派生一个子进程来处理每个客户端，这使得服务器能够同时为多个客户端服务，每个进程处理一个客户端。客户端数目的唯一限制是操作系统能拥有多少子进程的限制。

在为每个客户端的连接fork一个子进程是比较耗费CPU时间，一个系统能创建的进程个数也是有限的，如果创建的进程过多，就会发生CPU的“空转”，也就是CPU频繁的在进程间切换。对于像繁忙的Web服务器来说，这种做法显然是不可取的。

TCP预先派生子进程

在启动阶段预先派生一个数量的子进程，当各个客户连接到达时，这些子进程立即就能为它们服务。而不是为每个客户连接现场派生一个子进程。如图，展示了服务器父进程预先派生出N个子进程且正有2个客户连接着的情形。

process_pool.png

这种技术的优点在于无须引入父进程执行fork的开销就能处理新到的客户。缺点则是父进程必须在服务器启动阶段猜测需要预先派生多少个子进程。如果某个时刻客户端的数量恰好等于子进程总数，那么新到的客户端将被忽略，直到至少有一个子进程重新可用。这些客户端并未被完全忽略。内核将为每个新到的客户完成三路握手，直到达到相应套接字listen调用的blocklog数为止，然后在服务器调用accept时把这些已完成的连接传递给它。这么一来客户端就能知道服务器在响应时间的恶化，因为尽管它的connect调用立即返回，但是它的第一个请求可能是在一段时间之后才被服务器处理。

服务器可以通过父进程持续监视可用(即闲置)子进程数，来应对客户端负载的变动。一旦该值低于某个阈值就派生额外的子进程，同样，一旦该值超过一个阈值就终止一些过剩的子进程。

TCP进程池客户端测试程序

代码如下：

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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
#include <stdio.h>                                                        
#include <errno.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <string.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <sys/select.h>
#include <sys/time.h>
#include <fcntl.h>
#include <signal.h>
#include <time.h>

#include "error.h"

#define MAXLINE 4096

#define SA struct sockaddr

static int Socket(int family, int type, int protocol) {
    int n;

    if((n = socket(family, type, protocol)) < 0) {
        err_sys("socket error");
    }

    return n;
}

static void Inet_pton(int family, const char *strptr, void *addrptr) {
    if(inet_pton(family, strptr, addrptr) <= 0) {
        err_quit("inet_pton error for %s", strptr);
    }
}

static int Connect(int fd, struct sockaddr *sa, socklen_t salen) {
    if(connect(fd, sa, salen) < 0) {
        err_sys("connect error");
    }
}

static int tcp_connect(const char *ip, const char *port) {
    struct sockaddr_in servaddr;
    int sockfd;
    
    sockfd = Socket(AF_INET, SOCK_STREAM, 0);

    bzero(&servaddr, 0);
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(atoi(port));
    Inet_pton(AF_INET, ip, &servaddr.sin_addr);

    Connect(sockfd, (SA *) &servaddr, sizeof(servaddr));

    return sockfd;
}

static int writen(int fd, void *vptr, size_t nbytes) {
    const char *ptr = vptr;
    int left = nbytes;
    int written;

    while(left > 0) {
        if((written = write(fd, ptr, left)) < 0) {
            return -1;
        }

        ptr += written;
        left -= written;
    }

    return nbytes;
}

static void Writen(int fd, void *vptr, size_t nbytes) {
    if(writen(fd, vptr, nbytes) != nbytes) {
        err_sys("write error");
    }
}

static int readn(int fd, void *vptr, size_t nbytes) {
    char *ptr = vptr;
    int nread;
    int left = nbytes;

    while(left > 0) {
        nread = read(fd, vptr, left);

        if(nread < 0) {
            if(errno != EINTR) {
                return -1;
            }
        } else if(nread == 0) {
            break;
        } else {
            ptr += nread;
            left -= nread;
        }
    }
    return nbytes - left;
}

static int Readn(int fd, void *vptr, size_t nbytes) {
    int n;
    
    if((n = readn(fd, vptr, nbytes)) < 0) {
        err_sys("read error");
    }
    
    return n;
}

static void Close(int fd) {
    if(close(fd) == -1) {
        err_sys("close error");
    }
}

int main(int argc, char *argv[]) {
    int sockfd;
    int num_children;
    int nloops;
    int request_bytes;
    pid_t pid;
    int i;
    int j;
    int n;
    
    char request[MAXLINE];
    char reply[MAXLINE];

    if(argc != 6) {
        err_quit("%s <IPAddress> <port> <num_children> <num_loops> <request_bytes>", argv[0]);
    }

    num_children = atoi(argv[3]);
    nloops = atoi(argv[4]);
    request_bytes = atoi(argv[5]);

    snprintf(request, MAXLINE, "%d\n", request_bytes);

    for(i = 0; i < num_children; ++i) {
        pid = fork();

        if(pid == 0) {
            for(j = 0; j < nloops; ++j) {
                sockfd = tcp_connect(argv[1], argv[2]);
                Writen(sockfd, request, strlen(request));

                if((n = Readn(sockfd, reply, request_bytes)) != request_bytes) {
                    err_quit("server returned %d bytes", n);
                }
                Close(sockfd);
            }
            printf("child %d done\n", i);
            exit(0);
        } else if(pid < 0) {
            err_sys("fork error");
        }
    }

    while(wait(NULL) > 0) {
        ;
    }

    if(errno != ECHILD) {
        err_sys("wait error");
    }

    return 0;
}

每次运行客户程序时，需要指定服务器的主机名或IP地址、服务器的端口、由客户端fork的子进程数(以允许客户端并发地向同一个服务器发起多个连接)、每个子进程发送给服务器的请求数，以及每个请求要求服务器返回的数据字节数。

父进程调用fork派生指定个数的子进程，每个子进程再与服务器建立指定数目的连接。每次建立连接之后，子进程就在该连接上向服务器发送一行文本，指定需由服务器返回多少字节的数据，然后再该连接上读入这个数量的数据，最后关闭连接。父进程只是调用wait等待所有子进程都终止。需注意的是，这里关闭每个TCP连接的是客户端，因而TCP的TIME_WAIT状态在客户端而不是服务器。

用于执行本客户端程序的命令如下：

./client 192.168.1.156 9987 5 500 4096

这将建立2500个与服务器的TCP连接：5个子进程各自发起500次连接。在每个连接上，客户向服务器发送5字节数据(“4096\n”),服务器向客户返回4096字节数据。

TCP预先派生子进程服务器程序，accept无上锁保护

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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
#include <stdio.h>                                                        
#include <errno.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <string.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <sys/select.h>
#include <sys/time.h>
#include <fcntl.h>
#include <signal.h>
#include <time.h>
#include <sys/resource.h>
#include <sys/mman.h>

#include "error.h"

#define LISTENQ 1024
#define MAX_BUF 4096
#define SA struct sockaddr
#define MAX_CHILDREN 24
#define SERV_PORT 9987
#define min(a, b) ((a) < (b) ? (a) : (b))

static int num_children;
static pid_t pids[MAX_CHILDREN];
long *cptr;

static int Socket(int family, int type, int protocol) {
    int n;

    if((n = socket(family, type, protocol)) < 0) {
        err_sys("socket error");
    }

    return n;
}

static int Bind(int fd, struct sockaddr *sa, socklen_t salen) {
    if(bind(fd, sa, salen) < 0) {
        err_sys("bind error");
    }
}

static int Listen(int fd, int backlog) {
    const char *ptr;

    if((ptr = getenv("LISTENQ")) != NULL) {
        backlog = atoi(ptr);
    }

    if(listen(fd, backlog) < 0) {
        err_sys("listen error");
    }
}

static int tcp_listen(void) {
    int listenfd;
    struct sockaddr_in servaddr;
    
    listenfd = Socket(AF_INET, SOCK_STREAM, 0);    

    bzero(&servaddr, 0);
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(SERV_PORT);
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);

    Bind(listenfd, (SA *) &servaddr, sizeof(servaddr));

    Listen(listenfd, LISTENQ);

    return listenfd;
}

static int Accept(int fd, struct sockaddr *sa, socklen_t *salenptr) {
    int n;

    while((n = accept(fd, sa, salenptr)) < 0) {
        if(errno != EINTR && errno != ECONNABORTED) {
            err_sys("accept error");
        }
    }

    return n;
}

static int readline(int fd, void *vptr, size_t maxlen) {
    int i;
    char *ptr = vptr;
    int n;
    char c;

    for(i = 0; i < maxlen - 1; ++i) {
        if((n = read(fd, &c, 1)) == 1) {
            *ptr++ = c;
            if(c == '\n') {
                break;        /* newline */
            }
        } else if(n == 0) {
            if(i == 0) {
                return 0;     /* EOF, no data was read */
            } else {
                break;        /* EOF, some data was read */
            }
        } else {
            if(errno == EINTR) {
                continue;
            } else {
                return -1;     /* read error */
            }
        }
    }

    *ptr = '\0';

    return i + 1;
}

static int Readline(int fd, void *vptr, size_t maxlen) {
    int n;

    if((n = readline(fd, vptr, maxlen)) < 0) {
        err_sys("read error");
    }

    return n;
}

static int writen(int fd, void *vptr, size_t nbytes) {
    const char *ptr = vptr;
    size_t left = nbytes;
    int nwritten;

    while(left > 0) {
        if((nwritten = write(fd, vptr, left)) < 0) {
            return -1;
        }

        ptr += nwritten;
        left -= nwritten;
    }

    return nbytes;
}

static void Writen(int fd, void *vptr, size_t nbytes) {
    if(writen(fd, vptr, nbytes) != nbytes) {
        err_sys("write error");
    }
}

static void Close(fd) {
    if(close(fd) == -1) {
        err_sys("close error");
    }
}

static void process_request(int fd) {
    char recvline[MAX_BUF];
    char result[MAX_BUF];
    int nread;
    int towrite;

    for(;;) {
        if((nread = Readline(fd, recvline, MAX_BUF)) == 0) {
            return;
        }

        towrite = atol(recvline);

        if(towrite <= 0 || towrite > MAX_BUF) {
            err_quit("client request for %d bytes", towrite);
        }
        Writen(fd, result, towrite);
    }

}

static void child_main(int i, int listenfd) {
    int connfd;

    printf("child %d starting\n", i);

    for(;;) {
        connfd = Accept(listenfd, NULL, NULL);
        ++cptr[i];
        process_request(connfd);
        Close(connfd);
    }
}

static int child_make(int i, int listenfd) {
    pid_t pid;

    pid = fork();

    if(pid == 0) {
        child_main(i, listenfd);
    } else if(pid > 0) {
        return pid;
    } else {
        err_sys("fork error");
    }
}

static void sig_chld(int signo) {
    pid_t pid;
    int stat;
        
    while((pid = waitpid(-1, &stat, WNOHANG)) > 0) {
        ;
    }
    return ;
}

static void sig_int(int signo) {
    int i;

    /* terminate all chlidren */
    for(i = 0; i < num_children; ++i) {
        kill(pids[i], SIGTERM);
    }

    while(wait(NULL) > 0) { /* wait for all children */
        ;
    }
    
    if(errno != ECHILD) {
        err_sys("wait error");
    }

    printf("\n");
    
    for(i = 0; i < num_children; ++i) {
        printf("child %d, %ld connections\n", i, cptr[i]);
    }
    
    exit(0);
}

static long* meter(int num_children) {
    int fd;
    void *ptr;

    if((fd = open("/dev/zero", O_RDWR, 0)) < 0) {
        err_sys("open /dev/zero error");
    }

    if((ptr = mmap(0, num_children * sizeof(long), PROT_READ|PROT_WRITE, MAP_SHARED, fd, 0)) == (void*)-1) {
        err_sys("mmap error");
    }
    Close(fd);

    return (long *)ptr;
}

int main(int argc, char *argv[]) {
    int listenfd;
    int i;

    if(argc != 2) {
        err_quit("usage: %s <num_children>", argv[0]);
    }

    num_children = min(atoi(argv[1]), MAX_CHILDREN);
    listenfd = tcp_listen();

    cptr = meter(num_children);

    for(i = 0; i < num_children; ++i) {
        pids[i] = child_make(i, listenfd);
    }

    signal(SIGINT, sig_int);

    for(;;) {
        pause();
    }

    return 0;
}

main函数

调用tcp_listen函数创建一个监听套接字。调用child_make预先创建子进程，创建的个数由命令行指定，最大的个数不能超过MAX_CHILDREN。SIGINT信号处理函数，在终止服务器(CTRL+C)时被调用，用于统计各个子进程处理的连接数。

child_make函数

调用fork派生子进程，父进程返回子进程的pid。子进程调用child_main函数。

child_main函数

每个子进程调用accept返回一个已连接套接字，然后调用process_request函数处理客户请求，最后关闭连接。子进程一直在这个循环中反复，直到被父进程终止。

sig_int函数

在服务器终止(CTRL+C)时被调用，调用kill给每个子进程发送STGTERM信号终止它们，并通过调用wait等待子进程结束。最后打印每个子进程处理的连接数。

meter函数

在共享内存区中分配一个长整数计数器数组，用于查看客户端在阻塞于accept调用中的可用子进程池上的分布，每个子进程一个计数器，在accept返回后，对应子进程的计数器加1。SIGINT信号处理函数在所有子进程终止之后显示每个进程的计数器的值。

在分配共享内存区时，使用/dev/zero映射。数组是本进程在尚未派生各个子进程之前调用mmap创建的，它将由本进程(父进程)和后来fork的所有子进程共享。

运行服务器程序：

服务器程序创建了10个子进程，输出如下信息：

[heql@ubuntu socket]$ ./server 10
child 2 starting
child 1 starting
child 3 starting
child 0 starting
child 5 starting
child 7 starting
child 6 starting
child 4 starting
child 8 starting
child 9 starting

运行客户端程序：

客户端使用5个子进程各自发起5000次连接。在每个连接上，客户向服务器发送请求4096字节数据，服务器将向每个连接的客户返回4096字节数据。

[heql@ubuntu socket]$ ./client 192.168.1.156 9987 5 5000 4096
child 0 done
child 3 done
child 2 done
child 1 done
child 4 done

终止服务器

客户端程序运行完后，按下CTRL+C终止服务器，查看每个子进程处理的连接个数：

^C
child 0, 2466 connections
child 1, 2536 connections
child 2, 2500 connections
child 3, 2481 connections
child 4, 2516 connections
child 5, 2475 connections
child 6, 2550 connections
child 7, 2446 connections
child 8, 2524 connections
child 9, 2506 connections

多个进程在同一个监听描述符上调用accept：父进程在派生任何子进程之前创建监听套接字，而每次调用fork时，所有描述符也被复制。如图展示了proc结构(每个进程一个)、监听描述符的单个file结构以及单个socket结构之间的关系。

proc.png

描述符只是本进程应用file结构的proc结构中一个数组中某个元素的下标而已。fork调用执行期间为子进程复制描述符的特性之一是：子进程中一个给定描述符引用的file结构正是父进程中同一个描述符引用的file结构。每个file结构都有一个引用计数。当打开一个文件或套接字时，内核将为之构造一个file结构，并由此作为打开操作返回值的描述符引用，它的引用计数值为1，以后每当调用fork以派生子进程或对打开的操作返回的描述符(或其复制品)调用dup以复制描述符时，该file结构的引用计数就递增(每次递增1)。在我们的N个子进程的例子中，file结构的引用计数为N+1(别忘了父进程仍然保持该监听描述符打开着，不过它从不调用accept)。

服务器进程在程序启动阶段派生N个子进程，它们各自调用accept并因而均被内核投入睡眠。当第一个客户连接到达时，所有N个子进程均被唤醒。这是因为所有N个子进程所用的监听描述符执行同一个socket结构，致使它们在同一个等待通道上进行睡眠。尽管所有N个子进程均被唤醒，其中只有最先运行的子进程获得那个客户连接，其余N-1个子进程继续恢复睡眠。

这就是有时候称为 惊群 的问题，因为尽管只有一个子进程将获得连接，所有N个子进程却都被唤醒。如果被唤醒的子进程过多，这也会导致性能受损。

TCP预先派生子进程服务器程序，accept文件上锁保护

上面的实现允许多个进程在引用同一个监听套接字的描述符上调用accept，然而这种做法对于有些内核来说，是不允许这么做的，在客户开始连接到该服务器后不久，某个子进程的accept就会返回EPROTO错误(表示协议出错)。

解决办法是让应用进程在调用accept前后安置某种形式的锁(lock)，这样任意时刻只有一个子进程阻塞在accept调用中，其他子进程则阻塞在试图获取用于保护accept的锁上。

lock_init函数

代码如下：

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
static struct flock lock_it;
static struct flock unlock_it;
static int lock_fd = -1;

static void lock_init(const char *pathname) {
    char lock_file[1024];

    strncpy(lock_file, pathname, sizeof(lock_file));
    if((lock_fd = mkstemp(lock_file)) < 0) {
        err_sys("mkstemp error");
    }

    if(unlink(lock_file) < 0) {
        err_sys("unlink error");
    }

    lock_it.l_type = F_WRLCK;
    lock_it.l_whence = SEEK_SET;
    lock_it.l_start = 0;
    lock_it.l_len = 0;

    unlock_it.l_type = F_UNLCK;
    unlock_it.l_whence = SEEK_SET;
    unlock_it.l_start = 0;
    unlock_it.l_len = 0;    
}

lock_init函数由调用者指定一个路径名模板，mktemp函数根据该模板创建一个唯一的路径名，本函数随后创建一个具备该路径名的文件并立即unlink掉。通过从文件系统目录中删除该路径名，以后即使程序崩溃，这个临时文件也完全消失。只要有一个或多个进程打开着这个文件(也就是说它的引用计数大于0)，该文件本身就不会被删除。

初始化两个flock结构，一个用于上述文件，一个用于解锁文件。文件上述范围起自字节偏移量0(l_whence值为SEEK_SET， l_start值为0)，跨越整个文件(l_len值为0，表示锁住整个文件或到文件尾)。

在main函数中在派生子进程的循环之前增加lock_init函数的调用。

1
2
3
4
5
6
+   lock_init("/tmp/lock.XXXXXX");

    for(i = 0; i < nchildren; ++i) {
        cptr[i] = 0;
        pids[i] = child_make(i, listenfd);
    }

lock_wait、lock_release函数

代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
static void lock_wait(void) {
    while(fcntl(lock_fd, F_SETLKW, &lock_it) < 0) {
        if(errno == EINTR) {
            continue;
        } else {
            err_sys("fcntl error for lock_wait");
        }
    }
}

static void lock_release(void) {
    if(fcntl(lock_fd, F_SETLKW, &unlock_it) < 0) {
        err_sys("fcntl error for lock_release");
    }
}

lock_wait、lock_release函数用于加锁和解锁。在child_main函数中在调用accept之前获取文件锁，在accept返回之后释放文件锁。

1
2
3
4
5
6
7
8
9
for(;;) {
+   lock_wait();
    connfd = Accept(listenfd, NULL, NULL);
+   lock_release();
    
    ++cptr[i];
    process_request(connfd);
    Close(connfd);
}

TCP预先派生子进程服务器程序，accept使用线程上锁保护

上面使用的是文件上述方法，不过它涉及文件系统操作，可能比较耗时。可以改用线程上锁保护accept，因为这种方法不仅适用于同一进程内各线程之间的上锁，而且适用了不同进程之间的上锁。

在不同进程之间使用线程上锁要求：

  1. 互斥锁变量必须存放在由所有共享的内存区中
  2. 必须告知线程函数库这是在不同进程之间共享的互斥锁

要求线程库支持PTHREAD_PROCESS_SHARED属性

lock_init函数

代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
static pthread_mutex_t *mutex_ptr;

static void lock_init(void) {
    int fd;
    pthread_mutexattr_t mattr;
    void *ptr;

    if((fd = open("/dev/zero", O_RDWR, 0)) < 0) {
        err_sys("open /dev/zero error");
    }

    if((ptr = mmap(0, sizeof(pthread_mutex_t), PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0)) == (void *)-1) {
        err_sys("mmap error");
    }
    mutex_ptr = (pthread_mutex_t *)ptr;

    Close(fd);

    CHECK(pthread_mutexattr_init(&mattr));
    CHECK(pthread_mutexattr_setpshared(&mattr, PTHREAD_PROCESS_SHARED));
    CHECK(pthread_mutex_init(mutex_ptr, &mattr));
}

打开/dev/zero然后调用mmap。所映射的字节数是一个pthread_mutex_t类型变量的大小。随后关闭描述符，因为描述符已被内存映射了。

对于一个存放在共享内存区中的互斥锁，必须调用一些pthread库函数以告知该函数库：这是一个位于共享内存区中互斥锁，将用于不同进程之间的上锁。首先为一个互斥锁以默认属性初始化一个pthread_mutexattr_t结构，然后赋予该结构PTHREAD_PROCESS_SHARED属性(该属性的默认值为PTHREAD_PROCESS_PRIVATE，即只允许在单个进程内使用)。最后调用pthread_mutex_init函数以这些属性初始化共享内存区中的互斥锁。

lock_wait、lock_release函数

代码如下：

1
2
3
4
5
6
7
static void lock_wait(void) {
    CHECK(pthread_mutex_lock(mutex_ptr));
}

static void lock_release(void) {
    CHECK(pthread_mutex_unlock(mutex_ptr));
}