ACE/ace/Priority_Reactor.cpp

   1 #include "ace/Priority_Reactor.h"
   2 #include "ace/Malloc_T.h"
   3
   4
   5
   6 ACE_BEGIN_VERSIONED_NAMESPACE_DECL
   7
   8 typedef ACE_Unbounded_Queue_Iterator<ACE_Event_Tuple> QUEUE_ITERATOR;
   9 // Its iterator.
  10
  11 typedef ACE_Cached_Allocator<ACE_Node<ACE_Event_Tuple>, ACE_SYNCH_NULL_MUTEX> TUPLE_ALLOCATOR;
  12 // Defines the memory allocator used, no need for locking because it
  13 // is only used in one thread of control.
  14
  15 ACE_ALLOC_HOOK_DEFINE(ACE_Priority_Reactor)
  16
  17 // Initialize ACE_Select_Reactor.
  18
  19 #define npriorities \
  20         ACE_Event_Handler::HI_PRIORITY-ACE_Event_Handler::LO_PRIORITY+1
  21
  22 void
  23 ACE_Priority_Reactor::init_bucket (void)
  24 {
  25   // Allocate enough space for all the handles.
  26   // TODO: This can be wrong, maybe we should use other kind of
  27   // allocator here?
  28   ACE_NEW (this->tuple_allocator_,
  29            TUPLE_ALLOCATOR (ACE_Select_Reactor::DEFAULT_SIZE));
  30
  31   // The event handlers are assigned to a new As the Event
  32 #if defined (ACE_HAS_ALLOC_HOOKS)
  33   ACE_ALLOCATOR (this->bucket_,
  34                  static_cast<QUEUE **>(ACE_Allocator::instance()->malloc(sizeof(QUEUE *) * (npriorities))));
  35 #else
  36   ACE_NEW (this->bucket_,
  37            QUEUE *[npriorities]);
  38 #endif /* ACE_HAS_ALLOC_HOOKS */
  39
  40   // This loops "ensures" exception safety.
  41   for (int i = 0; i < npriorities; ++i)
  42     ACE_NEW (this->bucket_[i],
  43              QUEUE (this->tuple_allocator_));
  44 }
  45
  46 ACE_Priority_Reactor::ACE_Priority_Reactor (ACE_Sig_Handler *sh,
  47                                             ACE_Timer_Queue *tq)
  48   : ACE_Select_Reactor(sh, tq),
  49     bucket_ (0),
  50     tuple_allocator_ (0)
  51 {
  52   ACE_TRACE ("ACE_Priority_Reactor::ACE_Priority_Reactor");
  53   this->init_bucket ();
  54 }
  55
  56 ACE_Priority_Reactor::ACE_Priority_Reactor (size_t size,
  57                                             bool restart,
  58                                             ACE_Sig_Handler *sh,
  59                                             ACE_Timer_Queue *tq)
  60   : ACE_Select_Reactor (size, restart, sh, tq),
  61     bucket_ (0),
  62     tuple_allocator_ (0)
  63 {
  64   ACE_TRACE ("ACE_Priority_Reactor::ACE_Priority_Reactor");
  65   this->init_bucket ();
  66 }
  67
  68 ACE_Priority_Reactor::~ACE_Priority_Reactor (void)
  69 {
  70   ACE_TRACE ("ACE_Priority_Reactor::~ACE_Priority_Reactor");
  71
  72   for (int i = 0; i < npriorities; ++i)
  73     delete this->bucket_[i];
  74
  75 #if defined (ACE_HAS_ALLOC_HOOKS)
  76   ACE_Allocator::instance()->free(this->bucket_);
  77 #else
  78   delete[] this->bucket_;
  79 #endif /* ACE_HAS_ALLOC_HOOKS */
  80   delete tuple_allocator_;
  81 }
  82
  83 int
  84 ACE_Priority_Reactor::build_bucket (ACE_Handle_Set &dispatch_mask,
  85                                     int &min_priority,
  86                                     int &max_priority)
  87 {
  88   ACE_Handle_Set_Iterator handle_iter (dispatch_mask);
  89
  90   for (ACE_HANDLE handle;
  91        (handle = handle_iter ()) != ACE_INVALID_HANDLE;
  92        )
  93     {
  94       ACE_Event_Handler *event_handler =
  95         this->handler_rep_.find (handle);
  96       if (event_handler == 0)
  97         return -1;
  98
  99       ACE_Event_Tuple et (event_handler,
 100                           handle);
 101       int prio = et.event_handler_->priority ();
 102
 103       // If the priority is out of range assign the minimum priority.
 104       if (prio < ACE_Event_Handler::LO_PRIORITY
 105           || prio > ACE_Event_Handler::HI_PRIORITY)
 106         prio = ACE_Event_Handler::LO_PRIORITY;
 107
 108       if (bucket_[prio]->enqueue_tail (et) == -1)
 109         return -1;
 110
 111       // Update the priority ranges....
 112       if (min_priority > prio)
 113         min_priority = prio;
 114       if (max_priority < prio)
 115         max_priority = prio;
 116     }
 117
 118   return 0;
 119 }
 120
 121 int
 122 ACE_Priority_Reactor::dispatch_io_set (int number_of_active_handles,
 123                                        int& number_dispatched,
 124                                        int mask,
 125                                        ACE_Handle_Set& dispatch_mask,
 126                                        ACE_Handle_Set& ready_mask,
 127                                        ACE_EH_PTMF callback)
 128 {
 129   ACE_TRACE ("ACE_Priority_Reactor::dispatch_io_set");
 130
 131   if (number_of_active_handles == 0)
 132     return 0;
 133
 134   // The range for which there exists any Event_Tuple is computed on
 135   // the ordering loop, minimizing iterations on the dispatching loop.
 136   int min_priority =
 137     ACE_Event_Handler::HI_PRIORITY;
 138   int max_priority =
 139     ACE_Event_Handler::LO_PRIORITY;
 140
 141   if (this->build_bucket (dispatch_mask,
 142                           min_priority,
 143                           max_priority) == -1)
 144     return -1;
 145
 146   for (int i = max_priority; i >= min_priority; --i)
 147     {
 148       while (!bucket_[i]->is_empty ()
 149              && number_dispatched < number_of_active_handles)
 150         {
 151
 152           ACE_Event_Tuple et;
 153
 154           bucket_[i]->dequeue_head (et);
 155
 156           this->notify_handle (et.handle_,
 157                                mask,
 158                                ready_mask,
 159                                et.event_handler_,
 160                                callback);
 161           ++number_dispatched;
 162
 163           // clear the bit from that dispatch mask,
 164           // so when we need to restart the iteration (rebuilding the iterator...)
 165           // we will not dispatch the already dispatched handlers
 166           this->clear_dispatch_mask (et.handle_,
 167                                      mask);
 168
 169           if (this->state_changed_)
 170             this->state_changed_ = false; // so it will not rebuild it ...
 171         }
 172
 173       // Even if we are aborting the loop due to this->state_changed
 174       // or another error we still want to cleanup the buckets.
 175       bucket_[i]->reset ();
 176     }
 177
 178   return 0;
 179 }
 180
 181 void
 182 ACE_Priority_Reactor::dump (void) const
 183 {
 184 #if defined (ACE_HAS_DUMP)
 185   ACE_TRACE ("ACE_Priority_Reactor::dump");
 186
 187   ACELIB_DEBUG ((LM_DEBUG, ACE_BEGIN_DUMP, this));
 188
 189   ACE_Select_Reactor::dump ();
 190
 191   ACELIB_DEBUG ((LM_DEBUG, ACE_END_DUMP));
 192 #endif /* ACE_HAS_DUMP */
 193 }
 194
 195 ACE_END_VERSIONED_NAMESPACE_DECL