From 5a7cce168dc0c7f05fda9782c2c2d06449cfd402 Mon Sep 17 00:00:00 2001 From: Masahiro Sakamoto Date: Fri, 20 Sep 2019 21:15:33 +0900 Subject: [PATCH 1/6] Fix memory leak caused by deadline_timer holding object reference --- pulsar-client-cpp/lib/ClientConnection.cc | 7 +++++++ pulsar-client-cpp/lib/ClientImpl.cc | 1 + pulsar-client-cpp/lib/ConnectionPool.cc | 16 ++++++++++++++++ pulsar-client-cpp/lib/ConnectionPool.h | 2 ++ pulsar-client-cpp/lib/ProducerImpl.cc | 17 +++++++++++++---- pulsar-client-cpp/lib/ProducerImpl.h | 2 ++ 6 files changed, 41 insertions(+), 4 deletions(-) diff --git a/pulsar-client-cpp/lib/ClientConnection.cc b/pulsar-client-cpp/lib/ClientConnection.cc index f79cd423fc1cc..30c4c096f42c0 100644 --- a/pulsar-client-cpp/lib/ClientConnection.cc +++ b/pulsar-client-cpp/lib/ClientConnection.cc @@ -1338,11 +1338,14 @@ void ClientConnection::close() { if (keepAliveTimer_) { keepAliveTimer_->cancel(); + keepAliveTimer_.reset(); } if (consumerStatsRequestTimer_) { consumerStatsRequestTimer_->cancel(); + consumerStatsRequestTimer_.reset(); } + for (ProducersMap::iterator it = producers.begin(); it != producers.end(); ++it) { HandlerBase::handleDisconnection(ResultConnectError, shared_from_this(), it->second); } @@ -1382,6 +1385,10 @@ void ClientConnection::close() { if (tlsSocket_) { tlsSocket_->lowest_layer().close(); } + + if (executor_) { + executor_.reset(); + } } bool ClientConnection::isClosed() const { return state_ == Disconnected; } diff --git a/pulsar-client-cpp/lib/ClientImpl.cc b/pulsar-client-cpp/lib/ClientImpl.cc index 0eb9034902777..f61e2ff55103f 100644 --- a/pulsar-client-cpp/lib/ClientImpl.cc +++ b/pulsar-client-cpp/lib/ClientImpl.cc @@ -564,6 +564,7 @@ void ClientImpl::shutdown() { ioExecutorProvider_->close(); listenerExecutorProvider_->close(); partitionListenerExecutorProvider_->close(); + pool_.close(); } uint64_t ClientImpl::newProducerId() { diff --git a/pulsar-client-cpp/lib/ConnectionPool.cc b/pulsar-client-cpp/lib/ConnectionPool.cc index 86c89489979d4..83fbd21ce2cbd 100644 --- a/pulsar-client-cpp/lib/ConnectionPool.cc +++ b/pulsar-client-cpp/lib/ConnectionPool.cc @@ -80,4 +80,20 @@ Future ConnectionPool::getConnectionAsync( return future; } +void ConnectionPool::close() { + std::unique_lock lock(mutex_); + + if (poolConnections_) { + for (auto cnxIt = pool_.begin(); cnxIt != pool_.end(); cnxIt++) { + ClientConnectionPtr cnx = cnxIt->second.lock(); + if (cnx && !cnx->isClosed()) { + cnx->close(); + } + } + pool_.clear(); + } + + lock.unlock(); +} + } // namespace pulsar diff --git a/pulsar-client-cpp/lib/ConnectionPool.h b/pulsar-client-cpp/lib/ConnectionPool.h index f4c1f68f93acf..4f54c88f07b42 100644 --- a/pulsar-client-cpp/lib/ConnectionPool.h +++ b/pulsar-client-cpp/lib/ConnectionPool.h @@ -55,6 +55,8 @@ class PULSAR_PUBLIC ConnectionPool { Future getConnectionAsync(const std::string& logicalAddress, const std::string& physicalAddress); + void close(); + private: ClientConfiguration clientConfiguration_; ExecutorServiceProviderPtr executorProvider_; diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc index 666df3b27f184..8488c3d01b161 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.cc +++ b/pulsar-client-cpp/lib/ProducerImpl.cc @@ -88,9 +88,6 @@ ProducerImpl::ProducerImpl(ClientImplPtr client, const std::string& topic, const ProducerImpl::~ProducerImpl() { LOG_DEBUG(getName() << "~ProducerImpl"); - if (dataKeyGenTImer_) { - dataKeyGenTImer_->cancel(); - } closeAsync(ResultCallback()); printStats(); } @@ -473,6 +470,8 @@ void ProducerImpl::printStats() { void ProducerImpl::closeAsync(CloseCallback callback) { Lock lock(mutex_); + cancelTimers(); + if (state_ != Ready) { lock.unlock(); if (callback) { @@ -682,10 +681,20 @@ void ProducerImpl::start() { HandlerBase::start(); } void ProducerImpl::shutdown() { Lock lock(mutex_); state_ = Closed; + cancelTimers(); + producerCreatedPromise_.setFailed(ResultAlreadyClosed); +} + +void ProducerImpl::cancelTimers() { + if (dataKeyGenTImer_) { + dataKeyGenTImer_->cancel(); + dataKeyGenTImer_.reset(); + } + if (sendTimer_) { sendTimer_->cancel(); + sendTimer_.reset(); } - producerCreatedPromise_.setFailed(ResultAlreadyClosed); } bool ProducerImplCmp::operator()(const ProducerImplPtr& a, const ProducerImplPtr& b) const { diff --git a/pulsar-client-cpp/lib/ProducerImpl.h b/pulsar-client-cpp/lib/ProducerImpl.h index cb2a8a6e8a990..3a3780417d30f 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.h +++ b/pulsar-client-cpp/lib/ProducerImpl.h @@ -139,6 +139,8 @@ class ProducerImpl : public HandlerBase, bool encryptMessage(proto::MessageMetadata& metadata, SharedBuffer& payload, SharedBuffer& encryptedPayload); + void cancelTimers(); + typedef std::unique_lock Lock; ProducerConfiguration conf_; From 7ef79c2190d616418661bdd65c64243ef907315d Mon Sep 17 00:00:00 2001 From: Masahiro Sakamoto Date: Mon, 23 Sep 2019 01:00:53 +0900 Subject: [PATCH 2/6] Add ConnectionPool destructor --- pulsar-client-cpp/lib/ClientImpl.cc | 1 - pulsar-client-cpp/lib/ConnectionPool.cc | 32 ++++++++++++------------- pulsar-client-cpp/lib/ConnectionPool.h | 4 ++-- 3 files changed, 18 insertions(+), 19 deletions(-) diff --git a/pulsar-client-cpp/lib/ClientImpl.cc b/pulsar-client-cpp/lib/ClientImpl.cc index f61e2ff55103f..0eb9034902777 100644 --- a/pulsar-client-cpp/lib/ClientImpl.cc +++ b/pulsar-client-cpp/lib/ClientImpl.cc @@ -564,7 +564,6 @@ void ClientImpl::shutdown() { ioExecutorProvider_->close(); listenerExecutorProvider_->close(); partitionListenerExecutorProvider_->close(); - pool_.close(); } uint64_t ClientImpl::newProducerId() { diff --git a/pulsar-client-cpp/lib/ConnectionPool.cc b/pulsar-client-cpp/lib/ConnectionPool.cc index 83fbd21ce2cbd..1a79c80b8c557 100644 --- a/pulsar-client-cpp/lib/ConnectionPool.cc +++ b/pulsar-client-cpp/lib/ConnectionPool.cc @@ -42,6 +42,22 @@ ConnectionPool::ConnectionPool(const ClientConfiguration& conf, ExecutorServiceP poolConnections_(poolConnections), mutex_() {} +ConnectionPool::~ConnectionPool() { + std::unique_lock lock(mutex_); + + if (poolConnections_) { + for (auto cnxIt = pool_.begin(); cnxIt != pool_.end(); cnxIt++) { + ClientConnectionPtr cnx = cnxIt->second.lock(); + if (cnx && !cnx->isClosed()) { + cnx->close(); + } + } + pool_.clear(); + } + + lock.unlock(); +} + Future ConnectionPool::getConnectionAsync( const std::string& logicalAddress, const std::string& physicalAddress) { std::unique_lock lock(mutex_); @@ -80,20 +96,4 @@ Future ConnectionPool::getConnectionAsync( return future; } -void ConnectionPool::close() { - std::unique_lock lock(mutex_); - - if (poolConnections_) { - for (auto cnxIt = pool_.begin(); cnxIt != pool_.end(); cnxIt++) { - ClientConnectionPtr cnx = cnxIt->second.lock(); - if (cnx && !cnx->isClosed()) { - cnx->close(); - } - } - pool_.clear(); - } - - lock.unlock(); -} - } // namespace pulsar diff --git a/pulsar-client-cpp/lib/ConnectionPool.h b/pulsar-client-cpp/lib/ConnectionPool.h index 4f54c88f07b42..21011c5bdbf85 100644 --- a/pulsar-client-cpp/lib/ConnectionPool.h +++ b/pulsar-client-cpp/lib/ConnectionPool.h @@ -36,6 +36,8 @@ class PULSAR_PUBLIC ConnectionPool { ConnectionPool(const ClientConfiguration& conf, ExecutorServiceProviderPtr executorProvider, const AuthenticationPtr& authentication, bool poolConnections = true); + ~ConnectionPool(); + /** * Get a connection from the pool. *

@@ -55,8 +57,6 @@ class PULSAR_PUBLIC ConnectionPool { Future getConnectionAsync(const std::string& logicalAddress, const std::string& physicalAddress); - void close(); - private: ClientConfiguration clientConfiguration_; ExecutorServiceProviderPtr executorProvider_; From 4f76aaacb2a247de7e60ac236ca0e673a3360c81 Mon Sep 17 00:00:00 2001 From: Masahiro Sakamoto Date: Wed, 25 Sep 2019 11:21:38 +0900 Subject: [PATCH 3/6] Do not reset executor_ in ClientConnection::close() --- pulsar-client-cpp/lib/ClientConnection.cc | 4 ---- 1 file changed, 4 deletions(-) diff --git a/pulsar-client-cpp/lib/ClientConnection.cc b/pulsar-client-cpp/lib/ClientConnection.cc index 30c4c096f42c0..077f36fdf583d 100644 --- a/pulsar-client-cpp/lib/ClientConnection.cc +++ b/pulsar-client-cpp/lib/ClientConnection.cc @@ -1385,10 +1385,6 @@ void ClientConnection::close() { if (tlsSocket_) { tlsSocket_->lowest_layer().close(); } - - if (executor_) { - executor_.reset(); - } } bool ClientConnection::isClosed() const { return state_ == Disconnected; } From 7208ad013d8cd37ee1b64b13afc303ebf8ce7dfb Mon Sep 17 00:00:00 2001 From: Masahiro Sakamoto Date: Wed, 25 Sep 2019 11:30:37 +0900 Subject: [PATCH 4/6] Make ConnectionPool have shared_ptr instead of weak_ptr to ClientConnection --- pulsar-client-cpp/lib/ConnectionPool.cc | 8 ++++---- pulsar-client-cpp/lib/ConnectionPool.h | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/pulsar-client-cpp/lib/ConnectionPool.cc b/pulsar-client-cpp/lib/ConnectionPool.cc index 1a79c80b8c557..bd9c1399ae420 100644 --- a/pulsar-client-cpp/lib/ConnectionPool.cc +++ b/pulsar-client-cpp/lib/ConnectionPool.cc @@ -47,7 +47,7 @@ ConnectionPool::~ConnectionPool() { if (poolConnections_) { for (auto cnxIt = pool_.begin(); cnxIt != pool_.end(); cnxIt++) { - ClientConnectionPtr cnx = cnxIt->second.lock(); + ClientConnectionPtr cnx = cnxIt->second; if (cnx && !cnx->isClosed()) { cnx->close(); } @@ -65,17 +65,17 @@ Future ConnectionPool::getConnectionAsync( if (poolConnections_) { PoolMap::iterator cnxIt = pool_.find(logicalAddress); if (cnxIt != pool_.end()) { - ClientConnectionPtr cnx = cnxIt->second.lock(); + ClientConnectionPtr cnx = cnxIt->second; if (cnx && !cnx->isClosed()) { // Found a valid or pending connection in the pool LOG_DEBUG("Got connection from pool for " << logicalAddress << " use_count: " // - << (cnx.use_count() - 1) << " @ " << cnx.get()); + << (cnx.use_count() - 2) << " @ " << cnx.get()); return cnx->getConnectFuture(); } else { // Deleting stale connection LOG_INFO("Deleting stale connection from pool for " - << logicalAddress << " use_count: " << (cnx.use_count() - 1) << " @ " << cnx.get()); + << logicalAddress << " use_count: " << (cnx.use_count() - 2) << " @ " << cnx.get()); pool_.erase(logicalAddress); } } diff --git a/pulsar-client-cpp/lib/ConnectionPool.h b/pulsar-client-cpp/lib/ConnectionPool.h index 21011c5bdbf85..cafa11dda9313 100644 --- a/pulsar-client-cpp/lib/ConnectionPool.h +++ b/pulsar-client-cpp/lib/ConnectionPool.h @@ -61,7 +61,7 @@ class PULSAR_PUBLIC ConnectionPool { ClientConfiguration clientConfiguration_; ExecutorServiceProviderPtr executorProvider_; AuthenticationPtr authentication_; - typedef std::map PoolMap; + typedef std::map PoolMap; PoolMap pool_; bool poolConnections_; std::mutex mutex_; From d35bc2ad0b5c2c5413897357e441a15d0fc2571c Mon Sep 17 00:00:00 2001 From: Masahiro Sakamoto Date: Wed, 25 Sep 2019 13:18:01 +0900 Subject: [PATCH 5/6] Revert PoolMap type --- pulsar-client-cpp/lib/ConnectionPool.cc | 8 ++++---- pulsar-client-cpp/lib/ConnectionPool.h | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/pulsar-client-cpp/lib/ConnectionPool.cc b/pulsar-client-cpp/lib/ConnectionPool.cc index bd9c1399ae420..1a79c80b8c557 100644 --- a/pulsar-client-cpp/lib/ConnectionPool.cc +++ b/pulsar-client-cpp/lib/ConnectionPool.cc @@ -47,7 +47,7 @@ ConnectionPool::~ConnectionPool() { if (poolConnections_) { for (auto cnxIt = pool_.begin(); cnxIt != pool_.end(); cnxIt++) { - ClientConnectionPtr cnx = cnxIt->second; + ClientConnectionPtr cnx = cnxIt->second.lock(); if (cnx && !cnx->isClosed()) { cnx->close(); } @@ -65,17 +65,17 @@ Future ConnectionPool::getConnectionAsync( if (poolConnections_) { PoolMap::iterator cnxIt = pool_.find(logicalAddress); if (cnxIt != pool_.end()) { - ClientConnectionPtr cnx = cnxIt->second; + ClientConnectionPtr cnx = cnxIt->second.lock(); if (cnx && !cnx->isClosed()) { // Found a valid or pending connection in the pool LOG_DEBUG("Got connection from pool for " << logicalAddress << " use_count: " // - << (cnx.use_count() - 2) << " @ " << cnx.get()); + << (cnx.use_count() - 1) << " @ " << cnx.get()); return cnx->getConnectFuture(); } else { // Deleting stale connection LOG_INFO("Deleting stale connection from pool for " - << logicalAddress << " use_count: " << (cnx.use_count() - 2) << " @ " << cnx.get()); + << logicalAddress << " use_count: " << (cnx.use_count() - 1) << " @ " << cnx.get()); pool_.erase(logicalAddress); } } diff --git a/pulsar-client-cpp/lib/ConnectionPool.h b/pulsar-client-cpp/lib/ConnectionPool.h index cafa11dda9313..21011c5bdbf85 100644 --- a/pulsar-client-cpp/lib/ConnectionPool.h +++ b/pulsar-client-cpp/lib/ConnectionPool.h @@ -61,7 +61,7 @@ class PULSAR_PUBLIC ConnectionPool { ClientConfiguration clientConfiguration_; ExecutorServiceProviderPtr executorProvider_; AuthenticationPtr authentication_; - typedef std::map PoolMap; + typedef std::map PoolMap; PoolMap pool_; bool poolConnections_; std::mutex mutex_; From 74a836463dbe506b5275ac69537f33f10ddd3a6a Mon Sep 17 00:00:00 2001 From: Masahiro Sakamoto Date: Wed, 25 Sep 2019 14:43:11 +0900 Subject: [PATCH 6/6] Remove lock.unlock() from ~ConnectionPool() --- pulsar-client-cpp/lib/ConnectionPool.cc | 3 --- 1 file changed, 3 deletions(-) diff --git a/pulsar-client-cpp/lib/ConnectionPool.cc b/pulsar-client-cpp/lib/ConnectionPool.cc index 1a79c80b8c557..124f81ff855ca 100644 --- a/pulsar-client-cpp/lib/ConnectionPool.cc +++ b/pulsar-client-cpp/lib/ConnectionPool.cc @@ -44,7 +44,6 @@ ConnectionPool::ConnectionPool(const ClientConfiguration& conf, ExecutorServiceP ConnectionPool::~ConnectionPool() { std::unique_lock lock(mutex_); - if (poolConnections_) { for (auto cnxIt = pool_.begin(); cnxIt != pool_.end(); cnxIt++) { ClientConnectionPtr cnx = cnxIt->second.lock(); @@ -54,8 +53,6 @@ ConnectionPool::~ConnectionPool() { } pool_.clear(); } - - lock.unlock(); } Future ConnectionPool::getConnectionAsync(