message_manager.hpp

Go to the documentation of this file.

/*
Copyright (c) 2025 acrion innovations GmbH
Authors: Stefan Zipproth, s.zipproth@acrion.ch
 
This file is part of Cbeam, see https://github.com/acrion/cbeam and https://cbeam.org
 
Cbeam is offered under a commercial and under the AGPL license.
For commercial licensing, contact us at https://acrion.ch/sales. For AGPL licensing, see below.
 
AGPL licensing:
 
Cbeam is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
 
Cbeam 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 Affero General Public License for more details.
 
You should have received a copy of the GNU Affero General Public License
along with Cbeam. If not, see <https://www.gnu.org/licenses/>.
*/
 
#pragma once
 
#include <cbeam/concurrency/thread.hpp>
#include <cbeam/concurrency/threaded_object.hpp>
#include <cbeam/logging/log_manager.hpp>
#include <cbeam/platform/compiler_compatibility.hpp>
#include <cbeam/random/generators.hpp>
 
#include <atomic>             // for std::atomic
#include <condition_variable> // for std::condition_variable
#include <deque>
#include <functional>
#include <map>
#include <memory> // for std::unique_ptr, std::make_unique
#include <mutex>  // for std::mutex, std::lock_guard, std::unique_lock
#include <set>    // for std::set, std::operator!=, std::_Rb_tree_const_iterator
#include <thread> // for std::thread
#include <unordered_set>
#include <utility> // for std::forward
 
namespace cbeam::concurrency
{
    template <typename MessageDataType>
    class message_manager
    {
    public:
        using message_id_type = std::size_t;

        using message_logger_type = std::function<void(message_id_type, MessageDataType, bool)>;

        enum class order_type
        {
            FIFO,  
            FILO,  
            RANDOM 
        };
 
    private:
        struct message_queue
        {
            std::deque<MessageDataType>              queue;
            std::shared_ptr<std::mutex>              queue_mutex{std::make_shared<std::mutex>()};
            std::shared_ptr<std::condition_variable> queue_cv{std::make_shared<std::condition_variable>()};
            std::condition_variable                  queue_cv_empty;
            std::atomic<std::size_t>                 busy_count{0};
            message_logger_type                      message_logger;
            std::mutex                               message_logger_mutex;
        };
 
        class message_handler : public threaded_object<message_handler, MessageDataType>
        {
        public:
            message_handler(typename threaded_object<message_handler, MessageDataType>::construction_token,
                            const std::shared_ptr<message_queue>&                       queue,
                            const message_id_type                                       message_id,
                            std::function<void(MessageDataType message_type)>           on_message,
                            const std::function<void(const std::exception& exception)>& on_exception,
                            const std::function<void()>&                                on_exit,
                            const std::string&                                          thread_name,
                            const order_type                                            order)
                : threaded_object<message_handler, MessageDataType>{}
                , _message_queue{queue}
                , _message_id{message_id}
                , _on_message{on_message}
                , _on_exception{on_exception}
                , _on_exit{on_exit}
                , _thread_name{thread_name}
                , _order{order} { CBEAM_LOG_DEBUG("cbeam::concurrency::message_manager: constructor for thread + '" + thread_name + "' received message queue " + cbeam::convert::to_string(&queue->queue) + " to message handler"); }
 
            ~message_handler() override = default;
 
            void on_start() override
            {
                set_thread_name(_thread_name.c_str());
                CBEAM_LOG_DEBUG("cbeam::concurrency::message_manager: Thread '" + _thread_name + "' now waits for messages in queue " + cbeam::convert::to_string(&_message_queue->queue));
            }
 
            bool is_message_available() override { return !_message_queue->queue.empty(); }
 
            MessageDataType get_message() override
            {
                MessageDataType message;
 
                switch (_order)
                {
                case order_type::FIFO:
                    message = _message_queue->queue.front();
                    _message_queue->queue.pop_front();
                    break;
                case order_type::FILO:
                    message = _message_queue->queue.back();
                    _message_queue->queue.pop_back();
                    break;
                case order_type::RANDOM:
                {
                    const size_t rnd = cbeam::random::random_number(_message_queue->queue.size());
                    message          = _message_queue->queue.at(rnd);
                    _message_queue->queue.erase(_message_queue->queue.begin() + rnd);
                    break;
                }
                }
 
                _message_queue->busy_count.fetch_add(1, std::memory_order_acq_rel);
 
                CBEAM_SUPPRESS_WARNINGS_PUSH() // suppress false positive "may be used uninitialized"
                return message;
                CBEAM_SUPPRESS_WARNINGS_POP()
            }
 
            void on_message(const MessageDataType& message_data) noexcept override
            {
                try
                {
                    {
                        std::lock_guard<std::mutex> lock_logger(_message_queue->message_logger_mutex);
                        if (_message_queue->message_logger) { _message_queue->message_logger(_message_id, message_data, false); }
                    }
 
                    _on_message(message_data);
                }
                catch (const std::exception& ex)
                {
                    _on_exception(ex);
                }
 
                // Decrement the busy_count now that we're done
                auto still_busy = _message_queue->busy_count.fetch_sub(1, std::memory_order_acq_rel) - 1;
 
                // We also need to ensure we hold the queue mutex to safely check queue->queue.empty().
                // Because multiple threads might pop from the queue concurrently.
                std::lock_guard<std::mutex> lock_queue_mutex(*_message_queue->queue_mutex);
 
                // If the queue is empty and no thread is busy => notify waiting threads
                if (_message_queue->queue.empty() && still_busy == 0)
                {
                    _message_queue->queue_cv_empty.notify_all();
                }
            }
 
            void on_exit() override { _on_exit(); }
 
        private:
            std::shared_ptr<message_queue>                       _message_queue;
            const message_id_type                                _message_id;
            std::function<void(MessageDataType message_data)>    _on_message;
            std::function<void(const std::exception& exception)> _on_exception;
            std::function<void()>                                _on_exit;
            std::string                                          _thread_name;
            order_type                                           _order;
        }; // class message_handler
 
        std::mutex                                                                      _threads_mutex;
        std::map<message_id_type, std::unordered_set<std::unique_ptr<message_handler>>> _threads;
        std::mutex                                                                      _message_queues_mutex;
        std::map<message_id_type, std::shared_ptr<message_queue>>                       _message_queues;
 
    public:
        void send_message(const message_id_type message_id, const MessageDataType& message_data, const std::size_t max_queued_messages = 0)
        {
            std::shared_ptr<message_queue> queue;
            {
                std::lock_guard<std::mutex> lock(_message_queues_mutex);
                queue = _message_queues[message_id];
                if (!queue) { queue = _message_queues[message_id] = std::make_shared<message_queue>(); }
            }
 
            {
                std::unique_lock<std::mutex> lock(*queue->queue_mutex);
                queue->queue_cv->wait(lock, [&]
                                      { return max_queued_messages == 0 || queue->queue.size() < max_queued_messages; });
                queue->queue.emplace_back(message_data);
            }
 
            CBEAM_LOG_DEBUG("cbeam::concurrency::message_manager(" + cbeam::convert::to_string(this) + ")::send_message: adding message to receiver " + std::to_string(message_id) + " to queue " + cbeam::convert::to_string(&queue->queue));
            queue->queue_cv->notify_all();
 
            std::lock_guard<std::mutex> lock_logger(queue->message_logger_mutex);
            if (queue->message_logger) { queue->message_logger(message_id, message_data, true); }
        }

        void add_handler(const message_id_type                                message_id,
                         std::function<void(MessageDataType message_data)>    on_message,
                         std::function<void(const std::exception& exception)> on_exception = nullptr,
                         std::function<void()>                                on_exit      = nullptr,
                         const std::string&                                   thread_name  = {},
                         const order_type                                     order        = order_type::FIFO)
        {
            std::lock_guard<std::mutex> lock(_threads_mutex);
            std::lock_guard<std::mutex> lock2(_message_queues_mutex);
 
            auto& queue = _message_queues[message_id];
            if (!queue) { queue = std::make_shared<message_queue>(); }
            CBEAM_LOG_DEBUG("cbeam::concurrency::message_manager(" + cbeam::convert::to_string(this) + ")::add_handler(" + std::to_string(message_id) + ", ...): passing queue " + cbeam::convert::to_string(&queue->queue) + " to message handler");
            _threads[message_id].emplace(message_handler::create(queue->queue_mutex,
                                                                 queue->queue_cv,
                                                                 queue,
                                                                 message_id,
                                                                 on_message,
                                                                 on_exception,
                                                                 on_exit,
                                                                 thread_name + "_" + std::to_string(message_id),
                                                                 order));
        }
 
        void wait_until_empty(const message_id_type message_id)
        {
            std::shared_ptr<message_queue> queue;
            {
                std::lock_guard<std::mutex> lock(_message_queues_mutex);
                auto                        it = _message_queues.find(message_id);
                if (it == _message_queues.end() || !(it->second))
                {
                    // no known queue => nothing to wait for
                    return;
                }
                queue = it->second;
            }
 
            std::unique_lock<std::mutex> lock_q(*queue->queue_mutex);
 
            queue->queue_cv_empty.wait(lock_q, [&]
                                       { return queue->queue.empty() && (queue->busy_count.load(std::memory_order_acquire) == 0); });
        }
 
        void dispose(const message_id_type message_id)
        {
            std::lock_guard<std::mutex> lock(_threads_mutex);
            std::lock_guard<std::mutex> lock2(_message_queues_mutex);
            _threads[message_id].clear();
        }
 
        void set_logger(const message_id_type message_id,
                        message_logger_type   on_message)
        {
            std::lock_guard<std::mutex> lock_logger(_message_queues[message_id]->message_logger_mutex);
            _message_queues[message_id]->message_logger = on_message;
        }
    };
}