ACE_Select_Reactor 多线程通知机制分析

yunh

        ACE_Select_Reactor 是最为常用的一种 Reactor 实现，通常在单线程情况下使用，如果希望并发的处理连接上的事件，多使用 ACE_TP_Reactor，但这并不意味着 Select_Reactor 不可以在多线程环境中使用，当多个线程同时调用 Select_Reactor 的接口时，会发生什么情况，是这篇文章分析的主要内容。
        这里选取一个典型场景，主线程 M 负责检测并分派 Select_Reactor 上的 IO 事件，它主要调用 run_reactor_event_loop 接口，其中会使用 select 系统调用阻塞在被检测句柄上，直到超时(有定时器)或 IO 事件到达；子线程 N 负责在适当的时候改变 Select_Reactor 的状态，例如注册或反注册 IO 事件、定时器等，为简单起见，这里没有让线程 N 也调用 run_reactor_event_loop 运行事件循环，而只是调用一些改变状态的接口，如 register_handler/schedule_timer 等，实际例子只试验了调用定时器的情况，其它接口情况类似。
        首先上述场景是多线程安全的，不会因为线程 N 改变了 Select_Reactor 的状态，导致多线程竞争问题，这是由于 Select_Reactor 所有对外接口都加锁保护的原因。但这也引出了另一个问题，如果主线程在进入 select 后，陷入等待，而此时并没有释放锁，当线程 N 想要调用 schedule_timer 时，进入接口时获取锁必然也会陷入等待，直到主线程 M 从 select 调用中唤醒，并分派完所有事件，再次准备进入事件检测时，线程 N 才有机会获取锁，并进行更新。这样一分析，如果 Select_Reactor 没有任何事件(IO、定时器)到达，线程 N 岂不要一直等待下去? 这绝对是不可忍受的。
        写了一个小程序，想要验证下实际情况是否真的如此，这是一个简单的 Acceptor-Connector 实现的服务器，主线程等待连接到达、接收数据直到连接断开，启动一个单独的线程在 reactor 上不停的调度、取消定时器：

1.     while (thr_mgr ()->testcancel (ACE_Thread::self ()) == 0)
2.     {
3.         if(timer_ == 0)
4.         {
5.             timer_ = reactor ()->schedule_timer (this, 0, ACE_Time_Value(1, 0));
6.             ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) schedule timer %u.\n", timer_)));
7.         }
8.         else
9.         {
10.             reactor ()->cancel_timer (timer_);
11.             ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) cancel timer %u.\n", timer_)));
12.             timer_ = 0;
13.         }
15.         ACE_OS::sleep (3);
16.     }

复制代码

       根据上面的分析，如果在没有连接的情况下，子线程的所有对 Select_Reactor 接口的调用都将是被阻塞的，不会有任何输出，但实际情况却不是这样：

1. (7108) acceptor opened.
2. (7960) schedule timer 29490900.
3. (7108) handle timeout !
4. (7960) cancel timer 29490900.
5. (7960) schedule timer 29490900.
6. (7108) handle timeout !
7. (7960) cancel timer 29490900.

复制代码

       主线程 7108 打开侦听端口后，并没有连接连上来，也就没有任何 IO 事件，应当阻塞在 select 调用 (虽然没有任何句柄可侦听，但经过调试确实如此，Select_Reactor 有一些内部使用的句柄)，而此时子线程 7960 却可以成功的调用 schedule_timer！之后的 handle_timeout 是被主线程 7108 调度的，这可以理解，因为只有主线程在分派事件，但子线程 7960 是如何成功调用 schedule_timer 的呢? 按理说它必需在入口处等待锁呀：

1.   ACE_TRACE ("ACE_Select_Reactor_T::schedule_timer");
2.   ACE_MT (ACE_GUARD_RETURN (ACE_SELECT_REACTOR_TOKEN, ace_mon, this->token_, -1));
4.   return this->timer_queue_->schedule
5.     (handler,
6.      arg,
7.      timer_queue_->gettimeofday () + delay_time,
8.      interval);

复制代码

       看下锁的类型，才发现这里边有玄机，ACE_SELECT_REACTOR_TOKEN 不是一般的 ACE_Thread_Mutex 或 ACE_Recursive_Thread_Mutex，而是这个：

1. typedef ACE_Select_Reactor_Token_T<ACE_SELECT_TOKEN> ACE_Select_Reactor_Token;
2. typedef ACE_Select_Reactor_T<ACE_Select_Reactor_Token> ACE_Select_Reactor;

复制代码

       ACE_Select_Reactor 本身是个 typedef，是一个使用 ACE_Select_Reactor_Token 的模板，而后者又是一个typedef，是一个使用 ACE_SELECT_TOKEN 的模板，它的定义如下：

1. #if defined (ACE_MT_SAFE) && (ACE_MT_SAFE != 0)
2. typedef ACE_Token ACE_SELECT_TOKEN;
3. #else
4. typedef ACE_Noop_Token ACE_SELECT_TOKEN;
5. #endif /* ACE_MT_SAFE && ACE_MT_SAFE != 0 */

复制代码

       原来就是 ACE_Token，最终的锁类型是：

1. ACE_Select_Reactor_Token_T <ACE_Token>

复制代码

       经过一番探究，发现当线程 N 在此锁上调用 acquire 之前，它会调用锁的当前拥有者 (owner) 线程 M 之 sleep_hook 方法，而后者会向 Select_Reactor 发送一个通知，从而将主线程 M 从 select 调用中唤醒，在之后的事件分派过程中，当分派通知事件时，主线程会主动放弃拥有权，将自己加入锁的等待队列，并通知线程 N 已经获取了锁，线程 N 从 acquire 中醒来，更新 Select_Reactor 内部状态后 release 锁，再次唤醒等待的主线程 M，M 继续向下分派其它事件，从而完成一次线程安全的更新操作。这里由于代码过于复杂，就不具体罗列了。
        通过上面的分析可以得知，Select_Reactor 的这种线程合作机制有赖于 Notify 机制，如果我们将其关闭，上面的通知能力也就不复存在，为了验证这一点，在构造 Select_Reactor 时，传递第三个参数为 1，显示关闭通知机制：

1. ACE_NEW_RETURN(reactor, ACE_Reactor(new ACE_Select_Reactor(0, 0, 1), 1), -1);

复制代码

       重新编译工程，启动服务器端后，发现线程 N 的输出确实没有了：

1. (7756) acceptor opened.

复制代码

       如果此时使用一个客户端，连上服务器，由于 select 被 IO 事件唤醒，由线程 N 还可以继续运行并输出信息：

1. (7756) acceptor opened.
2. (7756) Connection from 127.0.0.1:57903
3. (6756) schedule timer 25427668.
4. (7756) recv message: Iteration 1.
5. (7756) handle timeout !
6. (7756) recv message: Iteration 2.
7. (6756) cancel timer 25427668.
8. (7756) recv message: Iteration 3.
9. (7756) recv message: Iteration 4.
10. (7756) recv message: Iteration 5.
11. (6756) schedule timer 25427668.
12. (7756) handle timeout !
13. (7756) Connection closed.
14. (2252|7756) handle_close
15. (6756) cancel timer 25427668.

复制代码

       事实上确实如此，线程 N 的输出被局限在有 IO 事件的时间段，一旦连接断开，它又被阻塞了。
        总结一下，ACE_Token  实现一种可重入锁，当新的线程在锁上等待时会通知当前 owner，而 ACE_Select_Reactor_Token_T 派生自它，重载了通知机制，用于向 Reactor 发送唤醒通知。如果子线程也运行 Select_Reactor 的事件循环，会怎么样? 如果 ACE 允许用户这样做，那么一个线程在进入 handle_events 时会通知另一个线程从 select 中离开，从而交替通知，无法正常工作。所以目前 ACE 禁止这样做，它会在 handle_events 中判断当前调用线程 ID，如果不是 open 时的线程，就直接返回 -1，用户可以通过 owner() 方法改变运行事件循环的线程为子线程，但主线程在下一次 handle_events 调用时会因为相同的原因退出，从而保证当前只有一个线程调用事件循环。
        如果用户在处理器回调(handle_input / handle_signal 等) 中调用 Reactor 接口修改其状态，又会如何呢? 答案是没问题，首先回调函数一般由侦测事件的线程来分派，所以这是同一个线程，不存在多线程竞争问题，其次 ACE_Token 本身是可重入锁，同一个线程在不释放它的情况下，可以再次取得它，只要最终释放的次数与取得的次数匹配即可。但 Select_Reactor 一旦在分派事件的过程中发现用户改变了自己的状态 (register/remove_handler)，它将停止继续分派剩余的事件，立即进入下一轮的事件检测，这是需要注意的地方。
        通过上面的分析，最后得到一个结论：如果你希望使用其它线程更新 Select_Reactor 的状态，最好不要禁用 Reactor 通知机制。
        本文的附件是测试程序，可以使用编译开关 DISABLE_NOTIFY 在两种情景下切换，查看运行效果。

freeeyes

如果Timerout事件里面有一个大的任务在处理，会导致select事件分派的性能问题。比如，你给你的timeout sleep 60秒。这时候，select会等待解锁。
所以我觉得，在设计上，定时器的reactor最好和实际负责IO事件分派的reactor完全独立分开，互不影响比较好。

yunh

        是的，这确实是 ACE_Select_Reactor 的最大问题，不光是在定时器，就是分派 IO 事件时，也不能在处理函数中堵塞，否则整个线程就挂死了。