improvement(hashmap): use folly ConcurrentHashMap replace unordered_map

.gitignore

@@ -33,7 +33,7 @@ cmake-build-release
 
 # Log path
 make_config.mk
-log/
+logs/
 lib/
 output/
 

src/helper.h

@@ -7,7 +7,7 @@
 
 #pragma once
 
-#include <cstdlib>
+#include <random>
 #include <vector>
 #include "pstring.h"
 
@@ -48,6 +48,58 @@ inline typename HASH::const_local_iterator RandomHashMember(const HASH& containe
   return typename HASH::const_local_iterator();
 }
 
+template <typename HASH>
+inline typename HASH::const_iterator FollyRandomHashMember(const HASH& container) {
+  auto it = container.cend();
+  if (container.empty()) {
+    return it;
+  }
+
+  it = container.cbegin();
+  std::random_device rd;
+  std::mt19937 gen(rd());
+  std::uniform_int_distribution<> dis(0, container.size() - 1);
+
+  size_t randomIdx = dis(gen);
+  while (randomIdx > 0) {
+    randomIdx--;
+    ++it;
+  }
+
+  return it;
+}
+
+template <typename HASH>
+inline size_t FollyScanHashMember(const HASH& container, size_t cursor, size_t count,
+                             std::vector<PString>& res) {
+  if (cursor >= container.size()) {
+    return 0;
+  }
+
+  // find corresponding iterator
+  size_t idx = cursor;
+  auto it = container.cbegin();
+  while (idx > 0) {
+    --idx;
+    ++it;
+  }
+
+  size_t new_cursor = cursor;
+  auto end = container.cend();
+  while (res.size() < count && it != end) {
+    ++new_cursor;
+    res.push_back(it->first);
+    ++it;
+  }
+
+  // not the last element of map
+  if (it != end) {
+    return new_cursor;
+  }
+
+  return 0;
+}
+
 // scan
 template <typename HASH>
 inline size_t ScanHashMember(const HASH& container, size_t cursor, size_t count,

src/store.cc

@@ -112,7 +112,7 @@ void PObject::freeValue() {
 
 int PStore::dirty_ = 0;
 
-void PStore::ExpiredDB::SetExpire(const PString& key, uint64_t when) { expireKeys_[key] = when; }
+void PStore::ExpiredDB::SetExpire(const PString& key, uint64_t when) { expireKeys_.insert_or_assign(key, when); }
 
 int64_t PStore::ExpiredDB::TTL(const PString& key, uint64_t now) {
   if (!PSTORE.ExistsKey(key)) {
@@ -147,6 +147,7 @@ PStore::ExpireResult PStore::ExpiredDB::ExpireIfNeed(const PString& key, uint64_
 
     WARN("Delete timeout key {}", it->first);
     PSTORE.DeleteKey(it->first);
+    // XXX: may throw exception if hash function crash
     expireKeys_.erase(it);
     return ExpireResult::expired;
   }
@@ -170,7 +171,7 @@ int PStore::ExpiredDB::LoopCheck(uint64_t now) {
       Propagate(params);
 
       PSTORE.DeleteKey(it->first);
-      expireKeys_.erase(it++);
+      it = expireKeys_.erase(it);
 
       ++nDel;
     } else {
@@ -188,8 +189,17 @@ bool PStore::BlockedClients::BlockClient(const PString& key, PClient* client, ui
     return false;
   }
 
-  Clients& clients = blockedClients_[key];
-  clients.push_back(Clients::value_type(std::static_pointer_cast<PClient>(client->shared_from_this()), timeout, pos));
+  auto it = blockedClients_.find(key);
+  if (it != blockedClients_.end()) {
+    // since folly doesn't support modify value directly, we have to make a copy to update atomically
+    auto clients = it->second;
+    clients.emplace_back(std::static_pointer_cast<PClient>(client->shared_from_this()), timeout, pos);
+    blockedClients_.assign(key, clients);
+  } else {
+    std::list<std::tuple<std::weak_ptr<PClient>, uint64_t, ListPosition>> clients;
+    clients.emplace_back(std::static_pointer_cast<PClient>(client->shared_from_this()), timeout, pos);
+    blockedClients_.insert(key, clients);
+  }
 
   INFO("{} is waited by {}, timeout {}", key, client->GetName(), timeout);
   return true;
@@ -200,7 +210,7 @@ size_t PStore::BlockedClients::UnblockClient(PClient* client) {
   const auto& keys = client->WaitingKeys();
 
   for (const auto& key : keys) {
-    Clients& clients = blockedClients_[key];
+    Clients clients = blockedClients_.find(key)->second;
     assert(!clients.empty());
 
     for (auto it(clients.begin()); it != clients.end(); ++it) {
@@ -213,6 +223,7 @@ size_t PStore::BlockedClients::UnblockClient(PClient* client) {
         break;
       }
     }
+    blockedClients_.assign(key, clients);
   }
 
   client->ClearWaitingKeys();
@@ -227,7 +238,7 @@ size_t PStore::BlockedClients::ServeClient(const PString& key, const PLIST& list
     return 0;
   }
 
-  Clients& clients = it->second;
+  Clients clients = it->second;
   if (clients.empty()) {
     return 0;
   }
@@ -300,6 +311,7 @@ size_t PStore::BlockedClients::ServeClient(const PString& key, const PLIST& list
       ++nServed;
     } else {
       clients.pop_front();
+      blockedClients_.assign(it->first, clients);
     }
   }
 
@@ -310,7 +322,8 @@ int PStore::BlockedClients::LoopCheck(uint64_t now) {
   int n = 0;
 
   for (auto it(blockedClients_.begin()); it != blockedClients_.end() && n < 100;) {
-    Clients& clients = it->second;
+    // iterator in folly::ConcurrentHashMap is always const
+    Clients clients = it->second;
     for (auto cli(clients.begin()); cli != clients.end();) {
       if (std::get<1>(*cli) < now) {  // timeout
         ++n;
@@ -326,13 +339,15 @@ int PStore::BlockedClients::LoopCheck(uint64_t now) {
         }
 
         clients.erase(cli++);
+        // clients is a copy, once it's modified, map should update it
+        blockedClients_.assign(it->first, clients);
       } else {
         ++cli;
       }
     }
 
     if (clients.empty()) {
-      blockedClients_.erase(it++);
+      it = blockedClients_.erase(it);
     } else {
       ++it;
     }
@@ -388,7 +403,7 @@ const PObject* PStore::GetObject(const PString& key) const {
 
   if (!backends_.empty()) {
     // if it's in dirty list, it must be deleted, wait sync to backend
-    if (waitSyncKeys_[dbno_].count(key)) {
+    if (waitSyncKeys_[dbno_].find(key) != waitSyncKeys_[dbno_].end()) {
       return nullptr;
     }
 
@@ -397,12 +412,14 @@ const PObject* PStore::GetObject(const PString& key) const {
     if (obj.type != PType_invalid) {
       DEBUG("GetKey from leveldb:{}", key);
 
-      (*db)[key] = std::move(obj);
-      PObject& realobj = (*db)[key];
-      realobj.lru = PObject::lruclock;
+      unsigned int remainTtlSecondsTemp = obj.lru;
+      // XXX: assign lru here cause everything get from db is immutable
+      obj.lru = PObject::lruclock;
+      db->try_emplace(key, std::move(obj));
+      const PObject& realobj = db->find(key)->second;
 
       // trick: use lru field to store the remain seconds to be expired.
-      unsigned int remainTtlSeconds = obj.lru;
+      unsigned int remainTtlSeconds = remainTtlSecondsTemp;
       if (remainTtlSeconds > 0) {
         SetExpire(key, ::Now() + remainTtlSeconds * 1000);
       }
@@ -418,10 +435,18 @@ bool PStore::DeleteKey(const PString& key) {
   auto db = &dbs_[dbno_];
   // add to dirty queue
   if (!waitSyncKeys_.empty()) {
-    waitSyncKeys_[dbno_][key] = nullptr;  // null implies delete data
+    waitSyncKeys_[dbno_].insert_or_assign(key, nullptr);  // null implies delete data
+  }
+
+  size_t ret = 0;
+  // erase() from folly ConcurrentHashmap will throw an exception if hash function crashes
+  try {
+   ret = db->erase(key);
+  } catch (const std::exception& e) {
+   return false;
   }
 
-  return db->erase(key) != 0;
+  return ret != 0;
 }
 
 bool PStore::ExistsKey(const PString& key) const {
@@ -439,9 +464,9 @@ PType PStore::KeyType(const PString& key) const {
 }
 
 static bool RandomMember(const PDB& hash, PString& res, PObject** val) {
-  PDB::const_local_iterator it = RandomHashMember(hash);
+  PDB::const_iterator it = FollyRandomHashMember(hash);
 
-  if (it != PDB::const_local_iterator()) {
+  if (it != hash.cend()) {
     res = it->first;
     if (val) {
       *val = const_cast<PObject*>(&it->second);
@@ -466,12 +491,12 @@ size_t PStore::ScanKey(size_t cursor, size_t count, std::vector<PString>& res) c
     return 0;
   }
 
-  std::vector<PDB::const_local_iterator> iters;
-  size_t newCursor = ScanHashMember(dbs_[dbno_], cursor, count, iters);
+  std::vector<PString> iters;
+  size_t newCursor = FollyScanHashMember(dbs_[dbno_], cursor, count, iters);
 
   res.reserve(iters.size());
-  for (auto it : iters) {
-    res.push_back(it->first);
+  for (const auto& it : iters) {
+    res.push_back(it);
   }
 
   return newCursor;
@@ -518,16 +543,17 @@ PError PStore::getValueByType(const PString& key, PObject*& value, PType type, b
 
 PObject* PStore::SetValue(const PString& key, PObject&& value) {
   auto db = &dbs_[dbno_];
-  (*db)[key] = std::move(value);
-  PObject& obj = (*db)[key];
-  obj.lru = PObject::lruclock;
+  value.lru = PObject::lruclock;
+  auto [realObj, status] = db->insert_or_assign(key, std::move(value));
+  const PObject& obj = realObj->second;
 
   // put this key to sync list
   if (!waitSyncKeys_.empty()) {
-    waitSyncKeys_[dbno_][key] = &obj;
+    waitSyncKeys_[dbno_].insert_or_assign(key, &obj);
   }
 
-  return &obj;
+  // XXX: any better solution without const_cast?
+  return const_cast<PObject*>(&obj);
 }
 
 void PStore::SetExpire(const PString& key, uint64_t when) const { expiredDBs_[dbno_].SetExpire(key, when); }
@@ -726,14 +752,14 @@ void PStore::AddDirtyKey(const PString& key) {
   if (!waitSyncKeys_.empty()) {
     PObject* obj = nullptr;
     GetValue(key, obj);
-    waitSyncKeys_[dbno_][key] = obj;
+    waitSyncKeys_[dbno_].insert_or_assign(key, obj);
   }
 }
 
 void PStore::AddDirtyKey(const PString& key, const PObject* value) {
   // put this key to sync list
   if (!waitSyncKeys_.empty()) {
-    waitSyncKeys_[dbno_][key] = value;
+    waitSyncKeys_[dbno_].insert_or_assign(key, value);
   }
 }
 

src/store.h

@@ -7,13 +7,16 @@
 
 #pragma once
 
+#define GLOG_NO_ABBREVIATED_SEVERITIES
+
 #include "common.h"
 #include "dump_interface.h"
 #include "hash.h"
 #include "list.h"
 #include "set.h"
 #include "sorted_set.h"
 
+#include <folly/concurrency/ConcurrentHashMap.h>
 #include <map>
 #include <memory>
 #include <vector>
@@ -79,7 +82,7 @@ struct PObject {
 
 class PClient;
 
-using PDB = std::unordered_map<PString, PObject, my_hash, std::equal_to<PString> >;
+using PDB = folly::ConcurrentHashMap<PString, PObject, my_hash, std::equal_to<PString>>;
 
 const int kMaxDBNum = 65536;
 
@@ -106,8 +109,6 @@ class PStore {
   // iterator
   PDB::const_iterator begin() const { return dbs_[dbno_].begin(); }
   PDB::const_iterator end() const { return dbs_[dbno_].end(); }
-  PDB::iterator begin() { return dbs_[dbno_].begin(); }
-  PDB::iterator end() { return dbs_[dbno_].end(); }
 
   const PObject* GetObject(const PString& key) const;
   PError GetValue(const PString& key, PObject*& value, bool touch = true);
@@ -171,7 +172,7 @@ class PStore {
     int LoopCheck(uint64_t now);
 
    private:
-    using P_EXPIRE_DB = std::unordered_map<PString, uint64_t, my_hash, std::equal_to<PString> >;
+    using P_EXPIRE_DB = folly::ConcurrentHashMap<PString, uint64_t, my_hash, std::equal_to<PString>>;
     P_EXPIRE_DB expireKeys_;  // all the keys to be expired, unordered.
   };
 
@@ -187,7 +188,7 @@ class PStore {
 
    private:
     using Clients = std::list<std::tuple<std::weak_ptr<PClient>, uint64_t, ListPosition> >;
-    using WaitingList = std::unordered_map<PString, Clients>;
+    using WaitingList = folly::ConcurrentHashMap<PString, Clients, my_hash, std::equal_to<PString>>;
 
     WaitingList blockedClients_;
   };
@@ -200,7 +201,7 @@ class PStore {
   std::vector<BlockedClients> blockedClients_;
   std::vector<std::unique_ptr<PDumpInterface> > backends_;
 
-  using ToSyncDB = std::unordered_map<PString, const PObject*, my_hash, std::equal_to<PString> >;
+  using ToSyncDB = folly::ConcurrentHashMap<PString, const PObject*, my_hash, std::equal_to<PString> >;
   std::vector<ToSyncDB> waitSyncKeys_;
   int dbno_ = -1;
 };