Miscellaneous bugfixes (#2)

* Fixes metadata tier ID parsing in monitoring system
* Fixes node join re-gossip bug: The join_remove_set was not being cleared after messaging, so the system attempted to gossip the same keys over and over again (and crashed).

Author: vsreekanti

README.md

@@ -20,4 +20,4 @@ More detailed instructions on [building](docs/building-anna.md) and [running](do
 
 ## License
 
-The Hydro Project is licensed under the [Apache v2 License](LICENSE).
+The Hydro Project is licensed under the [Apache v2 License](LICENSE).

include/metadata.hpp

@@ -100,7 +100,7 @@ extern hmap<Tier, TierMetadata, TierEnumHash> kTierMetadata;
 
 enum MetadataType { replication, server_stats, key_access, key_size };
 
-inline Key get_metadata_key(const ServerThread &st, unsigned tier_id,
+inline Key get_metadata_key(const ServerThread &st, Tier tier_id,
                             unsigned thread_num, MetadataType type) {
   string metadata_type;
 
@@ -122,7 +122,7 @@ inline Key get_metadata_key(const ServerThread &st, unsigned tier_id,
   return kMetadataIdentifier + kMetadataDelimiter + metadata_type +
          kMetadataDelimiter + st.public_ip() + kMetadataDelimiter +
          st.private_ip() + kMetadataDelimiter + std::to_string(thread_num) +
-         kMetadataDelimiter + std::to_string(tier_id);
+         kMetadataDelimiter + Tier_Name(tier_id);
 }
 
 // This version of the function should only be called with

include/proto/metadata.proto

@@ -1,11 +1,11 @@
 //  Copyright 2019 U.C. Berkeley RISE Lab
-// 
+//
 //  Licensed under the Apache License, Version 2.0 (the "License");
 //  you may not use this file except in compliance with the License.
 //  You may obtain a copy of the License at
-// 
+//
 //      http://www.apache.org/licenses/LICENSE-2.0
-// 
+//
 //  Unless required by applicable law or agreed to in writing, software
 //  distributed under the License is distributed on an "AS IS" BASIS,
 //  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -41,7 +41,6 @@ message KeyAccessData {
     // The number of times this key was accessed during this epoch.
     uint32 access_count = 2;
   }
-  
   // A list of all the key access frequencies tracked during this epoch.
   repeated KeyCount keys = 1;
 }
@@ -71,11 +70,11 @@ message ClusterMembership {
     message Server {
       // The public IP address for a server.
       string public_ip = 1;
-      
+
       // The private IP address for a server.
       string private_ip = 2;
     }
- 
+
     // The Tier represented by this message -- either MEMORY or DISK.
     Tier tier_id = 1;
 
@@ -93,15 +92,15 @@ message KeySizeData {
   message KeySize {
     // The key for which size metadata is being reported.
     string key = 1;
-    
+
     // The size of the above key.
     uint32 size = 2;
   }
-  
+
   // The list of key size metadata tuples being reported.
   repeated KeySize key_sizes = 1;
 }
-  
+
 // A message that captures the replication factor for an individual key.
 message ReplicationFactor {
   // A message representing the replication level for a single key at a

src/kvs/server.cpp

@@ -670,9 +670,8 @@ void run(unsigned thread_id, Address public_ip, Address private_ip,
     // redistribute data after node joins
     if (join_gossip_map.size() != 0) {
       set<Address> remove_address_set;
-
-      // assemble gossip
       AddressKeysetMap addr_keyset_map;
+
       for (const auto &join_pair : join_gossip_map) {
         Address address = join_pair.first;
         set<Key> key_set = join_pair.second;
@@ -710,6 +709,8 @@ void run(unsigned thread_id, Address public_ip, Address private_ip,
           serializers[stored_key_map[key].type_]->remove(key);
           stored_key_map.erase(key);
         }
+
+        join_remove_set.clear();
       }
     }
   }

src/monitor/stats_helpers.cpp

@@ -66,7 +66,8 @@ void collect_internal_stats(
           string metadata_type = tokens[1];
           Address ip_pair = tokens[2] + "/" + tokens[3];
           unsigned tid = stoi(tokens[4]);
-          unsigned tier_id = stoi(tokens[5]);
+          Tier tier;
+          Tier_Parse(tokens[5], &tier);
 
           LWWValue lww_value;
           lww_value.ParseFromString(tuple.payload());
@@ -76,7 +77,7 @@ void collect_internal_stats(
             ServerThreadStatistics stat;
             stat.ParseFromString(lww_value.value());
 
-            if (tier_id == 1) {
+            if (tier == MEMORY) {
               memory_storage[ip_pair][tid] = stat.storage_consumption();
               memory_occupancy[ip_pair][tid] =
                   std::pair<double, unsigned>(stat.occupancy(), stat.epoch());