threadsafe_queue.h

Go to the documentation of this file.

/*
 * threadsafe_queue.h
 * 
 * Copyright (C) 2007,2009,2010  Thomas A. Vaughan
 * All rights reserved.
 *
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the <organization> nor the
 *       names of its contributors may be used to endorse or promote products
 *       derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THOMAS A. VAUGHAN ''AS IS'' AND ANY
 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THOMAS A. VAUGHAN BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 *
 * Implementation of a simple threadsafe queue.
 */

#ifndef WAVEPACKET_THREADSAFE_QUEUE_H__
#define WAVEPACKET_THREADSAFE_QUEUE_H__

// includes --------------------------------------------------------------------
#include "smart_mutex.h"

#include <deque>


///\ingroup threadsafe
/*@{*/


/// A threadsafe FIFO queue.  Supports basic push/pop operations, and iteration.
template <class T>
class threadsafe_queue {
private:
        // private typedefs ----------------------------------------------------
        typedef typename std::deque<T>::iterator        iterator;
        typedef threadsafe_queue<T>                     queue_t;

public:
        // public typedefs -----------------------------------------------------
        class iterator_t {
        public:
                friend class threadsafe_queue<T>;
        private:
                iterator                iter;
                dword_t                 rvn;
        };

        /// push an element onto the back of the queue
        void push_back(IN T& t) {
                        mlock l(m_mutex);
                        ++m_rvn;        // queue has changed!  increment version
                        m_queue.push_back(t);
                }

        /// pop an element from the front of the queue.
        /// Returns false if the queue is empty
        bool pop_front(OUT T& t) {
                        bool retval = false;
                        mlock l(m_mutex);
                        if (m_queue.size()) {
                                ++m_rvn;        // queue has changed!
                                retval = true;
                                t = m_queue.front();
                                m_queue.pop_front();
                        }
                        return retval;
                }

        /// returns the number of elements in the queue
        int size(void) const throw() {
                        queue_t * pThis = this->getNonConst();
                        mlock l(pThis->m_mutex);
                        return pThis->m_queue.size();
                }

        /// resets the given iterator to point to the beginning of the queue
        void getIterator(OUT iterator_t& i) const throw() {
                        queue_t * pThis = this->getNonConst();
                        mlock l(pThis->m_mutex);
                        i.rvn = m_rvn;
                        i.iter = pThis->m_queue.begin();
                }

        /// gets element pointed to by iterator, and increments iterator
        bool getNextElement(IO iterator_t& i, OUT T& t) const {
                        queue_t * pThis = this->getNonConst();
                        mlock l(pThis->m_mutex);
                        if (i.rvn != m_rvn) {
                                return false;   // iterator has been invalidated
                        }
                        if (pThis->m_queue.end() == i.iter) {
                                return false;   // already at end of queue
                        }

                        // iterator is valid!  get element and increment
                        t = *i.iter;
                        ++i.iter;
                        return true;
                }

private:
        // private helper methods ----------------------------------------------
        // really ugly method to retrieve a non-const pointer.  This is because
        //      read-only APIs need to take the mutex, even though they
        //      otherwise won't modify the collection
        queue_t * getNonConst(void) const throw() {
                        return (queue_t *) this;
                }

        // private member data -------------------------------------------------
        smart_mutex             m_mutex;
        std::deque<T>           m_queue;
        dword_t                 m_rvn;          // record version number
};


#endif  // WAVEPACKET_THREADSAFE_QUEUE_H__