(Wangle) Simplify reduce, use the same implementation for func returning T/Future<T>

Summary:
I wanted to use `collectAll` for `reduce` if `func` does not return a
Future, because the overhead seemed smaller, but it has been suggested
that running the callback as soon as possible might be better. Not sure
which is. Note that this also makes n copies of the lambda and moves the
value at least n times.

I also plan to add a `streamingReduce` which calls `func` as soon as
results come in (but out of order).

Test Plan: Run all the tests.

Reviewed By: hans@fb.com

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

FB internal diff: D2015316

Tasks: 6025252

Signature: t1:2015316:1430349611:4f62a540ded85299a22670dd2add27cf1458e8f8

folly/futures/Future-inl.h

@@ -782,44 +782,28 @@ collectN(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) {
+template <class It, class T, class F>
+Future<T> reduce(It first, It last, T&& initial, F&& func) {
   if (first == last) {
     return makeFuture(std::move(initial));
   }
 
+  typedef typename std::iterator_traits<It>::value_type::value_type ItT;
+  typedef typename std::conditional<
+    detail::callableWith<F, T&&, Try<ItT>&&>::value, Try<ItT>, ItT>::type Arg;
   typedef isTry<Arg> IsTry;
 
-  return collectAll(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;
-    });
-}
+  folly::MoveWrapper<T> minitial(std::move(initial));
+  auto sfunc = std::make_shared<F>(std::move(func));
 
-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),
+  auto f = first->then([minitial, sfunc](Try<ItT>& head) mutable {
+    return (*sfunc)(std::move(*minitial),
                 head.template get<IsTry::value, Arg&&>());
   });
 
   for (++first; first != last; ++first) {
-    f = collectAll(f, *first).then([func](std::tuple<Try<T>, Try<ItT>>& t) {
-      return func(std::move(std::get<0>(t).value()),
+    f = collectAll(f, *first).then([sfunc](std::tuple<Try<T>, Try<ItT>>& t) {
+      return (*sfunc)(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&&>());

folly/futures/Future.h

@@ -407,7 +407,6 @@ class Future {
   void setExecutor(Executor* x) { core_->setExecutor(x); }
 };
 
-
 } // folly
 
 #include <folly/futures/Future-inl.h>

folly/futures/helpers.h

@@ -235,22 +235,14 @@ using isFutureResult = isFuture<typename std::result_of<F(T&&, Arg&&)>::type>;
 
     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);
+template <class It, class T, class F>
+Future<T> reduce(It first, It last, T&& initial, F&& func);
 
 /// Sugar for the most common case
 template <class Collection, class T, class F>
 auto reduce(Collection&& c, T&& initial, F&& func)
-    -> decltype(reduce(c.begin(), c.end(), initial, func)) {
+    -> decltype(reduce(c.begin(), c.end(), std::forward<T>(initial),
+                std::forward<F>(func))) {
   return reduce(
       c.begin(),
       c.end(),