ACE_Reactor是如何做到事件分发的

1. ACE_Reactor的创建

ctor: ACE_Reactor (ACE_Reactor_Impl *implementation = 0, int delete_implementation = 0);

你可以自己创建一个ACE_Reactor

但是大多数时候，我们都是通过调用ACE_Reactor::instance()这个静态方法来返回唯一的实例。

ACE_Reactor *

ACE_Reactor::instance (void)

{

ACE_TRACE ("ACE_Reactor::instance");

if (ACE_Reactor::reactor_ == 0) //1

{

// Perform Double-Checked Locking Optimization.

ACE_MT (ACE_GUARD_RETURN (ACE_Recursive_Thread_Mutex, ace_mon,

*ACE_Static_Object_Lock::instance (), 0));

if (ACE_Reactor::reactor_ == 0) //2

{

ACE_NEW_RETURN (ACE_Reactor::reactor_,

ACE_Reactor,

0);

ACE_Reactor::delete_reactor_ = 1;

ACE_REGISTER_FRAMEWORK_COMPONENT(ACE_Reactor, ACE_Reactor::reactor_)

}

}

return ACE_Reactor::reactor_;

}
注意这点使用了双检测的机制(代码1和2)，为了提高效率，不用每次都加锁。

2. Impl手法的运用

ACE_Reactor有个成员变量 ACE_Reactor_Impl *implementation_;

这个implementation_才是真正做事情的东西，典型的Impl手法。

为什么要多一个这个间隔层呢，主要是为了实现跨平台。

因为不同的平台的Reactor差异很大。

在Windows平台，实现这个是ACE_WFMO_Reactor

class ACE_Export ACE_WFMO_Reactor : public ACE_Reactor_Impl

3. Event_Handle的管理

ACE_WFMO_Reactor把要管理的Handle都放在 ACE_WFMO_Reactor_Handler_Repository handler_rep_;

这里要注意的是io_handle和event_handle的区别

io_handle是真正的handle，比如socket_handle, thread_handle

而event_handle是绑定在io_handle上面的事件handle

有代码为证:

1 int

2 ACE_WFMO_Reactor::register_handler_i (ACE_HANDLE event_handle,

3 ACE_HANDLE io_handle,

4 ACE_Event_Handler *event_handler,

5 ACE_Reactor_Mask new_masks)

6 {

7 // If this is a Winsock 1 system, the underlying event assignment will

8 // not work, so don't try. Winsock 1 must use ACE_Select_Reactor for

9 // reacting to socket activity.

10

11 // Make sure that the <handle> is valid

12 if (io_handle == ACE_INVALID_HANDLE)

13 io_handle = event_handler->get_handle ();

14

15 if (this->handler_rep_.invalid_handle (io_handle))

16 {

17 errno = ERROR_INVALID_HANDLE;

18 return -1;

19 }

20

21 long new_network_events = 0;

22 int delete_event = 0;

23 auto_ptr <ACE_Auto_Event> event;

24

25 // Look up the repository to see if the <event_handler> is already

26 // there.

27 ACE_Reactor_Mask old_masks;

28 int found = this->handler_rep_.modify_network_events_i (io_handle,

29 new_masks,

30 old_masks,

31 new_network_events,

32 event_handle,

33 delete_event,

34 ACE_Reactor::ADD_MASK);

35

36 // Check to see if the user passed us a valid event; If not then we

37 // need to create one

38 if (event_handle == ACE_INVALID_HANDLE)

39 {

40 // Note: don't change this since some C++ compilers have

41 // <auto_ptr>s that don't work properly


42 auto_ptr<ACE_Auto_Event> tmp (new ACE_Auto_Event);

43 event = tmp;

44 event_handle = event->handle ();

45 delete_event = 1;

46 }

47

48 int result = ::WSAEventSelect ((SOCKET) io_handle,

49 event_handle,

50 new_network_events);

可以看到在42行create event,在44复制到event_handle,最后通过WSAEventSelect将这个io_handle和event_handle绑定在一起了

这个register_handle调用一般都在Event_Handler的open函数中，实现了注册到Reactor中去。

4. 等待事件

如何检测到要发生的socket的事件呢？比如有新的client连接，收发。ACE并不直接调用select函数来取得。

ACE调用WaitForMultipleObjectsEx来等待事件的发生。ACE这样做的好处是不但可以捕捉socket事件，也可以捕捉到其他事件。前面说过了每一个对应的socket都有一个event_handle与之对应。WaitForMultipleObjectsEx会将发生事件的socket handle的index返回。这样ACE_Reactor就可以利用这个slot来查到io_handle和event_handler( 注意:event_handle和event_handler是不同的)

1 DWORD

2 ACE_WFMO_Reactor::wait_for_multiple_events (int timeout,

3 int alertable)

4 {

5 // Wait for any of handles_ to be active, or until timeout expires.

6 // If <alertable> is enabled allow asynchronous completion of

7 // ReadFile and WriteFile operations.

8

9 return ::WaitForMultipleObjectsEx (this->handler_rep_.max_handlep1 (),

10 this->handler_rep_.handles (),

11 FALSE,

12 timeout,

13 alertable);

14 }

5.分发事件

根据WaitForMultiObjectEx返回的slot就可以查询到event_handler来调用用户的处理函数了

1 int

2 ACE_WFMO_Reactor::complex_dispatch_handler (DWORD slot,

3 ACE_HANDLE event_handle)

4 {

5 // This dispatch is used for I/O entires.

6

7 ACE_WFMO_Reactor_Handler_Repository::Current_Info ¤t_info =

8 this->handler_rep_.current_info ()[slot];

9

10 WSANETWORKEVENTS events;

11 ACE_Reactor_Mask problems = ACE_Event_Handler::NULL_MASK;

12 if (::WSAEnumNetworkEvents ((SOCKET) current_info.io_handle_,

13 event_handle,

14 &events) == SOCKET_ERROR)

15 problems = ACE_Event_Handler::ALL_EVENTS_MASK;

16 else

17 {

18 // Prepare for upcalls. Clear the bits from <events> representing

19 // events the handler is not interested in. If there are any left,

20 // do the upcall(s). upcall will replace events.lNetworkEvents

21 // with bits representing any functions that requested a repeat

22 // callback before checking handles again. In this case, continue

23 // to call back unless the handler is unregistered as a result of

24 // one of the upcalls. The way this is written, the upcalls will

25 // keep being done even if one or more upcalls reported problems.

26 // In practice this may turn out not so good, but let's see. If any

27 // problems, please notify Steve Huston <shuston@riverace.com>

28 // before or after you change this code.

29 events.lNetworkEvents &= current_info.network_events_;

30 while (events.lNetworkEvents != 0)

31 {

32 ACE_Event_Handler *event_handler =

33 current_info.event_handler_;

34

35 int reference_counting_required =

36 event_handler->reference_counting_policy ().value () ==

37 ACE_Event_Handler::Reference_Counting_Policy::ENABLED;

38

39 // Call add_reference() if needed.

40 if (reference_counting_required)

41 {

42 event_handler->add_reference ();

43 }

44

45 // Upcall

46 problems |= this->upcall (current_info.event_handler_,

47 current_info.io_handle_,

48 events);

49

50 // Call remove_reference() if needed.

51 if (reference_counting_required)

52 {

53 event_handler->remove_reference ();

54 }

55

56 if (this->handler_rep_.scheduled_for_deletion (slot))

57 break;

58 }

59 }

60

61 if (problems != ACE_Event_Handler::NULL_MASK

62 && !this->handler_rep_.scheduled_for_deletion (slot) )

63 this->handler_rep_.unbind (event_handle, problems);

64

65 return 0;

66 }
这里值得注意的是ACE通过调用WSAEnumNetworkEvents来重置event_handle。