src/components/utils/include/utils/message_queue.h

   1 /**
   2  * Copyright (c) 2013, Ford Motor Company
   3  * 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 are met:
   7  *
   8  * Redistributions of source code must retain the above copyright notice, this
   9  * list of conditions and the following disclaimer.
  10  *
  11  * Redistributions in binary form must reproduce the above copyright notice,
  12  * this list of conditions and the following
  13  * disclaimer in the documentation and/or other materials provided with the
  14  * distribution.
  15  *
  16  * Neither the name of the Ford Motor Company nor the names of its contributors
  17  * may be used to endorse or promote products derived from this software
  18  * without specific prior written permission.
  19  *
  20  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  21  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  22  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  23  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
  24  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  25  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  26  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  27  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  28  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  29  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  30  * POSSIBILITY OF SUCH DAMAGE.
  31 */
  32
  33 #ifndef MESSAGE_QUEUE_CLASS
  34 #define MESSAGE_QUEUE_CLASS
  35
  36 #include <queue>
  37
  38 #include "utils/conditional_variable.h"
  39 #include "utils/lock.h"
  40 #include "utils/logger.h"
  41 #include "utils/prioritized_queue.h"
  42
  43 /**
  44  * \class MessageQueue
  45  * \brief Wrapper for multithreading queue.
  46  */
  47
  48 template<typename T, class Q = std::queue<T> > class MessageQueue {
  49   public:
  50     typedef Q Queue;
  51     /**
  52      * \brief Default constructor
  53      */
  54     MessageQueue();
  55
  56     /**
  57      * \brief Destructor
  58      */
  59     ~MessageQueue();
  60
  61     /**
  62      * \brief Returns size of the queue.
  63      * \return Size of the queue.
  64      */
  65     int32_t size() const;
  66
  67     /**
  68      * \brief If queue is empty.
  69      * \return Is queue empty.
  70      */
  71     bool empty() const;
  72
  73     /**
  74      * \brief Tells if queue is being shut down
  75      */
  76     bool IsShuttingDown() const;
  77
  78     /**
  79      * \brief Adds element to the queue.
  80      * \param element Element to be added to the queue.n
  81      */
  82     void push(const T& element);
  83
  84     /**
  85      * \brief Removes element from the queue and returns it.
  86      * \return To element of the queue.
  87      */
  88     T pop();
  89
  90     /**
  91      * \brief Conditional wait.
  92      */
  93     void wait();
  94
  95     /**
  96      * \brief Shutdown the queue.
  97      * This leads to waking up everyone waiting on the queue
  98      * Queue being shut down can be drained ( with pop() )
  99      * But nothing must be added to the queue after it began
 100      * shutting down
 101      */
 102     void Shutdown();
 103
 104     /**
 105       * \brief Clears queue.
 106       */
 107     void Reset();
 108
 109   private:
 110
 111     /**
 112      *\brief Queue
 113      */
 114     Queue queue_;
 115     volatile bool shutting_down_;
 116
 117     /**
 118      *\brief Platform specific syncronisation variable
 119      */
 120     mutable sync_primitives::Lock queue_lock_;
 121     sync_primitives::ConditionalVariable queue_new_items_;
 122 };
 123
 124 template<typename T, class Q> MessageQueue<T, Q>::MessageQueue()
 125     : shutting_down_(false) {
 126 }
 127
 128 template<typename T, class Q> MessageQueue<T, Q>::~MessageQueue() {
 129   if (!queue_.empty()) {
 130     log4cxx::LoggerPtr logger =
 131         log4cxx::LoggerPtr(log4cxx::Logger::getLogger("Utils"));
 132     LOG4CXX_ERROR(logger, "Destruction of non-drained queue");
 133   }
 134 }
 135
 136 template<typename T, class Q> void MessageQueue<T, Q>::wait() {
 137   sync_primitives::AutoLock auto_lock(queue_lock_);
 138   while ((!shutting_down_) && queue_.empty()) {
 139     queue_new_items_.Wait(auto_lock);
 140   }
 141 }
 142
 143 template<typename T, class Q> int32_t MessageQueue<T, Q>::size() const {
 144   sync_primitives::AutoLock auto_lock(queue_lock_);
 145   return queue_.size();
 146 }
 147
 148 template<typename T, class Q> bool MessageQueue<T, Q>::empty() const {
 149   sync_primitives::AutoLock auto_lock(queue_lock_);
 150   return queue_.empty();
 151 }
 152
 153 template<typename T, class Q> bool MessageQueue<T, Q>::IsShuttingDown() const {
 154   sync_primitives::AutoLock auto_lock(queue_lock_);
 155   return shutting_down_;
 156 }
 157
 158 template<typename T, class Q> void MessageQueue<T, Q>::push(const T& element) {
 159   sync_primitives::AutoLock auto_lock(queue_lock_);
 160   if (shutting_down_) {
 161     log4cxx::LoggerPtr logger =
 162         log4cxx::LoggerPtr(log4cxx::Logger::getLogger("Utils"));
 163     LOG4CXX_ERROR(logger, "Runtime error, pushing into queue"
 164                          " that is being shut down");
 165   }
 166   queue_.push(element);
 167   queue_new_items_.Broadcast();
 168 }
 169
 170 template<typename T, class Q> T MessageQueue<T, Q>::pop() {
 171   sync_primitives::AutoLock auto_lock(queue_lock_);
 172   if (queue_.empty()) {
 173     log4cxx::LoggerPtr logger =
 174         log4cxx::LoggerPtr(log4cxx::Logger::getLogger("Utils"));
 175     LOG4CXX_ERROR(logger, "Runtime error, popping out of empty que");
 176   }
 177   T result = queue_.front();
 178   queue_.pop();
 179   return result;
 180 }
 181
 182 template<typename T, class Q> void MessageQueue<T, Q>::Shutdown() {
 183   sync_primitives::AutoLock auto_lock(queue_lock_);
 184   shutting_down_ = true;
 185   queue_new_items_.Broadcast();
 186 }
 187
 188 template<typename T, class Q> void MessageQueue<T, Q>::Reset() {
 189   sync_primitives::AutoLock auto_lock(queue_lock_);
 190   shutting_down_ = false;
 191   if (!queue_.empty()) {
 192     Queue empty_queue;
 193     queue_.swap(empty_queue);
 194   }
 195 }
 196
 197 #endif  //  MESSAGE_QUEUE_CLASS