Stop thread-jumps before shutdown frees the proxies

Summary:
`CarbonRouterInstance::shutdownImpl()` calls `proxyEvbs_.clear()`, which
destroys the `VirtualEventBase`s one at a time and frees each `Proxy` along
with them (`Proxy::createProxy` ties the two together via `runOnDestruction`).
The other proxy IO threads are still routing at that point -- the join comes
later -- and `proxies_` is a vector of raw `Proxy*` that is never cleared. So a
request that thread-jumps on a live proxy can resolve a target that `clear()`
has already freed.

Confirmed on two coredumps (aarch64 and x86-64) from the `fbpkg` /
`fbpkg.fetch` CLIs, which stand up and tear down a router on every invocation:

  mov (%rcx,%rdx,8),%rax   ; rax = proxies_[idx]      -- ok
  mov 0x18(%rax),%rbx      ; rbx = Proxy->eventBase_  -- ok, read 0
  ...
  mov 0x18(%rbx),%rcx      ; SIGSEGV at 0x18, rbx == 0

`proxies_[i]` was still readable, but `eventBase_` -- a reference member that
cannot be null in a live `Proxy` -- read 0. The object had been freed and its
memory recycled. The x86 faulting `rip` matches the stack reported in S688594.

Two halves, because there are two populations of dangerous hops:

- New hops. `proxiesDraining_` is set at the start of proxy teardown (after
  `joinAuxiliaryThreads()`, before anything is destroyed) and checked by the
  two accessors that resolve a hop target, `getProxyForThreadJump()` and
  `getProxyBaseForThreadJump()`. While draining they return nullptr and the
  caller routes locally -- the existing "stay on this thread" path, which every
  hop site already handles.

- In-flight hops. `fenceProxyEventBases()` round-trips every proxy event base
  before any `VirtualEventBase` is destroyed. No hop site suspends between
  resolving a target and enqueueing on it, so a hop that read the flag as false
  is still running inline on its source proxy's thread, and the fence task
  queued there cannot run until it finishes. Once every proxy has passed the
  fence, each pending hop is covered by a keep-alive on its target, and
  `~VirtualEventBase()` blocks on those.

Relaxed ordering is sufficient: the happens-before comes from the fence's
notification queue, not from the flag.

Cost is one relaxed load of a read-mostly flag per hop resolution, behind a
consistent-hash lookup (SRRoute) or a full route-handle traverse
(Multi\*Routes). `getProxyFromHash()` is replaced by `getProxyForThreadJump()`;
it had one in-tree caller, but was public on an OSS-mirrored header, so this is
an API break for the GitHub mirror.

Reviewed By: vrishal, stuclar

Differential Revision: D115935603

fbshipit-source-id: 3927b8d85512262cf222b106e6aa3d8230cee68f

Author: Lenar Fatikhov

mcrouter/CarbonRouterInstance-inl.h

@@ -403,6 +403,17 @@ CarbonRouterInstance<RouterInfo>::CarbonRouterInstance(
     McrouterOptions inputOptions)
     : CarbonRouterInstanceBase(std::move(inputOptions)) {}
 
+template <class RouterInfo>
+void CarbonRouterInstance<RouterInfo>::fenceProxyEventBases() noexcept {
+  // Serial: proxyEvbs_.clear() pays the same N round-trips and dominates.
+  for (const auto& proxyEvb : proxyEvbs_) {
+    CHECK(!proxyEvb->getEventBase().inRunningEventBaseThread())
+        << "CarbonRouterInstance shutdown must not run on a proxy event base "
+           "thread";
+    proxyEvb->getEventBase().runInEventBaseThreadAndWait([] {});
+  }
+}
+
 template <class RouterInfo>
 void CarbonRouterInstance<RouterInfo>::shutdownImpl() noexcept {
   joinAuxiliaryThreads();
@@ -412,6 +423,9 @@ void CarbonRouterInstance<RouterInfo>::shutdownImpl() noexcept {
           destinationMap->disableProbes();
         });
   }
+  // Both must precede clear(), which frees proxies other threads still use.
+  setProxiesDraining();
+  fenceProxyEventBases();
   proxyEvbs_.clear();
   resetMetadata();
   resetAxonProxyClientFactory();

mcrouter/CarbonRouterInstance.h

@@ -15,6 +15,8 @@
 #include <thread>
 #include <vector>
 
+#include <folly/CppAttributes.h>
+#include <folly/Likely.h>
 #include <folly/Range.h>
 #include <folly/executors/IOThreadPoolExecutor.h>
 
@@ -241,8 +243,17 @@ class CarbonRouterInstance
     cpuStatsWorker_.reset();
   }
 
-  Proxy<RouterInfo>& getProxyFromHash(size_t hash) {
-    return *proxies_[hash % proxies_.size()];
+  /**
+   * Returns nullptr while the router is draining; route locally instead.
+   * Must be called from a proxy event base thread, and the result must not
+   * outlive the current turn of that thread's loop.
+   */
+  Proxy<RouterInfo>* FOLLY_NULLABLE getProxyFromHash(size_t hash) const {
+    if (FOLLY_UNLIKELY(proxiesDraining_.load(std::memory_order_relaxed))) {
+      return nullptr;
+    }
+    DCHECK(!proxies_.empty());
+    return proxies_[hash % proxies_.size()];
   }
 
   CarbonRouterInstance(const CarbonRouterInstance&) = delete;
@@ -267,6 +278,13 @@ class CarbonRouterInstance
 
   std::atomic<bool> shutdownStarted_{false};
 
+  // Relaxed: the happens-before comes from fenceProxyEventBases(), not this.
+  std::atomic<bool> proxiesDraining_{false};
+
+  void setProxiesDraining() noexcept {
+    proxiesDraining_.store(true, std::memory_order_relaxed);
+  }
+
   FileObserverHandle runtimeVarsObserverHandle_;
 
   ConfigApi::CallbackHandle configUpdateHandle_;
@@ -305,6 +323,10 @@ class CarbonRouterInstance
   folly::Expected<folly::Unit, std::string> setupProxy(
       const std::vector<folly::EventBase*>& evbs);
 
+  // Waits for every proxy event base to cycle. Not callable from a proxy
+  // thread.
+  void fenceProxyEventBases() noexcept;
+
   void spawnAuxiliaryThreads();
   void joinAuxiliaryThreads() noexcept;
   void shutdownImpl() noexcept;

mcrouter/test/cpp_unit_tests/CarbonRouterInstanceShutdownTest.cpp

@@ -0,0 +1,139 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#include <atomic>
+#include <chrono>
+#include <future>
+#include <memory>
+#include <thread>
+
+#include <gtest/gtest.h>
+
+#include <folly/ScopeGuard.h>
+
+#include "mcrouter/CarbonRouterInstance.h"
+// defaultTestOptions(); its own header does not build when included directly.
+#include "mcrouter/options.h"
+#include "mcrouter/routes/McrouterRouteHandle.h"
+
+using namespace facebook::memcache;
+using namespace facebook::memcache::mcrouter;
+
+namespace {
+
+// One thread to hold hostage, one to observe.
+constexpr size_t kNumProxies = 2;
+
+// create(), not init(): each test must own and tear down its own router.
+std::shared_ptr<CarbonRouterInstance<McrouterRouterInfo>> makeRouter(
+    size_t numProxies) {
+  McrouterOptions opts = defaultTestOptions();
+  opts.config = "{ \"route\": \"NullRoute\" }";
+  opts.num_proxies = numProxies;
+  return CarbonRouterInstance<McrouterRouterInfo>::create(std::move(opts));
+}
+
+} // namespace
+
+// Hold proxy 1's thread hostage and check what a hop resolved from it sees.
+// Teardown stalls on that thread either way, so the task observes the state
+// teardown reached before it needed proxy 1.
+TEST(CarbonRouterInstanceShutdownTest, ShutdownDrainsProxiesBeforeFreeingThem) {
+  auto router = makeRouter(kNumProxies);
+  ASSERT_NE(router, nullptr);
+  ASSERT_NE(router->getProxy(1), nullptr);
+
+  // Index arithmetic only; off a proxy thread, so do not copy this shape.
+  for (size_t hash = 0; hash < 2 * kNumProxies; ++hash) {
+    EXPECT_EQ(
+        router->getProxyFromHash(hash), router->getProxy(hash % kNumProxies));
+  }
+
+  std::promise<void> taskRunning;
+  std::future<void> taskRunningFuture = taskRunning.get_future();
+  std::promise<bool> observed;
+  std::future<bool> observedFuture = observed.get_future();
+
+  router->getProxy(1)->eventBase().getEventBase().runInEventBaseThread(
+      [&router, &taskRunning, &observed] {
+        taskRunning.set_value();
+        bool draining = false;
+        // Bounded so a regression fails instead of pinning this thread.
+        for (int i = 0; i < 1000 && !draining; ++i) {
+          draining = router->getProxyFromHash(0) == nullptr;
+          if (!draining) {
+            /* sleep override */
+            std::this_thread::sleep_for(std::chrono::milliseconds(10));
+          }
+        }
+        observed.set_value(draining);
+      });
+
+  taskRunningFuture.wait();
+
+  std::thread shutdownThread([&router] { router->shutdown(); });
+  SCOPE_EXIT {
+    shutdownThread.join();
+  };
+
+  EXPECT_TRUE(observedFuture.get())
+      << "shutdown() began tearing proxies down without first marking them as "
+         "draining, so a thread jump could still resolve a proxy that is "
+         "about to be freed";
+}
+
+// No proxy may be destroyed while any proxy thread is still running: a hop
+// that resolved a target just before the flag flipped is still inline on its
+// own thread. Occupy proxy 1 and watch proxy 0.
+TEST(CarbonRouterInstanceShutdownTest, ShutdownFencesBeforeDestroyingAnyProxy) {
+  auto router = makeRouter(kNumProxies);
+  ASSERT_NE(router, nullptr);
+  ASSERT_NE(router->getProxy(0), nullptr);
+  ASSERT_NE(router->getProxy(1), nullptr);
+
+  std::atomic<bool> proxy0Destroyed{false};
+  router->getProxy(0)->eventBase().runOnDestruction(
+      [&proxy0Destroyed] { proxy0Destroyed.store(true); });
+
+  std::promise<void> taskRunning;
+  std::future<void> taskRunningFuture = taskRunning.get_future();
+  std::promise<void> release;
+  std::shared_future<void> releaseFuture = release.get_future();
+
+  router->getProxy(1)->eventBase().getEventBase().runInEventBaseThread(
+      [&taskRunning, releaseFuture] {
+        taskRunning.set_value();
+        releaseFuture.wait();
+      });
+  taskRunningFuture.wait();
+
+  std::thread shutdownThread([&router] { router->shutdown(); });
+
+  // Fallback for early returns; the happy path joins inline below.
+  bool released = false;
+  SCOPE_EXIT {
+    if (!released) {
+      release.set_value();
+      shutdownThread.join();
+    }
+  };
+
+  // Bounded negative wait; can only pass spuriously, never fail spuriously.
+  // Kept short: this pins a proxy thread and other tests in this binary are
+  // latency-sensitive. Without the fence, clear() frees proxy 0 immediately.
+  /* sleep override */
+  std::this_thread::sleep_for(std::chrono::milliseconds(150));
+  EXPECT_FALSE(proxy0Destroyed.load())
+      << "shutdown() destroyed a proxy while another proxy thread was still "
+         "running, so a hop that had already resolved that proxy as its target "
+         "could still be about to use it";
+
+  released = true;
+  release.set_value();
+  shutdownThread.join();
+  EXPECT_TRUE(proxy0Destroyed.load()); // not vacuous
+}