general_executor.cpp

Go to the documentation of this file.

/*
 * Copyright 2023 MusicScience37 (Kenta Kabashima)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#include <atomic>
#include <cstdlib>
#include <exception>
#include <functional>
#include <memory>
#include <mutex>
#include <sstream>
#include <string>
#include <thread>
#include <utility>
#include <vector>
 
#include <asio/executor_work_guard.hpp>
 
#include "msgpack_rpc/common/msgpack_rpc_exception.h"
#include "msgpack_rpc/common/status_code.h"
#include "msgpack_rpc/config/executor_config.h"
#include "msgpack_rpc/executors/asio_context_type.h"
#include "msgpack_rpc/executors/i_async_executor.h"
#include "msgpack_rpc/executors/operation_type.h"
#include "msgpack_rpc/logging/logger.h"
 
namespace msgpack_rpc::executors {

class GeneralExecutor final : public IAsyncExecutor {
public:
    GeneralExecutor(std::shared_ptr<logging::Logger> logger,
        const config::ExecutorConfig& config)
        : transport_context_thread_pairs_(config.num_transport_threads()),
          callbacks_context_thread_pairs_(config.num_callback_threads()),
          logger_(std::move(logger)) {}
 
    GeneralExecutor(const GeneralExecutor&) = delete;
    GeneralExecutor(GeneralExecutor&&) = delete;
    GeneralExecutor& operator=(const GeneralExecutor&) = delete;
    GeneralExecutor& operator=(GeneralExecutor&&) = delete;

    ~GeneralExecutor() override { stop(); }

    void start() override {
        if (is_started_.exchange(true)) {
            throw MsgpackRPCException(StatusCode::PRECONDITION_NOT_MET,
                "An executor must not be run multiple times.");
        }
 
        MSGPACK_RPC_TRACE(logger_, "Start an executor.");
        start_threads();
    }

    void stop() override {
        if (is_stopped_.exchange(true)) {
            // Already stopped.
            return;
        }
 
        stop_threads();
 
        // Remove objects
        transport_context_thread_pairs_.clear();
        callbacks_context_thread_pairs_.clear();
 
        MSGPACK_RPC_TRACE(logger_, "Executor run stopped.");
    }

    AsioContextType& context(OperationType type) noexcept override {
        switch (type) {
        case OperationType::TRANSPORT:
            return transport_context_thread_pairs_
                [get_context_index(transport_context_index_,
                     transport_context_thread_pairs_.size())]
                    .context;
        case OperationType::CALLBACK:
            return callbacks_context_thread_pairs_
                [get_context_index(callback_context_index_,
                     callbacks_context_thread_pairs_.size())]
                    .context;
        }
        // This line won't be executed without a bug.
        std::abort();
    }

    [[nodiscard]] std::exception_ptr last_exception() override {
        std::unique_lock<std::mutex> lock(exception_in_thread_mutex_);
        return exception_in_thread_;
    }

    void on_exception(
        std::function<void(std::exception_ptr)> exception_callback) override {
        std::unique_lock<std::mutex> lock(exception_callbacks_mutex_);
        exception_callbacks_.push_back(std::move(exception_callback));
    }

    [[nodiscard]] bool is_running() override {
        std::unique_lock<std::mutex> lock(exception_in_thread_mutex_);
        if (exception_in_thread_) {
            return false;
        }
        return is_started_.load(std::memory_order_relaxed) &&
            !is_stopped_.load(std::memory_order_relaxed);
    }
 
private:
    void start_threads() {
        try {
            for (auto& context_thread_pair : transport_context_thread_pairs_) {
                context_thread_pair.thread =
                    std::thread{[this, &context_thread_pair] {
                        run_in_thread(context_thread_pair.context);
                    }};
            }
            for (auto& context_thread_pair : callbacks_context_thread_pairs_) {
                context_thread_pair.thread =
                    std::thread{[this, &context_thread_pair] {
                        run_in_thread(context_thread_pair.context);
                    }};
            }
        } catch (...) {
            stop_threads();
            throw;
        }
    }

    void stop_threads() {
        async_stop_threads_gently();
        for (auto& context_thread_pair : transport_context_thread_pairs_) {
            if (context_thread_pair.thread.joinable()) {
                context_thread_pair.thread.join();
            }
        }
        for (auto& context_thread_pair : callbacks_context_thread_pairs_) {
            if (context_thread_pair.thread.joinable()) {
                context_thread_pair.thread.join();
            }
        }
    }

    void interrupt_threads() {
        for (auto& context_thread_pair : transport_context_thread_pairs_) {
            context_thread_pair.context.stop();
        }
        for (auto& context_thread_pair : callbacks_context_thread_pairs_) {
            context_thread_pair.context.stop();
        }
    }

    void async_stop_threads_gently() {
        for (auto& context_thread_pair : transport_context_thread_pairs_) {
            context_thread_pair.work_guard.reset();
        }
        for (auto& context_thread_pair : callbacks_context_thread_pairs_) {
            context_thread_pair.work_guard.reset();
        }
    }

    void run_in_thread(AsioContextType& context) {
        const auto thread_id = create_thread_id_string();
        MSGPACK_RPC_TRACE(logger_, "Start an executor thread {}.", thread_id);
        try {
            context.run();
        } catch (const std::exception& e) {
            MSGPACK_RPC_CRITICAL(logger_,
                "Executor stops due to an exception thrown in thread {}: {}",
                thread_id, e.what());
            const std::exception_ptr exception = std::current_exception();
            {
                std::unique_lock<std::mutex> lock(exception_in_thread_mutex_);
                exception_in_thread_ = exception;
            }
            interrupt_threads();
            {
                std::unique_lock<std::mutex> lock(exception_callbacks_mutex_);
                for (const auto& exception_callback : exception_callbacks_) {
                    exception_callback(exception);
                }
            }
        }
        MSGPACK_RPC_TRACE(logger_, "Finish an executor thread {}.", thread_id);
    }

    static std::string create_thread_id_string() {
        std::ostringstream stream;
        stream << std::this_thread::get_id();
        return stream.str();
    }

    static std::size_t get_context_index(
        std::atomic<std::size_t>& index, std::size_t size) {
        // TODO Should I consider overflow?
        return index.fetch_add(1, std::memory_order_relaxed) % size;
    }

    struct ContextThreadPair {
    public:
        AsioContextType context;

        asio::executor_work_guard<AsioContextType::executor_type> work_guard;

        std::thread thread;

        ContextThreadPair()
            : context(1), work_guard(asio::make_work_guard(context)) {}
    };

    std::vector<ContextThreadPair> transport_context_thread_pairs_;

    std::vector<ContextThreadPair> callbacks_context_thread_pairs_;

    std::atomic<std::size_t> transport_context_index_{0};

    std::atomic<std::size_t> callback_context_index_{0};

    std::atomic<bool> is_started_{false};

    std::atomic<bool> is_stopped_{false};

    std::exception_ptr exception_in_thread_{};

    std::mutex exception_in_thread_mutex_{};

    std::vector<std::function<void(std::exception_ptr)>> exception_callbacks_{};

    std::mutex exception_callbacks_mutex_{};

    std::shared_ptr<logging::Logger> logger_;
};
 
std::shared_ptr<IAsyncExecutor> create_executor(
    std::shared_ptr<logging::Logger> logger,
    const config::ExecutorConfig& config) {
    return std::make_shared<GeneralExecutor>(std::move(logger), config);
}
 
}  // namespace msgpack_rpc::executors