Fix stack overflow in folly::window

Summary:
If the default inline executor is used for folly::window, and all the futures generated by the callback are also inline, then all spawn function calls (and their associated Futures) are added to the call stack and can overflow. This fixes that issue by using QueuedImmediateExecutor, which will put the executor callback into a thread-local queue, allowing the current function to exit and destruct all objects in its stack frame. Then, only the first call to QueuedImmediateExecutor in a thread is responsible for running each subsequent callback and will do so sequentially.

This will end up allocating dynamic queues that reach a max size of two. It is possible to avoid a dynamic queue by writing a new Executor that only stores a single function, if that is too much of a performance cost.

Reviewed By: yfeldblum

Differential Revision: D7597731

fbshipit-source-id: cd8899e6c086ecff812050f7241e80f39c546706

folly/futures/Future-inl.h

@@ -22,6 +22,7 @@
 
 #include <folly/Optional.h>
 #include <folly/executors/InlineExecutor.h>
+#include <folly/executors/QueuedImmediateExecutor.h>
 #include <folly/futures/Timekeeper.h>
 #include <folly/futures/detail/Core.h>
 #include <folly/synchronization/Baton.h>
@@ -1466,8 +1467,8 @@ Future<T> reduce(It first, It last, T&& initial, F&& func) {
 template <class Collection, class F, class ItT, class Result>
 std::vector<Future<Result>>
 window(Collection input, F func, size_t n) {
-  // Use global inline executor singleton
-  auto executor = &InlineExecutor::instance();
+  // Use global QueuedImmediateExecutor singleton to avoid stack overflow.
+  auto executor = &QueuedImmediateExecutor::instance();
   return window(executor, std::move(input), std::move(func), n);
 }
 

folly/futures/test/WindowTest.cpp

@@ -112,6 +112,40 @@ TEST(Window, exception) {
   EXPECT_EQ(2, res.get());
 }
 
+TEST(Window, stackOverflow) {
+  // Number of futures to spawn.
+  constexpr size_t m = 1000;
+  // Size of each block of input and output.
+  constexpr size_t n = 1000;
+
+  std::vector<std::array<int, n>> ints;
+  int64_t expectedSum = 0;
+  for (size_t i = 0; i < m; i++) {
+    std::array<int, n> next{};
+    next[i % n] = i;
+    ints.emplace_back(next);
+    expectedSum += i;
+  }
+
+  // Try to overflow window's executor.
+  auto res = reduce(
+      window(
+          ints,
+          [](std::array<int, n> i) {
+            return folly::Future<std::array<int, n>>(i);
+          },
+          1),
+      static_cast<int64_t>(0),
+      [](int64_t sum, const Try<std::array<int, n>>& b) {
+        for (int a : b.value()) {
+          sum += a;
+        }
+        return sum;
+      });
+
+  EXPECT_EQ(res.get(), expectedSum);
+}
+
 TEST(Window, parallel) {
   std::vector<int> input;
   std::vector<Promise<int>> ps(10);