Specialization of ACE_Select_Reactor to support thread-pool based event dispatching. More...

#include <TP_Reactor.h>

Inheritance diagram for ACE_TP_Reactor:

Collaboration diagram for ACE_TP_Reactor:

Public Member Functions
	ACE_TP_Reactor (ACE_Sig_Handler =0, ACE_Timer_Queue =0, bool mask_signals=true, int s_queue=ACE_Select_Reactor_Token::FIFO)
	Initialize ACE_TP_Reactor with the default size.
	ACE_TP_Reactor (size_t max_number_of_handles, bool restart=false, ACE_Sig_Handler sh=0, ACE_Timer_Queue tq=0, bool mask_signals=true, int s_queue=ACE_Select_Reactor_Token::FIFO)
virtual int	handle_events (ACE_Time_Value *max_wait_time=0)
virtual int	handle_events (ACE_Time_Value &max_wait_time)
virtual int	resumable_handler (void)
virtual int	owner (ACE_thread_t n_id, ACE_thread_t *o_id=0)
virtual int	owner (ACE_thread_t *t_id)
	Return the thread ID of the current Leader.
Static Public Member Functions
static void	no_op_sleep_hook (void *)
	Called from handle events.
Public Attributes
	ACE_ALLOC_HOOK_DECLARE
	Declare the dynamic allocation hooks.
Protected Member Functions
virtual void	clear_dispatch_mask (ACE_HANDLE handle, ACE_Reactor_Mask mask)
	Template method from the base class.
int	dispatch_i (ACE_Time_Value *max_wait_time, ACE_TP_Token_Guard &guard)
	Dispatch just 1 signal, timer, notification handlers.
int	get_event_for_dispatching (ACE_Time_Value *max_wait_time)
int	handle_timer_events (int &event_count, ACE_TP_Token_Guard &g)
	Handle timer events.
int	handle_notify_events (int &event_count, ACE_TP_Token_Guard &g)
	Handle notify events.
int	handle_socket_events (int &event_count, ACE_TP_Token_Guard &g)
	handle socket events
virtual void	notify_handle (ACE_HANDLE handle, ACE_Reactor_Mask mask, ACE_Handle_Set &, ACE_Event_Handler *eh, ACE_EH_PTMF callback)
	This method shouldn't get called.
Private Member Functions
ACE_HANDLE	get_notify_handle (void)
int	get_socket_event_info (ACE_EH_Dispatch_Info &info)
	Get socket event dispatch information.
int	dispatch_socket_event (ACE_EH_Dispatch_Info &dispatch_info)
void	clear_handle_read_set (ACE_HANDLE handle)
	Clear the handle from the read_set.
int	post_process_socket_event (ACE_EH_Dispatch_Info &dispatch_info, int status)
	ACE_TP_Reactor (const ACE_TP_Reactor &)
	Deny access since member-wise won't work...
ACE_TP_Reactor &	operator= (const ACE_TP_Reactor &)

Detailed Description

Specialization of ACE_Select_Reactor to support thread-pool based event dispatching.

One of the shortcomings of the ACE_Select_Reactor is that it does not support a thread pool-based event dispatching model, similar to the one in ACE_WFMO_Reactor. In ACE_Select_Reactor, only thread can call handle_events() at any given time. ACE_TP_Reactor removes this short-coming.

ACE_TP_Reactor is a specialization of ACE_Select_Reactor to support thread pool-based event dispatching. This reactor takes advantage of the fact that events reported by select() are persistent if not acted upon immediately. It works by remembering the event handler which was just activated, suspending it for further I/O activities, releasing the internal lock (so that another thread can start waiting in the event loop) and then dispatching the event's handler outside the scope of the reactor lock. After the event handler has been dispatched the event handler is resumed for further I/O activity.

This reactor implementation is best suited for situations when the callbacks to event handlers can take arbitrarily long and/or a number of threads are available to run the event loop. Note that I/O-processing callback code in event handlers (e.g. handle_input()) does not have to be modified or made thread-safe for this reactor. This is because before an I/O event is dispatched to an event handler, the handler is suspended; it is resumed by the reactor after the upcall completes. Therefore, multiple I/O events will not be made to one event handler multiple threads simultaneously. This suspend/resume protection does not apply to either timers scheduled with the reactor or to notifications requested via the reactor. When using timers and/or notifications you must provide proper protection for your class in the context of multiple threads.

Definition at line 176 of file TP_Reactor.h.

Constructor & Destructor Documentation

ACE_TP_Reactor::ACE_TP_Reactor	(	ACE_Sig_Handler *	sh = `0`,
		ACE_Timer_Queue *	tq = `0`,
		bool	mask_signals = `true`,
		int	s_queue = `ACE_Select_Reactor_Token::FIFO`
	)

Initialize ACE_TP_Reactor with the default size.

Definition at line 101 of file TP_Reactor.cpp.

  : ACE_Select_Reactor (sh, tq, ACE_DISABLE_NOTIFY_PIPE_DEFAULT, 0, mask_signals, s_queue)
{
  ACE_TRACE ("ACE_TP_Reactor::ACE_TP_Reactor");
  this->supress_notify_renew (1);
}

ACE_TP_Reactor::ACE_TP_Reactor	(	size_t	max_number_of_handles,
		bool	restart = `false`,
		ACE_Sig_Handler *	sh = `0`,
		ACE_Timer_Queue *	tq = `0`,
		bool	mask_signals = `true`,
		int	s_queue = `ACE_Select_Reactor_Token::FIFO`
	)

Initialize the ACE_TP_Reactor to manage max_number_of_handles. If restart is non-0 then the ACE_Reactor's handle_events() method will be restarted automatically when EINTR occurs. If sh or tq are non-0 they are used as the signal handler and timer queue, respectively.

Definition at line 111 of file TP_Reactor.cpp.

  : ACE_Select_Reactor (max_number_of_handles, restart, sh, tq, ACE_DISABLE_NOTIFY_PIPE_DEFAULT, 0, mask_signals, s_queue)
{
  ACE_TRACE ("ACE_TP_Reactor::ACE_TP_Reactor");
  this->supress_notify_renew (1);
}

ACE_TP_Reactor::ACE_TP_Reactor ( const ACE_TP_Reactor & ) [private]

Deny access since member-wise won't work...

Member Function Documentation

void ACE_TP_Reactor::clear_dispatch_mask	(	ACE_HANDLE	handle,
		ACE_Reactor_Mask	mask
	)			`[protected, virtual]`

Template method from the base class.

Reimplemented from ACE_Select_Reactor_Impl.

Definition at line 111 of file TP_Reactor.inl.

{
  this->ready_set_.rd_mask_.reset ();
  this->ready_set_.wr_mask_.reset ();
  this->ready_set_.ex_mask_.reset ();
}

void ACE_TP_Reactor::clear_handle_read_set ( ACE_HANDLE handle ) [private]

Clear the handle from the read_set.

Definition at line 103 of file TP_Reactor.inl.

{
  this->ready_set_.wr_mask_.clr_bit (handle);
  this->ready_set_.ex_mask_.clr_bit (handle);
  this->ready_set_.rd_mask_.clr_bit (handle);
}

int ACE_TP_Reactor::dispatch_i	(	ACE_Time_Value *	max_wait_time,
		ACE_TP_Token_Guard &	guard
	)			`[protected]`

Dispatch just 1 signal, timer, notification handlers.

Definition at line 177 of file TP_Reactor.cpp.

{
  int event_count = this->get_event_for_dispatching (max_wait_time);

  // We use this count to detect potential infinite loops as described
  // in bug 2540.
  int const initial_event_count = event_count;

  int result = 0;

  // Note: We are passing the <event_count> around, to have record of
  // how many events still need processing. May be this could be
  // useful in future.

#if 0
  // @Ciju
  // signal handling isn't in a production state yet.
  // Commenting it out for now.

  // Dispatch signals
  if (event_count == -1)
    {
      // Looks like we dont do any upcalls in dispatch signals. If at
      // a later point of time, we decide to handle signals we have to
      // release the lock before we make any upcalls.. What is here
      // now is not the right thing...
      //
      // @@ We need to do better..
      return this->handle_signals (event_count, guard);
    }
#endif // #if 0

  // If there are no signals and if we had received a proper
  // event_count then first look at dispatching timeouts. We need to
  // handle timers early since they may have higher latency
  // constraints than I/O handlers.  Ideally, the order of dispatching
  // should be a strategy...

  // NOTE: The event count does not have the number of timers that
  // needs dispatching. But we are still passing this along. We dont
  // need to do that. In the future we *may* have the timers also
  // returned through the <event_count>. Just passing that along for
  // that day.
  result = this->handle_timer_events (event_count, guard);

  if (result > 0)
    return result;

  // Else just go ahead fall through for further handling.

  if (event_count > 0)
    {
      // Next dispatch the notification handlers (if there are any to
      // dispatch).  These are required to handle multiple-threads
      // that are trying to update the <Reactor>.
      result = this->handle_notify_events (event_count, guard);

      if (result > 0)
        return result;

      // Else just fall through for further handling
    }

  if (event_count > 0)
    {
      // Handle socket events
      result = this->handle_socket_events (event_count, guard);
    }

  if (event_count != 0 && event_count == initial_event_count)
    {
      this->state_changed_ = true;
    }

  return result;
}

int ACE_TP_Reactor::dispatch_socket_event ( ACE_EH_Dispatch_Info & dispatch_info ) [private]

Notify the appropriate <callback> in the context of the <eh> associated with <handle> that a particular event has occurred.

Definition at line 557 of file TP_Reactor.cpp.

{
  ACE_TRACE ("ACE_TP_Reactor::dispatch_socket_event");

  ACE_Event_Handler * const event_handler = dispatch_info.event_handler_;
  ACE_EH_PTMF const callback = dispatch_info.callback_;

  // Check for removed handlers.
  if (event_handler == 0)
    return -1;

  // Upcall. If the handler returns positive value (requesting a
  // reactor callback) don't set the ready-bit because it will be
  // ignored if the reactor state has changed. Just call back
  // as many times as the handler requests it. Other threads are off
  // handling other things.
  int status = 1;
  while (status > 0)
    status = (event_handler->*callback) (dispatch_info.handle_);

  // Post process socket event
  return this->post_process_socket_event (dispatch_info, status);
}

int ACE_TP_Reactor::get_event_for_dispatching ( ACE_Time_Value * max_wait_time ) [protected]

Get the event that needs dispatching. It could be either a signal, timer, notification handlers or return possibly 1 I/O handler for dispatching. In the most common use case, this would return 1 I/O handler for dispatching

Definition at line 452 of file TP_Reactor.cpp.

{
  // If the reactor handler state has changed, clear any remembered
  // ready bits and re-scan from the master wait_set.
  if (this->state_changed_)
    {
      this->ready_set_.rd_mask_.reset ();
      this->ready_set_.wr_mask_.reset ();
      this->ready_set_.ex_mask_.reset ();

      this->state_changed_ = false;
    }
  else
    {
      // This is a hack... somewhere, under certain conditions (which
      // I don't understand...) the mask will have all of its bits clear,
      // yet have a size_ > 0. This is an attempt to remedy the affect,
      // without knowing why it happens.

      this->ready_set_.rd_mask_.sync (this->ready_set_.rd_mask_.max_set ());
      this->ready_set_.wr_mask_.sync (this->ready_set_.wr_mask_.max_set ());
      this->ready_set_.ex_mask_.sync (this->ready_set_.ex_mask_.max_set ());
    }

  return this->wait_for_multiple_events (this->ready_set_, max_wait_time);
}

ACE_HANDLE ACE_TP_Reactor::get_notify_handle ( void ) [private]

Get the handle of the notify pipe from the ready set if there is an event in the notify pipe.

Definition at line 667 of file TP_Reactor.cpp.

{
  // Call the notify handler to get a handle on which we would have a
  // notify waiting
  ACE_HANDLE const read_handle =
    this->notify_handler_->notify_handle ();

  // Check whether the rd_mask has been set on that handle. If so
  // return the handle.
  if (read_handle != ACE_INVALID_HANDLE &&
      this->ready_set_.rd_mask_.is_set (read_handle))
    {
      return read_handle;
    }

  // None found..
  return ACE_INVALID_HANDLE;
}

int ACE_TP_Reactor::get_socket_event_info ( ACE_EH_Dispatch_Info & info ) [private]

Get socket event dispatch information.

Definition at line 480 of file TP_Reactor.cpp.

{
  // Check for dispatch in write, except, read. Only catch one, but if
  // one is caught, be sure to clear the handle from each mask in case
  // there is more than one mask set for it. This would cause problems
  // if the handler is suspended for dispatching, but its set bit in
  // another part of ready_set_ kept it from being dispatched.
  int found_io = 0;
  ACE_HANDLE handle;

  // @@todo: We can do quite a bit of code reduction here. Let me get
  // it to work before I do this.
  {
    ACE_Handle_Set_Iterator handle_iter (this->ready_set_.wr_mask_);

    while (!found_io && (handle = handle_iter ()) != ACE_INVALID_HANDLE)
      {
        if (this->is_suspended_i (handle))
          continue;

        // Remember this info
        event.set (handle,
                   this->handler_rep_.find (handle),
                   ACE_Event_Handler::WRITE_MASK,
                   &ACE_Event_Handler::handle_output);

        this->clear_handle_read_set (handle);
        found_io = 1;
      }
  }

  if (!found_io)
    {
      ACE_Handle_Set_Iterator handle_iter (this->ready_set_.ex_mask_);

      while (!found_io && (handle = handle_iter ()) != ACE_INVALID_HANDLE)
        {
          if (this->is_suspended_i (handle))
            continue;

          // Remember this info
          event.set (handle,
                     this->handler_rep_.find (handle),
                     ACE_Event_Handler::EXCEPT_MASK,
                     &ACE_Event_Handler::handle_exception);

          this->clear_handle_read_set (handle);

          found_io = 1;
        }
    }

  if (!found_io)
    {
      ACE_Handle_Set_Iterator handle_iter (this->ready_set_.rd_mask_);

      while (!found_io && (handle = handle_iter ()) != ACE_INVALID_HANDLE)
        {
          if (this->is_suspended_i (handle))
            continue;

          // Remember this info
          event.set (handle,
                     this->handler_rep_.find (handle),
                     ACE_Event_Handler::READ_MASK,
                     &ACE_Event_Handler::handle_input);

          this->clear_handle_read_set (handle);
          found_io = 1;
        }
    }

  return found_io;
}

int ACE_TP_Reactor::handle_events ( ACE_Time_Value & max_wait_time ) [virtual]

This method is just like the one above, except the max_wait_time value is a reference and can therefore never be NULL.

Current <alertable_handle_events> is identical to <handle_events>.

Reimplemented from ACE_Select_Reactor_T< ACE_Select_Reactor_Token >.

Definition at line 646 of file TP_Reactor.cpp.

{
  return this->handle_events (&max_wait_time);
}

int ACE_TP_Reactor::handle_events ( ACE_Time_Value * max_wait_time = 0 ) [virtual]

This event loop driver that blocks for max_wait_time before returning. It will return earlier if timer events, I/O events, or signal events occur. Note that max_wait_time can be 0, in which case this method blocks indefinitely until events occur.

max_wait_time is decremented to reflect how much time this call took. For instance, if a time value of 3 seconds is passed to handle_events and an event occurs after 2 seconds, max_wait_time will equal 1 second. This can be used if an application wishes to handle events for some fixed amount of time.

Returns:: The total number of events that were dispatched; 0 if the max_wait_time elapsed without dispatching any handlers, or -1 if an error occurs (check errno for more information).

Reimplemented from ACE_Select_Reactor_T< ACE_Select_Reactor_Token >.

Definition at line 143 of file TP_Reactor.cpp.

{
  ACE_TRACE ("ACE_TP_Reactor::handle_events");

  // Stash the current time -- the destructor of this object will
  // automatically compute how much time elapsed since this method was
  // called.
  ACE_Countdown_Time countdown (max_wait_time);

  //
  // The order of these events is very subtle, modify with care.
  //

  // Instantiate the token guard which will try grabbing the token for
  // this thread.
  ACE_TP_Token_Guard guard (this->token_);

  int const result = guard.acquire_read_token (max_wait_time);

  // If the guard is NOT the owner just return the retval
  if (!guard.is_owner ())
    return result;

  // After getting the lock just just for deactivation..
  if (this->deactivated_)
    return -1;

  // Update the countdown to reflect time waiting for the token.
  countdown.update ();

  return this->dispatch_i (max_wait_time, guard);
}

int ACE_TP_Reactor::handle_notify_events	(	int &	event_count,
		ACE_TP_Token_Guard &	g
	)			`[protected]`

Handle notify events.

Definition at line 347 of file TP_Reactor.cpp.

{
  // Get the handle on which notify calls could have occured
  ACE_HANDLE notify_handle = this->get_notify_handle ();

  int result = 0;

  // The notify was not in the list returned by
  // wait_for_multiple_events ().
  if (notify_handle == ACE_INVALID_HANDLE)
    return result;

  // Now just do a read on the pipe..
  ACE_Notification_Buffer buffer;

  // Clear the handle of the read_mask of our <ready_set_>
  this->ready_set_.rd_mask_.clr_bit (notify_handle);

  // Keep reading notifies till we empty it or till we have a
  // dispatchable buffer
  while (this->notify_handler_->read_notify_pipe (notify_handle, buffer) > 0)
    {
      // Just figure out whether we can read any buffer that has
      // dispatchable info. If not we have just been unblocked by
      // another thread trying to update the reactor. If we get any
      // buffer that needs dispatching we will dispatch that after
      // releasing the lock
      if (this->notify_handler_->is_dispatchable (buffer) > 0)
        {
          // Release the token before dispatching notifies...
          guard.release_token ();

          // Dispatch the upcall for the notify
          this->notify_handler_->dispatch_notify (buffer);

          // We had a successful dispatch.
          result = 1;

          // break out of the while loop
          break;
        }
    }

  // If we did some work, then we just return 1 which will allow us
  // to get out of here. If we return 0, then we will be asked to do
  // some work ie. dispacth socket events
  return result;
}

int ACE_TP_Reactor::handle_socket_events	(	int &	event_count,
		ACE_TP_Token_Guard &	g
	)			`[protected]`

handle socket events

Definition at line 398 of file TP_Reactor.cpp.

{

  // We got the lock, lets handle some I/O events.
  ACE_EH_Dispatch_Info dispatch_info;

  this->get_socket_event_info (dispatch_info);

  // If there is any event handler that is ready to be dispatched, the
  // dispatch information is recorded in dispatch_info.
  if (!dispatch_info.dispatch ())
    {
      // Check for removed handlers.
      if (dispatch_info.event_handler_ == 0)
        {
          this->handler_rep_.unbind(dispatch_info.handle_,
                                    dispatch_info.mask_);
        }


      return 0;
    }

  // Suspend the handler so that other threads don't start dispatching
  // it, if we can't suspend then return directly
  //
  // NOTE: This check was performed in older versions of the
  // TP_Reactor. Looks like it is a waste..
  if (dispatch_info.event_handler_ != this->notify_handler_)
    if (this->suspend_i (dispatch_info.handle_) == -1)
      return 0;

  // Call add_reference() if needed.
  if (dispatch_info.reference_counting_required_)
    dispatch_info.event_handler_->add_reference ();

  // Release the lock.  Others threads can start waiting.
  guard.release_token ();

  int result = 0;

  // If there was an event handler ready, dispatch it.
  // Decrement the event left
  --event_count;

  // Dispatched an event
  if (this->dispatch_socket_event (dispatch_info) == 0)
    ++result;

  return result;
}

int ACE_TP_Reactor::handle_timer_events	(	int &	event_count,
		ACE_TP_Token_Guard &	g
	)			`[protected]`

Handle timer events.

Definition at line 307 of file TP_Reactor.cpp.

{
  if (this->timer_queue_ == 0 || this->timer_queue_->is_empty())
    { // Empty timer queue so cannot have any expired timers.
      return 0;
    }

  // Get the current time
  ACE_Time_Value cur_time (this->timer_queue_->gettimeofday () +
                           this->timer_queue_->timer_skew ());

  // Look for a node in the timer queue whose timer <= the present
  // time.
  ACE_Timer_Node_Dispatch_Info info;

  if (this->timer_queue_->dispatch_info (cur_time, info))
    {
      const void *upcall_act = 0;

      // Preinvoke.
      this->timer_queue_->preinvoke (info, cur_time, upcall_act);

      // Release the token before dispatching notifies...
      guard.release_token ();

      // call the functor
      this->timer_queue_->upcall (info, cur_time);

      // Postinvoke
      this->timer_queue_->postinvoke (info, cur_time, upcall_act);

      // We have dispatched a timer
      return 1;
    }

  return 0;
}

void ACE_TP_Reactor::no_op_sleep_hook ( void * ) [static]

Called from handle events.

Definition at line 98 of file TP_Reactor.inl.

{
}

void ACE_TP_Reactor::notify_handle	(	ACE_HANDLE	handle,
		ACE_Reactor_Mask	mask,
		ACE_Handle_Set &	,
		ACE_Event_Handler *	eh,
		ACE_EH_PTMF	callback
	)			`[protected, virtual]`

This method shouldn't get called.

Reimplemented from ACE_Select_Reactor_T< ACE_Select_Reactor_Token >.

Definition at line 652 of file TP_Reactor.cpp.

{
  ACE_ERROR ((LM_ERROR,
              ACE_TEXT ("ACE_TP_Reactor::notify_handle: ")
              ACE_TEXT ("Wrong version of notify_handle() got called\n")));

  ACE_ASSERT (eh == 0);
  ACE_UNUSED_ARG (eh);
}

ACE_TP_Reactor& ACE_TP_Reactor::operator= ( const ACE_TP_Reactor & ) [private]

int ACE_TP_Reactor::owner ( ACE_thread_t * t_id ) [virtual]

Return the thread ID of the current Leader.

Reimplemented from ACE_Select_Reactor_T< ACE_Select_Reactor_Token >.

Definition at line 134 of file TP_Reactor.cpp.

{
  ACE_TRACE ("ACE_TP_Reactor::owner");
  *t_id = ACE_Thread::self ();

  return 0;
}

int ACE_TP_Reactor::owner	(	ACE_thread_t	n_id,
		ACE_thread_t *	o_id = `0`
	)			`[virtual]`

The ACE_TP_Reactor implementation does not have a single owner thread. Attempts to set the owner explicitly are ignored. The reported owner thread is the current Leader in the pattern.

Reimplemented from ACE_Select_Reactor_T< ACE_Select_Reactor_Token >.

Definition at line 124 of file TP_Reactor.cpp.

{
  ACE_TRACE ("ACE_TP_Reactor::owner");
  if (o_id)
    *o_id = ACE_Thread::self ();

  return 0;
}

int ACE_TP_Reactor::post_process_socket_event	(	ACE_EH_Dispatch_Info &	dispatch_info,
		int	status
	)			`[private]`

Definition at line 582 of file TP_Reactor.cpp.

{
  int result = 0;

  // First check if we really have to post process something, if not, then
  // we don't acquire the token which saves us a lot of time.
  if (status < 0 ||
     (dispatch_info.event_handler_ != this->notify_handler_ &&
      dispatch_info.resume_flag_ ==
        ACE_Event_Handler::ACE_REACTOR_RESUMES_HANDLER))
    {
      // Get the reactor token and with this token acquired remove first the
      // handler and resume it at the same time. This must be atomic, see also
      // bugzilla 2395. When this is not atomic it can be that we resume the
      // handle after it is reused by the OS.
      ACE_TP_Token_Guard guard (this->token_);

      result = guard.acquire_token ();

      // If the guard is NOT the owner just return the retval
      if (!guard.is_owner ())
        return result;

      // A different event handler may have been registered during the
      // upcall if the handle was closed and then reopened, for
      // example.  Make sure we're removing and/or resuming the event
      // handler used during the upcall.
      ACE_Event_Handler const * const eh =
        this->handler_rep_.find (dispatch_info.handle_);

      // Only remove or resume the event handler used during the
      // upcall.
      if (eh == dispatch_info.event_handler_)
        {
          if (status < 0)
            {
              result =
                this->remove_handler_i (dispatch_info.handle_,
                                        dispatch_info.mask_);
            }

          // Resume handler if required.
          if (dispatch_info.event_handler_ != this->notify_handler_ &&
              dispatch_info.resume_flag_ ==
              ACE_Event_Handler::ACE_REACTOR_RESUMES_HANDLER)
            this->resume_i (dispatch_info.handle_);
        }
    }

  // Call remove_reference() if needed.
  if (dispatch_info.reference_counting_required_)
    dispatch_info.event_handler_->remove_reference ();

  return result;
}

int ACE_TP_Reactor::resumable_handler ( void ) [virtual]

Does the reactor allow the application to resume the handle on its own ie. can it pass on the control of handle resumption to the application. The TP reactor has can allow applications to resume handles. So return a positive value.

Reimplemented from ACE_Select_Reactor_Impl.

Definition at line 640 of file TP_Reactor.cpp.