fix deadlock

Author: BewareMyPower

lib/ClientConnection.cc

@@ -1322,8 +1322,12 @@ void ClientConnection::close(Result result, bool detach) {
             if (err) {
                 LOG_WARN(cnxString_ << "Failed to close socket: " << err.message());
             }
-        }  // else: there is an ongoing connect operation, the socket will be closed after the operation
-           // succeeds in `handleTcpConnected`
+        } else {
+            // There is an ongoing connect operation, the socket will be closed after the operation succeeds
+            // in `handleTcpConnected`
+            LOG_WARN(cnxString_
+                     << "Socket is still connecting, it will be closed once the connection attempt finishes");
+        }
     }
     state_ = Disconnected;
     if (tlsSocket_) {

lib/ClientConnection.h

@@ -404,8 +404,12 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
     typedef std::unordered_map<uint64_t, GetSchemaRequest> PendingGetSchemaMap;
     PendingGetSchemaMap pendingGetSchemaRequests_;
 
-    mutable std::mutex mutex_;
-    typedef std::unique_lock<std::mutex> Lock;
+    // Most accesses to this class happen in the event loop thread, but the `close` method could be called in
+    // a different thread by `ConnectionPool`, so we have to guard the state of the connection with a mutex.
+    // However, `close` could be called when a mutex is held, so we have to use a recursive mutex to avoid
+    // deadlock.
+    mutable std::recursive_mutex mutex_;
+    typedef std::unique_lock<std::recursive_mutex> Lock;
 
     // Pending buffers to write on the socket
     std::deque<std::any> pendingWriteBuffers_;

lib/MockServer.h

@@ -42,7 +42,7 @@ class MockServer : public std::enable_shared_from_this<MockServer> {
     }
 
     void setRequestDelay(std::initializer_list<typename RequestDelayType::value_type> delays) {
-        std::lock_guard<std::mutex> lock(mutex_);
+        std::lock_guard lock(mutex_);
         for (auto&& delay : delays) {
             requestDelays_[delay.first] = delay.second;
         }
@@ -53,15 +53,15 @@ class MockServer : public std::enable_shared_from_this<MockServer> {
         if (!connection) {
             return false;
         }
-        std::lock_guard<std::mutex> lock(mutex_);
+        std::lock_guard lock(mutex_);
         if (auto iter = requestDelays_.find(request); iter != requestDelays_.end()) {
             // Mock the `CLOSE_CONSUMER` command sent by broker, for simplicity, disconnect all consumers
             if (request == "SEEK") {
                 schedule(connection, "CLOSE_CONSUMER" + std::to_string(requestId),
                          requestDelays_["CLOSE_CONSUMER"], [connection] {
                              std::vector<uint64_t> consumerIds;
                              {
-                                 std::lock_guard<std::mutex> lock{connection->mutex_};
+                                 std::lock_guard lock{connection->mutex_};
                                  for (auto&& kv : connection->consumers_) {
                                      if (auto consumer = kv.second.lock()) {
                                          consumerIds.push_back(consumer->getConsumerId());
@@ -95,7 +95,7 @@ class MockServer : public std::enable_shared_from_this<MockServer> {
 
     // Return the number of pending timers cancelled
     auto close() {
-        std::lock_guard<std::mutex> lock(mutex_);
+        std::lock_guard lock(mutex_);
         auto result = pendingTimers_.size();
         for (auto&& kv : pendingTimers_) {
             try {
@@ -124,7 +124,7 @@ class MockServer : public std::enable_shared_from_this<MockServer> {
         auto self = shared_from_this();
         timer->async_wait([this, self, key, connection, task{std::move(task)}](const auto& ec) {
             {
-                std::lock_guard<std::mutex> lock(mutex_);
+                std::lock_guard lock(mutex_);
                 pendingTimers_.erase(key);
             }
             if (ec) {

tests/PulsarFriend.h

@@ -146,7 +146,7 @@ class PulsarFriend {
 
     static std::vector<ProducerImplPtr> getProducers(const ClientConnection& cnx) {
         std::vector<ProducerImplPtr> producers;
-        std::lock_guard<std::mutex> lock(cnx.mutex_);
+        std::lock_guard lock(cnx.mutex_);
         for (const auto& kv : cnx.producers_) {
             producers.emplace_back(kv.second.lock());
         }
@@ -155,7 +155,7 @@ class PulsarFriend {
 
     static std::vector<ConsumerImplPtr> getConsumers(const ClientConnection& cnx) {
         std::vector<ConsumerImplPtr> consumers;
-        std::lock_guard<std::mutex> lock(cnx.mutex_);
+        std::lock_guard lock(cnx.mutex_);
         for (const auto& kv : cnx.consumers_) {
             consumers.emplace_back(kv.second.lock());
         }