From d1c5aead61ef7f3f52913fb8b0e315c265005e55 Mon Sep 17 00:00:00 2001 From: Tyler Neely Date: Wed, 3 Aug 2022 15:07:08 +0000 Subject: [PATCH] Implement the rest of the KV-on-Raft RSM test --- tests/simulation/raft.cpp | 123 ++++++++++++++++++++++++++++---------- 1 file changed, 93 insertions(+), 30 deletions(-) diff --git a/tests/simulation/raft.cpp b/tests/simulation/raft.cpp index 9248a0522..7cf6144fd 100644 --- a/tests/simulation/raft.cpp +++ b/tests/simulation/raft.cpp @@ -44,7 +44,7 @@ struct CasRequest { }; struct CasResponse { - bool success; + bool cas_success; std::optional last_value; }; @@ -53,7 +53,7 @@ struct GetRequest { }; struct GetResponse { - int value; + std::optional value; }; class TestState { @@ -62,16 +62,17 @@ class TestState { public: GetResponse read(GetRequest request) { GetResponse ret; - ret.value = state_.at(request.key); + if (state_.contains(request.key)) { + ret.value = state_[request.key]; + } return ret; } CasResponse apply(CasRequest request) { CasResponse ret; - // TODO(gabor) remove my annoying-ass comments // Key exist - if (state_.find(request.key) != state_.end()) { + if (state_.contains(request.key)) { auto &val = state_[request.key]; /* @@ -79,7 +80,7 @@ class TestState { */ if (!request.new_value) { ret.last_value = val; - ret.success = true; + ret.cas_success = true; state_.erase(state_.find(request.key)); } @@ -90,12 +91,12 @@ class TestState { // Does old_value match? if (request.old_value == val) { ret.last_value = val; - ret.success = true; + ret.cas_success = true; val = request.new_value.value(); } else { ret.last_value = val; - ret.success = false; + ret.cas_success = false; } } /* @@ -103,9 +104,9 @@ class TestState { */ else { ret.last_value = std::nullopt; - ret.success = true; + ret.cas_success = true; - state_[request.key] = request.new_value.value(); + state_.emplace(request.key, std::move(request.new_value).value()); } return ret; @@ -162,45 +163,99 @@ void RunSimulation() { std::mt19937 cli_rng_{0}; Address server_addrs[]{srv_addr_1, srv_addr_2, srv_addr_3}; - bool success = false; Address leader = server_addrs[0]; - while (true) { + const int key = 0; + std::optional last_known_value; + + bool success = false; + + for (int i = 0; !success; i++) { // send request CasRequest cas_req; - cas_req.key = 0; - cas_req.old_value = std::nullopt; - cas_req.new_value = 12; + cas_req.key = key; + + cas_req.old_value = last_known_value; + + cas_req.new_value = i; WriteRequest cli_req; cli_req.operation = cas_req; - // TODO(tyler / gabor) replace Replication* with Cas/Read - - std::cout << "client sending ReplicationRequest to Leader " << leader.last_known_port << std::endl; - ResponseFuture> response_future = + std::cout << "client sending CasRequest to Leader " << leader.last_known_port << std::endl; + ResponseFuture> cas_response_future = cli_io.RequestWithTimeout, WriteResponse>(leader, cli_req, 50000); - // receive response - ResponseResult> response_result = response_future.Wait(); + // receive cas_response + ResponseResult> cas_response_result = cas_response_future.Wait(); - if (response_result.HasError()) { + if (cas_response_result.HasError()) { std::cout << "client timed out while trying to communicate with leader server " << std::endl; continue; } - ResponseEnvelope> response_envelope = response_result.GetValue(); - WriteResponse response = response_envelope.message; + ResponseEnvelope> cas_response_envelope = cas_response_result.GetValue(); + WriteResponse write_cas_response = cas_response_envelope.message; - if (response.success) { - success = true; - break; + if (write_cas_response.retry_leader) { + MG_ASSERT(!write_cas_response.success, "retry_leader should never be set for successful responses"); + leader = write_cas_response.retry_leader.value(); + std::cout << "client redirected to leader server " << leader.last_known_port << std::endl; + } else if (!write_cas_response.success) { + std::uniform_int_distribution addr_distrib(0, 2); + size_t addr_index = addr_distrib(cli_rng_); + leader = server_addrs[addr_index]; + + std::cout << "client NOT redirected to leader server, trying a random one at index " << addr_index + << " with port " << leader.last_known_port << std::endl; + continue; } - if (response.retry_leader) { - leader = response.retry_leader.value(); - std::cout << "client redirected to leader server " << leader.last_known_port << std::endl; + CasResponse cas_response = write_cas_response.write_return; + + bool cas_succeeded = cas_response.cas_success; + + std::cout << "Client received CasResponse! success: " << cas_succeeded + << " last_known_value: " << (int)*last_known_value << std::endl; + + if (cas_succeeded) { + last_known_value = i; } else { + last_known_value = cas_response.last_value; + continue; + } + + GetRequest get_req; + get_req.key = key; + + ReadRequest read_req; + read_req.operation = get_req; + + std::cout << "client sending GetRequest to Leader " << leader.last_known_port << std::endl; + ResponseFuture> get_response_future = + cli_io.RequestWithTimeout, ReadResponse>(leader, read_req, 50000); + + // receive response + ResponseResult> get_response_result = get_response_future.Wait(); + + if (get_response_result.HasError()) { + std::cout << "client timed out while trying to communicate with leader server " << std::endl; + continue; + } + + ResponseEnvelope> get_response_envelope = get_response_result.GetValue(); + ReadResponse read_get_response = get_response_envelope.message; + + if (!read_get_response.success) { + // sent to a non-leader + continue; + } + + if (read_get_response.retry_leader) { + MG_ASSERT(!read_get_response.success, "retry_leader should never be set for successful responses"); + leader = read_get_response.retry_leader.value(); + std::cout << "client redirected to leader server " << leader.last_known_port << std::endl; + } else if (!read_get_response.success) { std::uniform_int_distribution addr_distrib(0, 2); size_t addr_index = addr_distrib(cli_rng_); leader = server_addrs[addr_index]; @@ -208,6 +263,14 @@ void RunSimulation() { std::cout << "client NOT redirected to leader server, trying a random one at index " << addr_index << " with port " << leader.last_known_port << std::endl; } + + GetResponse get_response = read_get_response.read_return; + + MG_ASSERT(get_response.value == i); + + std::cout << "client successfully cas'd a value and read it back! value: " << i << std::endl; + + success = true; } MG_ASSERT(success);