include/dlvhex2/ConcurrentMessageQueueOwning.h

Go to the documentation of this file.

/* DMCS -- Distributed Nonmonotonic Multi-Context Systems.
 * Copyright (C) 2006-2015 Thomas Krennwallner
 *
 * This file is part of DMCS.
 *
 *  DMCS is free software: you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation, either version 3 of the License, or
 *  (at your option) any later version.
 *
 *  DMCS is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with DMCS.  If not, see <http://www.gnu.org/licenses/>.
 */

#ifndef _CONCURRENT_MESSAGE_QUEUE_OWNING_H
#define _CONCURRENT_MESSAGE_QUEUE_OWNING_H

#include <boost/shared_ptr.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition_variable.hpp>
#include <boost/thread/thread_time.hpp>
#include <boost/date_time/time_duration.hpp>

#include <queue>

namespace dlvhex
{

    template<class MessageBase>
        class ConcurrentMessageQueueOwning
    {
        private:
            typedef boost::shared_ptr<MessageBase> MessagePtr;

            std::queue<MessagePtr> q;
            const std::size_t n;
            std::size_t enq;
            std::size_t deq;

            mutable boost::mutex mtx;
            boost::condition_variable cnd;

            inline void
            notifyConsumer() {
                if (deq > 0) {   // is some consumer waiting?
                                 // notify one consuming thread
                    cnd.notify_one();
                }
            }

            inline void
            notifyProducer() {
                if (enq > 0) {   // is some producer waiting?
                                 // notify one producing thread
                    cnd.notify_one();
                }
            }

            inline void
            waitOnCapacity(boost::mutex::scoped_lock& lock) {
                                 // maximum capacity reached
                while (n == q.size()) {
                    ++enq;
                    cnd.wait(lock);
                    --enq;
                }

                notifyConsumer();
            }

            inline void
            waitOnEmpty(boost::mutex::scoped_lock& lock) {
                                 // minimum capacity reached
                while (q.empty()) {
                    ++deq;
                    cnd.wait(lock);
                    --deq;
                }

                notifyProducer();
            }

            inline bool
            waitOnTimedCapacity(boost::mutex::scoped_lock& lock, const boost::posix_time::time_duration& t) {
                bool no_timeout = true;

                                 // maximum capacity reached
                while (n == q.size() && no_timeout) {
                    ++enq;
                    no_timeout = cnd.timed_wait(lock, t);
                    --enq;
                }

                notifyConsumer();

                return no_timeout;
            }

            inline bool
            waitOnTimedEmpty(boost::mutex::scoped_lock& lock, const boost::posix_time::time_duration& t) {
                bool no_timeout = true;

                                 // minimum capacity reached
                while (q.empty() && no_timeout) {
                    ++deq;
                    no_timeout = cnd.timed_wait(lock, t);
                    --deq;
                }

                notifyProducer();

                return no_timeout;
            }

            inline void
            pushMessage (MessagePtr m) {
                q.push(m);
            }

            inline void
            popMessage (MessagePtr& m) {
                m = q.front();
                q.pop();
            }

        public:

            ConcurrentMessageQueueOwning()
                : n(1), enq(0), deq(0)
                { }

            ConcurrentMessageQueueOwning(std::size_t capacity)
                : n(capacity > 0 ? capacity : 1), enq(0), deq(0)
                { }

            ConcurrentMessageQueueOwning(const ConcurrentMessageQueueOwning<MessageBase>& q)
                : n(q.n), enq(0), deq(0)
                { }

            virtual
            ~ConcurrentMessageQueueOwning() {
                flush();
            }

            void flush() { {
                    boost::mutex::scoped_lock lock(mtx);
                    // just pop all elements from the queue (the smart pointers automatically
                    // destruct the elements and free the memory)
                    while (!q.empty())
                        q.pop();
                }
                notifyProducer();
            }

            bool
                empty () const
            {
                boost::mutex::scoped_lock lock(mtx);
                return q.empty();
            }

            bool
                size () const
            {
                return n;
            }

            void
            send (MessagePtr m, unsigned int prio) {
                boost::mutex::scoped_lock lock(mtx);
                waitOnCapacity(lock);
                pushMessage(m);
            }

            bool
            try_send (MessagePtr m, unsigned int prio) {
                boost::mutex::scoped_lock lock(mtx);

                if (q.size() < n) {
                    pushMessage(m);
                    notifyConsumer();
                    return true;
                }

                return false;
            }

            bool
            timed_send (MessagePtr m, unsigned int prio, const boost::posix_time::time_duration& t) {
                boost::mutex::scoped_lock lock(mtx);

                if (waitOnTimedCapacity(lock, t)) {
                    pushMessage(m);
                    return true;
                }

                return false;
            }

            void
            receive (MessagePtr& m, unsigned int& prio) {
                boost::mutex::scoped_lock lock(mtx);
                waitOnEmpty(lock);
                popMessage(m);
            }

            bool
            try_receive (MessagePtr& m, unsigned int prio) {
                boost::mutex::scoped_lock lock(mtx);

                if (!q.empty()) {
                    popMessage(m);
                    notifyProducer();
                    return true;
                }

                return false;
            }

            bool
            timed_receive (MessagePtr& m, unsigned int& prio, const boost::posix_time::time_duration& t) {
                boost::mutex::scoped_lock lock(mtx);

                if (waitOnTimedEmpty(lock, t)) {
                    popMessage(m);
                    return true;
                }

                return false;
            }

    };

}                                // namespace dlvhex
#endif                           // _CONCURRENT_MESSAGE_QUEUE_H


// vim:expandtab:ts=4:sw=4:
// mode: C++
// End: