ACE_Message_Queue_NT changes.

1 files changed, 205 insertions, 0 deletions

diff --git a/ace/Message_Queue.h b/ace/Message_Queue.h
index e9766f1ac95..76f7b05467c 100644
--- a/ace/Message_Queue.h
+++ b/ace/Message_Queue.h
@@ -52,6 +52,78 @@ public:
     WAS_INACTIVE = 2
     // Message queue was inactive before activate() or deactivate().
   };
+
+  ACE_Message_Queue_Base (void);
+
+  virtual int close (void) = 0;
+  // Close down the message queue and release all resources.
+
+  virtual ~ACE_Message_Queue_Base (void) = 0;
+  // Close down the message queue and release all resources.
+
+  // = Enqueue and dequeue methods.
+
+  // For the following enqueue and dequeue methods if <timeout> == 0,
+  // the caller will block until action is possible, else will wait
+  // until the absolute time specified in *<timeout> elapses).  These
+  // calls will return, however, when queue is closed, deactivated,
+  // when a signal occurs, or if the time specified in timeout
+  // elapses, (in which case errno = EWOULDBLOCK).
+
+  virtual int enqueue (ACE_Message_Block *new_item,
+                       ACE_Time_Value *timeout = 0) = 0;
+  // Enqueue a <ACE_Message_Block *> into the tail of the queue.
+  // Return -1 on failure, number of items in queue otherwise.
+
+  virtual int dequeue (ACE_Message_Block *&first_item,
+                       ACE_Time_Value *timeout = 0) = 0;
+  // Dequeue and return the <ACE_Message_Block *> at the head of the
+  // queue.  Returns -1 on failure, else the number of items still on
+  // the queue.
+
+  // = Check if queue is full/empty.
+  virtual int is_full (void) = 0;
+  // True if queue is full, else false.
+  virtual int is_empty (void) = 0;
+  // True if queue is empty, else false.
+
+  // = Queue statistic methods.
+  virtual size_t message_bytes (void) = 0;
+  // Number of total bytes on the queue.
+  virtual size_t message_count (void) = 0;
+  // Number of total messages on the queue.
+
+  // = Activation control methods.
+
+  virtual int deactivate (void) = 0;
+  // Deactivate the queue and wakeup all threads waiting on the queue
+  // so they can continue.  No messages are removed from the queue,
+  // however.  Any other operations called until the queue is
+  // activated again will immediately return -1 with <errno> ==
+  // ESHUTDOWN.  Returns WAS_INACTIVE if queue was inactive before the
+  // call and WAS_ACTIVE if queue was active before the call.
+
+  virtual int activate (void) = 0;
+  // Reactivate the queue so that threads can enqueue and dequeue
+  // messages again.  Returns WAS_INACTIVE if queue was inactive
+  // before the call and WAS_ACTIVE if queue was active before the
+  // call.
+
+  virtual int deactivated (void) = 0;
+  // Returns true if <deactivated_> is enabled.
+
+  // = Notification hook.
+
+  virtual void dump (void) const = 0;
+  // Dump the state of an object.
+
+  ACE_ALLOC_HOOK_DECLARE;
+  // Declare the dynamic allocation hooks.
+
+private:
+  // = Disallow these operations.
+  ACE_UNIMPLEMENTED_FUNC (void operator= (const ACE_Message_Queue_Base &))
+  ACE_UNIMPLEMENTED_FUNC (ACE_Message_Queue_Base (const ACE_Message_Queue_Base &))
 };
 
 // Include the templates here.
@@ -198,6 +270,139 @@ private:
 };
 #endif /* VXWORKS */
 
+#if defined (ACE_WIN32) && (ACE_HAS_WINNT4 != 0)
+class ACE_Export ACE_Message_Queue_NT : public ACE_Message_Queue_Base
+{
+  // = TITLE
+  //     Message Queue implementation using IO completion port on NT.
+  //
+  // = DESCRIPTION
+  //     Implementation of a strip-downed ACE_Message_Queue using NT's
+  //     IO completion port mechanism.
+  //
+  //     NOTE: *Many* ACE_Message_Queue features are not supported with
+  //     this implementation, including:
+  //     * open method have different signatures.
+  //     * dequeue_head () *requires* that the ACE_Message_Block
+  //       pointer argument point to an ACE_Message_Block that was
+  //       allocated by the caller.
+  //     * peek_dequeue_head ().
+  //     * ACE_Message_Queue_Iterators.
+  //     * No flow control.
+  //     * Message_Block chains.  The continuation field of ACE_Message_Block
+  //     *   is ignored; only the first block of a fragment chain is
+  //     *   recognized.
+public:
+  // = Initialization and termination methods.
+  ACE_Message_Queue_NT (size_t max_threads = ACE_Message_Queue_Base::DEFAULT_HWM);
+
+  virtual int open (size_t max_threads = ACE_Message_Queue_Base::DEFAULT_HWM);
+  // Initialize the Message Queue by creating a new NT I/O completion
+  // port.  The first arguemnt specifies the number of threads
+  // released by the MQ that are allowed to run concurrently.  Return
+  // 0 when succeeds, -1 otherwise.
+
+  virtual int close (void);
+  // Close down the underlying I/O completion port.  You need to
+  // re-open the MQ after this function is executed.
+
+  virtual ~ACE_Message_Queue_NT (void);
+  // Close down the message queue and release all resources.
+
+  // = Enqueue and dequeue methods.
+
+  virtual int enqueue (ACE_Message_Block *new_item,
+                       ACE_Time_Value *timeout = 0);
+  // Enqueue an <ACE_Message_Block *> at the end of the queue.
+  // Returns -1 on failure, else the number of items still on the
+  // queue.
+
+  virtual int dequeue (ACE_Message_Block *&first_item,
+                       ACE_Time_Value *timeout = 0);
+  // Dequeue and return the <ACE_Message_Block *> at the head of the
+  // queue.  Returns -1 on failure, else the number of items still on
+  // the queue.
+
+  // = Check if queue is full/empty.
+  virtual int is_full (void);
+  // Always return false.
+  virtual int is_empty (void);
+  // True if queue is empty, else false.  Notice the return value is
+  // only transient.
+
+  // = Queue statistic methods (transient.)
+  virtual size_t message_bytes (void);
+  // Number of total bytes on the queue.
+  virtual size_t message_count (void);
+  // Number of total messages on the queue.
+
+  virtual size_t max_threads (void);
+  // Get the max concurrent thread number.
+
+  // = Activation control methods.
+
+  virtual int deactivate (void);
+  // Deactivate the queue and wakeup all threads waiting on the queue
+  // so they can continue.  Messages already in the queue get removed.
+  // If there are more messages in the queue than there are threads
+  // waiting on the queue, the left over messages will not be removed.
+  // Any other enqueue/dequeue operations called until the queue is
+  // activated again will immediately return -1 with <errno> ==
+  // ESHUTDOWN.  Returns WAS_INACTIVE if queue was inactive before the
+  // call and WAS_ACTIVE if queue was active before the call.
+
+  virtual int activate (void);
+  // Reactivate the queue so that threads can enqueue and dequeue
+  // messages again.  Returns WAS_INACTIVE if queue was inactive
+  // before the call and WAS_ACTIVE if queue was active before the
+  // call.
+
+  virtual int deactivated (void);
+  // Returns true if <deactivated_> is enabled.
+
+  // = Notification hook.
+
+  virtual void dump (void) const;
+  // Dump the state of an object.
+
+  virtual ACE_HANDLE completion_port (void);
+  // Get the handle to the underlying completion port.
+
+  ACE_ALLOC_HOOK_DECLARE;
+  // Declare the dynamic allocation hooks.
+
+private:
+  // = Internal states.
+
+  size_t max_cthrs_;
+  // Maximum threads that can be released (and run) concurrently.
+
+  size_t cur_thrs_;
+  // Current number of threads waiting to dequeue messages.
+
+  size_t cur_bytes_;
+  // Current number of bytes in queue.
+
+  size_t cur_count_;
+  // Current number of messages in the queue.
+
+  ACE_Thread_Mutex lock_;
+  // Synchronizer.  This should really be an ACE_Recursive_Thread_Mutex
+  // but since this class is only supported on NT, it's okay to use
+  // ACE_Thread_Mutex here.
+
+  int deactivated_;
+  // Indicates that the queue is inactive.
+
+  ACE_HANDLE completion_port_;
+  // Underlying NT IoCompletionPort.
+
+  // = Disallow these operations.
+  ACE_UNIMPLEMENTED_FUNC (void operator= (const ACE_Message_Queue_NT &))
+  ACE_UNIMPLEMENTED_FUNC (ACE_Message_Queue_NT (const ACE_Message_Queue_NT &))
+};
+#endif /* ACE_WIN32 && ACE_HAS_WINNT4 != 0 */
+
 // This must go here to avoid problems with circular includes.
 #include "ace/Strategies.h"