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