ACE/ace/Message_Queue_NT.h

   1 // -*- C++ -*-
   2
   3 //=============================================================================
   4 /**
   5  *  @file    Message_Queue_NT.h
   6  *
   7  *  @author Douglas C. Schmidt <d.schmidt@vanderbilt.edu>
   8  */
   9 //=============================================================================
  10
  11 #ifndef ACE_MESSAGE_QUEUE_NT_H
  12 #define ACE_MESSAGE_QUEUE_NT_H
  13 #include /**/ "ace/pre.h"
  14
  15 #include "ace/Message_Queue.h"
  16
  17 #if !defined (ACE_LACKS_PRAGMA_ONCE)
  18 # pragma once
  19 #endif /* ACE_LACKS_PRAGMA_ONCE */
  20
  21 #if defined (ACE_HAS_WIN32_OVERLAPPED_IO)
  22 # include "ace/Synch_Traits.h"   /* Needed in ACE_Message_Queue_NT */
  23 # include "ace/Thread_Mutex.h"   /* Needed in ACE_Message_Queue_NT */
  24 #endif
  25
  26 ACE_BEGIN_VERSIONED_NAMESPACE_DECL
  27
  28 #if defined (ACE_HAS_WIN32_OVERLAPPED_IO)
  29 /**
  30  * @class ACE_Message_Queue_NT
  31  *
  32  * @brief Message Queue implementation using IO completion port on NT.
  33  *
  34  * Implementation of a strip-downed ACE_Message_Queue using NT's
  35  * IO completion port mechanism.
  36  * @note *Many* ACE_Message_Queue features are not supported with
  37  * this implementation, including:
  38  * * <open> method have different signatures.
  39  * * <dequeue_head> *requires* that the ACE_Message_Block
  40  * pointer argument point to an ACE_Message_Block that was
  41  * allocated by the caller.
  42  * * <peek_dequeue_head>.
  43  * * <ACE_Message_Queue_Iterators>.
  44  * * No flow control.
  45  */
  46 class ACE_Export ACE_Message_Queue_NT : public ACE_Message_Queue_Base
  47 {
  48 public:
  49   ACE_Message_Queue_NT (DWORD max_threads = ACE_Message_Queue_Base::DEFAULT_HWM);
  50
  51   /**
  52    * Initialize the Message Queue by creating a new NT I/O completion
  53    * port.  The first arguemnt specifies the number of threads
  54    * released by the MQ that are allowed to run concurrently.  Return
  55    * 0 when succeeds, -1 otherwise.
  56    */
  57   virtual int open (DWORD max_threads = ACE_Message_Queue_Base::DEFAULT_HWM);
  58
  59   /// Close down the underlying I/O completion port.  You need to
  60   /// re-open the MQ after this function is executed.
  61   virtual int close (void);
  62
  63   /// Close down the message queue and release all resources.
  64   virtual ~ACE_Message_Queue_NT (void);
  65
  66   // = Enqueue and dequeue methods.
  67
  68   /**
  69    * Enqueue an ACE_Message_Block * at the end of the queue.
  70    * Returns -1 on failure, else the number of items still on the
  71    * queue.
  72    */
  73   virtual int enqueue_tail (ACE_Message_Block *new_item,
  74                             ACE_Time_Value *timeout = 0);
  75   virtual int enqueue (ACE_Message_Block *new_item,
  76                        ACE_Time_Value *timeout = 0);
  77
  78   /**
  79    * Dequeue and return the ACE_Message_Block * at the head of the
  80    * queue.  Returns -1 on failure, else the number of items still on
  81    * the queue.
  82    */
  83   virtual int dequeue_head (ACE_Message_Block *&first_item,
  84                             ACE_Time_Value *timeout = 0);
  85   virtual int dequeue (ACE_Message_Block *&first_item,
  86                        ACE_Time_Value *timeout = 0);
  87
  88   // = Check if queue is full/empty.
  89   /**
  90    * Always return false.
  91    */
  92
  93   virtual bool is_full (void);
  94   /**
  95    * True if queue is empty, else false.  Notice the return value is
  96    * only transient.
  97    */
  98   virtual bool is_empty (void);
  99
 100   // = Queue statistic methods (transient.)
 101   /**
 102    * Number of total bytes on the queue, i.e., sum of the message
 103    * block sizes.
 104    */
 105   virtual size_t message_bytes (void);
 106
 107   /**
 108    * Number of total length on the queue, i.e., sum of the message
 109    * block lengths.
 110    */
 111   virtual size_t message_length (void);
 112
 113   /**
 114    * Number of total messages on the queue.
 115    */
 116   virtual size_t message_count (void);
 117
 118   // = Manual changes to these stats (used when queued message blocks
 119   // change size or lengths).
 120   /**
 121    * New value of the number of total bytes on the queue, i.e., sum of
 122    * the message block sizes.
 123    */
 124   virtual void message_bytes (size_t new_size);
 125
 126   /**
 127    * New value of the number of total length on the queue, i.e., sum
 128    * of the message block lengths.
 129    */
 130   virtual void message_length (size_t new_length);
 131
 132   /// Get the max concurrent thread number.
 133   virtual DWORD max_threads (void);
 134
 135   // = Activation control methods.
 136
 137   /**
 138    * Deactivate the queue and wake up all threads waiting on the queue
 139    * so they can continue.  No messages are removed from the queue,
 140    * however.  Any other operations called until the queue is
 141    * activated again will immediately return -1 with @c errno
 142    * ESHUTDOWN.
 143    *
 144    * @retval  The queue's state before this call.
 145    */
 146   virtual int deactivate (void);
 147
 148   /**
 149    * Reactivate the queue so that threads can enqueue and dequeue
 150    * messages again.  Returns the state of the queue before the call.
 151    */
 152   virtual int activate (void);
 153
 154   /**
 155    * Pulse the queue to wake up any waiting threads.  Changes the
 156    * queue state to PULSED; future enqueue/dequeue operations proceed
 157    * as in ACTIVATED state.
 158    *
 159    * @retval  The queue's state before this call.
 160    */
 161   virtual int pulse (void);
 162
 163   /// Returns true if the state of the queue is <DEACTIVATED>,
 164   /// but false if the queue's is <ACTIVATED> or <PULSED>.
 165   virtual int deactivated (void);
 166
 167   // = Not currently implemented...
 168   int peek_dequeue_head (ACE_Message_Block *&first_item,
 169                          ACE_Time_Value *timeout = 0);
 170   ACE_Notification_Strategy *notification_strategy (void);
 171   void notification_strategy (ACE_Notification_Strategy *s);
 172
 173   // = Notification hook.
 174
 175   /// Dump the state of an object.
 176   virtual void dump (void) const;
 177
 178   /// Get the handle to the underlying completion port.
 179   virtual ACE_HANDLE completion_port (void);
 180
 181   /// Declare the dynamic allocation hooks.
 182   ACE_ALLOC_HOOK_DECLARE;
 183
 184 private:
 185
 186   // Disallow copying and assignment.
 187   ACE_Message_Queue_NT (const ACE_Message_Queue_NT &);
 188   void operator= (const ACE_Message_Queue_NT &);
 189
 190 private:
 191   // = Internal states.
 192
 193   /// Maximum threads that can be released (and run) concurrently.
 194   DWORD max_cthrs_;
 195
 196   /// Current number of threads waiting to dequeue messages.
 197   DWORD cur_thrs_;
 198
 199   /// Current number of bytes in queue.
 200   size_t cur_bytes_;
 201
 202   /// Current length of messages in queue.
 203   size_t cur_length_;
 204
 205   /// Current number of messages in the queue.
 206   size_t cur_count_;
 207
 208   /**
 209    * Synchronizer.  This should really be an ACE_Recursive_Thread_Mutex
 210    * but since this class is only supported on NT, it's okay to use
 211    * ACE_Thread_Mutex here.
 212    */
 213   ACE_SYNCH_MUTEX lock_;
 214
 215   /// Underlying NT IoCompletionPort.
 216   ACE_HANDLE completion_port_;
 217
 218 };
 219 #endif /* ACE_HAS_WIN32_OVERLAPPED_IO */
 220
 221 ACE_END_VERSIONED_NAMESPACE_DECL
 222
 223 #if defined (__ACE_INLINE__)
 224 #include "ace/Message_Queue_NT.inl"
 225 #endif /* __ACE_INLINE__ */
 226
 227 #include /**/ "ace/post.h"
 228 #endif /* ACE_MESSAGE_QUEUE_NT_H */