wtf/MessageQueue.h

   1 /*
   2  * Copyright (C) 2008 Apple Inc. All rights reserved.
   3  * Copyright (C) 2009 Google Inc. All rights reserved.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions
   7  * are met:
   8  *
   9  * 1.  Redistributions of source code must retain the above copyright
  10  *     notice, this list of conditions and the following disclaimer.
  11  * 2.  Redistributions in binary form must reproduce the above copyright
  12  *     notice, this list of conditions and the following disclaimer in the
  13  *     documentation and/or other materials provided with the distribution.
  14  * 3.  Neither the name of Apple Computer, Inc. ("Apple") nor the names of
  15  *     its contributors may be used to endorse or promote products derived
  16  *     from this software without specific prior written permission.
  17  *
  18  * THIS SOFTWARE IS PROVIDED BY APPLE AND ITS CONTRIBUTORS "AS IS" AND ANY
  19  * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  20  * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  21  * DISCLAIMED. IN NO EVENT SHALL APPLE OR ITS CONTRIBUTORS BE LIABLE FOR ANY
  22  * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
  23  * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  24  * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  25  * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  26  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
  27  * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  28  */
  29
  30 #ifndef MessageQueue_h
  31 #define MessageQueue_h
  32
  33 #include <limits>
  34 #include <wtf/Assertions.h>
  35 #include <wtf/Deque.h>
  36 #include <wtf/Noncopyable.h>
  37 #include <wtf/Threading.h>
  38
  39 namespace WTF {
  40
  41     enum MessageQueueWaitResult {
  42         MessageQueueTerminated,       // Queue was destroyed while waiting for message.
  43         MessageQueueTimeout,          // Timeout was specified and it expired.
  44         MessageQueueMessageReceived,  // A message was successfully received and returned.
  45     };
  46
  47     // The queue takes ownership of messages and transfer it to the new owner
  48     // when messages are fetched from the queue.
  49     // Essentially, MessageQueue acts as a queue of OwnPtr<DataType>.
  50     template<typename DataType>
  51     class MessageQueue : public Noncopyable {
  52     public:
  53         MessageQueue() : m_killed(false) { }
  54         ~MessageQueue();
  55
  56         void append(PassOwnPtr<DataType>);
  57         bool appendAndCheckEmpty(PassOwnPtr<DataType>);
  58         void prepend(PassOwnPtr<DataType>);
  59
  60         PassOwnPtr<DataType> waitForMessage();
  61         PassOwnPtr<DataType> tryGetMessage();
  62         template<typename Predicate>
  63         PassOwnPtr<DataType> waitForMessageFilteredWithTimeout(MessageQueueWaitResult&, Predicate&, double absoluteTime);
  64
  65         template<typename Predicate>
  66         void removeIf(Predicate&);
  67
  68         void kill();
  69         bool killed() const;
  70
  71         // The result of isEmpty() is only valid if no other thread is manipulating the queue at the same time.
  72         bool isEmpty();
  73
  74         static double infiniteTime() { return std::numeric_limits<double>::max(); }
  75
  76     private:
  77         static bool alwaysTruePredicate(DataType*) { return true; }
  78
  79         mutable Mutex m_mutex;
  80         ThreadCondition m_condition;
  81         Deque<DataType*> m_queue;
  82         bool m_killed;
  83     };
  84
  85     template<typename DataType>
  86     MessageQueue<DataType>::~MessageQueue()
  87     {
  88         deleteAllValues(m_queue);
  89     }
  90
  91     template<typename DataType>
  92     inline void MessageQueue<DataType>::append(PassOwnPtr<DataType> message)
  93     {
  94         MutexLocker lock(m_mutex);
  95         m_queue.append(message.release());
  96         m_condition.signal();
  97     }
  98
  99     // Returns true if the queue was empty before the item was added.
 100     template<typename DataType>
 101     inline bool MessageQueue<DataType>::appendAndCheckEmpty(PassOwnPtr<DataType> message)
 102     {
 103         MutexLocker lock(m_mutex);
 104         bool wasEmpty = m_queue.isEmpty();
 105         m_queue.append(message.release());
 106         m_condition.signal();
 107         return wasEmpty;
 108     }
 109
 110     template<typename DataType>
 111     inline void MessageQueue<DataType>::prepend(PassOwnPtr<DataType> message)
 112     {
 113         MutexLocker lock(m_mutex);
 114         m_queue.prepend(message.release());
 115         m_condition.signal();
 116     }
 117
 118     template<typename DataType>
 119     inline PassOwnPtr<DataType> MessageQueue<DataType>::waitForMessage()
 120     {
 121         MessageQueueWaitResult exitReason;
 122         PassOwnPtr<DataType> result = waitForMessageFilteredWithTimeout(exitReason, MessageQueue<DataType>::alwaysTruePredicate, infiniteTime());
 123         ASSERT(exitReason == MessageQueueTerminated || exitReason == MessageQueueMessageReceived);
 124         return result;
 125     }
 126
 127     template<typename DataType>
 128     template<typename Predicate>
 129     inline PassOwnPtr<DataType> MessageQueue<DataType>::waitForMessageFilteredWithTimeout(MessageQueueWaitResult& result, Predicate& predicate, double absoluteTime)
 130     {
 131         MutexLocker lock(m_mutex);
 132         bool timedOut = false;
 133
 134         DequeConstIterator<DataType*> found = m_queue.end();
 135         while (!m_killed && !timedOut && (found = m_queue.findIf(predicate)) == m_queue.end())
 136             timedOut = !m_condition.timedWait(m_mutex, absoluteTime);
 137
 138         ASSERT(!timedOut || absoluteTime != infiniteTime());
 139
 140         if (m_killed) {
 141             result = MessageQueueTerminated;
 142             return 0;
 143         }
 144
 145         if (timedOut) {
 146             result = MessageQueueTimeout;
 147             return 0;
 148         }
 149
 150         ASSERT(found != m_queue.end());
 151         DataType* message = *found;
 152         m_queue.remove(found);
 153         result = MessageQueueMessageReceived;
 154         return message;
 155     }
 156
 157     template<typename DataType>
 158     inline PassOwnPtr<DataType> MessageQueue<DataType>::tryGetMessage()
 159     {
 160         MutexLocker lock(m_mutex);
 161         if (m_killed)
 162             return 0;
 163         if (m_queue.isEmpty())
 164             return 0;
 165
 166         DataType* message = m_queue.first();
 167         m_queue.removeFirst();
 168         return message;
 169     }
 170
 171     template<typename DataType>
 172     template<typename Predicate>
 173     inline void MessageQueue<DataType>::removeIf(Predicate& predicate)
 174     {
 175         MutexLocker lock(m_mutex);
 176         // See bug 31657 for why this loop looks so weird
 177         while (true) {
 178             DequeConstIterator<DataType*> found = m_queue.findIf(predicate);
 179             if (found == m_queue.end())
 180                 break;
 181
 182             DataType* message = *found;
 183             m_queue.remove(found);
 184             delete message;
 185        }
 186     }
 187
 188     template<typename DataType>
 189     inline bool MessageQueue<DataType>::isEmpty()
 190     {
 191         MutexLocker lock(m_mutex);
 192         if (m_killed)
 193             return true;
 194         return m_queue.isEmpty();
 195     }
 196
 197     template<typename DataType>
 198     inline void MessageQueue<DataType>::kill()
 199     {
 200         MutexLocker lock(m_mutex);
 201         m_killed = true;
 202         m_condition.broadcast();
 203     }
 204
 205     template<typename DataType>
 206     inline bool MessageQueue<DataType>::killed() const
 207     {
 208         MutexLocker lock(m_mutex);
 209         return m_killed;
 210     }
 211 } // namespace WTF
 212
 213 using WTF::MessageQueue;
 214 // MessageQueueWaitResult enum and all its values.
 215 using WTF::MessageQueueWaitResult;
 216 using WTF::MessageQueueTerminated;
 217 using WTF::MessageQueueTimeout;
 218 using WTF::MessageQueueMessageReceived;
 219
 220 #endif // MessageQueue_h