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; +};