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 template <class T>
 230 template <class Executor, class Arg, class... Args>
 231 auto Future<T>::then(Executor* x, Arg&& arg, Args&&... args)
 232   -> decltype(this->then(std::forward<Arg>(arg),
 233                          std::forward<Args>(args)...))
 234 {
 235   auto oldX = getExecutor();
 236   setExecutor(x);
 237   return this->then(std::forward<Arg>(arg), std::forward<Args>(args)...).
 238                via(oldX);
 239 }
 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 inline Future<T> Future<T>::via(Executor* executor) && {
 427   throwIfInvalid();
 428
 429   setExecutor(executor);
 430
 431   return std::move(*this);
 432 }
 433
 434 template <class T>
 435 inline Future<T> Future<T>::via(Executor* executor) & {
 436   throwIfInvalid();
 437
 438   MoveWrapper<Promise<T>> p;
 439   auto f = p->getFuture();
 440   then([p](Try<T>&& t) mutable { p->setTry(std::move(t)); });
 441   return std::move(f).via(executor);
 442 }
 443
 444 template <class T>
 445 bool Future<T>::isReady() const {
 446   throwIfInvalid();
 447   return core_->ready();
 448 }
 449
 450 template <class T>
 451 void Future<T>::raise(exception_wrapper exception) {
 452   core_->raise(std::move(exception));
 453 }
 454
 455 // makeFuture
 456
 457 template <class T>
 458 Future<typename std::decay<T>::type> makeFuture(T&& t) {
 459   Promise<typename std::decay<T>::type> p;
 460   p.setValue(std::forward<T>(t));
 461   return p.getFuture();
 462 }
 463
 464 inline // for multiple translation units
 465 Future<void> makeFuture() {
 466   Promise<void> p;
 467   p.setValue();
 468   return p.getFuture();
 469 }
 470
 471 template <class F>
 472 auto makeFutureWith(
 473     F&& func,
 474     typename std::enable_if<!std::is_reference<F>::value, bool>::type sdf)
 475     -> Future<decltype(func())> {
 476   Promise<decltype(func())> p;
 477   p.setWith(
 478     [&func]() {
 479       return (func)();
 480     });
 481   return p.getFuture();
 482 }
 483
 484 template <class F>
 485 auto makeFutureWith(F const& func) -> Future<decltype(func())> {
 486   F copy = func;
 487   return makeFutureWith(std::move(copy));
 488 }
 489
 490 template <class T>
 491 Future<T> makeFuture(std::exception_ptr const& e) {
 492   Promise<T> p;
 493   p.setException(e);
 494   return p.getFuture();
 495 }
 496
 497 template <class T>
 498 Future<T> makeFuture(exception_wrapper ew) {
 499   Promise<T> p;
 500   p.setException(std::move(ew));
 501   return p.getFuture();
 502 }
 503
 504 template <class T, class E>
 505 typename std::enable_if<std::is_base_of<std::exception, E>::value,
 506                         Future<T>>::type
 507 makeFuture(E const& e) {
 508   Promise<T> p;
 509   p.setException(make_exception_wrapper<E>(e));
 510   return p.getFuture();
 511 }
 512
 513 template <class T>
 514 Future<T> makeFuture(Try<T>&& t) {
 515   Promise<typename std::decay<T>::type> p;
 516   p.setTry(std::move(t));
 517   return p.getFuture();
 518 }
 519
 520 template <>
 521 inline Future<void> makeFuture(Try<void>&& t) {
 522   if (t.hasException()) {
 523     return makeFuture<void>(std::move(t.exception()));
 524   } else {
 525     return makeFuture();
 526   }
 527 }
 528
 529 // via
 530 inline Future<void> via(Executor* executor) {
 531   return makeFuture().via(executor);
 532 }
 533
 534 // when (variadic)
 535
 536 template <typename... Fs>
 537 typename detail::VariadicContext<
 538   typename std::decay<Fs>::type::value_type...>::type
 539 collectAll(Fs&&... fs) {
 540   auto ctx =
 541     new detail::VariadicContext<typename std::decay<Fs>::type::value_type...>();
 542   ctx->total = sizeof...(fs);
 543   auto f_saved = ctx->p.getFuture();
 544   detail::collectAllVariadicHelper(ctx,
 545     std::forward<typename std::decay<Fs>::type>(fs)...);
 546   return f_saved;
 547 }
 548
 549 // when (iterator)
 550
 551 template <class InputIterator>
 552 Future<
 553   std::vector<
 554   Try<typename std::iterator_traits<InputIterator>::value_type::value_type>>>
 555 collectAll(InputIterator first, InputIterator last) {
 556   typedef
 557     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 558
 559   if (first >= last) {
 560     return makeFuture(std::vector<Try<T>>());
 561   }
 562   size_t n = std::distance(first, last);
 563
 564   auto ctx = new detail::WhenAllContext<T>();
 565
 566   ctx->results.resize(n);
 567
 568   auto f_saved = ctx->p.getFuture();
 569
 570   for (size_t i = 0; first != last; ++first, ++i) {
 571      assert(i < n);
 572      auto& f = *first;
 573      f.setCallback_([ctx, i, n](Try<T> t) {
 574        ctx->results[i] = std::move(t);
 575        if (++ctx->count == n) {
 576          ctx->p.setValue(std::move(ctx->results));
 577          delete ctx;
 578        }
 579      });
 580   }
 581
 582   return f_saved;
 583 }
 584
 585 namespace detail {
 586
 587 template <class, class, typename = void> struct CollectContextHelper;
 588
 589 template <class T, class VecT>
 590 struct CollectContextHelper<T, VecT,
 591     typename std::enable_if<std::is_same<T, VecT>::value>::type> {
 592   static inline std::vector<T>&& getResults(std::vector<VecT>& results) {
 593     return std::move(results);
 594   }
 595 };
 596
 597 template <class T, class VecT>
 598 struct CollectContextHelper<T, VecT,
 599     typename std::enable_if<!std::is_same<T, VecT>::value>::type> {
 600   static inline std::vector<T> getResults(std::vector<VecT>& results) {
 601     std::vector<T> finalResults;
 602     finalResults.reserve(results.size());
 603     for (auto& opt : results) {
 604       finalResults.push_back(std::move(opt.value()));
 605     }
 606     return finalResults;
 607   }
 608 };
 609
 610 template <typename T>
 611 struct CollectContext {
 612
 613   typedef typename std::conditional<
 614     std::is_default_constructible<T>::value,
 615     T,
 616     Optional<T>
 617    >::type VecT;
 618
 619   explicit CollectContext(int n) : count(0), success_count(0), threw(false) {
 620     results.resize(n);
 621   }
 622
 623   Promise<std::vector<T>> p;
 624   std::vector<VecT> results;
 625   std::atomic<size_t> count, success_count;
 626   std::atomic_bool threw;
 627
 628   typedef std::vector<T> result_type;
 629
 630   static inline Future<std::vector<T>> makeEmptyFuture() {
 631     return makeFuture(std::vector<T>());
 632   }
 633
 634   inline void setValue() {
 635     p.setValue(CollectContextHelper<T, VecT>::getResults(results));
 636   }
 637
 638   inline void addResult(int i, Try<T>& t) {
 639     results[i] = std::move(t.value());
 640   }
 641 };
 642
 643 template <>
 644 struct CollectContext<void> {
 645
 646   explicit CollectContext(int n) : count(0), success_count(0), threw(false) {}
 647
 648   Promise<void> p;
 649   std::atomic<size_t> count, success_count;
 650   std::atomic_bool threw;
 651
 652   typedef void result_type;
 653
 654   static inline Future<void> makeEmptyFuture() {
 655     return makeFuture();
 656   }
 657
 658   inline void setValue() {
 659     p.setValue();
 660   }
 661
 662   inline void addResult(int i, Try<void>& t) {
 663     // do nothing
 664   }
 665 };
 666
 667 } // detail
 668
 669 template <class InputIterator>
 670 Future<typename detail::CollectContext<
 671   typename std::iterator_traits<InputIterator>::value_type::value_type
 672 >::result_type>
 673 collect(InputIterator first, InputIterator last) {
 674   typedef
 675     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 676
 677   if (first >= last) {
 678     return detail::CollectContext<T>::makeEmptyFuture();
 679   }
 680
 681   size_t n = std::distance(first, last);
 682   auto ctx = new detail::CollectContext<T>(n);
 683   auto f_saved = ctx->p.getFuture();
 684
 685   for (size_t i = 0; first != last; ++first, ++i) {
 686      assert(i < n);
 687      auto& f = *first;
 688      f.setCallback_([ctx, i, n](Try<T> t) {
 689
 690        if (t.hasException()) {
 691          if (!ctx->threw.exchange(true)) {
 692            ctx->p.setException(std::move(t.exception()));
 693          }
 694        } else if (!ctx->threw) {
 695          ctx->addResult(i, t);
 696          if (++ctx->success_count == n) {
 697            ctx->setValue();
 698          }
 699        }
 700
 701        if (++ctx->count == n) {
 702          delete ctx;
 703        }
 704      });
 705   }
 706
 707   return f_saved;
 708 }
 709
 710 template <class InputIterator>
 711 Future<
 712   std::pair<size_t,
 713             Try<
 714               typename
 715               std::iterator_traits<InputIterator>::value_type::value_type> > >
 716 collectAny(InputIterator first, InputIterator last) {
 717   typedef
 718     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 719
 720   auto ctx = new detail::WhenAnyContext<T>(std::distance(first, last));
 721   auto f_saved = ctx->p.getFuture();
 722
 723   for (size_t i = 0; first != last; first++, i++) {
 724     auto& f = *first;
 725     f.setCallback_([i, ctx](Try<T>&& t) {
 726       if (!ctx->done.exchange(true)) {
 727         ctx->p.setValue(std::make_pair(i, std::move(t)));
 728       }
 729       ctx->decref();
 730     });
 731   }
 732
 733   return f_saved;
 734 }
 735
 736 template <class InputIterator>
 737 Future<std::vector<std::pair<size_t, Try<typename
 738   std::iterator_traits<InputIterator>::value_type::value_type>>>>
 739 collectN(InputIterator first, InputIterator last, size_t n) {
 740   typedef typename
 741     std::iterator_traits<InputIterator>::value_type::value_type T;
 742   typedef std::vector<std::pair<size_t, Try<T>>> V;
 743
 744   struct ctx_t {
 745     V v;
 746     size_t completed;
 747     Promise<V> p;
 748   };
 749   auto ctx = std::make_shared<ctx_t>();
 750   ctx->completed = 0;
 751
 752   // for each completed Future, increase count and add to vector, until we
 753   // have n completed futures at which point we fulfill our Promise with the
 754   // vector
 755   auto it = first;
 756   size_t i = 0;
 757   while (it != last) {
 758     it->then([ctx, n, i](Try<T>&& t) {
 759       auto& v = ctx->v;
 760       auto c = ++ctx->completed;
 761       if (c <= n) {
 762         assert(ctx->v.size() < n);
 763         v.push_back(std::make_pair(i, std::move(t)));
 764         if (c == n) {
 765           ctx->p.setTry(Try<V>(std::move(v)));
 766         }
 767       }
 768     });
 769
 770     it++;
 771     i++;
 772   }
 773
 774   if (i < n) {
 775     ctx->p.setException(std::runtime_error("Not enough futures"));
 776   }
 777
 778   return ctx->p.getFuture();
 779 }
 780
 781 template <class It, class T, class F>
 782 Future<T> reduce(It first, It last, T&& initial, F&& func) {
 783   if (first == last) {
 784     return makeFuture(std::move(initial));
 785   }
 786
 787   typedef typename std::iterator_traits<It>::value_type::value_type ItT;
 788   typedef typename std::conditional<
 789     detail::callableWith<F, T&&, Try<ItT>&&>::value, Try<ItT>, ItT>::type Arg;
 790   typedef isTry<Arg> IsTry;
 791
 792   folly::MoveWrapper<T> minitial(std::move(initial));
 793   auto sfunc = std::make_shared<F>(std::move(func));
 794
 795   auto f = first->then([minitial, sfunc](Try<ItT>& head) mutable {
 796     return (*sfunc)(std::move(*minitial),
 797                 head.template get<IsTry::value, Arg&&>());
 798   });
 799
 800   for (++first; first != last; ++first) {
 801     f = collectAll(f, *first).then([sfunc](std::tuple<Try<T>, Try<ItT>>& t) {
 802       return (*sfunc)(std::move(std::get<0>(t).value()),
 803                   // Either return a ItT&& or a Try<ItT>&& depending
 804                   // on the type of the argument of func.
 805                   std::get<1>(t).template get<IsTry::value, Arg&&>());
 806     });
 807   }
 808
 809   return f;
 810 }
 811
 812 template <class T>
 813 template <class I, class F>
 814 Future<I> Future<T>::reduce(I&& initial, F&& func) {
 815   folly::MoveWrapper<I> minitial(std::move(initial));
 816   folly::MoveWrapper<F> mfunc(std::move(func));
 817   return then([minitial, mfunc](T& vals) mutable {
 818     auto ret = std::move(*minitial);
 819     for (auto& val : vals) {
 820       ret = (*mfunc)(std::move(ret), std::move(val));
 821     }
 822     return ret;
 823   });
 824 }
 825
 826 template <class T>
 827 Future<T> Future<T>::within(Duration dur, Timekeeper* tk) {
 828   return within(dur, TimedOut(), tk);
 829 }
 830
 831 template <class T>
 832 template <class E>
 833 Future<T> Future<T>::within(Duration dur, E e, Timekeeper* tk) {
 834
 835   struct Context {
 836     Context(E ex) : exception(std::move(ex)), promise(), token(false) {}
 837     E exception;
 838     Promise<T> promise;
 839     std::atomic<bool> token;
 840   };
 841   auto ctx = std::make_shared<Context>(std::move(e));
 842
 843   if (!tk) {
 844     tk = folly::detail::getTimekeeperSingleton();
 845   }
 846
 847   tk->after(dur)
 848     .then([ctx](Try<void> const& t) {
 849       if (ctx->token.exchange(true) == false) {
 850         if (t.hasException()) {
 851           ctx->promise.setException(std::move(t.exception()));
 852         } else {
 853           ctx->promise.setException(std::move(ctx->exception));
 854         }
 855       }
 856     });
 857
 858   this->then([ctx](Try<T>&& t) {
 859     if (ctx->token.exchange(true) == false) {
 860       ctx->promise.setTry(std::move(t));
 861     }
 862   });
 863
 864   return ctx->promise.getFuture();
 865 }
 866
 867 template <class T>
 868 Future<T> Future<T>::delayed(Duration dur, Timekeeper* tk) {
 869   return collectAll(*this, futures::sleep(dur, tk))
 870     .then([](std::tuple<Try<T>, Try<void>> tup) {
 871       Try<T>& t = std::get<0>(tup);
 872       return makeFuture<T>(std::move(t));
 873     });
 874 }
 875
 876 namespace detail {
 877
 878 template <class T>
 879 void waitImpl(Future<T>& f) {
 880   // short-circuit if there's nothing to do
 881   if (f.isReady()) return;
 882
 883   folly::fibers::Baton baton;
 884   f = f.then([&](Try<T> t) {
 885     baton.post();
 886     return makeFuture(std::move(t));
 887   });
 888   baton.wait();
 889
 890   // There's a race here between the return here and the actual finishing of
 891   // the future. f is completed, but the setup may not have finished on done
 892   // after the baton has posted.
 893   while (!f.isReady()) {
 894     std::this_thread::yield();
 895   }
 896 }
 897
 898 template <class T>
 899 void waitImpl(Future<T>& f, Duration dur) {
 900   // short-circuit if there's nothing to do
 901   if (f.isReady()) return;
 902
 903   auto baton = std::make_shared<folly::fibers::Baton>();
 904   f = f.then([baton](Try<T> t) {
 905     baton->post();
 906     return makeFuture(std::move(t));
 907   });
 908
 909   // Let's preserve the invariant that if we did not timeout (timed_wait returns
 910   // true), then the returned Future is complete when it is returned to the
 911   // caller. We need to wait out the race for that Future to complete.
 912   if (baton->timed_wait(dur)) {
 913     while (!f.isReady()) {
 914       std::this_thread::yield();
 915     }
 916   }
 917 }
 918
 919 template <class T>
 920 void waitViaImpl(Future<T>& f, DrivableExecutor* e) {
 921   while (!f.isReady()) {
 922     e->drive();
 923   }
 924 }
 925
 926 } // detail
 927
 928 template <class T>
 929 Future<T>& Future<T>::wait() & {
 930   detail::waitImpl(*this);
 931   return *this;
 932 }
 933
 934 template <class T>
 935 Future<T>&& Future<T>::wait() && {
 936   detail::waitImpl(*this);
 937   return std::move(*this);
 938 }
 939
 940 template <class T>
 941 Future<T>& Future<T>::wait(Duration dur) & {
 942   detail::waitImpl(*this, dur);
 943   return *this;
 944 }
 945
 946 template <class T>
 947 Future<T>&& Future<T>::wait(Duration dur) && {
 948   detail::waitImpl(*this, dur);
 949   return std::move(*this);
 950 }
 951
 952 template <class T>
 953 Future<T>& Future<T>::waitVia(DrivableExecutor* e) & {
 954   detail::waitViaImpl(*this, e);
 955   return *this;
 956 }
 957
 958 template <class T>
 959 Future<T>&& Future<T>::waitVia(DrivableExecutor* e) && {
 960   detail::waitViaImpl(*this, e);
 961   return std::move(*this);
 962 }
 963
 964 template <class T>
 965 T Future<T>::get() {
 966   return std::move(wait().value());
 967 }
 968
 969 template <>
 970 inline void Future<void>::get() {
 971   wait().value();
 972 }
 973
 974 template <class T>
 975 T Future<T>::get(Duration dur) {
 976   wait(dur);
 977   if (isReady()) {
 978     return std::move(value());
 979   } else {
 980     throw TimedOut();
 981   }
 982 }
 983
 984 template <>
 985 inline void Future<void>::get(Duration dur) {
 986   wait(dur);
 987   if (isReady()) {
 988     return;
 989   } else {
 990     throw TimedOut();
 991   }
 992 }
 993
 994 template <class T>
 995 T Future<T>::getVia(DrivableExecutor* e) {
 996   return std::move(waitVia(e).value());
 997 }
 998
 999 template <>
1000 inline void Future<void>::getVia(DrivableExecutor* e) {
1001   waitVia(e).value();
1002 }
1003
1004 namespace detail {
1005   template <class T>
1006   struct TryEquals {
1007     static bool equals(const Try<T>& t1, const Try<T>& t2) {
1008       return t1.value() == t2.value();
1009     }
1010   };
1011
1012   template <>
1013   struct TryEquals<void> {
1014     static bool equals(const Try<void>& t1, const Try<void>& t2) {
1015       return true;
1016     }
1017   };
1018 }
1019
1020 template <class T>
1021 Future<bool> Future<T>::willEqual(Future<T>& f) {
1022   return collectAll(*this, f).then([](const std::tuple<Try<T>, Try<T>>& t) {
1023     if (std::get<0>(t).hasValue() && std::get<1>(t).hasValue()) {
1024       return detail::TryEquals<T>::equals(std::get<0>(t), std::get<1>(t));
1025     } else {
1026       return false;
1027       }
1028   });
1029 }
1030
1031 template <class T>
1032 template <class F>
1033 Future<T> Future<T>::filter(F predicate) {
1034   auto p = folly::makeMoveWrapper(std::move(predicate));
1035   return this->then([p](T val) {
1036     T const& valConstRef = val;
1037     if (!(*p)(valConstRef)) {
1038       throw PredicateDoesNotObtain();
1039     }
1040     return val;
1041   });
1042 }
1043
1044 namespace futures {
1045   namespace {
1046     template <class Z>
1047     Future<Z> chainHelper(Future<Z> f) {
1048       return f;
1049     }
1050
1051     template <class Z, class F, class Fn, class... Callbacks>
1052     Future<Z> chainHelper(F f, Fn fn, Callbacks... fns) {
1053       return chainHelper<Z>(f.then(fn), fns...);
1054     }
1055   }
1056
1057   template <class A, class Z, class... Callbacks>
1058   std::function<Future<Z>(Try<A>)>
1059   chain(Callbacks... fns) {
1060     MoveWrapper<Promise<A>> pw;
1061     MoveWrapper<Future<Z>> fw(chainHelper<Z>(pw->getFuture(), fns...));
1062     return [=](Try<A> t) mutable {
1063       pw->setTry(std::move(t));
1064       return std::move(*fw);
1065     };
1066   }
1067
1068   template <class It, class F, class ItT, class Result>
1069   std::vector<Future<Result>> map(It first, It last, F func) {
1070     std::vector<Future<Result>> results;
1071     for (auto it = first; it != last; it++) {
1072       results.push_back(it->then(func));
1073     }
1074     return results;
1075   }
1076 }
1077
1078 // Instantiate the most common Future types to save compile time
1079 extern template class Future<void>;
1080 extern template class Future<bool>;
1081 extern template class Future<int>;
1082 extern template class Future<int64_t>;
1083 extern template class Future<std::string>;
1084 extern template class Future<double>;
1085
1086 } // namespace folly
1087
1088 // I haven't included a Future<T&> specialization because I don't forsee us
1089 // using it, however it is not difficult to add when needed. Refer to
1090 // Future<void> for guidance. std::future and boost::future code would also be
1091 // instructive.