artis-star/src/artis-star/common/utils/Multithreading.hpp

/**
 * @file common/utils/Multithreading.hpp
 * @author The ARTIS Development Team
 * See the AUTHORS or Authors.txt file
 */

/*
 * ARTIS - the multimodeling and simulation environment
 * This file is a part of the ARTIS environment
 *
 * Copyright (C) 2013-2019 ULCO http://www.univ-littoral.fr
 *
 * This program 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.
 *
 * This program 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 this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#ifndef COMMON_UTILS_MULTITHREADING
#define COMMON_UTILS_MULTITHREADING 1

#include <condition_variable>
#include <queue>
#include <mutex>

namespace artis {
    namespace common {

        struct BaseMessage {
            virtual ~BaseMessage() { }
        };

        template<typename Msg>
        struct Message : BaseMessage {
            explicit Message(Msg const& content)
                    :_content(content) { }

            Msg _content;
        };

        class Close {
        };

        class MessageQueue {
        public:
            MessageQueue() = default;

            template<typename T>
            void push(T const& msg)
            {
                std::lock_guard<std::mutex> lock(_mutex);

                _queue.push(std::make_shared<Message<T> >(msg));
                _condition.notify_all();
            }

            std::shared_ptr<BaseMessage> wait_and_pop()
            {
                std::unique_lock<std::mutex> lock(_mutex);

                _condition.wait(lock, [&] { return not _queue.empty(); });
                auto res = _queue.front();
                _queue.pop();
                return res;
            }

        private:
            std::mutex _mutex;
            std::condition_variable _condition;
            std::queue<std::shared_ptr<BaseMessage> > _queue;
        };

        class Sender {
        public:
            Sender()
                    :_queue(0) { }

            explicit Sender(MessageQueue* queue)
                    :_queue(queue) { }

            template<typename Message>
            void send(Message const& msg)
            {
                if (_queue) {
                    _queue->push(msg);
                }
            }

        private:
            MessageQueue* _queue;
        };

        template<typename PreviousDispatcher, typename Msg, typename Func>
        class TemplateDispatcher {
            template<typename Dispatcher, typename OtherMsg, typename OtherFunc>
            friend
            class TemplateDispatcher;

        public:
            TemplateDispatcher(TemplateDispatcher&& other)
                    :_queue(other._queue),
                     _previous(other._previous),
                     _function(
                             std::move(
                                     other._function)),
                     _chained(other._chained) { other._chained = true; }

            TemplateDispatcher(MessageQueue* queue,
                    PreviousDispatcher* previous,
                    Func&& function)
                    :
                    _queue(queue), _previous(previous),
                    _function(std::forward<Func>(function)),
                    _chained(false) { previous->_chained = true; }

            bool dispatch(std::shared_ptr<BaseMessage> const& msg)
            {
                Message<Msg>* message =
                        dynamic_cast < Message<Msg>* >(msg.get());

                if (message) {
                    _function(message->_content);
                    return true;
                } else {
                    return _previous->dispatch(msg);
                }
            }

            template<typename OtherMsg, typename OtherFunc>
            TemplateDispatcher<TemplateDispatcher<PreviousDispatcher, Msg, Func>,
                    OtherMsg, OtherFunc>
            handle(OtherFunc&& of)
            {
                return TemplateDispatcher<TemplateDispatcher<PreviousDispatcher,
                        Msg, Func>,
                        OtherMsg, OtherFunc>(
                        _queue, this, std::forward<OtherFunc>(of));
            }

            ~TemplateDispatcher() noexcept(false)
            {
                if (not _chained) {
                    wait_and_dispatch();
                }
            }

        private:
            TemplateDispatcher(TemplateDispatcher const&) = delete;

            TemplateDispatcher& operator=(TemplateDispatcher const&) = delete;

            void wait_and_dispatch()
            {
                for (;;) {
                    auto msg = _queue->wait_and_pop();

                    if (dispatch(msg)) {
                        break;
                    }
                }
            }

            MessageQueue* _queue;
            PreviousDispatcher* _previous;
            Func _function;
            bool _chained;
        };

        class Dispatcher {
            template<typename Dispatcher, typename Msg, typename Func>
            friend
            class TemplateDispatcher;

        public:
            Dispatcher(Dispatcher&& other)
                    :_queue(other._queue),
                     _chained(other._chained) { other._chained = true; }

            explicit Dispatcher(MessageQueue* queue)
                    :_queue(queue), _chained(false) { }

            template<typename Message, typename Func>
            TemplateDispatcher<Dispatcher, Message, Func>
            handle(Func&& function)
            {
                return TemplateDispatcher<Dispatcher, Message, Func>(
                        _queue, this, std::forward<Func>(function));
            }

            ~Dispatcher() noexcept(false)
            {
                if (not _chained) {
                    wait_and_dispatch();
                }
            }

        private:
            Dispatcher(Dispatcher const&) = delete;

            Dispatcher& operator=(Dispatcher const&) = delete;

            void wait_and_dispatch()
            {
                for (;;) {
                    auto msg = _queue->wait_and_pop();

                    dispatch(msg);
                }
            }

            bool dispatch(std::shared_ptr<BaseMessage> const& msg)
            {
                if (dynamic_cast < Message<Close>* >(msg.get())) {
                    throw Close();
                }
                return false;
            }

            MessageQueue* _queue;
            bool _chained;
        };

        class Receiver {
        public:
            Receiver() : _queue(new MessageQueue) { }

            ~Receiver() { delete _queue; }

            Sender get_sender() { return Sender(_queue); }

            Dispatcher wait() { return Dispatcher(_queue); }

        private:
            MessageQueue* _queue;
        };

    }
}  // namespace artis common

#endif