rpc_manager.hpp

Go to the documentation of this file.

#pragma once

#include <exception>
#include <functional>
#include <map>
#include <memory>
#include <mutex>
#include <vector>

#include "../derecho_type_definitions.hpp"
#include "../view.hpp"
#include "derecho_internal.hpp"
#include "p2p_connections.hpp"
#include "remote_invocable.hpp"
#include "rpc_utils.hpp"
#include "view_manager.hpp"
#include <derecho/mutils-serialization/SerializationSupport.hpp>
#include <derecho/utils/logger.hpp>

namespace derecho {

template <typename T>
class Replicated;
template <typename T>
class ExternalCaller;

struct IDeserializationContext : public mutils::RemoteDeserializationContext {};

namespace rpc {

using PendingBase_ref = std::reference_wrapper<PendingBase>;

class RPCManager {
    static_assert(std::is_trivially_copyable<Opcode>::value, "Oh no! Opcode is not trivially copyable!");
    const node_id_t nid;
    std::unique_ptr<std::map<Opcode, receive_fun_t>> receivers;
    // mutils::DeserializationManager dsm{{}};
    // Weijia: I prefer the deserialization context vector.
    mutils::RemoteDeserialization_v rdv;

    template <typename T>
    friend class ::derecho::Replicated;  //Give only Replicated access to view_manager
    template <typename T>
    friend class ::derecho::ExternalCaller;
    ViewManager& view_manager;

    std::unique_ptr<sst::P2PConnections> connections;

    std::mutex p2p_connections_mutex;
    std::mutex pending_results_mutex;
    std::condition_variable pending_results_cv;
    //Both maps contain one list of PendingResults references per subgroup
    std::map<subgroup_id_t, std::queue<PendingBase_ref>> pending_results_to_fulfill;
    std::map<subgroup_id_t, std::list<PendingBase_ref>> fulfilled_pending_results;

    bool thread_start = false;
    std::mutex thread_start_mutex;
    std::condition_variable thread_start_cv;
    std::atomic<bool> thread_shutdown{false};
    std::thread rpc_thread;
    std::thread fifo_worker_thread;
    struct fifo_req {
        node_id_t sender_id;
        char* msg_buf;
        uint32_t buffer_size;
        fifo_req() : sender_id(0),
                     msg_buf(nullptr),
                     buffer_size(0) {}
        fifo_req(node_id_t _sender_id,
                 char* _msg_buf,
                 uint32_t _buffer_size) : sender_id(_sender_id),
                                          msg_buf(_msg_buf),
                                          buffer_size(_buffer_size) {}
    };
    std::queue<fifo_req> fifo_queue;
    std::mutex fifo_queue_mutex;
    std::condition_variable fifo_queue_cv;

    void p2p_receive_loop();

    void fifo_worker();

    void p2p_message_handler(node_id_t sender_id, char* msg_buf, uint32_t buffer_size);

    std::exception_ptr receive_message(const Opcode& indx, const node_id_t& received_from,
                                       char const* const buf, std::size_t payload_size,
                                       const std::function<char*(int)>& out_alloc);

    std::exception_ptr parse_and_receive(char* buf, std::size_t size,
                                         const std::function<char*(int)>& out_alloc);

public:
    RPCManager(ViewManager& group_view_manager,
               IDeserializationContext* deserialization_context_ptr)
            // mutils::RemoteDeserializationContext_p deserialization_context_ptr = nullptr)
            : nid(getConfUInt32(CONF_DERECHO_LOCAL_ID)),
              receivers(new std::decay_t<decltype(*receivers)>()),
              view_manager(group_view_manager) {
        if(deserialization_context_ptr != nullptr) {
            rdv.push_back(deserialization_context_ptr);
        }
        rpc_thread = std::thread(&RPCManager::p2p_receive_loop, this);
    }

    ~RPCManager();

    void create_connections();

    void start_listening();
    template <typename UserProvidedClass, typename FunctionTuple>
    auto make_remote_invocable_class(std::unique_ptr<UserProvidedClass>* cls, uint32_t type_id, uint32_t instance_id, FunctionTuple funs) {
        //FunctionTuple is a std::tuple of partial_wrapped<Tag, Ret, UserProvidedClass, Args>,
        //which is the result of the user calling tag<Tag>(&UserProvidedClass::method) on each RPC method
        //Use callFunc to unpack the tuple into a variadic parameter pack for build_remoteinvocableclass
        return mutils::callFunc([&](const auto&... unpacked_functions) {
            return build_remote_invocable_class<UserProvidedClass>(nid, type_id, instance_id, *receivers,
                                                                   bind_to_instance(cls, unpacked_functions)...);
        },
                                funs);
    }

    void destroy_remote_invocable_class(uint32_t instance_id);

    template <typename UserProvidedClass, typename FunctionTuple>
    auto make_remote_invoker(uint32_t type_id, uint32_t instance_id, FunctionTuple funs) {
        return mutils::callFunc([&](const auto&... unpacked_functions) {
            //Supply the template parameters for build_remote_invoker_for_class by
            //asking bind_to_instance for the type of the wrapped<> that corresponds to each partial_wrapped<>
            return build_remote_invoker_for_class<UserProvidedClass,
                                                  decltype(bind_to_instance(std::declval<std::unique_ptr<UserProvidedClass>*>(),
                                                                            unpacked_functions))...>(nid, type_id,
                                                                                                     instance_id, *receivers);
        },
                                funs);
    }

    void new_view_callback(const View& new_view);

    void rpc_message_handler(subgroup_id_t subgroup_id, node_id_t sender_id,
                             char* msg_buf, uint32_t buffer_size);

    bool finish_rpc_send(subgroup_id_t subgroup_id, PendingBase& pending_results_handle);

    volatile char* get_sendbuffer_ptr(uint32_t dest_id, sst::REQUEST_TYPE type);

    void finish_p2p_send(node_id_t dest_node, subgroup_id_t dest_subgroup_id, PendingBase& pending_results_handle);
};

//Now that RPCManager is finished being declared, we can declare these convenience types
//(the declarations should really live in remote_invocable.h, but they depend on RPCManager existing)
template <typename T>
using RemoteInvocableOf = std::decay_t<decltype(*std::declval<RPCManager>()
                                                         .make_remote_invocable_class(std::declval<std::unique_ptr<T>*>(),
                                                                                      std::declval<uint32_t>(),
                                                                                      std::declval<uint32_t>(),
                                                                                      T::register_functions()))>;

template <typename T>
using RemoteInvokerFor = std::decay_t<decltype(*std::declval<RPCManager>()
                                                        .make_remote_invoker<T>(std::declval<uint32_t>(),
                                                                                std::declval<uint32_t>(),
                                                                                T::register_functions()))>;

// test if the current thread is in an RPC handler to tell if we are sending a cascading RPC message.
bool in_rpc_handler();

}  // namespace rpc
}  // namespace derecho