fix timer cancel invocations

Author: BewareMyPower

lib/AckGroupingTrackerEnabled.cc

@@ -121,8 +121,7 @@ AckGroupingTrackerEnabled::~AckGroupingTrackerEnabled() {
     this->flush();
     std::lock_guard<std::mutex> lock(this->mutexTimer_);
     if (this->timer_) {
-        ASIO_ERROR ec;
-        this->timer_->cancel(ec);
+        cancelTimer(*this->timer_);
     }
 }
 
@@ -172,7 +171,7 @@ void AckGroupingTrackerEnabled::scheduleTimer() {
 
     std::lock_guard<std::mutex> lock(this->mutexTimer_);
     this->timer_ = this->executor_->createDeadlineTimer();
-    this->timer_->expires_from_now(std::chrono::milliseconds(std::max(1L, this->ackGroupingTimeMs_)));
+    this->timer_->expires_after(std::chrono::milliseconds(std::max(1L, this->ackGroupingTimeMs_)));
     std::weak_ptr<AckGroupingTracker> weakSelf = shared_from_this();
     this->timer_->async_wait([this, weakSelf](const ASIO_ERROR& ec) -> void {
         auto self = weakSelf.lock();

lib/AsioTimer.h

@@ -29,3 +29,12 @@
 #include "AsioDefines.h"
 
 using DeadlineTimerPtr = std::shared_ptr<ASIO::steady_timer>;
+
+inline void cancelTimer(ASIO::steady_timer& timer) {
+    try {
+        timer.cancel();
+    } catch (const ASIO_SYSTEM_ERROR& ignored) {
+        // Most of the time the exception can be ignored unless the following logic depends on the fact that
+        // the timer is cancelled.
+    }
+}

lib/ClientConnection.cc

@@ -317,7 +317,7 @@ void ClientConnection::handlePulsarConnected(const proto::CommandConnected& cmdC
         // Only send keep-alive probes if the broker supports it
         keepAliveTimer_ = executor_->createDeadlineTimer();
         if (keepAliveTimer_) {
-            keepAliveTimer_->expires_from_now(std::chrono::seconds(keepAliveIntervalInSeconds_));
+            keepAliveTimer_->expires_after(std::chrono::seconds(keepAliveIntervalInSeconds_));
             auto weakSelf = weak_from_this();
             keepAliveTimer_->async_wait([weakSelf](const ASIO_ERROR&) {
                 auto self = weakSelf.lock();
@@ -362,7 +362,7 @@ void ClientConnection::startConsumerStatsTimer(std::vector<uint64_t> consumerSta
     // If the close operation has reset the consumerStatsRequestTimer_ then the use_count will be zero
     // Check if we have a timer still before we set the request timer to pop again.
     if (consumerStatsRequestTimer_) {
-        consumerStatsRequestTimer_->expires_from_now(operationsTimeout_);
+        consumerStatsRequestTimer_->expires_after(operationsTimeout_);
         auto weakSelf = weak_from_this();
         consumerStatsRequestTimer_->async_wait([weakSelf, consumerStatsRequests](const ASIO_ERROR& err) {
             auto self = weakSelf.lock();
@@ -621,32 +621,12 @@ void ClientConnection::handleResolve(ASIO_ERROR err, const tcp::resolver::result
         ptr->connectTimeoutTask_->stop();
     });
 
-    ASIO::async_connect(socket_, results, [weakSelf](const ASIO_ERROR& err, const tcp::endpoint& endpoint) {
+    ASIO::async_connect(*socket_, results, [weakSelf](const ASIO_ERROR& err, const tcp::endpoint& endpoint) {
         auto self = weakSelf.lock();
         if (self) {
             self->handleTcpConnected(err, endpoint);
         }
     });
-
-    // TODO: use the new resolver results API to iterate over endpoints
-    /*
-    LOG_DEBUG(cnxString_ << "Connecting to " << endpointIterator->endpoint() << "...");
-    connectTimeoutTask_->start();
-    if (endpointIterator != tcp::resolver::iterator()) {
-        LOG_DEBUG(cnxString_ << "Resolved hostname " << endpointIterator->host_name()  //
-                             << " to " << endpointIterator->endpoint());
-        socket_->async_connect(*endpointIterator, [weakSelf, endpointIterator](const ASIO_ERROR& err) {
-            auto self = weakSelf.lock();
-            if (self) {
-                self->handleTcpConnected(err, endpointIterator);
-            }
-        });
-    } else {
-        LOG_WARN(cnxString_ << "No IP address found");
-        close();
-        return;
-    }
-    */
 }
 
 void ClientConnection::readNextCommand() {
@@ -1050,7 +1030,7 @@ void ClientConnection::newLookup(const SharedBuffer& cmd, uint64_t requestId,
     LookupRequestData requestData;
     requestData.promise = promise;
     requestData.timer = executor_->createDeadlineTimer();
-    requestData.timer->expires_from_now(operationsTimeout_);
+    requestData.timer->expires_after(operationsTimeout_);
     auto weakSelf = weak_from_this();
     requestData.timer->async_wait([weakSelf, requestData](const ASIO_ERROR& ec) {
         auto self = weakSelf.lock();
@@ -1190,7 +1170,7 @@ Future<Result, ResponseData> ClientConnection::sendRequestWithId(const SharedBuf
 
     PendingRequestData requestData;
     requestData.timer = executor_->createDeadlineTimer();
-    requestData.timer->expires_from_now(operationsTimeout_);
+    requestData.timer->expires_after(operationsTimeout_);
     auto weakSelf = weak_from_this();
     requestData.timer->async_wait([weakSelf, requestData](const ASIO_ERROR& ec) {
         auto self = weakSelf.lock();
@@ -1245,7 +1225,7 @@ void ClientConnection::handleKeepAliveTimeout() {
         // be zero And we do not attempt to dereference the pointer.
         Lock lock(mutex_);
         if (keepAliveTimer_) {
-            keepAliveTimer_->expires_from_now(std::chrono::seconds(keepAliveIntervalInSeconds_));
+            keepAliveTimer_->expires_after(std::chrono::seconds(keepAliveIntervalInSeconds_));
             auto weakSelf = weak_from_this();
             keepAliveTimer_->async_wait([weakSelf](const ASIO_ERROR&) {
                 auto self = weakSelf.lock();
@@ -1307,12 +1287,12 @@ void ClientConnection::close(Result result, bool detach) {
     numOfPendingLookupRequest_ = 0;
 
     if (keepAliveTimer_) {
-        keepAliveTimer_->cancel();
+        cancelTimer(*keepAliveTimer_);
         keepAliveTimer_.reset();
     }
 
     if (consumerStatsRequestTimer_) {
-        consumerStatsRequestTimer_->cancel();
+        cancelTimer(*consumerStatsRequestTimer_);
         consumerStatsRequestTimer_.reset();
     }
 
@@ -1424,7 +1404,7 @@ Future<Result, GetLastMessageIdResponse> ClientConnection::newGetLastMessageId(u
     LastMessageIdRequestData requestData;
     requestData.promise = promise;
     requestData.timer = executor_->createDeadlineTimer();
-    requestData.timer->expires_from_now(operationsTimeout_);
+    requestData.timer->expires_after(operationsTimeout_);
     auto weakSelf = weak_from_this();
     requestData.timer->async_wait([weakSelf, requestData](const ASIO_ERROR& ec) {
         auto self = weakSelf.lock();
@@ -1472,7 +1452,7 @@ Future<Result, SchemaInfo> ClientConnection::newGetSchema(const std::string& top
     lock.unlock();
 
     auto weakSelf = weak_from_this();
-    timer->expires_from_now(operationsTimeout_);
+    timer->expires_after(operationsTimeout_);
     timer->async_wait([this, weakSelf, requestId](const ASIO_ERROR& ec) {
         auto self = weakSelf.lock();
         if (!self) {
@@ -1559,7 +1539,7 @@ void ClientConnection::handleSuccess(const proto::CommandSuccess& success) {
         lock.unlock();
 
         requestData.promise.setValue({});
-        requestData.timer->cancel();
+        cancelTimer(*requestData.timer);
     }
 }
 
@@ -1571,7 +1551,8 @@ void ClientConnection::handlePartitionedMetadataResponse(
     Lock lock(mutex_);
     auto it = pendingLookupRequests_.find(partitionMetadataResponse.request_id());
     if (it != pendingLookupRequests_.end()) {
-        it->second.timer->cancel();
+        cancelTimer(*it->second.timer);
+
         LookupDataResultPromisePtr lookupDataPromise = it->second.promise;
         pendingLookupRequests_.erase(it);
         numOfPendingLookupRequest_--;
@@ -1650,7 +1631,7 @@ void ClientConnection::handleLookupTopicRespose(
     Lock lock(mutex_);
     auto it = pendingLookupRequests_.find(lookupTopicResponse.request_id());
     if (it != pendingLookupRequests_.end()) {
-        it->second.timer->cancel();
+        cancelTimer(*it->second.timer);
         LookupDataResultPromisePtr lookupDataPromise = it->second.promise;
         pendingLookupRequests_.erase(it);
         numOfPendingLookupRequest_--;
@@ -1728,7 +1709,7 @@ void ClientConnection::handleProducerSuccess(const proto::CommandProducerSuccess
                 data.topicEpoch = boost::none;
             }
             requestData.promise.setValue(data);
-            requestData.timer->cancel();
+            cancelTimer(*requestData.timer);
         }
     }
 }
@@ -1748,7 +1729,7 @@ void ClientConnection::handleError(const proto::CommandError& error) {
         lock.unlock();
 
         requestData.promise.setFailed(result);
-        requestData.timer->cancel();
+        cancelTimer(*requestData.timer);
     } else {
         PendingGetLastMessageIdRequestsMap::iterator it =
             pendingGetLastMessageIdRequests_.find(error.request_id());
@@ -2041,8 +2022,8 @@ void ClientConnection::unsafeRemovePendingRequest(long requestId) {
     auto it = pendingRequests_.find(requestId);
     if (it != pendingRequests_.end()) {
         it->second.promise.setFailed(ResultDisconnected);
-        ASIO_ERROR ec;
-        it->second.timer->cancel(ec);
+        cancelTimer(*it->second.timer);
+
         pendingRequests_.erase(it);
     }
 }

lib/ConsumerImpl.cc

@@ -423,7 +423,7 @@ void ConsumerImpl::discardChunkMessages(const std::string& uuid, const MessageId
 }
 
 void ConsumerImpl::triggerCheckExpiredChunkedTimer() {
-    checkExpiredChunkedTimer_->expires_from_now(milliseconds(expireTimeOfIncompleteChunkedMessageMs_));
+    checkExpiredChunkedTimer_->expires_after(milliseconds(expireTimeOfIncompleteChunkedMessageMs_));
     std::weak_ptr<ConsumerImplBase> weakSelf{shared_from_this()};
     checkExpiredChunkedTimer_->async_wait([this, weakSelf](const ASIO_ERROR& ec) -> void {
         auto self = weakSelf.lock();
@@ -1690,7 +1690,7 @@ void ConsumerImpl::internalGetLastMessageIdAsync(const BackoffPtr& backoff, Time
         }
         remainTime -= next;
 
-        timer->expires_from_now(next);
+        timer->expires_after(next);
 
         auto self = shared_from_this();
         timer->async_wait([this, backoff, remainTime, timer, next, callback,
@@ -1814,9 +1814,8 @@ std::shared_ptr<ConsumerImpl> ConsumerImpl::get_shared_this_ptr() {
 }
 
 void ConsumerImpl::cancelTimers() noexcept {
-    ASIO_ERROR ec;
-    batchReceiveTimer_->cancel(ec);
-    checkExpiredChunkedTimer_->cancel(ec);
+    cancelTimer(*batchReceiveTimer_);
+    cancelTimer(*checkExpiredChunkedTimer_);
     unAckedMessageTrackerPtr_->stop();
     consumerStatsBasePtr_->stop();
 }

lib/ConsumerImplBase.cc

@@ -49,7 +49,7 @@ ConsumerImplBase::ConsumerImplBase(const ClientImplPtr& client, const std::strin
 
 void ConsumerImplBase::triggerBatchReceiveTimerTask(long timeoutMs) {
     if (timeoutMs > 0) {
-        batchReceiveTimer_->expires_from_now(std::chrono::milliseconds(timeoutMs));
+        batchReceiveTimer_->expires_after(std::chrono::milliseconds(timeoutMs));
         std::weak_ptr<ConsumerImplBase> weakSelf{shared_from_this()};
         batchReceiveTimer_->async_wait([weakSelf](const ASIO_ERROR& ec) {
             auto self = weakSelf.lock();

lib/ExecutorService.cc

@@ -38,8 +38,8 @@ void ExecutorService::start() {
             try {
                 io_context_.run();
                 LOG_DEBUG("Event loop of ExecutorService exits successfully");
-            } catch (const ASIO_ERROR &e) {
-                LOG_ERROR("Failed to run io_context: " << e.message());
+            } catch (const ASIO_SYSTEM_ERROR &e) {
+                LOG_ERROR("Failed to run io_context: " << e.what());
             }
         }
         {

lib/HandlerBase.cc

@@ -50,9 +50,8 @@ HandlerBase::HandlerBase(const ClientImplPtr& client, const std::string& topic,
       redirectedClusterURI_("") {}
 
 HandlerBase::~HandlerBase() {
-    ASIO_ERROR ignored;
-    timer_->cancel(ignored);
-    creationTimer_->cancel(ignored);
+    cancelTimer(*timer_);
+    cancelTimer(*creationTimer_);
 }
 
 void HandlerBase::start() {
@@ -61,15 +60,14 @@ void HandlerBase::start() {
     if (state_.compare_exchange_strong(state, Pending)) {
         grabCnx();
     }
-    creationTimer_->expires_from_now(operationTimeut_);
+    creationTimer_->expires_after(operationTimeut_);
     std::weak_ptr<HandlerBase> weakSelf{shared_from_this()};
     creationTimer_->async_wait([this, weakSelf](const ASIO_ERROR& error) {
         auto self = weakSelf.lock();
         if (self && !error) {
             LOG_WARN("Cancel the pending reconnection due to the start timeout");
             connectionFailed(ResultTimeout);
-            ASIO_ERROR ignored;
-            timer_->cancel(ignored);
+            cancelTimer(*timer_);
         }
     });
 }
@@ -133,8 +131,7 @@ void HandlerBase::grabCnx(const boost::optional<std::string>& assignedBrokerUrl)
                     connectionTimeMs_ =
                         duration_cast<milliseconds>(high_resolution_clock::now() - before).count();
                     // Prevent the creationTimer_ from cancelling the timer_ in future
-                    ASIO_ERROR ignored;
-                    creationTimer_->cancel(ignored);
+                    cancelTimer(*creationTimer_);
                     LOG_INFO("Finished connecting to broker after " << connectionTimeMs_ << " ms")
                 } else if (isResultRetryable(result)) {
                     scheduleReconnection();
@@ -188,7 +185,7 @@ void HandlerBase::scheduleReconnection(const boost::optional<std::string>& assig
         TimeDuration delay = assignedBrokerUrl ? std::chrono::milliseconds(0) : backoff_.next();
 
         LOG_INFO(getName() << "Schedule reconnection in " << (toMillis(delay) / 1000.0) << " s");
-        timer_->expires_from_now(delay);
+        timer_->expires_after(delay);
         // passing shared_ptr here since time_ will get destroyed, so tasks will be cancelled
         // so we will not run into the case where grabCnx is invoked on out of scope handler
         auto name = getName();

lib/MultiTopicsConsumerImpl.cc

@@ -507,7 +507,7 @@ void MultiTopicsConsumerImpl::closeAsync(const ResultCallback& originalCallback)
     failPendingBatchReceiveCallback();
 
     // cancel timer
-    batchReceiveTimer_->cancel();
+    cancelTimer(*batchReceiveTimer_);
 }
 
 void MultiTopicsConsumerImpl::messageReceived(const Consumer& consumer, const Message& msg) {
@@ -973,7 +973,7 @@ uint64_t MultiTopicsConsumerImpl::getNumberOfConnectedConsumer() {
     return numberOfConnectedConsumer;
 }
 void MultiTopicsConsumerImpl::runPartitionUpdateTask() {
-    partitionsUpdateTimer_->expires_from_now(partitionsUpdateInterval_);
+    partitionsUpdateTimer_->expires_after(partitionsUpdateInterval_);
     auto weakSelf = weak_from_this();
     partitionsUpdateTimer_->async_wait([weakSelf](const ASIO_ERROR& ec) {
         // If two requests call runPartitionUpdateTask at the same time, the timer will fail, and it
@@ -1126,8 +1126,7 @@ void MultiTopicsConsumerImpl::beforeConnectionChange(ClientConnection& cnx) {
 
 void MultiTopicsConsumerImpl::cancelTimers() noexcept {
     if (partitionsUpdateTimer_) {
-        ASIO_ERROR ec;
-        partitionsUpdateTimer_->cancel(ec);
+        cancelTimer(*partitionsUpdateTimer_);
     }
 }
 

lib/NegativeAcksTracker.cc

@@ -56,7 +56,7 @@ void NegativeAcksTracker::scheduleTimer() {
         return;
     }
     std::weak_ptr<NegativeAcksTracker> weakSelf{shared_from_this()};
-    timer_->expires_from_now(timerInterval_);
+    timer_->expires_after(timerInterval_);
     timer_->async_wait([weakSelf](const ASIO_ERROR &ec) {
         if (auto self = weakSelf.lock()) {
             self->handleTimer(ec);
@@ -135,8 +135,7 @@ void NegativeAcksTracker::add(const MessageId &m) {
 
 void NegativeAcksTracker::close() {
     closed_ = true;
-    ASIO_ERROR ec;
-    timer_->cancel(ec);
+    cancelTimer(*timer_);
     std::lock_guard<std::mutex> lock(mutex_);
     nackedMessages_.clear();
 }

lib/PartitionedProducerImpl.cc

@@ -421,7 +421,7 @@ void PartitionedProducerImpl::flushAsync(FlushCallback callback) {
 
 void PartitionedProducerImpl::runPartitionUpdateTask() {
     auto weakSelf = weak_from_this();
-    partitionsUpdateTimer_->expires_from_now(partitionsUpdateInterval_);
+    partitionsUpdateTimer_->expires_after(partitionsUpdateInterval_);
     partitionsUpdateTimer_->async_wait([weakSelf](const ASIO_ERROR& ec) {
         auto self = weakSelf.lock();
         if (self) {
@@ -524,8 +524,7 @@ uint64_t PartitionedProducerImpl::getNumberOfConnectedProducer() {
 
 void PartitionedProducerImpl::cancelTimers() noexcept {
     if (partitionsUpdateTimer_) {
-        ASIO_ERROR ec;
-        partitionsUpdateTimer_->cancel(ec);
+        cancelTimer(*partitionsUpdateTimer_);
     }
 }