include/wx/msgqueue.h

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