ACE/ace/Notification_Queue.cpp

   1 #include "ace/Notification_Queue.h"
   2
   3 #if !defined (__ACE_INLINE__)
   4 #include "ace/Notification_Queue.inl"
   5 #endif /* __ACE_INLINE__ */
   6
   7 #include "ace/Guard_T.h"
   8
   9 ACE_BEGIN_VERSIONED_NAMESPACE_DECL
  10
  11 ACE_ALLOC_HOOK_DEFINE(ACE_Notification_Queue_Node)
  12
  13 ACE_Notification_Queue::ACE_Notification_Queue()
  14   : alloc_queue_()
  15   , notify_queue_()
  16   , free_queue_()
  17 {
  18 }
  19
  20 ACE_Notification_Queue::~ACE_Notification_Queue()
  21 {
  22   reset();
  23 }
  24
  25 int
  26 ACE_Notification_Queue::open()
  27 {
  28   ACE_TRACE ("ACE_Notification_Queue::open");
  29
  30   ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, mon, this->notify_queue_lock_, -1);
  31
  32   if (!this->free_queue_.is_empty ())
  33     return 0;
  34
  35   return allocate_more_buffers();
  36 }
  37
  38 void
  39 ACE_Notification_Queue::reset()
  40 {
  41   ACE_TRACE ("ACE_Notification_Queue::reset");
  42
  43   // Release all the event handlers still in the queue ...
  44   for (ACE_Notification_Queue_Node * node = notify_queue_.head();
  45        node != 0;
  46        node = node->next())
  47     {
  48       if (node->get().eh_ == 0)
  49         {
  50           continue;
  51         }
  52       (void) node->get().eh_->remove_reference();
  53     }
  54
  55   // ... free up the dynamically allocated resources ...
  56   ACE_Notification_Queue_Node **b = 0;
  57   for (ACE_Unbounded_Queue_Iterator<ACE_Notification_Queue_Node *> alloc_iter (this->alloc_queue_);
  58        alloc_iter.next (b) != 0;
  59        alloc_iter.advance ())
  60     {
  61       delete [] *b;
  62       *b = 0;
  63     }
  64
  65   // ... cleanup the list of allocated blocks ...
  66   this->alloc_queue_.reset ();
  67
  68   // ... swap with empty lists to reset the contents ...
  69   Buffer_List().swap(notify_queue_);
  70   Buffer_List().swap(free_queue_);
  71 }
  72
  73 int
  74 ACE_Notification_Queue::allocate_more_buffers()
  75 {
  76   ACE_TRACE ("ACE_Notification_Queue::allocate_more_buffers");
  77
  78   ACE_Notification_Queue_Node *temp = 0;
  79
  80   ACE_NEW_RETURN (temp,
  81                   ACE_Notification_Queue_Node[ACE_REACTOR_NOTIFICATION_ARRAY_SIZE],
  82                   -1);
  83
  84   if (this->alloc_queue_.enqueue_head (temp) == -1)
  85     {
  86       delete [] temp;
  87       return -1;
  88     }
  89
  90   for (size_t i = 0; i < ACE_REACTOR_NOTIFICATION_ARRAY_SIZE; ++i)
  91     {
  92       free_queue_.push_front(temp + i);
  93     }
  94
  95   return 0;
  96 }
  97
  98 int
  99 ACE_Notification_Queue::purge_pending_notifications(
 100   ACE_Event_Handler * eh,
 101   ACE_Reactor_Mask mask)
 102 {
 103   ACE_TRACE ("ACE_Notification_Queue::purge_pending_notifications");
 104
 105   ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, mon, this->notify_queue_lock_, -1);
 106
 107   if (this->notify_queue_.is_empty ())
 108     return 0;
 109
 110   int number_purged = 0;
 111   ACE_Notification_Queue_Node * node = notify_queue_.head();
 112   while(node != 0)
 113     {
 114       if (!node->matches_for_purging(eh))
 115         {
 116           // Easy case, skip to the next node
 117           node = node->next();
 118           continue;
 119         }
 120
 121       if (!node->mask_disables_all_notifications(mask))
 122         {
 123           // ... another easy case, skip this node too, but clear the
 124           // mask first ...
 125           node->clear_mask(mask);
 126           node = node->next();
 127           continue;
 128         }
 129
 130       // ... this is the more complicated case, we want to remove the
 131       // node from the notify_queue_ list.  First save the next node
 132       // on the list:
 133       ACE_Notification_Queue_Node * next = node->next();
 134
 135       // ... then remove it ...
 136       notify_queue_.unsafe_remove(node);
 137       ++number_purged;
 138
 139       // ... release resources ...
 140       ACE_Event_Handler *event_handler = node->get().eh_;
 141       event_handler->remove_reference ();
 142
 143       // ... now this is a free node ...
 144       free_queue_.push_front(node);
 145
 146       // ... go to the next node, if there is one ...
 147       node = next;
 148     }
 149
 150   return number_purged;
 151 }
 152
 153 int
 154 ACE_Notification_Queue::push_new_notification(
 155   ACE_Notification_Buffer const & buffer)
 156 {
 157   ACE_TRACE ("ACE_Notification_Queue::push_new_notification");
 158
 159   bool notification_required = false;
 160
 161   ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, mon, this->notify_queue_lock_, -1);
 162
 163   // No pending notifications.
 164   if (this->notify_queue_.is_empty ())
 165     notification_required = true;
 166
 167   if (free_queue_.is_empty())
 168     {
 169       if (allocate_more_buffers() == -1)
 170         {
 171           return -1;
 172         }
 173     }
 174
 175   ACE_Notification_Queue_Node * node =
 176     free_queue_.pop_front();
 177
 178   ACE_ASSERT (node != 0);
 179   node->set(buffer);
 180
 181   notify_queue_.push_back(node);
 182
 183   if (!notification_required)
 184     {
 185       return 0;
 186     }
 187
 188   return 1;
 189 }
 190
 191 int
 192 ACE_Notification_Queue::pop_next_notification(
 193   ACE_Notification_Buffer & current,
 194   bool & more_messages_queued,
 195   ACE_Notification_Buffer & next)
 196 {
 197   ACE_TRACE ("ACE_Notification_Queue::pop_next_notification");
 198
 199   more_messages_queued = false;
 200
 201   ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, mon, this->notify_queue_lock_, -1);
 202
 203   if (notify_queue_.is_empty ())
 204     {
 205       return 0;
 206     }
 207
 208   ACE_Notification_Queue_Node * node = notify_queue_.pop_front();
 209
 210   current = node->get();
 211   free_queue_.push_front(node);
 212
 213   if(!this->notify_queue_.is_empty())
 214     {
 215       more_messages_queued = true;
 216       next = notify_queue_.head()->get();
 217     }
 218
 219   return 1;
 220 }
 221
 222 ACE_END_VERSIONED_NAMESPACE_DECL