create unorderedReduceSemiFuture method

Summary:
create a unorderedReduceSemiFuture method that returns a SemiFuture
instead of just a Future.

Reviewed By: LeeHowes

Differential Revision: D16946582

fbshipit-source-id: e28b4cb055b3a9c67adebb373c60a775f7005762

folly/futures/Future-inl.h

@@ -1090,8 +1090,7 @@ Future<T>::thenError(tag_t<ExceptionType>, F&& func) && {
   auto* ePtr = this->getExecutor();
   auto e = folly::getKeepAliveToken(ePtr ? *ePtr : InlineExecutor::instance());
 
-  this->setCallback_(
-      [state = futures::detail::makeCoreCallbackState(
+  this->setCallback_([state = futures::detail::makeCoreCallbackState(
                           std::move(p), std::forward<F>(func))](
                          Executor::KeepAlive<>&& ka, Try<T>&& t) mutable {
     if (auto ex = t.template tryGetExceptionObject<
@@ -1872,9 +1871,8 @@ Future<I> Future<T>::reduce(I&& initial, F&& func) && {
 
 // unorderedReduce (iterator)
 
-// TODO(T26439406): Make return SemiFuture
 template <class It, class T, class F>
-Future<T> unorderedReduce(It first, It last, T initial, F func) {
+SemiFuture<T> unorderedReduceSemiFuture(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>;
@@ -1914,6 +1912,9 @@ Future<T> unorderedReduce(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>(
       std::move(initial), std::move(func), std::distance(first, last));
   for (size_t i = 0; first != last; ++first, ++i) {
@@ -1936,8 +1937,7 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) {
           });
         }
       }
-      f.setCallback_(
-          [ctx, mp = std::move(p), mt = std::move(t)](
+      f.setCallback_([ctx, mp = std::move(p), mt = std::move(t)](
                          Executor::KeepAlive<>&&, Try<T>&& v) mutable {
         if (v.hasValue()) {
           try {
@@ -1957,7 +1957,24 @@ Future<T> unorderedReduce(It first, It last, T initial, F func) {
       });
     });
   }
-  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

folly/futures/test/ReduceTest.cpp

@@ -139,9 +139,29 @@ TEST(Reduce, unorderedReduce) {
     p1.setValue(1);
     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) {
+  {
     Promise<int> p1;
     Promise<int> p2;
     Promise<int> p3;
@@ -159,6 +179,26 @@ TEST(Reduce, unorderedReduceException) {
     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;
+    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 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);
+  }
 }
 
 TEST(Reduce, unorderedReduceFuture) {