SPServer : 一个基于线程池（包括HAHS和LF）的高并发 server 框架

spserver 是一个实现了半同步/半异步(Half-Sync/Half-Async)和领导者/追随者(Leader/Follower) 模式的服务器框架，能够简化 TCP server 的开发工作。
spserver 使用 c++ 实现，目前实现了以下功能：
1.封装了 TCP server 中接受连接的功能；
2.使用非阻塞型I/O和事件驱动模型，由主线程负责处理所有 TCP 连接上的数据读取和发送，因此连接数不受线程数的限制；
3.主线程读取到的数据放入队列，由一个线程池处理实际的业务。
4.一个 http 服务器框架，即嵌入式 web 服务器（请参考： SPWebServer：一个基于 SPServer 的 web 服务器框架）

0.6 版本之前只包含 Half-Sync/Half-Async 模式的实现，0.6 版本开始包含 Leader/Follower 模式的实现
0.7 版本开始支持 ssl 。把 socket 相关的操作抽象为一个 IOChannel 层，关于 openssl 的部分单独实现为一个 plugin 的形式，对于不使用 ssl 功能的用户，不需要引入 ssl 相关的头文件和库。
0.7.5 增加了一个 sptunnel 程序，是一个通用的 ssl proxy 。类似 stunnel 。
0.9.0 移植 spserver 到 windows 平台，需要在 windows 下编译 libevent 和 pthread 。
0.9.1 在 windows 平台，去掉了对 libevent 和 pthread 依赖，完全使用 iocp 和 windows 的线程机制实现了半同步半异步的框架。
0.9.2 移植了所有的功能到 windows 平台，同时新增加了 xyssl 的插件。

主页：
http://code.google.com/p/spserver/

源代码下载：
http://spserver.googlecode.com/files/spserver-0.6.src.tar.gz
http://code.google.com/p/spserver/downloads/list


在实现并发处理多事件的应用程序方面，有如下两种常见的编程模型：
ThreadPerConnection的多线程模型和事件驱动的单线程模型。

ThreadPerConnection的多线程模型
优点：简单易用，效率也不错。在这种模型中，开发者使用同步操作来编写程序，比如使用阻塞型I/O。使用同步操作的程序能够隐式地在线程的运行堆栈中维护应用程序的状态信息和执行历史，方便程序的开发。
缺点：没有足够的扩展性。如果应用程序只需处理少量的并发连接，那么对应地创建相应数量的线程，一般的机器都还能胜任；但如果应用程序需要处理成千上万个连接，那么为每个连接创建一个线程也许是不可行的。

事件驱动的单线程模型
优点：扩展性高，通常性能也比较好。在这种模型中，把会导致阻塞的操作转化为一个异步操作，主线程负责发起这个异步操作，并处理这个异步操作的结果。由于所有阻塞的操作都转化为异步操作，理论上主线程的大部分时间都是在处理实际的计算任务，少了多线程的调度时间，所以这种模型的性能通常会比较好。
缺点：要把所有会导致阻塞的操作转化为异步操作。一个是带来编程上的复杂度，异步操作需要由开发者来显示地管理应用程序的状态信息和执行历史。第二个是目前很多广泛使用的函数库都很难转为用异步操作来实现，即是可以用异步操作来实现，也将进一步增加编程的复杂度。

并发系统通常既包含异步处理服务，又包含同步处理服务。系统程序员有充分的理由使用异步特性改善性能。相反，应用程序员也有充分的理由使用同步处理简化他们的编程强度。

针对这种情况，ACE 的作者提出了 半同步/半异步 （Half-Sync/Half-Async） 模式。

引用

《POSA2》上对这个模式的描述如下：
半同步/半异步 体系结构模式将并发系统中的异步和同步服务处理分离，简化了编程，同时又没有降低性能。该模式介绍了两个通信层，一个用于异步服务处理，另一个用于同步服务处理。

目标：
需要同步处理的简易性的应用程序开发者无需考虑异步的复杂性。同时，必须将性能最大化的系统开发者不需要考虑同步处理的低效性。让同步和异步处理服务能够相互通信，而不会使它们的编程模型复杂化或者过度地降低它们的性能。

解决方案：
将系统中的服务分解成两层，同步和异步，并且在它们之间增加一个排队层协调异步和同步层中的服务之间的通信。在独立线程或进程中同步地处理高层服务（如耗时长的数据库查询或文件传输），从而简化并发编程。相反，异步地处理底层服务（如从网络连接上读取数据），以增强性能。如果驻留在相互独立的同步和异步层中的服务必须相互通信或同步它们的处理，则应允许它们通过一个排队层向对方传递消息。

模式原文：Half-Sync/Half-Async: An Architectural Pattern for Efficient and Well-structured Concurrent I/O
http://www.cs.wustl.edu/~schmidt/PDF/HS-HA.pdf
中文翻译：http://blog.chinaunix.net/u/31756/showart_245841.html

如果上面关于 半同步/半异步 的说明过于抽象，那么可以看一个《POSA2》中提到的例子：
许多餐厅使用 半同步/半异步 模式的变体。例如，餐厅常常雇佣一个领班负责迎接顾客，并在餐厅繁忙时留意给顾客安排桌位，为等待就餐的顾客按序排队是必要的。领班由所有顾客“共享”，不能被任何特定顾客占用太多时间。当顾客在一张桌子入坐后，有一个侍应生专门为这张桌子服务。

下面来看一个使用 spserver 实现的简单的 line echo server 。

class SP_EchoHandler : public SP_Handler {
public:
  SP_EchoHandler(){}
  virtual ~SP_EchoHandler(){}

  // return -1 : terminate session, 0 : continue
  virtual int start( SP_Request * request, SP_Response * response ) {
    request->setMsgDecoder( new SP_LineMsgDecoder() );
    response->getReply()->getMsg()->append(
      "Welcome to line echo server, enter 'quit' to quit.\r\n" );

    return 0;   
  }     

  // return -1 : terminate session, 0 : continue
  virtual int handle( SP_Request * request, SP_Response * response ) {
    SP_LineMsgDecoder * decoder = (SP_LineMsgDecoder*)request->getMsgDecoder();

    if( 0 != strcasecmp( (char*)decoder->getMsg(), "quit" ) ) {
      response->getReply()->getMsg()->append( (char*)decoder->getMsg() );
      response->getReply()->getMsg()->append( "\r\n" );
      return 0;         
    } else {    
      response->getReply()->getMsg()->append( "Byebye\r\n" );
      return -1;        
    }           
  }     
  virtual void error( SP_Response * response ) {}

  virtual void timeout( SP_Response * response ) {}

  virtual void close() {}
};

class SP_EchoHandlerFactory : public SP_HandlerFactory {
public:
  SP_EchoHandlerFactory() {}
  virtual ~SP_EchoHandlerFactory() {}

  virtual SP_Handler * create() const {
    return new SP_EchoHandler();
  }
};

int main( int argc, char * argv[] )
{
  int port = 3333;

  SP_Server server( "", port, new SP_EchoHandlerFactory() );
  server.runForever();

  return 0;
}

在最简单的情况下，使用 spserver 实现一个 TCP server 需要实现两个类：SP_Handler 的子类 和 SP_HandlerFactory 的子类。
SP_Handler 的子类负责处理具体业务。
SP_HandlerFactory 的子类协助 spserver 为每一个连接创建一个 SP_Handler 子类实例。

1.SP_Handler 生命周期
SP_Handler 和 TCP 连接一对一，SP_Handler 的生存周期和 TCP 连接一样。
当 TCP 连接被接受之后，SP_Handler 被创建，当 TCP 连接断开之后，SP_Handler将被 destroy。

2.SP_Handler 函数说明
SP_Handler 有 5 个纯虚方法需要由子类来重载。这 5 个方法分别是：
start：当一个连接成功接受之后，将首先被调用。返回 0 表示继续，-1 表示结束连接。
handle：当一个请求包接收完之后，将被调用。返回 0 表示继续，-1 表示结束连接。
error：当在一个连接上读取或者发送数据出错时，将被调用。error 之后，连接将被关闭。
timeout：当一个连接在约定的时间内没有发生可读或者可写事件，将被调用。timeout 之后，连接将被关闭。
close：当一个 TCP 连接被关闭时，无论是正常关闭，还是因为 error/timeout 而关闭。

3.SP_Handler 函数调用时机
当需要调用 SP_Handler 的 start/handle/error/timeout 方法时，相关的参数将被放入队列，然后由线程池来负责执行 SP_Handler 对应的方法。因此在 start/handle/error/timeout 中可以使用同步操作来编程，可以直接使用阻塞型 I/O 。
在发生 error 和 timeout 事件之后，close 紧跟着这两个方法之后被调用。
如果是程序正常指示结束连接，那么在主线程中直接调用 close 方法。

4.高级功能－－MsgDecoder
这个 line echo server 比起常见的 echo server 有一点不同：只有在读到一行时才进行 echo。
这个功能是通过一个称为 MsgDecoder 的接口来实现的。不同的 TCP server 在应用层的传输格式上各不相同。
比如在 SMTP/POP 这一类的协议中，大部分命令是使用 CRLF 作为分隔符的。而在 HTTP 中是使用 Header + Body 的形式。
为了适应不同的 TCP server，在 spserver 中有一个 MsgDecoder 接口，用来处理这些应用层的协议。
比如在这个 line echo server 中，把传输协议定义为：只有读到一行时将进行 echo 。
那么相应地就要实现一个 SP_LineMsgDecoder ，这个 LineMsgDecoder 负责判断目前的输入缓冲区中是否已经有完整的一行。

MsgDecoder 的接口如下：

class SP_MsgDecoder {
public:
  virtual ~SP_MsgDecoder();

  enum { eOK, eMoreData };
  virtual int decode( SP_Buffer * inBuffer ) = 0;
};

decode 方法对 inBuffer 里面的数据进行检查，看是否符合特定的要求。如果已经符合要求，那么返回 eOK ；如果还不满足要求，那么返回 eMoreData。比如 LineMsgDecoder 的 decode 方法的实现为：

int SP_LineMsgDecoder :: decode( SP_Buffer * inBuffer )
{               
  if( NULL != mLine ) free( mLine );
  mLine = inBuffer->getLine();
        
  return NULL == mLine ? eMoreData : eOK;
}   

spserver 默认提供了几个 MsgDecoder 的实现：
SP_DefaultMsgDecoder ：它的 decode 总是返回 eOK ，即只要有输入就当作是符合要求了。
    如果应用不设置 SP_Request->setMsgDecoder 的话，默认使用这个。
SP_LineMsgDecoder ： 检查到有一行的时候，返回 eOK ，按行读取输入。
SP_DotTermMsgDecoder ：检查到输入中包含了特定的 <CRLF>.<CRLF> 时，返回 eOK。

具体的使用例子可以参考示例：testsmtp 。

5.高级功能－－实现聊天室
spserver 还提供了一个广播消息的功能。使用消息广播功能可以方便地实现类似聊天室的功能。具体的实现可以参考示例：testchat 。

6.libevent
spserver 使用 c++ 实现，使用了一个第三方库－－libevent，以便在不同的平台上都能够使用最有效的事件驱动机制（Currently, libevent supports /dev/poll, kqueue(2), select(2), poll(2) and epoll(4). ）。

评论

16 楼 qiezi 2007-06-14  

你目前的实现方式是收到数据后，由msgdecoder来判断是否完成（是这样没错吧？），然后放入线程池我觉得欠点灵活性。

比如我有这样一个应用A，它非常简单，不需要放入线程池，收到后只需要根据长度就可以判断是否收完了，而且不耗CPU，所以不需要线程池，这时msgdecoder也是多余的了。

另一个应用B收到数据后要计算，所以交由线程池处理。

这样2个应用都兼顾的方式，我自已想了一种：
http://qiezi.iteye.com/blog/72929

写得很乱，个人习惯不好，我大致讲一下使用方式吧。由于C＋＋没有委托，所以实现起来可能要用static方法来做，下面的代码用D语法来演示。

class Protocol
{
	/**
	 * bytes: 要读取的字节数
	 * callback: 读取完成时的回调委托函数
	 * recvall: 是否要读取所有请求的字节数才回调
	 * threading: 是否交由线程池
	 *
	 * void delegate(char[] data, bool finished) 参数：
	 * data: 数据
	 * finished: 是否已经接收完成。对于recvall为true的，回调时这个参数肯定为true，否则可能为true或false
	 */
	void requestRead(size_t bytes, void delegate(char[], bool) callback, bool recvall, bool threading);
}

class AProtocol : public Protocol
{
    void connectionMade() // 连接建立时调用
	{
		// 请求接收4字节包头长度，必须收满4字节才回调，不放入线程池
		requestRead(4, &readPackageLength, true, false);
	}

	void readPackageLength(char[] data, bool finished)
	in {
		assert(data.length == 4);
		assert(finished);
	}
	body {
		int len = ntohl(*cast(int*)data.ptr);

		writefln("Received package length: %d", len);

		// 请求读取包内容
		requestRead(len, &readPackageData, true, false);
	}

	void readPackageData(char[] data, bool finished)
	{
		// send back
		send(data); // 此时没有放入线程池，可以直接设置EV_WRITE
	}
}


class BProtocol : public Protocol
{
    void connectionMade() // 连接建立时调用
	{
		// 请求接收4字节包头长度，必须收满4字节才回调，不放入线程池
		requestRead(4, &readPackageLength, true, false);
	}

	void readPackageLength(char[] data, bool finished)
	in {
		assert(data.length == 4);
		assert(finished);
	}
	body {
		int len = ntohl(*cast(int*)data.ptr);

		writefln("Received package length: %d", len);

		// 请求读取包内容，交由线程池处理
		requestRead(len, &readPackageData, true, true);
	}

	void readPackageData(char[] data, bool finished)
	{
		// 解出包头，密集型计算
		// ....
		send(data); // 此时不直接发送，放入发送缓冲区
	}
}

接口方式类似于twisted，改进了一些，算是reactor + proactor。目前还在构思阶段，实现还没开始。

15 楼 iunknown 2007-06-08  

引用

你的msgqueue对我非常有帮助，原来正为libevent 多线程困扰呢，把任务放到队列容易，要在完成时得到通知就比较麻烦

呵呵，的确是从 event loop 转到线程池容易，从线程池要转到 event loop 就比较麻烦了。

msgqueue 不是我实现的，是借来的。在 msgqueue 文件中保留了作者的声明。之前也发过 mail 给作者询问，在得到他允许之后才一起发布的。作者的原帖在这里
http://monkeymail.org/archives/libevent-users/2006-October/000257.html

引用

另一个问题，没考虑监听多个端口？多种协议？

这个没有问题啊。spserver 这个类有两个方法：
run 启动一个后台线程作为 event loop 线程，然后把执行流程返回给调用者；
runForever 不启动后台线程，直接使用调用者的线程。
如果要监听多个端口，那么只要初始化多个 spserver 对象就行了。每个 spserver 用不同的 HandlerFactory 来初始化。

引用

另外还可以考虑一下connector的实现

开始做的时候就考虑过这个问题，不过没有想到很好的办法，所以暂时先不实现，先把自己想清楚的部分先实现了。等想清楚了再看看是不是加进去。

引用

补充一下，说decoder写代码比较多，实际上也因为状态机的解析方式

的确，现在 spserver 的 msgdecoder 基本上可以认为是一个基于流的协议解释器。比如 spserver 自带的 HttpMsgDecoder 就是一个基于流的 http 协议解释器。因为这个框架最初做的时候是以实现 xmpp 服务器作为目标的，而 xmpp 的协议就是一个基于 xml 流的协议，所以整个框架就是倾向于这种做法。另外在开始实现 spserver 之前，刚好又看到了 mina 的介绍，所以这个 msgdecoder 的名字其实是从 mina 借来的，:)

引用

我在自己做协议时喜欢用长度前缀，先确定长度字节，收齐了这几字节就解出长度，再判断是否收齐剩下的，直接交给线程池就行了，中间只有一个状态，比较简单。

对于这种格式的协议，使用 msgdecoder 同样只有一个状态。原来的 msgdecoder 有一个 getMsg 的接口方法，现在已经去掉了，因为这个方法在框架中并不是必须的。也就是现在 msgdecoder 变成这样了

class SP_MsgDecoder {  
public:  
  virtual ~SP_MsgDecoder();  
   
  enum { eOK, eMoreData };  
  virtual int decode( SP_Buffer * inBuffer ) = 0;  
};  

在这段时间，我也一致在考虑是否有更好的实现 msgdecoder 这部分功能的方法。暂时还没想到更好的方法。只是对 msgdecoder 的认识更清晰了一点。就是这个接口只是定义了框架怎么回调特定的协议解释接口，至于在解释了协议之后，怎么获得解释的结果，就不是框架要关心的了，而是由具体的实现自己来决定。所以就去掉了原来的 getMsg 这个接口。

最近用 spserver 实现了一个 xmpp 服务器，在实现过程中，发现这个 msgdecoder 对协议解释部分和协议处理部分做了很好的解藕。在协议处理部分可以实现到 mina 宣称的“你只需撰写面向POJO的message而不是ByteBuffer的”。

引用

header解析比较简单不讨论了，主要是在收齐header以后，要开始根据文件长度字段收文件内容，边收边写文件

可以分成两个 msgdecoder ，一个用来解释 header ，在收齐 header 之后，decode 接口就返回 eOK 。然后设置 msgdecoder 为 defaultmsgdecoder ，这样就会“边收边调用handler::handle接口”，也就可以在handler::handle 接口中“边写文件”了。

14 楼 qiezi 2007-06-07  

补充一下，说decoder写代码比较多，实际上也因为状态机的解析方式，虽然高效，写出来感觉并不好看，当然有些协议也只能这么写。我在自己做协议时喜欢用长度前缀，先确定长度字节，收齐了这几字节就解出长度，再判断是否收齐剩下的，直接交给线程池就行了，中间只有一个状态，比较简单。

另外我有一个需求，你看看用你的架构是否有不便之处：

一个上传服务器协议格式：
[header][file content]

header是长度前缀的，包括4个字段，协议类型、要写入的路径、文件长度、MD5。header解析比较简单不讨论了，主要是在收齐header以后，要开始根据文件长度字段收文件内容，边收边写文件，如果用你的spserver架构，实现这个东东，如何才能有效利用起线程池？仅作讨论。

13 楼 qiezi 2007-06-07  

另一个问题，没考虑监听多个端口？多种协议？我目前实现的参照twisted接口，已经把多个端口、协议考虑在内，目前只实现了基本的单线程reactor，执行线程池还在计划之中。

另外还可以考虑一下connector的实现，twisted里面有些思想非常好，比如解析host也是异步的，这在libevent里面也有，我有时候还想是不是这2个不同语言的库也相互参照呢？

12 楼 qiezi 2007-06-07  

不错，我感觉稍稍重了一点，实现一个Message Decoder要写的代码稍稍多了些。另外有没有考虑使用libevent的bufferevent来做？我最近也在做空上，参照twisted的接口来做的，不过包装得比较浅。你的msgqueue对我非常有帮助，原来正为libevent 多线程困扰呢，把任务放到队列容易，要在完成时得到通知就比较麻烦，刚好看到你的实现。另外，没考虑过跨平台吗？

11 楼 iunknown 2007-05-22  

有了 epoll 和 iocp 之后，要做到高并发已经有了很好的基础了。但是真正要做好一个高并发的服务器却不是一件容易的事。

因为 epoll 和 iocp 只是提供了一个高效的事件通知机制，对于实现者来说，就是可以同时处理很多的句柄（比如 Socket，File等）而不用担心操作系统在事件通知上的消耗。

但同时我们可以看到，直接使用 epoll 或者 iocp ，要求是要直接使用底层的句柄，对于一些已经封装过的 api （比如 mysql client api），是没有办法使用 epoll 或者 iocp 来处理的。如果在高并发服务器中，用到了这些 api 的话，那么就需要做一些特殊的处理，否则就会由于这些 api 会阻塞，造成整个服务器阻塞。

fredzhang 写道

不过即便如此，epoll的并发能力比windows的iocp也要高。

在没有任何背景的情况下，这样说有点不合适。如果说服务器是一个什么都不做的，或者只是做一个简单的 echo server 的话，那么 epoll 可能比 iocp 的并发高。但是这样的比较没什么意义。
如果这个服务器要做一些很具体的业务处理，那么高并发的的瓶颈往往已经不在 epoll 或者 iocp 上，而是在怎么处理具体的业务上。包括怎样集成那些会造成 block 的 api ，怎么使用更好的算法，等等。
即使以 memcached 为例，估计用 epoll 或者 iocp 都能支持到同样的并发度。因为在并发数不断升高的时候， memcached 的瓶颈并不在 epoll 或者 iocp 的调度上，而是在于网络的传输速度和 hash table 的查找。即在达到能够区分出 epoll 和 iocp 谁的并发度更高的时刻之前，memcached 就已经因为其他的瓶颈而无法再支持更高的并发。

10 楼 fredzhang 2007-05-22  

aio在linux下的实现采用NPTL应当跟windows的iocp差不多了，都是用用户线程池来模拟的。linux下比较高效的并发模式还是采用epoll，aio只是个posix标准，在linux下没有真正实现。不过即便如此，epoll的并发能力比windows的iocp也要高。

9 楼 iunknown 2007-05-09  

qiek 写道

问一下这种情况原因是什么
在以下方法中:
XXXHandler :: handle( SP_Request * request, SP_Response * response )
{
....
    SP_Buffer * outBuffer = response->getReply()->getMsg();
    std::string strResult;
....
outBuffer->append(strResult.c_str(), strResult.size());
......
}

如果strResult的长度比较大，譬如是211984，则客户端会不停的收到数据，一直不能停止，不知道能否查一下原因！

谢谢你的测试，bug 已经查明，经本地测试已经没有问题。

最新代码已经 commit 到 googlecode，请使用如下命令行 update 最新源代码：

svn checkout http://spserver.googlecode.com/svn/trunk/ spserver

由于目前我正在实现一个基于 spserver 的嵌入式 http 框架，过段时间才会发布一个新版本。
所以现在请先用 svn 直接从源代码库中 update 。

8 楼 qiek 2007-05-08  

问一下这种情况原因是什么
在以下方法中:
XXXHandler :: handle( SP_Request * request, SP_Response * response )
{
....
    SP_Buffer * outBuffer = response->getReply()->getMsg();
    std::string strResult;
....
outBuffer->append(strResult.c_str(), strResult.size());
......
}

如果strResult的长度比较大，譬如是211984，则客户端会不停的收到数据，一直不能停止，不知道能否查一下原因！

7 楼 SteveGY 2007-03-31  

aio, Solaris的还是比较强的，Linux的实现不怎么样

6 楼 Arath 2007-03-15  

喔，谢谢楼上两位~

5 楼 iunknown 2007-03-15  

在 unix/linux 下和 Windows Completion Port 最类似的应该是 aio ，epoll 是 select/poll 的加强版。

4 楼 simohayha 2007-03-15  

Arath 写道

在Windows下只有Completion Port可以满足大量连接的需求，其实原理很简单，就是减少了线程，从而大量减少了系统开销.
对Unix/Linux开发不了解，不知道Unix/Linux下类似的方法是什么?

Linux下是epoll.

3 楼 Arath 2007-03-15  

在Windows下只有Completion Port可以满足大量连接的需求，其实原理很简单，就是减少了线程，从而大量减少了系统开销.
对Unix/Linux开发不了解，不知道Unix/Linux下类似的方法是什么?

2 楼 simohayha 2007-03-14  

并发效率的话可以看看这个
http://shootout.alioth.debian.org/gp4/benchmark.php?test=message&lang=all

http://shootout.alioth.debian.org/debian/benchmark.php?test=message&lang=all

1 楼 bigpanda 2007-03-14  

去年做过一段时间Streaming Server，钻研过一段时间这些东东，并发效率最高的，还是微软的IoCompletionPorts，好牛啊，可惜后来项目取消了，没有做下去。

http://www.microsoft.com/technet/sysinternals/information/IoCompletionPorts.mspx