Commit e9ed24ce authored by Shen Xu's avatar Shen Xu Committed by Facebook Github Bot

Lock the creation of unordered reduce chain but not the actual reduce

Summary: Before this diff, since `ctx->memo_` starts out in the `OnlyResult` state, the first reduce (and therefore all subsequent reduces) are run immediately inside the lock.

Reviewed By: yfeldblum

Differential Revision: D8319067

fbshipit-source-id: 5ed08885beefb42c6efb61768fcbc36e0a08a46b
parent e87ec591
...@@ -1684,6 +1684,16 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) { ...@@ -1684,6 +1684,16 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) {
: lock_(), memo_(makeFuture<T>(std::move(memo))), : lock_(), memo_(makeFuture<T>(std::move(memo))),
func_(std::move(fn)), numThens_(0), numFutures_(n), promise_() func_(std::move(fn)), numThens_(0), numFutures_(n), promise_()
{} {}
static void fulfillWithValueOrFuture(Promise<T>&& p, T&& v) {
p.setValue(std::move(v));
}
static void fulfillWithValueOrFuture(Promise<T>&& p, Future<T>&& f) {
f.setCallback_(
[p = std::move(p)](Try<T>&& t) mutable { p.setTry(std::move(t)); });
}
folly::MicroSpinLock lock_; // protects memo_ and numThens_ folly::MicroSpinLock lock_; // protects memo_ and numThens_
Future<T> memo_; Future<T> memo_;
F func_; F func_;
...@@ -1704,19 +1714,35 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) { ...@@ -1704,19 +1714,35 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) {
// the order of completion to reduce the values. // the order of completion to reduce the values.
// The spinlock just protects chaining a new Future, not actually // The spinlock just protects chaining a new Future, not actually
// executing the reduce, which should be really fast. // executing the reduce, which should be really fast.
folly::MSLGuard lock(ctx->lock_); Promise<T> p;
ctx->memo_ = auto f = p.getFuture();
ctx->memo_.then([ ctx, mt = std::move(t) ](T && v) mutable { {
// Either return a ItT&& or a Try<ItT>&& depending folly::MSLGuard lock(ctx->lock_);
// on the type of the argument of func. f = exchange(ctx->memo_, std::move(f));
return ctx->func_(std::move(v), if (++ctx->numThens_ == ctx->numFutures_) {
mt.template get<IsTry::value, Arg&&>()); // After reducing the value of the last Future, fulfill the Promise
}); ctx->memo_.setCallback_(
if (++ctx->numThens_ == ctx->numFutures_) { [ctx](Try<T>&& t2) { ctx->promise_.setValue(std::move(t2)); });
// After reducing the value of the last Future, fulfill the Promise }
ctx->memo_.setCallback_(
[ctx](Try<T>&& t2) { ctx->promise_.setValue(std::move(t2)); });
} }
f.setCallback_([ctx, mp = std::move(p), mt = std::move(t)](
Try<T>&& v) mutable {
if (v.hasValue()) {
try {
ctx->fulfillWithValueOrFuture(
std::move(mp),
ctx->func_(
std::move(v.value()),
mt.template get<IsTry::value, Arg&&>()));
} catch (std::exception& e) {
mp.setException(exception_wrapper(std::current_exception(), e));
} catch (...) {
mp.setException(exception_wrapper(std::current_exception()));
}
} else {
mp.setTry(std::move(v));
}
});
}); });
return ctx->promise_.getSemiFuture().via(&InlineExecutor::instance()); return ctx->promise_.getSemiFuture().via(&InlineExecutor::instance());
......
...@@ -174,3 +174,30 @@ TEST(Reduce, unorderedReduceException) { ...@@ -174,3 +174,30 @@ TEST(Reduce, unorderedReduceException) {
p1.setValue(1); p1.setValue(1);
EXPECT_THROW(f.get(), std::runtime_error); EXPECT_THROW(f.get(), std::runtime_error);
} }
TEST(Reduce, unorderedReduceFuture) {
Promise<int> p1;
Promise<int> p2;
Promise<int> p3;
std::vector<Future<int>> fs;
fs.push_back(p1.getFuture());
fs.push_back(p2.getFuture());
fs.push_back(p3.getFuture());
std::vector<Promise<double>> ps(3);
Future<int> f =
unorderedReduce(fs.begin(), fs.end(), 0.0, [&](double /* a */, int&& b) {
return ps[b - 1].getFuture();
});
p3.setValue(3);
p2.setValue(2);
p1.setValue(1);
ps[0].setValue(1.0);
ps[1].setValue(2.0);
ps[2].setValue(3.0);
EXPECT_EQ(1.0, f.get());
}
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