use read-write lock instead

Author: BewareMyPower

include/pulsar/Client.h

@@ -436,7 +436,7 @@ class PULSAR_PUBLIC Client {
      *
      * @return the current service information
      */
-    ServiceInfo getServiceInfo();
+    ServiceInfo getServiceInfo() const;
 
    private:
     Client(const std::shared_ptr<ClientImpl>&);

lib/Client.cc

@@ -200,6 +200,6 @@ void Client::getSchemaInfoAsync(const std::string& topic, int64_t version,
 
 void Client::updateServiceInfo(const ServiceInfo& serviceInfo) { impl_->updateServiceInfo(serviceInfo); }
 
-ServiceInfo Client::getServiceInfo() { return impl_->getServiceInfo(); }
+ServiceInfo Client::getServiceInfo() const { return impl_->getServiceInfo(); }
 
 }  // namespace pulsar

lib/ClientImpl.cc

@@ -24,7 +24,9 @@
 #include <algorithm>
 #include <chrono>
 #include <iterator>
+#include <mutex>
 #include <random>
+#include <shared_mutex>
 #include <sstream>
 
 #include "BinaryProtoLookupService.h"
@@ -74,8 +76,6 @@ std::string generateRandomName() {
     return randomName;
 }
 
-typedef std::unique_lock<std::mutex> Lock;
-
 typedef std::vector<std::string> StringList;
 
 namespace {
@@ -157,19 +157,26 @@ ExecutorServiceProviderPtr ClientImpl::getPartitionListenerExecutorProvider() {
 }
 
 LookupServicePtr ClientImpl::getLookup(const std::string& redirectedClusterURI) {
-    Lock lock(mutex_);
+    std::shared_lock readLock(mutex_);
     if (redirectedClusterURI.empty()) {
         return lookupServicePtr_;
     }
 
-    auto it = redirectedClusterLookupServicePtrs_.find(redirectedClusterURI);
-    if (it == redirectedClusterLookupServicePtrs_.end()) {
-        auto lookup = createLookup(redirectedClusterURI);
-        redirectedClusterLookupServicePtrs_.emplace(redirectedClusterURI, lookup);
-        return lookup;
+    if (auto it = redirectedClusterLookupServicePtrs_.find(redirectedClusterURI);
+        it != redirectedClusterLookupServicePtrs_.end()) {
+        return it->second;
     }
+    readLock.unlock();
 
-    return it->second;
+    std::unique_lock writeLock(mutex_);
+    // Double check in case another thread acquires the lock and inserts a pair first
+    if (auto it = redirectedClusterLookupServicePtrs_.find(redirectedClusterURI);
+        it != redirectedClusterLookupServicePtrs_.end()) {
+        return it->second;
+    }
+    auto lookup = createLookup(redirectedClusterURI);
+    redirectedClusterLookupServicePtrs_.emplace(redirectedClusterURI, lookup);
+    return lookup;
 }
 
 void ClientImpl::createProducerAsync(const std::string& topic, const ProducerConfiguration& conf,
@@ -179,7 +186,7 @@ void ClientImpl::createProducerAsync(const std::string& topic, const ProducerCon
     }
     TopicNamePtr topicName;
     {
-        Lock lock(mutex_);
+        std::shared_lock lock(mutex_);
         if (state_ != Open) {
             lock.unlock();
             callback(ResultAlreadyClosed, Producer());
@@ -267,7 +274,7 @@ void ClientImpl::createReaderAsync(const std::string& topic, const MessageId& st
                                    const ReaderConfiguration& conf, const ReaderCallback& callback) {
     TopicNamePtr topicName;
     {
-        Lock lock(mutex_);
+        std::shared_lock lock(mutex_);
         if (state_ != Open) {
             lock.unlock();
             callback(ResultAlreadyClosed, Reader());
@@ -289,7 +296,7 @@ void ClientImpl::createTableViewAsync(const std::string& topic, const TableViewC
                                       const TableViewCallback& callback) {
     TopicNamePtr topicName;
     {
-        Lock lock(mutex_);
+        std::shared_lock lock(mutex_);
         if (state_ != Open) {
             lock.unlock();
             callback(ResultAlreadyClosed, TableView());
@@ -355,7 +362,7 @@ void ClientImpl::subscribeWithRegexAsync(const std::string& regexPattern, const
                                          const SubscribeCallback& callback) {
     TopicNamePtr topicNamePtr = TopicName::get(regexPattern);
 
-    Lock lock(mutex_);
+    std::shared_lock lock(mutex_);
     if (state_ != Open) {
         lock.unlock();
         callback(ResultAlreadyClosed, Consumer());
@@ -441,7 +448,7 @@ void ClientImpl::subscribeAsync(const std::vector<std::string>& originalTopics,
     auto it = std::unique(topics.begin(), topics.end());
     auto newSize = std::distance(topics.begin(), it);
     topics.resize(newSize);
-    Lock lock(mutex_);
+    std::shared_lock lock(mutex_);
     if (state_ != Open) {
         lock.unlock();
         callback(ResultAlreadyClosed, Consumer());
@@ -477,7 +484,7 @@ void ClientImpl::subscribeAsync(const std::string& topic, const std::string& sub
                                 const ConsumerConfiguration& conf, const SubscribeCallback& callback) {
     TopicNamePtr topicName;
     {
-        Lock lock(mutex_);
+        std::shared_lock lock(mutex_);
         if (state_ != Open) {
             lock.unlock();
             callback(ResultAlreadyClosed, Consumer());
@@ -662,7 +669,7 @@ void ClientImpl::handleGetPartitions(Result result, const LookupDataResultPtr& p
 void ClientImpl::getPartitionsForTopicAsync(const std::string& topic, const GetPartitionsCallback& callback) {
     TopicNamePtr topicName;
     {
-        Lock lock(mutex_);
+        std::shared_lock lock(mutex_);
         if (state_ != Open) {
             lock.unlock();
             callback(ResultAlreadyClosed, StringList());
@@ -679,7 +686,9 @@ void ClientImpl::getPartitionsForTopicAsync(const std::string& topic, const GetP
 }
 
 void ClientImpl::closeAsync(const CloseCallback& callback) {
+    std::unique_lock lock(mutex_);
     if (state_ != Open) {
+        lock.unlock();
         if (callback) {
             callback(ResultAlreadyClosed);
         }
@@ -689,10 +698,12 @@ void ClientImpl::closeAsync(const CloseCallback& callback) {
     state_ = Closing;
 
     memoryLimitController_.close();
-    getLookup()->close();
+    lookupServicePtr_->close();
     for (const auto& it : redirectedClusterLookupServicePtrs_) {
         it.second->close();
     }
+    redirectedClusterLookupServicePtrs_.clear();
+    lock.unlock();
 
     auto producers = producers_.move();
     auto consumers = consumers_.move();
@@ -741,7 +752,7 @@ void ClientImpl::handleClose(Result result, const SharedInt& numberOfOpenHandler
         --(*numberOfOpenHandlers);
     }
     if (*numberOfOpenHandlers == 0) {
-        Lock lock(mutex_);
+        std::unique_lock lock(mutex_);
         if (state_ == Closed) {
             LOG_DEBUG("Client is already shutting down, possible race condition in handleClose");
             return;
@@ -821,12 +832,12 @@ void ClientImpl::shutdown() {
 }
 
 uint64_t ClientImpl::newProducerId() {
-    Lock lock(mutex_);
+    std::shared_lock lock(mutex_);
     return producerIdGenerator_++;
 }
 
 uint64_t ClientImpl::newConsumerId() {
-    Lock lock(mutex_);
+    std::shared_lock lock(mutex_);
     return consumerIdGenerator_++;
 }
 
@@ -870,51 +881,40 @@ std::chrono::nanoseconds ClientImpl::getOperationTimeout(const ClientConfigurati
 }
 
 void ClientImpl::updateServiceInfo(const ServiceInfo& serviceInfo) {
-    LookupServicePtr oldLookupServicePtr;
-    std::unordered_map<std::string, LookupServicePtr> oldRedirectedLookupServicePtrs;
-
-    {
-        Lock lock(mutex_);
-        if (state_ != Open) {
-            LOG_ERROR("Client is not open, cannot update connection info");
-            return;
-        }
-
-        if (serviceInfo.authentication.has_value() && *serviceInfo.authentication) {
-            clientConfiguration_.setAuth(*serviceInfo.authentication);
-        } else {
-            clientConfiguration_.setAuth(AuthFactory::Disabled());
-        }
-        if (serviceInfo.tlsTrustCertsFilePath.has_value()) {
-            clientConfiguration_.setTlsTrustCertsFilePath(*serviceInfo.tlsTrustCertsFilePath);
-        } else {
-            clientConfiguration_.setTlsTrustCertsFilePath("");
-        }
-        clientConfiguration_.setUseTls(ServiceNameResolver::useTls(ServiceURI(serviceInfo.serviceUrl)));
-        serviceInfo_ = {serviceInfo.serviceUrl, toOptionalAuthentication(clientConfiguration_.getAuthPtr()),
-                        clientConfiguration_.getTlsTrustCertsFilePath().empty()
-                            ? std::nullopt
-                            : std::make_optional(clientConfiguration_.getTlsTrustCertsFilePath())};
-
-        oldLookupServicePtr = std::move(lookupServicePtr_);
-        oldRedirectedLookupServicePtrs = std::move(redirectedClusterLookupServicePtrs_);
-
-        lookupServicePtr_ = createLookup(serviceInfo.serviceUrl);
-        redirectedClusterLookupServicePtrs_.clear();
+    std::unique_lock lock(mutex_);
+    if (state_ != Open) {
+        LOG_ERROR("Client is not open, cannot update connection info");
+        return;
     }
 
-    if (oldLookupServicePtr) {
-        oldLookupServicePtr->close();
+    if (serviceInfo.authentication.has_value() && *serviceInfo.authentication) {
+        clientConfiguration_.setAuth(*serviceInfo.authentication);
+    } else {
+        clientConfiguration_.setAuth(AuthFactory::Disabled());
     }
-    for (const auto& it : oldRedirectedLookupServicePtrs) {
-        it.second->close();
+    if (serviceInfo.tlsTrustCertsFilePath.has_value()) {
+        clientConfiguration_.setTlsTrustCertsFilePath(*serviceInfo.tlsTrustCertsFilePath);
+    } else {
+        clientConfiguration_.setTlsTrustCertsFilePath("");
     }
+    clientConfiguration_.setUseTls(ServiceNameResolver::useTls(ServiceURI(serviceInfo.serviceUrl)));
+    serviceInfo_ = {serviceInfo.serviceUrl, toOptionalAuthentication(clientConfiguration_.getAuthPtr()),
+                    clientConfiguration_.getTlsTrustCertsFilePath().empty()
+                        ? std::nullopt
+                        : std::make_optional(clientConfiguration_.getTlsTrustCertsFilePath())};
 
     pool_.resetConnections(clientConfiguration_.getAuthPtr(), clientConfiguration_);
+
+    lookupServicePtr_->close();
+    for (auto&& it : redirectedClusterLookupServicePtrs_) {
+        it.second->close();
+    }
+    redirectedClusterLookupServicePtrs_.clear();
+    lookupServicePtr_ = createLookup(serviceInfo.serviceUrl);
 }
 
-ServiceInfo ClientImpl::getServiceInfo() {
-    Lock lock(mutex_);
+ServiceInfo ClientImpl::getServiceInfo() const {
+    std::shared_lock lock(mutex_);
     return serviceInfo_;
 }
 

lib/ClientImpl.h

@@ -24,6 +24,7 @@
 #include <atomic>
 #include <cstdint>
 #include <memory>
+#include <shared_mutex>
 
 #include "ConnectionPool.h"
 #include "Future.h"
@@ -140,7 +141,7 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
     uint64_t getLookupCount() { return lookupCount_; }
 
     void updateServiceInfo(const ServiceInfo& serviceInfo);
-    ServiceInfo getServiceInfo();
+    ServiceInfo getServiceInfo() const;
 
     static std::chrono::nanoseconds getOperationTimeout(const ClientConfiguration& clientConfiguration);
 
@@ -191,7 +192,7 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
         Closed
     };
 
-    std::mutex mutex_;
+    mutable std::shared_mutex mutex_;
 
     State state_;
     ClientConfiguration clientConfiguration_;

tests/ClientTest.cc

@@ -509,7 +509,7 @@ TEST(ClientTest, testNoRetry) {
     }
 }
 
-TEST(ClientTest, testUpdateConnectionInfo) {
+TEST(ClientTest, testUpdateServiceInfo) {
     extern std::string getToken();  // from AuthToken.cc
 
     // Access "private/auth" namespace in cluster 1