Posted on 2007-06-19 16:31
小明 阅读(7608)
评论(1) 编辑 收藏 引用 所属分类:
Network/ACE
这个示例将用ACE Proactor来实现echo server
代码量差不多只有完成端口的1/3
ACE Reactor的实现
http://www.cppblog.com/sandy/archive/2006/02/17/3308.html
完成端口的实现
http://www.cppblog.com/sandy/archive/2007/06/06/25670.html当然如果你理解了完成端口，ACE Proactor就很简单了
大概的过程是这个样子的
1.使用ACE_Asynch_Acceptor来建立服务器端的socket,然后bind,listen等待Accept并创建IOCP
(相对于ACE Reactor的ACE_Acceptor）
2.当出现完成事件（GetQueuedCompletionStatus）返回时，调用用户的处理接口来处理，这是一个派生于ACE_Service_Handler的子类
（相对于ACE Reactor的ACE_Svc_Handler）
ACE Reactor和Proactor最大的不同就是
ACE Reactor在处理读写事件的时候，要使用peer().recv/send来接收/发送数据。
而ACE Proactor在处理读写事件的时候,数据已经被读到用户的缓冲区了，典型的"先斩后奏"
ACE Proactor比ACE Reactor麻烦的地方是需要手工去投递异步读写的请求。
可以通过下面的代码来去体会。
==========代码===============
#ifdef _DEBUG
#pragma comment(lib,"ACED.lib")
#else
#pragma comment(lib,"ACE.lib")
#endif
#include "ace/OS_main.h"
#include "ace/OS_NS_sys_socket.h"
#include "ace/ACE.h"
#include "ace/Service_Object.h"
#include "ace/Asynch_IO.h"
#include "ace/Proactor.h"
#include "ace/message_block.h"
#include "ace/Asynch_Acceptor.h"
#include "ace/INET_Addr.h"
class Echo_Service : public ACE_Service_Handler
{
public:
~Echo_Service ()
{
if (this->handle () != ACE_INVALID_HANDLE)
ACE_OS::closesocket (this->handle ());
}
virtual void open (ACE_HANDLE h, ACE_Message_Block&)
{
this->handle (h);
if (this->reader_.open (*this) != 0 || this->writer_.open (*this) != 0 )
{
ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"),ACE_TEXT ("Echo_Service open")));
delete this;
return;
}
ACE_Message_Block *mb;
ACE_NEW_NORETURN (mb, ACE_Message_Block (1024));
if (this->reader_.read (*mb, mb->space ()) != 0)
{
ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"),ACE_TEXT ("Echo_Service begin read")));
mb->release ();
delete this;
return;
}
}
virtual void handle_read_stream(const ACE_Asynch_Read_Stream::Result &result)
{
ACE_Message_Block &mb = result.message_block ();
if (!result.success () || result.bytes_transferred () == 0)
{
mb.release ();
delete this;
}
else
{
if (this->writer_.write (mb, mb.length ()) == -1)
{
ACE_ERROR ((LM_ERROR,
ACE_TEXT ("%p\n"),
ACE_TEXT ("starting write")));
mb.release ();
}
else
{
ACE_Message_Block *new_mb;
ACE_NEW_NORETURN (new_mb, ACE_Message_Block (1024));
this->reader_.read (*new_mb, new_mb->space ());
}
}
return;
}
virtual void handle_write_stream
(const ACE_Asynch_Write_Stream::Result &result)
{
ACE_Message_Block &mb = result.message_block ();
mb.release ();
return;
}
private:
ACE_Asynch_Read_Stream reader_;
ACE_Asynch_Write_Stream writer_;
};
typedef ACE_Asynch_Acceptor<Echo_Service> MyAcceptor;
int main(int argc, char *argv[])
{
ACE_INET_Addr addr(1500);
MyAcceptor server;
if(server.open(addr)==-1)
{
ACE_DEBUG ((LM_DEBUG,
ACE_TEXT ("(%P|%t) %p\n"),
ACE_TEXT ("bind failed")));
return 1;
}
while(1){
ACE_Proactor::instance ()->proactor_run_event_loop ();
}
return 0;
}