ACE_Select_Reactor的Notify功能实现浅析

作者 : ydogg 如需转载 , 请注明 .

1. 前言

本文关注平台为Win32,但在其它的类Unix 平台，就实现 框架而言并没有太多变 化 ，惟一不同的是用于底层 机制的ACE_Pipe 的实现 。
但是，为了解决某些特别的原因，如大量的通知存储、规避悬空处理器指针等， ACE 也提供了一种有别于 Pipe 的解决方案，其采用消息排队的方式工作。当采取这种方式时，需定义 ACE_HAS_REACTOR_NOTIFICATION_QUEUE 宏并重生成 ACE 。

2. Notify的能力、风险和风险规避

Notify 机制最重要的能力是：
1. 让反应器拥有了处理无限处理器的能力
2. 其次是提供了必要时解除反应器事件检查阻塞的能力。

Reactor的notify()让用户直接提供给Reactor待通知反应器的指针，而这些处理器无需注册到反应器上，从而提供了无限的扩展能力。但是，在使用ACE_Pipe的实现中，如果使用不当，可能会造成严重的后果。

潜在的风险：
1. 处理器被销毁后，排队等候的对应通知才被分派
2. ACE_Pipe的数据缓冲是有限的，大量通知到来可能会造成阻塞甚至死锁

因为通知的信息（包括处理器指针和掩码）以流数据的方式被写入ACE_Pipe，不能进行遍历查找，所以当对应处理器被销毁后，如果其在CE_Pipe的数据区中还有存储有通知，则ACE_Reactor将会在分派中使用悬空指针，造成未知的后果。另外，在ACE_Pipe的数据缓冲已满情况下，在处理器的回调中依然发送通知，就会因阻塞发生死锁。

在采用ACE_Pipe的Notify实现中，以上两个问题的规避只能依靠用户自己的应用机制。

不过，在采用队列的Notify实现中，ACE 给提供了问题的解决方法。
1. 采用队列方式可无限扩展Notify 的数量
2. 采用队列方式为通知的查询提供的方便，因此ACE_Reactor提供puerge_pending_notifications()方法来移除指定的通知，当处理器销毁时，可以使用该方法移除当前队列中所有和自己相关的通知，避免了使用悬空指针。

在采用ACE_Pipe方式中，puerge_pending_notifications()被实现为空方法。

3. ACE_Select_Reactor类图

可以看出，ACE_Select_Reactor_T 是具体的实现负责 者。

4. 参与Notify机制的类

参与Notify 机制的类 有：

ACE_Select_Reactor_Impl 、ACE_Select_Reactor_Notify 和ACE_Pipe 。从ACE_Select_Reactor 开始，采用自上而向下的顺 序进 行分析。

5. 实现原理

各类 型Reactor 的实现 方法有所不同，比如Select_Reactor 是通过 管道来实现notify 功能，而WFMO则 通过ACE_Event类 型 的的事件加通知队列来实现 。

Reactor 在初始化时 ，会创 建 一个pipe 。然后将该pipe 的句柄注册到Reactor 。这样 ，这 个Pipe 就处 于Reactor 的事件分派句柄集中，像其他普通句柄一样 ，被Reactor 所监测 和分派。当用户调 用notify()时 ，notify() 会把参数信息写入这 个 管道，通过事件监测，从而Reactor 就能得知有新notify 到来，从管道读 取 相关数据后，就可以进 行 相应 的分派处 理 。从而实现 了通知和事件分派的序列化。

详细 的来说 ，ACE为 所有的Reactor 都定义 了专为 它们 自己服务 的Notify类 ，如

ACE_Select_Reactor_Notify 和ACE_WFO_Reactor_Notify 。

在类Unix 平台上，支持管道并且select 可用于监测 管道事件。但是在Win32 平台并不支持管道功能，而且Winsock 的select 只能支持对socket 句柄的事件监测 。因此，ACE 在Win32 平台用一个本地连 接 来模拟Pipe 的功能，其中接受连 接 的socket 句柄作为Pipe读 句柄，而发 起连 接 的socket 句柄作为Pipe 的写句柄。

可以看出，notify 功能的实现机制类似于ACE_Reactor 的标 准 用法，并 且也有一个类 似于ACE_Event_Handler 的类 存在，只不过 是被包含在Reactor 中。在逻辑 意义 上，就是一个本地连 接 ，用于和进 程 内的其他部分的通信 （略有差别的地方在于事件分派的方式） 。

对 于ACE_Select_Reactor,ACE_Pipe 充当了ACE_Sock_Stream 的角色，而ACE_Select_Reactor_Notify则 充当了ACE_Event_Handler 的角色。ACE_Select_Reactor_Notify 派生自ACE_Reactor_Notify,而ACE_Reactor_Notify 正是派生自ACE_Event_Handler.

6. 相关变量

ACE_Select_Reactor_Impl类 中，定义 了类 型为ACE_Reactor_Notify* 的成员变 量

/ CallbackobjectthatunblockstheACE_Select_Reactorifit's
/// sleeping.
ACE_Reactor_Notify * notify_handler_;

7. 实现分析

7.1 初始化

ACE_Select_Reactor_T 的open 函数定义 如下：

virtual int open(size_tmax_number_of_handles = DEFAULT_SIZE,
restart 0 ,
ACE_Sig_Handler * disable_notify_pipe ACE_DISABLE_NOTIFY_PIPE_DEFAULT,
ACE_Reactor_Notify );

其中ACE_DISABLE_NOTIFY_PIPE_DEFAULT的值为0,表示默认要使用Notify功能。
这个值会传递给ACE_Select_Reactor_Notify的构造函数。在open()的实现中：

if (result != - 1 && this -> notify_handler_ == 0 )
... {
ACE_NEW_RETURN(this->notify_handler_,
ACE_Select_Reactor_Notify,
-1);

if(notify_handler_==0)
result=;
else
delete_notify_handler_true;
}

handler_rep_.open(size) )
result = ;
else ( open( disable_notify_pipe) )
... {
ACE_ERROR((LM_ERROR,
ACE_TEXT("%p ),0);">notificationpipeopenfailed)));
result;
}

如果没有使用外部的notify时，ACE_Select_Reactor_T将会让notify_handler指向一个new出来的ACE_Select_Reactor_Notify类型的对象，并调用它的open()方法进行初始化。

ACE_Select_Reactor_Notify的open()实现如下（删除了非关键代码）：

(disable_notify_pipe )
{
select_reactor_
dynamic_cast < ACE_Select_Reactor_Impl *> (r);

(select_reactor_ )
{
errno EINVAL;
return ;
}

notification_pipe_.open() )
;

(ACE::set_flags( notification_pipe_.read_handle(),
ACE_NONBLOCK)
register_handler
(
{
;
}

注意，这里的disable_notify_pipe就是ACE_Select_Reactor_T的open()方法中传递的参数，通过它，可以设置不使用notify功能。
可以看出，ACE_Select_Reactor_Notify的open()的主要功能是初始化自己的ACE_Pipe类型成员对象notification_pipe_，并把它注册到Reactor中，监测read事件。

ACE_Pipe的open()实现如下（Win32平台并删除了非关键代码）：

ACE_INET_Addrmy_addr;
ACE_SOCK_Acceptoracceptor;
ACE_SOCK_Connectorconnector;
ACE_SOCK_Streamreader;
ACE_SOCK_Streamwriter;
int result ;
# defined(ACE_WIN32)
ACE_INET_Addrlocal_any(static_cast u_short > #endif
(acceptor.open(local_any) || acceptor.get_local_addr(my_addr)
... {
ACE_INET_Addrsv_addr(my_addr.get_port_number(),
ACE_LOCALHOST);

//Establishaconnectionwithinthesameprocess.(connector.connect(writer,sv_addr)(acceptor.accept(reader))
...{
writer.close();
result;
}
}

acceptor.close();
handles_[ ] reader.get_handle();
writer.get_handle();