From d50b6c1abbb073d6a4c8062ff9c6d7d006978aca Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Thu, 18 Aug 2022 13:20:56 +0000 Subject: [PATCH 1/5] Move rsm_client to src/io/rsm --- {tests/simulation/utils => src/io/rsm}/rsm_client.hpp | 4 ++++ tests/simulation/raft.cpp | 3 ++- tests/simulation/sharded_map.cpp | 3 ++- 3 files changed, 8 insertions(+), 2 deletions(-) rename {tests/simulation/utils => src/io/rsm}/rsm_client.hpp (98%) diff --git a/tests/simulation/utils/rsm_client.hpp b/src/io/rsm/rsm_client.hpp similarity index 98% rename from tests/simulation/utils/rsm_client.hpp rename to src/io/rsm/rsm_client.hpp index 5b6031f2a..f83456d73 100644 --- a/tests/simulation/utils/rsm_client.hpp +++ b/src/io/rsm/rsm_client.hpp @@ -17,6 +17,8 @@ #include "io/rsm/raft.hpp" #include "utils/result.hpp" +namespace memgraph::io::rsm { + using memgraph::io::Address; using memgraph::io::Duration; using memgraph::io::ResponseEnvelope; @@ -131,3 +133,5 @@ class RsmClient { return TimedOut{}; } }; + +} // namespace memgraph::io::rsm diff --git a/tests/simulation/raft.cpp b/tests/simulation/raft.cpp index 8928f7300..09b21fb96 100644 --- a/tests/simulation/raft.cpp +++ b/tests/simulation/raft.cpp @@ -20,9 +20,9 @@ #include "io/address.hpp" #include "io/rsm/raft.hpp" +#include "io/rsm/rsm_client.hpp" #include "io/simulator/simulator.hpp" #include "io/simulator/simulator_transport.hpp" -#include "utils/rsm_client.hpp" using memgraph::io::Address; using memgraph::io::Duration; @@ -34,6 +34,7 @@ using memgraph::io::Time; using memgraph::io::rsm::Raft; using memgraph::io::rsm::ReadRequest; using memgraph::io::rsm::ReadResponse; +using memgraph::io::rsm::RsmClient; using memgraph::io::rsm::WriteRequest; using memgraph::io::rsm::WriteResponse; using memgraph::io::simulator::Simulator; diff --git a/tests/simulation/sharded_map.cpp b/tests/simulation/sharded_map.cpp index c5d48a1a9..677932067 100644 --- a/tests/simulation/sharded_map.cpp +++ b/tests/simulation/sharded_map.cpp @@ -22,11 +22,11 @@ #include "io/errors.hpp" #include "io/rsm/coordinator_rsm.hpp" #include "io/rsm/raft.hpp" +#include "io/rsm/rsm_client.hpp" #include "io/rsm/shard_rsm.hpp" #include "io/simulator/simulator.hpp" #include "io/simulator/simulator_transport.hpp" #include "utils/result.hpp" -#include "utils/rsm_client.hpp" using memgraph::coordinator::Address; using memgraph::coordinator::AddressAndStatus; @@ -48,6 +48,7 @@ using memgraph::io::rsm::CoordinatorRsm; using memgraph::io::rsm::Raft; using memgraph::io::rsm::ReadRequest; using memgraph::io::rsm::ReadResponse; +using memgraph::io::rsm::RsmClient; using memgraph::io::rsm::StorageGetRequest; using memgraph::io::rsm::StorageGetResponse; using memgraph::io::rsm::StorageRsm; From 3794693356dbd7721820b403e5d193dbdc016eaa Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Thu, 18 Aug 2022 13:34:11 +0000 Subject: [PATCH 2/5] Make shard_rsm.hpp a bit more clear --- src/io/rsm/rsm_client.hpp | 4 ++-- src/io/rsm/shard_rsm.hpp | 18 ++++++++++++++---- 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/src/io/rsm/rsm_client.hpp b/src/io/rsm/rsm_client.hpp index f83456d73..3b09a98d8 100644 --- a/src/io/rsm/rsm_client.hpp +++ b/src/io/rsm/rsm_client.hpp @@ -75,7 +75,7 @@ class RsmClient { const Time before = io_.Now(); do { - spdlog::debug("client sending CasRequest to Leader {}", leader_.ToString()); + spdlog::debug("client sending WriteRequest to Leader {}", leader_.ToString()); ResponseFuture> response_future = io_.template Request, WriteResponse>(leader_, client_req); ResponseResult> response_result = std::move(response_future).Wait(); @@ -107,7 +107,7 @@ class RsmClient { const Time before = io_.Now(); do { - spdlog::debug("client sending GetRequest to Leader {}", leader_.ToString()); + spdlog::debug("client sending ReadRequest to Leader {}", leader_.ToString()); ResponseFuture> get_response_future = io_.template Request, ReadResponse>(leader_, read_req); diff --git a/src/io/rsm/shard_rsm.hpp b/src/io/rsm/shard_rsm.hpp index 8cce21664..38ae1e280 100644 --- a/src/io/rsm/shard_rsm.hpp +++ b/src/io/rsm/shard_rsm.hpp @@ -11,6 +11,16 @@ #pragma once +/// The StorageRsm is a simple in-memory raft-backed kv store that can be used for simple testing +/// and implementation of some query engine logic before storage engines are fully implemented. +/// +/// To implement multiple read and write commands, change the StorageRead* and StorageWrite* requests +/// and responses to a std::variant of the different options, and route them to specific handlers in +/// the StorageRsm's Read and Apply methods. Remember that Read is called immediately when the Raft +/// leader receives the request, and does not replicate anything over Raft. Apply is called only +/// AFTER the StorageWriteRequest is replicated to a majority of Raft peers, and the result of calling +/// StorageRsm::Apply(StorageWriteRequest) is returned to the client that submitted the request. + #include #include #include @@ -60,11 +70,11 @@ struct StorageWriteResponse { std::optional latest_known_shard_map_version{std::nullopt}; }; -struct StorageGetRequest { +struct StorageReadRequest { ShardRsmKey key; }; -struct StorageGetResponse { +struct StorageReadResponse { bool shard_rsm_success; std::optional value; // Only has a value if the given shard does not contain the requested key @@ -86,8 +96,8 @@ class StorageRsm { } public: - StorageGetResponse Read(StorageGetRequest request) { - StorageGetResponse ret; + StorageReadResponse Read(StorageReadRequest request) { + StorageReadResponse ret; if (!IsKeyInRange(request.key)) { ret.latest_known_shard_map_version = shard_map_version_; From 2320f95dd1456c678da7de134097557a821e5940 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Thu, 18 Aug 2022 14:01:47 +0000 Subject: [PATCH 3/5] Update the coordinator to include request for initializing a new shard map --- src/coordinator/coordinator.hpp | 39 ++++++++++++++++++++++++++++---- src/coordinator/shard_map.hpp | 33 ++++++++++++++++++++------- tests/simulation/sharded_map.cpp | 15 ++++++------ 3 files changed, 67 insertions(+), 20 deletions(-) diff --git a/src/coordinator/coordinator.hpp b/src/coordinator/coordinator.hpp index fcd073d15..8eb070c72 100644 --- a/src/coordinator/coordinator.hpp +++ b/src/coordinator/coordinator.hpp @@ -87,10 +87,22 @@ struct DeregisterStorageEngineResponse { bool success; }; -using WriteRequests = std::variant; -using WriteResponses = std::variant; +struct InitializeLabelRequest { + std::string label_name; + Hlc last_shard_map_version; +}; + +struct InitializeLabelResponse { + bool success; + std::optional fresher_shard_map; +}; + +using WriteRequests = + std::variant; +using WriteResponses = + std::variant; using ReadRequests = std::variant; using ReadResponses = std::variant; @@ -123,7 +135,7 @@ class Coordinator { MG_ASSERT(!(hlc_request.last_shard_map_version.logical_id > hlc_shard_map.logical_id)); - res.new_hlc = shard_map_.UpdateShardMapVersion(); + res.new_hlc = shard_map_.IncrementShardMapVersion(); // res.fresher_shard_map = hlc_request.last_shard_map_version.logical_id < hlc_shard_map.logical_id // ? std::make_optional(shard_map_) @@ -199,6 +211,23 @@ class Coordinator { return res; } + WriteResponses ApplyWrite(InitializeLabelRequest &&initialize_label_request) { + InitializeLabelResponse res{}; + + bool success = shard_map_.InitializeNewLabel(initialize_label_request.label_name, + initialize_label_request.last_shard_map_version); + + if (success) { + res.fresher_shard_map = shard_map_; + res.success = false; + } else { + res.fresher_shard_map = std::nullopt; + res.success = true; + } + + return res; + } + public: explicit Coordinator(ShardMap sm) : shard_map_{(sm)} {} diff --git a/src/coordinator/shard_map.hpp b/src/coordinator/shard_map.hpp index bc891db5e..7624ff904 100644 --- a/src/coordinator/shard_map.hpp +++ b/src/coordinator/shard_map.hpp @@ -46,16 +46,9 @@ struct ShardMap { Hlc shard_map_version; std::map shards; - // TODO(gabor) later we will want to update the wallclock time with - // the given Io's time as well. This function should just be - // replaced with operator== since it is already overloaded for Hlc - // objects. - bool CompareShardMapVersions(Hlc one, Hlc two) { return one.logical_id == two.logical_id; } - - public: // TODO(gabor) later we will want to update the wallclock time with // the given Io's time as well - Hlc UpdateShardMapVersion() noexcept { + Hlc IncrementShardMapVersion() noexcept { ++shard_map_version.logical_id; return shard_map_version; } @@ -83,12 +76,29 @@ struct ShardMap { // Apply the split shards_in_map[key] = shard_to_map_to; + return true; } return false; } + bool InitializeNewLabel(std::string label_name, Hlc last_shard_map_version) { + if (shard_map_version != last_shard_map_version) { + return false; + } + + if (shards.contains(label_name)) { + return false; + } + + shards.emplace(label_name, Shards{}); + + IncrementShardMapVersion(); + + return true; + } + void AddServer(Address server_address) { // Find a random place for the server to plug in } @@ -106,6 +116,13 @@ struct ShardMap { return asd2; } + + private: + // TODO(gabor) later we will want to update the wallclock time with + // the given Io's time as well. This function should just be + // replaced with operator== since it is already overloaded for Hlc + // objects. + bool CompareShardMapVersions(Hlc one, Hlc two) { return one.logical_id == two.logical_id; } }; } // namespace memgraph::coordinator diff --git a/tests/simulation/sharded_map.cpp b/tests/simulation/sharded_map.cpp index 677932067..6a1f82220 100644 --- a/tests/simulation/sharded_map.cpp +++ b/tests/simulation/sharded_map.cpp @@ -49,8 +49,8 @@ using memgraph::io::rsm::Raft; using memgraph::io::rsm::ReadRequest; using memgraph::io::rsm::ReadResponse; using memgraph::io::rsm::RsmClient; -using memgraph::io::rsm::StorageGetRequest; -using memgraph::io::rsm::StorageGetResponse; +using memgraph::io::rsm::StorageReadRequest; +using memgraph::io::rsm::StorageReadResponse; using memgraph::io::rsm::StorageRsm; using memgraph::io::rsm::StorageWriteRequest; using memgraph::io::rsm::StorageWriteResponse; @@ -62,8 +62,8 @@ using memgraph::io::simulator::SimulatorStats; using memgraph::io::simulator::SimulatorTransport; using memgraph::utils::BasicResult; -using StorageClient = - RsmClient, StorageWriteRequest, StorageWriteResponse, StorageGetRequest, StorageGetResponse>; +using StorageClient = RsmClient, StorageWriteRequest, StorageWriteResponse, StorageReadRequest, + StorageReadResponse>; namespace { ShardMap CreateDummyShardmap(memgraph::coordinator::Address a_io_1, memgraph::coordinator::Address a_io_2, @@ -122,11 +122,12 @@ std::optional DetermineShardLocation(Shard target_shard, const st using ConcreteCoordinatorRsm = CoordinatorRsm; using ConcreteStorageRsm = Raft; + StorageReadRequest, StorageReadResponse>; template void RunStorageRaft( - Raft server) { + Raft + server) { server.Run(); } @@ -307,7 +308,7 @@ int main() { // Have client use shard map to decide which shard to communicate // with to read that same value back - StorageGetRequest storage_get_req; + StorageReadRequest storage_get_req; storage_get_req.key = {write_key_1, write_key_2}; auto get_response_result = storage_client.SendReadRequest(storage_get_req); From 2a3a3338f0c2c1331969925eef5ef2093fa9c9dd Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Thu, 18 Aug 2022 14:04:28 +0000 Subject: [PATCH 4/5] Clean up coordinator a bit --- src/coordinator/coordinator.hpp | 22 +++------------------- 1 file changed, 3 insertions(+), 19 deletions(-) diff --git a/src/coordinator/coordinator.hpp b/src/coordinator/coordinator.hpp index 8eb070c72..6a8e38afc 100644 --- a/src/coordinator/coordinator.hpp +++ b/src/coordinator/coordinator.hpp @@ -128,7 +128,7 @@ class Coordinator { uint64_t highest_allocated_edge_id_; /// Increment our - ReadResponses Read(HlcRequest hlc_request) { + ReadResponses HandleRead(HlcRequest &&hlc_request) { HlcResponse res{}; auto hlc_shard_map = shard_map_.GetHlc(); @@ -147,7 +147,7 @@ class Coordinator { return res; } - GetShardMapResponse Read(GetShardMapRequest &&get_shard_map_request) { + ReadResponses HandleRead(GetShardMapRequest &&get_shard_map_request) { GetShardMapResponse res; res.shard_map = shard_map_; return res; @@ -232,23 +232,7 @@ class Coordinator { explicit Coordinator(ShardMap sm) : shard_map_{(sm)} {} ReadResponses Read(ReadRequests requests) { - // if (std::get_if(&requests)) { - // std::cout << "HlcRequest" << std::endl; - // } else if (std::get_if(&requests)) { - // std::cout << "GetShardMapRequest" << std::endl; - // } else { - // std::cout << "idk requests" << std::endl; - // } - // std::cout << "Coordinator Read()" << std::endl; - auto ret = std::visit([&](auto requests) { return Read(requests); }, (requests)); - // if (std::get_if(&ret)) { - // std::cout << "HlcResponse" << std::endl; - // } else if (std::get_if(&ret)) { - // std::cout << "GetShardMapResponse" << std::endl; - // } else { - // std::cout << "idk response" << std::endl; - // } - return ret; + return std::visit([&](auto &&requests) { return HandleRead(std::move(requests)); }, std::move(requests)); } WriteResponses Apply(WriteRequests requests) { From d42fa1fc8702367eeaeec78c4b805691ca75b0ce Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Thu, 18 Aug 2022 15:14:04 +0000 Subject: [PATCH 5/5] Fix ShardMap::GetShardForKey --- src/coordinator/shard_map.hpp | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/src/coordinator/shard_map.hpp b/src/coordinator/shard_map.hpp index 7624ff904..9109dba65 100644 --- a/src/coordinator/shard_map.hpp +++ b/src/coordinator/shard_map.hpp @@ -108,13 +108,20 @@ struct ShardMap { Shards GetShardsForRange(Label label, CompoundKey start, CompoundKey end); Shard GetShardForKey(Label label, CompoundKey key) { - // return shards.at(label).at(key); - std::cout << "label" << std::endl; - auto asd1 = shards.at(label); - std::cout << "key" << std::endl; - auto asd2 = asd1[key]; + auto shard_for_label = shards.at(label); - return asd2; + auto max = (--shard_for_label.end())->first; + + if (key > max) { + return shard_for_label[max]; + } + + for (auto it = shard_for_label.lower_bound(key);; --it) { + MG_ASSERT(it->first <= key); + return it->second; + } + + MG_ASSERT(false, "failed to find shard with a key that is less than or equal to the provided key"); } private: