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