folly/futures/Future-inl.h

   1 /*
   2  * Copyright 2017 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 <algorithm>
  20 #include <cassert>
  21 #include <chrono>
  22 #include <thread>
  23
  24 #include <folly/Optional.h>
  25 #include <folly/executors/InlineExecutor.h>
  26 #include <folly/futures/Timekeeper.h>
  27 #include <folly/futures/detail/Core.h>
  28 #include <folly/synchronization/Baton.h>
  29
  30 #ifndef FOLLY_FUTURE_USING_FIBER
  31 #if FOLLY_MOBILE || defined(__APPLE__)
  32 #define FOLLY_FUTURE_USING_FIBER 0
  33 #else
  34 #define FOLLY_FUTURE_USING_FIBER 1
  35 #include <folly/fibers/Baton.h>
  36 #endif
  37 #endif
  38
  39 namespace folly {
  40
  41 class Timekeeper;
  42
  43 namespace futures {
  44 namespace detail {
  45 #if FOLLY_FUTURE_USING_FIBER
  46 typedef folly::fibers::Baton FutureBatonType;
  47 #else
  48 typedef folly::Baton<> FutureBatonType;
  49 #endif
  50 } // namespace detail
  51 } // namespace futures
  52
  53 namespace detail {
  54 std::shared_ptr<Timekeeper> getTimekeeperSingleton();
  55 } // namespace detail
  56
  57 namespace futures {
  58 namespace detail {
  59 //  Guarantees that the stored functor is destructed before the stored promise
  60 //  may be fulfilled. Assumes the stored functor to be noexcept-destructible.
  61 template <typename T, typename F>
  62 class CoreCallbackState {
  63  public:
  64   template <typename FF>
  65   CoreCallbackState(Promise<T>&& promise, FF&& func) noexcept(
  66       noexcept(F(std::declval<FF>())))
  67       : func_(std::forward<FF>(func)), promise_(std::move(promise)) {
  68     assert(before_barrier());
  69   }
  70
  71   CoreCallbackState(CoreCallbackState&& that) noexcept(
  72       noexcept(F(std::declval<F>()))) {
  73     if (that.before_barrier()) {
  74       new (&func_) F(std::move(that.func_));
  75       promise_ = that.stealPromise();
  76     }
  77   }
  78
  79   CoreCallbackState& operator=(CoreCallbackState&&) = delete;
  80
  81   ~CoreCallbackState() {
  82     if (before_barrier()) {
  83       stealPromise();
  84     }
  85   }
  86
  87   template <typename... Args>
  88   auto invoke(Args&&... args) noexcept(
  89       noexcept(std::declval<F&&>()(std::declval<Args&&>()...))) {
  90     assert(before_barrier());
  91     return std::move(func_)(std::forward<Args>(args)...);
  92   }
  93
  94   template <typename... Args>
  95   auto tryInvoke(Args&&... args) noexcept {
  96     return makeTryWith([&] { return invoke(std::forward<Args>(args)...); });
  97   }
  98
  99   void setTry(Try<T>&& t) {
 100     stealPromise().setTry(std::move(t));
 101   }
 102
 103   void setException(exception_wrapper&& ew) {
 104     stealPromise().setException(std::move(ew));
 105   }
 106
 107   Promise<T> stealPromise() noexcept {
 108     assert(before_barrier());
 109     func_.~F();
 110     return std::move(promise_);
 111   }
 112
 113  private:
 114   bool before_barrier() const noexcept {
 115     return !promise_.isFulfilled();
 116   }
 117
 118   union {
 119     F func_;
 120   };
 121   Promise<T> promise_{Promise<T>::makeEmpty()};
 122 };
 123
 124 template <typename T, typename F>
 125 inline auto makeCoreCallbackState(Promise<T>&& p, F&& f) noexcept(
 126     noexcept(CoreCallbackState<T, _t<std::decay<F>>>(
 127         std::declval<Promise<T>&&>(),
 128         std::declval<F&&>()))) {
 129   return CoreCallbackState<T, _t<std::decay<F>>>(
 130       std::move(p), std::forward<F>(f));
 131 }
 132
 133 template <class T>
 134 FutureBase<T>::FutureBase(SemiFuture<T>&& other) noexcept : core_(other.core_) {
 135   other.core_ = nullptr;
 136 }
 137
 138 template <class T>
 139 FutureBase<T>::FutureBase(Future<T>&& other) noexcept : core_(other.core_) {
 140   other.core_ = nullptr;
 141 }
 142
 143 template <class T>
 144 template <class T2, typename>
 145 FutureBase<T>::FutureBase(T2&& val)
 146     : core_(new futures::detail::Core<T>(Try<T>(std::forward<T2>(val)))) {}
 147
 148 template <class T>
 149 template <typename T2>
 150 FutureBase<T>::FutureBase(
 151     typename std::enable_if<std::is_same<Unit, T2>::value>::type*)
 152     : core_(new futures::detail::Core<T>(Try<T>(T()))) {}
 153
 154 template <class T>
 155 template <
 156     class... Args,
 157     typename std::enable_if<std::is_constructible<T, Args&&...>::value, int>::
 158         type>
 159 FutureBase<T>::FutureBase(in_place_t, Args&&... args)
 160     : core_(
 161           new futures::detail::Core<T>(in_place, std::forward<Args>(args)...)) {
 162 }
 163
 164 template <class T>
 165 template <class FutureType>
 166 void FutureBase<T>::assign(FutureType& other) noexcept {
 167   std::swap(core_, other.core_);
 168 }
 169
 170 template <class T>
 171 FutureBase<T>::~FutureBase() {
 172   detach();
 173 }
 174
 175 template <class T>
 176 T& FutureBase<T>::value() & {
 177   throwIfInvalid();
 178
 179   return core_->getTry().value();
 180 }
 181
 182 template <class T>
 183 T const& FutureBase<T>::value() const& {
 184   throwIfInvalid();
 185
 186   return core_->getTry().value();
 187 }
 188
 189 template <class T>
 190 T&& FutureBase<T>::value() && {
 191   throwIfInvalid();
 192
 193   return std::move(core_->getTry().value());
 194 }
 195
 196 template <class T>
 197 T const&& FutureBase<T>::value() const&& {
 198   throwIfInvalid();
 199
 200   return std::move(core_->getTry().value());
 201 }
 202
 203 template <class T>
 204 bool FutureBase<T>::isReady() const {
 205   throwIfInvalid();
 206   return core_->ready();
 207 }
 208
 209 template <class T>
 210 bool FutureBase<T>::hasValue() {
 211   return core_->getTry().hasValue();
 212 }
 213
 214 template <class T>
 215 bool FutureBase<T>::hasException() {
 216   return core_->getTry().hasException();
 217 }
 218
 219 template <class T>
 220 void FutureBase<T>::detach() {
 221   if (core_) {
 222     core_->detachFuture();
 223     core_ = nullptr;
 224   }
 225 }
 226
 227 template <class T>
 228 void FutureBase<T>::throwIfInvalid() const {
 229   if (!core_) {
 230     throwNoState();
 231   }
 232 }
 233
 234 template <class T>
 235 Optional<Try<T>> FutureBase<T>::poll() {
 236   Optional<Try<T>> o;
 237   if (core_->ready()) {
 238     o = std::move(core_->getTry());
 239   }
 240   return o;
 241 }
 242
 243 template <class T>
 244 void FutureBase<T>::raise(exception_wrapper exception) {
 245   core_->raise(std::move(exception));
 246 }
 247
 248 template <class T>
 249 template <class F>
 250 void FutureBase<T>::setCallback_(F&& func) {
 251   throwIfInvalid();
 252   core_->setCallback(std::forward<F>(func));
 253 }
 254
 255 template <class T>
 256 FutureBase<T>::FutureBase(futures::detail::EmptyConstruct) noexcept
 257     : core_(nullptr) {}
 258
 259 // then
 260
 261 // Variant: returns a value
 262 // e.g. f.then([](Try<T>&& t){ return t.value(); });
 263 template <class T>
 264 template <typename F, typename R, bool isTry, typename... Args>
 265 typename std::enable_if<!R::ReturnsFuture::value, typename R::Return>::type
 266 FutureBase<T>::thenImplementation(
 267     F&& func,
 268     futures::detail::argResult<isTry, F, Args...>) {
 269   static_assert(sizeof...(Args) <= 1, "Then must take zero/one argument");
 270   typedef typename R::ReturnsFuture::Inner B;
 271
 272   this->throwIfInvalid();
 273
 274   Promise<B> p;
 275   p.core_->setInterruptHandlerNoLock(this->core_->getInterruptHandler());
 276
 277   // grab the Future now before we lose our handle on the Promise
 278   auto f = p.getFuture();
 279   f.core_->setExecutorNoLock(this->getExecutor());
 280
 281   /* This is a bit tricky.
 282
 283      We can't just close over *this in case this Future gets moved. So we
 284      make a new dummy Future. We could figure out something more
 285      sophisticated that avoids making a new Future object when it can, as an
 286      optimization. But this is correct.
 287
 288      core_ can't be moved, it is explicitly disallowed (as is copying). But
 289      if there's ever a reason to allow it, this is one place that makes that
 290      assumption and would need to be fixed. We use a standard shared pointer
 291      for core_ (by copying it in), which means in essence obj holds a shared
 292      pointer to itself.  But this shouldn't leak because Promise will not
 293      outlive the continuation, because Promise will setException() with a
 294      broken Promise if it is destructed before completed. We could use a
 295      weak pointer but it would have to be converted to a shared pointer when
 296      func is executed (because the Future returned by func may possibly
 297      persist beyond the callback, if it gets moved), and so it is an
 298      optimization to just make it shared from the get-go.
 299
 300      Two subtle but important points about this design. futures::detail::Core
 301      has no back pointers to Future or Promise, so if Future or Promise get
 302      moved (and they will be moved in performant code) we don't have to do
 303      anything fancy. And because we store the continuation in the
 304      futures::detail::Core, not in the Future, we can execute the continuation
 305      even after the Future has gone out of scope. This is an intentional design
 306      decision. It is likely we will want to be able to cancel a continuation
 307      in some circumstances, but I think it should be explicit not implicit
 308      in the destruction of the Future used to create it.
 309      */
 310   this->setCallback_(
 311       [state = futures::detail::makeCoreCallbackState(
 312            std::move(p), std::forward<F>(func))](Try<T>&& t) mutable {
 313
 314         if (!isTry && t.hasException()) {
 315           state.setException(std::move(t.exception()));
 316         } else {
 317           state.setTry(makeTryWith(
 318               [&] { return state.invoke(t.template get<isTry, Args>()...); }));
 319         }
 320       });
 321   return f;
 322 }
 323
 324 // Variant: returns a Future
 325 // e.g. f.then([](T&& t){ return makeFuture<T>(t); });
 326 template <class T>
 327 template <typename F, typename R, bool isTry, typename... Args>
 328 typename std::enable_if<R::ReturnsFuture::value, typename R::Return>::type
 329 FutureBase<T>::thenImplementation(
 330     F&& func,
 331     futures::detail::argResult<isTry, F, Args...>) {
 332   static_assert(sizeof...(Args) <= 1, "Then must take zero/one argument");
 333   typedef typename R::ReturnsFuture::Inner B;
 334   this->throwIfInvalid();
 335
 336   Promise<B> p;
 337   p.core_->setInterruptHandlerNoLock(this->core_->getInterruptHandler());
 338
 339   // grab the Future now before we lose our handle on the Promise
 340   auto f = p.getFuture();
 341   f.core_->setExecutorNoLock(this->getExecutor());
 342
 343   this->setCallback_(
 344       [state = futures::detail::makeCoreCallbackState(
 345            std::move(p), std::forward<F>(func))](Try<T>&& t) mutable {
 346         if (!isTry && t.hasException()) {
 347           state.setException(std::move(t.exception()));
 348         } else {
 349           auto tf2 = state.tryInvoke(t.template get<isTry, Args>()...);
 350           if (tf2.hasException()) {
 351             state.setException(std::move(tf2.exception()));
 352           } else {
 353             tf2->setCallback_([p = state.stealPromise()](Try<B> && b) mutable {
 354               p.setTry(std::move(b));
 355             });
 356           }
 357         }
 358       });
 359
 360   return f;
 361 }
 362 } // namespace detail
 363 } // namespace futures
 364
 365 template <class T>
 366 SemiFuture<typename std::decay<T>::type> makeSemiFuture(T&& t) {
 367   return makeSemiFuture(Try<typename std::decay<T>::type>(std::forward<T>(t)));
 368 }
 369
 370 // makeSemiFutureWith(SemiFuture<T>()) -> SemiFuture<T>
 371 template <class F>
 372 typename std::enable_if<
 373     isSemiFuture<typename std::result_of<F()>::type>::value,
 374     typename std::result_of<F()>::type>::type
 375 makeSemiFutureWith(F&& func) {
 376   using InnerType =
 377       typename isSemiFuture<typename std::result_of<F()>::type>::Inner;
 378   try {
 379     return std::forward<F>(func)();
 380   } catch (std::exception& e) {
 381     return makeSemiFuture<InnerType>(
 382         exception_wrapper(std::current_exception(), e));
 383   } catch (...) {
 384     return makeSemiFuture<InnerType>(
 385         exception_wrapper(std::current_exception()));
 386   }
 387 }
 388
 389 // makeSemiFutureWith(T()) -> SemiFuture<T>
 390 // makeSemiFutureWith(void()) -> SemiFuture<Unit>
 391 template <class F>
 392 typename std::enable_if<
 393     !(isSemiFuture<typename std::result_of<F()>::type>::value),
 394     SemiFuture<Unit::LiftT<typename std::result_of<F()>::type>>>::type
 395 makeSemiFutureWith(F&& func) {
 396   using LiftedResult = Unit::LiftT<typename std::result_of<F()>::type>;
 397   return makeSemiFuture<LiftedResult>(
 398       makeTryWith([&func]() mutable { return std::forward<F>(func)(); }));
 399 }
 400
 401 template <class T>
 402 SemiFuture<T> makeSemiFuture(std::exception_ptr const& e) {
 403   return makeSemiFuture(Try<T>(e));
 404 }
 405
 406 template <class T>
 407 SemiFuture<T> makeSemiFuture(exception_wrapper ew) {
 408   return makeSemiFuture(Try<T>(std::move(ew)));
 409 }
 410
 411 template <class T, class E>
 412 typename std::
 413     enable_if<std::is_base_of<std::exception, E>::value, SemiFuture<T>>::type
 414     makeSemiFuture(E const& e) {
 415   return makeSemiFuture(Try<T>(make_exception_wrapper<E>(e)));
 416 }
 417
 418 template <class T>
 419 SemiFuture<T> makeSemiFuture(Try<T>&& t) {
 420   return SemiFuture<T>(new futures::detail::Core<T>(std::move(t)));
 421 }
 422
 423 // This must be defined after the constructors to avoid a bug in MSVC
 424 // https://connect.microsoft.com/VisualStudio/feedback/details/3142777/out-of-line-constructor-definition-after-implicit-reference-causes-incorrect-c2244
 425 inline SemiFuture<Unit> makeSemiFuture() {
 426   return makeSemiFuture(Unit{});
 427 }
 428
 429 template <class T>
 430 SemiFuture<T> SemiFuture<T>::makeEmpty() {
 431   return SemiFuture<T>(futures::detail::EmptyConstruct{});
 432 }
 433
 434 template <class T>
 435 SemiFuture<T>::SemiFuture(SemiFuture<T>&& other) noexcept
 436     : futures::detail::FutureBase<T>(std::move(other)) {}
 437
 438 template <class T>
 439 SemiFuture<T>::SemiFuture(Future<T>&& other) noexcept
 440     : futures::detail::FutureBase<T>(std::move(other)) {
 441   // SemiFuture should not have an executor on construction
 442   if (this->core_) {
 443     this->setExecutor(nullptr);
 444   }
 445 }
 446
 447 template <class T>
 448 SemiFuture<T>& SemiFuture<T>::operator=(SemiFuture<T>&& other) noexcept {
 449   this->assign(other);
 450   return *this;
 451 }
 452
 453 template <class T>
 454 SemiFuture<T>& SemiFuture<T>::operator=(Future<T>&& other) noexcept {
 455   this->assign(other);
 456   // SemiFuture should not have an executor on construction
 457   if (this->core_) {
 458     this->setExecutor(nullptr);
 459   }
 460   return *this;
 461 }
 462
 463 template <class T>
 464 void SemiFuture<T>::boost_() {
 465   // If a SemiFuture has an executor it should be deferred, so boost it
 466   if (auto e = this->getExecutor()) {
 467     // We know in a SemiFuture that if we have an executor it should be
 468     // DeferredExecutor. Verify this in debug mode.
 469     DCHECK(nullptr != dynamic_cast<DeferredExecutor*>(e));
 470
 471     auto ka = static_cast<DeferredExecutor*>(e)->getKeepAliveToken();
 472     static_cast<DeferredExecutor*>(e)->boost();
 473   }
 474 }
 475
 476 template <class T>
 477 inline Future<T> SemiFuture<T>::via(Executor* executor, int8_t priority) && {
 478   throwIfInvalid();
 479   if (!executor) {
 480     throwNoExecutor();
 481   }
 482
 483   // If current executor is deferred, boost block to ensure that work
 484   // progresses and is run on the new executor.
 485   auto oldExecutor = this->getExecutor();
 486   if (oldExecutor && executor && (executor != oldExecutor)) {
 487     // We know in a SemiFuture that if we have an executor it should be
 488     // DeferredExecutor. Verify this in debug mode.
 489     DCHECK(nullptr != dynamic_cast<DeferredExecutor*>(this->getExecutor()));
 490     if (static_cast<DeferredExecutor*>(oldExecutor)) {
 491       executor->add([oldExecutorKA = oldExecutor->getKeepAliveToken()]() {
 492         static_cast<DeferredExecutor*>(oldExecutorKA.get())->boost();
 493       });
 494     }
 495   }
 496
 497   this->setExecutor(executor, priority);
 498
 499   auto newFuture = Future<T>(this->core_);
 500   this->core_ = nullptr;
 501   return newFuture;
 502 }
 503
 504 template <class T>
 505 template <typename F>
 506 SemiFuture<typename futures::detail::callableResult<T, F>::Return::value_type>
 507 SemiFuture<T>::defer(F&& func) && {
 508   // If we already have a deferred executor, use it, otherwise create one
 509   auto defKeepAlive = this->getExecutor()
 510       ? this->getExecutor()->getKeepAliveToken()
 511       : DeferredExecutor::create();
 512   auto e = defKeepAlive.get();
 513   // We know in a SemiFuture that if we have an executor it should be
 514   // DeferredExecutor (either it was that way before, or we just created it).
 515   // Verify this in debug mode.
 516   DCHECK(nullptr != dynamic_cast<DeferredExecutor*>(e));
 517   // Convert to a folly::future with a deferred executor
 518   // Will be low-cost if this is not a new executor as via optimises for that
 519   // case
 520   auto sf =
 521       std::move(*this)
 522           .via(e)
 523           // Then add the work, with a wrapper function that captures the
 524           // keepAlive so the executor is destroyed at the right time.
 525           .then(
 526               DeferredExecutor::wrap(std::move(defKeepAlive), std::move(func)))
 527           // Finally, convert back o a folly::SemiFuture to hide the executor
 528           .semi();
 529   // Carry deferred executor through chain as constructor from Future will
 530   // nullify it
 531   sf.setExecutor(e);
 532   return sf;
 533 }
 534
 535 template <class T>
 536 Future<T> Future<T>::makeEmpty() {
 537   return Future<T>(futures::detail::EmptyConstruct{});
 538 }
 539
 540 template <class T>
 541 Future<T>::Future(Future<T>&& other) noexcept
 542     : futures::detail::FutureBase<T>(std::move(other)) {}
 543
 544 template <class T>
 545 Future<T>& Future<T>::operator=(Future<T>&& other) noexcept {
 546   this->assign(other);
 547   return *this;
 548 }
 549
 550 template <class T>
 551 template <
 552     class T2,
 553     typename std::enable_if<
 554         !std::is_same<T, typename std::decay<T2>::type>::value &&
 555             std::is_constructible<T, T2&&>::value &&
 556             std::is_convertible<T2&&, T>::value,
 557         int>::type>
 558 Future<T>::Future(Future<T2>&& other)
 559     : Future(std::move(other).then([](T2&& v) { return T(std::move(v)); })) {}
 560
 561 template <class T>
 562 template <
 563     class T2,
 564     typename std::enable_if<
 565         !std::is_same<T, typename std::decay<T2>::type>::value &&
 566             std::is_constructible<T, T2&&>::value &&
 567             !std::is_convertible<T2&&, T>::value,
 568         int>::type>
 569 Future<T>::Future(Future<T2>&& other)
 570     : Future(std::move(other).then([](T2&& v) { return T(std::move(v)); })) {}
 571
 572 template <class T>
 573 template <
 574     class T2,
 575     typename std::enable_if<
 576         !std::is_same<T, typename std::decay<T2>::type>::value &&
 577             std::is_constructible<T, T2&&>::value,
 578         int>::type>
 579 Future<T>& Future<T>::operator=(Future<T2>&& other) {
 580   return operator=(
 581       std::move(other).then([](T2&& v) { return T(std::move(v)); }));
 582 }
 583
 584 // unwrap
 585
 586 template <class T>
 587 template <class F>
 588 typename std::
 589     enable_if<isFuture<F>::value, Future<typename isFuture<T>::Inner>>::type
 590     Future<T>::unwrap() {
 591   return then([](Future<typename isFuture<T>::Inner> internal_future) {
 592     return internal_future;
 593   });
 594 }
 595
 596 template <class T>
 597 inline Future<T> Future<T>::via(Executor* executor, int8_t priority) && {
 598   this->throwIfInvalid();
 599
 600   this->setExecutor(executor, priority);
 601
 602   auto newFuture = Future<T>(this->core_);
 603   this->core_ = nullptr;
 604   return newFuture;
 605 }
 606
 607 template <class T>
 608 inline Future<T> Future<T>::via(Executor* executor, int8_t priority) & {
 609   this->throwIfInvalid();
 610   Promise<T> p;
 611   auto f = p.getFuture();
 612   auto func = [p = std::move(p)](Try<T>&& t) mutable {
 613     p.setTry(std::move(t));
 614   };
 615   using R = futures::detail::callableResult<T, decltype(func)>;
 616   this->template thenImplementation<decltype(func), R>(
 617       std::move(func), typename R::Arg());
 618   return std::move(f).via(executor, priority);
 619 }
 620
 621 template <typename T>
 622 template <typename R, typename Caller, typename... Args>
 623   Future<typename isFuture<R>::Inner>
 624 Future<T>::then(R(Caller::*func)(Args...), Caller *instance) {
 625   typedef typename std::remove_cv<typename std::remove_reference<
 626       typename futures::detail::ArgType<Args...>::FirstArg>::type>::type
 627       FirstArg;
 628
 629   return then([instance, func](Try<T>&& t){
 630     return (instance->*func)(t.template get<isTry<FirstArg>::value, Args>()...);
 631   });
 632 }
 633
 634 template <class T>
 635 Future<Unit> Future<T>::then() {
 636   return then([] () {});
 637 }
 638
 639 // onError where the callback returns T
 640 template <class T>
 641 template <class F>
 642 typename std::enable_if<
 643     !futures::detail::callableWith<F, exception_wrapper>::value &&
 644         !futures::detail::callableWith<F, exception_wrapper&>::value &&
 645         !futures::detail::Extract<F>::ReturnsFuture::value,
 646     Future<T>>::type
 647 Future<T>::onError(F&& func) {
 648   typedef std::remove_reference_t<
 649       typename futures::detail::Extract<F>::FirstArg>
 650       Exn;
 651   static_assert(
 652       std::is_same<typename futures::detail::Extract<F>::RawReturn, T>::value,
 653       "Return type of onError callback must be T or Future<T>");
 654
 655   Promise<T> p;
 656   p.core_->setInterruptHandlerNoLock(this->core_->getInterruptHandler());
 657   auto f = p.getFuture();
 658
 659   this->setCallback_(
 660       [state = futures::detail::makeCoreCallbackState(
 661            std::move(p), std::forward<F>(func))](Try<T>&& t) mutable {
 662         if (auto e = t.template tryGetExceptionObject<Exn>()) {
 663           state.setTry(makeTryWith([&] { return state.invoke(*e); }));
 664         } else {
 665           state.setTry(std::move(t));
 666         }
 667       });
 668
 669   return f;
 670 }
 671
 672 // onError where the callback returns Future<T>
 673 template <class T>
 674 template <class F>
 675 typename std::enable_if<
 676     !futures::detail::callableWith<F, exception_wrapper>::value &&
 677         !futures::detail::callableWith<F, exception_wrapper&>::value &&
 678         futures::detail::Extract<F>::ReturnsFuture::value,
 679     Future<T>>::type
 680 Future<T>::onError(F&& func) {
 681   static_assert(
 682       std::is_same<typename futures::detail::Extract<F>::Return, Future<T>>::
 683           value,
 684       "Return type of onError callback must be T or Future<T>");
 685   typedef std::remove_reference_t<
 686       typename futures::detail::Extract<F>::FirstArg>
 687       Exn;
 688
 689   Promise<T> p;
 690   auto f = p.getFuture();
 691
 692   this->setCallback_(
 693       [state = futures::detail::makeCoreCallbackState(
 694            std::move(p), std::forward<F>(func))](Try<T>&& t) mutable {
 695         if (auto e = t.template tryGetExceptionObject<Exn>()) {
 696           auto tf2 = state.tryInvoke(*e);
 697           if (tf2.hasException()) {
 698             state.setException(std::move(tf2.exception()));
 699           } else {
 700             tf2->setCallback_([p = state.stealPromise()](Try<T> && t3) mutable {
 701               p.setTry(std::move(t3));
 702             });
 703           }
 704         } else {
 705           state.setTry(std::move(t));
 706         }
 707       });
 708
 709   return f;
 710 }
 711
 712 template <class T>
 713 template <class F>
 714 Future<T> Future<T>::ensure(F&& func) {
 715   return this->then([funcw = std::forward<F>(func)](Try<T> && t) mutable {
 716     std::move(funcw)();
 717     return makeFuture(std::move(t));
 718   });
 719 }
 720
 721 template <class T>
 722 template <class F>
 723 Future<T> Future<T>::onTimeout(Duration dur, F&& func, Timekeeper* tk) {
 724   return within(dur, tk).onError([funcw = std::forward<F>(func)](
 725       TimedOut const&) { return std::move(funcw)(); });
 726 }
 727
 728 template <class T>
 729 template <class F>
 730 typename std::enable_if<
 731     futures::detail::callableWith<F, exception_wrapper>::value &&
 732         futures::detail::Extract<F>::ReturnsFuture::value,
 733     Future<T>>::type
 734 Future<T>::onError(F&& func) {
 735   static_assert(
 736       std::is_same<typename futures::detail::Extract<F>::Return, Future<T>>::
 737           value,
 738       "Return type of onError callback must be T or Future<T>");
 739
 740   Promise<T> p;
 741   auto f = p.getFuture();
 742   this->setCallback_(
 743       [state = futures::detail::makeCoreCallbackState(
 744            std::move(p), std::forward<F>(func))](Try<T> t) mutable {
 745         if (t.hasException()) {
 746           auto tf2 = state.tryInvoke(std::move(t.exception()));
 747           if (tf2.hasException()) {
 748             state.setException(std::move(tf2.exception()));
 749           } else {
 750             tf2->setCallback_([p = state.stealPromise()](Try<T> && t3) mutable {
 751               p.setTry(std::move(t3));
 752             });
 753           }
 754         } else {
 755           state.setTry(std::move(t));
 756         }
 757       });
 758
 759   return f;
 760 }
 761
 762 // onError(exception_wrapper) that returns T
 763 template <class T>
 764 template <class F>
 765 typename std::enable_if<
 766     futures::detail::callableWith<F, exception_wrapper>::value &&
 767         !futures::detail::Extract<F>::ReturnsFuture::value,
 768     Future<T>>::type
 769 Future<T>::onError(F&& func) {
 770   static_assert(
 771       std::is_same<typename futures::detail::Extract<F>::Return, Future<T>>::
 772           value,
 773       "Return type of onError callback must be T or Future<T>");
 774
 775   Promise<T> p;
 776   auto f = p.getFuture();
 777   this->setCallback_(
 778       [state = futures::detail::makeCoreCallbackState(
 779            std::move(p), std::forward<F>(func))](Try<T>&& t) mutable {
 780         if (t.hasException()) {
 781           state.setTry(makeTryWith(
 782               [&] { return state.invoke(std::move(t.exception())); }));
 783         } else {
 784           state.setTry(std::move(t));
 785         }
 786       });
 787
 788   return f;
 789 }
 790
 791 template <class Func>
 792 auto via(Executor* x, Func&& func)
 793     -> Future<typename isFuture<decltype(std::declval<Func>()())>::Inner> {
 794   // TODO make this actually more performant. :-P #7260175
 795   return via(x).then(std::forward<Func>(func));
 796 }
 797
 798 // makeFuture
 799
 800 template <class T>
 801 Future<typename std::decay<T>::type> makeFuture(T&& t) {
 802   return makeFuture(Try<typename std::decay<T>::type>(std::forward<T>(t)));
 803 }
 804
 805 inline Future<Unit> makeFuture() {
 806   return makeFuture(Unit{});
 807 }
 808
 809 // makeFutureWith(Future<T>()) -> Future<T>
 810 template <class F>
 811 typename std::enable_if<isFuture<typename std::result_of<F()>::type>::value,
 812                         typename std::result_of<F()>::type>::type
 813 makeFutureWith(F&& func) {
 814   using InnerType =
 815       typename isFuture<typename std::result_of<F()>::type>::Inner;
 816   try {
 817     return std::forward<F>(func)();
 818   } catch (std::exception& e) {
 819     return makeFuture<InnerType>(
 820         exception_wrapper(std::current_exception(), e));
 821   } catch (...) {
 822     return makeFuture<InnerType>(exception_wrapper(std::current_exception()));
 823   }
 824 }
 825
 826 // makeFutureWith(T()) -> Future<T>
 827 // makeFutureWith(void()) -> Future<Unit>
 828 template <class F>
 829 typename std::enable_if<
 830     !(isFuture<typename std::result_of<F()>::type>::value),
 831     Future<Unit::LiftT<typename std::result_of<F()>::type>>>::type
 832 makeFutureWith(F&& func) {
 833   using LiftedResult = Unit::LiftT<typename std::result_of<F()>::type>;
 834   return makeFuture<LiftedResult>(
 835       makeTryWith([&func]() mutable { return std::forward<F>(func)(); }));
 836 }
 837
 838 template <class T>
 839 Future<T> makeFuture(std::exception_ptr const& e) {
 840   return makeFuture(Try<T>(e));
 841 }
 842
 843 template <class T>
 844 Future<T> makeFuture(exception_wrapper ew) {
 845   return makeFuture(Try<T>(std::move(ew)));
 846 }
 847
 848 template <class T, class E>
 849 typename std::enable_if<std::is_base_of<std::exception, E>::value,
 850                         Future<T>>::type
 851 makeFuture(E const& e) {
 852   return makeFuture(Try<T>(make_exception_wrapper<E>(e)));
 853 }
 854
 855 template <class T>
 856 Future<T> makeFuture(Try<T>&& t) {
 857   return Future<T>(new futures::detail::Core<T>(std::move(t)));
 858 }
 859
 860 // via
 861 Future<Unit> via(Executor* executor, int8_t priority) {
 862   return makeFuture().via(executor, priority);
 863 }
 864
 865 // mapSetCallback calls func(i, Try<T>) when every future completes
 866
 867 template <class T, class InputIterator, class F>
 868 void mapSetCallback(InputIterator first, InputIterator last, F func) {
 869   for (size_t i = 0; first != last; ++first, ++i) {
 870     first->setCallback_([func, i](Try<T>&& t) {
 871       func(i, std::move(t));
 872     });
 873   }
 874 }
 875
 876 // collectAll (variadic)
 877
 878 template <typename... Fs>
 879 typename futures::detail::CollectAllVariadicContext<
 880     typename std::decay<Fs>::type::value_type...>::type
 881 collectAll(Fs&&... fs) {
 882   auto ctx = std::make_shared<futures::detail::CollectAllVariadicContext<
 883       typename std::decay<Fs>::type::value_type...>>();
 884   futures::detail::collectVariadicHelper<
 885       futures::detail::CollectAllVariadicContext>(ctx, std::forward<Fs>(fs)...);
 886   return ctx->p.getFuture();
 887 }
 888
 889 // collectAll (iterator)
 890
 891 template <class InputIterator>
 892 Future<
 893   std::vector<
 894   Try<typename std::iterator_traits<InputIterator>::value_type::value_type>>>
 895 collectAll(InputIterator first, InputIterator last) {
 896   typedef
 897     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 898
 899   struct CollectAllContext {
 900     CollectAllContext(size_t n) : results(n) {}
 901     ~CollectAllContext() {
 902       p.setValue(std::move(results));
 903     }
 904     Promise<std::vector<Try<T>>> p;
 905     std::vector<Try<T>> results;
 906   };
 907
 908   auto ctx =
 909       std::make_shared<CollectAllContext>(size_t(std::distance(first, last)));
 910   mapSetCallback<T>(first, last, [ctx](size_t i, Try<T>&& t) {
 911     ctx->results[i] = std::move(t);
 912   });
 913   return ctx->p.getFuture();
 914 }
 915
 916 // collect (iterator)
 917
 918 namespace futures {
 919 namespace detail {
 920
 921 template <typename T>
 922 struct CollectContext {
 923   struct Nothing {
 924     explicit Nothing(int /* n */) {}
 925   };
 926
 927   using Result = typename std::conditional<
 928     std::is_void<T>::value,
 929     void,
 930     std::vector<T>>::type;
 931
 932   using InternalResult = typename std::conditional<
 933     std::is_void<T>::value,
 934     Nothing,
 935     std::vector<Optional<T>>>::type;
 936
 937   explicit CollectContext(size_t n) : result(n) {}
 938   ~CollectContext() {
 939     if (!threw.exchange(true)) {
 940       // map Optional<T> -> T
 941       std::vector<T> finalResult;
 942       finalResult.reserve(result.size());
 943       std::transform(result.begin(), result.end(),
 944                      std::back_inserter(finalResult),
 945                      [](Optional<T>& o) { return std::move(o.value()); });
 946       p.setValue(std::move(finalResult));
 947     }
 948   }
 949   inline void setPartialResult(size_t i, Try<T>& t) {
 950     result[i] = std::move(t.value());
 951   }
 952   Promise<Result> p;
 953   InternalResult result;
 954   std::atomic<bool> threw {false};
 955 };
 956
 957 } // namespace detail
 958 } // namespace futures
 959
 960 template <class InputIterator>
 961 Future<typename futures::detail::CollectContext<typename std::iterator_traits<
 962     InputIterator>::value_type::value_type>::Result>
 963 collect(InputIterator first, InputIterator last) {
 964   typedef
 965     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 966
 967   auto ctx = std::make_shared<futures::detail::CollectContext<T>>(
 968       std::distance(first, last));
 969   mapSetCallback<T>(first, last, [ctx](size_t i, Try<T>&& t) {
 970     if (t.hasException()) {
 971        if (!ctx->threw.exchange(true)) {
 972          ctx->p.setException(std::move(t.exception()));
 973        }
 974      } else if (!ctx->threw) {
 975        ctx->setPartialResult(i, t);
 976      }
 977   });
 978   return ctx->p.getFuture();
 979 }
 980
 981 // collect (variadic)
 982
 983 template <typename... Fs>
 984 typename futures::detail::CollectVariadicContext<
 985     typename std::decay<Fs>::type::value_type...>::type
 986 collect(Fs&&... fs) {
 987   auto ctx = std::make_shared<futures::detail::CollectVariadicContext<
 988       typename std::decay<Fs>::type::value_type...>>();
 989   futures::detail::collectVariadicHelper<
 990       futures::detail::CollectVariadicContext>(ctx, std::forward<Fs>(fs)...);
 991   return ctx->p.getFuture();
 992 }
 993
 994 // collectAny (iterator)
 995
 996 template <class InputIterator>
 997 Future<
 998   std::pair<size_t,
 999             Try<
1000               typename
1001               std::iterator_traits<InputIterator>::value_type::value_type>>>
1002 collectAny(InputIterator first, InputIterator last) {
1003   typedef
1004     typename std::iterator_traits<InputIterator>::value_type::value_type T;
1005
1006   struct CollectAnyContext {
1007     CollectAnyContext() {}
1008     Promise<std::pair<size_t, Try<T>>> p;
1009     std::atomic<bool> done {false};
1010   };
1011
1012   auto ctx = std::make_shared<CollectAnyContext>();
1013   mapSetCallback<T>(first, last, [ctx](size_t i, Try<T>&& t) {
1014     if (!ctx->done.exchange(true)) {
1015       ctx->p.setValue(std::make_pair(i, std::move(t)));
1016     }
1017   });
1018   return ctx->p.getFuture();
1019 }
1020
1021 // collectAnyWithoutException (iterator)
1022
1023 template <class InputIterator>
1024 Future<std::pair<
1025     size_t,
1026     typename std::iterator_traits<InputIterator>::value_type::value_type>>
1027 collectAnyWithoutException(InputIterator first, InputIterator last) {
1028   typedef
1029       typename std::iterator_traits<InputIterator>::value_type::value_type T;
1030
1031   struct CollectAnyWithoutExceptionContext {
1032     CollectAnyWithoutExceptionContext(){}
1033     Promise<std::pair<size_t, T>> p;
1034     std::atomic<bool> done{false};
1035     std::atomic<size_t> nFulfilled{0};
1036     size_t nTotal;
1037   };
1038
1039   auto ctx = std::make_shared<CollectAnyWithoutExceptionContext>();
1040   ctx->nTotal = size_t(std::distance(first, last));
1041
1042   mapSetCallback<T>(first, last, [ctx](size_t i, Try<T>&& t) {
1043     if (!t.hasException() && !ctx->done.exchange(true)) {
1044       ctx->p.setValue(std::make_pair(i, std::move(t.value())));
1045     } else if (++ctx->nFulfilled == ctx->nTotal) {
1046       ctx->p.setException(t.exception());
1047     }
1048   });
1049   return ctx->p.getFuture();
1050 }
1051
1052 // collectN (iterator)
1053
1054 template <class InputIterator>
1055 Future<std::vector<std::pair<size_t, Try<typename
1056   std::iterator_traits<InputIterator>::value_type::value_type>>>>
1057 collectN(InputIterator first, InputIterator last, size_t n) {
1058   typedef typename
1059     std::iterator_traits<InputIterator>::value_type::value_type T;
1060   typedef std::vector<std::pair<size_t, Try<T>>> V;
1061
1062   struct CollectNContext {
1063     V v;
1064     std::atomic<size_t> completed = {0};
1065     Promise<V> p;
1066   };
1067   auto ctx = std::make_shared<CollectNContext>();
1068
1069   if (size_t(std::distance(first, last)) < n) {
1070     ctx->p.setException(std::runtime_error("Not enough futures"));
1071   } else {
1072     // for each completed Future, increase count and add to vector, until we
1073     // have n completed futures at which point we fulfil our Promise with the
1074     // vector
1075     mapSetCallback<T>(first, last, [ctx, n](size_t i, Try<T>&& t) {
1076       auto c = ++ctx->completed;
1077       if (c <= n) {
1078         assert(ctx->v.size() < n);
1079         ctx->v.emplace_back(i, std::move(t));
1080         if (c == n) {
1081           ctx->p.setTry(Try<V>(std::move(ctx->v)));
1082         }
1083       }
1084     });
1085   }
1086
1087   return ctx->p.getFuture();
1088 }
1089
1090 // reduce (iterator)
1091
1092 template <class It, class T, class F>
1093 Future<T> reduce(It first, It last, T&& initial, F&& func) {
1094   if (first == last) {
1095     return makeFuture(std::move(initial));
1096   }
1097
1098   typedef typename std::iterator_traits<It>::value_type::value_type ItT;
1099   typedef typename std::conditional<
1100       futures::detail::callableWith<F, T&&, Try<ItT>&&>::value,
1101       Try<ItT>,
1102       ItT>::type Arg;
1103   typedef isTry<Arg> IsTry;
1104
1105   auto sfunc = std::make_shared<F>(std::move(func));
1106
1107   auto f = first->then(
1108       [ minitial = std::move(initial), sfunc ](Try<ItT> & head) mutable {
1109         return (*sfunc)(
1110             std::move(minitial), head.template get<IsTry::value, Arg&&>());
1111       });
1112
1113   for (++first; first != last; ++first) {
1114     f = collectAll(f, *first).then([sfunc](std::tuple<Try<T>, Try<ItT>>& t) {
1115       return (*sfunc)(std::move(std::get<0>(t).value()),
1116                   // Either return a ItT&& or a Try<ItT>&& depending
1117                   // on the type of the argument of func.
1118                   std::get<1>(t).template get<IsTry::value, Arg&&>());
1119     });
1120   }
1121
1122   return f;
1123 }
1124
1125 // window (collection)
1126
1127 template <class Collection, class F, class ItT, class Result>
1128 std::vector<Future<Result>>
1129 window(Collection input, F func, size_t n) {
1130   // Use global inline executor singleton
1131   auto executor = &InlineExecutor::instance();
1132   return window(executor, std::move(input), std::move(func), n);
1133 }
1134
1135 template <class Collection, class F, class ItT, class Result>
1136 std::vector<Future<Result>>
1137 window(Executor* executor, Collection input, F func, size_t n) {
1138   struct WindowContext {
1139     WindowContext(Executor* executor_, Collection&& input_, F&& func_)
1140         : executor(executor_),
1141           input(std::move(input_)),
1142           promises(input.size()),
1143           func(std::move(func_)) {}
1144     std::atomic<size_t> i{0};
1145     Executor* executor;
1146     Collection input;
1147     std::vector<Promise<Result>> promises;
1148     F func;
1149
1150     static inline void spawn(std::shared_ptr<WindowContext> ctx) {
1151       size_t i = ctx->i++;
1152       if (i < ctx->input.size()) {
1153         auto fut = ctx->func(std::move(ctx->input[i]));
1154         fut.setCallback_([ctx = std::move(ctx), i](Try<Result>&& t) mutable {
1155           const auto executor_ = ctx->executor;
1156           executor_->add([ctx = std::move(ctx), i, t = std::move(t)]() mutable {
1157             ctx->promises[i].setTry(std::move(t));
1158             // Chain another future onto this one
1159             spawn(std::move(ctx));
1160           });
1161         });
1162       }
1163     }
1164   };
1165
1166   auto max = std::min(n, input.size());
1167
1168   auto ctx = std::make_shared<WindowContext>(
1169       executor, std::move(input), std::move(func));
1170
1171   // Start the first n Futures
1172   for (size_t i = 0; i < max; ++i) {
1173     executor->add([ctx]() mutable { WindowContext::spawn(std::move(ctx)); });
1174   }
1175
1176   std::vector<Future<Result>> futures;
1177   futures.reserve(ctx->promises.size());
1178   for (auto& promise : ctx->promises) {
1179     futures.emplace_back(promise.getFuture());
1180   }
1181
1182   return futures;
1183 }
1184
1185 // reduce
1186
1187 template <class T>
1188 template <class I, class F>
1189 Future<I> Future<T>::reduce(I&& initial, F&& func) {
1190   return then([
1191     minitial = std::forward<I>(initial),
1192     mfunc = std::forward<F>(func)
1193   ](T& vals) mutable {
1194     auto ret = std::move(minitial);
1195     for (auto& val : vals) {
1196       ret = mfunc(std::move(ret), std::move(val));
1197     }
1198     return ret;
1199   });
1200 }
1201
1202 // unorderedReduce (iterator)
1203
1204 template <class It, class T, class F, class ItT, class Arg>
1205 Future<T> unorderedReduce(It first, It last, T initial, F func) {
1206   if (first == last) {
1207     return makeFuture(std::move(initial));
1208   }
1209
1210   typedef isTry<Arg> IsTry;
1211
1212   struct UnorderedReduceContext {
1213     UnorderedReduceContext(T&& memo, F&& fn, size_t n)
1214         : lock_(), memo_(makeFuture<T>(std::move(memo))),
1215           func_(std::move(fn)), numThens_(0), numFutures_(n), promise_()
1216       {}
1217     folly::MicroSpinLock lock_; // protects memo_ and numThens_
1218     Future<T> memo_;
1219     F func_;
1220     size_t numThens_; // how many Futures completed and called .then()
1221     size_t numFutures_; // how many Futures in total
1222     Promise<T> promise_;
1223   };
1224
1225   auto ctx = std::make_shared<UnorderedReduceContext>(
1226     std::move(initial), std::move(func), std::distance(first, last));
1227
1228   mapSetCallback<ItT>(
1229       first,
1230       last,
1231       [ctx](size_t /* i */, Try<ItT>&& t) {
1232         // Futures can be completed in any order, simultaneously.
1233         // To make this non-blocking, we create a new Future chain in
1234         // the order of completion to reduce the values.
1235         // The spinlock just protects chaining a new Future, not actually
1236         // executing the reduce, which should be really fast.
1237         folly::MSLGuard lock(ctx->lock_);
1238         ctx->memo_ =
1239             ctx->memo_.then([ ctx, mt = std::move(t) ](T && v) mutable {
1240               // Either return a ItT&& or a Try<ItT>&& depending
1241               // on the type of the argument of func.
1242               return ctx->func_(std::move(v),
1243                                 mt.template get<IsTry::value, Arg&&>());
1244             });
1245         if (++ctx->numThens_ == ctx->numFutures_) {
1246           // After reducing the value of the last Future, fulfill the Promise
1247           ctx->memo_.setCallback_(
1248               [ctx](Try<T>&& t2) { ctx->promise_.setValue(std::move(t2)); });
1249         }
1250       });
1251
1252   return ctx->promise_.getFuture();
1253 }
1254
1255 // within
1256
1257 template <class T>
1258 Future<T> Future<T>::within(Duration dur, Timekeeper* tk) {
1259   return within(dur, TimedOut(), tk);
1260 }
1261
1262 template <class T>
1263 template <class E>
1264 Future<T> Future<T>::within(Duration dur, E e, Timekeeper* tk) {
1265
1266   struct Context {
1267     Context(E ex) : exception(std::move(ex)), promise() {}
1268     E exception;
1269     Future<Unit> thisFuture;
1270     Promise<T> promise;
1271     std::atomic<bool> token {false};
1272   };
1273
1274   if (this->isReady()) {
1275     return std::move(*this);
1276   }
1277
1278   std::shared_ptr<Timekeeper> tks;
1279   if (LIKELY(!tk)) {
1280     tks = folly::detail::getTimekeeperSingleton();
1281     tk = tks.get();
1282   }
1283
1284   if (UNLIKELY(!tk)) {
1285     return makeFuture<T>(NoTimekeeper());
1286   }
1287
1288   auto ctx = std::make_shared<Context>(std::move(e));
1289
1290   ctx->thisFuture = this->then([ctx](Try<T>&& t) mutable {
1291     if (ctx->token.exchange(true) == false) {
1292       ctx->promise.setTry(std::move(t));
1293     }
1294   });
1295
1296   // Have time keeper use a weak ptr to hold ctx,
1297   // so that ctx can be deallocated as soon as the future job finished.
1298   tk->after(dur).then([weakCtx = to_weak_ptr(ctx)](Try<Unit> const& t) mutable {
1299     auto lockedCtx = weakCtx.lock();
1300     if (!lockedCtx) {
1301       // ctx already released. "this" completed first, cancel "after"
1302       return;
1303     }
1304     // "after" completed first, cancel "this"
1305     lockedCtx->thisFuture.raise(TimedOut());
1306     if (lockedCtx->token.exchange(true) == false) {
1307       if (t.hasException()) {
1308         lockedCtx->promise.setException(std::move(t.exception()));
1309       } else {
1310         lockedCtx->promise.setException(std::move(lockedCtx->exception));
1311       }
1312     }
1313   });
1314
1315   return ctx->promise.getFuture().via(this->getExecutor());
1316 }
1317
1318 // delayed
1319
1320 template <class T>
1321 Future<T> Future<T>::delayed(Duration dur, Timekeeper* tk) {
1322   return collectAll(*this, futures::sleep(dur, tk))
1323       .then([](std::tuple<Try<T>, Try<Unit>> tup) {
1324         Try<T>& t = std::get<0>(tup);
1325         return makeFuture<T>(std::move(t));
1326       });
1327 }
1328
1329 namespace futures {
1330 namespace detail {
1331
1332 template <class T>
1333 void doBoost(folly::Future<T>& /* usused */) {}
1334
1335 template <class T>
1336 void doBoost(folly::SemiFuture<T>& f) {
1337   f.boost_();
1338 }
1339
1340 template <class FutureType, typename T = typename FutureType::value_type>
1341 void waitImpl(FutureType& f) {
1342   // short-circuit if there's nothing to do
1343   if (f.isReady()) {
1344     return;
1345   }
1346
1347   FutureBatonType baton;
1348   f.setCallback_([&](const Try<T>& /* t */) { baton.post(); });
1349   doBoost(f);
1350   baton.wait();
1351   assert(f.isReady());
1352 }
1353
1354 template <class FutureType, typename T = typename FutureType::value_type>
1355 void waitImpl(FutureType& f, Duration dur) {
1356   // short-circuit if there's nothing to do
1357   if (f.isReady()) {
1358     return;
1359   }
1360
1361   Promise<T> promise;
1362   auto ret = promise.getFuture();
1363   auto baton = std::make_shared<FutureBatonType>();
1364   f.setCallback_([baton, promise = std::move(promise)](Try<T>&& t) mutable {
1365     promise.setTry(std::move(t));
1366     baton->post();
1367   });
1368   doBoost(f);
1369   f = std::move(ret);
1370   if (baton->try_wait_for(dur)) {
1371     assert(f.isReady());
1372   }
1373 }
1374
1375 template <class T>
1376 void waitViaImpl(Future<T>& f, DrivableExecutor* e) {
1377   // Set callback so to ensure that the via executor has something on it
1378   // so that once the preceding future triggers this callback, drive will
1379   // always have a callback to satisfy it
1380   if (f.isReady()) {
1381     return;
1382   }
1383   f = f.via(e).then([](T&& t) { return std::move(t); });
1384   while (!f.isReady()) {
1385     e->drive();
1386   }
1387   assert(f.isReady());
1388 }
1389
1390 template <class T>
1391 void waitViaImpl(SemiFuture<T>& f, DrivableExecutor* e) {
1392   // Set callback so to ensure that the via executor has something on it
1393   // so that once the preceding future triggers this callback, drive will
1394   // always have a callback to satisfy it
1395   if (f.isReady()) {
1396     return;
1397   }
1398   f = std::move(f).via(e).then([](T&& t) { return std::move(t); });
1399   while (!f.isReady()) {
1400     e->drive();
1401   }
1402   assert(f.isReady());
1403 }
1404
1405 } // namespace detail
1406 } // namespace futures
1407
1408 template <class T>
1409 SemiFuture<T>& SemiFuture<T>::wait() & {
1410   futures::detail::waitImpl(*this);
1411   return *this;
1412 }
1413
1414 template <class T>
1415 SemiFuture<T>&& SemiFuture<T>::wait() && {
1416   futures::detail::waitImpl(*this);
1417   return std::move(*this);
1418 }
1419
1420 template <class T>
1421 SemiFuture<T>& SemiFuture<T>::wait(Duration dur) & {
1422   futures::detail::waitImpl(*this, dur);
1423   return *this;
1424 }
1425
1426 template <class T>
1427 SemiFuture<T>&& SemiFuture<T>::wait(Duration dur) && {
1428   futures::detail::waitImpl(*this, dur);
1429   return std::move(*this);
1430 }
1431
1432 template <class T>
1433 SemiFuture<T>& SemiFuture<T>::waitVia(DrivableExecutor* e) & {
1434   futures::detail::waitViaImpl(*this, e);
1435   return *this;
1436 }
1437
1438 template <class T>
1439 SemiFuture<T>&& SemiFuture<T>::waitVia(DrivableExecutor* e) && {
1440   futures::detail::waitViaImpl(*this, e);
1441   return std::move(*this);
1442 }
1443
1444 template <class T>
1445 T SemiFuture<T>::get() && {
1446   return std::move(wait().value());
1447 }
1448
1449 template <class T>
1450 T SemiFuture<T>::get(Duration dur) && {
1451   wait(dur);
1452   if (this->isReady()) {
1453     return std::move(this->value());
1454   } else {
1455     throwTimedOut();
1456   }
1457 }
1458
1459 template <class T>
1460 Try<T> SemiFuture<T>::getTry() && {
1461   wait();
1462   return std::move(this->core_->getTry());
1463 }
1464
1465 template <class T>
1466 T SemiFuture<T>::getVia(DrivableExecutor* e) && {
1467   return std::move(waitVia(e).value());
1468 }
1469
1470 template <class T>
1471 Try<T> SemiFuture<T>::getTryVia(DrivableExecutor* e) && {
1472   waitVia(e);
1473   return std::move(this->core_->getTry());
1474 }
1475
1476 template <class T>
1477 Future<T>& Future<T>::wait() & {
1478   futures::detail::waitImpl(*this);
1479   return *this;
1480 }
1481
1482 template <class T>
1483 Future<T>&& Future<T>::wait() && {
1484   futures::detail::waitImpl(*this);
1485   return std::move(*this);
1486 }
1487
1488 template <class T>
1489 Future<T>& Future<T>::wait(Duration dur) & {
1490   futures::detail::waitImpl(*this, dur);
1491   return *this;
1492 }
1493
1494 template <class T>
1495 Future<T>&& Future<T>::wait(Duration dur) && {
1496   futures::detail::waitImpl(*this, dur);
1497   return std::move(*this);
1498 }
1499
1500 template <class T>
1501 Future<T>& Future<T>::waitVia(DrivableExecutor* e) & {
1502   futures::detail::waitViaImpl(*this, e);
1503   return *this;
1504 }
1505
1506 template <class T>
1507 Future<T>&& Future<T>::waitVia(DrivableExecutor* e) && {
1508   futures::detail::waitViaImpl(*this, e);
1509   return std::move(*this);
1510 }
1511
1512 template <class T>
1513 T Future<T>::get() {
1514   return std::move(wait().value());
1515 }
1516
1517 template <class T>
1518 T Future<T>::get(Duration dur) {
1519   wait(dur);
1520   if (this->isReady()) {
1521     return std::move(this->value());
1522   } else {
1523     throwTimedOut();
1524   }
1525 }
1526
1527 template <class T>
1528 Try<T>& Future<T>::getTry() {
1529   throwIfInvalid();
1530
1531   return this->core_->getTry();
1532 }
1533
1534 template <class T>
1535 T Future<T>::getVia(DrivableExecutor* e) {
1536   return std::move(waitVia(e).value());
1537 }
1538
1539 template <class T>
1540 Try<T>& Future<T>::getTryVia(DrivableExecutor* e) {
1541   return waitVia(e).getTry();
1542 }
1543
1544 namespace futures {
1545 namespace detail {
1546 template <class T>
1547 struct TryEquals {
1548   static bool equals(const Try<T>& t1, const Try<T>& t2) {
1549     return t1.value() == t2.value();
1550   }
1551 };
1552 } // namespace detail
1553 } // namespace futures
1554
1555 template <class T>
1556 Future<bool> Future<T>::willEqual(Future<T>& f) {
1557   return collectAll(*this, f).then([](const std::tuple<Try<T>, Try<T>>& t) {
1558     if (std::get<0>(t).hasValue() && std::get<1>(t).hasValue()) {
1559       return futures::detail::TryEquals<T>::equals(
1560           std::get<0>(t), std::get<1>(t));
1561     } else {
1562       return false;
1563       }
1564   });
1565 }
1566
1567 template <class T>
1568 template <class F>
1569 Future<T> Future<T>::filter(F&& predicate) {
1570   return this->then([p = std::forward<F>(predicate)](T val) {
1571     T const& valConstRef = val;
1572     if (!p(valConstRef)) {
1573       throwPredicateDoesNotObtain();
1574     }
1575     return val;
1576   });
1577 }
1578
1579 template <class F>
1580 inline Future<Unit> when(bool p, F&& thunk) {
1581   return p ? std::forward<F>(thunk)().unit() : makeFuture();
1582 }
1583
1584 template <class P, class F>
1585 Future<Unit> whileDo(P&& predicate, F&& thunk) {
1586   if (predicate()) {
1587     auto future = thunk();
1588     return future.then([
1589       predicate = std::forward<P>(predicate),
1590       thunk = std::forward<F>(thunk)
1591     ]() mutable {
1592       return whileDo(std::forward<P>(predicate), std::forward<F>(thunk));
1593     });
1594   }
1595   return makeFuture();
1596 }
1597
1598 template <class F>
1599 Future<Unit> times(const int n, F&& thunk) {
1600   return folly::whileDo(
1601       [ n, count = std::make_unique<std::atomic<int>>(0) ]() mutable {
1602         return count->fetch_add(1) < n;
1603       },
1604       std::forward<F>(thunk));
1605 }
1606
1607 namespace futures {
1608 template <class It, class F, class ItT, class Result>
1609 std::vector<Future<Result>> map(It first, It last, F func) {
1610   std::vector<Future<Result>> results;
1611   for (auto it = first; it != last; it++) {
1612     results.push_back(it->then(func));
1613   }
1614   return results;
1615 }
1616 } // namespace futures
1617
1618 // Instantiate the most common Future types to save compile time
1619 extern template class Future<Unit>;
1620 extern template class Future<bool>;
1621 extern template class Future<int>;
1622 extern template class Future<int64_t>;
1623 extern template class Future<std::string>;
1624 extern template class Future<double>;
1625 } // namespace folly