Implements priority based dispatching. More...

#include <Priority_Reactor.h>

Inheritance diagram for ACE_Priority_Reactor:

Collaboration diagram for ACE_Priority_Reactor:

Public Member Functions
	ACE_Priority_Reactor (ACE_Sig_Handler =0, ACE_Timer_Queue =0)
	Initialize ACE_Priority_Reactor with the default size.
	ACE_Priority_Reactor (size_t size, bool restart=false, ACE_Sig_Handler =0, ACE_Timer_Queue =0)
	Initialize ACE_Priority_Reactor with size size.
virtual	~ACE_Priority_Reactor (void)
	Close down the select_reactor and release all of its resources.
void	dump (void) const
	Dump the state of an object.
Public Attributes
	ACE_ALLOC_HOOK_DECLARE
	Declare the dynamic allocation hooks.
Protected Member Functions
virtual int	dispatch_io_set (int number_of_active_handles, int &number_dispatched, int mask, ACE_Handle_Set &dispatch_mask, ACE_Handle_Set &ready_mask, ACE_EH_PTMF callback)
Private Types
typedef ACE_Unbounded_Queue < ACE_Event_Tuple >	QUEUE
Private Member Functions
void	init_bucket (void)
	A small helper to initialize the bucket.
int	build_bucket (ACE_Handle_Set &dispatch_mask, int &min_priority, int &max_priority)
	ACE_Priority_Reactor (const ACE_Priority_Reactor &)
	Deny access since member-wise won't work...
ACE_Priority_Reactor &	operator= (const ACE_Priority_Reactor &)
Private Attributes
QUEUE **	bucket_
ACE_Allocator *	tuple_allocator_

Detailed Description

Implements priority based dispatching.

This class refines the dispatching mechanism for the Select_Reactor by taking advantage of the priority method on ACE_Event_Handler.

Definition at line 37 of file Priority_Reactor.h.

Member Typedef Documentation

typedef ACE_Unbounded_Queue<ACE_Event_Tuple> ACE_Priority_Reactor::QUEUE [private]

There is a queue per-priority, which simply holds the Event_Handlers until we know who goes first.

Definition at line 85 of file Priority_Reactor.h.

Constructor & Destructor Documentation

ACE_Priority_Reactor::ACE_Priority_Reactor	(	ACE_Sig_Handler *	sh = `0`,
		ACE_Timer_Queue *	tq = `0`
	)

Initialize ACE_Priority_Reactor with the default size.

Definition at line 43 of file Priority_Reactor.cpp.

  : ACE_Select_Reactor(sh, tq),
    bucket_ (0),
    tuple_allocator_ (0)
{
  ACE_TRACE ("ACE_Priority_Reactor::ACE_Priority_Reactor");
  this->init_bucket ();
}

ACE_Priority_Reactor::ACE_Priority_Reactor	(	size_t	size,
		bool	restart = `false`,
		ACE_Sig_Handler *	sh = `0`,
		ACE_Timer_Queue *	tq = `0`
	)

Initialize ACE_Priority_Reactor with size size.

Definition at line 53 of file Priority_Reactor.cpp.

  : ACE_Select_Reactor (size, restart, sh, tq),
    bucket_ (0),
    tuple_allocator_ (0)
{
  ACE_TRACE ("ACE_Priority_Reactor::ACE_Priority_Reactor");
  this->init_bucket ();
}

ACE_Priority_Reactor::~ACE_Priority_Reactor ( void ) [virtual]

Close down the select_reactor and release all of its resources.

Definition at line 65 of file Priority_Reactor.cpp.

{
  ACE_TRACE ("ACE_Priority_Reactor::~ACE_Priority_Reactor");

  for (int i = 0; i < npriorities; ++i)
    delete this->bucket_[i];

  delete[] this->bucket_;
  delete tuple_allocator_;
}

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

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

Member Function Documentation

int ACE_Priority_Reactor::build_bucket	(	ACE_Handle_Set &	dispatch_mask,
		int &	min_priority,
		int &	max_priority
	)			`[private]`

Build the bucket from the given dispatch_mask. Return -1 on failure, 0 otherwise.

Definition at line 77 of file Priority_Reactor.cpp.

{
  ACE_Handle_Set_Iterator handle_iter (dispatch_mask);

  for (ACE_HANDLE handle;
       (handle = handle_iter ()) != ACE_INVALID_HANDLE;
       )
    {
      ACE_Event_Handler *event_handler =
        this->handler_rep_.find (handle);
      if (event_handler == 0)
        return -1;

      ACE_Event_Tuple et (event_handler,
                          handle);
      int prio = et.event_handler_->priority ();

      // If the priority is out of range assign the minimum priority.
      if (prio < ACE_Event_Handler::LO_PRIORITY
          || prio > ACE_Event_Handler::HI_PRIORITY)
        prio = ACE_Event_Handler::LO_PRIORITY;

      if (bucket_[prio]->enqueue_tail (et) == -1)
        return -1;

      // Update the priority ranges....
      if (min_priority > prio)
        min_priority = prio;
      if (max_priority < prio)
        max_priority = prio;
    }

  return 0;
}

int ACE_Priority_Reactor::dispatch_io_set	(	int	number_of_active_handles,
		int &	number_dispatched,
		int	mask,
		ACE_Handle_Set &	dispatch_mask,
		ACE_Handle_Set &	ready_mask,
		ACE_EH_PTMF	callback
	)			`[protected, virtual]`

We simply override this function to implement the priority dispatching.

Reimplemented from ACE_Select_Reactor_T< ACE_Select_Reactor_Token >.

Definition at line 115 of file Priority_Reactor.cpp.

{
  ACE_TRACE ("ACE_Priority_Reactor::dispatch_io_set");

  if (number_of_active_handles == 0)
    return 0;

  // The range for which there exists any Event_Tuple is computed on
  // the ordering loop, minimizing iterations on the dispatching loop.
  int min_priority =
    ACE_Event_Handler::HI_PRIORITY;
  int max_priority =
    ACE_Event_Handler::LO_PRIORITY;

  if (this->build_bucket (dispatch_mask,
                          min_priority,
                          max_priority) == -1)
    return -1;

  for (int i = max_priority; i >= min_priority; --i)
    {
      while (!bucket_[i]->is_empty ()
             && number_dispatched < number_of_active_handles)
        {

          ACE_Event_Tuple et;

          bucket_[i]->dequeue_head (et);

          this->notify_handle (et.handle_,
                               mask,
                               ready_mask,
                               et.event_handler_,
                               callback);
          number_dispatched++;

          // clear the bit from that dispatch mask,
          // so when we need to restart the iteration (rebuilding the iterator...)
          // we will not dispatch the already dipatched handlers
          this->clear_dispatch_mask (et.handle_,
                                     mask);

          if (this->state_changed_)
            this->state_changed_ = false; // so it will not rebuild it ...
        }

      // Even if we are aborting the loop due to this->state_changed
      // or another error we still want to cleanup the buckets.
      bucket_[i]->reset ();
    }

  return 0;
}

void ACE_Priority_Reactor::dump ( void ) const [virtual]

Dump the state of an object.

Reimplemented from ACE_Select_Reactor_T< ACE_Select_Reactor_Token >.

Definition at line 175 of file Priority_Reactor.cpp.

{
#if defined (ACE_HAS_DUMP)
  ACE_TRACE ("ACE_Priority_Reactor::dump");

  ACE_DEBUG ((LM_DEBUG, ACE_BEGIN_DUMP, this));

  ACE_Select_Reactor::dump ();

  ACE_DEBUG ((LM_DEBUG, ACE_END_DUMP));
#endif /* ACE_HAS_DUMP */
}

void ACE_Priority_Reactor::init_bucket ( void ) [private]

A small helper to initialize the bucket.

Definition at line 25 of file Priority_Reactor.cpp.

{
  // Allocate enough space for all the handles.
  // TODO: This can be wrong, maybe we should use other kind of
  // allocator here?
  ACE_NEW (this->tuple_allocator_,
           TUPLE_ALLOCATOR (ACE_Select_Reactor::DEFAULT_SIZE));

  // The event handlers are assigned to a new As the Event
  ACE_NEW (this->bucket_,
           QUEUE *[npriorities]);

  // This loops "ensures" exception safety.
  for (int i = 0; i < npriorities; ++i)
    ACE_NEW (this->bucket_[i],
             QUEUE (this->tuple_allocator_));
}