SynchronizedQueue.h

Go to the documentation of this file.

/* ============================================================
 *
 * This file is a part of RSC project
 *
 * Copyright (C) 2010 by Johannes Wienke <jwienke at techfak dot uni-bielefeld dot de>
 *
 * This file may be licensed under the terms of the
 * GNU Lesser General Public License Version 3 (the ``LGPL''),
 * or (at your option) any later version.
 *
 * Software distributed under the License is distributed
 * on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
 * express or implied. See the LGPL for the specific language
 * governing rights and limitations.
 *
 * You should have received a copy of the LGPL along with this
 * program. If not, go to http://www.gnu.org/licenses/lgpl.html
 * or write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 *
 * The development of this software was supported by:
 *   CoR-Lab, Research Institute for Cognition and Robotics
 *     Bielefeld University
 *
 * ============================================================ */

#pragma once

#include <queue>

#include <boost/format.hpp>
#include <boost/function.hpp>
#include <boost/noncopyable.hpp>
#include <boost/thread/condition.hpp>
#include <boost/thread/recursive_mutex.hpp>

#include "InterruptedException.h"
#include "rsc/rscexports.h"

namespace rsc {
namespace threading {

class RSC_EXPORT QueueEmptyException: public std::runtime_error {
public:
    QueueEmptyException();
    explicit QueueEmptyException(const std::string& message);
};

template<class M>
class SynchronizedQueue: public boost::noncopyable {
public:

    typedef boost::function<void(const M& drop)> dropHandlerType;

private:

    volatile bool interrupted;
    std::queue<M> queue;
    mutable boost::recursive_mutex mutex;
    boost::condition condition;
    unsigned int sizeLimit;
    dropHandlerType dropHandler;

public:

    explicit SynchronizedQueue(const unsigned int& sizeLimit = 0,
                               dropHandlerType dropHandler = 0) :
        interrupted(false), sizeLimit(sizeLimit), dropHandler(dropHandler) {
    }

    virtual ~SynchronizedQueue() {
        // TODO what to do on destruction if still someone in waiting?
        // Simply calling interrupt does not help because threads still access
        // the queue variables at waking up from the interruption while this
        // class is being destructed and the condition variable gets destructed
    }

    void push(const M& message) {
        {
            boost::recursive_mutex::scoped_lock lock(this->mutex);
            if (this->sizeLimit > 0) {
                while (this->queue.size() > this->sizeLimit - 1) {
                    if (this->dropHandler) {
                        this->dropHandler(this->queue.front());
                    }
                    this->queue.pop();
                }
            }
            this->queue.push(message);
        }
        this->condition.notify_one();
    }

    M peek(const boost::uint32_t& timeoutMs = 0) {

        boost::recursive_mutex::scoped_lock lock(this->mutex);

        while (!this->interrupted && this->queue.empty()) {
            if (timeoutMs == 0) {
                this->condition.wait(lock);
            } else {
#if BOOST_VERSION >= 105000
                if (!this->condition.timed_wait(lock, boost::posix_time::milliseconds(timeoutMs))) {
#else
                const boost::system_time timeout = boost::get_system_time()
                    + boost::posix_time::milliseconds(timeoutMs);
                if (!this->condition.timed_wait(lock, timeout)) {
#endif
                    throw QueueEmptyException(boost::str(boost::format("No element available on queue within %d ms.") % timeoutMs));
                }
            }
        }

        if (this->interrupted) {
            throw InterruptedException("Queue was interrupted");
        }

        return this->queue.front();
    }

    M pop(const boost::uint32_t& timeoutMs = 0) {

        boost::recursive_mutex::scoped_lock lock(this->mutex);

        while (!this->interrupted && this->queue.empty()) {
            if (timeoutMs == 0) {
                this->condition.wait(lock);
            } else {
#if BOOST_VERSION >= 105000
                if (!this->condition.timed_wait(lock, boost::posix_time::milliseconds(timeoutMs))) {
#else
                const boost::system_time timeout = boost::get_system_time()
                        + boost::posix_time::milliseconds(timeoutMs);
                if (!this->condition.timed_wait(lock, timeout)) {
#endif
                    throw QueueEmptyException(boost::str(boost::format("No element available on queue within %d ms.") % timeoutMs));
                }
            }
        }

        if (this->interrupted) {
            throw InterruptedException("Queue was interrupted");
        }

        M message = this->queue.front();
        this->queue.pop();
        return message;

    }

    M tryPop() {

        boost::recursive_mutex::scoped_lock lock(this->mutex);

        if (this->queue.empty()) {
            throw QueueEmptyException();
        }

        M message = this->queue.front();
        this->queue.pop();
        return message;

    }

    bool empty() const {
        boost::recursive_mutex::scoped_lock lock(this->mutex);
        return this->queue.empty();
    }

    std::size_t size() const {
        boost::recursive_mutex::scoped_lock lock(this->mutex);
        return this->queue.size();
    }

    void clear() {
        boost::recursive_mutex::scoped_lock lock(this->mutex);
        while (!this->queue.empty()) {
            this->queue.pop();
        }
    }

    void interrupt() {
        {
            boost::recursive_mutex::scoped_lock lock(this->mutex);
            this->interrupted = true;
        }
        this->condition.notify_all();
    }

};

}
}