Fix the wrong backoff computation when retrying (apache#296)

### Motivation

All the retryable operations share the same `Backoff` object in
`RetryableLookupService`, so if the reconnection happens for some times,
the delay of retrying will keeps the maximum value (30 seconds).

### Modifications

Refactor the design of the `RetryableLookupService`:
- Add a `RetryableOperation` class to represent a retryable operation,
  each instance has its own `Backoff` object. The operation could only
  be executed once.
- Add a `RetryableOperationCache` class to represent a map that maps a
  specific name to its associated operation. It's an optimization that
  if an operation (e.g. find the owner topic of topic A) was not
  complete while the same operation was executed, the future would be
  reused.
- In `RetryableLookupService`, just maintain some caches for different
  operations.
- Add `RetryableOperationCacheTest` to verify the behaviors.

Author: BewareMyPower

lib/ClientImpl.cc

@@ -598,6 +598,7 @@ void ClientImpl::closeAsync(CloseCallback callback) {
     state_ = Closing;
 
     memoryLimitController_.close();
+    lookupServicePtr_->close();
 
     auto producers = producers_.move();
     auto consumers = consumers_.move();

lib/LookupService.h

@@ -86,6 +86,8 @@ class LookupService {
                                                  const std::string& version = "") = 0;
 
     virtual ~LookupService() {}
+
+    virtual void close() {}
 };
 
 typedef std::shared_ptr<LookupService> LookupServicePtr;

lib/RetryableLookupService.h

@@ -18,23 +18,17 @@
  */
 #pragma once
 
-#include <algorithm>
-#include <memory>
-
-#include "Backoff.h"
-#include "ExecutorService.h"
-#include "LogUtils.h"
 #include "LookupDataResult.h"
 #include "LookupService.h"
-#include "SynchronizedHashMap.h"
+#include "NamespaceName.h"
+#include "RetryableOperationCache.h"
 #include "TopicName.h"
 
 namespace pulsar {
 
-class RetryableLookupService : public LookupService,
-                               public std::enable_shared_from_this<RetryableLookupService> {
+class RetryableLookupService : public LookupService {
    private:
-    friend class PulsarFriend;
+    friend class LookupServiceTest;
     struct PassKey {
         explicit PassKey() {}
     };
@@ -44,123 +38,58 @@ class RetryableLookupService : public LookupService,
     explicit RetryableLookupService(PassKey, Args&&... args)
         : RetryableLookupService(std::forward<Args>(args)...) {}
 
+    void close() override {
+        lookupCache_->clear();
+        partitionLookupCache_->clear();
+        namespaceLookupCache_->clear();
+        getSchemaCache_->clear();
+    }
+
     template <typename... Args>
     static std::shared_ptr<RetryableLookupService> create(Args&&... args) {
         return std::make_shared<RetryableLookupService>(PassKey{}, std::forward<Args>(args)...);
     }
 
     LookupResultFuture getBroker(const TopicName& topicName) override {
-        return executeAsync<LookupResult>("get-broker-" + topicName.toString(),
-                                          [this, topicName] { return lookupService_->getBroker(topicName); });
+        return lookupCache_->run("get-broker-" + topicName.toString(),
+                                 [this, topicName] { return lookupService_->getBroker(topicName); });
     }
 
     Future<Result, LookupDataResultPtr> getPartitionMetadataAsync(const TopicNamePtr& topicName) override {
-        return executeAsync<LookupDataResultPtr>(
+        return partitionLookupCache_->run(
             "get-partition-metadata-" + topicName->toString(),
             [this, topicName] { return lookupService_->getPartitionMetadataAsync(topicName); });
     }
 
     Future<Result, NamespaceTopicsPtr> getTopicsOfNamespaceAsync(
         const NamespaceNamePtr& nsName, CommandGetTopicsOfNamespace_Mode mode) override {
-        return executeAsync<NamespaceTopicsPtr>(
+        return namespaceLookupCache_->run(
             "get-topics-of-namespace-" + nsName->toString(),
             [this, nsName, mode] { return lookupService_->getTopicsOfNamespaceAsync(nsName, mode); });
     }
 
     Future<Result, SchemaInfo> getSchema(const TopicNamePtr& topicName, const std::string& version) override {
-        return executeAsync<SchemaInfo>("get-schema" + topicName->toString(), [this, topicName, version] {
+        return getSchemaCache_->run("get-schema" + topicName->toString(), [this, topicName, version] {
             return lookupService_->getSchema(topicName, version);
         });
     }
 
-    template <typename T>
-    Future<Result, T> executeAsync(const std::string& key, std::function<Future<Result, T>()> f) {
-        Promise<Result, T> promise;
-        executeAsyncImpl(key, f, promise, timeout_);
-        return promise.getFuture();
-    }
-
    private:
     const std::shared_ptr<LookupService> lookupService_;
-    const TimeDuration timeout_;
-    Backoff backoff_;
-    const ExecutorServiceProviderPtr executorProvider_;
-
-    SynchronizedHashMap<std::string, DeadlineTimerPtr> backoffTimers_;
+    RetryableOperationCachePtr<LookupResult> lookupCache_;
+    RetryableOperationCachePtr<LookupDataResultPtr> partitionLookupCache_;
+    RetryableOperationCachePtr<NamespaceTopicsPtr> namespaceLookupCache_;
+    RetryableOperationCachePtr<SchemaInfo> getSchemaCache_;
 
     RetryableLookupService(std::shared_ptr<LookupService> lookupService, int timeoutSeconds,
                            ExecutorServiceProviderPtr executorProvider)
         : lookupService_(lookupService),
-          timeout_(boost::posix_time::seconds(timeoutSeconds)),
-          backoff_(boost::posix_time::milliseconds(100), timeout_ + timeout_,
-                   boost::posix_time::milliseconds(0)),
-          executorProvider_(executorProvider) {}
-
-    std::weak_ptr<RetryableLookupService> weak_from_this() noexcept { return shared_from_this(); }
-
-    // NOTE: Set the visibility to fix compilation error in GCC 6
-    template <typename T>
-#ifndef _WIN32
-    __attribute__((visibility("hidden")))
-#endif
-    void
-    executeAsyncImpl(const std::string& key, std::function<Future<Result, T>()> f, Promise<Result, T> promise,
-                     TimeDuration remainingTime) {
-        auto weakSelf = weak_from_this();
-        f().addListener([this, weakSelf, key, f, promise, remainingTime](Result result, const T& value) {
-            auto self = weakSelf.lock();
-            if (!self) {
-                return;
-            }
-
-            if (result == ResultOk) {
-                backoffTimers_.remove(key);
-                promise.setValue(value);
-            } else if (result == ResultRetryable) {
-                if (remainingTime.total_milliseconds() <= 0) {
-                    backoffTimers_.remove(key);
-                    promise.setFailed(ResultTimeout);
-                    return;
-                }
-
-                DeadlineTimerPtr timerPtr;
-                try {
-                    timerPtr = executorProvider_->get()->createDeadlineTimer();
-                } catch (const std::runtime_error& e) {
-                    LOG_ERROR("Failed to retry lookup for " << key << ": " << e.what());
-                    promise.setFailed(ResultConnectError);
-                    return;
-                }
-                auto it = backoffTimers_.emplace(key, timerPtr);
-                auto& timer = *(it.first->second);
-                auto delay = std::min(backoff_.next(), remainingTime);
-                timer.expires_from_now(delay);
-
-                auto nextRemainingTime = remainingTime - delay;
-                LOG_INFO("Reschedule " << key << " for " << delay.total_milliseconds()
-                                       << " ms, remaining time: " << nextRemainingTime.total_milliseconds()
-                                       << " ms");
-                timer.async_wait([this, weakSelf, key, f, promise,
-                                  nextRemainingTime](const boost::system::error_code& ec) {
-                    auto self = weakSelf.lock();
-                    if (!self || ec) {
-                        if (self && ec != boost::asio::error::operation_aborted) {
-                            LOG_ERROR("The timer for " << key << " failed: " << ec.message());
-                        }
-                        // The lookup service has been destructed or the timer has been cancelled
-                        promise.setFailed(ResultTimeout);
-                        return;
-                    }
-                    executeAsyncImpl(key, f, promise, nextRemainingTime);
-                });
-            } else {
-                backoffTimers_.remove(key);
-                promise.setFailed(result);
-            }
-        });
-    }
-
-    DECLARE_LOG_OBJECT()
+          lookupCache_(RetryableOperationCache<LookupResult>::create(executorProvider, timeoutSeconds)),
+          partitionLookupCache_(
+              RetryableOperationCache<LookupDataResultPtr>::create(executorProvider, timeoutSeconds)),
+          namespaceLookupCache_(
+              RetryableOperationCache<NamespaceTopicsPtr>::create(executorProvider, timeoutSeconds)),
+          getSchemaCache_(RetryableOperationCache<SchemaInfo>::create(executorProvider, timeoutSeconds)) {}
 };
 
 }  // namespace pulsar

lib/RetryableOperation.h

@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#pragma once
+
+#include <pulsar/Result.h>
+
+#include <algorithm>
+#include <atomic>
+#include <functional>
+#include <memory>
+
+#include "Backoff.h"
+#include "ExecutorService.h"
+#include "Future.h"
+#include "LogUtils.h"
+
+namespace pulsar {
+
+template <typename T>
+class RetryableOperation : public std::enable_shared_from_this<RetryableOperation<T>> {
+    struct PassKey {
+        explicit PassKey() {}
+    };
+
+    RetryableOperation(const std::string& name, std::function<Future<Result, T>()>&& func, int timeoutSeconds,
+                       DeadlineTimerPtr timer)
+        : name_(name),
+          func_(std::move(func)),
+          timeout_(boost::posix_time::seconds(timeoutSeconds)),
+          backoff_(boost::posix_time::milliseconds(100), timeout_ + timeout_,
+                   boost::posix_time::milliseconds(0)),
+          timer_(timer) {}
+
+   public:
+    template <typename... Args>
+    explicit RetryableOperation(PassKey, Args&&... args) : RetryableOperation(std::forward<Args>(args)...) {}
+
+    template <typename... Args>
+    static std::shared_ptr<RetryableOperation<T>> create(Args&&... args) {
+        return std::make_shared<RetryableOperation<T>>(PassKey{}, std::forward<Args>(args)...);
+    }
+
+    Future<Result, T> run() {
+        bool expected = false;
+        if (!started_.compare_exchange_strong(expected, true)) {
+            return promise_.getFuture();
+        }
+        return runImpl(timeout_);
+    }
+
+    void cancel() {
+        promise_.setFailed(ResultDisconnected);
+        boost::system::error_code ec;
+        timer_->cancel(ec);
+    }
+
+   private:
+    const std::string name_;
+    std::function<Future<Result, T>()> func_;
+    const TimeDuration timeout_;
+    Backoff backoff_;
+    Promise<Result, T> promise_;
+    std::atomic_bool started_{false};
+    DeadlineTimerPtr timer_;
+
+    Future<Result, T> runImpl(TimeDuration remainingTime) {
+        std::weak_ptr<RetryableOperation<T>> weakSelf{this->shared_from_this()};
+        func_().addListener([this, weakSelf, remainingTime](Result result, const T& value) {
+            auto self = weakSelf.lock();
+            if (!self) {
+                return;
+            }
+            if (result == ResultOk) {
+                promise_.setValue(value);
+                return;
+            }
+            if (result != ResultRetryable) {
+                promise_.setFailed(result);
+                return;
+            }
+            if (remainingTime.total_milliseconds() <= 0) {
+                promise_.setFailed(ResultTimeout);
+                return;
+            }
+
+            auto delay = std::min(backoff_.next(), remainingTime);
+            timer_->expires_from_now(delay);
+
+            auto nextRemainingTime = remainingTime - delay;
+            LOG_INFO("Reschedule " << name_ << " for " << delay.total_milliseconds()
+                                   << " ms, remaining time: " << nextRemainingTime.total_milliseconds()
+                                   << " ms");
+            timer_->async_wait([this, weakSelf, nextRemainingTime](const boost::system::error_code& ec) {
+                auto self = weakSelf.lock();
+                if (!self) {
+                    return;
+                }
+                if (ec) {
+                    if (ec == boost::asio::error::operation_aborted) {
+                        LOG_DEBUG("Timer for " << name_ << " is cancelled");
+                        promise_.setFailed(ResultTimeout);
+                    } else {
+                        LOG_WARN("Timer for " << name_ << " failed: " << ec.message());
+                    }
+                } else {
+                    LOG_DEBUG("Run operation " << name_ << ", remaining time: "
+                                               << nextRemainingTime.total_milliseconds() << " ms");
+                    runImpl(nextRemainingTime);
+                }
+            });
+        });
+        return promise_.getFuture();
+    }
+
+    DECLARE_LOG_OBJECT()
+};
+
+}  // namespace pulsar