您当前的位置： 首页 > 

phymat.nico

暂无认证

  • 1浏览

    0关注

    1967博文

    0收益

  • 0浏览

    0点赞

    0打赏

    0留言

私信
关注
热门博文

ACE_Select_Reactor 一 ——入门

phymat.nico 发布时间：2017-12-20 23:54:05 ，浏览量：1

ACE_Select_Reactor 1 ——入门

    ACE Reactor 框架实现了Reactor模式，允许事件驱动的应用对源自许多不同事件源的事件做出反映，比如IO句柄，定时器，以及信号，应用重新定义框架所定义的挂钩方法。框架随机对其进行分派来处理事件，Reactor负责：（1）检查多路分离器来自各种事件源的、不同类型的连接和数据事件，（2）将这些事件分派给应用所定义的处理器，由它进行处理。

    反应式服务器响应来自一个或多个事件源的时间，在理想情况下，对时间的响应会足够快，以使所有请求看起来像是被同时处理的，尽管事件处理通常是由单个线程处理的。同步事件多路分离器位于各个反应式服务器的心脏处。这种机制检测源自许多事件源的事件并对其作出响应，从而同步地使事件作为服务器正常执行路径的一部分提供给服务器。

    select（）函数是最为常见的同步事件多路分离器。这个系统函数在同一组IO句柄上等待指定的事件发生，当一个或者是多个IO句柄开始活动时，或是在指定的时间过去后，select函数就会返回。

    ACE_Select_Reactor是ACE_Reactor接口的一种实现，它使用select同步时间多路分离器函数来检测IO和定时器事件，除了支持ACE_Reactor接口的所有特性外，ACE_Select_Reactor类还提供了以下能力：

    1、它支持重入的反应器调用，应用可以从正在由统一反应器分派的事件处理器中调用handle_event方法；

    2、它可以被配置为同步化的或异步化的，在线程安全性和降低开销之间进行折中；

    3、它在再次调用select函数之前，分派其句柄集中的所有活动句柄，从而保证了公正性。

 

 

[cpp] view plain copy
  1. #include "ace/streams.h"  
  2. #include "ace/Reactor.h"  
  3. #include "ace/Select_Reactor.h"  
  4. #include "ace/Thread_Manager.h"  
  5.   
  6. #include   
  7.   
  8. #include "Reactor_Logging_Server_T.h"  
  9. #include "Logging_Acceptor_Ex.h"  
  10.   
  11. typedef Reactor_Logging_Server  
  12.         Server_Logging_Daemon;  
  13.   
  14.   
  15. static ACE_THR_FUNC_RETURN event_loop (void *arg)   
  16. {  
  17.   ACE_Reactor *reactor = static_cast (arg);  
  18.   
  19.   reactor->owner (ACE_OS::thr_self ());  
  20.   reactor->run_reactor_event_loop ();  
  21.   return 0;  
  22. }  
  23.   
  24.   
  25. static ACE_THR_FUNC_RETURN controller (void *arg)   
  26. {  
  27.   for (;;)   
  28.   {  
  29.     std::string user_input;  
  30.     std::getline (cin, user_input, '\n');  
  31.     if (user_input == "quit")   
  32.     {  
  33.       break;  
  34.     }  
  35.   }  
  36.   return 0;  
  37. }  
  38.   
  39.   
  40. int main(int argc,char **argv)  
  41. {  
  42.   ACE_Select_Reactor select_reactor;  
  43.   ACE_Reactor reactor (&select_reactor);  
  44.   
  45.   Server_Logging_Daemon *server;  
  46.   // Ignore argv[0]...  
  47.   --argc; ++argv;  
  48.   ACE_NEW_RETURN (server,  
  49.                   Server_Logging_Daemon (argc, argv, &reactor),  
  50.                   1);  
  51.   ACE_Thread_Manager::instance ()->spawn (event_loop, &reactor);  
  52.   ACE_Thread_Manager::instance ()->spawn (controller, &reactor);  
  53.   return ACE_Thread_Manager::instance ()->wait ();  
  54. }  

    在上面的代码中，在51行，由ACE_Thread_Manager单体派生一个线程，并让其运行event_loop()函数，在52行，由ACE_Thread_Manager单体派生一个线程，让其运行controller()函数。

    在从main函数返回之前，等待其他两个线程推出，ACE_Thread_Manager:wait()还会收取两个线程的退出状态，以免内存泄漏。

ACE_Select_Reactor 2 —— 服务器网关

    ACE中的流包装提供面向连接的通信。流数据传输包装类包括ACE_SOCK_Stream和ACE_LSOCK_Stream，他们分别包装TCP/IP和UNIX域Socket协议数据传输功能。连接建立类包括针对TCP/IP的ACE_SOCK_Connector和ACE_SOCK_Acceptor，以及针对UNIX域Socket的ACE_LSOCK_Connector和ACE_LSOCK_Acceptor。

 

 

[cpp] view plain copy
  1. class Server  
  2. {  
  3. public:  
  4.     Server (int port): server_addr_(port),peer_acceptor_(server_addr_)  
  5.     {  
  6.         data_buf_= new char[SIZE_BUF];  
  7.     }  
  8.     int handle_connection()  
  9.     {  
  10.         // Read data from client  
  11.             int byte_count=0;  
  12.             if( (byte_count=new_stream_.recv (data_buf_, SIZE_DATA, 0))==-1)  
  13.                 ACE_ERROR ((LM_ERROR, "%p\n", "Error in recv"));  
  14.             else  
  15.             {  
  16.                 data_buf_[byte_count]=0;  
  17.                 ACE_DEBUG((LM_DEBUG,"Server received %s \n",data_buf_));  
  18.             }  
  19.         // Close new endpoint  
  20.         if (new_stream_.close () == -1)  
  21.             ACE_ERROR ((LM_ERROR, "%p\n", "close"));  
  22.         return 0;  
  23.     }  
  24.     int accept_connections ()  
  25.     {  
  26.         if (peer_acceptor_.get_local_addr (server_addr_) == -1)  
  27.             ACE_ERROR_RETURN ((LM_ERROR,"%p\n","Error in get_local_addr"),1);  
  28.         ACE_DEBUG ((LM_DEBUG,"Starting server at port %d\n",  
  29.             server_addr_.get_port_number ()));  
  30.         while(1)  
  31.         {  
  32.             ACE_Time_Value timeout (ACE_DEFAULT_TIMEOUT);  
  33.             if (peer_acceptor_.accept (new_stream_, &client_addr_, &timeout)== -1)  
  34.             {  
  35.                 ACE_ERROR ((LM_ERROR, "%p\n", "accept"));  
  36.                 continue;  
  37.             }  
  38.             else  
  39.             {  
  40.                 ACE_DEBUG((LM_DEBUG,  
  41.                     "Connection established with remote %s:%d\n",  
  42.                     client_addr_.get_host_name(),client_addr_.get_port_number()));  
  43.                 //Handle the connection  
  44.                 handle_connection();  
  45.             }  
  46.         }  
  47.     }  
  48. private:  
  49.     char *data_buf_;  
  50.     ACE_INET_Addr server_addr_;  
  51.     ACE_INET_Addr client_addr_;  
  52.     ACE_SOCK_Acceptor peer_acceptor_;  
  53.     ACE_SOCK_Stream new_stream_;  
  54. };  

     在上面的Server类中，创建了一个被动服务器，侦听到来的客户端连接，在连接建立之后，服务器接收来自客户端的数据，然后关闭链接。

    Server类包含的accept_connection()方法使用接收器来将连接接受“进”ACE_SOCK_Stream new_stream_。该操作完成的基本流程是：调用接收器上的accept()，并将流作为参数传入其中。一旦连接已经建立进流中，流的包装方法send()和recv()就可以用来在新建立的链路上发送和接收数据，还有一个空的ACE_INET_Addr也被传入接收器的accept()方法，并在其中被设定为发起连接的远程机器地址。

    在连接建立后，服务器调用handle_connection()方法，它开始从客户端那里收取一个预先知道的单词，然后将流关闭。连接关闭通过调用流上的close()方法来完成，该方法会释放所有的Socket资源并终止连接。

 

    http://acme-ltt.iteye.com/blog/1455556中提到的ACE_Select_Reactor，在static ACE_THR_FUNC_RETURN controller (void *arg)函数中，调用上述的Server类，搭建基于ACE_Select_Reactor的Socket服务器网关。

 

 

    客户端程序：

 

[cpp] view plain copy
  1. #include "ace/SOCK_Connector.h"  
  2. #include "ace/INET_Addr.h"  
  3. #include "ace/OS.h"  
  4. #include "ace/Log_Msg.h"  
  5. #include   
  6. #include "text.h"  
  7. #include "ace/Thread_Manager.h"  
  8. #include   
  9. #define SIZE_BUF 128  
  10. static const ACE_Time_Value TIME_INTERVAL(0, 1000000);  
  11. class Client  
  12. {  
  13. public:  
  14.     Client(char *hostname, int port):remote_addr_(port,hostname)  
  15.     {     
  16.           
  17.           
  18.     }  
  19.     int connect_to_server()  
  20.     {  
  21.           
  22.         ACE_DEBUG ((LM_DEBUG, "(%P|%t) Starting connect to %s:%d\n",  
  23.             remote_addr_.get_host_name(),remote_addr_.get_port_number()));  
  24.         if (connector_.connect (client_stream_, remote_addr_) == -1)  
  25.             ACE_ERROR_RETURN ((LM_ERROR,"(%P|%t) %p\n","connection failed"),-1);  
  26.         else  
  27.             ACE_DEBUG ((LM_DEBUG,"(%P|%t) connected to %s\n",  
  28.             remote_addr_.get_host_name ()));  
  29.         return 0;  
  30.     }  
  31.       
  32.     int send_to_server()  
  33.     {  
  34.         iovec iov[3];  
  35.         iov[0].iov_base = (void *)"get";  
  36.         iov[0].iov_len = 3;  
  37.         iov[1].iov_base = getdata()/* some data */;  
  38.         iov[1].iov_len = strlen(getdata());  
  39.         iov[2].iov_base = (void *)"end.";  
  40.         iov[2].iov_len = 4;  
  41.           
  42.         if (client_stream_.sendv_n (iov,3) == -1)  
  43.         {  
  44.             ACE_ERROR_RETURN ((LM_ERROR,"(%P|%t) %p\n","send_n"),0);  
  45.             //break;  
  46.             exit(0);  
  47.         }  
  48.         close();  
  49.     }  
  50.     int close()  
  51.     {  
  52.         if (client_stream_.close () == -1)  
  53.             ACE_ERROR_RETURN ((LM_ERROR,"(%P|%t) %p\n","close"),-1);  
  54.         else  
  55.             return 0;  
  56.     }  
  57. private:  
  58.     ACE_SOCK_Stream client_stream_;  
  59.     ACE_INET_Addr remote_addr_;  
  60.     ACE_SOCK_Connector connector_;  
  61.     char *data_buf_;  
  62. };  
  63. int main (int argc, char *argv[])  
  64. {  
  65.     if(argc  acceptor_type;
 
      EchoService()
     {
          printf( "EchoService 创建\n");
     }
 
     ~ EchoService()
     {
          printf( "EchoService 销毁\n");
     }
 
      //  响应socket 已经打开，连接已经建立事件
      int  open( void*)
     {
          //  注册读事件
          if ( reactor()-> register_handler( this, ACE_Event_Handler:: READ_MASK))
         {
               //  无法注册handler
               return -1;
         }
 
          //  注册写事件
          if ( reactor()-> register_handler( this, ACE_Event_Handler:: WRITE_MASK))
         {
               //  无法注册handler
               return -1;
         }
 
          //  取消写事件，等待有数据时唤醒
          reactor()-> cancel_wakeup( this, ACE_Event_Handler:: WRITE_MASK);
 
          printf( "EchoService 已打开\n");
 
          return 0;
     }
 
      //  响应有数据可读事件
      int  handle_input( ACE_HANDLE)
     {
          char  buf[24];
          ssize_t  c =  _peer. recv( buf, sizeof( buf) - 1);
 
          if ( c == 0)
         {
               //  连接已经关闭
               return -1;
         }
 
          _bufs. push_back( std:: string( buf, c));
 
          if ( _bufs. size() == 1)
         {
               //  缓冲区尺寸为1 说明原来缓冲区为空，写事件是取消的，这里唤醒它
               reactor()-> schedule_wakeup( this, ACE_Event_Handler:: WRITE_MASK);
         }
 
          return 0;
     }
 
      //  响应可以发送数据了事件
      int  handle_output( ACE_HANDLE)
     {
          while (! _bufs. empty())
         {
               std:: string& buf(* _bufs. begin());
               char  const*       s( buf. c_str());
               char  const* const  e( s + buf. size());
               while ( s != e)
              {
                    ssize_t  c( _peer. send( s, e -  s));
                    if ( c == -1 || c == 0)
                   {
                        //  发送不成功不论发送过程中是否发生阻塞，
                        if ( ACE_OS:: last_error() == EWOULDBLOCK)
                       {
                             //  输出缓冲区满，无法再发送数据了（如果你还是继续发送数据，发送会阻塞的）
                             break;
                       } else{
                             //  连接已关闭
                             break;
                       }
                   } else{
                        s +=  c;
                   }
              }
               if ( s == e)
              {
                    _bufs. pop_front();
              } else{
                    buf =  std:: string( s, e -  s);
                    break;
              }
         }
          if ( _bufs. empty())
         {
               //  缓冲区为空，取消写事件监听
               reactor()-> cancel_wakeup( this, ACE_Event_Handler:: WRITE_MASK);
         }
          //  不论发送是否成功都返回0，因为，如果发送失败，handle_input 也会发生读失败事件，
          //  错误处理有handle_input 返回-1 来触发
          return 0;
     }
    
      int  handle_close( ACE_HANDLE =  ACE_INVALID_HANDLE,  ACE_Reactor_Mask  mask = ACE_Event_Handler:: ALL_EVENTS_MASK)
     {
          if ( mask == ACE_Event_Handler:: WRITE_MASK)
               return 0;
          _peer. close();
          delete  this;
          return 0;
     }
 
      //  这个函数主要给reactor::register_handler 时使用的
      ACE_HANDLE  get_handle ()  const
     {
          return  _peer. get_handle();
     }
 
      //  这个函数主要给acceptor 使用的
      stream_type&  peer()
     {
          return  _peer;
     }
 
      //  这个函数主要给acceptor 使用的
      int  close ( u_long = 0)
     {
          return  handle_close();
     }
 
private:
      stream_type  _peer;
      std:: list< std:: string> _bufs;
};
 
int  main( int  /*argc*/, char*  /*argv*/[])
{
      u_short  port = 20001;
      ACE_Reactor:: instance( new ACE_Reactor( new ACE_Select_Reactor,  true));
      EchoService:: acceptor_type acceptor;
      ACE_INET_Addr  svrAddr( port);
      if ( acceptor. open( svrAddr))
     {
          fprintf( stderr, " 服务打开失败：%s\n", ACE_OS:: strerror( ACE_OS:: last_error()));
          return 1;
     } else{
          fprintf( stdout, " 服务已打开，端口为：%u\n", port);
          ACE_Reactor:: instance()-> run_reactor_event_loop();
          return 0;
     }
}
关注
打赏
1659628745
查看更多评论

最近更新

热门博客

[ 申请 ]友情链接：

ClashX教程 绘画宝宝 配音宝宝
立即登录/注册

微信扫码登录

0.0487s