[incubator-kvrocks] branch unstable updated: Add separate mutex that protects blocked stream consumers (#1471)

[hu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

hulk pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/incubator-kvrocks.git


The following commit(s) were added to refs/heads/unstable by this push:
     new 271c0dfa Add separate mutex that protects blocked stream consumers (#1471)
271c0dfa is described below

commit 271c0dfa9d837f1ddd308445566f3c9786be9a4e
Author: Yaroslav <to...@gmail.com>
AuthorDate: Fri May 26 07:15:16 2023 +0300

    Add separate mutex that protects blocked stream consumers (#1471)
    
    This separates blocking reads on streams from blocking reads on other data structures
    (lists/sorted sets/etc) reducing lock contention on a single mutex for all the blocking keys.
---
 src/server/server.cc | 9 +++------
 src/server/server.h  | 3 +++
 2 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/src/server/server.cc b/src/server/server.cc
index 5e20984d..a8ba3e63 100644
--- a/src/server/server.cc
+++ b/src/server/server.cc
@@ -20,7 +20,6 @@
 
 #include "server.h"
 
-#include <fcntl.h>
 #include <glog/logging.h>
 #include <rocksdb/convenience.h>
 #include <rocksdb/statistics.h>
@@ -41,14 +40,12 @@
 #include "fmt/format.h"
 #include "lua.h"
 #include "redis_connection.h"
-#include "redis_request.h"
 #include "storage/compaction_checker.h"
 #include "storage/redis_db.h"
 #include "storage/scripting.h"
 #include "string_util.h"
 #include "thread_util.h"
 #include "time_util.h"
-#include "tls_util.h"
 #include "version.h"
 #include "worker.h"
 
@@ -529,7 +526,7 @@ void Server::UnblockOnKey(const std::string &key, redis::Connection *conn) {
 
 void Server::BlockOnStreams(const std::vector<std::string> &keys, const std::vector<redis::StreamEntryID> &entry_ids,
                             redis::Connection *conn) {
-  std::lock_guard<std::mutex> guard(blocking_keys_mu_);
+  std::lock_guard<std::mutex> guard(blocked_stream_consumers_mu_);
 
   IncrBlockedClientNum();
 
@@ -546,7 +543,7 @@ void Server::BlockOnStreams(const std::vector<std::string> &keys, const std::vec
 }
 
 void Server::UnblockOnStreams(const std::vector<std::string> &keys, redis::Connection *conn) {
-  std::lock_guard<std::mutex> guard(blocking_keys_mu_);
+  std::lock_guard<std::mutex> guard(blocked_stream_consumers_mu_);
 
   DecrBlockedClientNum();
 
@@ -590,7 +587,7 @@ void Server::WakeupBlockingConns(const std::string &key, size_t n_conns) {
 }
 
 void Server::OnEntryAddedToStream(const std::string &ns, const std::string &key, const redis::StreamEntryID &entry_id) {
-  std::lock_guard<std::mutex> guard(blocking_keys_mu_);
+  std::lock_guard<std::mutex> guard(blocked_stream_consumers_mu_);
 
   auto iter = blocked_stream_consumers_.find(key);
   if (iter == blocked_stream_consumers_.end() || iter->second.empty()) {
diff --git a/src/server/server.h b/src/server/server.h
index ee31df5e..1f56d644 100644
--- a/src/server/server.h
+++ b/src/server/server.h
@@ -287,7 +287,10 @@ class Server {
   std::mutex pubsub_channels_mu_;
   std::map<std::string, std::list<ConnContext *>> blocking_keys_;
   std::mutex blocking_keys_mu_;
+
   std::atomic<int> blocked_clients_{0};
+
+  std::mutex blocked_stream_consumers_mu_;
   std::map<std::string, std::set<std::shared_ptr<StreamConsumer>>> blocked_stream_consumers_;
 
   // threads