From 2e552991e49fffaf3cd184a98febbb4514326808 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Thu, 6 Feb 2020 01:16:11 +0000 Subject: [PATCH 1/6] WIP --- include/kvs/server_utils.hpp | 32 +++++++++++++++++++++++++++----- 1 file changed, 27 insertions(+), 5 deletions(-) diff --git a/include/kvs/server_utils.hpp b/include/kvs/server_utils.hpp index 40e82b2b..ba42484e 100644 --- a/include/kvs/server_utils.hpp +++ b/include/kvs/server_utils.hpp @@ -60,14 +60,24 @@ class MemoryLWWSerializer : public Serializer { public: MemoryLWWSerializer(MemoryLWWKVS *kvs) : kvs_(kvs) {} - string get(const Key &key, AnnaError &error) { + string get(const Key &key, AnnaError &error, bool delta, const string &previous_payload) { + auto val = kvs_->get(key, error); - if (val.reveal().value == "") { + if (val.size().reveal() == 0) { error = AnnaError::KEY_DNE; + } + + if (!delta) { + return serialize(val); + } else { + if (val.reveal().timestamp != deserialize_lww(previous_payload).reveal().timestamp) { + return serialize(val); + } else { + return "ACK"; + } } - return serialize(val); } unsigned put(const Key &key, const string &serialized) { @@ -85,12 +95,22 @@ class MemorySetSerializer : public Serializer { public: MemorySetSerializer(MemorySetKVS *kvs) : kvs_(kvs) {} - string get(const Key &key, AnnaError &error) { + string get(const Key &key, AnnaError &error, bool delta, const string &previous_payload) { auto val = kvs_->get(key, error); + if (val.size().reveal() == 0) { error = AnnaError::KEY_DNE; } - return serialize(val); + + if (!delta) { + return serialize(val); + } else { + if (val.value.reveal() != previous_payload) { + return serialize(val); + } else { + return "ACK"; + } + } } unsigned put(const Key &key, const string &serialized) { @@ -842,3 +862,5 @@ struct PendingGossip { }; #endif // INCLUDE_KVS_SERVER_UTILS_HPP_ + + From 16f886bee2ac33ef2c808458cfcf1ff49bc52fa4 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Sat, 15 Feb 2020 19:28:06 +0000 Subject: [PATCH 2/6] finish process_get and client.py --- include/kvs/server_utils.hpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/include/kvs/server_utils.hpp b/include/kvs/server_utils.hpp index ba42484e..72466bcf 100644 --- a/include/kvs/server_utils.hpp +++ b/include/kvs/server_utils.hpp @@ -105,7 +105,7 @@ class MemorySetSerializer : public Serializer { if (!delta) { return serialize(val); } else { - if (val.value.reveal() != previous_payload) { + if (val.reveal().value != previous_payload) { return serialize(val); } else { return "ACK"; From 94669cd5cf855cf085e2ecb4b405fd59210fd9ea Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Wed, 19 Feb 2020 19:35:35 +0000 Subject: [PATCH 3/6] rsub breaks down 2/19 --- client/python/anna/base_client.py | 25 +++++++++++ client/python/anna/client.py | 75 +++++++++++++++++++++++++++++++ src/kvs/user_request_handler.cpp | 5 ++- src/kvs/utils.cpp | 4 +- 4 files changed, 106 insertions(+), 3 deletions(-) diff --git a/client/python/anna/base_client.py b/client/python/anna/base_client.py index 82f4072c..096b5294 100644 --- a/client/python/anna/base_client.py +++ b/client/python/anna/base_client.py @@ -233,6 +233,31 @@ def _prepare_data_request(self, keys): return (req, tuples) + # Helper function to create a KeyRequest (see + # hydro-project/common/proto/anna.proto). Takes in a key name and returns a + # tuple containing a KeyRequest and a KeyTuple contained in that KeyRequest + # with response_address, request_id, and address_cache_size automatically + # populated. + def _prepare_delta_data_request(self, keys, serialized_previous): + req = KeyRequest() + req.request_id = self._get_request_id() + req.response_address = self.response_address + + tuples = [] + + for key in keys: + # TODO: is tup included in req? + tup = req.tuples.add() + tuples.append(tup) + tup.key = key + tup.previous_payload = serialized_previous + + if self.address_cache and key in self.address_cache: + tup.address_cache_size = len(self.address_cache[key]) + + return (req, tuples) + + # Returns and increments a request ID. Loops back after 10,000 requests. def _get_request_id(self): response = self.ut.get_ip() + ':' + str(self.rid) diff --git a/client/python/anna/client.py b/client/python/anna/client.py index 90ba6c0b..6b15168b 100644 --- a/client/python/anna/client.py +++ b/client/python/anna/client.py @@ -32,6 +32,16 @@ SocketCache ) +from anna.lattice import ( + LWWPairLattice, + SetLattice +) + +from anna.anna_pb2 import ( + # Anna's lattice types as an enum + LWW, SET) + + class AnnaTcpClient(BaseAnnaClient): def __init__(self, elb_addr, ip, local=False, offset=0): @@ -76,6 +86,8 @@ def __init__(self, elb_addr, ip, local=False, offset=0): self.rid = 0 + self.cache = {} + def get(self, keys): if type(keys) != list: keys = [keys] @@ -115,6 +127,69 @@ def get(self, keys): return kv_pairs + def get_delta(self, keys): + if type(keys) != list: + keys = [keys] + + worker_addresses = {} + for key in keys: + worker_addresses[key] = (self._get_worker_address(key)) + + # Initialize all KV pairs to 0. Only change a value if we get a valid + # response from the server. + kv_pairs = {} + for key in keys: + kv_pairs[key] = None + + request_ids = [] + for key in worker_addresses: + if key in self.cache: + # TODO: if the key is in cache, + # send the previous_payplad along with the request + kv_pairs[key] = self.cache[key] + + if worker_addresses[key]: + send_sock = self.pusher_cache.get(worker_addresses[key]) + + req, _ = self._prepare_delta_data_request([key], self._serialize(self.cache[key])[0]) + req.type = GET + + send_request(req, send_sock) + request_ids.append(req.request_id) + + else: + if worker_addresses[key]: + send_sock = self.pusher_cache.get(worker_addresses[key]) + + req, _ = self._prepare_data_request([key]) + req.type = GET + + send_request(req, send_sock) + request_ids.append(req.request_id) + + # Wait for all responses to return. + responses = recv_response(request_ids, self.response_puller, + KeyResponse) + + for response in responses: + for tup in response.tuples: + if tup.invalidate: + self._invalidate_cache(tup.key) + + lattice_value = self._deserialize(tup) + if tup.error == NO_ERROR: + kv_pairs[tup.key] = lattice_value + cache[tup.key] = lattice_value + + # if tup.lattice_type == LWW: + # cache[tup.key] = LWWPairLattice(tup.timestamp, "") + # if tup.lattice_type == SET: + # cache[tup.key] = SetLattice(tup.payload) + + return kv_pairs + + + def get_all(self, keys): if type(keys) != list: keys = [keys] diff --git a/src/kvs/user_request_handler.cpp b/src/kvs/user_request_handler.cpp index 40fba97a..f0ccda84 100644 --- a/src/kvs/user_request_handler.cpp +++ b/src/kvs/user_request_handler.cpp @@ -39,6 +39,8 @@ void user_request_handler( // first check if the thread is responsible for the key Key key = tuple.key(); string payload = tuple.payload(); + bool delta = tuple.delta(); + bytes previous_payload = tuple.previous_payload(); ServerThreadList threads = kHashRingUtil->get_responsible_threads( wt.replication_response_connect_address(), key, is_metadata(key), @@ -76,10 +78,11 @@ void user_request_handler( tp->set_error(AnnaError::KEY_DNE); } else { - auto res = process_get(key, serializers[stored_key_map[key].type_]); + auto res = process_get(key, serializers[stored_key_map[key].type_], delta, previous_payload); tp->set_lattice_type(stored_key_map[key].type_); tp->set_payload(res.first); tp->set_error(res.second); + // TODO: do I need to set anything else here? } } else if (request_type == RequestType::PUT) { if (tuple.lattice_type() == LatticeType::NONE) { diff --git a/src/kvs/utils.cpp b/src/kvs/utils.cpp index 69c30493..08328a23 100644 --- a/src/kvs/utils.cpp +++ b/src/kvs/utils.cpp @@ -51,9 +51,9 @@ void send_gossip(AddressKeysetMap &addr_keyset_map, SocketCache &pushers, } std::pair process_get(const Key &key, - Serializer *serializer) { + Serializer *serializer, bool delta, const string &previous_payload) { AnnaError error = AnnaError::NO_ERROR; - auto res = serializer->get(key, error); + auto res = serializer->get(key, error, delta, previous_payload); return std::pair(std::move(res), error); } From 448836e25c9e1cf8298f94781fb8b37424a640ca Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Thu, 20 Feb 2020 18:47:24 +0000 Subject: [PATCH 4/6] finish delta logic --- client/python/anna/client.py | 4 ++++ include/kvs/server_utils.hpp | 3 ++- src/kvs/user_request_handler.cpp | 1 - 3 files changed, 6 insertions(+), 2 deletions(-) diff --git a/client/python/anna/client.py b/client/python/anna/client.py index 6b15168b..da0f1f3b 100644 --- a/client/python/anna/client.py +++ b/client/python/anna/client.py @@ -177,6 +177,10 @@ def get_delta(self, keys): self._invalidate_cache(tup.key) lattice_value = self._deserialize(tup) + + if (lattice_value.payload == "ACK"): + lattice_value = self.cache[key] + if tup.error == NO_ERROR: kv_pairs[tup.key] = lattice_value cache[tup.key] = lattice_value diff --git a/include/kvs/server_utils.hpp b/include/kvs/server_utils.hpp index 72466bcf..561c5a4a 100644 --- a/include/kvs/server_utils.hpp +++ b/include/kvs/server_utils.hpp @@ -105,7 +105,8 @@ class MemorySetSerializer : public Serializer { if (!delta) { return serialize(val); } else { - if (val.reveal().value != previous_payload) { + SetLattice deserialized_payload = deserialize_set(previous_payload); + if (val.reveal().value operator!= previous_payload.reveal()) { return serialize(val); } else { return "ACK"; diff --git a/src/kvs/user_request_handler.cpp b/src/kvs/user_request_handler.cpp index f0ccda84..2602f287 100644 --- a/src/kvs/user_request_handler.cpp +++ b/src/kvs/user_request_handler.cpp @@ -82,7 +82,6 @@ void user_request_handler( tp->set_lattice_type(stored_key_map[key].type_); tp->set_payload(res.first); tp->set_error(res.second); - // TODO: do I need to set anything else here? } } else if (request_type == RequestType::PUT) { if (tuple.lattice_type() == LatticeType::NONE) { From faa426eefa526a80bddfb8ae16ebfeb475c77d0b Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Tue, 3 Mar 2020 18:00:22 +0000 Subject: [PATCH 5/6] finish delta --- client/python/anna/base_client.py | 1 + client/python/anna/client.py | 7 +---- common | 2 +- include/kvs/kvs_handlers.hpp | 2 +- include/kvs/server_utils.hpp | 38 ++++++++++++++++--------- src/kvs/user_request_handler.cpp | 14 +++++++-- src/kvs/utils.cpp | 2 ++ tests/kvs/server_handler_base.hpp | 17 +++++++++++ tests/kvs/test_user_request_handler.hpp | 38 +++++++++++++++++++++++++ 9 files changed, 96 insertions(+), 25 deletions(-) diff --git a/client/python/anna/base_client.py b/client/python/anna/base_client.py index 096b5294..8e1e450d 100644 --- a/client/python/anna/base_client.py +++ b/client/python/anna/base_client.py @@ -251,6 +251,7 @@ def _prepare_delta_data_request(self, keys, serialized_previous): tuples.append(tup) tup.key = key tup.previous_payload = serialized_previous + tup.delta = True if self.address_cache and key in self.address_cache: tup.address_cache_size = len(self.address_cache[key]) diff --git a/client/python/anna/client.py b/client/python/anna/client.py index da0f1f3b..ca35cf41 100644 --- a/client/python/anna/client.py +++ b/client/python/anna/client.py @@ -178,18 +178,13 @@ def get_delta(self, keys): lattice_value = self._deserialize(tup) - if (lattice_value.payload == "ACK"): + if (lattice_value.identical): lattice_value = self.cache[key] if tup.error == NO_ERROR: kv_pairs[tup.key] = lattice_value cache[tup.key] = lattice_value - # if tup.lattice_type == LWW: - # cache[tup.key] = LWWPairLattice(tup.timestamp, "") - # if tup.lattice_type == SET: - # cache[tup.key] = SetLattice(tup.payload) - return kv_pairs diff --git a/common b/common index b576ca55..d215a560 160000 --- a/common +++ b/common @@ -1 +1 @@ -Subproject commit b576ca556c5fc2c0a648e6402ae3626316f825f1 +Subproject commit d215a5604ad591a2a119be3e7c898fea13d0c58c diff --git a/include/kvs/kvs_handlers.hpp b/include/kvs/kvs_handlers.hpp index 2403a014..b945ca49 100644 --- a/include/kvs/kvs_handlers.hpp +++ b/include/kvs/kvs_handlers.hpp @@ -92,7 +92,7 @@ void send_gossip(AddressKeysetMap &addr_keyset_map, SocketCache &pushers, map &stored_key_map); std::pair process_get(const Key &key, - Serializer *serializer); + Serializer *serializer, bool delta = false, const string& previous_payload = ""); void process_put(const Key &key, LatticeType lattice_type, const string &payload, Serializer *serializer, diff --git a/include/kvs/server_utils.hpp b/include/kvs/server_utils.hpp index 561c5a4a..e96463cb 100644 --- a/include/kvs/server_utils.hpp +++ b/include/kvs/server_utils.hpp @@ -48,7 +48,7 @@ typedef map> AddressKeysetMap; class Serializer { public: - virtual string get(const Key &key, AnnaError &error) = 0; + virtual string get(const Key &key, AnnaError &error, bool delta = false, const string &previous_payload = "") = 0; virtual unsigned put(const Key &key, const string &serialized) = 0; virtual void remove(const Key &key) = 0; virtual ~Serializer(){}; @@ -60,7 +60,7 @@ class MemoryLWWSerializer : public Serializer { public: MemoryLWWSerializer(MemoryLWWKVS *kvs) : kvs_(kvs) {} - string get(const Key &key, AnnaError &error, bool delta, const string &previous_payload) { + string get(const Key &key, AnnaError &error, bool delta = false, const string &previous_payload = "") { auto val = kvs_->get(key, error); @@ -95,7 +95,7 @@ class MemorySetSerializer : public Serializer { public: MemorySetSerializer(MemorySetKVS *kvs) : kvs_(kvs) {} - string get(const Key &key, AnnaError &error, bool delta, const string &previous_payload) { + string get(const Key &key, AnnaError &error, bool delta = false, const string &previous_payload = "") { auto val = kvs_->get(key, error); if (val.size().reveal() == 0) { @@ -106,7 +106,17 @@ class MemorySetSerializer : public Serializer { return serialize(val); } else { SetLattice deserialized_payload = deserialize_set(previous_payload); - if (val.reveal().value operator!= previous_payload.reveal()) { + + bool same_set = true; + if (deserialized_payload.reveal().size() != val.reveal().size()) { + same_set = false; + } + // if (std::set_difference(val.reveal(), val.reveal() + val.reveal().size(), deserialized_payload.reveal(), deserialized_payload.reveal() + deserialized_payload.reveal.size(), v1.begin()) - v1.begin() != 0) { + // for (std::string::iterator it=str.begin(); it!=str.end(); ++it) { + + // } + // if (!same_set) { + if (val.reveal() == deserialized_payload.reveal()) { return serialize(val); } else { return "ACK"; @@ -129,7 +139,7 @@ class MemoryOrderedSetSerializer : public Serializer { public: MemoryOrderedSetSerializer(MemoryOrderedSetKVS *kvs) : kvs_(kvs) {} - string get(const Key &key, AnnaError &error) { + string get(const Key &key, AnnaError &error, bool delta = false, const string &previous_payload = "") { auto val = kvs_->get(key, error); if (val.size().reveal() == 0) { error = AnnaError::KEY_DNE; @@ -152,7 +162,7 @@ class MemorySingleKeyCausalSerializer : public Serializer { public: MemorySingleKeyCausalSerializer(MemorySingleKeyCausalKVS *kvs) : kvs_(kvs) {} - string get(const Key &key, AnnaError &error) { + string get(const Key &key, AnnaError &error, bool delta = false, const string &previous_payload = "") { auto val = kvs_->get(key, error); if (val.reveal().value.size().reveal() == 0) { error = AnnaError::KEY_DNE; @@ -177,7 +187,7 @@ class MemoryMultiKeyCausalSerializer : public Serializer { public: MemoryMultiKeyCausalSerializer(MemoryMultiKeyCausalKVS *kvs) : kvs_(kvs) {} - string get(const Key &key, AnnaError &error) { + string get(const Key &key, AnnaError &error, bool delta = false, const string &previous_payload = "") { auto val = kvs_->get(key, error); if (val.reveal().value.size().reveal() == 0) { error = AnnaError::KEY_DNE; @@ -203,7 +213,7 @@ class MemoryPrioritySerializer : public Serializer { public: MemoryPrioritySerializer(MemoryPriorityKVS *kvs) : kvs_(kvs) {} - string get(const Key &key, AnnaError &error) { + string get(const Key &key, AnnaError &error, bool delta = false, const string &previous_payload = "") { auto val = kvs_->get(key, error); if (val.reveal().value == "") { error = AnnaError::KEY_DNE; @@ -235,7 +245,7 @@ class DiskLWWSerializer : public Serializer { } } - string get(const Key &key, AnnaError &error) { + string get(const Key &key, AnnaError &error, bool delta = false, const string &previous_payload = "") { string res; LWWValue value; @@ -320,7 +330,7 @@ class DiskSetSerializer : public Serializer { } } - string get(const Key &key, AnnaError &error) { + string get(const Key &key, AnnaError &error, bool delta = false, const string &previous_payload = "") { string res; SetValue value; @@ -416,7 +426,7 @@ class DiskOrderedSetSerializer : public Serializer { } } - string get(const Key &key, AnnaError &error) { + string get(const Key &key, AnnaError &error, bool delta = false, const string &previous_payload = "") { string res; SetValue value; @@ -512,7 +522,7 @@ class DiskSingleKeyCausalSerializer : public Serializer { } } - string get(const Key &key, AnnaError &error) { + string get(const Key &key, AnnaError &error, bool delta = false, const string &previous_payload = "") { string res; SingleKeyCausalValue value; @@ -629,7 +639,7 @@ class DiskMultiKeyCausalSerializer : public Serializer { } } - string get(const Key &key, AnnaError &error) { + string get(const Key &key, AnnaError &error, bool delta = false, const string &previous_payload = "") { string res; MultiKeyCausalValue value; @@ -782,7 +792,7 @@ class DiskPrioritySerializer : public Serializer { ebs_root_ += "/"; } - string get(const Key &key, AnnaError &error) override { + string get(const Key &key, AnnaError &error, bool delta = false, const string &previous_payload = "") override { string res; PriorityValue value; diff --git a/src/kvs/user_request_handler.cpp b/src/kvs/user_request_handler.cpp index 2602f287..70b2bcc7 100644 --- a/src/kvs/user_request_handler.cpp +++ b/src/kvs/user_request_handler.cpp @@ -40,7 +40,7 @@ void user_request_handler( Key key = tuple.key(); string payload = tuple.payload(); bool delta = tuple.delta(); - bytes previous_payload = tuple.previous_payload(); + const string previous_payload = tuple.previous_payload(); ServerThreadList threads = kHashRingUtil->get_responsible_threads( wt.replication_response_connect_address(), key, is_metadata(key), @@ -76,12 +76,20 @@ void user_request_handler( if (stored_key_map.find(key) == stored_key_map.end() || stored_key_map[key].type_ == LatticeType::NONE) { - tp->set_error(AnnaError::KEY_DNE); + tp->set_error(AnnaError::KEY_DNE); } else { + auto res = process_get(key, serializers[stored_key_map[key].type_], delta, previous_payload); tp->set_lattice_type(stored_key_map[key].type_); - tp->set_payload(res.first); + if (res.first == "ACK") { + tp->set_identical(true); + std::cout << "receive ACK in u_r_h" << std::endl; + + } else { + tp->set_payload(res.first); + } tp->set_error(res.second); + } } else if (request_type == RequestType::PUT) { if (tuple.lattice_type() == LatticeType::NONE) { diff --git a/src/kvs/utils.cpp b/src/kvs/utils.cpp index 08328a23..99bd2f93 100644 --- a/src/kvs/utils.cpp +++ b/src/kvs/utils.cpp @@ -34,6 +34,8 @@ void send_gossip(AddressKeysetMap &addr_keyset_map, SocketCache &pushers, type = stored_key_map[key].type_; } + + auto res = process_get(key, serializers[type]); if (res.second == 0) { diff --git a/tests/kvs/server_handler_base.hpp b/tests/kvs/server_handler_base.hpp index e33caf19..014f145f 100644 --- a/tests/kvs/server_handler_base.hpp +++ b/tests/kvs/server_handler_base.hpp @@ -124,6 +124,23 @@ class ServerHandlerTest : public ::testing::Test { return request_str; } + string get_key_request_delta(Key key, string ip, string previous_payload) { + KeyRequest request; + request.set_type(RequestType::GET); + request.set_response_address(UserThread(ip, 0).response_connect_address()); + request.set_request_id(kRequestId); + + KeyTuple *tp = request.add_tuples(); + tp->set_key(std::move(key)); + tp->set_delta(true); + tp->set_previous_payload(previous_payload); + + string request_str; + request.SerializeToString(&request_str); + + return request_str; + } + string put_key_request(Key key, LatticeType lattice_type, string payload, string ip) { KeyRequest request; diff --git a/tests/kvs/test_user_request_handler.hpp b/tests/kvs/test_user_request_handler.hpp index 3ca27f6e..5e2029e1 100644 --- a/tests/kvs/test_user_request_handler.hpp +++ b/tests/kvs/test_user_request_handler.hpp @@ -52,6 +52,44 @@ TEST_F(ServerHandlerTest, UserGetLWWTest) { EXPECT_EQ(key_access_tracker[key].size(), 1); } +TEST_F(ServerHandlerTest, UserGetDeltaLWWTest) { + Key key = "key"; + string value = "value"; + serializers[LatticeType::LWW]->put(key, serialize(0, value)); + stored_key_map[key].type_ = LatticeType::LWW; + + string get_request = get_key_request_delta(key, ip, value); + + unsigned access_count = 0; + unsigned seed = 0; + + EXPECT_EQ(local_changeset.size(), 0); + + user_request_handler(access_count, seed, get_request, log_, global_hash_rings, + local_hash_rings, pending_requests, key_access_tracker, + stored_key_map, key_replication_map, local_changeset, wt, + serializers, pushers); + + vector messages = get_zmq_messages(); + EXPECT_EQ(messages.size(), 1); + + KeyResponse response; + response.ParseFromString(messages[0]); + + EXPECT_EQ(response.response_id(), kRequestId); + EXPECT_EQ(response.tuples().size(), 1); + + KeyTuple rtp = response.tuples(0); + + EXPECT_EQ(rtp.key(), key); + EXPECT_EQ(rtp.identical(), true); + EXPECT_EQ(rtp.error(), 0); + + EXPECT_EQ(local_changeset.size(), 0); + EXPECT_EQ(access_count, 1); + EXPECT_EQ(key_access_tracker[key].size(), 1); +} + TEST_F(ServerHandlerTest, UserGetSetTest) { Key key = "key"; set s; From 1a36e442fca97f73702636e5654c9fbdc00ce76b Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Thu, 5 Mar 2020 23:38:59 +0000 Subject: [PATCH 6/6] finish delta and test --- include/kvs/server_utils.hpp | 16 +++------- include/metadata.hpp | 1 + src/kvs/user_request_handler.cpp | 4 +-- tests/kvs/test_user_request_handler.hpp | 42 +++++++++++++++++++++++++ 4 files changed, 49 insertions(+), 14 deletions(-) diff --git a/include/kvs/server_utils.hpp b/include/kvs/server_utils.hpp index e96463cb..1e4c0765 100644 --- a/include/kvs/server_utils.hpp +++ b/include/kvs/server_utils.hpp @@ -24,6 +24,7 @@ #include "kvs_common.hpp" #include "lattices/lww_pair_lattice.hpp" #include "yaml-cpp/yaml.h" +#include "metadata.hpp" // Define the garbage collect threshold #define GARBAGE_COLLECT_THRESHOLD 10000000 @@ -74,7 +75,7 @@ class MemoryLWWSerializer : public Serializer { if (val.reveal().timestamp != deserialize_lww(previous_payload).reveal().timestamp) { return serialize(val); } else { - return "ACK"; + return kDeltaRequestIdentical; } } @@ -107,19 +108,10 @@ class MemorySetSerializer : public Serializer { } else { SetLattice deserialized_payload = deserialize_set(previous_payload); - bool same_set = true; - if (deserialized_payload.reveal().size() != val.reveal().size()) { - same_set = false; - } - // if (std::set_difference(val.reveal(), val.reveal() + val.reveal().size(), deserialized_payload.reveal(), deserialized_payload.reveal() + deserialized_payload.reveal.size(), v1.begin()) - v1.begin() != 0) { - // for (std::string::iterator it=str.begin(); it!=str.end(); ++it) { - - // } - // if (!same_set) { - if (val.reveal() == deserialized_payload.reveal()) { + if (val.reveal() != deserialized_payload.reveal()) { return serialize(val); } else { - return "ACK"; + return kDeltaRequestIdentical; } } } diff --git a/include/metadata.hpp b/include/metadata.hpp index a8c47392..304d5a5a 100644 --- a/include/metadata.hpp +++ b/include/metadata.hpp @@ -19,6 +19,7 @@ #include "threads.hpp" const string kMetadataTypeReplication = "replication"; +const string kDeltaRequestIdentical = "identical"; struct TierEnumHash { template std::size_t operator()(T t) const { diff --git a/src/kvs/user_request_handler.cpp b/src/kvs/user_request_handler.cpp index 70b2bcc7..afd7e2ef 100644 --- a/src/kvs/user_request_handler.cpp +++ b/src/kvs/user_request_handler.cpp @@ -13,6 +13,7 @@ // limitations under the License. #include "kvs/kvs_handlers.hpp" +#include "metadata.hpp" void user_request_handler( unsigned &access_count, unsigned &seed, string &serialized, logger log, @@ -81,9 +82,8 @@ void user_request_handler( auto res = process_get(key, serializers[stored_key_map[key].type_], delta, previous_payload); tp->set_lattice_type(stored_key_map[key].type_); - if (res.first == "ACK") { + if (res.first == kDeltaRequestIdentical) { tp->set_identical(true); - std::cout << "receive ACK in u_r_h" << std::endl; } else { tp->set_payload(res.first); diff --git a/tests/kvs/test_user_request_handler.hpp b/tests/kvs/test_user_request_handler.hpp index 5e2029e1..114e952d 100644 --- a/tests/kvs/test_user_request_handler.hpp +++ b/tests/kvs/test_user_request_handler.hpp @@ -131,6 +131,48 @@ TEST_F(ServerHandlerTest, UserGetSetTest) { EXPECT_EQ(key_access_tracker[key].size(), 1); } +TEST_F(ServerHandlerTest, UserGetDeltaSetTest) { + Key key = "key"; + set s; + s.emplace("value1"); + s.emplace("value2"); + s.emplace("value3"); + serializers[LatticeType::SET]->put(key, serialize(SetLattice(s))); + stored_key_map[key].type_ = LatticeType::SET; + string serialized_s = serialize(SetLattice(s)); + + string get_request = get_key_request_delta(key, ip, serialized_s); + + unsigned access_count = 0; + unsigned seed = 0; + + EXPECT_EQ(local_changeset.size(), 0); + + user_request_handler(access_count, seed, get_request, log_, global_hash_rings, + local_hash_rings, pending_requests, key_access_tracker, + stored_key_map, key_replication_map, local_changeset, wt, + serializers, pushers); + + vector messages = get_zmq_messages(); + EXPECT_EQ(messages.size(), 1); + + KeyResponse response; + response.ParseFromString(messages[0]); + + EXPECT_EQ(response.response_id(), kRequestId); + EXPECT_EQ(response.tuples().size(), 1); + + KeyTuple rtp = response.tuples(0); + + EXPECT_EQ(rtp.key(), key); + EXPECT_EQ(rtp.identical(), true); + EXPECT_EQ(rtp.error(), 0); + + EXPECT_EQ(local_changeset.size(), 0); + EXPECT_EQ(access_count, 1); + EXPECT_EQ(key_access_tracker[key].size(), 1); +} + TEST_F(ServerHandlerTest, UserGetOrderedSetTest) { Key key = "key"; ordered_set s;