ACE/ace/Message_Queue_Vx.cpp

   1 #include "ace/Message_Queue.h"
   2 #include "ace/Message_Queue_Vx.h"
   3 #include "ace/Log_Category.h"
   4
   5 #if !defined (__ACE_INLINE__)
   6 #include "ace/Message_Queue_Vx.inl"
   7 #endif /* __ACE_INLINE__ */
   8
   9 ACE_BEGIN_VERSIONED_NAMESPACE_DECL
  10
  11 #if defined (ACE_VXWORKS)
  12
  13 ////////////////////////////////
  14 // class ACE_Message_Queue_Vx //
  15 ////////////////////////////////
  16
  17 ACE_ALLOC_HOOK_DEFINE (ACE_Message_Queue_Vx);
  18
  19 void
  20 ACE_Message_Queue_Vx::dump (void) const
  21 {
  22 #if defined (ACE_HAS_DUMP)
  23   ACE_TRACE ("ACE_Message_Queue_Vx::dump");
  24   ACELIB_DEBUG ((LM_DEBUG, ACE_BEGIN_DUMP, this));
  25   switch (this->state_)
  26     {
  27     case ACE_Message_Queue_Base::ACTIVATED:
  28       ACELIB_DEBUG ((LM_DEBUG,
  29                   ACE_TEXT ("state = ACTIVATED\n")));
  30       break;
  31     case ACE_Message_Queue_Base::DEACTIVATED:
  32       ACELIB_DEBUG ((LM_DEBUG,
  33                   ACE_TEXT ("state = DEACTIVATED\n")));
  34       break;
  35     case ACE_Message_Queue_Base::PULSED:
  36       ACELIB_DEBUG ((LM_DEBUG,
  37                   ACE_TEXT ("state = PULSED\n")));
  38       break;
  39     }
  40   ACELIB_DEBUG ((LM_DEBUG,
  41               ACE_TEXT ("low_water_mark = %d\n")
  42               ACE_TEXT ("high_water_mark = %d\n")
  43               ACE_TEXT ("cur_bytes = %d\n")
  44               ACE_TEXT ("cur_length = %d\n")
  45               ACE_TEXT ("cur_count = %d\n")
  46               ACE_TEXT ("head_ = %u\n")
  47               ACE_TEXT ("MSG_Q_ID = %u\n"),
  48               this->low_water_mark_,
  49               this->high_water_mark_,
  50               this->cur_bytes_,
  51               this->cur_length_,
  52               this->cur_count_,
  53               this->head_,
  54               this->tail_));
  55   ACELIB_DEBUG ((LM_DEBUG, ACE_END_DUMP));
  56 #endif /* ACE_HAS_DUMP */
  57 }
  58
  59 ACE_Message_Queue_Vx::ACE_Message_Queue_Vx (size_t max_messages,
  60                                             size_t max_message_length,
  61                                             ACE_Notification_Strategy *ns)
  62   : ACE_Message_Queue<ACE_NULL_SYNCH> (0, 0, ns),
  63     max_messages_ (static_cast<int> (max_messages)),
  64     max_message_length_ (static_cast<int> (max_message_length))
  65 {
  66   ACE_TRACE ("ACE_Message_Queue_Vx::ACE_Message_Queue_Vx");
  67
  68   if (this->open (max_messages_, max_message_length_, ns) == -1)
  69     ACELIB_ERROR ((LM_ERROR, ACE_TEXT ("open")));
  70 }
  71
  72 ACE_Message_Queue_Vx::~ACE_Message_Queue_Vx (void)
  73 {
  74   ACE_TRACE ("ACE_Message_Queue_Vx::~ACE_Message_Queue_Vx");
  75
  76   if (this->tail_ != 0  &&  this->close () == -1)
  77     ACELIB_ERROR ((LM_ERROR, ACE_TEXT ("close")));
  78 }
  79
  80 // Don't bother locking since if someone calls this function more than
  81 // once for the same queue, we're in bigger trouble than just
  82 // concurrency control!
  83
  84 int
  85 ACE_Message_Queue_Vx::open (size_t max_messages,
  86                             size_t max_message_length,
  87                             ACE_Notification_Strategy *ns)
  88 {
  89   ACE_TRACE ("ACE_Message_Queue_Vx::open");
  90   this->high_water_mark_ = 0;
  91   this->low_water_mark_  = 0;
  92   this->cur_bytes_ = 0;
  93   this->cur_length_ = 0;
  94   this->cur_count_ = 0;
  95   this->head_ = 0;
  96   this->notification_strategy_ = ns;
  97   this->max_messages_ = static_cast<int> (max_messages);
  98   this->max_message_length_ = static_cast<int> (max_message_length);
  99
 100   if (tail_)
 101     {
 102       // Had already created a msgQ, so delete it.
 103       close ();
 104       activate_i ();
 105     }
 106
 107   return (this->tail_ =
 108           reinterpret_cast<ACE_Message_Block *> (
 109               ::msgQCreate (max_messages_,
 110                             max_message_length_,
 111                             MSG_Q_FIFO))) == 0 ? -1 : 0;
 112 }
 113
 114 // Clean up the queue if we have not already done so!
 115
 116 int
 117 ACE_Message_Queue_Vx::close (void)
 118 {
 119   ACE_TRACE ("ACE_Message_Queue_Vx::close");
 120   // Don't lock, because we don't have a lock.  It shouldn't be
 121   // necessary, anyways.
 122
 123   this->deactivate_i ();
 124
 125   // Don't bother to free up the remaining message on the list,
 126   // because we don't have any way to iterate over what's in the
 127   // queue.
 128
 129   return ::msgQDelete (msgq ());
 130 }
 131
 132 bool
 133 ACE_Message_Queue_Vx::is_empty_i (void)
 134 {
 135   ACE_TRACE ("ACE_Message_Queue_Vx::is_empty_i");
 136   return ::msgQNumMsgs (msgq ()) == 0;
 137 }
 138
 139 bool
 140 ACE_Message_Queue_Vx::is_full_i (void)
 141 {
 142   ACE_TRACE ("ACE_Message_Queue_Vx::is_full_i");
 143   return ::msgQNumMsgs (msgq ()) >= max_messages_;
 144 }
 145
 146 size_t
 147 ACE_Message_Queue_Vx::high_water_mark (void)
 148 {
 149   ACE_TRACE ("ACE_Message_Queue_Vx::high_water_mark");
 150   ACE_NOTSUP_RETURN ((size_t) -1);
 151 }
 152
 153 void
 154 ACE_Message_Queue_Vx::high_water_mark (size_t)
 155 {
 156   ACE_TRACE ("ACE_Message_Queue_Vx::high_water_mark");
 157   ACE_NOTSUP;
 158 }
 159
 160 size_t
 161 ACE_Message_Queue_Vx::low_water_mark (void)
 162 {
 163   ACE_TRACE ("ACE_Message_Queue_Vx::low_water_mark");
 164   // Don't need to guard, because this is fixed.
 165
 166   ACE_NOTSUP_RETURN ((size_t) -1);
 167 }
 168
 169 void
 170 ACE_Message_Queue_Vx::low_water_mark (size_t)
 171 {
 172   ACE_TRACE ("ACE_Message_Queue_Vx::low_water_mark");
 173   ACE_NOTSUP;
 174 }
 175
 176 size_t
 177 ACE_Message_Queue_Vx::message_bytes (void)
 178 {
 179   ACE_TRACE ("ACE_Message_Queue_Vx::message_bytes");
 180   ACE_NOTSUP_RETURN ((size_t) -1);
 181 }
 182
 183 size_t
 184 ACE_Message_Queue_Vx::message_length (void)
 185 {
 186   ACE_TRACE ("ACE_Message_Queue_Vx::message_length");
 187   ACE_NOTSUP_RETURN ((size_t) -1);
 188 }
 189
 190 size_t
 191 ACE_Message_Queue_Vx::message_count (void)
 192 {
 193   ACE_TRACE ("ACE_Message_Queue_Vx::message_count");
 194   // Don't need to guard, because this is a system call.
 195
 196   return ::msgQNumMsgs (msgq ());
 197 }
 198
 199 void
 200 ACE_Message_Queue_Vx::message_bytes (size_t)
 201 {
 202   ACE_TRACE ("ACE_Message_Queue_Vx::message_bytes");
 203   ACE_NOTSUP;
 204 }
 205
 206 void
 207 ACE_Message_Queue_Vx::message_length (size_t)
 208 {
 209   ACE_TRACE ("ACE_Message_Queue_Vx::message_length");
 210   ACE_NOTSUP;
 211 }
 212
 213 int
 214 ACE_Message_Queue_Vx::signal_enqueue_waiters (void)
 215 {
 216   // No-op.
 217   return 0;
 218 }
 219
 220 int
 221 ACE_Message_Queue_Vx::signal_dequeue_waiters (void)
 222 {
 223   // No-op.
 224   return 0;
 225 }
 226
 227 int
 228 ACE_Message_Queue_Vx::enqueue_tail_i (ACE_Message_Block *new_item)
 229 {
 230   ACE_TRACE ("ACE_Message_Queue_Vx::enqueue_tail_i");
 231
 232   if (new_item == 0)
 233     return -1;
 234
 235   // Don't try to send a composite message!!!!  Only the first
 236   // block will be sent.
 237
 238   ++this->cur_count_;
 239
 240   // Always use this method to actually send a message on the queue.
 241   if (::msgQSend (msgq (),
 242                   new_item->rd_ptr (),
 243                   new_item->size (),
 244                   WAIT_FOREVER,
 245                   MSG_PRI_NORMAL) == OK)
 246     return ::msgQNumMsgs (msgq ());
 247   else
 248     return -1;
 249 }
 250
 251 int
 252 ACE_Message_Queue_Vx::enqueue_head_i (ACE_Message_Block *new_item)
 253 {
 254   ACE_TRACE ("ACE_Message_Queue_Vx::enqueue_head_i");
 255
 256   // Just delegate to enqueue_tail_i.
 257   return enqueue_tail_i (new_item);
 258 }
 259
 260 int
 261 ACE_Message_Queue_Vx::enqueue_i (ACE_Message_Block *new_item)
 262 {
 263   ACE_TRACE ("ACE_Message_Queue_Vx::enqueue_i");
 264
 265   if (new_item == 0)
 266     return -1;
 267
 268   if (this->head_ == 0)
 269     // Should always take this branch.
 270     return this->enqueue_head_i (new_item);
 271   else
 272     ACE_NOTSUP_RETURN (-1);
 273 }
 274
 275 int
 276 ACE_Message_Queue_Vx::enqueue_deadline_i (ACE_Message_Block *new_item)
 277 {
 278   ACE_TRACE ("ACE_Message_Queue_Vx::enqueue_deadline_i");
 279
 280   // Just delegate to enqueue_tail_i.
 281   return enqueue_tail_i (new_item);
 282 }
 283
 284 // Actually get the first ACE_Message_Block (no locking, so must be
 285 // called with locks held).  This method assumes that the queue has at
 286 // least one item in it when it is called.
 287
 288 int
 289 ACE_Message_Queue_Vx::dequeue_head_i (ACE_Message_Block *&first_item)
 290 {
 291   ACE_TRACE ("ACE_Message_Queue_Vx::dequeue_head_i");
 292
 293   // We don't allocate a new Message_Block:  the caller must provide
 294   // it, and must ensure that it is big enough (without chaining).
 295
 296   if (first_item == 0  ||  first_item->wr_ptr () == 0)
 297     return -1;
 298
 299   if (::msgQReceive (msgq (),
 300                      first_item->wr_ptr (),
 301                      first_item->size (),
 302                      WAIT_FOREVER) == ERROR)
 303     return -1;
 304   else
 305     return ::msgQNumMsgs (msgq ());
 306 }
 307
 308 int
 309 ACE_Message_Queue_Vx::dequeue_prio_i (ACE_Message_Block *& /*dequeued*/)
 310 {
 311   ACE_TRACE ("ACE_Message_Queue_Vx::dequeue_prio_i");
 312   ACE_NOTSUP_RETURN (-1);
 313 }
 314
 315 int
 316 ACE_Message_Queue_Vx::dequeue_tail_i (ACE_Message_Block *& /*dequeued*/)
 317 {
 318   ACE_TRACE ("ACE_Message_Queue_Vx::dequeue_tail_i");
 319   ACE_NOTSUP_RETURN (-1);
 320 }
 321
 322 int
 323 ACE_Message_Queue_Vx::dequeue_deadline_i (ACE_Message_Block *& /*dequeued*/)
 324 {
 325   ACE_TRACE ("ACE_Message_Queue_Vx::dequeue_deadline_i");
 326   ACE_NOTSUP_RETURN (-1);
 327 }
 328
 329 // Take a look at the first item without removing it.
 330
 331 int
 332 ACE_Message_Queue_Vx::wait_not_full_cond (ACE_Time_Value *)
 333 {
 334   // Always return here, and let the VxWorks message queue handle blocking.
 335   return 0;
 336 }
 337
 338 int
 339 ACE_Message_Queue_Vx::wait_not_empty_cond (ACE_Time_Value *)
 340 {
 341   // Always return here, and let the VxWorks message queue handle blocking.
 342   return 0;
 343 }
 344
 345 #endif /* ACE_VXWORKS */
 346
 347 ACE_END_VERSIONED_NAMESPACE_DECL