Avoid calling cancel() in destructors

apache · Mar 20, 2023 · d5b000a · d5b000a
1 parent 8242c4b
commit d5b000a
Show file tree

Hide file tree

Showing 8 changed files with 18 additions and 2 deletions.
diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc
@@ -1251,6 +1251,7 @@ void ConsumerImpl::shutdown() {
         client->cleanupConsumer(this);
     }
     negativeAcksTracker_.close();
+    consumerStatsBasePtr_->close();
     cancelTimers();
     consumerCreatedPromise_.setFailed(ResultAlreadyClosed);
     failPendingReceiveCallback();

diff --git a/lib/ProducerImpl.cc b/lib/ProducerImpl.cc
@@ -946,6 +946,7 @@ void ProducerImpl::shutdown() {
     if (client) {
         client->cleanupProducer(this);
     }
+    producerStatsBasePtr_->close();
     cancelTimers();
     producerCreatedPromise_.setFailed(ResultAlreadyClosed);
     state_ = Closed;

diff --git a/lib/stats/ConsumerStatsBase.h b/lib/stats/ConsumerStatsBase.h
@@ -28,6 +28,7 @@ namespace pulsar {
 class ConsumerStatsBase {
    public:
     virtual void start() {}
+    virtual void close() {}
     virtual void receivedMessage(Message&, Result) = 0;
     virtual void messageAcknowledged(Result, CommandAck_AckType, uint32_t ackNums = 1) = 0;
     virtual ~ConsumerStatsBase() {}

diff --git a/lib/stats/ConsumerStatsImpl.cc b/lib/stats/ConsumerStatsImpl.cc
@@ -63,10 +63,15 @@ void ConsumerStatsImpl::flushAndReset(const boost::system::error_code& ec) {
     LOG_INFO(tmp);
 }
 
-ConsumerStatsImpl::~ConsumerStatsImpl() { timer_->cancel(); }
+ConsumerStatsImpl::~ConsumerStatsImpl() {}
 
 void ConsumerStatsImpl::start() { scheduleTimer(); }
 
+void ConsumerStatsImpl::close() {
+    boost::system::error_code ec;
+    timer_->cancel(ec);
+}
+
 void ConsumerStatsImpl::receivedMessage(Message& msg, Result res) {
     Lock lock(mutex_);
     if (res == ResultOk) {

diff --git a/lib/stats/ConsumerStatsImpl.h b/lib/stats/ConsumerStatsImpl.h
@@ -60,6 +60,7 @@ class ConsumerStatsImpl : public std::enable_shared_from_this<ConsumerStatsImpl>
     ConsumerStatsImpl(const ConsumerStatsImpl& stats);
     void flushAndReset(const boost::system::error_code&);
     void start() override;
+    void close() override;
     void receivedMessage(Message&, Result) override;
     void messageAcknowledged(Result, CommandAck_AckType, uint32_t ackNums) override;
     virtual ~ConsumerStatsImpl();

diff --git a/lib/stats/ProducerStatsBase.h b/lib/stats/ProducerStatsBase.h
@@ -28,6 +28,7 @@ namespace pulsar {
 class ProducerStatsBase {
    public:
     virtual void start() {}
+    virtual void close() {}
     virtual void messageSent(const Message& msg) = 0;
     virtual void messageReceived(Result, const boost::posix_time::ptime&) = 0;
     virtual ~ProducerStatsBase(){};

diff --git a/lib/stats/ProducerStatsImpl.cc b/lib/stats/ProducerStatsImpl.cc
@@ -102,7 +102,12 @@ void ProducerStatsImpl::messageReceived(Result res, const boost::posix_time::pti
     totalSendMap_[res] += 1;  // Value will automatically be initialized to 0 in the constructor
 }
 
-ProducerStatsImpl::~ProducerStatsImpl() { timer_->cancel(); }
+ProducerStatsImpl::~ProducerStatsImpl() {}
+
+void ProducerStatsImpl::close() {
+    boost::system::error_code ec;
+    timer_->cancel(ec);
+}
 
 void ProducerStatsImpl::scheduleTimer() {
     timer_->expires_from_now(boost::posix_time::seconds(statsIntervalInSeconds_));

diff --git a/lib/stats/ProducerStatsImpl.h b/lib/stats/ProducerStatsImpl.h
@@ -82,6 +82,7 @@ class ProducerStatsImpl : public std::enable_shared_from_this<ProducerStatsImpl>
     ProducerStatsImpl(const ProducerStatsImpl& stats);
 
     void start() override;
+    void close() override;
 
     void flushAndReset(const boost::system::error_code&);