Fix a segfault in folly::window

Summary:
If the callback passed to `window` throws an exception, the process will
segfault.

This happens because `Core::doCallback` will execute a callback that throws
an exception twice. Once with the original arguments, and again with the wrapped
exception: https://fburl.com/wdmztc4l

In the case of `window`, the callback in question moves its `ctx` to the next
iteration of `spawn`, and so the second call to the callback segfaults.

This prevents the issue by wrapping the call to `func` in a future, which captures
the exception and returns it to the caller, rather than allowing it to bubble out and
break things.

Reviewed By: yfeldblum

Differential Revision: D6802396

fbshipit-source-id: 3595c622e8e8f58209c7b20f553f7642bc49c089

folly/futures/Future-inl.h

@@ -1170,7 +1170,8 @@ window(Executor* executor, Collection input, F func, size_t n) {
     static inline void spawn(std::shared_ptr<WindowContext> ctx) {
       size_t i = ctx->i++;
       if (i < ctx->input.size()) {
-        auto fut = ctx->func(std::move(ctx->input[i]));
+        auto fut =
+            makeFutureWith([&] { return ctx->func(std::move(ctx->input[i])); });
         fut.setCallback_([ctx = std::move(ctx), i](Try<Result>&& t) mutable {
           const auto executor_ = ctx->executor;
           executor_->add([ctx = std::move(ctx), i, t = std::move(t)]() mutable {

folly/futures/test/WindowTest.cpp

@@ -84,6 +84,34 @@ TEST(Window, basic) {
   }
 }
 
+TEST(Window, exception) {
+  std::vector<int> ints = {1, 2, 3, 4};
+  std::vector<Promise<int>> ps(4);
+
+  auto res = reduce(
+      window(
+          ints,
+          [&ps](int i) {
+            if (i > 2) {
+              throw std::runtime_error("exception should not kill process");
+            }
+            return ps[i].getFuture();
+          },
+          2),
+      0,
+      [](int sum, const Try<int>& b) {
+        sum += b.hasException<std::exception>() ? 1 : 0;
+        return sum;
+      });
+
+  for (auto& p : ps) {
+    p.setValue(0);
+  }
+
+  // Should have received 2 exceptions.
+  EXPECT_EQ(2, res.get());
+}
+
 TEST(Window, parallel) {
   std::vector<int> input;
   std::vector<Promise<int>> ps(10);