OrderedQueueDispatcherPool.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 <boost/bind.hpp>
#include <boost/thread/condition.hpp>
#include <boost/function.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>

#include "../misc/IllegalStateException.h"
#include "SynchronizedQueue.h"

namespace rsc {
namespace threading {

template<class M, class R>
class OrderedQueueDispatcherPool {
public:

    typedef boost::function<void(boost::shared_ptr<R>& receiver,
            const M& message)> deliverFunction;

    typedef boost::function<bool(boost::shared_ptr<R>& receiver,
            const M& message)> filterFunction;

    class DeliveryHandler {
    public:

        virtual ~DeliveryHandler() {
        }

        virtual void
        deliver(boost::shared_ptr<R>& receiver, const M& message) = 0;

    };

    typedef boost::shared_ptr<DeliveryHandler> DeliveryHandlerPtr;

    class FilterHandler {
    public:

        virtual ~FilterHandler() {
        }

        virtual bool
        filter(boost::shared_ptr<R>& receiver, const M& message) = 0;

    };

    typedef boost::shared_ptr<FilterHandler> FilterHandlerPtr;

private:

    class TrueFilter: public FilterHandler {
    public:
        bool filter(boost::shared_ptr<R>& /*receiver*/, const M& /*message*/) {
            return true;
        }
    };

    class FilterFunctionAdapter: public FilterHandler {
    public:

        explicit FilterFunctionAdapter(filterFunction function) :
            function(function) {
        }

        bool filter(boost::shared_ptr<R>& receiver, const M& message) {
            return function(receiver, message);
        }

    private:
        filterFunction function;
    };

    class DeliverFunctionAdapter: public DeliveryHandler {
    public:

        explicit DeliverFunctionAdapter(deliverFunction function) :
            function(function) {
        }

        void deliver(boost::shared_ptr<R>& receiver, const M& message) {
            function(receiver, message);
        }

    private:
        deliverFunction function;
    };

    class Receiver {
    public:

        Receiver(boost::shared_ptr<R> receiver) :
            receiver(receiver), processing(false) {
        }

        boost::shared_ptr<R> receiver;
        // TODO think about if this really requires a synchronized queue if
        // all message dispatching to worker threads is synchronized
        SynchronizedQueue<M> queue;

        boost::condition processingCondition;

        volatile bool processing;

    };

    // TODO make this a set to only allow unique subscriptions?
    boost::mutex receiversMutex;
    std::vector<boost::shared_ptr<Receiver> > receivers;
    size_t currentPosition;

    volatile bool jobsAvailable;
    boost::condition jobsAvailableCondition;
    volatile bool interrupted;

    volatile bool parallelCalls;

    volatile bool started;

    void nextJob(const unsigned int& /*workerNum*/,
            boost::shared_ptr<Receiver>& receiver) {

        boost::mutex::scoped_lock lock(receiversMutex);

        //        std::cout << "Worker " << workerNum << " requests a new job"
        //                << std::endl;

        // wait until a job is available
        bool gotJob = false;
        while (!gotJob) {

            while (!jobsAvailable && !interrupted) {
                //                std::cout << "Worker " << workerNum
                //                        << ": no jobs available, waiting" << std::endl;
                jobsAvailableCondition.wait(lock);
            }

            if (interrupted) {
                throw InterruptedException("Processing was interrupted");
            }

            // search for the next job
            for (size_t pos = 0; pos < receivers.size(); ++pos) {

                // TODO is this selection fair in any case?
                ++currentPosition;
                size_t realPos = currentPosition % receivers.size();

                // TODO maybe provide an atomic pop and tell if successful
                // operation in SynchronizedQueue
                if ((parallelCalls || !receivers[realPos]->processing)
                        && !receivers[realPos]->queue.empty()) {

                    // found a job

                    receiver = receivers[realPos];
                    receiver->processing = true;
                    gotJob = true;
                    break;

                }

            }

            // did not find a job, hence there are no other jobs right now
            if (!gotJob) {
                jobsAvailable = false;
            }

        }

        // if I got a job there are certainly others interested in jobs
        //        std::cout << "Worker " << workerNum << ": got job for receiver"
        //                << receiver->receiver << ", notify_one" << std::endl;
        lock.unlock();
        jobsAvailableCondition.notify_one();

    }

    unsigned int threadPoolSize;

    void finishedWork(boost::shared_ptr<Receiver> receiver) {

        boost::mutex::scoped_lock lock(receiversMutex);
        // changing this flag must already be locked as it is read by the
        // globally synchronized nextJob method to determine if a job is
        // available
        receiver->processing = false;
        receiver->processingCondition.notify_all();
        // maybe avoid informing other threads about new jobs of the receiver
        // will be removed by adding a flag? Right now they will start to search
        // for a new job and may not find one.
        if (!receiver->queue.empty()) {
            jobsAvailable = true;
            lock.unlock();
            jobsAvailableCondition.notify_one();
        }

    }

    void worker(const unsigned int& workerNum) {

        try {
            while (true) {

                boost::shared_ptr<Receiver> receiver;
                nextJob(workerNum, receiver);
                M message = receiver->queue.pop();
                //                std::cout << "Worker " << workerNum << " got new job: "
                //                        << message << " for receiver " << *(receiver->receiver)
                //                        << std::endl;
                if (filterHandler->filter(receiver->receiver, message)) {
                    deliveryHandler->deliver(receiver->receiver, message);
                }
                finishedWork(receiver);

            }
        } catch (InterruptedException& e) {
            //            std::cout << "Worker " << workerNum << " was interrupted"
            //                    << std::endl;
        }

    }

    std::vector<boost::shared_ptr<boost::thread> > threadPool;

    DeliveryHandlerPtr deliveryHandler;
    FilterHandlerPtr filterHandler;

public:

    OrderedQueueDispatcherPool(const unsigned int& threadPoolSize,
            deliverFunction delFunc) :
            currentPosition(0), jobsAvailable(false), interrupted(false), parallelCalls(
                    false), started(false), threadPoolSize(threadPoolSize), deliveryHandler(
                    new DeliverFunctionAdapter(delFunc)), filterHandler(
                    new TrueFilter()) {
    }

    OrderedQueueDispatcherPool(const unsigned int& threadPoolSize,
            deliverFunction delFunc, filterFunction filterFunc) :
            currentPosition(0), jobsAvailable(false), interrupted(false), parallelCalls(
                    false), started(false), threadPoolSize(threadPoolSize), deliveryHandler(
                    new DeliverFunctionAdapter(delFunc)), filterHandler(
                    new FilterFunctionAdapter(filterFunc)) {
    }

    OrderedQueueDispatcherPool(const unsigned int& threadPoolSize,
            DeliveryHandlerPtr deliveryHandler) :
            currentPosition(0), jobsAvailable(false), interrupted(false), parallelCalls(
                    false), started(false), threadPoolSize(threadPoolSize), deliveryHandler(
                    deliveryHandler), filterHandler(new TrueFilter) {
    }

    OrderedQueueDispatcherPool(const unsigned int& threadPoolSize,
            DeliveryHandlerPtr deliveryHandler, FilterHandlerPtr filterHandler) :
            currentPosition(0), jobsAvailable(false), interrupted(false), parallelCalls(
                    false), started(false), threadPoolSize(threadPoolSize), deliveryHandler(
                    deliveryHandler), filterHandler(filterHandler) {
    }

    virtual ~OrderedQueueDispatcherPool() {
        stop();
    }

    void registerReceiver(boost::shared_ptr<R> receiver) {
        boost::mutex::scoped_lock lock(receiversMutex);
        boost::shared_ptr<Receiver> rec(new Receiver(receiver));
        receivers.push_back(rec);
    }

    bool unregisterReceiver(boost::shared_ptr<R> receiver) {

        boost::mutex::scoped_lock lock(receiversMutex);

        for (typename std::vector<boost::shared_ptr<Receiver> >::iterator it =
                receivers.begin(); it != receivers.end(); ++it) {
            boost::shared_ptr<Receiver> rec = *it;
            if (rec->receiver == receiver) {
                it = receivers.erase(it);
                while (rec->processing) {
                    rec ->processingCondition.wait(lock);
                }
                return true;
            }
        }
        return false;

    }

    void setParallelCalls(const bool& allow) {
        parallelCalls = allow;
    }

    void start() {

        boost::mutex::scoped_lock lock(receiversMutex);
        if (started) {
            throw rsc::misc::IllegalStateException("Pool already running");
        }

        interrupted = false;

        for (unsigned int i = 0; i < threadPoolSize; ++i) {
            boost::function<void()> workerMethod = boost::bind(
                    &OrderedQueueDispatcherPool::worker, this, i);
            boost::shared_ptr<boost::thread> w(new boost::thread(workerMethod));
            threadPool.push_back(w);
        }

        started = true;

    }

    void stop() {

        {
            boost::mutex::scoped_lock lock(receiversMutex);
            interrupted = true;
        }
        jobsAvailableCondition.notify_all();

        for (unsigned int i = 0; i < threadPool.size(); ++i) {
            threadPool[i]->join();
        }
        threadPool.clear();

        started = false;

    }

    void push(const M& message) {

        //        std::cout << "new job " << message << std::endl;
        {
            boost::mutex::scoped_lock lock(receiversMutex);
            for (typename std::vector<boost::shared_ptr<Receiver> >::iterator
                    it = receivers.begin(); it != receivers.end(); ++it) {
                (*it)->queue.push(message);
            }
            jobsAvailable = true;
        }
        jobsAvailableCondition.notify_one();

    }

};

}
}