remote_invocable.hpp

Go to the documentation of this file.

#pragma once

#include <functional>
#include <numeric>
#include <type_traits>

#include "rpc_utils.hpp"
#include <derecho/mutils-serialization/SerializationSupport.hpp>
#include <derecho/utils/logger.hpp>
#include <mutils/FunctionalMap.hpp>
#include <mutils/tuple_extras.hpp>
#include <spdlog/spdlog.h>

namespace derecho {

namespace rpc {

bool in_rpc_handler();

//Technically, RemoteInvocable "specializes" this template for the case where
//the second parameter is a std::function<Ret(Args...)>. However, there is no
//implementation for any other specialization, so this template is meaningless.
template <FunctionTag, typename>
struct RemoteInvocable;

template <FunctionTag, typename>
struct RemoteInvoker;

template <FunctionTag Tag, typename Ret, typename... Args>
struct RemoteInvoker<Tag, std::function<Ret(Args...)>> {
    using remote_function_type = std::function<Ret(Args...)>;
    const Opcode invoke_opcode;
    const Opcode reply_opcode;

    Ret* returnRet() {
        return nullptr;
    }

    /* Maps invocation-instance IDs to results sets
       Weijia: Although the maximum pending RPC is controlled by the window
       size option in the configuration, the user application may not retrieve
       the results on time. Therefore, the results may piled up and we need more
       slots for the results than window size. We hardwired this to 4K here.
       TODO: Obviously, this is not space efficient. Let's find a better RPC
       return value mechanism.
     */

    #define MAX_CONCURRENT_RPCS_PER_INVOKER (4096)
    PendingResults<Ret> results_vector[MAX_CONCURRENT_RPCS_PER_INVOKER];
    std::atomic<unsigned short> invocation_id_sequencer;
    // std::mutex map_lock; - we don't need a lock on the result map anymore.
    using lock_t = std::unique_lock<std::mutex>;

    /* use this from within a derived class to retrieve precisely this RemoteInvoker
     * (this way, all the inherited RemoteInvoker methods in the subclass do not need
     * to worry about type collisions)*/
    inline RemoteInvoker& get_invoker(
            std::integral_constant<FunctionTag, Tag> const* const,
            const Args&...) {
        return *this;
    }

    using barray = char*;
    using cbarray = const char*;

    inline auto serialize_one(barray) { return 0; }

    template <typename A, typename... Rest>
    inline auto serialize_one(barray v, const A& a, const Rest&... rest) {
        auto size = mutils::to_bytes(a, v);
        return size + serialize_one(v + size, rest...);
    }

    inline auto serialize_all(barray v, const Args&... args) {
        return serialize_one(v, args...);
    }

    struct send_return {
        std::size_t size;
        char* buf;
        QueryResults<Ret> results;
        PendingResults<Ret>& pending;
    };

    send_return send(const std::function<char*(int)>& out_alloc,
                     const std::decay_t<Args>&... remote_args) {
        // auto invocation_id = mutils::long_rand();
        std::size_t invocation_id = invocation_id_sequencer++;
        invocation_id %= MAX_CONCURRENT_RPCS_PER_INVOKER;
        std::size_t size = mutils::bytes_size(invocation_id);
        {
            auto t = {std::size_t{0}, std::size_t{0}, mutils::bytes_size(remote_args)...};
            size += std::accumulate(t.begin(), t.end(), 0);
        }
        char* serialized_args = out_alloc(size);
        {
            auto v = serialized_args + mutils::to_bytes(invocation_id, serialized_args);
            auto check_size = mutils::bytes_size(invocation_id) + serialize_all(v, remote_args...);
            assert_always(check_size == size);
        }

        // lock_t l{map_lock};
        results_vector[invocation_id].reset();
        PendingResults<Ret>& pending_results = results_vector[invocation_id];

        dbg_default_trace("Ready to send an RPC call message with invocation ID {}", invocation_id);
        return send_return{size, serialized_args, pending_results.get_future(),
                           pending_results};
    }

    template <typename definitely_char>
    inline recv_ret receive_response(
            std::false_type*,
            mutils::DeserializationManager* dsm,
            const node_id_t& nid, const char* response,
            const std::function<definitely_char*(int)>&) {
        //note: find where this exception is set on the sending side!
        bool is_exception = response[0];
        long int invocation_id = ((long int*)(response + 1))[0];
        // lock_t l{map_lock};
        // we unlock the map here to avoid the deadlock:
        // The p2p handler thread, on receiving an RPC REPLY may get this lock
        // before sst_detect thread finish handling the corresponding ordered
        // send locally. However, sst_detect thread may be so slow that it
        // hasn't finish deliverying of a previous ordered send, therefore waits
        // for the map_lock in the following call stack:
        // - RPCManager::rpc_message_handler() ->
        // - RPCManager::parse_and_receive() ->
        // - RPCManager::receive_message()->
        // - receivers->at(indx)->
        // - RemoteInvoker::receive_response().
        // therefore, the promise for the next ordered_send, on which the
        // p2p handler thread is waiting for, cannot be fulfilled.
        // dead lock!!!
        // We use this workaround by just release the results_map lock as early
        // because we have 64K slots and we assume after 64K messages, this
        // corresponding ordered_send has been finished already.
        // TODO: make a better plan along with garbage collection.
        // l.unlock();
        // TODO: garbage collection for the responses.
        // More on May-7th (Weijia):
        // We don't need to lock the results container (now results_vector)
        // anymore. The vector is initialized at the beginning. The p2p rpc
        // reuse the slot by reseting it. No deleting, insertion, or
        // replacement operations at all. But we still need a better garbage
        // collection mechanism.
        if(is_exception) {
            results_vector[invocation_id].set_exception(nid, std::make_exception_ptr(remote_exception_occurred{nid}));
        } else {
            dbg_default_trace("Received an RPC response for invocation ID {} from node {}", invocation_id, nid);
            results_vector[invocation_id].set_value(nid, *mutils::from_bytes<Ret>(dsm, response + 1 + sizeof(invocation_id)));
        }
        return recv_ret{Opcode(), 0, nullptr, nullptr};
    }

    inline recv_ret receive_response(std::true_type*,
                                     mutils::DeserializationManager*,
                                     const node_id_t& nid, const char* response,
                                     const std::function<char*(int)>&) {
        if(response[0]) throw remote_exception_occurred{nid};
        assert_always(false && "was not expecting a response!");
    }

    inline recv_ret receive_response(  //mutils::DeserializationManager* dsm,
            mutils::RemoteDeserialization_v* rdv,
            const node_id_t& nid, const char* response,
            const std::function<char*(int)>& f) {
        constexpr std::is_same<void, Ret>* choice{nullptr};
        //        return receive_response(choice, dsm, nid, response, f);
        mutils::DeserializationManager dsm{*rdv};
        return receive_response(choice, &dsm, nid, response, f);
    }

    inline void fulfill_pending_results_vector(long int invocation_id, const node_list_t& who) {
        // I think this function is never called
        assert_always(false);
        results_vector[invocation_id].fulfill_map(who);
    }

    RemoteInvoker(uint32_t class_id, uint32_t instance_id,
                  std::map<Opcode, receive_fun_t>& receivers)
            : invoke_opcode{class_id, instance_id, Tag, false},
              reply_opcode{class_id, instance_id, Tag, true},
              invocation_id_sequencer(0) {
        receivers.emplace(reply_opcode, [this](auto... a) {
            return this->receive_response(a...);
        });
    }
};

template <FunctionTag Tag, typename Ret, typename... Args>
struct RemoteInvocable<Tag, std::function<Ret(Args...)>> {
    using remote_function_type = std::function<Ret(Args...)>;
    const remote_function_type remote_invocable_function;
    const Opcode invoke_opcode;
    const Opcode reply_opcode;

    /* use this from within a derived class to retrieve precisely this RemoteInvocable
     * (this way, all the inherited RemoteInvocable methods in the subclass do not need
     * to worry about type collisions)*/
    inline RemoteInvocable& get_handler(
            std::integral_constant<FunctionTag, Tag> const* const,
            const Args&...) {
        return *this;
    }

    inline recv_ret receive_call(std::false_type const* const,
                                 mutils::DeserializationManager* dsm,
                                 const node_id_t& caller, const char* _recv_buf,
                                 const std::function<char*(int)>& out_alloc) {
        long int invocation_id = ((long int*)_recv_buf)[0];
        auto recv_buf = _recv_buf + sizeof(long int);
        try {
            const auto result = mutils::deserialize_and_run(dsm, recv_buf, remote_invocable_function);
            const auto result_size = mutils::bytes_size(result) + sizeof(long int) + 1;
            auto out = out_alloc(result_size);
            out[0] = false;
            ((long int*)(out + 1))[0] = invocation_id;
            mutils::to_bytes(result, out + sizeof(invocation_id) + 1);
            dbg_default_trace("Ready to send an RPC reply for invocation ID {} to node {}", invocation_id, caller);
            return recv_ret{reply_opcode, result_size, out, nullptr};
        } catch(...) {
            char* out = out_alloc(sizeof(long int) + 1);
            out[0] = true;
            ((long int*)(out + 1))[0] = invocation_id;
            return recv_ret{reply_opcode, sizeof(long int) + 1, out,
                            std::current_exception()};
        }
    }

    inline recv_ret receive_call(std::true_type const* const,
                                 mutils::DeserializationManager* dsm,
                                 const node_id_t&, const char* _recv_buf,
                                 const std::function<char*(int)>&) {
        //TODO: Need to catch exceptions here, and possibly send them back, since void functions can still throw exceptions!
        auto recv_buf = _recv_buf + sizeof(long int);
        mutils::deserialize_and_run(dsm, recv_buf, remote_invocable_function);
        return recv_ret{reply_opcode, 0, nullptr};
    }

    inline recv_ret receive_call(  //mutils::DeserializationManager* dsm,
            mutils::RemoteDeserialization_v* rdv,
            const node_id_t& who, const char* recv_buf,
            const std::function<char*(int)>& out_alloc) {
        constexpr std::is_same<Ret, void>* choice{nullptr};
        //      return this->receive_call(choice, dsm, who, recv_buf, out_alloc);
        mutils::DeserializationManager dsm{*rdv};
        return this->receive_call(choice, &dsm, who, recv_buf, out_alloc);
    }

    RemoteInvocable(uint32_t class_id, uint32_t instance_id,
                    std::map<Opcode, receive_fun_t>& receivers,
                    std::function<Ret(Args...)> f)
            : remote_invocable_function(f),
              invoke_opcode{class_id, instance_id, Tag, false},
              reply_opcode{class_id, instance_id, Tag, true} {
        receivers.emplace(invoke_opcode, [this](auto... a) {
            return this->receive_call(a...);
        });
    }
};

template <FunctionTag Tag, typename NotAFunction>
struct wrapped;

template <FunctionTag Tag, typename Ret, typename... Arguments>
struct wrapped<Tag, std::function<Ret(Arguments...)>> {
    using fun_t = std::function<Ret(Arguments...)>;
    fun_t fun;
};

template <FunctionTag Tag, typename Ret, typename Class, typename... Arguments>
struct partial_wrapped {
    using fun_t = Ret (Class::*)(Arguments...);
    fun_t fun;
};

template <FunctionTag Tag, typename Ret, typename Class, typename... Arguments>
struct const_partial_wrapped {
    using fun_t = Ret (Class::*)(Arguments...) const;
    fun_t fun;
};

template <typename NewClass, FunctionTag Tag, typename Ret, typename... Args>
wrapped<Tag, std::function<Ret(Args...)>> bind_to_instance(std::unique_ptr<NewClass>* _this,
                                                           const partial_wrapped<Tag, Ret, NewClass, Args...>& partial) {
    assert(_this);
    return wrapped<Tag, std::function<Ret(Args...)>>{
            [_this, fun = partial.fun](Args... arguments) {
                assert(_this);
                assert(_this->get());
                return (_this->get()->*fun)(arguments...);
            }};
}

template <typename NewClass, FunctionTag Tag, typename Ret, typename... Args>
wrapped<Tag, std::function<Ret(Args...)>> bind_to_instance(std::unique_ptr<NewClass>* _this,
                                                           const const_partial_wrapped<Tag, Ret, NewClass, Args...>& partial) {
    assert(_this);
    return wrapped<Tag, std::function<Ret(Args...)>>{
            [_this, fun = partial.fun](Args... arguments) {
                return (_this->get()->*fun)(arguments...);
            }};
}

template <FunctionTag Tag, typename NewClass, typename Ret, typename... Args>
partial_wrapped<Tag, Ret, NewClass, Args...> tag(Ret (NewClass::*fun)(Args...)) {
    static_assert(!std::is_reference<Ret>::value && !std::is_pointer<Ret>::value, "RPC-registered functions cannot return references or pointers!");
    //Fold-expression asserts the boolean expression for all Args in the parameter pack
    static_assert(((std::is_reference<Args>::value || sizeof(Args) < 2 * sizeof(void*)) && ...), "RPC-registered functions must take non-pointer-size arguments by reference to avoid extra copying.");
    return partial_wrapped<Tag, Ret, NewClass, Args...>{fun};
}

template <FunctionTag Tag, typename NewClass, typename Ret, typename... Args>
const_partial_wrapped<Tag, Ret, NewClass, Args...> tag(Ret (NewClass::*fun)(Args...) const) {
    static_assert(!std::is_reference<Ret>::value && !std::is_pointer<Ret>::value, "RPC-registered functions cannot return references or pointers!");
    static_assert(((std::is_reference<Args>::value || sizeof(Args) < 2 * sizeof(void*)) && ...), "RPC-registered functions must take non-pointer-size arguments by reference to avoid extra copying.");
    return const_partial_wrapped<Tag, Ret, NewClass, Args...>{fun};
}

/* Technically, RemoteInvocablePairs specializes this template for the cases
 * where the parameter pack is a list of types of the form wrapped<id, FunType>
 * However, there is only one specialization, so using RemoteInvocablePairs for
 * anything other than wrapped<id, FunType> is meaningless.
 */
template <typename...>
struct RemoteInvocablePairs;

template <FunctionTag id, typename FunType>
struct RemoteInvocablePairs<wrapped<id, FunType>>
        : public RemoteInvoker<id, FunType>, public RemoteInvocable<id, FunType> {
    RemoteInvocablePairs(uint32_t class_id,
                         uint32_t instance_id,
                         std::map<Opcode, receive_fun_t>& receivers, FunType function_ptr)
            : RemoteInvoker<id, FunType>(class_id, instance_id, receivers),
              RemoteInvocable<id, FunType>(class_id, instance_id, receivers, function_ptr) {}

    using RemoteInvoker<id, FunType>::get_invoker;
    using RemoteInvocable<id, FunType>::get_handler;
};

template <FunctionTag id, typename FunType, typename... rest>
struct RemoteInvocablePairs<wrapped<id, FunType>, rest...>
        : public RemoteInvoker<id, FunType>, public RemoteInvocable<id, FunType>, public RemoteInvocablePairs<rest...> {
    template <typename... RestFunTypes>
    RemoteInvocablePairs(uint32_t class_id,
                         uint32_t instance_id,
                         std::map<Opcode, receive_fun_t>& receivers,
                         FunType function_ptr,
                         RestFunTypes&&... function_ptrs)
            : RemoteInvoker<id, FunType>(class_id, instance_id, receivers),
              RemoteInvocable<id, FunType>(class_id, instance_id, receivers, function_ptr),
              RemoteInvocablePairs<rest...>(class_id, instance_id, receivers, std::forward<RestFunTypes>(function_ptrs)...) {}

    //Ensure the inherited functions from RemoteInvoker and RemoteInvokable are visible
    using RemoteInvoker<id, FunType>::get_invoker;
    using RemoteInvocable<id, FunType>::get_handler;
    using RemoteInvocablePairs<rest...>::get_invoker;
    using RemoteInvocablePairs<rest...>::get_handler;
};

template <typename...>
struct RemoteInvokers;

template <FunctionTag Tag, typename FunType>
struct RemoteInvokers<wrapped<Tag, FunType>> : public RemoteInvoker<Tag, FunType> {
    RemoteInvokers(uint32_t class_id,
                   uint32_t instance_id,
                   std::map<Opcode, receive_fun_t>& receivers)
            : RemoteInvoker<Tag, FunType>(class_id, instance_id, receivers) {}

    using RemoteInvoker<Tag, FunType>::get_invoker;
};

template <FunctionTag Tag, typename FunType, typename... RestWrapped>
struct RemoteInvokers<wrapped<Tag, FunType>, RestWrapped...>
        : public RemoteInvoker<Tag, FunType>, public RemoteInvokers<RestWrapped...> {
    RemoteInvokers(uint32_t class_id,
                   uint32_t instance_id,
                   std::map<Opcode, receive_fun_t>& receivers)
            : RemoteInvoker<Tag, FunType>(class_id, instance_id, receivers),
              RemoteInvokers<RestWrapped...>(class_id, instance_id, receivers) {}

    using RemoteInvoker<Tag, FunType>::get_invoker;
    using RemoteInvokers<RestWrapped...>::get_invoker;
};

template <class IdentifyingClass, typename... WrappedFuns>
class RemoteInvocableClass : private RemoteInvocablePairs<WrappedFuns...> {
public:
    const node_id_t nid;

    RemoteInvocableClass(node_id_t nid, uint32_t type_id, uint32_t instance_id,
                         std::map<Opcode, receive_fun_t>& rvrs, const WrappedFuns&... fs)
            : RemoteInvocablePairs<WrappedFuns...>(type_id, instance_id, rvrs, fs.fun...),
              nid(nid) {}

    template <FunctionTag Tag, typename... Args>
    std::size_t get_size_for_ordered_send(Args&&... a) {
        //only used for size calculation
        long int invocation_id = 0;
        std::size_t function_call_size = mutils::bytes_size(invocation_id);
        {
            function_call_size += (0 + ... + mutils::bytes_size(a));
        }
        //Add the header_size that send() adds to the invoker's out_alloc
        return function_call_size + remote_invocation_utilities::header_space();
    }

    template <FunctionTag Tag, typename... Args>
    auto* getReturnType(Args&&... args) {
        constexpr std::integral_constant<FunctionTag, Tag>* choice{nullptr};
        return this->get_invoker(choice, args...).returnRet();
    }

    template <FunctionTag Tag, typename... Args>
    auto send(const std::function<char*(int)>& out_alloc, Args&&... args) {
        using namespace remote_invocation_utilities;

        constexpr std::integral_constant<FunctionTag, Tag>* choice{nullptr};
        auto& invoker = this->get_invoker(choice, args...);
        const auto header_size = header_space();
        auto sent_return = invoker.send(
                [&out_alloc, &header_size](std::size_t size) {
                    return out_alloc(size + header_size) + header_size;
                },
                std::forward<Args>(args)...);

        std::size_t payload_size = sent_return.size;
        char* buf = sent_return.buf - header_size;
        uint32_t flags = 0;
        /*
         set the cascading flag if necessary.
         This is not important because, unlike p2p_send, ordered_send/query
         does not distinguish cascading and non cascading sends. However,
         to keep the format consistency, we keep the flags field in the RPC
         message header reserved for future use.

        if (in_rpc_handler()) {
            RPC_HEADER_FLAG_SET(flags,CASCADE);
            dbg_default_info("send cascading message.");
        }
        */
        populate_header(buf, payload_size, invoker.invoke_opcode, nid, flags);

        using Ret = typename decltype(sent_return.results)::type;
        /*
          much like previous definition, except with
          two fewer fields
        */
        struct send_return {
            QueryResults<Ret> results;
            PendingResults<Ret>& pending;
            //LifeTracker
            /*
class LifeTracker {
std::function<void () > deleter;
~LifeTraker(){
deleter();
}
};

LifeTracker{[invocation_id, &map](){map.erase(invocation_id);}};
       */
        };
        return send_return{std::move(sent_return.results),
                           sent_return.pending};
    }

    using specialized_to = IdentifyingClass;
    RemoteInvocableClass& for_class(IdentifyingClass*) {
        return *this;
    }
};

template <class IdentifyingClass>
class RemoteInvocableClass<IdentifyingClass> {
public:
    const node_id_t nid;

    RemoteInvocableClass(node_id_t nid, uint32_t type_id, uint32_t instance_id,
                         std::map<Opcode, receive_fun_t>& rvrs)
            : nid(nid) {}

    template <FunctionTag Tag, typename... Args>
    std::size_t get_size(Args&&... a) {
        auto invocation_id = mutils::long_rand();
        std::size_t size = mutils::bytes_size(invocation_id);
        {
            auto t = {std::size_t{0}, std::size_t{0}, mutils::bytes_size(a)...};
            size += std::accumulate(t.begin(), t.end(), 0);
        }
        return size;
    }

    template <FunctionTag Tag, typename... Args>
    auto* getReturnType(Args&&... args) {
        void* null_and_void(nullptr);
        return null_and_void;
    }

    using specialized_to = IdentifyingClass;
    RemoteInvocableClass& for_class(IdentifyingClass*) {
        return *this;
    }
};

template <class IdentifyingClass, typename... WrappedFuns>
auto build_remote_invocable_class(const node_id_t nid, const uint32_t type_id, const uint32_t instance_id,
                                  std::map<Opcode, receive_fun_t>& rvrs,
                                  const WrappedFuns&... fs) {
    return std::make_unique<RemoteInvocableClass<IdentifyingClass, WrappedFuns...>>(nid, type_id, instance_id, rvrs, fs...);
}

template <class IdentifyingClass, typename... WrappedFuns>
class RemoteInvokerForClass : private RemoteInvokers<WrappedFuns...> {
public:
    const node_id_t nid;

    RemoteInvokerForClass(node_id_t nid, uint32_t type_id, uint32_t instance_id,
                          std::map<Opcode, receive_fun_t>& rvrs)
            : RemoteInvokers<WrappedFuns...>(type_id, instance_id, rvrs),
              nid(nid) {}

    template <FunctionTag Tag, typename... Args>
    auto send(const std::function<char*(int)>& out_alloc, Args&&... args) {
        using namespace remote_invocation_utilities;

        constexpr std::integral_constant<FunctionTag, Tag>* choice{nullptr};
        auto& invoker = this->get_invoker(choice, args...);
        const auto header_size = header_space();
        auto sent_return = invoker.send(
                [&out_alloc, &header_size](std::size_t size) {
                    return out_alloc(size + header_size) + header_size;
                },
                std::forward<Args>(args)...);

        std::size_t payload_size = sent_return.size;
        char* buf = sent_return.buf - header_size;
        uint32_t flags = 0;
        if(in_rpc_handler()) {
            RPC_HEADER_FLAG_SET(flags, CASCADE);
            dbg_default_info("sending cascading RPC.");
        }
        populate_header(buf, payload_size, invoker.invoke_opcode, nid, flags);

        using Ret = typename decltype(sent_return.results)::type;
        /*
          much like previous definition, except with
          two fewer fields
        */
        struct send_return {
            QueryResults<Ret> results;
            PendingResults<Ret>& pending;
        };
        return send_return{std::move(sent_return.results),
                           sent_return.pending};
    }
};

template <class IdentifyingClass>
class RemoteInvokerForClass<IdentifyingClass> {
public:
    const node_id_t nid;

    RemoteInvokerForClass(node_id_t nid, uint32_t type_id, uint32_t instance_id,
                          std::map<Opcode, receive_fun_t>& rvrs)
            : nid(nid) {}
};

template <class IdentifyingClass, typename... WrappedFuns>
auto build_remote_invoker_for_class(const node_id_t nid, const uint32_t type_id, const uint32_t instance_id,
                                    std::map<Opcode, receive_fun_t>& rvrs) {
    return std::make_unique<RemoteInvokerForClass<IdentifyingClass, WrappedFuns...>>(nid, type_id, instance_id, rvrs);
}
}  // namespace rpc
}  // namespace derecho