Commit d7808b52 authored by Amol Bhave's avatar Amol Bhave Committed by Facebook Github Bot

Back out "[folly] create unorderedReduceSemiFuture method"

Summary: Original commit changeset: e28b4cb055b3

Reviewed By: yfeldblum

Differential Revision: D17429976

fbshipit-source-id: e2f6f59c22854f1b2fd25913ad779f4edc83f8ee
parent 92dd0f71
...@@ -1090,25 +1090,26 @@ Future<T>::thenError(tag_t<ExceptionType>, F&& func) && { ...@@ -1090,25 +1090,26 @@ Future<T>::thenError(tag_t<ExceptionType>, F&& func) && {
auto* ePtr = this->getExecutor(); auto* ePtr = this->getExecutor();
auto e = folly::getKeepAliveToken(ePtr ? *ePtr : InlineExecutor::instance()); auto e = folly::getKeepAliveToken(ePtr ? *ePtr : InlineExecutor::instance());
this->setCallback_([state = futures::detail::makeCoreCallbackState( this->setCallback_(
std::move(p), std::forward<F>(func))]( [state = futures::detail::makeCoreCallbackState(
Executor::KeepAlive<>&& ka, Try<T>&& t) mutable { std::move(p), std::forward<F>(func))](
if (auto ex = t.template tryGetExceptionObject< Executor::KeepAlive<>&& ka, Try<T>&& t) mutable {
std::remove_reference_t<ExceptionType>>()) { if (auto ex = t.template tryGetExceptionObject<
auto tf2 = state.tryInvoke(std::move(*ex)); std::remove_reference_t<ExceptionType>>()) {
if (tf2.hasException()) { auto tf2 = state.tryInvoke(std::move(*ex));
state.setException(std::move(ka), std::move(tf2.exception())); if (tf2.hasException()) {
} else { state.setException(std::move(ka), std::move(tf2.exception()));
tf2->setCallback_( } else {
[p = state.stealPromise()]( tf2->setCallback_(
Executor::KeepAlive<>&& innerKA, Try<T>&& t3) mutable { [p = state.stealPromise()](
p.setTry(std::move(innerKA), std::move(t3)); Executor::KeepAlive<>&& innerKA, Try<T>&& t3) mutable {
}); p.setTry(std::move(innerKA), std::move(t3));
} });
} else { }
state.setTry(std::move(ka), std::move(t)); } else {
} state.setTry(std::move(ka), std::move(t));
}); }
});
return std::move(sf).via(std::move(e)); return std::move(sf).via(std::move(e));
} }
...@@ -1871,8 +1872,9 @@ Future<I> Future<T>::reduce(I&& initial, F&& func) && { ...@@ -1871,8 +1872,9 @@ Future<I> Future<T>::reduce(I&& initial, F&& func) && {
// unorderedReduce (iterator) // unorderedReduce (iterator)
// TODO(T26439406): Make return SemiFuture
template <class It, class T, class F> template <class It, class T, class F>
SemiFuture<T> unorderedReduceSemiFuture(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 ItF = typename std::iterator_traits<It>::value_type;
using ItT = typename ItF::value_type; using ItT = typename ItF::value_type;
using Arg = MaybeTryArg<F, T, ItT>; using Arg = MaybeTryArg<F, T, ItT>;
...@@ -1912,9 +1914,6 @@ SemiFuture<T> unorderedReduceSemiFuture(It first, It last, T initial, F func) { ...@@ -1912,9 +1914,6 @@ SemiFuture<T> unorderedReduceSemiFuture(It first, It last, T initial, F func) {
} }
}; };
std::vector<futures::detail::DeferredWrapper> executors;
futures::detail::stealDeferredExecutors(executors, first, last);
auto ctx = std::make_shared<Context>( auto ctx = std::make_shared<Context>(
std::move(initial), std::move(func), std::distance(first, last)); std::move(initial), std::move(func), std::distance(first, last));
for (size_t i = 0; first != last; ++first, ++i) { for (size_t i = 0; first != last; ++first, ++i) {
...@@ -1937,44 +1936,28 @@ SemiFuture<T> unorderedReduceSemiFuture(It first, It last, T initial, F func) { ...@@ -1937,44 +1936,28 @@ SemiFuture<T> unorderedReduceSemiFuture(It first, It last, T initial, F func) {
}); });
} }
} }
f.setCallback_([ctx, mp = std::move(p), mt = std::move(t)]( f.setCallback_(
Executor::KeepAlive<>&&, Try<T>&& v) mutable { [ctx, mp = std::move(p), mt = std::move(t)](
if (v.hasValue()) { Executor::KeepAlive<>&&, Try<T>&& v) mutable {
try { if (v.hasValue()) {
Fulfill{}( try {
std::move(mp), Fulfill{}(
ctx->func_( std::move(mp),
std::move(v.value()), ctx->func_(
mt.template get<IsTry::value, Arg&&>())); std::move(v.value()),
} catch (std::exception& e) { mt.template get<IsTry::value, Arg&&>()));
mp.setException(exception_wrapper(std::current_exception(), e)); } catch (std::exception& e) {
} catch (...) { mp.setException(exception_wrapper(std::current_exception(), e));
mp.setException(exception_wrapper(std::current_exception())); } catch (...) {
} mp.setException(exception_wrapper(std::current_exception()));
} else { }
mp.setTry(std::move(v)); } else {
} mp.setTry(std::move(v));
}); }
});
}); });
} }
return ctx->promise_.getSemiFuture().via(&InlineExecutor::instance());
auto future = ctx->promise_.getSemiFuture();
if (!executors.empty()) {
future = std::move(future).defer(
[](Try<typename decltype(future)::value_type>&& t) {
return std::move(t).value();
});
const auto& deferredExecutor = futures::detail::getDeferredExecutor(future);
deferredExecutor->setNestedExecutors(std::move(executors));
}
return future;
}
template <class It, class T, class F>
Future<T> unorderedReduce(It first, It last, T initial, F func) {
return unorderedReduceSemiFuture(
first, last, std::move(initial), std::move(func))
.via(&InlineExecutor::instance());
} }
// within // within
......
...@@ -139,66 +139,26 @@ TEST(Reduce, unorderedReduce) { ...@@ -139,66 +139,26 @@ TEST(Reduce, unorderedReduce) {
p1.setValue(1); p1.setValue(1);
EXPECT_EQ(1.0, std::move(f).get()); EXPECT_EQ(1.0, std::move(f).get());
} }
{
Promise<int> p1;
Promise<int> p2;
Promise<int> p3;
std::vector<SemiFuture<int>> fs;
fs.push_back(p1.getSemiFuture());
fs.push_back(p2.getSemiFuture());
fs.push_back(p3.getSemiFuture());
SemiFuture<double> f = unorderedReduceSemiFuture(
fs.begin(), fs.end(), 0.0, [](double /* a */, int&& b) {
return double(b);
});
p3.setValue(3);
p2.setValue(2);
p1.setValue(1);
EXPECT_EQ(1.0, std::move(f).get());
}
} }
TEST(Reduce, unorderedReduceException) { TEST(Reduce, unorderedReduceException) {
{ Promise<int> p1;
Promise<int> p1; Promise<int> p2;
Promise<int> p2; Promise<int> p3;
Promise<int> p3;
std::vector<Future<int>> fs;
fs.push_back(p1.getFuture());
fs.push_back(p2.getFuture());
fs.push_back(p3.getFuture());
Future<double> f =
unorderedReduce(fs.begin(), fs.end(), 0.0, [](double /* a */, int&& b) {
return b + 0.0;
});
p3.setValue(3);
p2.setException(exception_wrapper(std::runtime_error("blah")));
p1.setValue(1);
EXPECT_THROW(std::move(f).get(), std::runtime_error);
}
{
Promise<int> p1;
Promise<int> p2;
Promise<int> p3;
std::vector<SemiFuture<int>> fs; std::vector<Future<int>> fs;
fs.push_back(p1.getSemiFuture()); fs.push_back(p1.getFuture());
fs.push_back(p2.getSemiFuture()); fs.push_back(p2.getFuture());
fs.push_back(p3.getSemiFuture()); fs.push_back(p3.getFuture());
SemiFuture<double> f = unorderedReduceSemiFuture( Future<double> f =
fs.begin(), fs.end(), 0.0, [](double /* a */, int&& b) { unorderedReduce(fs.begin(), fs.end(), 0.0, [](double /* a */, int&& b) {
return b + 0.0; return b + 0.0;
}); });
p3.setValue(3); p3.setValue(3);
p2.setException(exception_wrapper(std::runtime_error("blah"))); p2.setException(exception_wrapper(std::runtime_error("blah")));
p1.setValue(1); p1.setValue(1);
EXPECT_THROW(std::move(f).get(), std::runtime_error); EXPECT_THROW(std::move(f).get(), std::runtime_error);
}
} }
TEST(Reduce, unorderedReduceFuture) { TEST(Reduce, unorderedReduceFuture) {
......
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