SPProcPool: Unix/Linux 上的进程池服务器框架

SPProcPool 是一个 linux/unix 平台上的进程池服务器框架，使用 c++ 实现。主要包含了几种不同类型的进程池的实现：

一个基于 Leader/Follower 模式的服务器端进程池（类似 apache 的 prefork 模型）；

一个组合了 Prefork 和线程池的服务器端框架（类似 apache 的 worker 模型）；

一个基于文件句柄传递的服务器端进程池；

一个用于非服务器端的，能够在多线程或事件驱动环境下使用的进程池。

主页：[url]http://code.google.com/p/spprocpool/[/url]

下载：[url=http://freshmeat.net/redir/spprocpool/72480/url_tgz/spprocpool-0.1.src.tar.gz]spprocpool[/url]

关于进程池服务器框架，在《unix网络编程（第二版）》的第 27 章，详细地描述了多种不同的实现方式。

基于 Leader/Follower 模式的实现：27.6 TCP 预先派生子进程服务器程序，accept 无上锁保护

基于文件句柄传递：27.9 TCP 预先派生子进程服务器程序，传递描述字

关于非服务器端的，能够在多线程或事件驱动环境下使用的进程池，这里做一个比较详细的说明。

多线程的好处是各个线程能够共享一个地址空间，因此对一些需要全局排序来调度的任务，使用多线程可以比较方便地实现。比如类似 postfix/qmgr 的模块，如果使用多线程的话，那么所有的邮件能够在一个优先队列中排队，按队列顺序进行投递；如果投递失败了，那么重新插入队列。

但另一方面，如果具体的任务处理部分已经有了实现，但不是线程安全的，这种问题要怎么来解决呢？

一个最直观的解决方法是每个任务直接 fork 一次。但是这种做法有几个细节问题需要认真考虑：

1.在子进程中如何把结果返回给调度进程？常见的方法是使用 pipe

2.每个任务 fork 一次，估计很多人的第一反应就是性能会怎么样？

3.如果调度进程中，除了负责 fork 的线程，还有其他的线程存在，那么就存在 fork-with-thread 的问题。

>>具体的内容可以参考：[url]http://www.opengroup.org/onlinepubs/000095399/functions/fork.html[/url]

针对上面存在的问题，在每个任务直接 fork 一次的基础上，做了一些改进，就形成了这样的一个进程池实现：

1.在前面的方案中，存在调度进程（Sched）和工作进程（Worker）这两类进程。

>>为了避免 fork-with-thread 的问题，再增加一类管理进程（Manager）。管理进程的主要职责就是负责 fork 工作进程。

2.通常 Manager 是由 Sched fork 出来的，它们之间存在一条通信的 pipe （MgrPipe） 。

>>创建一个新的工作进程的流程如下：Sched 创建一个 pipe （WorkerPipe），把其中的一端用 send_fd 的方法发送给 Manager，

>>然后 Manager fork 一个 Worker 出来，并且把 WorkerPipe 传递给 Worker 。这样就在 Sched 和 Worker 之间建立了一个 Pipe 。

3.Worker 在被 fork 出来之后，通常就阻塞在读 WorkerPipe 上面。Sched 通过 WorkerPipe 发送任务给 Worker 。

>>Worker 完成任务之后，通过 WorkerPipe 发送结果给 Sched 。Worker 可以不断地重复这个过程，这样就达到了一个池的效果。

4.对于使用 libevent 这类事件驱动类型的程序，这个进程池也能方便地被调用。

>>因为 Worker 曝露出来的是一个 PipeFd，能够方便地加入到 libevent 的事件循环中。这类事件驱动类的程序，

>>通常使用单线程实现，当具体的任务处理可能需要耗费比较长时间的时候，就需要使用多线程或者多进程来辅助了。

SPProcPool 提供了 3 个服务器框架：

SP_ProcInetServer（传递文件句柄）

SP_ProcLFServer（Leader/Follower模型）

SP_ProcMTServer（Worker模型）

它们的定义很类似：

[code]

class SP_ProcInetServer {

public:

SP_ProcInetServer( const char * bindIP, int port,

SP_ProcInetServiceFactory * factory );

virtual ~SP_ProcInetServer();

virtual int start();

};

class SP_ProcLFServer {

public:

SP_ProcLFServer( const char * bindIP, int port,

SP_ProcInetServiceFactory * factory );

virtual ~SP_ProcLFServer();

virtual int start();

};

class SP_ProcMTServer {

public:

SP_ProcMTServer( const char * bindIP, int port,

SP_ProcInetServiceFactory * factory );

virtual ~SP_ProcMTServer();

virtual int start();

};

[/code]

从接口中可以看到，要使用这些框架，需要提供一个 SP_ProcInetServiceFactory 类的实例。这个类相关的定义如下：

[code]

class SP_ProcInetService {

public:

virtual ~SP_ProcInetService();

virtual void handle( int socketFd ) = 0;

};

class SP_ProcInetServiceFactory {

public:

virtual ~SP_ProcInetServiceFactory();

virtual SP_ProcInetService * create() const = 0;

virtual void workerInit( const SP_ProcInfo * procInfo );

virtual void workerEnd( const SP_ProcInfo * procInfo );

};

[/code]

这里使用的是典型的抽象工厂方法。在工厂类中，除了 create 方法之外，还有两个特别的方法：workerInit 和 workerEnd 。workerInit 在子进程开始运行的时候被调用，workerEnd 在子进程退出的时候被调用。在 Service 类中，只有一个 handle 方法，它的参数就是已经 accept 到 socket 。

下面以一个简单的服务器为例进行说明。这个服务器是模仿《unix网络编程（第二版）》（中文版）第27章的服务器。服务器从 socket 读入包含一个数字的一行，然后根据这个数字返回相应的内容。

要实现这个简单的服务器例子，代码如下：

[code]

class SP_ProcUnpService : public SP_ProcInetService {

public:

SP_ProcUnpService() {}

virtual ~SP_ProcUnpService() {}

virtual void handle( int sockfd ) {

int ntowrite;

ssize_t nread;

char line[MAXLINE], result[MAXN];

for ( ; ; ) {

if ( (nread = read(sockfd, line, MAXLINE)) == 0) {

return;

}

ntowrite = atol(line);

if ((ntowrite <= 0) || (ntowrite > MAXN)) {

syslog( LOG_WARNING, "WARN: client request for %d bytes", ntowrite);

exit( -1 );

}

SP_ProcPduUtils::writen(sockfd, result, ntowrite);

}

}

};

class SP_ProcUnpServiceFactory : public SP_ProcInetServiceFactory {

public:

SP_ProcUnpServiceFactory() {}

virtual ~SP_ProcUnpServiceFactory() {}

virtual SP_ProcInetService * create() const {

return new SP_ProcUnpService();

}

virtual void workerInit( const SP_ProcInfo * procInfo ) {

signal( SIGINT, SIG_DFL );

printf( "pid %d start\n", (int)procInfo->getPid() );

}

virtual void workerEnd( const SP_ProcInfo * procInfo ) {

printf( "pid %d exit, pipeFd %d, requests %d, lastActiveTime %ld\n",

(int)procInfo->getPid(), procInfo->getPipeFd(),

procInfo->getRequests(), procInfo->getLastActiveTime() );

}

};

int main( int argc, char * argv[] )

{

SP_ProcMTServer server( "", 1770, new SP_ProcUnpServiceFactory() );

server.start();

return 0;

}

[/code]