dep/ACE_wrappers/ace/Message_Queue_NT.cpp

   1 // $Id: Message_Queue_NT.cpp 80826 2008-03-04 14:51:23Z wotte $
   2
   3 #include "ace/Message_Queue_NT.h"
   4 #include "ace/Log_Msg.h"
   5
   6 #if !defined (__ACE_INLINE__)
   7 #include "ace/Message_Queue_NT.inl"
   8 #endif /* __ACE_INLINE__ */
   9
  10 ACE_RCSID (ace,
  11            Message_Queue_NT,
  12            "$Id: Message_Queue_NT.cpp 80826 2008-03-04 14:51:23Z wotte $")
  13
  14
  15 ACE_BEGIN_VERSIONED_NAMESPACE_DECL
  16
  17 #if defined (ACE_HAS_WIN32_OVERLAPPED_IO)
  18
  19 ACE_Message_Queue_NT::ACE_Message_Queue_NT (DWORD max_threads)
  20   : max_cthrs_ (max_threads),
  21     cur_thrs_ (0),
  22     cur_bytes_ (0),
  23     cur_length_ (0),
  24     cur_count_ (0),
  25     completion_port_ (ACE_INVALID_HANDLE)
  26 {
  27   ACE_TRACE ("ACE_Message_Queue_NT::ACE_Message_Queue_NT");
  28   this->open (max_threads);
  29 }
  30
  31 int
  32 ACE_Message_Queue_NT::open (DWORD max_threads)
  33 {
  34   ACE_TRACE ("ACE_Message_Queue_NT::open");
  35   this->max_cthrs_ = max_threads;
  36   this->completion_port_ = ::CreateIoCompletionPort (ACE_INVALID_HANDLE,
  37                                                      0,
  38                                                      ACE_Message_Queue_Base::ACTIVATED,
  39                                                      max_threads);
  40   return (this->completion_port_ == 0 ? -1 : 0);
  41 }
  42
  43 int
  44 ACE_Message_Queue_NT::close (void)
  45 {
  46   ACE_TRACE ("ACE_Message_Queue_NT::close");
  47   ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->lock_, -1);
  48   this->deactivate ();
  49   return (::CloseHandle (this->completion_port_) ? 0 : -1 );
  50 }
  51
  52 ACE_Message_Queue_NT::~ACE_Message_Queue_NT (void)
  53 {
  54   ACE_TRACE ("ACE_Message_Queue_NT::~ACE_Message_Queue_NT");
  55   this->close ();
  56 }
  57
  58 int
  59 ACE_Message_Queue_NT::enqueue (ACE_Message_Block *new_item,
  60                                ACE_Time_Value *)
  61 {
  62   ACE_TRACE ("ACE_Message_Queue_NT::enqueue");
  63   ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->lock_, -1);
  64   if (this->state_ != ACE_Message_Queue_Base::DEACTIVATED)
  65     {
  66       size_t const msize = new_item->total_size ();
  67       size_t const mlength = new_item->total_length ();
  68       // Note - we send ACTIVATED in the 3rd arg to tell the completion
  69       // routine it's _NOT_ being woken up because of deactivate().
  70       ULONG_PTR state_to_post;
  71       state_to_post = ACE_Message_Queue_Base::ACTIVATED;
  72       if (::PostQueuedCompletionStatus (this->completion_port_,
  73                                         static_cast<DWORD> (msize),
  74                                         state_to_post,
  75                                         reinterpret_cast<LPOVERLAPPED> (new_item)))
  76         {
  77           // Update the states once I succeed.
  78           this->cur_bytes_ += msize;
  79           this->cur_length_ += mlength;
  80           return ACE_Utils::truncate_cast<int> (++this->cur_count_);
  81         }
  82     }
  83   else
  84     errno = ESHUTDOWN;
  85
  86   // Fail to enqueue the message.
  87   return -1;
  88 }
  89
  90 int
  91 ACE_Message_Queue_NT::dequeue (ACE_Message_Block *&first_item,
  92                                ACE_Time_Value *timeout)
  93 {
  94   ACE_TRACE ("ACE_Message_Queue_NT::dequeue_head");
  95
  96   {
  97     ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->lock_, -1);
  98
  99     // Make sure the MQ is not deactivated before proceeding.
 100     if (this->state_ == ACE_Message_Queue_Base::DEACTIVATED)
 101       {
 102         errno = ESHUTDOWN;      // Operation on deactivated MQ not allowed.
 103         return -1;
 104       }
 105     else
 106       ++this->cur_thrs_;        // Increase the waiting thread count.
 107   }
 108
 109   ULONG_PTR queue_state;
 110   DWORD msize;
 111   // Get a message from the completion port.
 112   int retv = ::GetQueuedCompletionStatus (this->completion_port_,
 113                                           &msize,
 114                                           &queue_state,
 115                                           reinterpret_cast<LPOVERLAPPED *> (&first_item),
 116                                           (timeout == 0 ? INFINITE : timeout->msec ()));
 117   {
 118     ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->lock_, -1);
 119     --this->cur_thrs_;          // Decrease waiting thread count.
 120     if (retv)
 121       {
 122         if (queue_state == ACE_Message_Queue_Base::ACTIVATED)
 123           {                     // Really get a valid MB from the queue.
 124             --this->cur_count_;
 125             this->cur_bytes_ -= msize;
 126             this->cur_length_ -= first_item->total_length ();
 127             return ACE_Utils::truncate_cast<int> (this->cur_count_);
 128           }
 129         else                    // Woken up by deactivate () or pulse ().
 130             errno = ESHUTDOWN;
 131       }
 132   }
 133   return -1;
 134 }
 135
 136 int
 137 ACE_Message_Queue_NT::deactivate (void)
 138 {
 139   ACE_TRACE ("ACE_Message_Queue_NT::deactivate");
 140   ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->lock_, -1);
 141
 142   int const previous_state = this->state_;
 143   if (previous_state != ACE_Message_Queue_Base::DEACTIVATED)
 144     {
 145       this->state_ = ACE_Message_Queue_Base::DEACTIVATED;
 146
 147       // Get the number of shutdown messages necessary to wake up all
 148       // waiting threads.
 149       DWORD cntr =
 150         this->cur_thrs_ - static_cast<DWORD> (this->cur_count_);
 151       while (cntr-- > 0)
 152         ::PostQueuedCompletionStatus (this->completion_port_,
 153                                       0,
 154                                       this->state_,
 155                                       0);
 156     }
 157   return previous_state;
 158 }
 159
 160 int
 161 ACE_Message_Queue_NT::activate (void)
 162 {
 163   ACE_TRACE ("ACE_Message_Queue_NT::activate");
 164   ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->lock_, -1);
 165   int const previous_status = this->state_;
 166   this->state_ = ACE_Message_Queue_Base::ACTIVATED;
 167   return previous_status;
 168 }
 169
 170 int
 171 ACE_Message_Queue_NT::pulse (void)
 172 {
 173   ACE_TRACE ("ACE_Message_Queue_NT::pulse");
 174   ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->lock_, -1);
 175
 176   int const previous_state = this->state_;
 177   if (previous_state != ACE_Message_Queue_Base::DEACTIVATED)
 178     {
 179       this->state_ = ACE_Message_Queue_Base::PULSED;
 180
 181       // Get the number of shutdown messages necessary to wake up all
 182       // waiting threads.
 183
 184       DWORD cntr =
 185         this->cur_thrs_ - static_cast<DWORD> (this->cur_count_);
 186       while (cntr-- > 0)
 187         ::PostQueuedCompletionStatus (this->completion_port_,
 188                                       0,
 189                                       this->state_,
 190                                       0);
 191     }
 192   return previous_state;
 193 }
 194
 195 void
 196 ACE_Message_Queue_NT::dump (void) const
 197 {
 198 #if defined (ACE_HAS_DUMP)
 199   ACE_TRACE ("ACE_Message_Queue_NT::dump");
 200
 201   ACE_DEBUG ((LM_DEBUG, ACE_BEGIN_DUMP, this));
 202   switch (this->state_)
 203     {
 204     case ACE_Message_Queue_Base::ACTIVATED:
 205       ACE_DEBUG ((LM_DEBUG,
 206                   ACE_TEXT ("state = ACTIVATED\n")));
 207       break;
 208     case ACE_Message_Queue_Base::DEACTIVATED:
 209       ACE_DEBUG ((LM_DEBUG,
 210                   ACE_TEXT ("state = DEACTIVATED\n")));
 211       break;
 212     case ACE_Message_Queue_Base::PULSED:
 213       ACE_DEBUG ((LM_DEBUG,
 214                   ACE_TEXT ("state = PULSED\n")));
 215       break;
 216     }
 217
 218   ACE_DEBUG ((LM_DEBUG,
 219               ACE_TEXT ("max_cthrs_ = %d\n")
 220               ACE_TEXT ("cur_thrs_ = %d\n")
 221               ACE_TEXT ("cur_bytes = %d\n")
 222               ACE_TEXT ("cur_length = %d\n")
 223               ACE_TEXT ("cur_count = %d\n")
 224               ACE_TEXT ("completion_port_ = %x\n"),
 225               this->max_cthrs_,
 226               this->cur_thrs_,
 227               this->cur_bytes_,
 228               this->cur_length_,
 229               this->cur_count_,
 230               this->completion_port_));
 231   ACE_DEBUG ((LM_DEBUG, ACE_END_DUMP));
 232 #endif /* ACE_HAS_DUMP */
 233 }
 234
 235 #endif /* ACE_HAS_WIN32_OVERLAPPED_IO */
 236
 237 ACE_END_VERSIONED_NAMESPACE_DECL