Merge "raft: RPC module implementation" from Pavel Solodovnikov

This series provides additional RPC verbs and corresponding methods in `messaging_service` class, as well as a scylla-specific Raft RPC module implementation that uses `netw::messaging_service` under the hood to dispatch RPC messages. * https://github.com/ManManson/scylla/commits/raft-api-rpc-impl-v6: raft: introduce `raft_rpc` class raft: add Raft RPC verbs to `messaging_service` and wire up the RPC calls configure.py: compile serializer.cc
2026-06-09 08:23:29 +00:00 · 2021-02-03 10:39:37 +01:00
parent 55e3df8a72 b9a280161d
commit f8ae46f294
5 changed files with 226 additions and 10 deletions
--- a/configure.py
+++ b/configure.py
@@ -547,6 +547,14 @@ extra_cxxflags = {
    'sanitize': {}
 }

+scylla_raft_core = [
+    'raft/raft.cc',
+    'raft/server.cc',
+    'raft/fsm.cc',
+    'raft/progress.cc',
+    'raft/log.cc',
+]
+
 scylla_core = (['database.cc',
                'absl-flat_hash_map.cc',
                'table.cc',
@@ -868,7 +876,10 @@ scylla_core = (['database.cc',
                'lua.cc',
                'service/raft/schema_raft_state_machine.cc',
                'service/raft/raft_sys_table_storage.cc',
-                ] + [Antlr3Grammar('cql3/Cql.g')] + [Thrift('interface/cassandra.thrift', 'Cassandra')]
+                'serializer.cc',
+                'service/raft/raft_rpc.cc',
+                ] + [Antlr3Grammar('cql3/Cql.g')] + [Thrift('interface/cassandra.thrift', 'Cassandra')] \
+                  + scylla_raft_core
               )

 api = ['api/api.cc',
@@ -989,14 +1000,7 @@ scylla_tests_dependencies = scylla_core + idls + scylla_tests_generic_dependenci
    'test/lib/random_schema.cc',
 ]

-scylla_raft_dependencies = [
-    'raft/raft.cc',
-    'raft/server.cc',
-    'raft/fsm.cc',
-    'raft/progress.cc',
-    'raft/log.cc',
-    'utils/uuid.cc'
-]
+scylla_raft_dependencies = scylla_raft_core + ['utils/uuid.cc']

 deps = {
    'scylla': idls + ['main.cc', 'release.cc', 'utils/build_id.cc'] + scylla_core + api + alternator + redis,
--- a/message/messaging_service.cc
+++ b/message/messaging_service.cc
@@ -43,6 +43,7 @@
 #include "repair/repair.hh"
 #include "digest_algorithm.hh"
 #include "service/paxos/proposal.hh"
+#include "serializer.hh"
 #include "idl/consistency_level.dist.hh"
 #include "idl/tracing.dist.hh"
 #include "idl/result.dist.hh"
@@ -64,6 +65,7 @@
 #include "idl/mutation.dist.hh"
 #include "idl/messaging_service.dist.hh"
 #include "idl/paxos.dist.hh"
+#include "idl/raft.dist.hh"
 #include "serializer_impl.hh"
 #include "serialization_visitors.hh"
 #include "idl/consistency_level.dist.impl.hh"
@@ -86,6 +88,7 @@
 #include "idl/mutation.dist.impl.hh"
 #include "idl/messaging_service.dist.impl.hh"
 #include "idl/paxos.dist.impl.hh"
+#include "idl/raft.dist.impl.hh"
 #include <seastar/rpc/lz4_compressor.hh>
 #include <seastar/rpc/lz4_fragmented_compressor.hh>
 #include <seastar/rpc/multi_algo_compressor_factory.hh>
@@ -524,6 +527,11 @@ static constexpr unsigned do_get_rpc_client_idx(messaging_verb verb) {
    case messaging_verb::PAXOS_ACCEPT:
    case messaging_verb::PAXOS_LEARN:
    case messaging_verb::PAXOS_PRUNE:
+    case messaging_verb::RAFT_SEND_SNAPSHOT:
+    case messaging_verb::RAFT_APPEND_ENTRIES:
+    case messaging_verb::RAFT_APPEND_ENTRIES_REPLY:
+    case messaging_verb::RAFT_VOTE_REQUEST:
+    case messaging_verb::RAFT_VOTE_REPLY:
        return 2;
    case messaging_verb::MUTATION_DONE:
    case messaging_verb::MUTATION_FAILED:
@@ -1435,6 +1443,56 @@ future<> messaging_service::send_hint_mutation(msg_addr id, clock_type::time_poi
        std::move(reply_to), shard, std::move(response_id), std::move(trace_info));
 }

+void messaging_service::register_raft_send_snapshot(std::function<future<> (const rpc::client_info&, rpc::opt_time_point, raft::server_id from_id, raft::server_id dst_id, raft::install_snapshot)>&& func) {
+   register_handler(this, netw::messaging_verb::RAFT_SEND_SNAPSHOT, std::move(func));
+}
+future<> messaging_service::unregister_raft_send_snapshot() {
+   return unregister_handler(netw::messaging_verb::RAFT_SEND_SNAPSHOT);
+}
+future<> messaging_service::send_raft_send_snapshot(msg_addr id, clock_type::time_point timeout, raft::server_id from_id, raft::server_id dst_id, const raft::install_snapshot& install_snapshot) {
+   return send_message_timeout<void>(this, messaging_verb::RAFT_SEND_SNAPSHOT, std::move(id), timeout, std::move(from_id), std::move(dst_id), install_snapshot);
+}
+
+void messaging_service::register_raft_append_entries(std::function<future<> (const rpc::client_info&, rpc::opt_time_point, raft::server_id from_id, raft::server_id dst_id, raft::append_request)>&& func) {
+   register_handler(this, netw::messaging_verb::RAFT_APPEND_ENTRIES, std::move(func));
+}
+future<> messaging_service::unregister_raft_append_entries() {
+   return unregister_handler(netw::messaging_verb::RAFT_APPEND_ENTRIES);
+}
+future<> messaging_service::send_raft_append_entries(msg_addr id, clock_type::time_point timeout, raft::server_id from_id, raft::server_id dst_id, const raft::append_request& append_request) {
+   return send_message_oneway_timeout(this, timeout, messaging_verb::RAFT_APPEND_ENTRIES, std::move(id), std::move(from_id), std::move(dst_id), append_request);
+}
+
+void messaging_service::register_raft_append_entries_reply(std::function<future<> (const rpc::client_info&, rpc::opt_time_point, raft::server_id from_id, raft::server_id dst_id, raft::append_reply)>&& func) {
+   register_handler(this, netw::messaging_verb::RAFT_APPEND_ENTRIES_REPLY, std::move(func));
+}
+future<> messaging_service::unregister_raft_append_entries_reply() {
+   return unregister_handler(netw::messaging_verb::RAFT_APPEND_ENTRIES_REPLY);
+}
+future<> messaging_service::send_raft_append_entries_reply(msg_addr id, clock_type::time_point timeout, raft::server_id from_id, raft::server_id dst_id, const raft::append_reply& reply) {
+   return send_message_oneway_timeout(this, timeout, messaging_verb::RAFT_APPEND_ENTRIES_REPLY, std::move(id), std::move(from_id), std::move(dst_id), reply);
+}
+
+void messaging_service::register_raft_vote_request(std::function<future<> (const rpc::client_info&, rpc::opt_time_point, raft::server_id from_id, raft::server_id dst_id, raft::vote_request)>&& func) {
+   register_handler(this, netw::messaging_verb::RAFT_VOTE_REQUEST, std::move(func));
+}
+future<> messaging_service::unregister_raft_vote_request() {
+   return unregister_handler(netw::messaging_verb::RAFT_VOTE_REQUEST);
+}
+future<> messaging_service::send_raft_vote_request(msg_addr id, clock_type::time_point timeout, raft::server_id from_id, raft::server_id dst_id, const raft::vote_request& vote_request) {
+   return send_message_oneway_timeout(this, timeout, messaging_verb::RAFT_VOTE_REQUEST, std::move(id), std::move(from_id), std::move(dst_id), vote_request);
+}
+
+void messaging_service::register_raft_vote_reply(std::function<future<> (const rpc::client_info&, rpc::opt_time_point, raft::server_id from_id, raft::server_id dst_id, raft::vote_reply)>&& func) {
+   register_handler(this, netw::messaging_verb::RAFT_VOTE_REPLY, std::move(func));
+}
+future<> messaging_service::unregister_raft_vote_reply() {
+   return unregister_handler(netw::messaging_verb::RAFT_VOTE_REPLY);
+}
+future<> messaging_service::send_raft_vote_reply(msg_addr id, clock_type::time_point timeout, raft::server_id from_id, raft::server_id dst_id, const raft::vote_reply& vote_reply) {
+   return send_message_oneway_timeout(this, timeout, messaging_verb::RAFT_VOTE_REPLY, std::move(id), std::move(from_id), std::move(dst_id), vote_reply);
+}
+
 void init_messaging_service(sharded<messaging_service>& ms,
                messaging_service::config mscfg, netw::messaging_service::scheduling_config scfg,
                sstring ms_trust_store, sstring ms_cert, sstring ms_key, sstring ms_tls_prio, bool ms_client_auth) {
--- a/message/messaging_service.hh
+++ b/message/messaging_service.hh
@@ -39,6 +39,7 @@
 #include "streaming/stream_mutation_fragments_cmd.hh"
 #include "cache_temperature.hh"
 #include "service/paxos/prepare_response.hh"
+#include "raft/raft.hh"

 #include <list>
 #include <vector>
@@ -144,7 +145,12 @@ enum class messaging_verb : int32_t {
    PAXOS_PRUNE = 43,
    GOSSIP_GET_ENDPOINT_STATES = 44,
    NODE_OPS_CMD = 45,
-    LAST = 46,
+    RAFT_SEND_SNAPSHOT = 46,
+    RAFT_APPEND_ENTRIES = 47,
+    RAFT_APPEND_ENTRIES_REPLY = 48,
+    RAFT_VOTE_REQUEST = 49,
+    RAFT_VOTE_REPLY = 50,
+    LAST = 51,
 };

 } // namespace netw
@@ -547,6 +553,27 @@ public:
    future<> send_hint_mutation(msg_addr id, clock_type::time_point timeout, const frozen_mutation& fm, std::vector<inet_address> forward,
        inet_address reply_to, unsigned shard, response_id_type response_id, std::optional<tracing::trace_info> trace_info = std::nullopt);

+    // RAFT verbs
+    void register_raft_send_snapshot(std::function<future<> (const rpc::client_info&, rpc::opt_time_point, raft::server_id from_id, raft::server_id dst_id, raft::install_snapshot)>&& func);
+    future<> unregister_raft_send_snapshot();
+    future<> send_raft_send_snapshot(msg_addr id, clock_type::time_point timeout, raft::server_id from_id, raft::server_id dst_id, const raft::install_snapshot& install_snapshot);
+
+    void register_raft_append_entries(std::function<future<> (const rpc::client_info&, rpc::opt_time_point, raft::server_id from_id, raft::server_id dst_id, raft::append_request)>&& func);
+    future<> unregister_raft_append_entries();
+    future<> send_raft_append_entries(msg_addr id, clock_type::time_point timeout, raft::server_id from_id, raft::server_id dst_id, const raft::append_request& append_request);
+
+    void register_raft_append_entries_reply(std::function<future<> (const rpc::client_info&, rpc::opt_time_point, raft::server_id from_id, raft::server_id dst_id, raft::append_reply)>&& func);
+    future<> unregister_raft_append_entries_reply();
+    future<> send_raft_append_entries_reply(msg_addr id, clock_type::time_point timeout, raft::server_id from_id, raft::server_id dst_id, const raft::append_reply& reply); 
+
+    void register_raft_vote_request(std::function<future<> (const rpc::client_info&, rpc::opt_time_point, raft::server_id from_id, raft::server_id dst_id, raft::vote_request)>&& func);
+    future<> unregister_raft_vote_request();
+    future<> send_raft_vote_request(msg_addr id, clock_type::time_point timeout, raft::server_id from_id, raft::server_id dst_id, const raft::vote_request& vote_request);
+
+    void register_raft_vote_reply(std::function<future<> (const rpc::client_info&, rpc::opt_time_point, raft::server_id from_id, raft::server_id dst_id, raft::vote_reply)>&& func);
+    future<> unregister_raft_vote_reply();
+    future<> send_raft_vote_reply(msg_addr id, clock_type::time_point timeout, raft::server_id from_id, raft::server_id dst_id, const raft::vote_reply& vote_reply);
+
    void foreach_server_connection_stats(std::function<void(const rpc::client_info&, const rpc::stats&)>&& f) const;
 private:
    bool remove_rpc_client_one(clients_map& clients, msg_addr id, bool dead_only);
--- a/service/raft/raft_rpc.cc
+++ b/service/raft/raft_rpc.cc
@@ -0,0 +1,77 @@
+/*
+ * Copyright (C) 2020 ScyllaDB
+ */
+
+/*
+ * This file is part of Scylla.
+ *
+ * Scylla 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.
+ *
+ * Scylla 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
+ */
+#include "service/raft/raft_rpc.hh"
+#include "gms/inet_address_serializer.hh"
+#include "serializer_impl.hh"
+#include "message/msg_addr.hh"
+#include "message/messaging_service.hh"
+#include "db/timeout_clock.hh"
+
+raft_rpc::raft_rpc(netw::messaging_service& ms, uint64_t group_id, raft::server_id srv_id)
+    : _group_id(group_id), _server_id(srv_id), _messaging(ms)
+{}
+
+future<> raft_rpc::send_snapshot(raft::server_id id, const raft::install_snapshot& snap) {
+    return _messaging.send_raft_send_snapshot(
+        netw::msg_addr(get_inet_address(id)), db::no_timeout, _server_id, id, snap);
+}
+
+future<> raft_rpc::send_append_entries(raft::server_id id, const raft::append_request& append_request) {
+    return _messaging.send_raft_append_entries(
+        netw::msg_addr(get_inet_address(id)), db::no_timeout, _server_id, id, append_request);
+}
+
+future<> raft_rpc::send_append_entries_reply(raft::server_id id, const raft::append_reply& reply) {
+    return _messaging.send_raft_append_entries_reply(
+        netw::msg_addr(get_inet_address(id)), db::no_timeout, _server_id, id, reply);
+}
+
+future<> raft_rpc::send_vote_request(raft::server_id id, const raft::vote_request& vote_request) {
+    return _messaging.send_raft_vote_request(
+        netw::msg_addr(get_inet_address(id)), db::no_timeout, _server_id, id, vote_request);
+}
+
+future<> raft_rpc::send_vote_reply(raft::server_id id, const raft::vote_reply& vote_reply) {
+    return _messaging.send_raft_vote_reply(
+        netw::msg_addr(get_inet_address(id)), db::no_timeout, _server_id, id, vote_reply);
+}
+
+void raft_rpc::add_server(raft::server_id id, raft::server_info info) {
+    // Parse gms::inet_address from server_info
+    auto in = ser::as_input_stream(bytes_view(info));
+    _server_addresses.emplace(std::pair(id, ser::deserialize(in, boost::type<gms::inet_address>())));
+}
+
+void raft_rpc::remove_server(raft::server_id id) {
+    _server_addresses.erase(id);
+}
+
+future<> raft_rpc::abort() {
+    return make_ready_future<>();
+}
+
+gms::inet_address raft_rpc::get_inet_address(raft::server_id id) const {
+    auto it = _server_addresses.find(id);
+    if (it == _server_addresses.end()) {
+        throw std::runtime_error(format("Destination raft server not found (group id: {}, server id: {})", _group_id, id));
+    }
+    return it->second;
+}
--- a/service/raft/raft_rpc.hh
+++ b/service/raft/raft_rpc.hh
@@ -0,0 +1,50 @@
+/*
+ * Copyright (C) 2020 ScyllaDB
+ */
+
+/*
+ * This file is part of Scylla.
+ *
+ * Scylla 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.
+ *
+ * Scylla 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
+ */
+#pragma once
+
+#include "raft/raft.hh"
+#include "gms/inet_address.hh"
+#include "message/messaging_service_fwd.hh"
+
+// Scylla-specific implementation of raft RPC module.
+//
+// Uses `netw::messaging_service` as an underlying implementation for
+// actually sending RPC messages.
+class raft_rpc : public raft::rpc {
+    uint64_t _group_id;
+    raft::server_id _server_id;
+    std::unordered_map<raft::server_id, gms::inet_address> _server_addresses;
+    netw::messaging_service& _messaging;
+
+public:
+    explicit raft_rpc(netw::messaging_service& ms, uint64_t group_id, raft::server_id srv_id);
+
+    future<> send_snapshot(raft::server_id server_id, const raft::install_snapshot& snap) override;
+    future<> send_append_entries(raft::server_id id, const raft::append_request& append_request) override;
+    future<> send_append_entries_reply(raft::server_id id, const raft::append_reply& reply) override;
+    future<> send_vote_request(raft::server_id id, const raft::vote_request& vote_request) override;
+    future<> send_vote_reply(raft::server_id id, const raft::vote_reply& vote_reply) override;
+    void add_server(raft::server_id id, raft::server_info info) override;
+    void remove_server(raft::server_id id) override;
+    future<> abort() override;
+    // Map raft server_id to inet_address to be consumed by `messaging_service`
+    gms::inet_address get_inet_address(raft::server_id id) const;
+};