ACE - Reactor源码总结整理
ACE源码约10万行，是c++中非常大的一个网络编程代码库，包含了网络编程的边边角角。
ACE代码可以分三个层次：OS层、OO层和框架层：
- OS层主要是为了兼容各个平台，将网络底层API统一化，这一层用户不关心。
- OO层则是对一些常用的数据结构或方法进行OO封装，方便上层使用，包括socket方法，进程、线程和他们的同步机制等。
- 框架层实现了一些优秀的网络框架。
1 ACE_INET_Addr
类名 | ACE_INET_Addr |
|---|---|
所在文件 | INET_Addr.h |
功能 | 处理和存储IP、端口、协议类型的数据类，提供了一组操作这些数据的方法。 |
关键数据 | 1 /// Creates an ACE_INET_Addr from a @a port_number and ACE_INET_Addr (u_short port_number, 2 3 const char host_name[], 4 5 int address_family = AF_UNSPEC); 6 7 union 8 9 { 10 11 sockaddr_in in4_; 12 13 #if defined (ACE_HAS_IPV6) 14 15 sockaddr_in6 in6_; 16 17 #endif /* ACE_HAS_IPV6 */ 18 19 } inet_addr_; |
在构造时入参端口和IP地址完成赋值。
#define LISTEN_PORT 5010
#define SERVER_IP ACE_LOCALHOST
ACE_INET_Addr Server_Info(LISTEN_PORT,(char *)SERVER_IP);
2 ACE_SOCK_Acceptor
类名 | ACE_SOCK_Acceptor |
|---|---|
所在文件 | SOCK_Acceptor.h |
功能 | 接受器，被动建立连接，用于Socket Server。 |
关键数据 | 1 /** 2 3 * Initialize a passive-mode BSD-style acceptor socket (no QoS). 4 5 */ 6 7 int open (const ACE_Addr &local_sap, 8 9 int reuse_addr = 0, 10 11 int protocol_family = PF_UNSPEC, 12 13 int backlog = ACE_DEFAULT_BACKLOG, 14 15 int protocol = 0); 16 17 /** 18 19 * Accept a new ACE_SOCK_Stream connection. A @a timeout of 0 20 21 */ 22 23 int accept (ACE_SOCK_Stream &new_stream, 24 25 ACE_Addr *remote_addr = 0, 26 27 ACE_Time_Value *timeout = 0, 28 29 bool restart = true, 30 31 bool reset_new_handle = false) const; |
Open：创建socket，打开监听端口，绑定socket。完成socket、listen、bind操作。
文件：SOCK.cpp 代码行：141 - 146 |
|---|
1 this->set_handle (ACE_OS::socket (protocol_family, 2 3 type, 4 5 protocol, 6 7 protocolinfo, 8 9 g, 10 11 flags)); |
文件：SOCK_Acceptor.cpp 代码行：279 - 282 |
|---|
1 1 else if (ACE_OS::bind (this->get_handle (), 2 2 3 3 (sockaddr *) local_sap.get_addr (), 4 4 5 5 local_sap.get_size ()) == -1) 6 6 7 7 error = 1; |
文件：SOCK_Acceptor.cpp 代码行：284 - 291 |
|---|
1 if (error != 0 2 3 || ACE_OS::listen (this->get_handle (), 4 5 backlog) == -1) 6 7 { 8 9 ACE_Errno_Guard g (errno); // Preserve across close() below. 10 11 error = 1; 12 13 this->close (); 14 15 } |
Accept：创建数据通道。完成accept操作。
文件：SOCK_Acceptor.cpp 代码行：134 - 136 |
|---|
1 new_stream.set_handle (ACE_OS::accept (this->get_handle (), 2 3 addr, 4 5 len_ptr)); |
3 ACE_SOCK_Stream
类名 | ACE_SOCK_Stream |
|---|---|
所在文件 | SOCK_Stream.h |
功能 | 传输数据的流,用于传输数据。 |
关键数据 | 1 /// Recv @a n bytes via Win32 @c ReadFile using overlapped I/O. 2 3 ssize_t recv (void *buf, 4 5 size_t n, 6 7 ACE_OVERLAPPED *overlapped) const; 8 9 /// Send an @a n byte buffer to the connected socket. 10 11 ssize_t send (const void *buf, 12 13 size_t n, 14 15 int flags, 16 17 const ACE_Time_Value *timeout = 0) const; |
ACE_SOCK_Stream在ACE_SOCK_Acceptor执行accept时作为参数传入，accept把返回的IOhandle赋值给ACE_SOCK_Stream，ACE_SOCK_Stream利用IO进行recv和send数据收发。
Recv：接收accept连接上来的IO数据。
文件：SOCK_IO.cpp 代码行：143 - 145 |
|---|
1 ssize_t const result = ACE_OS::recvv (this->get_handle (), 2 3 iovp, 4 5 total_tuples); |
Send：向IO发送数据。
文件：SOCK_IO.cpp 代码行：103 - 105 |
|---|
1 ssize_t const result = ACE_OS::sendv (this->get_handle (), 2 3 iovp, 4 5 total_tuples); |
4 ACE_Task
类名 | ACE_Task |
|---|---|
所在文件 | Task_T.h |
功能 | 提供线程实例，也可以创建一组线程。包含一个消息队列。 |
关键数据 | 1 /// Insert message into the message queue. Note that @a timeout uses 2 3 int putq (ACE_Message_Block *, ACE_Time_Value *timeout = 0); 4 5 /** 6 7 * Extract the first message from the queue (blocking). 8 9 */ 10 11 int getq (ACE_Message_Block *&mb, ACE_Time_Value *timeout = 0); 12 13 // = Active object activation method. 14 15 virtual int activate (long flags = THR_NEW_LWP | THR_JOINABLE | THR_INHERIT_SCHED, 16 17 int n_threads = 1, 18 19 int force_active = 0, 20 21 long priority = ACE_DEFAULT_THREAD_PRIORITY, 22 23 int grp_id = -1, 24 25 ACE_Task_Base *task = 0, 26 27 ACE_hthread_t thread_handles[] = 0, 28 29 void *stack[] = 0, 30 31 size_t stack_size[] = 0, 32 33 ACE_thread_t thread_ids[] = 0, 34 35 const char* thr_name[] = 0); 36 37 /// Run by a daemon thread to handle deferred processing. 38 39 virtual int svc (void); 40 41 /// Queue of messages on the ACE_Task.. 42 43 ACE_Message_Queue<ACE_SYNCH_USE> *msg_queue_; |
Putq：将消息插入队列，ACE_Message_Queue<ACE_SYNCH_USE> *msg_queue_是ACE实现的一个消息队列。提供队列的存储和方法。
文件：Task_T.inl 代码行：29 |
|---|
1 this->msg_queue_->enqueue_tail (mb, tv); |
Getq：从消息队列中取一条数据记录。
文件：Task_T.inl 代码行：22 |
|---|
1 return this->msg_queue_->dequeue_head (mb, tv); |
Activate：创建并激活线程。
文件：Task.cpp 代码行：161 - 172 |
|---|
1 grp_spawned = 2 3 this->thr_mgr_->spawn_n (n_threads, 4 5 &ACE_Task_Base::svc_run, 6 7 (void *) this, 8 9 flags, 10 11 priority, 12 13 grp_id, 14 15 task, 16 17 thread_handles, 18 19 stack, 20 21 stack_size, 22 23 thr_name); |
Svc：线程函数，需要继承类重写，创建时调用svc_run，入参为this当前类，以面向对象形式封装线程。
文件：Task.cpp 代码行：270 - 271 |
|---|
1 // Call the Task's svc() hook method. 2 3 int const svc_status = t->svc (); |
5 ACE_Event_Handler
类名 | ACE_Event_Handler |
|---|---|
所在文件 | Event_Handler.h |
功能 | 事件接收器，与I/O绑定，当I/O产生不同事件时，执行ACE_Event_Handler的不同方法。 |
关键数据 | 1 /// Get the I/O handle. 2 3 virtual ACE_HANDLE get_handle (void) const; 4 5 /// Set the I/O handle. 6 7 virtual void set_handle (ACE_HANDLE); 8 9 /// Called when input events occur (e.g., connection or data). 10 11 virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE); 12 13 /// Called when output events are possible (e.g., when flow control 14 15 /// abates or non-blocking connection completes). 16 17 virtual int handle_output (ACE_HANDLE fd = ACE_INVALID_HANDLE); |
get_handle、set_handle：继承重写或调用它们去设置和获取I/O。
文件：Event_Handler.h 代码行：91 – 94 |
|---|
1 virtual ACE_HANDLE get_handle (void) const; 2 3 virtual void set_handle (ACE_HANDLE); |
handle_input、handle_output：继承重写它们，当产生相应的I/O事件时将被执行。
文件：Event_Handler.h 代码行：91 – 94 |
|---|
1 virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE); 2 3 virtual int handle_output (ACE_HANDLE fd = ACE_INVALID_HANDLE); |
6 ACE_Reactor
类名 | ACE_Reactor |
|---|---|
所在文件 | Reactor.h |
功能 | Reactor框架，提供IO复用、消息循环和消息分发。 |
关键数据 | 1 int register_handler (ACE_Event_Handler *event_handler, 2 3 ACE_HANDLE event_handle = ACE_INVALID_HANDLE); 4 5 int remove_handler (ACE_HANDLE handle, 6 7 ACE_Reactor_Mask masks); 8 9 int run_reactor_event_loop (REACTOR_EVENT_HOOK = 0); |
Reactor是I/O多路复用模式的一种，ACE提供了Reactor框架。
主要参与Reactor的对象有：
l 事件I/O，ACE_SOCK_Acceptor、ACE_SOCK_Stream的get_handle均提供对应I/O。
l 注册事件，继承于ACE_Event_Handler的对象。对象绑定了I/O，当产生I/O事件后将调用对象相对应的方法。
ACE_Reactor内部调用复杂，以下提供最后执行的关键代码。
register_handler：注册事件到Reactor，需要传入ACE_Event_Handler的派生类，即指定触发事件后将执行的函数，以及传入将被复用监听的I/O，I/O可以与ACE_Event_Handler绑定，也可以单独传入。Reactor将这两个信息绑定在handler_rep_容器内，并且将I/O以ADD形式添加到select_reactor_.wait_set_。
文件：Select_Reactor_T.cpp Select_Reactor_Base.cpp Select_Reactor_Base.cpp 代码行：993 249 265-268 |
|---|
1 return this->handler_rep_.bind (handle, event_handler, mask); |
1 this->event_handlers_[handle]= event_handler; |
1 this->select_reactor_.bit_ops (handle, 2 3 mask, 4 5 this->select_reactor_.wait_set_, 6 7 ACE_Reactor::ADD_MASK); |
run_reactor_event_loop：消息循环主要进行两个操作，1.监听注册的I/O。2.当I/O产生事件后通知到绑定的ACE_Event_Handler具体执行函数。在监听部分使用select监听注册进去的I/O select_reactor_.wait_set_。通知时发送到指定的handle_input，handle_output等相应函数。
文件：Select_Reactor_T.cpp Select_Reactor_T.cpp Select_Reactor_Base.cpp 代码行： 1435 - 1441 1093 – 1100 737 - 746 |
|---|
1 int number_of_active_handles = 2 3 this->wait_for_multiple_events (this->dispatch_set_, 4 5 max_wait_time); 6 7 result = 8 9 this->dispatch (number_of_active_handles, 10 11 this->dispatch_set_); |
1 dispatch_set.rd_mask_ = this->wait_set_.rd_mask_; 2 3 dispatch_set.wr_mask_ = this->wait_set_.wr_mask_; 4 5 dispatch_set.ex_mask_ = this->wait_set_.ex_mask_; 6 7 number_of_active_handles = ACE_OS::select (width, 8 9 dispatch_set.rd_mask_, 10 11 dispatch_set.wr_mask_, 12 13 dispatch_set.ex_mask_, 14 15 this_timeout); |
1 ACE_HANDLE const read_handle = 2 3 this->notification_pipe_.read_handle (); 4 5 if (read_handle != ACE_INVALID_HANDLE 6 7 && rd_mask.is_set (read_handle)) 8 9 { 10 11 --number_of_active_handles; 12 13 rd_mask.clr_bit (read_handle); 14 15 return this->handle_input (read_handle); 16 17 } |