rpc.hh

/*
 * This file is open source software, licensed to you under the terms
 * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
 * distributed with this work for additional information regarding copyright
 * ownership.  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.
 */
/*
 * Copyright (C) 2015 Cloudius Systems, Ltd.
 */
 
#pragma once
 
#include <unordered_map>
#include <unordered_set>
#include <list>
#include <variant>
#include <boost/intrusive/list.hpp>
#include <seastar/core/future.hh>
#include <seastar/core/seastar.hh>
#include <seastar/net/api.hh>
#include <seastar/core/iostream.hh>
#include <seastar/core/shared_ptr.hh>
#include <seastar/core/condition-variable.hh>
#include <seastar/core/gate.hh>
#include <seastar/rpc/rpc_types.hh>
#include <seastar/core/byteorder.hh>
#include <seastar/core/shared_future.hh>
#include <seastar/core/queue.hh>
#include <seastar/core/weak_ptr.hh>
#include <seastar/core/scheduling.hh>
#include <seastar/util/backtrace.hh>
#include <seastar/util/log.hh>
 
namespace bi = boost::intrusive;
 
namespace seastar {
 
namespace rpc {
 
 
using id_type = int64_t;
 
using rpc_semaphore = basic_semaphore<semaphore_default_exception_factory, rpc_clock_type>;
using resource_permit = semaphore_units<semaphore_default_exception_factory, rpc_clock_type>;
 
static constexpr char rpc_magic[] = "SSTARRPC";
 
 
struct isolation_config {
    scheduling_group sched_group = current_scheduling_group();
};
 
isolation_config default_isolate_connection(sstring isolation_cookie);
 
struct resource_limits {
    size_t basic_request_size = 0; 
    unsigned bloat_factor = 1;     
    size_t max_memory = rpc_semaphore::max_counter(); 
    using syncronous_isolation_function = std::function<isolation_config (sstring isolation_cookie)>;
    using asyncronous_isolation_function = std::function<future<isolation_config> (sstring isolation_cookie)>;
    using isolation_function_alternatives = std::variant<syncronous_isolation_function, asyncronous_isolation_function>;
    isolation_function_alternatives isolate_connection = default_isolate_connection;
};
 
struct client_options {
    std::optional<net::tcp_keepalive_params> keepalive;
    bool tcp_nodelay = true;
    bool reuseaddr = false;
    compressor::factory* compressor_factory = nullptr;
    bool send_timeout_data = true;
    connection_id stream_parent = invalid_connection_id;
    sstring isolation_cookie;
    sstring metrics_domain = "default";
    bool send_handler_duration = true;
};
 
 
// RPC call that passes stream connection id as a parameter
// may arrive to a different shard from where the stream connection
// was opened, so the connection id is not known to a server that handles
// the RPC call. The shard that the stream connection belong to is know
// since it is a part of connection id, but this is not enough to locate
// a server instance the connection belongs to if there are more than one
// server on the shard. Stream domain parameter is here to help with that.
// Different servers on all shards logically belonging to the same service should
// belong to the same streaming domain. Only one server on each shard can belong to
// a particulr streaming domain.
class streaming_domain_type {
    uint64_t _id;
public:
    explicit streaming_domain_type(uint64_t id) : _id(id) {}
    bool operator==(const streaming_domain_type& o) const {
        return _id == o._id;
    }
    friend struct std::hash<streaming_domain_type>;
    friend std::ostream& operator<<(std::ostream&, const streaming_domain_type&);
};
 
 
class server;
 
struct server_options {
    compressor::factory* compressor_factory = nullptr;
    bool tcp_nodelay = true;
    std::optional<streaming_domain_type> streaming_domain;
    server_socket::load_balancing_algorithm load_balancing_algorithm = server_socket::load_balancing_algorithm::default_;
    // optional filter function. If set, will be called with remote
    // (connecting) address.
    // Returning false will refuse the incoming connection.
    // Returning true will allow the mechanism to proceed.
    std::function<bool(const socket_address&)> filter_connection = {};
};
 
 
inline
size_t
estimate_request_size(const resource_limits& lim, size_t serialized_size) {
    return lim.basic_request_size + serialized_size * lim.bloat_factor;
}
 
struct negotiation_frame {
    char magic[sizeof(rpc_magic) - 1];
    uint32_t len; // additional negotiation data length; multiple negotiation_frame_feature_record structs
};
 
enum class protocol_features : uint32_t {
    COMPRESS = 0,
    TIMEOUT = 1,
    CONNECTION_ID = 2,
    STREAM_PARENT = 3,
    ISOLATION = 4,
    HANDLER_DURATION = 5,
};
 
// internal representation of feature data
using feature_map = std::map<protocol_features, sstring>;
 
// An rpc signature, in the form signature<Ret (In0, In1, In2)>.
template <typename Function>
struct signature;
 
class logger {
    ::seastar::logger* _seastar_logger = nullptr;
 
    void log(const sstring& str) const {
        if (_seastar_logger) {
            // default level for log messages is `info`
            _seastar_logger->info("{}", str);
        }
    }
 
    // _seastar_logger will always be used first if it's available
    template <typename... Args>
#ifdef SEASTAR_LOGGER_COMPILE_TIME_FMT
    void log(log_level level, fmt::format_string<Args...> fmt, Args&&... args) const {
#else
    void log(log_level level, const char* fmt, Args&&... args) const {
#endif
        if (_seastar_logger) {
            _seastar_logger->log(level, fmt, std::forward<Args>(args)...);
        }
    }
 
public:
    void set(::seastar::logger* logger) {
        _seastar_logger = logger;
    }
 
    void operator()(const client_info& info, id_type msg_id, const sstring& str) const;
    void operator()(const client_info& info, id_type msg_id, log_level level, std::string_view str) const;
 
    void operator()(const client_info& info, const sstring& str) const;
    void operator()(const client_info& info, log_level level, std::string_view str) const;
 
    void operator()(const socket_address& addr, const sstring& str) const;
    void operator()(const socket_address& addr, log_level level, std::string_view str) const;
};
 
class connection {
protected:
    connected_socket _fd;
    input_stream<char> _read_buf;
    output_stream<char> _write_buf;
    bool _error = false;
    bool _connected = false;
    std::optional<shared_promise<>> _negotiated = shared_promise<>();
    promise<> _stopped;
    stats _stats;
    const logger& _logger;
    // The owner of the pointer below is an instance of rpc::protocol<typename Serializer> class.
    // The type of the pointer is erased here, but the original type is Serializer
    void* _serializer;
    struct outgoing_entry : public bi::list_base_hook<bi::link_mode<bi::auto_unlink>> {
        timer<rpc_clock_type> t;
        snd_buf buf;
        promise<> done;
        cancellable* pcancel = nullptr;
        outgoing_entry(snd_buf b) : buf(std::move(b)) {}
 
        outgoing_entry(outgoing_entry&&) = delete;
        outgoing_entry(const outgoing_entry&) = delete;
 
        void uncancellable() {
            t.cancel();
            if (pcancel) {
                pcancel->cancel_send = std::function<void()>();
            }
        }
 
        ~outgoing_entry() {
            if (pcancel) {
                pcancel->cancel_send = std::function<void()>();
                pcancel->send_back_pointer = nullptr;
            }
        }
 
        using container_t = bi::list<outgoing_entry, bi::constant_time_size<false>>;
    };
    void withdraw(outgoing_entry::container_t::iterator it, std::exception_ptr ex = nullptr);
    future<> _outgoing_queue_ready = _negotiated->get_shared_future();
    outgoing_entry::container_t _outgoing_queue;
    size_t _outgoing_queue_size = 0;
    std::unique_ptr<compressor> _compressor;
    bool _propagate_timeout = false;
    bool _timeout_negotiated = false;
    bool _handler_duration_negotiated = false;
    // stream related fields
    bool _is_stream = false;
    connection_id _id = invalid_connection_id;
 
    std::unordered_map<connection_id, xshard_connection_ptr> _streams;
    queue<rcv_buf> _stream_queue = queue<rcv_buf>(max_queued_stream_buffers);
    semaphore _stream_sem = semaphore(max_stream_buffers_memory);
    bool _sink_closed = true;
    bool _source_closed = true;
    // the future holds if sink is already closed
    // if it is not ready it means the sink is been closed
    future<bool> _sink_closed_future = make_ready_future<bool>(false);
 
    void set_negotiated() noexcept;
 
    bool is_stream() const noexcept {
        return _is_stream;
    }
 
    snd_buf compress(snd_buf buf);
    future<> send_buffer(snd_buf buf);
    future<> send(snd_buf buf, std::optional<rpc_clock_type::time_point> timeout = {}, cancellable* cancel = nullptr);
    future<> send_entry(outgoing_entry& d) noexcept;
    future<> stop_send_loop(std::exception_ptr ex);
    future<std::optional<rcv_buf>>  read_stream_frame_compressed(input_stream<char>& in);
    bool stream_check_twoway_closed() const noexcept {
        return _sink_closed && _source_closed;
    }
    future<> stream_close();
    future<> stream_process_incoming(rcv_buf&&);
    future<> handle_stream_frame();
 
public:
    connection(connected_socket&& fd, const logger& l, void* s, connection_id id = invalid_connection_id) : connection(l, s, id) {
        set_socket(std::move(fd));
    }
    connection(const logger& l, void* s, connection_id id = invalid_connection_id) : _logger(l), _serializer(s), _id(id) {}
    virtual ~connection() {}
    size_t outgoing_queue_length() const noexcept {
        return _outgoing_queue_size;
    }
 
    void set_socket(connected_socket&& fd);
    future<> send_negotiation_frame(feature_map features);
    bool error() const noexcept { return _error; }
    void abort();
    future<> stop() noexcept;
    future<> stream_receive(circular_buffer<foreign_ptr<std::unique_ptr<rcv_buf>>>& bufs);
    future<> close_sink() {
        _sink_closed = true;
        if (stream_check_twoway_closed()) {
            return stream_close();
        }
        return make_ready_future();
    }
    bool sink_closed() const noexcept {
        return _sink_closed;
    }
    future<> close_source() {
        _source_closed = true;
        if (stream_check_twoway_closed()) {
            return stream_close();
        }
        return make_ready_future();
    }
    connection_id get_connection_id() const noexcept {
        return _id;
    }
    stats& get_stats_internal() noexcept {
        return _stats;
    }
    xshard_connection_ptr get_stream(connection_id id) const;
    void register_stream(connection_id id, xshard_connection_ptr c);
    virtual socket_address peer_address() const = 0;
 
    const logger& get_logger() const noexcept {
        return _logger;
    }
 
    template<typename Serializer>
    Serializer& serializer() {
        return *static_cast<Serializer*>(_serializer);
    }
 
    template <typename FrameType>
    future<typename FrameType::return_type> read_frame(socket_address info, input_stream<char>& in);
 
    template <typename FrameType>
    future<typename FrameType::return_type> read_frame_compressed(socket_address info, std::unique_ptr<compressor>& compressor, input_stream<char>& in);
    friend class client;
    template<typename Serializer, typename... Out>
    friend class sink_impl;
    template<typename Serializer, typename... In>
    friend class source_impl;
 
    void suspend_for_testing(promise<>& p) {
        _outgoing_queue_ready.get();
        auto dummy = std::make_unique<outgoing_entry>(snd_buf());
        _outgoing_queue.push_back(*dummy);
        _outgoing_queue_ready = dummy->done.get_future();
        (void)p.get_future().then([dummy = std::move(dummy)] { dummy->done.set_value(); });
    }
};
 
struct deferred_snd_buf {
    promise<> pr;
    snd_buf data;
};
 
// send data Out...
template<typename Serializer, typename... Out>
class sink_impl : public sink<Out...>::impl {
    // Used on the shard *this lives on.
    alignas (cache_line_size) uint64_t _next_seq_num = 1;
 
    // Used on the shard the _conn lives on.
    struct alignas (cache_line_size) {
        uint64_t last_seq_num = 0;
        std::map<uint64_t, deferred_snd_buf> out_of_order_bufs;
    } _remote_state;
public:
    sink_impl(xshard_connection_ptr con) : sink<Out...>::impl(std::move(con)) { this->_con->get()->_sink_closed = false; }
    future<> operator()(const Out&... args) override;
    future<> close() override;
    future<> flush() override;
    ~sink_impl() override;
};
 
// receive data In...
template<typename Serializer, typename... In>
class source_impl : public source<In...>::impl {
public:
    source_impl(xshard_connection_ptr con) : source<In...>::impl(std::move(con)) { this->_con->get()->_source_closed = false; }
    future<std::optional<std::tuple<In...>>> operator()() override;
};
 
class client : public rpc::connection, public weakly_referencable<client> {
    socket _socket;
    id_type _message_id = 1;
    struct reply_handler_base {
        timer<rpc_clock_type> t;
        cancellable* pcancel = nullptr;
        rpc_clock_type::time_point start;
        virtual void operator()(client&, id_type, rcv_buf data) = 0;
        virtual void timeout() {}
        virtual void cancel() {}
        virtual ~reply_handler_base() {
            if (pcancel) {
                pcancel->cancel_wait = std::function<void()>();
                pcancel->wait_back_pointer = nullptr;
            }
        };
    };
 
    class metrics {
        struct domain;
 
        using domain_member_hook_t = boost::intrusive::list_member_hook<boost::intrusive::link_mode<boost::intrusive::auto_unlink>>;
 
        const client& _c;
        domain& _domain;
        domain_member_hook_t _link;
 
        using domain_list_t = boost::intrusive::list<metrics,
              boost::intrusive::member_hook<metrics, metrics::domain_member_hook_t, &metrics::_link>,
              boost::intrusive::constant_time_size<false>>;
 
    public:
        metrics(const client&);
        ~metrics();
    };
 
    void enqueue_zero_frame();
public:
    template<typename Reply, typename Func>
    struct reply_handler final : reply_handler_base {
        Func func;
        Reply reply;
        reply_handler(Func&& f) : func(std::move(f)) {}
        virtual void operator()(client& client, id_type msg_id, rcv_buf data) override {
            return func(reply, client, msg_id, std::move(data));
        }
        virtual void timeout() override {
            reply.done = true;
            reply.p.set_exception(timeout_error());
        }
        virtual void cancel() override {
            reply.done = true;
            reply.p.set_exception(canceled_error());
        }
        virtual ~reply_handler() {}
    };
private:
    std::unordered_map<id_type, std::unique_ptr<reply_handler_base>> _outstanding;
    socket_address _server_addr, _local_addr;
    client_options _options;
    weak_ptr<client> _parent; // for stream clients
 
    metrics _metrics;
 
private:
    future<> negotiate_protocol(feature_map map);
    void negotiate(feature_map server_features);
    // Returned future is
    // - message id
    // - optional server-side handler duration
    // - message payload
    future<std::tuple<int64_t, std::optional<uint32_t>, std::optional<rcv_buf>>>
    read_response_frame_compressed(input_stream<char>& in);
public:
    client(const logger& l, void* s, const socket_address& addr, const socket_address& local = {});
    client(const logger& l, void* s, client_options options, const socket_address& addr, const socket_address& local = {});
 
    client(const logger& l, void* s, socket socket, const socket_address& addr, const socket_address& local = {});
    client(const logger& l, void* s, client_options options, socket socket, const socket_address& addr, const socket_address& local = {});
 
    stats get_stats() const;
    size_t incoming_queue_length() const noexcept {
        return _outstanding.size();
    }
 
    auto next_message_id() { return _message_id++; }
    void wait_for_reply(id_type id, std::unique_ptr<reply_handler_base>&& h, std::optional<rpc_clock_type::time_point> timeout, cancellable* cancel);
    void wait_timed_out(id_type id);
    future<> stop() noexcept;
    void abort_all_streams();
    void deregister_this_stream();
    socket_address peer_address() const override {
        return _server_addr;
    }
    future<> await_connection() {
        if (!_negotiated) {
            return make_ready_future<>();
        } else {
            return _negotiated->get_shared_future();
        }
    }
    template<typename Serializer, typename... Out>
    future<sink<Out...>> make_stream_sink(socket socket) {
        return await_connection().then([this, socket = std::move(socket)] () mutable {
            if (!this->get_connection_id()) {
                return make_exception_future<sink<Out...>>(std::runtime_error("Streaming is not supported by the server"));
            }
            client_options o = _options;
            o.stream_parent = this->get_connection_id();
            o.send_timeout_data = false;
            o.metrics_domain += "_stream";
            auto c = make_shared<client>(_logger, _serializer, o, std::move(socket), _server_addr, _local_addr);
            c->_parent = this->weak_from_this();
            c->_is_stream = true;
            return c->await_connection().then([c, this] {
                if (_error) {
                    throw closed_error();
                }
                xshard_connection_ptr s = make_lw_shared(make_foreign(static_pointer_cast<rpc::connection>(c)));
                this->register_stream(c->get_connection_id(), s);
                return sink<Out...>(make_shared<sink_impl<Serializer, Out...>>(std::move(s)));
            }).handle_exception([c] (std::exception_ptr eptr) {
                // If await_connection fails we need to stop the client
                // before destroying it.
                return c->stop().then([eptr, c] {
                    return make_exception_future<sink<Out...>>(eptr);
                });
            });
        });
    }
    template<typename Serializer, typename... Out>
    future<sink<Out...>> make_stream_sink() {
        return make_stream_sink<Serializer, Out...>(make_socket());
    }
 
    future<> request(uint64_t type, int64_t id, snd_buf buf, std::optional<rpc_clock_type::time_point> timeout = {}, cancellable* cancel = nullptr);
};
 
class protocol_base;
 
class server {
private:
    static thread_local std::unordered_map<streaming_domain_type, server*> _servers;
 
public:
    class connection : public rpc::connection, public enable_shared_from_this<connection> {
        client_info _info;
        connection_id _parent_id = invalid_connection_id;
        std::optional<isolation_config> _isolation_config;
    private:
        future<> negotiate_protocol();
        future<std::tuple<std::optional<uint64_t>, uint64_t, int64_t, std::optional<rcv_buf>>>
        read_request_frame_compressed(input_stream<char>& in);
        future<feature_map> negotiate(feature_map requested);
        future<> send_unknown_verb_reply(std::optional<rpc_clock_type::time_point> timeout, int64_t msg_id, uint64_t type);
    public:
        connection(server& s, connected_socket&& fd, socket_address&& addr, const logger& l, void* seralizer, connection_id id);
        future<> process();
        future<> respond(int64_t msg_id, snd_buf&& data, std::optional<rpc_clock_type::time_point> timeout, std::optional<rpc_clock_type::duration> handler_duration);
        client_info& info() { return _info; }
        const client_info& info() const { return _info; }
        stats get_stats() const {
            stats res = _stats;
            res.pending = outgoing_queue_length();
            return res;
        }
        socket_address peer_address() const override {
            return _info.addr;
        }
        // Resources will be released when this goes out of scope
        future<resource_permit> wait_for_resources(size_t memory_consumed,  std::optional<rpc_clock_type::time_point> timeout) {
            if (timeout) {
                return get_units(get_server()._resources_available, memory_consumed, *timeout);
            } else {
                return get_units(get_server()._resources_available, memory_consumed);
            }
        }
        size_t estimate_request_size(size_t serialized_size) {
            return rpc::estimate_request_size(get_server()._limits, serialized_size);
        }
        size_t max_request_size() const {
            return get_server()._limits.max_memory;
        }
        server& get_server() {
            return _info.server;
        }
        const server& get_server() const {
            return _info.server;
        }
        future<> deregister_this_stream();
        future<> abort_all_streams();
    };
private:
    protocol_base& _proto;
    server_socket _ss;
    resource_limits _limits;
    rpc_semaphore _resources_available;
    std::unordered_map<connection_id, shared_ptr<connection>> _conns;
    promise<> _ss_stopped;
    gate _reply_gate;
    server_options _options;
    bool _shutdown = false;
    uint64_t _next_client_id = 1;
 
public:
    server(protocol_base* proto, const socket_address& addr, resource_limits memory_limit = resource_limits());
    server(protocol_base* proto, server_options opts, const socket_address& addr, resource_limits memory_limit = resource_limits());
    server(protocol_base* proto, server_socket, resource_limits memory_limit = resource_limits(), server_options opts = server_options{});
    server(protocol_base* proto, server_options opts, server_socket, resource_limits memory_limit = resource_limits());
    void accept();
    future<> stop();
    future<> shutdown();
    template<typename Func>
    void foreach_connection(Func&& f) {
        for (auto c : _conns) {
            f(*c.second);
        }
    }
    void abort_connection(connection_id id);
    gate& reply_gate() {
        return _reply_gate;
    }
    friend connection;
    friend client;
};
 
using rpc_handler_func = std::function<future<> (shared_ptr<server::connection>, std::optional<rpc_clock_type::time_point> timeout, int64_t msgid,
                                                 rcv_buf data, gate::holder guard)>;
 
struct rpc_handler {
    scheduling_group sg;
    rpc_handler_func func;
    gate use_gate;
};
 
class protocol_base {
public:
    virtual ~protocol_base() {};
    virtual shared_ptr<server::connection> make_server_connection(rpc::server& server, connected_socket fd, socket_address addr, connection_id id) = 0;
protected:
    friend class server;
 
    struct handler_with_holder {
        rpc_handler& handler;
        gate::holder holder;
    };
    virtual std::optional<handler_with_holder> get_handler(uint64_t msg_id) = 0;
};
 
 
template<typename Serializer, typename MsgType = uint32_t>
class protocol final : public protocol_base {
public:
    class server : public rpc::server {
    public:
        server(protocol& proto, const socket_address& addr, resource_limits memory_limit = resource_limits()) :
            rpc::server(&proto, addr, memory_limit) {}
        server(protocol& proto, server_options opts, const socket_address& addr, resource_limits memory_limit = resource_limits()) :
            rpc::server(&proto, opts, addr, memory_limit) {}
        server(protocol& proto, server_socket socket, resource_limits memory_limit = resource_limits(), server_options = server_options{}) :
            rpc::server(&proto, std::move(socket), memory_limit) {}
        server(protocol& proto, server_options opts, server_socket socket, resource_limits memory_limit = resource_limits()) :
            rpc::server(&proto, opts, std::move(socket), memory_limit) {}
    };
    class client : public rpc::client {
    public:
        /*
         * Create client object which will attempt to connect to the remote address.
         *
         * @param addr the remote address identifying this client
         * @param local the local address of this client
         */
        client(protocol& p, const socket_address& addr, const socket_address& local = {}) :
            rpc::client(p.get_logger(), &p._serializer, addr, local) {}
        client(protocol& p, client_options options, const socket_address& addr, const socket_address& local = {}) :
            rpc::client(p.get_logger(), &p._serializer, options, addr, local) {}
 
        client(protocol& p, socket socket, const socket_address& addr, const socket_address& local = {}) :
            rpc::client(p.get_logger(), &p._serializer, std::move(socket), addr, local) {}
        client(protocol& p, client_options options, socket socket, const socket_address& addr, const socket_address& local = {}) :
            rpc::client(p.get_logger(), &p._serializer, options, std::move(socket), addr, local) {}
    };
 
    friend server;
private:
    std::unordered_map<MsgType, rpc_handler> _handlers;
    Serializer _serializer;
    logger _logger;
 
public:
    protocol(Serializer&& serializer) : _serializer(std::forward<Serializer>(serializer)) {}
 
    template<typename Func>
    auto make_client(MsgType t);
 
    template<typename Func>
    auto register_handler(MsgType t, Func&& func);
 
    template <typename Func>
    auto register_handler(MsgType t, scheduling_group sg, Func&& func);
 
    future<> unregister_handler(MsgType t);
 
 
    void set_logger(::seastar::logger* logger) {
        _logger.set(logger);
    }
 
    const logger& get_logger() const {
        return _logger;
    }
 
    shared_ptr<rpc::server::connection> make_server_connection(rpc::server& server, connected_socket fd, socket_address addr, connection_id id) override {
        return make_shared<rpc::server::connection>(server, std::move(fd), std::move(addr), _logger, &_serializer, id);
    }
 
    bool has_handler(MsgType msg_id);
 
    bool has_handlers() const noexcept {
        return !_handlers.empty();
    }
 
private:
    std::optional<handler_with_holder> get_handler(uint64_t msg_id) override;
 
    template<typename Ret, typename... In>
    auto make_client(signature<Ret(In...)> sig, MsgType t);
 
    void register_receiver(MsgType t, rpc_handler&& handler) {
        auto r = _handlers.emplace(t, std::move(handler));
        if (!r.second) {
            throw_with_backtrace<std::runtime_error>("registered handler already exists");
        }
    }
};
 
 
}
 
}
 
#include "rpc_impl.hh"