From e30a55ba2f6687729d2a685e704c754ab42378a9 Mon Sep 17 00:00:00 2001 From: Pavel Solodovnikov Date: Sat, 30 Jan 2021 01:09:32 +0300 Subject: [PATCH 1/3] configure.py: compile serializer.cc This file was not added to the configure.py, which `raft_sys_table_storage` series was supposed to do. Signed-off-by: Pavel Solodovnikov --- configure.py | 1 + 1 file changed, 1 insertion(+) diff --git a/configure.py b/configure.py index 29f8fc34e3..63d1219f16 100755 --- a/configure.py +++ b/configure.py @@ -865,6 +865,7 @@ scylla_core = (['database.cc', 'lua.cc', 'service/raft/schema_raft_state_machine.cc', 'service/raft/raft_sys_table_storage.cc', + 'serializer.cc', ] + [Antlr3Grammar('cql3/Cql.g')] + [Thrift('interface/cassandra.thrift', 'Cassandra')] ) From 1a979dbba23472699d4c6e6b8bae6866297a1b8b Mon Sep 17 00:00:00 2001 From: Pavel Solodovnikov Date: Wed, 9 Dec 2020 01:26:18 +0300 Subject: [PATCH 2/3] raft: add Raft RPC verbs to `messaging_service` and wire up the RPC calls All RPC module APIs except for `send_snapshot` should resolve as soon as the message is sent, so these messages are passed via `send_message_oneway_timeout`. `send_snapshot` message is sent via `send_message_timeout` and returns a `future<>`, which resolves when snapshot transfer finishes or fails with an exception. All necessary functions to wire the new Raft RPC verbs are also provided (such as `register` and `unregister` handlers). Signed-off-by: Pavel Solodovnikov --- message/messaging_service.cc | 58 ++++++++++++++++++++++++++++++++++++ message/messaging_service.hh | 29 +++++++++++++++++- 2 files changed, 86 insertions(+), 1 deletion(-) diff --git a/message/messaging_service.cc b/message/messaging_service.cc index d46036f33f..30b91b3a7b 100644 --- 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 #include #include @@ -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 (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(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 (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 (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 (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 (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& 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) { diff --git a/message/messaging_service.hh b/message/messaging_service.hh index 3e3f28aece..e78c9a9c59 100644 --- 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 #include @@ -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 forward, inet_address reply_to, unsigned shard, response_id_type response_id, std::optional trace_info = std::nullopt); + // RAFT verbs + void register_raft_send_snapshot(std::function (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 (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 (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 (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 (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&& f) const; private: bool remove_rpc_client_one(clients_map& clients, msg_addr id, bool dead_only); From b9a280161dfdd76b9bfceff5857a9d847bd90bfa Mon Sep 17 00:00:00 2001 From: Pavel Solodovnikov Date: Wed, 9 Dec 2020 01:09:01 +0300 Subject: [PATCH 3/3] raft: introduce `raft_rpc` class The patch contains a skeleton implementation for the Scylla-specific Raft RPC module. It uses `netw::messaging_service` as underlying mechanism to send RPC messages. The instance is supposed to be bound to a single raft group. Signed-off-by: Pavel Solodovnikov --- configure.py | 21 ++++++----- service/raft/raft_rpc.cc | 77 ++++++++++++++++++++++++++++++++++++++++ service/raft/raft_rpc.hh | 50 ++++++++++++++++++++++++++ 3 files changed, 139 insertions(+), 9 deletions(-) create mode 100644 service/raft/raft_rpc.cc create mode 100644 service/raft/raft_rpc.hh diff --git a/configure.py b/configure.py index 63d1219f16..ac4d3e42fe 100755 --- a/configure.py +++ b/configure.py @@ -544,6 +544,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', @@ -866,7 +874,9 @@ scylla_core = (['database.cc', 'service/raft/schema_raft_state_machine.cc', 'service/raft/raft_sys_table_storage.cc', 'serializer.cc', - ] + [Antlr3Grammar('cql3/Cql.g')] + [Thrift('interface/cassandra.thrift', 'Cassandra')] + 'service/raft/raft_rpc.cc', + ] + [Antlr3Grammar('cql3/Cql.g')] + [Thrift('interface/cassandra.thrift', 'Cassandra')] \ + + scylla_raft_core ) api = ['api/api.cc', @@ -987,14 +997,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, diff --git a/service/raft/raft_rpc.cc b/service/raft/raft_rpc.cc new file mode 100644 index 0000000000..f50b10bd63 --- /dev/null +++ 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 . + */ +#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()))); +} + +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; +} diff --git a/service/raft/raft_rpc.hh b/service/raft/raft_rpc.hh new file mode 100644 index 0000000000..e75515f630 --- /dev/null +++ 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 . + */ +#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 _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; +};