include/wx/msgqueue.h

   1 ///////////////////////////////////////////////////////////////////////////////
   2 // Name:        wx/thread.h
   3 // Purpose:     Message queues for inter-thread communication
   4 // Author:      Evgeniy Tarassov
   5 // Created:     2007-10-31
   6 // RCS-ID:      $Id$
   7 // Copyright:   (C) 2007 TT-Solutions SARL
   8 // Licence:     wxWindows licence
   9 ///////////////////////////////////////////////////////////////////////////////
  10
  11 #ifndef _WX_MSGQUEUE_H_
  12 #define _WX_MSGQUEUE_H_
  13
  14 // ----------------------------------------------------------------------------
  15 // headers
  16 // ----------------------------------------------------------------------------
  17
  18 #include "wx/thread.h"
  19
  20 #if wxUSE_THREADS
  21
  22 #include "wx/stopwatch.h"
  23
  24 #include "wx/beforestd.h"
  25 #include <queue>
  26 #include "wx/afterstd.h"
  27
  28 enum wxMessageQueueError
  29 {
  30     wxMSGQUEUE_NO_ERROR = 0, // operation completed successfully
  31     wxMSGQUEUE_TIMEOUT,      // no messages received before timeout expired
  32     wxMSGQUEUE_MISC_ERROR    // some unexpected (and fatal) error has occurred
  33 };
  34
  35 // ---------------------------------------------------------------------------
  36 // Message queue allows passing message between threads.
  37 //
  38 // This class is typically used for communicating between the main and worker
  39 // threads. The main thread calls Post() and the worker thread calls Receive().
  40 //
  41 // For this class a message is an object of arbitrary type T. Notice that
  42 // typically there must be some special message indicating that the thread
  43 // should terminate as there is no other way to gracefully shutdown a thread
  44 // waiting on the message queue.
  45 // ---------------------------------------------------------------------------
  46 template <typename T>
  47 class wxMessageQueue
  48 {
  49 public:
  50     // The type of the messages transported by this queue
  51     typedef T Message;
  52
  53     // Default ctor creates an initially empty queue
  54     wxMessageQueue()
  55        : m_conditionNotEmpty(m_mutex)
  56     {
  57     }
  58
  59     // Add a message to this queue and signal the threads waiting for messages.
  60     //
  61     // This method is safe to call from multiple threads in parallel.
  62     wxMessageQueueError Post(const Message& msg)
  63     {
  64         wxMutexLocker locker(m_mutex);
  65
  66         wxCHECK( locker.IsOk(), wxMSGQUEUE_MISC_ERROR );
  67
  68         m_messages.push(msg);
  69
  70         m_conditionNotEmpty.Signal();
  71
  72         return wxMSGQUEUE_NO_ERROR;
  73     }
  74
  75     // Remove all messages from the queue.
  76     //
  77     // This method is meant to be called from the same thread(s) that call
  78     // Post() to discard any still pending requests if they became unnecessary.
  79     wxMessageQueueError Clear()
  80     {
  81         wxCHECK( IsOk(), wxMSGQUEUE_MISC_ERROR );
  82
  83         wxMutexLocker locker(m_mutex);
  84
  85         std::queue<T> empty;
  86         std::swap(m_messages, empty);
  87
  88         return wxMSGQUEUE_NO_ERROR;
  89     }
  90
  91     // Wait no more than timeout milliseconds until a message becomes available.
  92     //
  93     // Setting timeout to 0 is equivalent to an infinite timeout. See Receive().
  94     wxMessageQueueError ReceiveTimeout(long timeout, T& msg)
  95     {
  96         wxCHECK( IsOk(), wxMSGQUEUE_MISC_ERROR );
  97
  98         wxMutexLocker locker(m_mutex);
  99
 100         wxCHECK( locker.IsOk(), wxMSGQUEUE_MISC_ERROR );
 101
 102         const wxMilliClock_t waitUntil = wxGetLocalTimeMillis() + timeout;
 103         while ( m_messages.empty() )
 104         {
 105             wxCondError result = m_conditionNotEmpty.WaitTimeout(timeout);
 106
 107             if ( result == wxCOND_NO_ERROR )
 108                 continue;
 109
 110             wxCHECK( result == wxCOND_TIMEOUT, wxMSGQUEUE_MISC_ERROR );
 111
 112             const wxMilliClock_t now = wxGetLocalTimeMillis();
 113
 114             if ( now >= waitUntil )
 115                 return wxMSGQUEUE_TIMEOUT;
 116
 117             timeout = (waitUntil - now).ToLong();
 118             wxASSERT(timeout > 0);
 119         }
 120
 121         msg = m_messages.front();
 122         m_messages.pop();
 123
 124         return wxMSGQUEUE_NO_ERROR;
 125     }
 126
 127     // Same as ReceiveTimeout() but waits for as long as it takes for a message
 128     // to become available (so it can't return wxMSGQUEUE_TIMEOUT)
 129     wxMessageQueueError Receive(T& msg)
 130     {
 131         wxCHECK( IsOk(), wxMSGQUEUE_MISC_ERROR );
 132
 133         wxMutexLocker locker(m_mutex);
 134
 135         wxCHECK( locker.IsOk(), wxMSGQUEUE_MISC_ERROR );
 136
 137         while ( m_messages.empty() )
 138         {
 139             wxCondError result = m_conditionNotEmpty.Wait();
 140
 141             wxCHECK( result == wxCOND_NO_ERROR, wxMSGQUEUE_MISC_ERROR );
 142         }
 143
 144         msg = m_messages.front();
 145         m_messages.pop();
 146
 147         return wxMSGQUEUE_NO_ERROR;
 148     }
 149
 150     // Return false only if there was a fatal error in ctor
 151     bool IsOk() const
 152     {
 153         return m_conditionNotEmpty.IsOk();
 154     }
 155
 156 private:
 157     // Disable copy ctor and assignment operator
 158     wxMessageQueue(const wxMessageQueue<T>& rhs);
 159     wxMessageQueue<T>& operator=(const wxMessageQueue<T>& rhs);
 160
 161     mutable wxMutex m_mutex;
 162     wxCondition     m_conditionNotEmpty;
 163
 164     std::queue<T>   m_messages;
 165 };
 166
 167 #endif // wxUSE_THREADS
 168
 169 #endif // _WX_MSGQUEUE_H_