用ACE Reactor框架编程

peakzhang

用ACE Reactor框架编程

高歌

按照我的理解，我大致讲一下windows平台下的原理。在windows平台下ACE_Reactor的实现类是ACE_WMFO_Reactor，事件的多路分离是通过WaitForMultiObject和WSAEventSelect着两个函数来实现的。WaitForMultiObject的功能是等待一组(64)handle中的一个或者所有被操作系统激活，程序返回，具体使用看msdn，WSAEventSelect的功能是指定一个事件对象event object与 某个SOCKET的FD_XXX网络事件关联。ACE_WMFO_Reactor对象构造成功后，我们调用run_event_loop()，事件的多路分离线程就开始了， int
ACE_Reactor::run_reactor_event_loop (REACTOR_EVENT_HOOK eh)
{
　　ACE_TRACE ("ACE_Reactor::run_reactor_event_loop");
　　if (this->reactor_event_loop_done ())
　　　　return 0;
　　while (1)
　　{
　　　　int result = this->implementation_->handle_events ();//循环调用实现类的handle_events;
　　　　if (eh != 0 && (*eh)(this))
　　　　　　continue;
　　　　else if (result == -1 && this->implementation_->deactivated ())
　　　　　　return 0;
　　　　else if (result == -1)
　　　　return -1;
　　}
　　ACE_NOTREACHED (return 0;)
}
下面是handle_events调用过程
ACE_WFMO_Reactor::handle_events (ACE_Time_Value *how_long)
-ACE_WFMO_Reactor::event_handling (ACE_Time_Value *max_wait_time,int alertable)
--ACE_WFMO_Reactor::ok_to_wait (ACE_Time_Value *max_wait_time,int alertable)
--ACE_WFMO_Reactor::ok_to_wait (ACE_Time_Value *max_wait_time,int alertable)
--ACE_WFMO_Reactor::calculate_timeout (ACE_Time_Value *max_wait_time)
--ACE_WFMO_Reactor::calculate_timeout (ACE_Time_Value *max_wait_time)
--ACE_WFMO_Reactor::wait_for_multiple_events (int timeout,int alertable)
上面那个函数就是关键的多路分离函数。Reactor是如何知道应该调用那个事件处理器呢(event_handler)，请看下面那个函数，
ACE_WFMO_Reactor::register_handler (ACE_Event_Handler *event_handler,ACE_Reactor_Mask mask)这是一个简单的事件处理器的注册函数，没有提供IO句柄，因为我们可以从ACE_Event_Handler派生一个类，,并在这个类中重载get_handle()这个函数提供IO句柄。如下
class Server_Acceptor : public ACE_Event_Handler
{
public:
　　virtual int open (const ACE_INET_Addr &local_addr);
　　virtual ACE_HANDLE get_handle (void) const {return acceptor_.get_handle (); };
　　virtual int handle_close (ACE_HANDLE = ACE_INVALID_HANDLE,ACE_Reactor_Mask = 0);
　　virtual int handle_input (ACE_HANDLE = ACE_INVALID_HANDLE);

protected:
　　ACE_SOCK_Acceptor acceptor_;
};
这个注册过程最终是通过register_handler_i这个成员函数进行的，
int
ACE_WFMO_Reactor::register_handler_i (ACE_HANDLE event_handle,
ACE_HANDLE io_handle,
ACE_Event_Handler *event_handler,
ACE_Reactor_Mask new_masks)
{
　　// If this is a Winsock 1 system, the underlying event assignment will
　　// not work, so don't try. Winsock 1 must use ACE_Select_Reactor for
　　// reacting to socket activity.
#if !defined (ACE_HAS_WINSOCK2) || (ACE_HAS_WINSOCK2 == 0)
ACE_UNUSED_ARG (event_handle);
ACE_UNUSED_ARG (io_handle);
ACE_UNUSED_ARG (event_handler);
ACE_UNUSED_ARG (new_masks);
ACE_NOTSUP_RETURN (-1);
#else
　　// Make sure that the <handle> is valid
　　if (io_handle == ACE_INVALID_HANDLE)
　　io_handle = event_handler->get_handle ();
　　if (this->handler_rep_.invalid_handle (io_handle))
　　{
　　　　errno = ERROR_INVALID_HANDLE;
　　　　return -1;
　　}
　　long new_network_events = 0;
　　int delete_event = 0;
　　auto_ptr <ACE_Auto_Event> event;
　　// Look up the repository to see if the <event_handler> is already
　　// there.
　　ACE_Reactor_Mask old_masks;
　　int found = this->handler_rep_.modify_network_events_i (io_handle,
　　new_masks,
　　old_masks,
　　new_network_events,
　　event_handle,
　　delete_event,
　　ACE_Reactor::ADD_MASK);
　　// Check to see if the user passed us a valid event; If not then we
　　// need to create one
　　if (event_handle == ACE_INVALID_HANDLE)
　　{
　　　　// Note: don't change this since some C++ compilers have
　　　　// <auto_ptr>s that don't work properly...
　　　　auto_ptr<ACE_Auto_Event> tmp (new ACE_Auto_Event);
　　　　event = tmp;
　　　　event_handle = event->handle ();
　　　　delete_event = 1;
　　}
　　int result = ::WSAEventSelect ((SOCKET) io_handle,
　　event_handle,
　　new_network_events);
　　// If we had found the <Event_Handler> there is nothing more to do
　　if (found)
　　　　return result;
　　else if (result != SOCKET_ERROR &&
　　this->handler_rep_.bind_i (1,
　　　　event_handler,
　　　　new_network_events,
　　　　io_handle,
　　　　event_handle,
　　　　delete_event) != -1)
　　{
　　　　// The <event_handler> was not found in the repository, add to
　　　　// the repository.
　　　　if (delete_event)
　　　　{
　　　　　　　// Clear out the handle in the ACE_Auto_Event so that when
　　　　　　　// it is destroyed, the handle isn't closed out from under
　　　　　　　// the reactor. After setting it, running down the event
　　　　　　　// (via auto_ptr<> event, above) at function return will
　　　　　　　// cause an error because it'll try to close an invalid handle.
　　　　　　　// To avoid that smashing the errno value, save the errno
　　　　　　　// here, explicitly remove the event so the dtor won't do it
　　　　　　　// again, then restore errno.
　　　　　　　ACE_Errno_Guard guard (errno);
　　　　　　　event->handle (ACE_INVALID_HANDLE);
　　　　　　　event->remove ();
　　　　}
　　　　return 0;
　　}
　　else
　　　　return -1;
#endif /* ACE_HAS_WINSOCK2 || ACE_HAS_WINSOCK2 == 0 */
}
这个函数先调用WSAEventSelect进行Event_Handler的注册，然后调用handler_rep_.bind_i来修改句柄相关的信息。handler_rep是ACE_WFMO_Reactor的一个句柄仓库。
在使用ACE_WFMO_Reactor时也遇到个这么几个问题，
如何安全的结束run_event_loop线程？
定义入下一个类
class Quit_Handler : public ACE_Event_Handler {
friend class ace_dewarn_gplusplus;
public:
Quit_Handler (ACE_Reactor *r) : ACE_Event_Handler (r) {}
virtual int handle_exception (ACE_HANDLE) {
reactor ()->end_reactor_event_loop ();
return -1; // Trigger call to handle_close() method.
}
virtual int handle_close (ACE_HANDLE, ACE_Reactor_Mask)
{ delete this; return 0; }
private:
// Private destructor ensures dynamic allocation.
virtual ~Quit_Handler () {}
};
然后调用
Quit_Handler *quit_handler_;
ACE_Reactor *r = ACE_Reactor::instance();
quit_handler_ = new Quit_Handler (r);
r->notify (quit_handler_);
如何安全的删除事件处理器？
ACE_Reactor::instance()->remove_handler(acceptor_.get_handle(),ACE_Event_Handler::READ_MASK);
不要直接调用handle_close
如何安全的结束工作者，消费者线程？
如果线程使用同步队列的话，可以插入以下消息，结束线程
ACE_Message_Block *shutdown_message = 0;
ACE_NEW_RETURN
(shutdown_message,
ACE_Message_Block (0, ACE_Message_Block::MB_STOP), -1);
if( my_card_queue.enqueue_tail (shutdown_message) == -1)
shutdown_message->release();
MFC中如何使用ACE编程？
在程序开始运行是调用ACE::init()，结束前调用ACE:fini()。
Reactor存在的问题是WaitForMultiObject只能 传入64个局柄，这个问题会在前摄式框架中解决。