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