diff --git a/src/coordinator/coordinator.hpp b/src/coordinator/coordinator.hpp index fcd073d15..6a8e38afc 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; @@ -116,14 +128,14 @@ 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(); 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_) @@ -135,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; @@ -199,27 +211,28 @@ 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)} {} 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) { diff --git a/src/coordinator/shard_map.hpp b/src/coordinator/shard_map.hpp index bc891db5e..9109dba65 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 } @@ -98,14 +108,28 @@ 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: + // 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/utils/rsm_client.hpp b/src/io/rsm/rsm_client.hpp similarity index 95% rename from tests/simulation/utils/rsm_client.hpp rename to src/io/rsm/rsm_client.hpp index 5b6031f2a..3b09a98d8 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; @@ -73,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(); @@ -105,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); @@ -131,3 +133,5 @@ class RsmClient { return TimedOut{}; } }; + +} // namespace memgraph::io::rsm 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_; 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 8fa7a4b05..4f8d630a2 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,8 +48,9 @@ using memgraph::io::rsm::CoordinatorRsm; using memgraph::io::rsm::Raft; using memgraph::io::rsm::ReadRequest; using memgraph::io::rsm::ReadResponse; -using memgraph::io::rsm::StorageGetRequest; -using memgraph::io::rsm::StorageGetResponse; +using memgraph::io::rsm::RsmClient; +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; @@ -61,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, memgraph::coordinator::Address a_io_3, memgraph::coordinator::Address b_io_1, @@ -120,11 +121,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(); } @@ -303,7 +305,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);