(Wangle) Reduce

Summary:
1] The lambda should ble able to return a `Try<T>`. Maybe? Can a `then`
return a `Try<T>` actually? Can fix this with `resultOf`. (Doubling the number of functions to 4.)

2] `initial` and `func` have to be copyable.

Test Plan: Added tests.

Reviewed By: hans@fb.com

Subscribers: trunkagent, folly-diffs@, jsedgwick, yfeldblum

FB internal diff: D1870996

Tasks: 6025252

Signature: t1:1870996:1427318511:2ae5894b79022da88990835b26d35c4520fdbd29

folly/futures/Future-inl.h

@@ -589,6 +589,53 @@ whenN(InputIterator first, InputIterator last, size_t n) {
   return ctx->p.getFuture();
 }
 
+template <class It, class T, class F, class ItT, class Arg>
+typename std::enable_if<!isFutureResult<F, T, Arg>::value, Future<T>>::type
+reduce(It first, It last, T initial, F func) {
+  if (first == last) {
+    return makeFuture(std::move(initial));
+  }
+
+  typedef isTry<Arg> IsTry;
+
+  return whenAll(first, last)
+    .then([initial, func](std::vector<Try<ItT>>& vals) mutable {
+      for (auto& val : vals) {
+        initial = func(std::move(initial),
+                       // Either return a ItT&& or a Try<ItT>&& depending
+                       // on the type of the argument of func.
+                       val.template get<IsTry::value, Arg&&>());
+      }
+      return initial;
+    });
+}
+
+template <class It, class T, class F, class ItT, class Arg>
+typename std::enable_if<isFutureResult<F, T, Arg>::value, Future<T>>::type
+reduce(It first, It last, T initial, F func) {
+  if (first == last) {
+    return makeFuture(std::move(initial));
+  }
+
+  typedef isTry<Arg> IsTry;
+
+  auto f = first->then([initial, func](Try<ItT>& head) mutable {
+    return func(std::move(initial),
+                head.template get<IsTry::value, Arg&&>());
+  });
+
+  for (++first; first != last; ++first) {
+    f = whenAll(f, *first).then([func](std::tuple<Try<T>, Try<ItT>>& t) {
+      return func(std::move(std::get<0>(t).value()),
+                  // Either return a ItT&& or a Try<ItT>&& depending
+                  // on the type of the argument of func.
+                  std::get<1>(t).template get<IsTry::value, Arg&&>());
+    });
+  }
+
+  return f;
+}
+
 template <class T>
 Future<T> Future<T>::within(Duration dur, Timekeeper* tk) {
   return within(dur, TimedOut(), tk);

folly/futures/Future.h

@@ -622,6 +622,32 @@ Future<std::vector<std::pair<
   Try<typename std::iterator_traits<InputIterator>::value_type::value_type>>>>
 whenN(InputIterator first, InputIterator last, size_t n);
 
+template <typename F, typename T, typename ItT>
+using MaybeTryArg = typename std::conditional<
+  detail::callableWith<F, T&&, Try<ItT>&&>::value, Try<ItT>, ItT>::type;
+
+template<typename F, typename T, typename Arg>
+using isFutureResult = isFuture<typename std::result_of<F(T&&, Arg&&)>::type>;
+
+/** repeatedly calls func on every result, e.g.
+    reduce(reduce(reduce(T initial, result of first), result of second), ...)
+
+    The type of the final result is a Future of the type of the initial value.
+
+    Func can either return a T, or a Future<T>
+  */
+template <class It, class T, class F,
+          class ItT = typename std::iterator_traits<It>::value_type::value_type,
+          class Arg = MaybeTryArg<F, T, ItT>>
+typename std::enable_if<!isFutureResult<F, T, Arg>::value, Future<T>>::type
+reduce(It first, It last, T initial, F func);
+
+template <class It, class T, class F,
+          class ItT = typename std::iterator_traits<It>::value_type::value_type,
+          class Arg = MaybeTryArg<F, T, ItT>>
+typename std::enable_if<isFutureResult<F, T, Arg>::value, Future<T>>::type
+reduce(It first, It last, T initial, F func);
+
 } // folly
 
 #include <folly/futures/Future-inl.h>

folly/futures/test/FutureTest.cpp

@@ -1442,3 +1442,79 @@ TEST(Future, Unwrap_FutureNotReady) {
   ASSERT_TRUE(unwrapped.isReady());
   EXPECT_EQ(5484, unwrapped.value());
 }
+
+TEST(Reduce, Basic) {
+  auto makeFutures = [](int count) {
+    std::vector<Future<int>> fs;
+    for (int i = 1; i <= count; ++i) {
+      fs.emplace_back(makeFuture(i));
+    }
+    return fs;
+  };
+
+  // Empty (Try)
+  {
+    auto fs = makeFutures(0);
+
+    Future<double> f1 = reduce(fs.begin(), fs.end(), 1.2,
+      [](double a, Try<int>&& b){
+        return a + *b + 0.1;
+      });
+    EXPECT_EQ(1.2, f1.get());
+  }
+
+  // One (Try)
+  {
+    auto fs = makeFutures(1);
+
+    Future<double> f1 = reduce(fs.begin(), fs.end(), 0.0,
+      [](double a, Try<int>&& b){
+        return a + *b + 0.1;
+      });
+    EXPECT_EQ(1.1, f1.get());
+  }
+
+  // Returning values (Try)
+  {
+    auto fs = makeFutures(3);
+
+    Future<double> f1 = reduce(fs.begin(), fs.end(), 0.0,
+      [](double a, Try<int>&& b){
+        return a + *b + 0.1;
+      });
+    EXPECT_EQ(6.3, f1.get());
+  }
+
+  // Returning values
+  {
+    auto fs = makeFutures(3);
+
+    Future<double> f1 = reduce(fs.begin(), fs.end(), 0.0,
+      [](double a, int&& b){
+        return a + b + 0.1;
+      });
+    EXPECT_EQ(6.3, f1.get());
+  }
+
+  // Returning futures (Try)
+  {
+    auto fs = makeFutures(3);
+
+    Future<double> f2 = reduce(fs.begin(), fs.end(), 0.0,
+      [](double a, Try<int>&& b){
+        return makeFuture<double>(a + *b + 0.1);
+      });
+    EXPECT_EQ(6.3, f2.get());
+  }
+
+  // Returning futures
+  {
+    auto fs = makeFutures(3);
+
+    Future<double> f2 = reduce(fs.begin(), fs.end(), 0.0,
+      [](double a, int&& b){
+        return makeFuture<double>(a + b + 0.1);
+      });
+    EXPECT_EQ(6.3, f2.get());
+  }
+}