（转）Linux进程池的实现

原文地址： http://m.blog.csdn.net/blog/yhc13429826359/39258067
通过动态创建子进程（或者子线程）来实现并发服务器的。这样做有如下缺点：

1、 动态创建进程（或线程）是比较耗费时间的，这将导致较慢的客户响应。

2、动态创建的子进程（或子线程）通常只用来为一个客户服务（除非我们做特殊处理），这将导致系统上产生大量的细微进程（或者线程）。进程（或者线程）间的切换消费大量CPU时间。

3、动态创建的子进程是当前进程的完整映像。当前进程必须谨慎地管理其分配的文件描述符和堆内存等系统资源，从而使系统的可用资源急剧下降，进而影响服务器的性能。

1、进程池和线程池的概述

进程池和线程池相似，所以这里我们只以进程池为例子进行介绍。下面对进程池的讨论完全适用于线程池。

进程池是由服务器预先创建的一组子进程，这些子进程的数目在3-10个之间。httpd守护进程就是使用了包含7个子进程的进程池来实现并发的。线程池中的线程数量应该和CPU数量差不多。

进程池中的所有子进程都运行着相同的代码，并具有相同的属性，比如优先级，PGID等等。因为进程池在服务器启动之初就创建好了，所以每个子进程都相对“干净”，即他们没有打开不必要的文件描述符（从父进程继承而来），也不会错误地使用大块的堆内存（从父进程复制得到）。

当有新的任务到来时，主进程将通过某种方式选择进程池中的某一个子进程来为之服务。相对于动态创建子进程，选择一个已经存在的子进程的代价明显要小的多。至于主进程选择哪个进程来为新任务服务，则有两种方式：

1、主进程使用某种算法主动选择子进程。最简单、最常用的算法是随机算法和Round-Robin(轮流选取)算法，但更优秀、更智能的算法将使任务在各个工作进程中更均匀地分配、从而减轻服务器的整体压力。

2、主进程和所有子进程通过一个共享的工作队列来同步，子进程都睡眠在该工作队列上。当有心的任务到来时，主进程将任务添加到工作队列中。这将唤醒正在等待任务的子进程，不过只有一个子进程获得新任务的“接管权”，它可以从工作队列中取出并执行之，而其他子进程将继续睡眠在工作队列上。

当选择好子进程后，主进程还需要使用某种通知机制来告诉目标子进程有新进程需要处理，并传递必要的数据。最简单的方法是，在父进程和子进程之间预先建立好一条管道，然后通过该管道来实现所有进程间通信（当然，要预先定义好一套协议来规范管道的使用）。在父线程和子线程之间传递数据就要简单的多，因为我们可以把这些数据定义全局的，那么他们本身就是被所有线程共享的。

综合上面的论述，我们将进程池的一般模型描述为图15-1所示的形式。



2、处理多客户

在使用进程池处理多客户任务时，首先考虑的一个问题是：监听socket和连接socket是否都由主进程来统一管理。并发模型，其中半同步/半反应堆模式是由主进程统一管理这两种socket的。而高效的半同步/半异步和领导者/追随者模式，则是由主进程管理所有监听socket，而各个子进程分别管理属于自己的连接socket的。对于前一种情况，主进程接受新的连接以得到连接socket，然后它需要将该socket传递给子进程（对于线程池而言，父线程将socket传递给子线程是很简单的。因为他们可以很容易地共享该socket。但对于进程池而言，必须通过管道传输）。后一种情况的灵活性更大一些，因为子进程可以自己调用accept来接受新的连接，这样该父进程就无须向子进程传递socket。而只需要简单地通知一声：“我检测到新的连接，你来接受它。

常连接，即一个客户的多次请求可以复用一个TCP连接。那么，在设计进程池时还需要考虑：一个客户连接上的所有任务是否始终由一个子进程来处理。如果说客户任务是无状态的，那么我们可以考虑使用不同的进程为该客户不同请求服务。




但如果客户任务是存在上下文关系的，则最好一直用同一个进程来为之服务，否则实现起来比较麻烦，因为我们不得不在各个子进程传递上下文数据，我们采用epoll的EPOLLONESHOT事件，这一事件能够确保一个客户连接在整个生命周期中仅被一个线程处理。

3、半同步/半异步进程池实现

综合前面的讨论，我们可以实现这个进程池，为了避免在父、子进程之间传递文件描述符，我们将接受新连接的操作放到子进程中，很显然，对于这种模式而言，一个客户连接上的所有任务始终是由一个子进程来处理的。



1、包含的头文件：

C++ Code

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

#include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> #include <assert.h> #include <stdio.h> #include <unistd.h> #include <errno.h> #include <string.h> #include <fcntl.h> #include <stdlib.h> #include <sys/epoll.h> #include <signal.h> #include <sys/wait.h> #include <sys/stat.h>

2、process结构体实现：



C++ Code

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

//描述一个子进程的类， //m_pid是目标子进程的PID，m_pipefd是父进程和子进程通信用的管道 class process { public: process() : m_pid( -1 ){} public: pid_t m_pid; int m_pipefd[2]; };

3、进程池类的实现



C++ Code

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

//将它定义为模板类是为了代码复用 //其模板参数是处理逻辑任务的类 template< typename T > class processpool { private: //将构造函数定义为私有，因此我们只能通过后面的create静态函数来创建 //processpool实例 processpool( int listenfd, int process_number = 8 ); public: //单例模式，以保证程序最多创建一个processpool实例，这是程序正确处理信号的必要条件 static processpool< T > *create( int listenfd, int process_number = 8 ) { if( !m_instance ) { m_instance = new processpool< T >( listenfd, process_number ); } return m_instance; } ~processpool() { delete [] m_sub_process; } //启动进程池 void run(); private: void setup_sig_pipe(); void run_parent(); void run_child(); private: //进程允许的最大子进程数量 static const int MAX_PROCESS_NUMBER = 16; //每个子进程最多能处理的客户数量 static const int USER_PER_PROCESS = 65536; //epoll最多能处理的事件数 static const int MAX_EVENT_NUMBER = 10000; //进程池中的进程总数 int m_process_number; //子进程在池中的序号，从0开始 int m_idx; //每个进程都有一个epoll内核事件表，用m_epoolfd标识 int m_epollfd; //监听socket int m_listenfd; //子进程通过m_stop来决定是否停止运行 int m_stop; //保存所有子进程的描述信息 process *m_sub_process; //进程池静态实例 static processpool< T > *m_instance; };

4、细节实现如下：

C++ Code

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 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360

template< typename T > processpool< T > *processpool< T >::m_instance = NULL; //用于处理信号的管道，以实现统一事件源，后面称之为信号管道 static int sig_pipefd[2]; static int setnonblocking( int fd ) { int old_option = fcntl( fd, F_GETFL ); int new_option = old_option | O_NONBLOCK; fcntl( fd, F_SETFL, new_option ); return old_option; } static void addfd( int epollfd, int fd ) { epoll_event event; event.data.fd = fd; event.events = EPOLLIN | EPOLLET; epoll_ctl( epollfd, EPOLL_CTL_ADD, fd, &event ); setnonblocking( fd ); } //从epollfd标识的epoll内核事件表中删除fd上的所有注册事件 static void removefd( int epollfd, int fd ) { epoll_ctl( epollfd, EPOLL_CTL_DEL, fd, 0 ); close( fd ); } static void sig_handler( int sig ) { int save_errno = errno; int msg = sig; send( sig_pipefd[1], ( char * )&msg, 1, 0 ); errno = save_errno; } static void addsig( int sig, void( handler )(int), bool restart = true ) { struct sigaction sa; memset( &sa, '\0', sizeof( sa ) ); sa.sa_handler = handler; if( restart ) { sa.sa_flags |= SA_RESTART; } sigfillset( &sa.sa_mask ); assert( sigaction( sig, &sa, NULL ) != -1 ); } //进程池构造函数。 //参数listenfd是监听socket，它必须在创建进程池之前被创建，否则 //子进程无法直接引用它，参数process_number指定进程池中子进程的数量。 template< typename T > processpool< T >::processpool( int listenfd, int process_number ) : m_listenfd( listenfd ), m_process_number( process_number ), m_idx( -1 ), m_stop( false ) { assert( ( process_number > 0 ) && ( process_number <= MAX_PROCESS_NUMBER ) ); m_sub_process = new process[ process_number ]; assert( m_sub_process ); //创建process_number个子进程，并建立他们和父进程之间的管道 for( int i = 0; i < process_number; ++i ) { int ret = socketpair( PF_UNIX, SOCK_STREAM, 0, m_sub_process[i].m_pipefd ); assert( ret == 0 ); m_sub_process[i].m_pid = fork(); assert( m_sub_process[i].m_pid >= 0 ); if( m_sub_process[i].m_pid > 0 ) { close( m_sub_process[i].m_pipefd[1] ); continue; } else { close( m_sub_process[i].m_pipefd[0] ); m_idx = i; break; } } } //统一事件源 template< typename T > void processpool< T >::setup_sig_pipe() { //创建epoll事件监听表和信号管道 m_epollfd = epoll_create( 5 ); assert( m_epollfd != -1 ); int ret = socketpair( PF_UNIX, SOCK_STREAM, 0, sig_pipefd ); assert( ret != -1 ); setnonblocking( sig_pipefd[1] ); addfd( m_epollfd, sig_pipefd[0] ); //设置信号处理函数 addsig( SIGCHLD, sig_handler ); addsig( SIGTERM, sig_handler ); addsig( SIGINT, sig_handler ); addsig( SIGPIPE, SIG_IGN ); } //父进程中m_idx值为-1，子进程中m_idx值大于等于0，我们据此判断下来 //要运行的是父进程代码还是子进程代码 template< typename T > void processpool< T >::run() { if( m_idx != -1 ) { run_child(); return; } run_parent(); } template< typename T > void processpool< T >::run_child() { setup_sig_pipe(); //每个子进程都通过其在进程池中的序号值m_idx找到与父进程通信的管道 int pipefd = m_sub_process[m_idx].m_pipefd[ 1 ]; //子进程需要监听管道文件描述pipefd，因为父进程将通过它来通知子进程 //accept新连接 addfd( m_epollfd, pipefd ); epoll_event events[ MAX_EVENT_NUMBER ]; T *users = new T [ USER_PER_PROCESS ]; assert( users ); int number = 0; int ret = -1; while( ! m_stop ) { number = epoll_wait( m_epollfd, events, MAX_EVENT_NUMBER, -1 ); if ( ( number < 0 ) && ( errno != EINTR ) ) { printf( "epoll failure\n" ); break; } for ( int i = 0; i < number; i++ ) { int sockfd = events[i].data.fd; if( ( sockfd == pipefd ) && ( events[i].events & EPOLLIN ) ) { int client = 0; //从父/子进程之间的管道读取数据，并将结果保存在变量client中。 //如果读取成功，则表示有新的客户连接到来。 ret = recv( sockfd, ( char * )&client, sizeof( client ), 0 ); if( ( ( ret < 0 ) && ( errno != EAGAIN ) ) || ret == 0 ) { continue; } else { struct sockaddr_in client_address; socklen_t client_addrlength = sizeof( client_address ); int connfd = accept( m_listenfd, ( struct sockaddr * )&client_address, &client_addrlength ); if ( connfd < 0 ) { printf( "errno is: %d\n", errno ); continue; } addfd( m_epollfd, connfd ); //模板T必须实现init方法，以初始化一个客户连接 //我们直接使用connfd来索引逻辑处理对象 //T类型的对象，以提高程序效率 users[connfd].init( m_epollfd, connfd, client_address ); } } //下面处理子进程接收到的信号 else if( ( sockfd == sig_pipefd[0] ) && ( events[i].events & EPOLLIN ) ) { int sig; char signals[1024]; ret = recv( sig_pipefd[0], signals, sizeof( signals ), 0 ); if( ret <= 0 ) { continue; } else { for( int i = 0; i < ret; ++i ) { switch( signals[i] ) { case SIGCHLD: { pid_t pid; int stat; while ( ( pid = waitpid( -1, &stat, WNOHANG ) ) > 0 ) { continue; } break; } case SIGTERM: case SIGINT: { m_stop = true; break; } default: { break; } } } } } //如果是其他可读数据，那么必然是客户请求到来。 //调用逻辑对象的process方法处理之 else if( events[i].events & EPOLLIN ) { users[sockfd].process(); } else { continue; } } } delete [] users; users = NULL; close( pipefd ); //close( m_listenfd ); //我们将这句话注销掉，以提醒读者，应该有m_listenfd的创建者 //来关闭这个文件描述符，即所谓的“对象（比如一个文件描述符，又或者一 //堆内存）由那个函数创建，就应该由那个函数销毁 close( m_epollfd ); } template< typename T > void processpool< T >::run_parent() { setup_sig_pipe(); //父进程监听m_listenfd addfd( m_epollfd, m_listenfd ); epoll_event events[ MAX_EVENT_NUMBER ]; int sub_process_counter = 0; int new_conn = 1; int number = 0; int ret = -1; while( ! m_stop ) { number = epoll_wait( m_epollfd, events, MAX_EVENT_NUMBER, -1 ); if ( ( number < 0 ) && ( errno != EINTR ) ) { printf( "epoll failure\n" ); break; } for ( int i = 0; i < number; i++ ) { int sockfd = events[i].data.fd; if( sockfd == m_listenfd ) { //如果有新连接到来，就采用RR方式将其分配给一个子进程处理 int i = sub_process_counter; do { if( m_sub_process[i].m_pid != -1 ) { break; } i = (i + 1) % m_process_number; } while( i != sub_process_counter ); if( m_sub_process[i].m_pid == -1 ) { m_stop = true; break; } sub_process_counter = (i + 1) % m_process_number; //send( m_sub_process[sub_process_counter++].m_pipefd[0], ( char* )&new_conn, sizeof( new_conn ), 0 ); send( m_sub_process[i].m_pipefd[0], ( char * )&new_conn, sizeof( new_conn ), 0 ); printf( "send request to child %d\n", i ); //sub_process_counter %= m_process_number; } //下面处理父进程接收到的信号 else if( ( sockfd == sig_pipefd[0] ) && ( events[i].events & EPOLLIN ) ) { int sig; char signals[1024]; ret = recv( sig_pipefd[0], signals, sizeof( signals ), 0 ); if( ret <= 0 ) { continue; } else { for( int i = 0; i < ret; ++i ) { //如果进程池中第i个子进程退出了， //则主进程关闭通信管道，并设置相应的m_pid为-1，以标记该子进程已退出 switch( signals[i] ) { case SIGCHLD: { pid_t pid; int stat; while ( ( pid = waitpid( -1, &stat, WNOHANG ) ) > 0 ) { for( int i = 0; i < m_process_number; ++i ) { if( m_sub_process[i].m_pid == pid ) { printf( "child %d join\n", i ); close( m_sub_process[i].m_pipefd[0] ); m_sub_process[i].m_pid = -1; } } } //如果所有子进程都已经退出了，则父进程也退出 m_stop = true; for( int i = 0; i < m_process_number; ++i ) { if( m_sub_process[i].m_pid != -1 ) { m_stop = false; } } break; } case SIGTERM: case SIGINT: { //如果父进程接收到终止信号，那么就杀死所有子进程，并等待它们全部结束，当然， //通知子进程结束更好的方法是向父/子进程之间的通信管道发送特殊数据 printf( "kill all the clild now\n" ); for( int i = 0; i < m_process_number; ++i ) { int pid = m_sub_process[i].m_pid; if( pid != -1 ) { kill( pid, SIGTERM ); } } break; } default: { break; } } } } } else { continue; } } } //由创建者关闭这个文件描述符 //close( m_listenfd ); close( m_epollfd ); }

4、用进程池实现的简单CGI服务器

利用进程池来重新实现一个并发的CGI服务器，代码如下所示：

C++ Code

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

#include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> #include <assert.h> #include <stdio.h> #include <unistd.h> #include <errno.h> #include <string.h> #include <fcntl.h> #include <stdlib.h> #include <sys/epoll.h> #include <signal.h> #include <sys/wait.h> #include <sys/stat.h> #include "processpool.h" /*引用上一节介绍的进程池*/ /*用户处理客户CGI请求的类，它可以作为processpoll类的模板类*/ class cgi_conn { public: cgi_conn(){} ~cgi_conn(){} /*初始化客户连接，清空读缓冲区*/ void init( int epollfd, int sockfd, const sockaddr_in& client_addr ) { m_epollfd = epollfd; m_sockfd = sockfd; m_address = client_addr; memset(m_buf, '\0', BUFFER_SIZE); m_read_idx = 0; } void process() { int idx = 0; int ret = -1; /*循环读取和分析客户数据*/ while( true ) { idx = m_read_idx; ret = recv( m_sockfd, m_buf+idx, BUFFER_SIZE-1-idx, 0); /*如果读操作发生错误，则关闭客户连接，但如果是暂时无数据可读，则退出循环*/ if( ret < 0 ) { if( errno != EAGAIN ) { removefd( m_epollfd, m_sockfd ); } break; } else if( ret == 0 ) { removefd( m_epollfd, m_sockfd ); break; } else { m_read_idx += ret; printf("user content is:%s\n", m_buf); /*如果遇到字符"\r\n"，则开始处理客户请求*/ for(; idx<m_read_idx; ++idx) { if( (idx>=1) && (m_buf[idx-1] == '\r') && (m_buf[idx] == '\n') ) { break; } } /*如果没有遇到字符“\r\n”，则需要读取更多客户数据*/ if( idx == m_read_idx) { continue; } m_buf[idx-1] = '\0'; char* file_name =m_buf; /*判断客户要运行的CGI程序是否存在*/ if( access(file_name, F_OK) == -1 ) { removefd( m_epollfd, m_sockfd ); break; } /*创建子进程来执行CGI程序*/ ret = fork(); if( ret == -1) { removefd( m_epollfd, m_sockfd); break; } else if( ret > 0 ) { /*父进程只需关闭连接*/ removefd( m_epollfd, m_sockfd); break; } else { /*子进程将标准输出定向到m_sockfd,并执行CGI程序*/ close( STDOUT_FILENO ) ; dup( m_sockfd ); execl( m_buf, m_buf, 0 ); exit(0); } } } } private: /*读缓冲区的大小*/ static const int BUFFER_SIZE = 1024; static int m_epollfd; int m_sockfd; sockaddr_in m_address; char m_buf[ BUFFER_SIZE ]; /*标记读缓冲区中已经读入的客户数据最后一个字节的下一个位置*/ int m_read_idx; }; int cgi_conn::m_epollfd = -1; /*主函数*/ int main( int argc, char* argv[] ) { if( argc <= 2) { printf( "usage: %s ip_address port_number\n", basename(argv[0]) ); return 1; } const char* ip = argv[1]; int port = atoi( argv[2] ); int listenfd = socket(PF_INET, SOCK_STREAM, 0); assert( listenfd >= 0 ); int ret = 0; struct sockaddr_in address; bzero( &address, sizeof(address) ); address.sin_family = AF_INET; inet_pton(AF_INET, ip, &address.sin_addr); address.sin_port = htons( port ); ret = bind(listenfd, (struct sockaddr*)&address, sizeof( address )); assert( ret != -1); ret = listen( listenfd, 5 ); assert(ret != -1); processpool<cgi_conn>* pool = processpool<cgi_conn>::create( listenfd ); if( pool ) { pool->run(); delete pool; } close(listenfd); /*正如前文提到，main函数创建了文件描述符listenfd，那么就由它亲自关闭*/ return 0; }