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   std::shared_ptr<Timekeeper> tks;
1071   if (!tk) {
1072     tks = folly::detail::getTimekeeperSingleton();
1073     tk = DCHECK_NOTNULL(tks.get());
1074   }
1075
1076   auto ctx = std::make_shared<Context>(std::move(e));
1077
1078   ctx->thisFuture = this->then([ctx](Try<T>&& t) mutable {
1079     if (ctx->token.exchange(true) == false) {
1080       ctx->promise.setTry(std::move(t));
1081     }
1082   });
1083
1084   // Have time keeper use a weak ptr to hold ctx,
1085   // so that ctx can be deallocated as soon as the future job finished.
1086   tk->after(dur).then([weakCtx = to_weak_ptr(ctx)](Try<Unit> const& t) mutable {
1087     auto lockedCtx = weakCtx.lock();
1088     if (!lockedCtx) {
1089       // ctx already released. "this" completed first, cancel "after"
1090       return;
1091     }
1092     // "after" completed first, cancel "this"
1093     lockedCtx->thisFuture.raise(TimedOut());
1094     if (lockedCtx->token.exchange(true) == false) {
1095       if (t.hasException()) {
1096         lockedCtx->promise.setException(std::move(t.exception()));
1097       } else {
1098         lockedCtx->promise.setException(std::move(lockedCtx->exception));
1099       }
1100     }
1101   });
1102
1103   return ctx->promise.getFuture().via(getExecutor());
1104 }
1105
1106 // delayed
1107
1108 template <class T>
1109 Future<T> Future<T>::delayed(Duration dur, Timekeeper* tk) {
1110   return collectAll(*this, futures::sleep(dur, tk))
1111     .then([](std::tuple<Try<T>, Try<Unit>> tup) {
1112       Try<T>& t = std::get<0>(tup);
1113       return makeFuture<T>(std::move(t));
1114     });
1115 }
1116
1117 namespace futures {
1118 namespace detail {
1119
1120 template <class T>
1121 void waitImpl(Future<T>& f) {
1122   // short-circuit if there's nothing to do
1123   if (f.isReady()) return;
1124
1125   FutureBatonType baton;
1126   f.setCallback_([&](const Try<T>& /* t */) { baton.post(); });
1127   baton.wait();
1128   assert(f.isReady());
1129 }
1130
1131 template <class T>
1132 void waitImpl(Future<T>& f, Duration dur) {
1133   // short-circuit if there's nothing to do
1134   if (f.isReady()) {
1135     return;
1136   }
1137
1138   Promise<T> promise;
1139   auto ret = promise.getFuture();
1140   auto baton = std::make_shared<FutureBatonType>();
1141   f.setCallback_([ baton, promise = std::move(promise) ](Try<T> && t) mutable {
1142     promise.setTry(std::move(t));
1143     baton->post();
1144   });
1145   f = std::move(ret);
1146   if (baton->timed_wait(dur)) {
1147     assert(f.isReady());
1148   }
1149 }
1150
1151 template <class T>
1152 void waitViaImpl(Future<T>& f, DrivableExecutor* e) {
1153   // Set callback so to ensure that the via executor has something on it
1154   // so that once the preceding future triggers this callback, drive will
1155   // always have a callback to satisfy it
1156   if (f.isReady())
1157     return;
1158   f = f.via(e).then([](T&& t) { return std::move(t); });
1159   while (!f.isReady()) {
1160     e->drive();
1161   }
1162   assert(f.isReady());
1163 }
1164
1165 } // namespace detail
1166 } // namespace futures
1167
1168 template <class T>
1169 Future<T>& Future<T>::wait() & {
1170   futures::detail::waitImpl(*this);
1171   return *this;
1172 }
1173
1174 template <class T>
1175 Future<T>&& Future<T>::wait() && {
1176   futures::detail::waitImpl(*this);
1177   return std::move(*this);
1178 }
1179
1180 template <class T>
1181 Future<T>& Future<T>::wait(Duration dur) & {
1182   futures::detail::waitImpl(*this, dur);
1183   return *this;
1184 }
1185
1186 template <class T>
1187 Future<T>&& Future<T>::wait(Duration dur) && {
1188   futures::detail::waitImpl(*this, dur);
1189   return std::move(*this);
1190 }
1191
1192 template <class T>
1193 Future<T>& Future<T>::waitVia(DrivableExecutor* e) & {
1194   futures::detail::waitViaImpl(*this, e);
1195   return *this;
1196 }
1197
1198 template <class T>
1199 Future<T>&& Future<T>::waitVia(DrivableExecutor* e) && {
1200   futures::detail::waitViaImpl(*this, e);
1201   return std::move(*this);
1202 }
1203
1204 template <class T>
1205 T Future<T>::get() {
1206   return std::move(wait().value());
1207 }
1208
1209 template <class T>
1210 T Future<T>::get(Duration dur) {
1211   wait(dur);
1212   if (isReady()) {
1213     return std::move(value());
1214   } else {
1215     throwTimedOut();
1216   }
1217 }
1218
1219 template <class T>
1220 T Future<T>::getVia(DrivableExecutor* e) {
1221   return std::move(waitVia(e).value());
1222 }
1223
1224 namespace futures {
1225 namespace detail {
1226 template <class T>
1227 struct TryEquals {
1228   static bool equals(const Try<T>& t1, const Try<T>& t2) {
1229     return t1.value() == t2.value();
1230   }
1231 };
1232 } // namespace detail
1233 } // namespace futures
1234
1235 template <class T>
1236 Future<bool> Future<T>::willEqual(Future<T>& f) {
1237   return collectAll(*this, f).then([](const std::tuple<Try<T>, Try<T>>& t) {
1238     if (std::get<0>(t).hasValue() && std::get<1>(t).hasValue()) {
1239       return futures::detail::TryEquals<T>::equals(
1240           std::get<0>(t), std::get<1>(t));
1241     } else {
1242       return false;
1243       }
1244   });
1245 }
1246
1247 template <class T>
1248 template <class F>
1249 Future<T> Future<T>::filter(F&& predicate) {
1250   return this->then([p = std::forward<F>(predicate)](T val) {
1251     T const& valConstRef = val;
1252     if (!p(valConstRef)) {
1253       throwPredicateDoesNotObtain();
1254     }
1255     return val;
1256   });
1257 }
1258
1259 template <class F>
1260 inline Future<Unit> when(bool p, F&& thunk) {
1261   return p ? std::forward<F>(thunk)().unit() : makeFuture();
1262 }
1263
1264 template <class P, class F>
1265 Future<Unit> whileDo(P&& predicate, F&& thunk) {
1266   if (predicate()) {
1267     auto future = thunk();
1268     return future.then([
1269       predicate = std::forward<P>(predicate),
1270       thunk = std::forward<F>(thunk)
1271     ]() mutable {
1272       return whileDo(std::forward<P>(predicate), std::forward<F>(thunk));
1273     });
1274   }
1275   return makeFuture();
1276 }
1277
1278 template <class F>
1279 Future<Unit> times(const int n, F&& thunk) {
1280   return folly::whileDo(
1281       [ n, count = std::make_unique<std::atomic<int>>(0) ]() mutable {
1282         return count->fetch_add(1) < n;
1283       },
1284       std::forward<F>(thunk));
1285 }
1286
1287 namespace futures {
1288   template <class It, class F, class ItT, class Result>
1289   std::vector<Future<Result>> map(It first, It last, F func) {
1290     std::vector<Future<Result>> results;
1291     for (auto it = first; it != last; it++) {
1292       results.push_back(it->then(func));
1293     }
1294     return results;
1295   }
1296 }
1297
1298 namespace futures {
1299
1300 namespace detail {
1301
1302 struct retrying_policy_raw_tag {};
1303 struct retrying_policy_fut_tag {};
1304
1305 template <class Policy>
1306 struct retrying_policy_traits {
1307   using result = std::result_of_t<Policy(size_t, const exception_wrapper&)>;
1308   using is_raw = std::is_same<result, bool>;
1309   using is_fut = std::is_same<result, Future<bool>>;
1310   using tag = typename std::conditional<
1311         is_raw::value, retrying_policy_raw_tag, typename std::conditional<
1312         is_fut::value, retrying_policy_fut_tag, void>::type>::type;
1313 };
1314
1315 template <class Policy, class FF, class Prom>
1316 void retryingImpl(size_t k, Policy&& p, FF&& ff, Prom prom) {
1317   using F = typename std::result_of<FF(size_t)>::type;
1318   using T = typename F::value_type;
1319   auto f = makeFutureWith([&] { return ff(k++); });
1320   f.then([
1321     k,
1322     prom = std::move(prom),
1323     pm = std::forward<Policy>(p),
1324     ffm = std::forward<FF>(ff)
1325   ](Try<T> && t) mutable {
1326     if (t.hasValue()) {
1327       prom.setValue(std::move(t).value());
1328       return;
1329     }
1330     auto& x = t.exception();
1331     auto q = pm(k, x);
1332     q.then([
1333       k,
1334       prom = std::move(prom),
1335       xm = std::move(x),
1336       pm = std::move(pm),
1337       ffm = std::move(ffm)
1338     ](bool shouldRetry) mutable {
1339       if (shouldRetry) {
1340         retryingImpl(k, std::move(pm), std::move(ffm), std::move(prom));
1341       } else {
1342         prom.setException(std::move(xm));
1343       };
1344     });
1345   });
1346 }
1347
1348 template <class Policy, class FF>
1349 typename std::result_of<FF(size_t)>::type
1350 retrying(size_t k, Policy&& p, FF&& ff) {
1351   using F = typename std::result_of<FF(size_t)>::type;
1352   using T = typename F::value_type;
1353   auto prom = Promise<T>();
1354   auto f = prom.getFuture();
1355   retryingImpl(
1356       k, std::forward<Policy>(p), std::forward<FF>(ff), std::move(prom));
1357   return f;
1358 }
1359
1360 template <class Policy, class FF>
1361 typename std::result_of<FF(size_t)>::type
1362 retrying(Policy&& p, FF&& ff, retrying_policy_raw_tag) {
1363   auto q = [pm = std::forward<Policy>(p)](size_t k, exception_wrapper x) {
1364     return makeFuture<bool>(pm(k, x));
1365   };
1366   return retrying(0, std::move(q), std::forward<FF>(ff));
1367 }
1368
1369 template <class Policy, class FF>
1370 typename std::result_of<FF(size_t)>::type
1371 retrying(Policy&& p, FF&& ff, retrying_policy_fut_tag) {
1372   return retrying(0, std::forward<Policy>(p), std::forward<FF>(ff));
1373 }
1374
1375 //  jittered exponential backoff, clamped to [backoff_min, backoff_max]
1376 template <class URNG>
1377 Duration retryingJitteredExponentialBackoffDur(
1378     size_t n,
1379     Duration backoff_min,
1380     Duration backoff_max,
1381     double jitter_param,
1382     URNG& rng) {
1383   using d = Duration;
1384   auto dist = std::normal_distribution<double>(0.0, jitter_param);
1385   auto jitter = std::exp(dist(rng));
1386   auto backoff = d(d::rep(jitter * backoff_min.count() * std::pow(2, n - 1)));
1387   return std::max(backoff_min, std::min(backoff_max, backoff));
1388 }
1389
1390 template <class Policy, class URNG>
1391 std::function<Future<bool>(size_t, const exception_wrapper&)>
1392 retryingPolicyCappedJitteredExponentialBackoff(
1393     size_t max_tries,
1394     Duration backoff_min,
1395     Duration backoff_max,
1396     double jitter_param,
1397     URNG&& rng,
1398     Policy&& p) {
1399   return [
1400     pm = std::forward<Policy>(p),
1401     max_tries,
1402     backoff_min,
1403     backoff_max,
1404     jitter_param,
1405     rngp = std::forward<URNG>(rng)
1406   ](size_t n, const exception_wrapper& ex) mutable {
1407     if (n == max_tries) {
1408       return makeFuture(false);
1409     }
1410     return pm(n, ex).then(
1411         [ n, backoff_min, backoff_max, jitter_param, rngp = std::move(rngp) ](
1412             bool v) mutable {
1413           if (!v) {
1414             return makeFuture(false);
1415           }
1416           auto backoff = detail::retryingJitteredExponentialBackoffDur(
1417               n, backoff_min, backoff_max, jitter_param, rngp);
1418           return futures::sleep(backoff).then([] { return true; });
1419         });
1420   };
1421 }
1422
1423 template <class Policy, class URNG>
1424 std::function<Future<bool>(size_t, const exception_wrapper&)>
1425 retryingPolicyCappedJitteredExponentialBackoff(
1426     size_t max_tries,
1427     Duration backoff_min,
1428     Duration backoff_max,
1429     double jitter_param,
1430     URNG&& rng,
1431     Policy&& p,
1432     retrying_policy_raw_tag) {
1433   auto q = [pm = std::forward<Policy>(p)](
1434       size_t n, const exception_wrapper& e) {
1435     return makeFuture(pm(n, e));
1436   };
1437   return retryingPolicyCappedJitteredExponentialBackoff(
1438       max_tries,
1439       backoff_min,
1440       backoff_max,
1441       jitter_param,
1442       std::forward<URNG>(rng),
1443       std::move(q));
1444 }
1445
1446 template <class Policy, class URNG>
1447 std::function<Future<bool>(size_t, const exception_wrapper&)>
1448 retryingPolicyCappedJitteredExponentialBackoff(
1449     size_t max_tries,
1450     Duration backoff_min,
1451     Duration backoff_max,
1452     double jitter_param,
1453     URNG&& rng,
1454     Policy&& p,
1455     retrying_policy_fut_tag) {
1456   return retryingPolicyCappedJitteredExponentialBackoff(
1457       max_tries,
1458       backoff_min,
1459       backoff_max,
1460       jitter_param,
1461       std::forward<URNG>(rng),
1462       std::forward<Policy>(p));
1463 }
1464 }
1465
1466 template <class Policy, class FF>
1467 typename std::result_of<FF(size_t)>::type
1468 retrying(Policy&& p, FF&& ff) {
1469   using tag = typename detail::retrying_policy_traits<Policy>::tag;
1470   return detail::retrying(std::forward<Policy>(p), std::forward<FF>(ff), tag());
1471 }
1472
1473 inline
1474 std::function<bool(size_t, const exception_wrapper&)>
1475 retryingPolicyBasic(
1476     size_t max_tries) {
1477   return [=](size_t n, const exception_wrapper&) { return n < max_tries; };
1478 }
1479
1480 template <class Policy, class URNG>
1481 std::function<Future<bool>(size_t, const exception_wrapper&)>
1482 retryingPolicyCappedJitteredExponentialBackoff(
1483     size_t max_tries,
1484     Duration backoff_min,
1485     Duration backoff_max,
1486     double jitter_param,
1487     URNG&& rng,
1488     Policy&& p) {
1489   using tag = typename detail::retrying_policy_traits<Policy>::tag;
1490   return detail::retryingPolicyCappedJitteredExponentialBackoff(
1491       max_tries,
1492       backoff_min,
1493       backoff_max,
1494       jitter_param,
1495       std::forward<URNG>(rng),
1496       std::forward<Policy>(p),
1497       tag());
1498 }
1499
1500 inline
1501 std::function<Future<bool>(size_t, const exception_wrapper&)>
1502 retryingPolicyCappedJitteredExponentialBackoff(
1503     size_t max_tries,
1504     Duration backoff_min,
1505     Duration backoff_max,
1506     double jitter_param) {
1507   auto p = [](size_t, const exception_wrapper&) { return true; };
1508   return retryingPolicyCappedJitteredExponentialBackoff(
1509       max_tries,
1510       backoff_min,
1511       backoff_max,
1512       jitter_param,
1513       ThreadLocalPRNG(),
1514       std::move(p));
1515 }
1516
1517 }
1518
1519 // Instantiate the most common Future types to save compile time
1520 extern template class Future<Unit>;
1521 extern template class Future<bool>;
1522 extern template class Future<int>;
1523 extern template class Future<int64_t>;
1524 extern template class Future<std::string>;
1525 extern template class Future<double>;
1526
1527 } // namespace folly