Poco::TCPServer框架解析

Poco::TCPServer框架解析

POCO C++ Libraries提供一套 C++ 的类库用以开发基于网络的可移植的应用程序，功能涉及线程、文件、流，网络协议包括：HTTP、FTP、SMTP 等，还提供 XML 的解析和 SQL 数据库的访问接口。不仅给我的工作带来极大的便利，而且设计巧妙，代码易读，注释丰富，也是非常好的学习材料，我个人非常喜欢。POCO的创始人在这个开源项目的基础上做了一些收费产品，也成立了自己的公司，"I am in the lucky position
to work for my own company"，真是让人羡慕啊。

POCO C++ Libraries



基于POCO的产品



Poco主页：http://pocoproject.org/

Poco文档：http://pocoproject.org/docs/

创始人主页：http://obiltschnig.com/

公司主页：http://www.appinf.com/

我主要用过Net，Data，XML部分，Net里对socket的封装类，实现TCP，HTTP，SMTP协议的框架，Data里对MySQL接口封装，XML里对DOM标准的实现。我目前主要是做C++网络编程，曾经尝试接触ACE库，但觉得太复杂难理解，而且也没有条件用于项目，后来发现了Poco，不仅简单易用，而且也包含《C++ Networking Programming》中提到的各种模式和框架，更难得的是文档注释丰富，看起源码来相当舒服。因此想写一些笔记，一方面借此加深对网络编程的理解，另一方面也希望和大家多多交流。

第一篇讨论Poco::TCPServer框架，先看看Poco::TCPServer有哪些类


class Net_API TCPServer: public Poco::Runnable


class Net_API TCPServerConnection: public Poco::Runnable


class Net_API TCPServerConnectionFactory


class Net_API TCPServerDispatcher: public Poco::Runnable


class Net_API TCPServerParams: public Poco::RefCountedObject

Runnable是一个抽象类，要求其子类实现一个run()函数，run()一般作为线程的入口函数.

先看看run函数是如何被调用的，以TCPServer为例。


void TCPServer::start()


{


poco_assert (_stopped);




_stopped = false;


_thread.start(*this);//这里传入本类对象地址


}

再看看Thread类的start函数


class Foundation_API Thread: private ThreadImpl


void Thread::start(Runnable& target)


{


startImpl(target);


}

这里的ThreadImpl::startImpl是POSIX实现版本


void ThreadImpl::startImpl(Runnable& target)


{




_pData->pRunnableTarget = ⌖


if (pthread_create(&_pData->thread, &attributes, runnableEntry, this))


{


_pData->pRunnableTarget = 0;


throw SystemException("cannot start thread");


}



pRunnableTarget就指向Runnable了


void* ThreadImpl::runnableEntry(void* pThread)


{




ThreadImpl* pThreadImpl = reinterpret_cast<ThreadImpl*>(pThread);


AutoPtr<ThreadData> pData = pThreadImpl->_pData;


try


{


pData->pRunnableTarget->run();//在这里


}


catch (Exception& exc)



知道run()是如何被调用了，再看看TcpServer::run()的实现

1

void TCPServer::run()

2

{

3

while (!_stopped)

4

{

5

Poco::Timespan timeout(250000);

6

if (_socket.poll(timeout, Socket::SELECT_READ))

7

{

8

try

9

{

10

StreamSocket ss = _socket.acceptConnection();

11

// enabe nodelay per default: OSX really needs that

12

ss.setNoDelay(true);

13

_pDispatcher->enqueue(ss);

14

}

15

catch (Poco::Exception& exc)

16




在第6行调用标准select函数，准备好读后，第10行调用标准accept建立连接，然后把StreamSocke对象放入TCPServerDispatcher对象内的队列中，可见TCPServer只是建立连接，之后的工作就都交给TCPServerDispatcher了。

在讲TCPServerDispatcher之前，先需要说明的是第10行TCPServer的_socket成员变量类型是ServerSocket，ServerSocket在一个构造函数中调用了bind和listen，具体如下


ServerSocket::ServerSocket(Poco::UInt16 port, int backlog): Socket(new ServerSocketImpl)


{


IPAddress wildcardAddr;


SocketAddress address(wildcardAddr, port);


impl()->bind(address, true);


impl()->listen(backlog);


}
要注意，尽管ServerSocket类提供了无参构造函数，但使用无参构造函数创建的对象，在TCPServer对象调用start()之前，必须先bind和listen。

" /// Before start() is called, the ServerSocket passed to

/// TCPServer must have been bound and put into listening state."

继续看TCPServerDispatcher，沿着TcpServer::run()第13行，我们看它的equeue()

1

void TCPServerDispatcher::enqueue(const StreamSocket& socket)

2

{

3

FastMutex::ScopedLock lock(_mutex);

4


5

if (_queue.size() < _pParams->getMaxQueued())

6

{

7

_queue.enqueueNotification(new TCPConnectionNotification(socket));

8

if (!_queue.hasIdleThreads() && _currentThreads < _pParams->getMaxThreads())

9

{

10

try

11

{

12

static const std::string threadName("TCPServerConnection");

13

_threadPool.start(*this, threadName);

14

++_currentThreads;

15

}

16

catch (Poco::Exception&)

17




第7行enqueueNotification把一个TCPConnectionNotification入队，第13行把this交给ThreadPool启动一个线程，那么这个线程就要运行TCPServerDispatcher的run函数了，看TCPServerDispatcher::run()

1

void TCPServerDispatcher::run()

2

{

3

AutoPtr<TCPServerDispatcher> guard(this, true); // ensure object stays alive

4


5

int idleTime = (int) _pParams->getThreadIdleTime().totalMilliseconds();

6


7

for (;;)

8

{

9

AutoPtr<Notification> pNf = _queue.waitDequeueNotification(idleTime);

10

if (pNf)

11

{

12

TCPConnectionNotification* pCNf = dynamic_cast<TCPConnectionNotification*>(pNf.get());

13

if (pCNf)

14

{

15

std::auto_ptr<TCPServerConnection> pConnection(_pConnectionFactory->createConnection(pCNf->socket()));

16

poco_check_ptr(pConnection.get());

17

beginConnection();

18

pConnection->start();

19

endConnection();

20

}

21

}



第9行waitDequeueNotification一个Notification，第12行把这个通知类型转换成TCPConnectionNotification，联系之前的enqueueNotification，大概能才到是什么意思。第15行又出现个TCPServerConnection。好吧，看来搞清楚TCPServerDispatcher还是要先看下TCPServerConnection，还有TCPConnectionNotification。

尽管TCPServerConnection继承了Runable，但没有实现run()，它的start()如下


void TCPServerConnection::start()


{


try


{


run();


}





用户需要继承TCPServerConnection实现run函数，看下源码中的说明，要完成对这个socket的处理，因为run函数返回时，连接就自动关闭了。


class Net_API TCPServerConnection: public Poco::Runnable


/// The abstract base class for TCP server connections


/// created by TCPServer.


///


/// Derived classes must override the run() method


/// (inherited from Runnable). Furthermore, a


/// TCPServerConnectionFactory must be provided for the subclass.


///


/// The run() method must perform the complete handling


/// of the client connection. As soon as the run() method


/// returns, the server connection object is destroyed and


/// the connection is automatically closed.


///


/// A new TCPServerConnection object will be created for


/// each new client connection that is accepted by


/// TCPServer.

连接在哪被关闭的呢，注释说随着TCPServerConnection对象销毁而关闭。具体就是，TCPServerConnection类有个StreamSocket成员，StreamSocket继承自Socket，Socket类又包含了个SocketImpl成员，所以就有~TCPServerConnection()->~StreamSocket()->~Socket()->~SocketImpl()，最后~SocketImpl()调用close()关闭了连接。

那么TCPServerConnection对象何时被创建何时被销毁呢，这下又回到TCPServerDispatcher::run()来了，看TCPServerDispatcher::run()的第15行，创建了TCPServerConnection对象，第18行调用TCPServerConnection::start()进而调用TCPServerConnection::run()，第20行块结束，TCPServerConnection对象随着智能指针销毁而被销毁。

还剩TCPConnectionNotification或者说Notification要搞清楚了，但是要对Poco的Notifactions模块源码进行分析的话，本文的篇幅也就太长了，就从文档来大致看看吧

Notifications

Facilities for type-safe sending and delivery of notification objects within a single thread or from one thread to another, also well suited for the implementation of notification mechanisms. A notification queue class for distributing tasks to worker threads,
simplifying the implementation of multithreaded servers.

简单的说Notifications模块可用于线程间传递消息，简化多线程服务器的实现。具体到TCPServer，就是把已连接的socket，放到NotificationQueue中，并从TheadPool出来一个线程，线程从NotificationQueue取到这个socket，从而用TCPServerConnection::run()里的逻辑对socket进行处理。

至此TCPServer基本分析完毕，还有TCPServerConnectionFactory和TCPServerParams，分别用于产生TCPServerConnection和设置参数，就不细说了。

纵观Poco::TCPServer，套一下《Unix Network Programming》里讲的服务器模型，属于“在程序启动阶段创建一个线程池之后让主线程调用accept，并把每个客户连接传递给池中某个可用线程”。Poco1.3.6版里用select作为IO multiplexing。1.3.7版正在尝试epoll(windows平台依然是select)，但还未release，跟踪svn来看也没有使用edge-triggered模式。套用《The
C10K problem》里讲的服务器模型，属于"Serve many clients with each thread, and use nonblocking I/O and level-triggered readiness notification"。总结起来就是：non-blocking IO + IO multiplexing(level-triggered) + thread pool。

Poco::TCPServer也许并不能算一个性能很高的TCP服务器，但我非常喜欢它的设计和编码风格。顺便提一下对底层socket的封装，由socket类派生的各种子类，ServerSocket在构造函数中进行bind和listen，StreamSocket在构造函数进行connect，都是非常贴心的设计。

下一篇打算分析Poco的Reator框架。