Commit a8990a74 authored by Yedidya Feldblum's avatar Yedidya Feldblum Committed by Facebook Github Bot

Refactor non-variadic collect variants

Summary:
[Folly] Refactor non-variadic `collect` variants.

No change in behavior - just moving code around.

Reviewed By: marshallcline

Differential Revision: D8440630

fbshipit-source-id: a6349c73c2109237f1be174c2260e9c9eb2d1c0e
parent 698d1e78
...@@ -1315,16 +1315,6 @@ FOLLY_ALWAYS_INLINE FOLLY_ATTR_VISIBILITY_HIDDEN void foreach( ...@@ -1315,16 +1315,6 @@ FOLLY_ALWAYS_INLINE FOLLY_ATTR_VISIBILITY_HIDDEN void foreach(
} // namespace detail } // namespace detail
} // namespace futures } // namespace futures
// mapSetCallback calls func(i, Try<T>) when every future completes
template <class T, class InputIterator, class F>
void mapSetCallback(InputIterator first, InputIterator last, F func) {
for (size_t i = 0; first != last; ++first, ++i) {
first->setCallback_([func, i](Try<T>&& t) {
func(i, std::move(t));
});
}
}
// collectAll (variadic) // collectAll (variadic)
template <typename... Fs> template <typename... Fs>
...@@ -1362,23 +1352,23 @@ template <class InputIterator> ...@@ -1362,23 +1352,23 @@ template <class InputIterator>
SemiFuture<std::vector< SemiFuture<std::vector<
Try<typename std::iterator_traits<InputIterator>::value_type::value_type>>> Try<typename std::iterator_traits<InputIterator>::value_type::value_type>>>
collectAllSemiFuture(InputIterator first, InputIterator last) { collectAllSemiFuture(InputIterator first, InputIterator last) {
typedef using F = typename std::iterator_traits<InputIterator>::value_type;
typename std::iterator_traits<InputIterator>::value_type::value_type T; using T = typename F::value_type;
struct CollectAllContext { struct Context {
CollectAllContext(size_t n) : results(n) {} explicit Context(size_t n) : results(n) {}
~CollectAllContext() { ~Context() {
p.setValue(std::move(results)); p.setValue(std::move(results));
} }
Promise<std::vector<Try<T>>> p; Promise<std::vector<Try<T>>> p;
std::vector<Try<T>> results; std::vector<Try<T>> results;
}; };
auto ctx = auto ctx = std::make_shared<Context>(size_t(std::distance(first, last)));
std::make_shared<CollectAllContext>(size_t(std::distance(first, last))); for (size_t i = 0; first != last; ++first, ++i) {
mapSetCallback<T>(first, last, [ctx](size_t i, Try<T>&& t) { first->setCallback_(
ctx->results[i] = std::move(t); [i, ctx](Try<T>&& t) { ctx->results[i] = std::move(t); });
}); }
return ctx->p.getSemiFuture(); return ctx->p.getSemiFuture();
} }
...@@ -1391,68 +1381,47 @@ collectAll(InputIterator first, InputIterator last) { ...@@ -1391,68 +1381,47 @@ collectAll(InputIterator first, InputIterator last) {
// collect (iterator) // collect (iterator)
namespace futures { // TODO(T26439406): Make return SemiFuture
namespace detail { template <class InputIterator>
Future<std::vector<
template <typename T> typename std::iterator_traits<InputIterator>::value_type::value_type>>
struct CollectContext { collect(InputIterator first, InputIterator last) {
struct Nothing { using F = typename std::iterator_traits<InputIterator>::value_type;
explicit Nothing(int /* n */) {} using T = typename F::value_type;
};
using Result = typename std::conditional<
std::is_void<T>::value,
void,
std::vector<T>>::type;
using InternalResult = typename std::conditional<
std::is_void<T>::value,
Nothing,
std::vector<Optional<T>>>::type;
explicit CollectContext(size_t n) : result(n) { struct Context {
explicit Context(size_t n) : result(n) {
finalResult.reserve(n); finalResult.reserve(n);
} }
~CollectContext() { ~Context() {
if (!threw.exchange(true)) { if (!threw.exchange(true)) {
// map Optional<T> -> T // map Optional<T> -> T
std::transform(result.begin(), result.end(), std::transform(
result.begin(),
result.end(),
std::back_inserter(finalResult), std::back_inserter(finalResult),
[](Optional<T>& o) { return std::move(o.value()); }); [](Optional<T>& o) { return std::move(o.value()); });
p.setValue(std::move(finalResult)); p.setValue(std::move(finalResult));
} }
} }
void setPartialResult(size_t i, Try<T>& t) { Promise<std::vector<T>> p;
result[i] = std::move(t.value()); std::vector<Optional<T>> result;
} std::vector<T> finalResult;
Promise<Result> p; std::atomic<bool> threw{false};
InternalResult result; };
Result finalResult;
std::atomic<bool> threw {false};
};
} // namespace detail
} // namespace futures
// TODO(T26439406): Make return SemiFuture
template <class InputIterator>
Future<typename futures::detail::CollectContext<typename std::iterator_traits<
InputIterator>::value_type::value_type>::Result>
collect(InputIterator first, InputIterator last) {
typedef
typename std::iterator_traits<InputIterator>::value_type::value_type T;
auto ctx = std::make_shared<futures::detail::CollectContext<T>>( auto ctx = std::make_shared<Context>(std::distance(first, last));
std::distance(first, last)); for (size_t i = 0; first != last; ++first, ++i) {
mapSetCallback<T>(first, last, [ctx](size_t i, Try<T>&& t) { first->setCallback_([i, ctx](Try<T>&& t) {
if (t.hasException()) { if (t.hasException()) {
if (!ctx->threw.exchange(true)) { if (!ctx->threw.exchange(true)) {
ctx->p.setException(std::move(t.exception())); ctx->p.setException(std::move(t.exception()));
} }
} else if (!ctx->threw) { } else if (!ctx->threw) {
ctx->setPartialResult(i, t); ctx->result[i] = std::move(t.value());
} }
}); });
}
return ctx->p.getSemiFuture().via(&InlineExecutor::instance()); return ctx->p.getSemiFuture().via(&InlineExecutor::instance());
} }
...@@ -1501,21 +1470,22 @@ Future< ...@@ -1501,21 +1470,22 @@ Future<
typename typename
std::iterator_traits<InputIterator>::value_type::value_type>>> std::iterator_traits<InputIterator>::value_type::value_type>>>
collectAny(InputIterator first, InputIterator last) { collectAny(InputIterator first, InputIterator last) {
typedef using F = typename std::iterator_traits<InputIterator>::value_type;
typename std::iterator_traits<InputIterator>::value_type::value_type T; using T = typename F::value_type;
struct CollectAnyContext { struct Context {
CollectAnyContext() {}
Promise<std::pair<size_t, Try<T>>> p; Promise<std::pair<size_t, Try<T>>> p;
std::atomic<bool> done {false}; std::atomic<bool> done {false};
}; };
auto ctx = std::make_shared<CollectAnyContext>(); auto ctx = std::make_shared<Context>();
mapSetCallback<T>(first, last, [ctx](size_t i, Try<T>&& t) { for (size_t i = 0; first != last; ++first, ++i) {
first->setCallback_([i, ctx](Try<T>&& t) {
if (!ctx->done.exchange(true)) { if (!ctx->done.exchange(true)) {
ctx->p.setValue(std::make_pair(i, std::move(t))); ctx->p.setValue(std::make_pair(i, std::move(t)));
} }
}); });
}
return ctx->p.getSemiFuture().via(&InlineExecutor::instance()); return ctx->p.getSemiFuture().via(&InlineExecutor::instance());
} }
...@@ -1527,27 +1497,27 @@ Future<std::pair< ...@@ -1527,27 +1497,27 @@ Future<std::pair<
size_t, size_t,
typename std::iterator_traits<InputIterator>::value_type::value_type>> typename std::iterator_traits<InputIterator>::value_type::value_type>>
collectAnyWithoutException(InputIterator first, InputIterator last) { collectAnyWithoutException(InputIterator first, InputIterator last) {
typedef using F = typename std::iterator_traits<InputIterator>::value_type;
typename std::iterator_traits<InputIterator>::value_type::value_type T; using T = typename F::value_type;
struct CollectAnyWithoutExceptionContext { struct Context {
CollectAnyWithoutExceptionContext(){} Context(size_t n) : nTotal(n) {}
Promise<std::pair<size_t, T>> p; Promise<std::pair<size_t, T>> p;
std::atomic<bool> done{false}; std::atomic<bool> done{false};
std::atomic<size_t> nFulfilled{0}; std::atomic<size_t> nFulfilled{0};
size_t nTotal; size_t nTotal;
}; };
auto ctx = std::make_shared<CollectAnyWithoutExceptionContext>(); auto ctx = std::make_shared<Context>(size_t(std::distance(first, last)));
ctx->nTotal = size_t(std::distance(first, last)); for (size_t i = 0; first != last; ++first, ++i) {
first->setCallback_([i, ctx](Try<T>&& t) {
mapSetCallback<T>(first, last, [ctx](size_t i, Try<T>&& t) {
if (!t.hasException() && !ctx->done.exchange(true)) { if (!t.hasException() && !ctx->done.exchange(true)) {
ctx->p.setValue(std::make_pair(i, std::move(t.value()))); ctx->p.setValue(std::make_pair(i, std::move(t.value())));
} else if (++ctx->nFulfilled == ctx->nTotal) { } else if (++ctx->nFulfilled == ctx->nTotal) {
ctx->p.setException(t.exception()); ctx->p.setException(t.exception());
} }
}); });
}
return ctx->p.getSemiFuture().via(&InlineExecutor::instance()); return ctx->p.getSemiFuture().via(&InlineExecutor::instance());
} }
...@@ -1558,62 +1528,57 @@ SemiFuture<std::vector<std::pair< ...@@ -1558,62 +1528,57 @@ SemiFuture<std::vector<std::pair<
size_t, size_t,
Try<typename std::iterator_traits<InputIterator>::value_type::value_type>>>> Try<typename std::iterator_traits<InputIterator>::value_type::value_type>>>>
collectN(InputIterator first, InputIterator last, size_t n) { collectN(InputIterator first, InputIterator last, size_t n) {
using T = using F = typename std::iterator_traits<InputIterator>::value_type;
typename std::iterator_traits<InputIterator>::value_type::value_type; using T = typename F::value_type;
using V = std::vector<Optional<Try<T>>>;
using Result = std::vector<std::pair<size_t, Try<T>>>; using Result = std::vector<std::pair<size_t, Try<T>>>;
assert(n > 0); struct Context {
assert(std::distance(first, last) >= 0); explicit Context(size_t numFutures, size_t min_)
: v(numFutures), min(min_) {}
struct CollectNContext {
explicit CollectNContext(size_t numFutures) : v(numFutures) {}
V v; std::vector<Optional<Try<T>>> v;
size_t min;
std::atomic<size_t> completed = {0}; // # input futures completed std::atomic<size_t> completed = {0}; // # input futures completed
std::atomic<size_t> stored = {0}; // # output values stored std::atomic<size_t> stored = {0}; // # output values stored
Promise<Result> p; Promise<Result> p;
};
void setPartialResult(size_t index, Try<T>&& t) { assert(n > 0);
v[index] = std::move(t); assert(std::distance(first, last) >= 0);
}
void complete() { if (size_t(std::distance(first, last)) < n) {
Result result; return SemiFuture<Result>(
result.reserve(completed.load()); exception_wrapper(std::runtime_error("Not enough futures")));
for (size_t i = 0; i < v.size(); ++i) {
auto& entry = v[i];
if (entry.hasValue()) {
result.emplace_back(i, std::move(entry).value());
} }
}
p.setTry(Try<Result>(std::move(result)));
}
};
auto numFutures = static_cast<size_t>(std::distance(first, last));
auto ctx = std::make_shared<CollectNContext>(numFutures);
if (numFutures < n) {
ctx->p.setException(std::runtime_error("Not enough futures"));
} else {
// for each completed Future, increase count and add to vector, until we // for each completed Future, increase count and add to vector, until we
// have n completed futures at which point we fulfil our Promise with the // have n completed futures at which point we fulfil our Promise with the
// vector // vector
mapSetCallback<T>(first, last, [ctx, n](size_t i, Try<T>&& t) { auto ctx = std::make_shared<Context>(size_t(std::distance(first, last)), n);
for (size_t i = 0; first != last; ++first, ++i) {
first->setCallback_([i, ctx](Try<T>&& t) {
// relaxed because this guards control but does not guard data // relaxed because this guards control but does not guard data
auto const c = 1 + ctx->completed.fetch_add(1, std::memory_order_relaxed); auto const c = 1 + ctx->completed.fetch_add(1, std::memory_order_relaxed);
if (c > n) { if (c > ctx->min) {
return; return;
} }
ctx->setPartialResult(i, std::move(t)); ctx->v[i] = std::move(t);
// release because the stored values in all threads must be visible below // release because the stored values in all threads must be visible below
// acquire because no stored value is permitted to be fetched early // acquire because no stored value is permitted to be fetched early
auto const s = 1 + ctx->stored.fetch_add(1, std::memory_order_acq_rel); auto const s = 1 + ctx->stored.fetch_add(1, std::memory_order_acq_rel);
if (s < n) { if (s < ctx->min) {
return; return;
} }
ctx->complete(); Result result;
result.reserve(ctx->completed.load());
for (size_t j = 0; j < ctx->v.size(); ++j) {
auto& entry = ctx->v[j];
if (entry.hasValue()) {
result.emplace_back(j, std::move(entry).value());
}
}
ctx->p.setTry(Try<Result>(std::move(result)));
}); });
} }
...@@ -1741,28 +1706,26 @@ Future<I> Future<T>::reduce(I&& initial, F&& func) { ...@@ -1741,28 +1706,26 @@ Future<I> Future<T>::reduce(I&& initial, F&& func) {
// unorderedReduce (iterator) // unorderedReduce (iterator)
// TODO(T26439406): Make return SemiFuture // TODO(T26439406): Make return SemiFuture
template <class It, class T, class F, class ItT, class Arg> template <class It, class T, class F>
Future<T> unorderedReduce(It first, It last, T initial, F func) { Future<T> unorderedReduce(It first, It last, T initial, F func) {
using ItF = typename std::iterator_traits<It>::value_type;
using ItT = typename ItF::value_type;
using Arg = MaybeTryArg<F, T, ItT>;
if (first == last) { if (first == last) {
return makeFuture(std::move(initial)); return makeFuture(std::move(initial));
} }
typedef isTry<Arg> IsTry; typedef isTry<Arg> IsTry;
struct UnorderedReduceContext { struct Context {
UnorderedReduceContext(T&& memo, F&& fn, size_t n) Context(T&& memo, F&& fn, size_t n)
: lock_(), memo_(makeFuture<T>(std::move(memo))), : lock_(),
func_(std::move(fn)), numThens_(0), numFutures_(n), promise_() memo_(makeFuture<T>(std::move(memo))),
{} func_(std::move(fn)),
numThens_(0),
static void fulfillWithValueOrFuture(Promise<T>&& p, T&& v) { numFutures_(n),
p.setValue(std::move(v)); promise_() {}
}
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_;
...@@ -1772,13 +1735,21 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) { ...@@ -1772,13 +1735,21 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) {
Promise<T> promise_; Promise<T> promise_;
}; };
auto ctx = std::make_shared<UnorderedReduceContext>( struct Fulfill {
std::move(initial), std::move(func), std::distance(first, last)); void operator()(Promise<T>&& p, T&& v) const {
p.setValue(std::move(v));
}
void operator()(Promise<T>&& p, Future<T>&& f) const {
f.setCallback_(
[p = std::move(p)](Try<T>&& t) mutable { p.setTry(std::move(t)); });
}
};
mapSetCallback<ItT>( auto ctx = std::make_shared<Context>(
first, std::move(initial), std::move(func), std::distance(first, last));
last, for (size_t i = 0; first != last; ++first, ++i) {
[ctx](size_t /* i */, Try<ItT>&& t) { first->setCallback_([i, ctx](Try<ItT>&& t) {
(void)i;
// Futures can be completed in any order, simultaneously. // Futures can be completed in any order, simultaneously.
// To make this non-blocking, we create a new Future chain in // To make this non-blocking, we create a new Future chain in
// the order of completion to reduce the values. // the order of completion to reduce the values.
...@@ -1795,11 +1766,11 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) { ...@@ -1795,11 +1766,11 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) {
[ctx](Try<T>&& t2) { ctx->promise_.setValue(std::move(t2)); }); [ctx](Try<T>&& t2) { ctx->promise_.setValue(std::move(t2)); });
} }
} }
f.setCallback_([ctx, mp = std::move(p), mt = std::move(t)]( f.setCallback_(
Try<T>&& v) mutable { [ctx, mp = std::move(p), mt = std::move(t)](Try<T>&& v) mutable {
if (v.hasValue()) { if (v.hasValue()) {
try { try {
ctx->fulfillWithValueOrFuture( Fulfill{}(
std::move(mp), std::move(mp),
ctx->func_( ctx->func_(
std::move(v.value()), std::move(v.value()),
...@@ -1814,7 +1785,7 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) { ...@@ -1814,7 +1785,7 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) {
} }
}); });
}); });
}
return ctx->promise_.getSemiFuture().via(&InlineExecutor::instance()); return ctx->promise_.getSemiFuture().via(&InlineExecutor::instance());
} }
......
...@@ -72,7 +72,6 @@ namespace futures { ...@@ -72,7 +72,6 @@ namespace futures {
namespace detail { namespace detail {
template <class> class Core; template <class> class Core;
template <class> struct CollectContext;
template <typename...> template <typename...>
struct ArgType; struct ArgType;
......
...@@ -307,8 +307,8 @@ Future<std::tuple<Try<typename remove_cvref_t<Fs>::value_type>...>> collectAll( ...@@ -307,8 +307,8 @@ Future<std::tuple<Try<typename remove_cvref_t<Fs>::value_type>...>> collectAll(
/// type of the returned Future is std::vector<T> instead of /// type of the returned Future is std::vector<T> instead of
/// std::vector<Try<T>> /// std::vector<Try<T>>
template <class InputIterator> template <class InputIterator>
Future<typename futures::detail::CollectContext<typename std::iterator_traits< Future<std::vector<
InputIterator>::value_type::value_type>::result_type> typename std::iterator_traits<InputIterator>::value_type::value_type>>
collect(InputIterator first, InputIterator last); collect(InputIterator first, InputIterator last);
/// Sugar for the most common case /// Sugar for the most common case
...@@ -435,12 +435,7 @@ auto reduce(Collection&& c, T&& initial, F&& func) ...@@ -435,12 +435,7 @@ auto reduce(Collection&& c, T&& initial, F&& func)
/** like reduce, but calls func on finished futures as they complete /** like reduce, but calls func on finished futures as they complete
does NOT keep the order of the input does NOT keep the order of the input
*/ */
template < template <class It, class T, class F>
class It,
class T,
class F,
class ItT = typename std::iterator_traits<It>::value_type::value_type,
class Arg = MaybeTryArg<F, T, ItT>>
Future<T> unorderedReduce(It first, It last, T initial, F func); Future<T> unorderedReduce(It first, It last, T initial, F func);
/// Sugar for the most common case /// Sugar for the most common 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