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