protect executor callback destruction

Summary:
Callback destruction can involve user logic.  In practice, we
found quite a bunch of patterns like:

```
executor->add([x = someGuardObjectThatTriggersWorkInDtor] {
  ...
});
```

So ensure that dtor is

* Executed under `RequestContextScopeGuard`,
* Protected from leaking unhandled exceptions.

Reviewed By: ot, luciang

Differential Revision: D26744394

fbshipit-source-id: 5fcca0287a98de919c3649a9613e6d54cc1a1ba3

folly/executors/ThreadPoolExecutor.cpp

@@ -74,6 +74,23 @@ ThreadPoolExecutor::Task::Task(
   enqueueTime_ = std::chrono::steady_clock::now();
 }
 
+namespace {
+
+template <class F>
+void nothrow(const char* name, F&& f) {
+  try {
+    f();
+  } catch (const std::exception& e) {
+    LOG(ERROR) << "ThreadPoolExecutor: " << name << " threw unhandled "
+               << typeid(e).name() << " exception: " << e.what();
+  } catch (...) {
+    LOG(ERROR) << "ThreadPoolExecutor: " << name
+               << " threw unhandled non-exception object";
+  }
+}
+
+} // namespace
+
 void ThreadPoolExecutor::runTask(const ThreadPtr& thread, Task&& task) {
   thread->idle.store(false, std::memory_order_relaxed);
   auto startTime = std::chrono::steady_clock::now();
@@ -84,30 +101,29 @@ void ThreadPoolExecutor::runTask(const ThreadPtr& thread, Task&& task) {
   }
   stats.waitTime = startTime - task.enqueueTime_;
 
-  if (task.expiration_ > std::chrono::milliseconds(0) &&
-      stats.waitTime >= task.expiration_) {
-    stats.expired = true;
-    if (task.expireCallback_ != nullptr) {
-      task.expireCallback_();
-    }
-  } else {
+  {
     folly::RequestContextScopeGuard rctx(task.context_);
-    try {
-      task.func_();
-    } catch (const std::exception& e) {
-      LOG(ERROR) << "ThreadPoolExecutor: func threw unhandled "
-                 << typeid(e).name() << " exception: " << e.what();
-    } catch (...) {
-      LOG(ERROR) << "ThreadPoolExecutor: func threw unhandled non-exception "
-                    "object";
+    if (task.expiration_ > std::chrono::milliseconds(0) &&
+        stats.waitTime >= task.expiration_) {
+      stats.expired = true;
+      if (task.expireCallback_ != nullptr) {
+        nothrow("expireCallback", [&] { task.expireCallback_(); });
+      }
+    } else {
+      nothrow("func", [&] { task.func_(); });
     }
+    // Callback destruction might involve user logic, protect it as well.
+    nothrow("~func", [&] { task.func_ = nullptr; });
+    nothrow("~expireCallback", [&] { task.expireCallback_ = nullptr; });
+  }
+  if (!stats.expired) {
     stats.runTime = std::chrono::steady_clock::now() - startTime;
   }
 
   // Times in this USDT use granularity of std::chrono::steady_clock::duration,
   // which is platform dependent. On Facebook servers, the granularity is
   // nanoseconds. We explicitly do not perform any unit conversions to avoid
-  // unneccessary costs and leave it to consumers of this data to know what
+  // unnecessary costs and leave it to consumers of this data to know what
   // effective clock resolution is.
   FOLLY_SDT(
       folly,

folly/executors/test/ThreadPoolExecutorTest.cpp

@@ -590,19 +590,27 @@ class TestData : public folly::RequestData {
 };
 
 TEST(ThreadPoolExecutorTest, RequestContext) {
-  CPUThreadPoolExecutor executor(1);
-
   RequestContextScopeGuard rctx; // create new request context for this scope
   EXPECT_EQ(nullptr, RequestContext::get()->getContextData("test"));
   RequestContext::get()->setContextData("test", std::make_unique<TestData>(42));
   auto data = RequestContext::get()->getContextData("test");
   EXPECT_EQ(42, dynamic_cast<TestData*>(data)->data_);
 
-  executor.add([] {
-    auto data2 = RequestContext::get()->getContextData("test");
-    ASSERT_TRUE(data2 != nullptr);
-    EXPECT_EQ(42, dynamic_cast<TestData*>(data2)->data_);
-  });
+  struct VerifyRequestContext {
+    ~VerifyRequestContext() {
+      auto data2 = RequestContext::get()->getContextData("test");
+      EXPECT_TRUE(data2 != nullptr);
+      if (data2 != nullptr) {
+        EXPECT_EQ(42, dynamic_cast<TestData*>(data2)->data_);
+      }
+    }
+  };
+
+  {
+    CPUThreadPoolExecutor executor(1);
+    executor.add([] { VerifyRequestContext(); });
+    executor.add([x = VerifyRequestContext()] {});
+  }
 }
 
 std::atomic<int> g_sequence{};