Drain admitted TKO probes before Proxy EventBase teardown

Summary:
During `CarbonRouterInstance` shutdown, a TKO health probe could still be armed on a Proxy's EventBase after the proxy `VirtualEventBase`s were dropped. The probe re-entered `VirtualEventBase::destroyImpl()` on an already-completed keepalive and aborted with `std::future_error: Promise already satisfied`.

Close probe admission before tearing the proxies down:

- `shutdownImpl()` queues a terminal `disableProbes()` on each Proxy EventBase, after the auxiliary threads join and before the proxy `VirtualEventBase`s are dropped.
- `ProxyDestinationBase` checks the gate before arming the probe timer and again in the timer callback, so nothing is admitted once the gate is set.
- Probes admitted before the gate retain the existing `VirtualEventBase` keepalive and drain before Proxy destruction; the steady-state probe path is unchanged.

Reviewed By: ghostonhuang

Differential Revision: D115378060

fbshipit-source-id: 39d93973192e6396e60e82f559924017790db8a9

Author: Joshua Pollack

mcrouter/CarbonRouterInstance-inl.h

@@ -406,6 +406,12 @@ CarbonRouterInstance<RouterInfo>::CarbonRouterInstance(
 template <class RouterInfo>
 void CarbonRouterInstance<RouterInfo>::shutdownImpl() noexcept {
   joinAuxiliaryThreads();
+  for (size_t i = 0; i < proxies_.size(); ++i) {
+    proxyEvbs_[i]->runInEventBaseThread(
+        [destinationMap = proxies_[i]->destinationMap()] {
+          destinationMap->disableProbes();
+        });
+  }
   proxyEvbs_.clear();
   resetMetadata();
   resetAxonProxyClientFactory();

mcrouter/ProxyDestinationBase.cpp

@@ -224,11 +224,19 @@ void ProxyDestinationBase::scheduleNextProbe() {
 }
 
 void ProxyDestinationBase::startSendingProbes() {
+  if (proxy().destinationMap()->probesDisabled()) {
+    return;
+  }
+
   // Reset the RequestContext before scheduling repeated tasks.
   folly::RequestContextScopeGuard guard{nullptr};
   probeDelayNextMs = proxy().router().opts().probe_delay_initial_ms;
   probeTimer_ =
       folly::AsyncTimeout::make(proxy().eventBase(), [this]() noexcept {
+        if (proxy().destinationMap()->probesDisabled()) {
+          return;
+        }
+
         // Note that the previous probe might still be in flight
         if (!probeInflight_) {
           probeInflight_ = true;

mcrouter/ProxyDestinationMap.h

@@ -103,6 +103,14 @@ class ProxyDestinationMap {
    */
   void setResetTimer(std::chrono::milliseconds interval);
 
+  void disableProbes() {
+    probesDisabled_ = true;
+  }
+
+  bool probesDisabled() const {
+    return probesDisabled_;
+  }
+
   /**
    * Calls f(const ProxyDestination&) for each destination stored
    * in ProxyDestinationMap. The whole map is locked during the call.
@@ -145,6 +153,7 @@ class ProxyDestinationMap {
 
   uint32_t inactivityTimeout_;
   std::unique_ptr<folly::AsyncTimeout> resetTimer_;
+  bool probesDisabled_{false};
 
   /**
    * Schedules timeout for resetting inactive connections.

mcrouter/test/cpp_unit_tests/McrouterClientUsage.cpp

@@ -10,16 +10,20 @@
 #include <utility>
 #include <vector>
 
+#include <fmt/core.h>
+#include <glog/logging.h>
 #include <gtest/gtest.h>
 
 #include <folly/fibers/Baton.h>
 #include <folly/io/async/EventBase.h>
+#include <thrift/lib/cpp2/util/ScopedServerInterfaceThread.h>
 
 #include "mcrouter/CarbonRouterClient.h"
 #include "mcrouter/CarbonRouterInstance.h"
 #include "mcrouter/ExecutorObserver.h"
 #include "mcrouter/McReqUtil.h"
 #include "mcrouter/Proxy.h"
+#include "mcrouter/ProxyDestinationMap.h"
 #include "mcrouter/config.h"
 #include "mcrouter/lib/carbon/RequestReplyUtil.h"
 #include "mcrouter/lib/carbon/example/gen/HelloGoodbyeRouterInfo.h"
@@ -28,13 +32,16 @@
 #include "mcrouter/lib/network/AsyncMcServerWorker.h"
 #include "mcrouter/lib/network/McServerRequestContext.h"
 #include "mcrouter/lib/network/gen/MemcacheRouterInfo.h"
+#include "mcrouter/lib/network/gen/gen-cpp2/Memcache.h"
 
 #include "mcrouter/stats.h"
 
 using facebook::memcache::McGetReply;
 using facebook::memcache::McGetRequest;
 using facebook::memcache::McStatsReply;
 using facebook::memcache::McStatsRequest;
+using facebook::memcache::McVersionReply;
+using facebook::memcache::McVersionRequest;
 using facebook::memcache::MemcacheRouterInfo;
 using facebook::memcache::mcrouter::CarbonRouterClient;
 using facebook::memcache::mcrouter::CarbonRouterInstance;
@@ -756,3 +763,59 @@ TEST(CarbonRouterClient, requestExpiryTestWithLatencyInjectionRoute) {
   server.shutdown();
   EXPECT_TRUE(replyReceived);
 }
+
+class TkoProbeHandler final : public apache::thrift::ServiceHandler<
+                                  facebook::memcache::thrift::Memcache> {
+ public:
+  void async_eb_mcGet(
+      apache::thrift::HandlerCallbackPtr<McGetReply> callback,
+      const McGetRequest&) override final {
+    callback->result(McGetReply(carbon::Result::TIMEOUT));
+  }
+
+  void async_eb_mcVersion(
+      apache::thrift::HandlerCallbackPtr<McVersionReply> callback,
+      const McVersionRequest&) override final {
+    versionRequested_.post();
+    callback->result(McVersionReply(carbon::Result::OK));
+  }
+
+  folly::fibers::Baton& versionRequested() {
+    return versionRequested_;
+  }
+
+ private:
+  folly::fibers::Baton versionRequested_;
+};
+
+TEST(CarbonRouterInstance, disableProbesPreventsScheduledTkoProbe) {
+  auto handler = std::make_shared<TkoProbeHandler>();
+  apache::thrift::ScopedServerInterfaceThread server(handler, "127.0.0.1", 0);
+
+  auto opts = defaultTestOptions();
+  opts.failures_until_tko = 1;
+  opts.probe_delay_initial_ms = 1;
+  opts.config_str = fmt::format(
+      R"({{"route":{{"type":"PoolRoute","pool":{{"name":"A","servers":["127.0.0.1:{}"],"protocol":"thrift"}}}}}})",
+      server.getPort());
+  auto router = CarbonRouterInstance<MemcacheRouterInfo>::init(
+      "disableProbesPreventsScheduledTkoProbe", opts);
+
+  auto& proxy = *CHECK_NOTNULL(router->getProxy(0));
+  auto client = router->createClient(0, false);
+  auto& clientRef = *CHECK_NOTNULL(client.get());
+  folly::fibers::Baton replyReceived;
+  const McGetRequest request("key");
+  clientRef.send(
+      request,
+      [&proxy, &replyReceived](const McGetRequest&, McGetReply&& reply) {
+        EXPECT_EQ(carbon::Result::TIMEOUT, *reply.result());
+        proxy.destinationMap()->disableProbes();
+        replyReceived.post();
+      });
+  replyReceived.wait();
+
+  EXPECT_FALSE(
+      handler->versionRequested().try_wait_for(std::chrono::seconds(1)));
+  router->shutdown();
+}