diff --git a/src/io/rsm/raft.hpp b/src/io/rsm/raft.hpp index 09f3ea655..ffea65930 100644 --- a/src/io/rsm/raft.hpp +++ b/src/io/rsm/raft.hpp @@ -38,29 +38,32 @@ using memgraph::io::simulator::SimulatorConfig; using memgraph::io::simulator::SimulatorStats; using memgraph::io::simulator::SimulatorTransport; -using Op = std::vector; using Term = uint64_t; using LogIndex = uint64_t; using Time = uint64_t; using Duration = uint64_t; using RequestId = uint64_t; -/// The request that a client sends to request that -/// the cluster replicates their data. -struct ReplicationRequest { - std::vector opaque_data; -}; - -struct ReplicationResponse { +template +struct WriteResponse { bool success; + WriteValue write_value; std::optional
retry_leader; }; +template +struct ReadResponse { + bool success; + ReadValue read_value; + std::optional
retry_leader; +}; + +template struct AppendRequest { Term term = 0; LogIndex last_log_index; Term last_log_term; - std::vector> entries; + std::vector> entries; LogIndex leader_commit; }; @@ -87,9 +90,10 @@ struct VoteResponse { bool vote_granted = false; }; +template struct CommonState { Term term = 0; - std::vector> log; + std::vector> log; LogIndex committed_log_size = 0; LogIndex last_applied = 0; }; @@ -130,17 +134,48 @@ struct Follower { using Role = std::variant; -template Rsm*/> +/* +TODO make concept that expresses the fact that any RSM must +be able to have a specific type applied to it, returning +another interesting result type. + +all ReplicatedState classes should have an apply method +that returns our WriteResponseValue: + +ReadResponse read(ReadRequest); +WriteResponseValue ReplicatedState::apply(WriteRequest); + +for examples: +if the state is uint64_t, and WriteRequest is `struct PlusOne {};`, +and WriteResponseValue is also uint64_t (the new value), then +each call to state.apply(PlusOne{}) will return the new value +after incrementing it. 0, 1, 2, 3... and this will be sent back +to the client that requested the mutation. + +In practice, these mutations will usually be predicated on some +previous value, so that they are idempotent, functioning similarly +to a CAS operation. + +template +concept Rsm = requires(T t, Write w) +{ + { t.read(r) } -> std::same_as; + { t.apply(w) } -> std::same_as; +}; +*/ + +template class Raft { CommonState state_; Role role_ = Candidate{}; Io io_; std::vector
peers_; - // Rsm rsm_; + ReplicatedState replicated_state_; public: - Raft(Io &&io, std::vector
peers /*, Rsm &&rsm */) - : io_(std::move(io)), peers_(peers) /*, rsm_(std::move(rsm)*/ {} + Raft(Io &&io, std::vector
peers, ReplicatedState &&replicated_state) + : io_(std::move(io)), peers_(peers) replicated_state_(std::move(replicated_state)) {} void Run() { Time last_cron = io_.Now(); @@ -155,9 +190,8 @@ class Raft { Duration receive_timeout = RandomTimeout(10000, 50000); - auto request_result = - io_.template ReceiveWithTimeout( - receive_timeout); + auto request_result = io_.template ReceiveWithTimeout(receive_timeout); if (request_result.HasError()) { continue; } @@ -182,7 +216,17 @@ class Raft { } std::ranges::sort(indices, std::ranges::greater()); // assuming reverse sort (using std::ranges::greater) - state_.committed_log_size = indices[(indices.size() / 2)]; + size_t new_committed_log_size = indices[(indices.size() / 2)]; + + // TODO(tyler / gabor) for each index between the old + // index and the new one, apply that log's WriteRequest + // to our replicated_state_, and use the specific return + // value of the ReplicatedState::apply method (WriteResponseValue) + // to respondto the requester. + // + // this will completely replace the while loop below + + state_.committed_log_size = new_committed_log_size; Log("committed_log_size is now ", state_.committed_log_size); @@ -190,7 +234,7 @@ class Raft { const auto &front = leader.pending_client_requests.front(); if (front.log_index <= state_.committed_log_size) { Log("responding SUCCESS to client"); - ReplicationResponse rr{ + WriteResponse rr{ .success = true, .retry_leader = std::nullopt, }; @@ -208,7 +252,7 @@ class Raft { for (auto &[address, follower] : followers) { const LogIndex index = follower.confirmed_contiguous_index; - std::vector> entries; + std::vector> entries; if (state_.log.size() > index) { entries.insert(entries.begin(), state_.log.begin() + index, state_.log.end()); @@ -273,6 +317,29 @@ class Raft { return term; } + template + void Log(Ts &&...args) { + const Time now = io_.Now(); + const Term term = state_.term; + + std::cout << '\t' << now << "\t" << term << "\t" << io_.GetAddress().last_known_port; + + std::visit([&](auto &&role) { role.Print(); }, role_); + + (std::cout << ... << args) << std::endl; + } + + ///////////////////////////////////////////////////////////// + /// Raft-related Cron methods + /// + /// Cron + std::visit is how events are dispatched + /// to certain code based on Raft role. + /// + /// Cron(role) takes as the first argument a reference to its + /// role, and as the second argument, the message that has + /// been received. + ///////////////////////////////////////////////////////////// + /// Periodic protocol maintenance. void Cron() { // dispatch periodic logic based on our role to a specific Cron method. @@ -351,7 +418,9 @@ class Raft { return std::nullopt; } - /// ********************************************** + ///////////////////////////////////////////////////////////// + /// Raft-related Handle methods + /// /// Handle + std::visit is how events are dispatched /// to certain code based on Raft role. /// @@ -359,10 +428,10 @@ class Raft { /// takes as the first argument a reference /// to its role, and as the second argument, the /// message that has been received. - /// ********************************************** - void Handle( - std::variant &&message_variant, - RequestId request_id, Address from_address) { + ///////////////////////////////////////////////////////////// + + void Handle(std::variant &&message_variant, + RequestId request_id, Address from_address) { // dispatch the message to a handler based on our role, // which can be specified in the Handle first argument, // or it can be `auto` if it's a handler for several roles @@ -470,57 +539,6 @@ class Raft { return std::nullopt; } - // only leaders actually handle replication requests from clients - std::optional Handle(Leader &leader, ReplicationRequest &&req, RequestId request_id, Address from_address) { - Log("received ReplicationRequest"); - - // we are the leader. add item to log and send Append to peers - state_.log.emplace_back(std::pair(state_.term, std::move(req.opaque_data))); - - PendingClientRequest pcr{ - .log_index = state_.log.size(), - .request_id = request_id, - .address = from_address, - }; - - leader.pending_client_requests.push_back(pcr); - - BroadcastAppendEntries(leader.followers); - - // TODO(tyler) add message to pending requests buffer, reply asynchronously - return std::nullopt; - } - - // Raft paper - 8 - // When a client first starts up, it connects to a randomly chosen - // server. If the client’s first choice is not the leader, that - // server will reject the client’s request and supply information - // about the most recent leader it has heard from. - std::optional Handle(Follower &follower, ReplicationRequest &&req, RequestId request_id, Address from_address) { - auto res = ReplicationResponse{}; - - res.success = false; - Log("redirecting client to known Leader with port ", follower.leader_address.last_known_port); - res.retry_leader = follower.leader_address; - - io_.Send(from_address, request_id, res); - - return std::nullopt; - } - - std::optional Handle(Candidate &, ReplicationRequest &&req, RequestId request_id, Address from_address) { - Log("received ReplicationRequest - not redirecting because no Leader is known"); - auto res = ReplicationResponse{}; - - res.success = false; - - Cron(); - - io_.Send(from_address, request_id, res); - - return std::nullopt; - } - template std::optional Handle(AllRoles &role, AppendRequest &&req, RequestId request_id, Address from_address) { AppendResponse res{ @@ -635,16 +653,77 @@ class Raft { return std::nullopt; } - template - void Log(Ts &&...args) { - const Time now = io_.Now(); - const Term term = state_.term; + ///////////////////////////////////////////////////////////// + /// RSM-related handle methods + ///////////////////////////////////////////////////////////// - std::cout << '\t' << now << "\t" << term << "\t" << io_.GetAddress().last_known_port; + // Leaders are able to immediately respond to the requester (with a ReadResponseValue) applied to the ReplicatedState + std::optional Handle(Leader &leader, ReadRequest &&res, RequestId request_id, Address from_address) { + // TODO(tyler / gabor) implement + return std::nullopt; + } - std::visit([&](auto &&role) { role.Print(); }, role_); + // Candidates should respond with a failure, similar to the Candidate + WriteRequest failure below + std::optional Handle(Candidate &, ReadRequest &&req, RequestId request_id, Address from_address) { + // TODO(tyler / gabor) implement + return std::nullopt; + } - (std::cout << ... << args) << std::endl; + // Leaders should respond with a redirection, similar to the Follower + WriteRequest response below + std::optional Handle(Follower &, ReadRequest &&req, RequestId request_id, Address from_address) { + // TODO(tyler / gabor) implement + return std::nullopt; + } + + // Raft paper - 8 + // When a client first starts up, it connects to a randomly chosen + // server. If the client’s first choice is not the leader, that + // server will reject the client’s request and supply information + // about the most recent leader it has heard from. + std::optional Handle(Follower &follower, WriteRequest &&req, RequestId request_id, Address from_address) { + auto res = WriteResponse{}; + + res.success = false; + Log("redirecting client to known Leader with port ", follower.leader_address.last_known_port); + res.retry_leader = follower.leader_address; + + io_.Send(from_address, request_id, res); + + return std::nullopt; + } + + std::optional Handle(Candidate &, WriteRequest &&req, RequestId request_id, Address from_address) { + Log("received WriteRequest - not redirecting because no Leader is known"); + auto res = WriteResponse{}; + + res.success = false; + + Cron(); + + io_.Send(from_address, request_id, res); + + return std::nullopt; + } + + // only leaders actually handle replication requests from clients + std::optional Handle(Leader &leader, WriteRequest &&req, RequestId request_id, Address from_address) { + Log("received WriteRequest"); + + // we are the leader. add item to log and send Append to peers + state_.log.emplace_back(std::pair(state_.term, std::move(req.opaque_data))); + + PendingClientRequest pcr{ + .log_index = state_.log.size() - 1, + .request_id = request_id, + .address = from_address, + }; + + leader.pending_client_requests.push_back(pcr); + + BroadcastAppendEntries(leader.followers); + + // TODO(tyler) add message to pending requests buffer, reply asynchronously + return std::nullopt; } }; diff --git a/tests/simulation/raft.cpp b/tests/simulation/raft.cpp index 28e60f78b..2baa3996d 100644 --- a/tests/simulation/raft.cpp +++ b/tests/simulation/raft.cpp @@ -32,8 +32,40 @@ using memgraph::io::simulator::SimulatorConfig; using memgraph::io::simulator::SimulatorStats; using memgraph::io::simulator::SimulatorTransport; +class TestState { + std::map state_; + + public: + ReadResponse read(ReadRequest request) { + // TODO(tyler / gabor) implement + } + + CasResponse apply(CasRequest request) { + // TODO(tyler / gabor) implement + } +}; + +struct Cas { + int key; + int old_value; + int new_value; +}; + +struct CasResponse { + bool success; + std::optional last_value; +}; + +struct ReadRequest { + int key; +}; + +struct ReadResponse { + int value; +}; + template -void RunRaft(Raft server) { +void RunRaft(Raft server) { server.Run(); } @@ -63,6 +95,7 @@ void RunSimulation() { std::vector
srv_2_peers = {srv_addr_1, srv_addr_3}; std::vector
srv_3_peers = {srv_addr_1, srv_addr_2}; + // TODO(tyler / gabor) supply default TestState to Raft constructor Raft srv_1{std::move(srv_io_1), srv_1_peers}; Raft srv_2{std::move(srv_io_2), srv_2_peers}; Raft srv_3{std::move(srv_io_3), srv_3_peers}; @@ -88,6 +121,8 @@ void RunSimulation() { ReplicationRequest cli_req; cli_req.opaque_data = std::vector{1, 2, 3, 4}; + // TODO(tyler / gabor) replace Replication* with Cas/Read + std::cout << "client sending ReplicationRequest to Leader " << leader.last_known_port << std::endl; ResponseFuture response_future = cli_io.RequestWithTimeout(leader, cli_req, 50000);