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