A queue of Routing_Slips waiting to be persisted. More...

#include <Routing_Slip_Queue.h>

Collaboration diagram for TAO_Notify::Routing_Slip_Queue:

Public Member Functions
	Routing_Slip_Queue (size_t allowed=1)
	Construct setting "allowed".
	~Routing_Slip_Queue ()
	Destructor.
void	add (const Routing_Slip_Ptr &routing_slip)
	Add a routing slip to the tail of the queue and dispatch if necessary.
void	complete ()
	A call back to indicate that processing is complete for a previously-queued Routing_Slip.
void	set_allowed (size_t allowed)
Private Types
typedef ACE_Unbounded_Queue < Routing_Slip_Ptr >	Queue
typedef ACE_Guard < TAO_SYNCH_MUTEX >	Guard
Private Member Functions
void	dispatch (Guard &guard)
bool	dispatch_one (Guard &guard)
	Routing_Slip_Queue (const Routing_Slip_Queue &rhs)
Routing_Slip_Queue &	operator= (const Routing_Slip_Queue &rhs)
Private Attributes
size_t	allowed_
TAO_SYNCH_MUTEX	internals_
	Protection for internal information.
size_t	active_
Queue	queue_

Detailed Description

A queue of Routing_Slips waiting to be persisted.

The Routing_Slip_Queue keeps a queue of Routing_Slips waiting to be written to persistent storage. The "allowed" parameter determines how many Routing_Slips can be handled simultaneously by the persistent storage. Until this threshold is reached, Routing_Slips are not held in the queue, but pass straight through.

Once the allowe number of Routing_Slips are being handled, any additional requests are held in the queue until persistence is complete for another Routing_Slips.

Having Routing_Slips waiting in the queue is "a good thing" [TM] because it allows delivery completions to be applied to the routing slip before it is written -- thereby reducing or completely eliminating the number of actual writes to persistent storage.

Experimentation indicates that a good value for "allowed" is 1.

Allowed == 0 is treated as a special case meaning pass all Routing_Slips through the queue immediately. Setting it a good way to test how well your storage device withstands continuous beating.

Definition at line 43 of file Routing_Slip_Queue.h.

Member Typedef Documentation

typedef ACE_Guard< TAO_SYNCH_MUTEX > TAO_Notify::Routing_Slip_Queue::Guard [private]

Definition at line 46 of file Routing_Slip_Queue.h.

typedef ACE_Unbounded_Queue<Routing_Slip_Ptr> TAO_Notify::Routing_Slip_Queue::Queue [private]

Definition at line 45 of file Routing_Slip_Queue.h.

Constructor & Destructor Documentation

TAO_Notify::Routing_Slip_Queue::Routing_Slip_Queue ( size_t allowed = 1 )

Construct setting "allowed".

Parameters:

allowed

the number of Routing_Slips that can be handled simultaneously by the persistent store.

Definition at line 18 of file Routing_Slip_Queue.cpp.

    : allowed_ (allowed)
    , active_ (0)
  {
  }

TAO_Notify::Routing_Slip_Queue::~Routing_Slip_Queue ( )

Destructor.

Definition at line 24 of file Routing_Slip_Queue.cpp.

  {
  }

TAO_Notify::Routing_Slip_Queue::Routing_Slip_Queue ( const Routing_Slip_Queue & rhs ) [private]

Member Function Documentation

void TAO_Notify::Routing_Slip_Queue::add ( const Routing_Slip_Ptr & routing_slip )

Add a routing slip to the tail of the queue and dispatch if necessary.

Definition at line 29 of file Routing_Slip_Queue.cpp.

  {
    Guard guard (internals_);
    ACE_ASSERT (guard.locked()); // check recursion
    if (this->allowed_ == 0)
    {
      ++this->active_;
      guard.release ();
      routing_slip->at_front_of_persist_queue ();
//      guard.acquire ();
    }
    else
    {
      this->queue_.enqueue_tail (routing_slip);
      dispatch (guard);
    }
  }

void TAO_Notify::Routing_Slip_Queue::complete ( )

A call back to indicate that processing is complete for a previously-queued Routing_Slip.

Definition at line 47 of file Routing_Slip_Queue.cpp.

  {
    Guard guard (internals_);
    ACE_ASSERT (guard.locked()); // check recursion
    ACE_ASSERT (this->active_ > 0);
    --this->active_;
    dispatch (guard);
  }

void TAO_Notify::Routing_Slip_Queue::dispatch ( Guard & guard ) [private]

Definition at line 57 of file Routing_Slip_Queue.cpp.

  {
    // we start out pretty nice,
    // but the more work we do for other people
    // the less nice we get.
    size_t nice = this->allowed_ + 1;
    while (nice > 0 && (this->active_ < this->allowed_))
    {
      if (dispatch_one (guard))
      {
        --nice;
      }
      else
      {
        // that's about as nice as I get.
        nice = 0;
      }
    }
  }

bool TAO_Notify::Routing_Slip_Queue::dispatch_one ( Guard & guard ) [private]

Definition at line 78 of file Routing_Slip_Queue.cpp.

  {
    bool ok = false;
    Routing_Slip_Ptr routing_slip;
    if (this->queue_.dequeue_head (routing_slip) == 0)
    {
      ++this->active_;
      guard.release ();
      routing_slip->at_front_of_persist_queue ();
      guard.acquire ();
    }
    return ok;
  }

Routing_Slip_Queue& TAO_Notify::Routing_Slip_Queue::operator= ( const Routing_Slip_Queue & rhs ) [private]

void TAO_Notify::Routing_Slip_Queue::set_allowed ( size_t allowed )

/brief Adjust the "allowed" value on-the-fly (not recommended, but it works.)

Definition at line 93 of file Routing_Slip_Queue.cpp.

  {
    Guard guard (internals_);
    size_t allowed_was = this->allowed_;
    this->allowed_ = allowed;
    if (allowed == 0 && allowed_was != 0)
    {
      while (dispatch_one (guard))
      {
        ; // work happens in dispatc_one
      }
    }
    else
    {
      dispatch (guard);
    }
  }