lmdb-safe: Improve the scalability of transaction maps

This commit improves the scalability of the transaction maps by
moving from a `mutex` to a `shared_mutex` and making the stored value
atomic. This allows accessing the maps concurrently from different threads
as long as the entry for the thread exists, and we fall back to
taking a full lock otherwise.

Signed-off-by: Remi Gacogne <remi.gacogne@powerdns.com>
(cherry picked from commit c340aa91bf37d8105d2b2390eecbadfca88c1d27)

diff --git a/ext/lmdb-safe/lmdb-safe.cc b/ext/lmdb-safe/lmdb-safe.cc
index 128e0c398b8dbcf2c5bc914cb839d9b5e61d0017..b2ed4291f984c3d1039d258d2ad07ecb9659dcb0 100644 (file)
--- a/ext/lmdb-safe/lmdb-safe.cc
+++ b/ext/lmdb-safe/lmdb-safe.cc
@@ -140,37 +140,84 @@ Various other options may also need to be set before opening the handle, e.g. md
 
 void MDBEnv::incROTX()
 {
-  std::lock_guard<std::mutex> l(d_countmutex);
-  ++d_ROtransactionsOut[std::this_thread::get_id()];
+  auto threadId = std::this_thread::get_id();
+  {
+    std::shared_lock<std::shared_mutex> lock(d_countmutex);
+    if (auto transactionsIt = d_ROtransactionsOut.find(threadId); transactionsIt != d_ROtransactionsOut.end()) {
+      ++transactionsIt->second;
+      return;
+    }
+  }
+
+  {
+    std::unique_lock<std::shared_mutex> lock(d_countmutex);
+    auto [transactionsIt, inserted] = d_ROtransactionsOut.emplace(threadId, 1);
+    if (!inserted) {
+      ++transactionsIt->second;
+    }
+  }
 }
 
 void MDBEnv::decROTX()
 {
-  std::lock_guard<std::mutex> l(d_countmutex);
-  --d_ROtransactionsOut[std::this_thread::get_id()];
+  auto threadId = std::this_thread::get_id();
+  {
+    std::shared_lock<std::shared_mutex> lock(d_countmutex);
+    d_ROtransactionsOut.at(threadId)--;
+  }
 }
 
 void MDBEnv::incRWTX()
 {
-  std::lock_guard<std::mutex> l(d_countmutex);
-  ++d_RWtransactionsOut[std::this_thread::get_id()];
+  auto threadId = std::this_thread::get_id();
+  {
+    std::shared_lock<std::shared_mutex> lock(d_countmutex);
+    if (auto transactionsIt = d_RWtransactionsOut.find(threadId); transactionsIt != d_RWtransactionsOut.end()) {
+      ++transactionsIt->second;
+      return;
+    }
+  }
+
+  {
+    std::unique_lock<std::shared_mutex> lock(d_countmutex);
+    auto [transactionsIt, inserted] = d_RWtransactionsOut.emplace(threadId, 1);
+    if (!inserted) {
+      ++transactionsIt->second;
+    }
+  }
 }
 
 void MDBEnv::decRWTX()
 {
-  std::lock_guard<std::mutex> l(d_countmutex);
-  --d_RWtransactionsOut[std::this_thread::get_id()];
+  auto threadId = std::this_thread::get_id();
+  {
+    std::shared_lock<std::shared_mutex> lock(d_countmutex);
+    d_RWtransactionsOut.at(threadId)--;
+  }
 }
 
 int MDBEnv::getRWTX()
 {
-  std::lock_guard<std::mutex> l(d_countmutex);
-  return d_RWtransactionsOut[std::this_thread::get_id()];
+  auto threadId = std::this_thread::get_id();
+  {
+    std::shared_lock<std::shared_mutex> lock(d_countmutex);
+    if (auto transactionsIt = d_RWtransactionsOut.find(threadId); transactionsIt != d_RWtransactionsOut.end()) {
+      return transactionsIt->second.load();
+    }
+  }
+  return 0;
 }
+
 int MDBEnv::getROTX()
 {
-  std::lock_guard<std::mutex> l(d_countmutex);
-  return d_ROtransactionsOut[std::this_thread::get_id()];
+  auto threadId = std::this_thread::get_id();
+  {
+    std::shared_lock<std::shared_mutex> lock(d_countmutex);
+    if (auto transactionsIt = d_RWtransactionsOut.find(threadId); transactionsIt != d_RWtransactionsOut.end()) {
+      return transactionsIt->second.load();
+    }
+  }
+  return 0;
 }
 
 

diff --git a/ext/lmdb-safe/lmdb-safe.hh b/ext/lmdb-safe/lmdb-safe.hh
index f9b4dbf5ce58d9c65eb021a30e613d89222628d3..2f19124c698f5b1b96e4b9cf6c3e2fa9fba94957 100644 (file)
--- a/ext/lmdb-safe/lmdb-safe.hh
+++ b/ext/lmdb-safe/lmdb-safe.hh
@@ -13,6 +13,7 @@
 #include <mutex>
 #include <vector>
 #include <algorithm>
+#include <shared_mutex>
 #include <string>
 #include <string_view>
 #include <atomic>
@@ -109,9 +110,9 @@ public:
   void decROTX();
 private:
   std::mutex d_openmut;
-  std::mutex d_countmutex;
-  std::map<std::thread::id, int> d_RWtransactionsOut;
-  std::map<std::thread::id, int> d_ROtransactionsOut;
+  std::shared_mutex d_countmutex;
+  std::map<std::thread::id, std::atomic<int>> d_RWtransactionsOut;
+  std::map<std::thread::id, std::atomic<int>> d_ROtransactionsOut;
 };
 
 std::shared_ptr<MDBEnv> getMDBEnv(const char* fname, int flags, int mode, uint64_t mapsizeMB=(sizeof(void *)==4) ? 100 : 16000);