fix thread safety when calling isClosed and add explanations

Author: BewareMyPower

lib/ClientConnection.cc

@@ -417,8 +417,10 @@ void ClientConnection::handleTcpConnected(const ASIO_ERROR& err, const tcp::endp
     Lock lock(mutex_);
     if (!err) {
         if (isClosed()) {
-            ASIO_ERROR err;
-            socket_->close(err);
+            if (socket_) {
+                ASIO_ERROR err;
+                socket_->close(err);
+            }
             connectPromise_.setFailed(ResultAlreadyClosed);
             LOG_INFO(cnxString_ << "Connection already closed");
             return;
@@ -674,9 +676,11 @@ void ClientConnection::readNextCommand() {
 }
 
 void ClientConnection::handleRead(const ASIO_ERROR& err, size_t bytesTransferred, uint32_t minReadSize) {
+    Lock lock{mutex_};
     if (isClosed()) {
         return;
     }
+    lock.unlock();
     // Update buffer write idx with new data
     incomingBuffer_.bytesWritten(bytesTransferred);
 
@@ -1145,9 +1149,11 @@ void ClientConnection::sendMessage(const std::shared_ptr<SendArguments>& args) {
 }
 
 void ClientConnection::handleSend(const ASIO_ERROR& err, const SharedBuffer&) {
+    Lock lock{mutex_};
     if (isClosed()) {
         return;
     }
+    lock.unlock();
     if (err) {
         LOG_WARN(cnxString_ << "Could not send message on connection: " << err << " " << err.message());
         close(ResultDisconnected);
@@ -1269,9 +1275,11 @@ void ClientConnection::handleGetLastMessageIdTimeout(const ASIO_ERROR& ec,
 }
 
 void ClientConnection::handleKeepAliveTimeout() {
+    Lock lock{mutex_};
     if (isClosed()) {
         return;
     }
+    lock.unlock();
 
     if (havePendingPingRequest_) {
         LOG_WARN(cnxString_ << "Forcing connection to close after keep-alive timeout");
@@ -1325,8 +1333,9 @@ void ClientConnection::close(Result result, bool detach) {
         } 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");
+            LOG_DEBUG(
+                cnxString_
+                << "Socket is still connecting, it will be closed once the connection attempt finishes");
         }
     }
     state_ = Disconnected;

lib/ClientConnection.h

@@ -441,6 +441,15 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
     const std::string clientVersion_;
     ConnectionPool& pool_;
     const size_t poolIndex_;
+
+    // This field is used to check if there is any pending operation when closing the connection pool. It's
+    // because the event loop thread, which runs `io_context::run()`, will be stopped immediately by
+    // `io_context::stop()` after closing the pool. If the pending operation's callback handler is not called
+    // before the event loop is stopped, it could crash because the destructor of `io_context` will destroy
+    // internal task queues so that the asynchronous operation's callback handler could refer and access a
+    // destroyed object.
+    // This counter is added to track the pending operation count so that the pool can wait for a certain time
+    // until all pending operations are completed when closing this connection.
     std::atomic_int pendingOperations_{0};
 
     friend class PulsarFriend;