Commit ebded22a authored by Walker Mills's avatar Walker Mills Committed by Facebook Github Bot

Add window overload that takes an executor to prevent stack overflow

Summary:
AIUI, if there is no executor available, then callbacks are executed inline. `folly::window` uses a recursive helper function (`spawn`) to handle chaining callbacks. So if `window` is used on a large enough collection of `Future`s without executors (e.g., created by `makeFuture`, or have otherwise already completed), you get a stack overflow. A minimal repro looks like:
```

int main(int argc, char** argv) {
  std::vector<int> v(100000);
  for(int i=0; i < v.size(); i++) {
    v[i] = i;
  }

  std::vector<folly::Future<folly::Unit>> f =
      folly::window(
          std::move(v),
          [](int /* unused */) { return folly::makeFuture(); },
          1);
  folly::collectAll(f).get();
}
```

This diff resolves the issue by adding an overload of `folly::window` which takes an executor as its first parameter. The executor-less `window` overload calls through to the new function using an `InlineExecutor` as the default executor.

Reviewed By: yfeldblum

Differential Revision: D6038733

fbshipit-source-id: 5dcab575592650efa2e106f12632ec06817a0009
parent 34a2f64f
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include <folly/Baton.h> #include <folly/Baton.h>
#include <folly/Optional.h> #include <folly/Optional.h>
#include <folly/futures/InlineExecutor.h>
#include <folly/futures/Timekeeper.h> #include <folly/futures/Timekeeper.h>
#include <folly/futures/detail/Core.h> #include <folly/futures/detail/Core.h>
...@@ -1067,27 +1068,38 @@ Future<T> reduce(It first, It last, T&& initial, F&& func) { ...@@ -1067,27 +1068,38 @@ Future<T> reduce(It first, It last, T&& initial, F&& func) {
template <class Collection, class F, class ItT, class Result> template <class Collection, class F, class ItT, class Result>
std::vector<Future<Result>> std::vector<Future<Result>>
window(Collection input, F func, size_t n) { window(Collection input, F func, size_t n) {
struct WindowContext { // Use global inline executor singleton
WindowContext(Collection&& i, F&& fn) auto executor = &InlineExecutor::instance();
: input_(std::move(i)), promises_(input_.size()), return window(executor, std::move(input), std::move(func), n);
func_(std::move(fn)) }
{}
std::atomic<size_t> i_ {0};
Collection input_;
std::vector<Promise<Result>> promises_;
F func_;
static inline void spawn(const std::shared_ptr<WindowContext>& ctx) { template <class Collection, class F, class ItT, class Result>
size_t i = ctx->i_++; std::vector<Future<Result>>
if (i < ctx->input_.size()) { window(Executor* executor, Collection input, F func, size_t n) {
// Using setCallback_ directly since we don't need the Future struct WindowContext {
ctx->func_(std::move(ctx->input_[i])).setCallback_( WindowContext(Executor* executor_, Collection&& input_, F&& func_)
// ctx is captured by value : executor(executor_),
[ctx, i](Try<Result>&& t) { input(std::move(input_)),
ctx->promises_[i].setTry(std::move(t)); promises(input.size()),
func(std::move(func_)) {}
std::atomic<size_t> i{0};
Executor* executor;
Collection input;
std::vector<Promise<Result>> promises;
F func;
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]));
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 {
ctx->promises[i].setTry(std::move(t));
// Chain another future onto this one // Chain another future onto this one
spawn(std::move(ctx)); spawn(std::move(ctx));
}); });
});
} }
} }
}; };
...@@ -1095,16 +1107,16 @@ window(Collection input, F func, size_t n) { ...@@ -1095,16 +1107,16 @@ window(Collection input, F func, size_t n) {
auto max = std::min(n, input.size()); auto max = std::min(n, input.size());
auto ctx = std::make_shared<WindowContext>( auto ctx = std::make_shared<WindowContext>(
std::move(input), std::move(func)); executor, std::move(input), std::move(func));
// Start the first n Futures
for (size_t i = 0; i < max; ++i) { for (size_t i = 0; i < max; ++i) {
// Start the first n Futures executor->add([ctx]() mutable { WindowContext::spawn(std::move(ctx)); });
WindowContext::spawn(ctx);
} }
std::vector<Future<Result>> futures; std::vector<Future<Result>> futures;
futures.reserve(ctx->promises_.size()); futures.reserve(ctx->promises.size());
for (auto& promise : ctx->promises_) { for (auto& promise : ctx->promises) {
futures.emplace_back(promise.getFuture()); futures.emplace_back(promise.getFuture());
} }
......
...@@ -328,6 +328,15 @@ template < ...@@ -328,6 +328,15 @@ template <
class Result = typename futures::detail::resultOf<F, ItT&&>::value_type> class Result = typename futures::detail::resultOf<F, ItT&&>::value_type>
std::vector<Future<Result>> window(Collection input, F func, size_t n); std::vector<Future<Result>> window(Collection input, F func, size_t n);
template <
class Collection,
class F,
class ItT = typename std::iterator_traits<
typename Collection::iterator>::value_type,
class Result = typename futures::detail::resultOf<F, ItT&&>::value_type>
std::vector<Future<Result>>
window(Executor* executor, Collection input, F func, size_t n);
template <typename F, typename T, typename ItT> template <typename F, typename T, typename ItT>
using MaybeTryArg = typename std::conditional< using MaybeTryArg = typename std::conditional<
futures::detail::callableWith<F, T&&, Try<ItT>&&>::value, futures::detail::callableWith<F, T&&, Try<ItT>&&>::value,
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#include <folly/Conv.h> #include <folly/Conv.h>
#include <folly/futures/Future.h> #include <folly/futures/Future.h>
#include <folly/futures/ManualExecutor.h>
#include <folly/portability/GTest.h> #include <folly/portability/GTest.h>
#include <vector> #include <vector>
...@@ -42,17 +43,17 @@ TEST(Window, basic) { ...@@ -42,17 +43,17 @@ TEST(Window, basic) {
EXPECT_EQ(expect, res); EXPECT_EQ(expect, res);
}; };
{ {
// 2 in-flight at a time SCOPED_TRACE("2 in-flight at a time");
std::vector<int> input = {1, 2, 3}; std::vector<int> input = {1, 2, 3};
fn(input, 2, 6); fn(input, 2, 6);
} }
{ {
// 4 in-flight at a time SCOPED_TRACE("4 in-flight at a time");
std::vector<int> input = {1, 2, 3}; std::vector<int> input = {1, 2, 3};
fn(input, 4, 6); fn(input, 4, 6);
} }
{ {
// empty input SCOPED_TRACE("empty input");
std::vector<int> input; std::vector<int> input;
fn(input, 1, 0); fn(input, 1, 0);
} }
...@@ -104,8 +105,8 @@ TEST(Window, parallel) { ...@@ -104,8 +105,8 @@ TEST(Window, parallel) {
barrier.wait(); barrier.wait();
for (size_t i = 0; i < ps.size(); i++) { for (auto& t : ts) {
ts[i].join(); t.join();
} }
EXPECT_TRUE(f.isReady()); EXPECT_TRUE(f.isReady());
...@@ -139,8 +140,8 @@ TEST(Window, parallelWithError) { ...@@ -139,8 +140,8 @@ TEST(Window, parallelWithError) {
barrier.wait(); barrier.wait();
for (size_t i = 0; i < ps.size(); i++) { for (auto& t : ts) {
ts[i].join(); t.join();
} }
EXPECT_TRUE(f.isReady()); EXPECT_TRUE(f.isReady());
...@@ -172,8 +173,8 @@ TEST(Window, allParallelWithError) { ...@@ -172,8 +173,8 @@ TEST(Window, allParallelWithError) {
barrier.wait(); barrier.wait();
for (size_t i = 0; i < ps.size(); i++) { for (auto& t : ts) {
ts[i].join(); t.join();
} }
EXPECT_TRUE(f.isReady()); EXPECT_TRUE(f.isReady());
...@@ -186,3 +187,173 @@ TEST(Window, allParallelWithError) { ...@@ -186,3 +187,173 @@ TEST(Window, allParallelWithError) {
} }
} }
} }
TEST(WindowExecutor, basic) {
ManualExecutor executor;
// int -> Future<int>
auto fn = [executor_ = &executor](
std::vector<int> input, size_t window_size, size_t expect) {
auto res = reduce(
window(
executor_, input, [](int i) { return makeFuture(i); }, window_size),
0,
[](int sum, const Try<int>& b) { return sum + *b; });
executor_->waitFor(res);
EXPECT_EQ(expect, res.get());
};
{
SCOPED_TRACE("2 in-flight at a time");
std::vector<int> input = {1, 2, 3};
fn(input, 2, 6);
}
{
SCOPED_TRACE("4 in-flight at a time");
std::vector<int> input = {1, 2, 3};
fn(input, 4, 6);
}
{
SCOPED_TRACE("empty input");
std::vector<int> input;
fn(input, 1, 0);
}
{
// int -> Future<Unit>
auto res = reduce(
window(
&executor,
std::vector<int>({1, 2, 3}),
[](int /* i */) { return makeFuture(); },
2),
0,
[](int sum, const Try<Unit>& b) {
EXPECT_TRUE(b.hasValue());
return sum + 1;
});
executor.waitFor(res);
EXPECT_EQ(3, res.get());
}
{
// string -> return Future<int>
auto res = reduce(
window(
&executor,
std::vector<std::string>{"1", "2", "3"},
[](std::string s) { return makeFuture<int>(folly::to<int>(s)); },
2),
0,
[](int sum, const Try<int>& b) { return sum + *b; });
executor.waitFor(res);
EXPECT_EQ(6, res.get());
}
}
TEST(WindowExecutor, parallel) {
ManualExecutor executor;
std::vector<int> input;
std::vector<Promise<int>> ps(10);
for (size_t i = 0; i < ps.size(); i++) {
input.emplace_back(i);
}
auto f = collect(
window(&executor, input, [&](int i) { return ps[i].getFuture(); }, 3));
std::vector<std::thread> ts;
boost::barrier barrier(ps.size() + 1);
for (size_t i = 0; i < ps.size(); i++) {
ts.emplace_back([&ps, &barrier, i]() {
barrier.wait();
ps[i].setValue(i);
});
}
barrier.wait();
for (auto& t : ts) {
t.join();
}
executor.waitFor(f);
EXPECT_TRUE(f.isReady());
for (size_t i = 0; i < ps.size(); i++) {
EXPECT_EQ(i, f.value()[i]);
}
}
TEST(WindowExecutor, parallelWithError) {
ManualExecutor executor;
std::vector<int> input;
std::vector<Promise<int>> ps(10);
for (size_t i = 0; i < ps.size(); i++) {
input.emplace_back(i);
}
auto f = collect(
window(&executor, input, [&](int i) { return ps[i].getFuture(); }, 3));
std::vector<std::thread> ts;
boost::barrier barrier(ps.size() + 1);
for (size_t i = 0; i < ps.size(); i++) {
ts.emplace_back([&ps, &barrier, i]() {
barrier.wait();
if (i == (ps.size() / 2)) {
ps[i].setException(eggs);
} else {
ps[i].setValue(i);
}
});
}
barrier.wait();
for (auto& t : ts) {
t.join();
}
executor.waitFor(f);
EXPECT_TRUE(f.isReady());
EXPECT_THROW(f.value(), eggs_t);
}
TEST(WindowExecutor, allParallelWithError) {
ManualExecutor executor;
std::vector<int> input;
std::vector<Promise<int>> ps(10);
for (size_t i = 0; i < ps.size(); i++) {
input.emplace_back(i);
}
auto f = collectAll(
window(&executor, input, [&](int i) { return ps[i].getFuture(); }, 3));
std::vector<std::thread> ts;
boost::barrier barrier(ps.size() + 1);
for (size_t i = 0; i < ps.size(); i++) {
ts.emplace_back([&ps, &barrier, i]() {
barrier.wait();
if (i == (ps.size() / 2)) {
ps[i].setException(eggs);
} else {
ps[i].setValue(i);
}
});
}
barrier.wait();
for (auto& t : ts) {
t.join();
}
executor.waitFor(f);
EXPECT_TRUE(f.isReady());
for (size_t i = 0; i < ps.size(); i++) {
if (i == (ps.size() / 2)) {
EXPECT_THROW(f.value()[i].value(), eggs_t);
} else {
EXPECT_TRUE(f.value()[i].hasValue());
EXPECT_EQ(i, f.value()[i].value());
}
}
}
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment