Avoid McRouter leakage

Summary:
McRouter leaks the client instance if it encounters errors, like config failure, SMC unavailability, etc. This has caused [some issues](https://fb.workplace.com/groups/604299349618684/permalink/27403334835955105/) for us over the past year.

**This diff:** Moving the client instance to the aux thread, so it can be auto-destructed after its SR dependencies are released.

With the [context](https://fb.workplace.com/groups/604299349618684/posts/27403334835955105/?comment_id=27406017052353550&reply_comment_id=27406822342273021) disylh shared, I was curious to see if this approach might work, and what might be the gaps we need to tackle.

I followed the discussion on the [previous attempt](https://www.internalfb.com/diff/D44641566?dst_version_fbid=1590697228110996&transaction_fbid=143930055086555) where this was discussed. Iiuc, #1 is addressed now as we manually do the SR deps release before destruction. I didn't quite follow #2, and why clearing the event base would cause the deadlock (maybe it wouldn't anymore, since we destruct from a different thread). But curious to hear opinions!

Reviewed By: disylh

Differential Revision: D66252318

fbshipit-source-id: 498ea74ab78c6dc0e70b1471a0779551002e27ac

mcrouter/CarbonRouterInstance-inl.h

@@ -18,6 +18,8 @@
 #include <folly/io/async/EventBase.h>
 #include <folly/json/DynamicConverter.h>
 
+#include <common/logging/logging.h>
+
 #include "mcrouter/AsyncWriter.h"
 #include "mcrouter/CarbonRouterInstanceBase.h"
 #include "mcrouter/ExecutorObserver.h"
@@ -127,7 +129,15 @@ CarbonRouterInstance<RouterInfo>* CarbonRouterInstance<RouterInfo>::createRaw(
     initFailureLogger();
   }
 
-  auto router = new CarbonRouterInstance<RouterInfo>(std::move(input_options));
+  // Deleter is only called if unique_ptr::get() returns non-null.
+  auto deleter = [](CarbonRouterInstance<RouterInfo>* inst) {
+    LOG_EVERY_MS(WARNING, 10000) << "Destroying CarbonRouterInstance";
+    delete inst;
+  };
+  auto router =
+      std::unique_ptr<CarbonRouterInstance<RouterInfo>, decltype(deleter)>(
+          new CarbonRouterInstance<RouterInfo>(std::move(input_options)),
+          deleter);
 
   folly::Expected<folly::Unit, std::string> result;
   try {
@@ -141,7 +151,7 @@ CarbonRouterInstance<RouterInfo>* CarbonRouterInstance<RouterInfo>::createRaw(
 
     result = router->spinUp();
     if (result.hasValue()) {
-      return router;
+      return router.release();
     }
   } catch (...) {
     result = folly::makeUnexpected(
@@ -161,10 +171,20 @@ CarbonRouterInstance<RouterInfo>* CarbonRouterInstance<RouterInfo>::createRaw(
   // can be released.
   // We schedule the deletion on auxiliary thread pool
   // to avoid potential deadlock with the current thread.
-  auxThreadPool->add([router, auxThreadPool]() mutable {
-    router->resetMetadata();
-    router->resetAxonProxyClientFactory();
-  });
+  auxThreadPool->add(
+      [router = std::move(router),
+       auxThreadPool,
+       deleteRouter =
+           input_options.delete_carbon_instance_upon_init_failure]() mutable {
+        router->resetMetadata();
+        router->resetAxonProxyClientFactory();
+        if (!deleteRouter) {
+          // Leak it. unique_ptr::get() is guaranteed to return nullptr after
+          // this.
+          router.release();
+        }
+        // router gets destroyed here, unless configuration asks to leak it.
+      });
 
   throw std::runtime_error(std::move(result.error()));
 }

mcrouter/mcrouter_options_list.h

@@ -958,6 +958,13 @@ MCROUTER_OPTION_INTEGER(
     "Measure proxy CPU utilization every proxy_cpu_interval_s seconds. "
     "0 means disabled.")
 
+MCROUTER_OPTION_TOGGLE(
+    delete_carbon_instance_upon_init_failure,
+    true,
+    "delete-carbon-instance-upon-init-failure",
+    no_short,
+    "Controls whether we delete the carbon instance upon failures or leak it")
+
 #ifdef ADDITIONAL_OPTIONS_FILE
 #include ADDITIONAL_OPTIONS_FILE
 #endif