Commit 51e2bb94 authored by Brian Bustamante's avatar Brian Bustamante Committed by Facebook Github Bot

collectAllSemiFuture propagate completeKA to permit inlining

Summary: Allow propagation of completing executor KA in order to permit inlining

Reviewed By: LeeHowes

Differential Revision: D17123836

fbshipit-source-id: 7524868284f9482f56e5b8cd8b9e33bc4212de74
parent bba44900
......@@ -1415,12 +1415,15 @@ collectAllSemiFuture(InputIterator first, InputIterator last) {
using T = typename F::value_type;
struct Context {
explicit Context(size_t n) : results(n) {}
explicit Context(size_t n) : results(n), count(n) {}
~Context() {
p.setValue(std::move(results));
futures::detail::setTry(
p, std::move(ka), Try<std::vector<Try<T>>>(std::move(results)));
}
Promise<std::vector<Try<T>>> p;
Executor::KeepAlive<> ka;
std::vector<Try<T>> results;
std::atomic<size_t> count;
};
std::vector<futures::detail::DeferredWrapper> executors;
......@@ -1429,9 +1432,14 @@ collectAllSemiFuture(InputIterator first, InputIterator last) {
auto ctx = std::make_shared<Context>(size_t(std::distance(first, last)));
for (size_t i = 0; first != last; ++first, ++i) {
first->setCallback_([i, ctx](Executor::KeepAlive<>&&, Try<T>&& t) {
ctx->results[i] = std::move(t);
});
first->setCallback_(
[i, ctx](Executor::KeepAlive<>&& ka, Try<T>&& t) {
ctx->results[i] = std::move(t);
if (ctx->count.fetch_sub(1, std::memory_order_acq_rel) == 1) {
ctx->ka = std::move(ka);
}
},
futures::detail::InlineContinuation::permit);
}
auto future = ctx->p.getSemiFuture();
......@@ -1936,25 +1944,24 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) {
});
}
}
f.setCallback_(
[ctx, mp = std::move(p), mt = std::move(t)](
Executor::KeepAlive<>&&, Try<T>&& v) mutable {
if (v.hasValue()) {
try {
Fulfill{}(
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));
}
});
f.setCallback_([ctx, mp = std::move(p), mt = std::move(t)](
Executor::KeepAlive<>&&, Try<T>&& v) mutable {
if (v.hasValue()) {
try {
Fulfill{}(
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());
......
......@@ -35,6 +35,10 @@ void coreDetachPromiseMaybeWithResult(Core<T>& core) {
}
core.detachPromise();
}
template <typename T>
void setTry(Promise<T>& p, Executor::KeepAlive<>&& ka, Try<T>&& t) {
p.setTry(std::move(ka), std::move(t));
}
} // namespace detail
} // namespace futures
......
......@@ -58,6 +58,8 @@ template <class T>
class SemiFuture;
template <class T>
class Future;
template <class T>
class Promise;
namespace futures {
namespace detail {
......@@ -66,6 +68,8 @@ class FutureBase;
struct EmptyConstruct {};
template <typename T, typename F>
class CoreCallbackState;
template <typename T>
void setTry(Promise<T>& p, Executor::KeepAlive<>&& ka, Try<T>&& t);
} // namespace detail
} // namespace futures
......@@ -407,6 +411,10 @@ class Promise {
friend class Future;
template <class, class>
friend class futures::detail::CoreCallbackState;
friend void futures::detail::setTry<T>(
Promise<T>& p,
Executor::KeepAlive<>&& ka,
Try<T>&& t);
// Whether the Future has been retrieved (a one-time operation).
bool retrieved_;
......
......@@ -19,6 +19,7 @@
#include <boost/thread/barrier.hpp>
#include <folly/Random.h>
#include <folly/executors/ManualExecutor.h>
#include <folly/futures/Future.h>
#include <folly/portability/GTest.h>
#include <folly/small_vector.h>
......@@ -111,6 +112,67 @@ TEST(Collect, collectAll) {
}
}
TEST(Collect, collectAllSemiFutureInline) {
// inline future collection on same executor
{
ManualExecutor x;
std::vector<Future<int>> futures;
futures.emplace_back(makeFuture(42).via(&x));
futures.emplace_back(makeFuture(42).via(&x));
futures.emplace_back(makeFuture(42).via(&x));
auto allf =
collectAllSemiFuture(futures).via(&x).thenTryInline([](auto&&) {});
EXPECT_FALSE(allf.isReady());
EXPECT_EQ(3, x.run());
EXPECT_TRUE(allf.isReady());
}
// inline defered semi-future collection on same executor
{
ManualExecutor x;
std::vector<SemiFuture<int>> futures;
futures.emplace_back(makeSemiFuture(42).defer([](auto&&) { return 42; }));
futures.emplace_back(makeSemiFuture(42).defer([](auto&&) { return 42; }));
futures.emplace_back(makeSemiFuture(42).defer([](auto&&) { return 42; }));
auto allf = collectAllSemiFuture(futures).defer([](auto&&) {}).via(&x);
EXPECT_FALSE(allf.isReady());
EXPECT_EQ(3, x.run());
EXPECT_TRUE(allf.isReady());
}
// inline future collection lastly fullfilled on same executor
{
ManualExecutor x1, x2;
std::vector<Future<int>> futures;
futures.emplace_back(makeFuture(42).via(&x1));
futures.emplace_back(makeFuture(42).via(&x2));
auto allf = collectAllSemiFuture(futures).defer([](auto&&) {}).via(&x1);
EXPECT_FALSE(allf.isReady());
EXPECT_EQ(1, x2.run());
EXPECT_FALSE(allf.isReady());
EXPECT_EQ(1, x1.run());
EXPECT_TRUE(allf.isReady());
}
// prevent inlining of future collection lastly fullfilled on different
// executor
{
ManualExecutor x1, x2;
std::vector<Future<int>> futures;
futures.emplace_back(makeFuture(42).via(&x1));
futures.emplace_back(makeFuture(42).via(&x2));
auto allf = collectAllSemiFuture(futures).defer([](auto&&) {}).via(&x1);
EXPECT_FALSE(allf.isReady());
EXPECT_EQ(1, x1.run());
EXPECT_FALSE(allf.isReady());
EXPECT_EQ(1, x2.run());
EXPECT_FALSE(allf.isReady());
EXPECT_EQ(1, x1.run());
EXPECT_TRUE(allf.isReady());
}
}
TEST(Collect, collect) {
// success case
{
......
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