folly/futures/Future-inl.h

   1 /*
   2  * Copyright 2015 Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #pragma once
  18
  19 #include <chrono>
  20 #include <thread>
  21
  22 #include <folly/experimental/fibers/Baton.h>
  23 #include <folly/Optional.h>
  24 #include <folly/futures/detail/Core.h>
  25 #include <folly/futures/Timekeeper.h>
  26
  27 namespace folly {
  28
  29 class Timekeeper;
  30
  31 namespace detail {
  32   Timekeeper* getTimekeeperSingleton();
  33 }
  34
  35 template <class T>
  36 Future<T>::Future(Future<T>&& other) noexcept : core_(other.core_) {
  37   other.core_ = nullptr;
  38 }
  39
  40 template <class T>
  41 Future<T>& Future<T>::operator=(Future<T>&& other) noexcept {
  42   std::swap(core_, other.core_);
  43   return *this;
  44 }
  45
  46 template <class T>
  47 template <class T2>
  48 Future<T>::Future(T2&& val) : core_(nullptr) {
  49   Promise<T> p;
  50   p.setValue(std::forward<T2>(val));
  51   *this = p.getFuture();
  52 }
  53
  54 template <class T>
  55 template <class T2,
  56           typename std::enable_if<
  57             folly::is_void_or_unit<T2>::value,
  58             int>::type>
  59 Future<T>::Future() : core_(nullptr) {
  60   Promise<T> p;
  61   p.setValue();
  62   *this = p.getFuture();
  63 }
  64
  65
  66 template <class T>
  67 Future<T>::~Future() {
  68   detach();
  69 }
  70
  71 template <class T>
  72 void Future<T>::detach() {
  73   if (core_) {
  74     core_->detachFuture();
  75     core_ = nullptr;
  76   }
  77 }
  78
  79 template <class T>
  80 void Future<T>::throwIfInvalid() const {
  81   if (!core_)
  82     throw NoState();
  83 }
  84
  85 template <class T>
  86 template <class F>
  87 void Future<T>::setCallback_(F&& func) {
  88   throwIfInvalid();
  89   core_->setCallback(std::move(func));
  90 }
  91
  92 // unwrap
  93
  94 template <class T>
  95 template <class F>
  96 typename std::enable_if<isFuture<F>::value,
  97                         Future<typename isFuture<T>::Inner>>::type
  98 Future<T>::unwrap() {
  99   return then([](Future<typename isFuture<T>::Inner> internal_future) {
 100       return internal_future;
 101   });
 102 }
 103
 104 // then
 105
 106 // Variant: returns a value
 107 // e.g. f.then([](Try<T>&& t){ return t.value(); });
 108 template <class T>
 109 template <typename F, typename R, bool isTry, typename... Args>
 110 typename std::enable_if<!R::ReturnsFuture::value, typename R::Return>::type
 111 Future<T>::thenImplementation(F func, detail::argResult<isTry, F, Args...>) {
 112   static_assert(sizeof...(Args) <= 1, "Then must take zero/one argument");
 113   typedef typename R::ReturnsFuture::Inner B;
 114
 115   throwIfInvalid();
 116
 117   // wrap these so we can move them into the lambda
 118   folly::MoveWrapper<Promise<B>> p;
 119   folly::MoveWrapper<F> funcm(std::forward<F>(func));
 120
 121   // grab the Future now before we lose our handle on the Promise
 122   auto f = p->getFuture();
 123   if (getExecutor()) {
 124     f.setExecutor(getExecutor());
 125   }
 126
 127   /* This is a bit tricky.
 128
 129      We can't just close over *this in case this Future gets moved. So we
 130      make a new dummy Future. We could figure out something more
 131      sophisticated that avoids making a new Future object when it can, as an
 132      optimization. But this is correct.
 133
 134      core_ can't be moved, it is explicitly disallowed (as is copying). But
 135      if there's ever a reason to allow it, this is one place that makes that
 136      assumption and would need to be fixed. We use a standard shared pointer
 137      for core_ (by copying it in), which means in essence obj holds a shared
 138      pointer to itself.  But this shouldn't leak because Promise will not
 139      outlive the continuation, because Promise will setException() with a
 140      broken Promise if it is destructed before completed. We could use a
 141      weak pointer but it would have to be converted to a shared pointer when
 142      func is executed (because the Future returned by func may possibly
 143      persist beyond the callback, if it gets moved), and so it is an
 144      optimization to just make it shared from the get-go.
 145
 146      We have to move in the Promise and func using the MoveWrapper
 147      hack. (func could be copied but it's a big drag on perf).
 148
 149      Two subtle but important points about this design. detail::Core has no
 150      back pointers to Future or Promise, so if Future or Promise get moved
 151      (and they will be moved in performant code) we don't have to do
 152      anything fancy. And because we store the continuation in the
 153      detail::Core, not in the Future, we can execute the continuation even
 154      after the Future has gone out of scope. This is an intentional design
 155      decision. It is likely we will want to be able to cancel a continuation
 156      in some circumstances, but I think it should be explicit not implicit
 157      in the destruction of the Future used to create it.
 158      */
 159   setCallback_(
 160     [p, funcm](Try<T>&& t) mutable {
 161       if (!isTry && t.hasException()) {
 162         p->setException(std::move(t.exception()));
 163       } else {
 164         p->setWith([&]() {
 165           return (*funcm)(t.template get<isTry, Args>()...);
 166         });
 167       }
 168     });
 169
 170   return f;
 171 }
 172
 173 // Variant: returns a Future
 174 // e.g. f.then([](T&& t){ return makeFuture<T>(t); });
 175 template <class T>
 176 template <typename F, typename R, bool isTry, typename... Args>
 177 typename std::enable_if<R::ReturnsFuture::value, typename R::Return>::type
 178 Future<T>::thenImplementation(F func, detail::argResult<isTry, F, Args...>) {
 179   static_assert(sizeof...(Args) <= 1, "Then must take zero/one argument");
 180   typedef typename R::ReturnsFuture::Inner B;
 181
 182   throwIfInvalid();
 183
 184   // wrap these so we can move them into the lambda
 185   folly::MoveWrapper<Promise<B>> p;
 186   folly::MoveWrapper<F> funcm(std::forward<F>(func));
 187
 188   // grab the Future now before we lose our handle on the Promise
 189   auto f = p->getFuture();
 190   if (getExecutor()) {
 191     f.setExecutor(getExecutor());
 192   }
 193
 194   setCallback_(
 195     [p, funcm](Try<T>&& t) mutable {
 196       if (!isTry && t.hasException()) {
 197         p->setException(std::move(t.exception()));
 198       } else {
 199         try {
 200           auto f2 = (*funcm)(t.template get<isTry, Args>()...);
 201           // that didn't throw, now we can steal p
 202           f2.setCallback_([p](Try<B>&& b) mutable {
 203             p->setTry(std::move(b));
 204           });
 205         } catch (const std::exception& e) {
 206           p->setException(exception_wrapper(std::current_exception(), e));
 207         } catch (...) {
 208           p->setException(exception_wrapper(std::current_exception()));
 209         }
 210       }
 211     });
 212
 213   return f;
 214 }
 215
 216 template <typename T>
 217 template <typename R, typename Caller, typename... Args>
 218   Future<typename isFuture<R>::Inner>
 219 Future<T>::then(R(Caller::*func)(Args...), Caller *instance) {
 220   typedef typename std::remove_cv<
 221     typename std::remove_reference<
 222       typename detail::ArgType<Args...>::FirstArg>::type>::type FirstArg;
 223   return then([instance, func](Try<T>&& t){
 224     return (instance->*func)(t.template get<isTry<FirstArg>::value, Args>()...);
 225   });
 226 }
 227
 228 // TODO(6838553)
 229 #ifndef __clang__
 230 template <class T>
 231 template <class... Args>
 232 auto Future<T>::then(Executor* x, Args&&... args)
 233   -> decltype(this->then(std::forward<Args>(args)...))
 234 {
 235   auto oldX = getExecutor();
 236   setExecutor(x);
 237   return this->then(std::forward<Args>(args)...).via(oldX);
 238 }
 239 #endif
 240
 241 template <class T>
 242 Future<void> Future<T>::then() {
 243   return then([] (Try<T>&& t) {});
 244 }
 245
 246 // onError where the callback returns T
 247 template <class T>
 248 template <class F>
 249 typename std::enable_if<
 250   !detail::callableWith<F, exception_wrapper>::value &&
 251   !detail::Extract<F>::ReturnsFuture::value,
 252   Future<T>>::type
 253 Future<T>::onError(F&& func) {
 254   typedef typename detail::Extract<F>::FirstArg Exn;
 255   static_assert(
 256       std::is_same<typename detail::Extract<F>::RawReturn, T>::value,
 257       "Return type of onError callback must be T or Future<T>");
 258
 259   Promise<T> p;
 260   auto f = p.getFuture();
 261   auto pm = folly::makeMoveWrapper(std::move(p));
 262   auto funcm = folly::makeMoveWrapper(std::move(func));
 263   setCallback_([pm, funcm](Try<T>&& t) mutable {
 264     if (!t.template withException<Exn>([&] (Exn& e) {
 265           pm->setWith([&]{
 266             return (*funcm)(e);
 267           });
 268         })) {
 269       pm->setTry(std::move(t));
 270     }
 271   });
 272
 273   return f;
 274 }
 275
 276 // onError where the callback returns Future<T>
 277 template <class T>
 278 template <class F>
 279 typename std::enable_if<
 280   !detail::callableWith<F, exception_wrapper>::value &&
 281   detail::Extract<F>::ReturnsFuture::value,
 282   Future<T>>::type
 283 Future<T>::onError(F&& func) {
 284   static_assert(
 285       std::is_same<typename detail::Extract<F>::Return, Future<T>>::value,
 286       "Return type of onError callback must be T or Future<T>");
 287   typedef typename detail::Extract<F>::FirstArg Exn;
 288
 289   Promise<T> p;
 290   auto f = p.getFuture();
 291   auto pm = folly::makeMoveWrapper(std::move(p));
 292   auto funcm = folly::makeMoveWrapper(std::move(func));
 293   setCallback_([pm, funcm](Try<T>&& t) mutable {
 294     if (!t.template withException<Exn>([&] (Exn& e) {
 295           try {
 296             auto f2 = (*funcm)(e);
 297             f2.setCallback_([pm](Try<T>&& t2) mutable {
 298               pm->setTry(std::move(t2));
 299             });
 300           } catch (const std::exception& e2) {
 301             pm->setException(exception_wrapper(std::current_exception(), e2));
 302           } catch (...) {
 303             pm->setException(exception_wrapper(std::current_exception()));
 304           }
 305         })) {
 306       pm->setTry(std::move(t));
 307     }
 308   });
 309
 310   return f;
 311 }
 312
 313 template <class T>
 314 template <class F>
 315 Future<T> Future<T>::ensure(F func) {
 316   MoveWrapper<F> funcw(std::move(func));
 317   return this->then([funcw](Try<T>&& t) {
 318     (*funcw)();
 319     return makeFuture(std::move(t));
 320   });
 321 }
 322
 323 template <class T>
 324 template <class F>
 325 Future<T> Future<T>::onTimeout(Duration dur, F&& func, Timekeeper* tk) {
 326   auto funcw = folly::makeMoveWrapper(std::forward<F>(func));
 327   return within(dur, tk)
 328     .onError([funcw](TimedOut const&) { return (*funcw)(); });
 329 }
 330
 331 template <class T>
 332 template <class F>
 333 typename std::enable_if<
 334   detail::callableWith<F, exception_wrapper>::value &&
 335   detail::Extract<F>::ReturnsFuture::value,
 336   Future<T>>::type
 337 Future<T>::onError(F&& func) {
 338   static_assert(
 339       std::is_same<typename detail::Extract<F>::Return, Future<T>>::value,
 340       "Return type of onError callback must be T or Future<T>");
 341
 342   Promise<T> p;
 343   auto f = p.getFuture();
 344   auto pm = folly::makeMoveWrapper(std::move(p));
 345   auto funcm = folly::makeMoveWrapper(std::move(func));
 346   setCallback_([pm, funcm](Try<T> t) mutable {
 347     if (t.hasException()) {
 348       try {
 349         auto f2 = (*funcm)(std::move(t.exception()));
 350         f2.setCallback_([pm](Try<T> t2) mutable {
 351           pm->setTry(std::move(t2));
 352         });
 353       } catch (const std::exception& e2) {
 354         pm->setException(exception_wrapper(std::current_exception(), e2));
 355       } catch (...) {
 356         pm->setException(exception_wrapper(std::current_exception()));
 357       }
 358     } else {
 359       pm->setTry(std::move(t));
 360     }
 361   });
 362
 363   return f;
 364 }
 365
 366 // onError(exception_wrapper) that returns T
 367 template <class T>
 368 template <class F>
 369 typename std::enable_if<
 370   detail::callableWith<F, exception_wrapper>::value &&
 371   !detail::Extract<F>::ReturnsFuture::value,
 372   Future<T>>::type
 373 Future<T>::onError(F&& func) {
 374   static_assert(
 375       std::is_same<typename detail::Extract<F>::Return, Future<T>>::value,
 376       "Return type of onError callback must be T or Future<T>");
 377
 378   Promise<T> p;
 379   auto f = p.getFuture();
 380   auto pm = folly::makeMoveWrapper(std::move(p));
 381   auto funcm = folly::makeMoveWrapper(std::move(func));
 382   setCallback_([pm, funcm](Try<T> t) mutable {
 383     if (t.hasException()) {
 384       pm->setWith([&]{
 385         return (*funcm)(std::move(t.exception()));
 386       });
 387     } else {
 388       pm->setTry(std::move(t));
 389     }
 390   });
 391
 392   return f;
 393 }
 394
 395 template <class T>
 396 typename std::add_lvalue_reference<T>::type Future<T>::value() {
 397   throwIfInvalid();
 398
 399   return core_->getTry().value();
 400 }
 401
 402 template <class T>
 403 typename std::add_lvalue_reference<const T>::type Future<T>::value() const {
 404   throwIfInvalid();
 405
 406   return core_->getTry().value();
 407 }
 408
 409 template <class T>
 410 Try<T>& Future<T>::getTry() {
 411   throwIfInvalid();
 412
 413   return core_->getTry();
 414 }
 415
 416 template <class T>
 417 Optional<Try<T>> Future<T>::poll() {
 418   Optional<Try<T>> o;
 419   if (core_->ready()) {
 420     o = std::move(core_->getTry());
 421   }
 422   return o;
 423 }
 424
 425 template <class T>
 426 template <typename Executor>
 427 inline Future<T> Future<T>::via(Executor* executor) && {
 428   throwIfInvalid();
 429
 430   setExecutor(executor);
 431
 432   return std::move(*this);
 433 }
 434
 435 template <class T>
 436 template <typename Executor>
 437 inline Future<T> Future<T>::via(Executor* executor) & {
 438   throwIfInvalid();
 439
 440   MoveWrapper<Promise<T>> p;
 441   auto f = p->getFuture();
 442   then([p](Try<T>&& t) mutable { p->setTry(std::move(t)); });
 443   return std::move(f).via(executor);
 444 }
 445
 446 template <class T>
 447 bool Future<T>::isReady() const {
 448   throwIfInvalid();
 449   return core_->ready();
 450 }
 451
 452 template <class T>
 453 void Future<T>::raise(exception_wrapper exception) {
 454   core_->raise(std::move(exception));
 455 }
 456
 457 // makeFuture
 458
 459 template <class T>
 460 Future<typename std::decay<T>::type> makeFuture(T&& t) {
 461   Promise<typename std::decay<T>::type> p;
 462   p.setValue(std::forward<T>(t));
 463   return p.getFuture();
 464 }
 465
 466 inline // for multiple translation units
 467 Future<void> makeFuture() {
 468   Promise<void> p;
 469   p.setValue();
 470   return p.getFuture();
 471 }
 472
 473 template <class F>
 474 auto makeFutureWith(
 475     F&& func,
 476     typename std::enable_if<!std::is_reference<F>::value, bool>::type sdf)
 477     -> Future<decltype(func())> {
 478   Promise<decltype(func())> p;
 479   p.setWith(
 480     [&func]() {
 481       return (func)();
 482     });
 483   return p.getFuture();
 484 }
 485
 486 template <class F>
 487 auto makeFutureWith(F const& func) -> Future<decltype(func())> {
 488   F copy = func;
 489   return makeFutureWith(std::move(copy));
 490 }
 491
 492 template <class T>
 493 Future<T> makeFuture(std::exception_ptr const& e) {
 494   Promise<T> p;
 495   p.setException(e);
 496   return p.getFuture();
 497 }
 498
 499 template <class T>
 500 Future<T> makeFuture(exception_wrapper ew) {
 501   Promise<T> p;
 502   p.setException(std::move(ew));
 503   return p.getFuture();
 504 }
 505
 506 template <class T, class E>
 507 typename std::enable_if<std::is_base_of<std::exception, E>::value,
 508                         Future<T>>::type
 509 makeFuture(E const& e) {
 510   Promise<T> p;
 511   p.setException(make_exception_wrapper<E>(e));
 512   return p.getFuture();
 513 }
 514
 515 template <class T>
 516 Future<T> makeFuture(Try<T>&& t) {
 517   Promise<typename std::decay<T>::type> p;
 518   p.setTry(std::move(t));
 519   return p.getFuture();
 520 }
 521
 522 template <>
 523 inline Future<void> makeFuture(Try<void>&& t) {
 524   if (t.hasException()) {
 525     return makeFuture<void>(std::move(t.exception()));
 526   } else {
 527     return makeFuture();
 528   }
 529 }
 530
 531 // via
 532 template <typename Executor>
 533 Future<void> via(Executor* executor) {
 534   return makeFuture().via(executor);
 535 }
 536
 537 // when (variadic)
 538
 539 template <typename... Fs>
 540 typename detail::VariadicContext<
 541   typename std::decay<Fs>::type::value_type...>::type
 542 collectAll(Fs&&... fs) {
 543   auto ctx =
 544     new detail::VariadicContext<typename std::decay<Fs>::type::value_type...>();
 545   ctx->total = sizeof...(fs);
 546   auto f_saved = ctx->p.getFuture();
 547   detail::collectAllVariadicHelper(ctx,
 548     std::forward<typename std::decay<Fs>::type>(fs)...);
 549   return f_saved;
 550 }
 551
 552 // when (iterator)
 553
 554 template <class InputIterator>
 555 Future<
 556   std::vector<
 557   Try<typename std::iterator_traits<InputIterator>::value_type::value_type>>>
 558 collectAll(InputIterator first, InputIterator last) {
 559   typedef
 560     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 561
 562   if (first >= last) {
 563     return makeFuture(std::vector<Try<T>>());
 564   }
 565   size_t n = std::distance(first, last);
 566
 567   auto ctx = new detail::WhenAllContext<T>();
 568
 569   ctx->results.resize(n);
 570
 571   auto f_saved = ctx->p.getFuture();
 572
 573   for (size_t i = 0; first != last; ++first, ++i) {
 574      assert(i < n);
 575      auto& f = *first;
 576      f.setCallback_([ctx, i, n](Try<T> t) {
 577        ctx->results[i] = std::move(t);
 578        if (++ctx->count == n) {
 579          ctx->p.setValue(std::move(ctx->results));
 580          delete ctx;
 581        }
 582      });
 583   }
 584
 585   return f_saved;
 586 }
 587
 588 namespace detail {
 589
 590 template <class, class, typename = void> struct CollectContextHelper;
 591
 592 template <class T, class VecT>
 593 struct CollectContextHelper<T, VecT,
 594     typename std::enable_if<std::is_same<T, VecT>::value>::type> {
 595   static inline std::vector<T>&& getResults(std::vector<VecT>& results) {
 596     return std::move(results);
 597   }
 598 };
 599
 600 template <class T, class VecT>
 601 struct CollectContextHelper<T, VecT,
 602     typename std::enable_if<!std::is_same<T, VecT>::value>::type> {
 603   static inline std::vector<T> getResults(std::vector<VecT>& results) {
 604     std::vector<T> finalResults;
 605     finalResults.reserve(results.size());
 606     for (auto& opt : results) {
 607       finalResults.push_back(std::move(opt.value()));
 608     }
 609     return finalResults;
 610   }
 611 };
 612
 613 template <typename T>
 614 struct CollectContext {
 615
 616   typedef typename std::conditional<
 617     std::is_default_constructible<T>::value,
 618     T,
 619     Optional<T>
 620    >::type VecT;
 621
 622   explicit CollectContext(int n) : count(0), threw(false) {
 623     results.resize(n);
 624   }
 625
 626   Promise<std::vector<T>> p;
 627   std::vector<VecT> results;
 628   std::atomic<size_t> count;
 629   std::atomic_bool threw;
 630
 631   typedef std::vector<T> result_type;
 632
 633   static inline Future<std::vector<T>> makeEmptyFuture() {
 634     return makeFuture(std::vector<T>());
 635   }
 636
 637   inline void setValue() {
 638     p.setValue(CollectContextHelper<T, VecT>::getResults(results));
 639   }
 640
 641   inline void addResult(int i, Try<T>& t) {
 642     results[i] = std::move(t.value());
 643   }
 644 };
 645
 646 template <>
 647 struct CollectContext<void> {
 648
 649   explicit CollectContext(int n) : count(0), threw(false) {}
 650
 651   Promise<void> p;
 652   std::atomic<size_t> count;
 653   std::atomic_bool threw;
 654
 655   typedef void result_type;
 656
 657   static inline Future<void> makeEmptyFuture() {
 658     return makeFuture();
 659   }
 660
 661   inline void setValue() {
 662     p.setValue();
 663   }
 664
 665   inline void addResult(int i, Try<void>& t) {
 666     // do nothing
 667   }
 668 };
 669
 670 } // detail
 671
 672 template <class InputIterator>
 673 Future<typename detail::CollectContext<
 674   typename std::iterator_traits<InputIterator>::value_type::value_type
 675 >::result_type>
 676 collect(InputIterator first, InputIterator last) {
 677   typedef
 678     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 679
 680   if (first >= last) {
 681     return detail::CollectContext<T>::makeEmptyFuture();
 682   }
 683
 684   size_t n = std::distance(first, last);
 685   auto ctx = new detail::CollectContext<T>(n);
 686   auto f_saved = ctx->p.getFuture();
 687
 688   for (size_t i = 0; first != last; ++first, ++i) {
 689      assert(i < n);
 690      auto& f = *first;
 691      f.setCallback_([ctx, i, n](Try<T> t) {
 692        auto c = ++ctx->count;
 693
 694        if (t.hasException()) {
 695          if (!ctx->threw.exchange(true)) {
 696            ctx->p.setException(std::move(t.exception()));
 697          }
 698        } else if (!ctx->threw) {
 699          ctx->addResult(i, t);
 700          if (c == n) {
 701            ctx->setValue();
 702          }
 703        }
 704
 705        if (c == n) {
 706          delete ctx;
 707        }
 708      });
 709   }
 710
 711   return f_saved;
 712 }
 713
 714 template <class InputIterator>
 715 Future<
 716   std::pair<size_t,
 717             Try<
 718               typename
 719               std::iterator_traits<InputIterator>::value_type::value_type> > >
 720 collectAny(InputIterator first, InputIterator last) {
 721   typedef
 722     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 723
 724   auto ctx = new detail::WhenAnyContext<T>(std::distance(first, last));
 725   auto f_saved = ctx->p.getFuture();
 726
 727   for (size_t i = 0; first != last; first++, i++) {
 728     auto& f = *first;
 729     f.setCallback_([i, ctx](Try<T>&& t) {
 730       if (!ctx->done.exchange(true)) {
 731         ctx->p.setValue(std::make_pair(i, std::move(t)));
 732       }
 733       ctx->decref();
 734     });
 735   }
 736
 737   return f_saved;
 738 }
 739
 740 template <class InputIterator>
 741 Future<std::vector<std::pair<size_t, Try<typename
 742   std::iterator_traits<InputIterator>::value_type::value_type>>>>
 743 collectN(InputIterator first, InputIterator last, size_t n) {
 744   typedef typename
 745     std::iterator_traits<InputIterator>::value_type::value_type T;
 746   typedef std::vector<std::pair<size_t, Try<T>>> V;
 747
 748   struct ctx_t {
 749     V v;
 750     size_t completed;
 751     Promise<V> p;
 752   };
 753   auto ctx = std::make_shared<ctx_t>();
 754   ctx->completed = 0;
 755
 756   // for each completed Future, increase count and add to vector, until we
 757   // have n completed futures at which point we fulfill our Promise with the
 758   // vector
 759   auto it = first;
 760   size_t i = 0;
 761   while (it != last) {
 762     it->then([ctx, n, i](Try<T>&& t) {
 763       auto& v = ctx->v;
 764       auto c = ++ctx->completed;
 765       if (c <= n) {
 766         assert(ctx->v.size() < n);
 767         v.push_back(std::make_pair(i, std::move(t)));
 768         if (c == n) {
 769           ctx->p.setTry(Try<V>(std::move(v)));
 770         }
 771       }
 772     });
 773
 774     it++;
 775     i++;
 776   }
 777
 778   if (i < n) {
 779     ctx->p.setException(std::runtime_error("Not enough futures"));
 780   }
 781
 782   return ctx->p.getFuture();
 783 }
 784
 785 template <class It, class T, class F, class ItT, class Arg>
 786 typename std::enable_if<!isFutureResult<F, T, Arg>::value, Future<T>>::type
 787 reduce(It first, It last, T initial, F func) {
 788   if (first == last) {
 789     return makeFuture(std::move(initial));
 790   }
 791
 792   typedef isTry<Arg> IsTry;
 793
 794   return collectAll(first, last)
 795     .then([initial, func](std::vector<Try<ItT>>& vals) mutable {
 796       for (auto& val : vals) {
 797         initial = func(std::move(initial),
 798                        // Either return a ItT&& or a Try<ItT>&& depending
 799                        // on the type of the argument of func.
 800                        val.template get<IsTry::value, Arg&&>());
 801       }
 802       return initial;
 803     });
 804 }
 805
 806 template <class It, class T, class F, class ItT, class Arg>
 807 typename std::enable_if<isFutureResult<F, T, Arg>::value, Future<T>>::type
 808 reduce(It first, It last, T initial, F func) {
 809   if (first == last) {
 810     return makeFuture(std::move(initial));
 811   }
 812
 813   typedef isTry<Arg> IsTry;
 814
 815   auto f = first->then([initial, func](Try<ItT>& head) mutable {
 816     return func(std::move(initial),
 817                 head.template get<IsTry::value, Arg&&>());
 818   });
 819
 820   for (++first; first != last; ++first) {
 821     f = collectAll(f, *first).then([func](std::tuple<Try<T>, Try<ItT>>& t) {
 822       return func(std::move(std::get<0>(t).value()),
 823                   // Either return a ItT&& or a Try<ItT>&& depending
 824                   // on the type of the argument of func.
 825                   std::get<1>(t).template get<IsTry::value, Arg&&>());
 826     });
 827   }
 828
 829   return f;
 830 }
 831
 832 template <class T>
 833 Future<T> Future<T>::within(Duration dur, Timekeeper* tk) {
 834   return within(dur, TimedOut(), tk);
 835 }
 836
 837 template <class T>
 838 template <class E>
 839 Future<T> Future<T>::within(Duration dur, E e, Timekeeper* tk) {
 840
 841   struct Context {
 842     Context(E ex) : exception(std::move(ex)), promise(), token(false) {}
 843     E exception;
 844     Promise<T> promise;
 845     std::atomic<bool> token;
 846   };
 847   auto ctx = std::make_shared<Context>(std::move(e));
 848
 849   if (!tk) {
 850     tk = folly::detail::getTimekeeperSingleton();
 851   }
 852
 853   tk->after(dur)
 854     .then([ctx](Try<void> const& t) {
 855       if (ctx->token.exchange(true) == false) {
 856         if (t.hasException()) {
 857           ctx->promise.setException(std::move(t.exception()));
 858         } else {
 859           ctx->promise.setException(std::move(ctx->exception));
 860         }
 861       }
 862     });
 863
 864   this->then([ctx](Try<T>&& t) {
 865     if (ctx->token.exchange(true) == false) {
 866       ctx->promise.setTry(std::move(t));
 867     }
 868   });
 869
 870   return ctx->promise.getFuture();
 871 }
 872
 873 template <class T>
 874 Future<T> Future<T>::delayed(Duration dur, Timekeeper* tk) {
 875   return collectAll(*this, futures::sleep(dur, tk))
 876     .then([](std::tuple<Try<T>, Try<void>> tup) {
 877       Try<T>& t = std::get<0>(tup);
 878       return makeFuture<T>(std::move(t));
 879     });
 880 }
 881
 882 namespace detail {
 883
 884 template <class T>
 885 void waitImpl(Future<T>& f) {
 886   // short-circuit if there's nothing to do
 887   if (f.isReady()) return;
 888
 889   folly::fibers::Baton baton;
 890   f = f.then([&](Try<T> t) {
 891     baton.post();
 892     return makeFuture(std::move(t));
 893   });
 894   baton.wait();
 895
 896   // There's a race here between the return here and the actual finishing of
 897   // the future. f is completed, but the setup may not have finished on done
 898   // after the baton has posted.
 899   while (!f.isReady()) {
 900     std::this_thread::yield();
 901   }
 902 }
 903
 904 template <class T>
 905 void waitImpl(Future<T>& f, Duration dur) {
 906   // short-circuit if there's nothing to do
 907   if (f.isReady()) return;
 908
 909   auto baton = std::make_shared<folly::fibers::Baton>();
 910   f = f.then([baton](Try<T> t) {
 911     baton->post();
 912     return makeFuture(std::move(t));
 913   });
 914
 915   // Let's preserve the invariant that if we did not timeout (timed_wait returns
 916   // true), then the returned Future is complete when it is returned to the
 917   // caller. We need to wait out the race for that Future to complete.
 918   if (baton->timed_wait(dur)) {
 919     while (!f.isReady()) {
 920       std::this_thread::yield();
 921     }
 922   }
 923 }
 924
 925 template <class T>
 926 void waitViaImpl(Future<T>& f, DrivableExecutor* e) {
 927   while (!f.isReady()) {
 928     e->drive();
 929   }
 930 }
 931
 932 } // detail
 933
 934 template <class T>
 935 Future<T>& Future<T>::wait() & {
 936   detail::waitImpl(*this);
 937   return *this;
 938 }
 939
 940 template <class T>
 941 Future<T>&& Future<T>::wait() && {
 942   detail::waitImpl(*this);
 943   return std::move(*this);
 944 }
 945
 946 template <class T>
 947 Future<T>& Future<T>::wait(Duration dur) & {
 948   detail::waitImpl(*this, dur);
 949   return *this;
 950 }
 951
 952 template <class T>
 953 Future<T>&& Future<T>::wait(Duration dur) && {
 954   detail::waitImpl(*this, dur);
 955   return std::move(*this);
 956 }
 957
 958 template <class T>
 959 Future<T>& Future<T>::waitVia(DrivableExecutor* e) & {
 960   detail::waitViaImpl(*this, e);
 961   return *this;
 962 }
 963
 964 template <class T>
 965 Future<T>&& Future<T>::waitVia(DrivableExecutor* e) && {
 966   detail::waitViaImpl(*this, e);
 967   return std::move(*this);
 968 }
 969
 970 template <class T>
 971 T Future<T>::get() {
 972   return std::move(wait().value());
 973 }
 974
 975 template <>
 976 inline void Future<void>::get() {
 977   wait().value();
 978 }
 979
 980 template <class T>
 981 T Future<T>::get(Duration dur) {
 982   wait(dur);
 983   if (isReady()) {
 984     return std::move(value());
 985   } else {
 986     throw TimedOut();
 987   }
 988 }
 989
 990 template <>
 991 inline void Future<void>::get(Duration dur) {
 992   wait(dur);
 993   if (isReady()) {
 994     return;
 995   } else {
 996     throw TimedOut();
 997   }
 998 }
 999
1000 template <class T>
1001 T Future<T>::getVia(DrivableExecutor* e) {
1002   return std::move(waitVia(e).value());
1003 }
1004
1005 template <>
1006 inline void Future<void>::getVia(DrivableExecutor* e) {
1007   waitVia(e).value();
1008 }
1009
1010 template <class T>
1011 Future<bool> Future<T>::willEqual(Future<T>& f) {
1012   return collectAll(*this, f).then([](const std::tuple<Try<T>, Try<T>>& t) {
1013     if (std::get<0>(t).hasValue() && std::get<1>(t).hasValue()) {
1014       return std::get<0>(t).value() == std::get<1>(t).value();
1015     } else {
1016       return false;
1017       }
1018   });
1019 }
1020
1021 template <class T>
1022 template <class F>
1023 Future<T> Future<T>::filter(F predicate) {
1024   auto p = folly::makeMoveWrapper(std::move(predicate));
1025   return this->then([p](T val) {
1026     T const& valConstRef = val;
1027     if (!(*p)(valConstRef)) {
1028       throw PredicateDoesNotObtain();
1029     }
1030     return val;
1031   });
1032 }
1033
1034 namespace futures {
1035   namespace {
1036     template <class Z>
1037     Future<Z> chainHelper(Future<Z> f) {
1038       return f;
1039     }
1040
1041     template <class Z, class F, class Fn, class... Callbacks>
1042     Future<Z> chainHelper(F f, Fn fn, Callbacks... fns) {
1043       return chainHelper<Z>(f.then(fn), fns...);
1044     }
1045   }
1046
1047   template <class A, class Z, class... Callbacks>
1048   std::function<Future<Z>(Try<A>)>
1049   chain(Callbacks... fns) {
1050     MoveWrapper<Promise<A>> pw;
1051     MoveWrapper<Future<Z>> fw(chainHelper<Z>(pw->getFuture(), fns...));
1052     return [=](Try<A> t) mutable {
1053       pw->setTry(std::move(t));
1054       return std::move(*fw);
1055     };
1056   }
1057
1058   template <class It, class F, class ItT, class Result>
1059   std::vector<Future<Result>> map(It first, It last, F func) {
1060     std::vector<Future<Result>> results;
1061     for (auto it = first; it != last; it++) {
1062       results.push_back(it->then(func));
1063     }
1064     return results;
1065   }
1066 }
1067
1068 } // namespace folly
1069
1070 // I haven't included a Future<T&> specialization because I don't forsee us
1071 // using it, however it is not difficult to add when needed. Refer to
1072 // Future<void> for guidance. std::future and boost::future code would also be
1073 // instructive.