folly/futures/Future-inl.h

   1 /*
   2  * Copyright 2015 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 <chrono>
  20 #include <thread>
  21
  22 #include <folly/Baton.h>
  23 #include <folly/Optional.h>
  24 #include <folly/futures/detail/Core.h>
  25 #include <folly/futures/Timekeeper.h>
  26
  27 namespace folly {
  28
  29 class Timekeeper;
  30
  31 namespace detail {
  32   Timekeeper* getTimekeeperSingleton();
  33 }
  34
  35 template <class T>
  36 Future<T>::Future(Future<T>&& other) noexcept : core_(other.core_) {
  37   other.core_ = nullptr;
  38 }
  39
  40 template <class T>
  41 Future<T>& Future<T>::operator=(Future<T>&& other) noexcept {
  42   std::swap(core_, other.core_);
  43   return *this;
  44 }
  45
  46 template <class T>
  47 template <class F>
  48 Future<T>::Future(
  49   const typename std::enable_if<!std::is_void<F>::value, F>::type& val)
  50     : core_(nullptr) {
  51   Promise<F> p;
  52   p.setValue(val);
  53   *this = p.getFuture();
  54 }
  55
  56 template <class T>
  57 template <class F>
  58 Future<T>::Future(
  59   typename std::enable_if<!std::is_void<F>::value, F>::type&& val)
  60     : core_(nullptr) {
  61   Promise<F> p;
  62   p.setValue(std::forward<F>(val));
  63   *this = p.getFuture();
  64 }
  65
  66 template <>
  67 template <class F,
  68           typename std::enable_if<std::is_void<F>::value, int>::type>
  69 Future<void>::Future() : core_(nullptr) {
  70   Promise<void> p;
  71   p.setValue();
  72   *this = p.getFuture();
  73 }
  74
  75
  76 template <class T>
  77 Future<T>::~Future() {
  78   detach();
  79 }
  80
  81 template <class T>
  82 void Future<T>::detach() {
  83   if (core_) {
  84     core_->detachFuture();
  85     core_ = nullptr;
  86   }
  87 }
  88
  89 template <class T>
  90 void Future<T>::throwIfInvalid() const {
  91   if (!core_)
  92     throw NoState();
  93 }
  94
  95 template <class T>
  96 template <class F>
  97 void Future<T>::setCallback_(F&& func) {
  98   throwIfInvalid();
  99   core_->setCallback(std::move(func));
 100 }
 101
 102 // unwrap
 103
 104 template <class T>
 105 template <class F>
 106 typename std::enable_if<isFuture<F>::value,
 107                         Future<typename isFuture<T>::Inner>>::type
 108 Future<T>::unwrap() {
 109   return then([](Future<typename isFuture<T>::Inner> internal_future) {
 110       return internal_future;
 111   });
 112 }
 113
 114 // then
 115
 116 // Variant: returns a value
 117 // e.g. f.then([](Try<T>&& t){ return t.value(); });
 118 template <class T>
 119 template <typename F, typename R, bool isTry, typename... Args>
 120 typename std::enable_if<!R::ReturnsFuture::value, typename R::Return>::type
 121 Future<T>::thenImplementation(F func, detail::argResult<isTry, F, Args...>) {
 122   static_assert(sizeof...(Args) <= 1, "Then must take zero/one argument");
 123   typedef typename R::ReturnsFuture::Inner B;
 124
 125   throwIfInvalid();
 126
 127   // wrap these so we can move them into the lambda
 128   folly::MoveWrapper<Promise<B>> p;
 129   folly::MoveWrapper<F> funcm(std::forward<F>(func));
 130
 131   // grab the Future now before we lose our handle on the Promise
 132   auto f = p->getFuture();
 133   if (getExecutor()) {
 134     f.setExecutor(getExecutor());
 135   }
 136
 137   /* This is a bit tricky.
 138
 139      We can't just close over *this in case this Future gets moved. So we
 140      make a new dummy Future. We could figure out something more
 141      sophisticated that avoids making a new Future object when it can, as an
 142      optimization. But this is correct.
 143
 144      core_ can't be moved, it is explicitly disallowed (as is copying). But
 145      if there's ever a reason to allow it, this is one place that makes that
 146      assumption and would need to be fixed. We use a standard shared pointer
 147      for core_ (by copying it in), which means in essence obj holds a shared
 148      pointer to itself.  But this shouldn't leak because Promise will not
 149      outlive the continuation, because Promise will setException() with a
 150      broken Promise if it is destructed before completed. We could use a
 151      weak pointer but it would have to be converted to a shared pointer when
 152      func is executed (because the Future returned by func may possibly
 153      persist beyond the callback, if it gets moved), and so it is an
 154      optimization to just make it shared from the get-go.
 155
 156      We have to move in the Promise and func using the MoveWrapper
 157      hack. (func could be copied but it's a big drag on perf).
 158
 159      Two subtle but important points about this design. detail::Core has no
 160      back pointers to Future or Promise, so if Future or Promise get moved
 161      (and they will be moved in performant code) we don't have to do
 162      anything fancy. And because we store the continuation in the
 163      detail::Core, not in the Future, we can execute the continuation even
 164      after the Future has gone out of scope. This is an intentional design
 165      decision. It is likely we will want to be able to cancel a continuation
 166      in some circumstances, but I think it should be explicit not implicit
 167      in the destruction of the Future used to create it.
 168      */
 169   setCallback_(
 170     [p, funcm](Try<T>&& t) mutable {
 171       if (!isTry && t.hasException()) {
 172         p->setException(std::move(t.exception()));
 173       } else {
 174         p->setWith([&]() {
 175           return (*funcm)(t.template get<isTry, Args>()...);
 176         });
 177       }
 178     });
 179
 180   return f;
 181 }
 182
 183 // Variant: returns a Future
 184 // e.g. f.then([](T&& t){ return makeFuture<T>(t); });
 185 template <class T>
 186 template <typename F, typename R, bool isTry, typename... Args>
 187 typename std::enable_if<R::ReturnsFuture::value, typename R::Return>::type
 188 Future<T>::thenImplementation(F func, detail::argResult<isTry, F, Args...>) {
 189   static_assert(sizeof...(Args) <= 1, "Then must take zero/one argument");
 190   typedef typename R::ReturnsFuture::Inner B;
 191
 192   throwIfInvalid();
 193
 194   // wrap these so we can move them into the lambda
 195   folly::MoveWrapper<Promise<B>> p;
 196   folly::MoveWrapper<F> funcm(std::forward<F>(func));
 197
 198   // grab the Future now before we lose our handle on the Promise
 199   auto f = p->getFuture();
 200   if (getExecutor()) {
 201     f.setExecutor(getExecutor());
 202   }
 203
 204   setCallback_(
 205     [p, funcm](Try<T>&& t) mutable {
 206       if (!isTry && t.hasException()) {
 207         p->setException(std::move(t.exception()));
 208       } else {
 209         try {
 210           auto f2 = (*funcm)(t.template get<isTry, Args>()...);
 211           // that didn't throw, now we can steal p
 212           f2.setCallback_([p](Try<B>&& b) mutable {
 213             p->setTry(std::move(b));
 214           });
 215         } catch (const std::exception& e) {
 216           p->setException(exception_wrapper(std::current_exception(), e));
 217         } catch (...) {
 218           p->setException(exception_wrapper(std::current_exception()));
 219         }
 220       }
 221     });
 222
 223   return f;
 224 }
 225
 226 template <typename T>
 227 template <typename R, typename Caller, typename... Args>
 228   Future<typename isFuture<R>::Inner>
 229 Future<T>::then(R(Caller::*func)(Args...), Caller *instance) {
 230   typedef typename std::remove_cv<
 231     typename std::remove_reference<
 232       typename detail::ArgType<Args...>::FirstArg>::type>::type FirstArg;
 233   return then([instance, func](Try<T>&& t){
 234     return (instance->*func)(t.template get<isTry<FirstArg>::value, Args>()...);
 235   });
 236 }
 237
 238 template <class T>
 239 template <class... Args>
 240 auto Future<T>::then(Executor* x, Args&&... args)
 241   -> decltype(this->then(std::forward<Args>(args)...))
 242 {
 243   auto oldX = getExecutor();
 244   setExecutor(x);
 245   return this->then(std::forward<Args>(args)...).via(oldX);
 246 }
 247
 248 template <class T>
 249 Future<void> Future<T>::then() {
 250   return then([] (Try<T>&& t) {});
 251 }
 252
 253 // onError where the callback returns T
 254 template <class T>
 255 template <class F>
 256 typename std::enable_if<
 257   !detail::callableWith<F, exception_wrapper>::value &&
 258   !detail::Extract<F>::ReturnsFuture::value,
 259   Future<T>>::type
 260 Future<T>::onError(F&& func) {
 261   typedef typename detail::Extract<F>::FirstArg Exn;
 262   static_assert(
 263       std::is_same<typename detail::Extract<F>::RawReturn, T>::value,
 264       "Return type of onError callback must be T or Future<T>");
 265
 266   Promise<T> p;
 267   auto f = p.getFuture();
 268   auto pm = folly::makeMoveWrapper(std::move(p));
 269   auto funcm = folly::makeMoveWrapper(std::move(func));
 270   setCallback_([pm, funcm](Try<T>&& t) mutable {
 271     if (!t.template withException<Exn>([&] (Exn& e) {
 272           pm->setWith([&]{
 273             return (*funcm)(e);
 274           });
 275         })) {
 276       pm->setTry(std::move(t));
 277     }
 278   });
 279
 280   return f;
 281 }
 282
 283 // onError where the callback returns Future<T>
 284 template <class T>
 285 template <class F>
 286 typename std::enable_if<
 287   !detail::callableWith<F, exception_wrapper>::value &&
 288   detail::Extract<F>::ReturnsFuture::value,
 289   Future<T>>::type
 290 Future<T>::onError(F&& func) {
 291   static_assert(
 292       std::is_same<typename detail::Extract<F>::Return, Future<T>>::value,
 293       "Return type of onError callback must be T or Future<T>");
 294   typedef typename detail::Extract<F>::FirstArg Exn;
 295
 296   Promise<T> p;
 297   auto f = p.getFuture();
 298   auto pm = folly::makeMoveWrapper(std::move(p));
 299   auto funcm = folly::makeMoveWrapper(std::move(func));
 300   setCallback_([pm, funcm](Try<T>&& t) mutable {
 301     if (!t.template withException<Exn>([&] (Exn& e) {
 302           try {
 303             auto f2 = (*funcm)(e);
 304             f2.setCallback_([pm](Try<T>&& t2) mutable {
 305               pm->setTry(std::move(t2));
 306             });
 307           } catch (const std::exception& e2) {
 308             pm->setException(exception_wrapper(std::current_exception(), e2));
 309           } catch (...) {
 310             pm->setException(exception_wrapper(std::current_exception()));
 311           }
 312         })) {
 313       pm->setTry(std::move(t));
 314     }
 315   });
 316
 317   return f;
 318 }
 319
 320 template <class T>
 321 template <class F>
 322 Future<T> Future<T>::ensure(F func) {
 323   MoveWrapper<F> funcw(std::move(func));
 324   return this->then([funcw](Try<T>&& t) {
 325     (*funcw)();
 326     return makeFuture(std::move(t));
 327   });
 328 }
 329
 330 template <class T>
 331 template <class F>
 332 Future<T> Future<T>::onTimeout(Duration dur, F&& func, Timekeeper* tk) {
 333   auto funcw = folly::makeMoveWrapper(std::forward<F>(func));
 334   return within(dur, tk)
 335     .onError([funcw](TimedOut const&) { return (*funcw)(); });
 336 }
 337
 338 template <class T>
 339 template <class F>
 340 typename std::enable_if<
 341   detail::callableWith<F, exception_wrapper>::value &&
 342   detail::Extract<F>::ReturnsFuture::value,
 343   Future<T>>::type
 344 Future<T>::onError(F&& func) {
 345   static_assert(
 346       std::is_same<typename detail::Extract<F>::Return, Future<T>>::value,
 347       "Return type of onError callback must be T or Future<T>");
 348
 349   Promise<T> p;
 350   auto f = p.getFuture();
 351   auto pm = folly::makeMoveWrapper(std::move(p));
 352   auto funcm = folly::makeMoveWrapper(std::move(func));
 353   setCallback_([pm, funcm](Try<T> t) mutable {
 354     if (t.hasException()) {
 355       try {
 356         auto f2 = (*funcm)(std::move(t.exception()));
 357         f2.setCallback_([pm](Try<T> t2) mutable {
 358           pm->setTry(std::move(t2));
 359         });
 360       } catch (const std::exception& e2) {
 361         pm->setException(exception_wrapper(std::current_exception(), e2));
 362       } catch (...) {
 363         pm->setException(exception_wrapper(std::current_exception()));
 364       }
 365     } else {
 366       pm->setTry(std::move(t));
 367     }
 368   });
 369
 370   return f;
 371 }
 372
 373 // onError(exception_wrapper) that returns T
 374 template <class T>
 375 template <class F>
 376 typename std::enable_if<
 377   detail::callableWith<F, exception_wrapper>::value &&
 378   !detail::Extract<F>::ReturnsFuture::value,
 379   Future<T>>::type
 380 Future<T>::onError(F&& func) {
 381   static_assert(
 382       std::is_same<typename detail::Extract<F>::Return, Future<T>>::value,
 383       "Return type of onError callback must be T or Future<T>");
 384
 385   Promise<T> p;
 386   auto f = p.getFuture();
 387   auto pm = folly::makeMoveWrapper(std::move(p));
 388   auto funcm = folly::makeMoveWrapper(std::move(func));
 389   setCallback_([pm, funcm](Try<T> t) mutable {
 390     if (t.hasException()) {
 391       pm->setWith([&]{
 392         return (*funcm)(std::move(t.exception()));
 393       });
 394     } else {
 395       pm->setTry(std::move(t));
 396     }
 397   });
 398
 399   return f;
 400 }
 401
 402 template <class T>
 403 typename std::add_lvalue_reference<T>::type Future<T>::value() {
 404   throwIfInvalid();
 405
 406   return core_->getTry().value();
 407 }
 408
 409 template <class T>
 410 typename std::add_lvalue_reference<const T>::type Future<T>::value() const {
 411   throwIfInvalid();
 412
 413   return core_->getTry().value();
 414 }
 415
 416 template <class T>
 417 Try<T>& Future<T>::getTry() {
 418   throwIfInvalid();
 419
 420   return core_->getTry();
 421 }
 422
 423 template <class T>
 424 Optional<Try<T>> Future<T>::poll() {
 425   Optional<Try<T>> o;
 426   if (core_->ready()) {
 427     o = std::move(core_->getTry());
 428   }
 429   return o;
 430 }
 431
 432 template <class T>
 433 template <typename Executor>
 434 inline Future<T> Future<T>::via(Executor* executor) && {
 435   throwIfInvalid();
 436
 437   setExecutor(executor);
 438
 439   return std::move(*this);
 440 }
 441
 442 template <class T>
 443 template <typename Executor>
 444 inline Future<T> Future<T>::via(Executor* executor) & {
 445   throwIfInvalid();
 446
 447   MoveWrapper<Promise<T>> p;
 448   auto f = p->getFuture();
 449   then([p](Try<T>&& t) mutable { p->setTry(std::move(t)); });
 450   return std::move(f).via(executor);
 451 }
 452
 453 template <class T>
 454 bool Future<T>::isReady() const {
 455   throwIfInvalid();
 456   return core_->ready();
 457 }
 458
 459 template <class T>
 460 void Future<T>::raise(exception_wrapper exception) {
 461   core_->raise(std::move(exception));
 462 }
 463
 464 // makeFuture
 465
 466 template <class T>
 467 Future<typename std::decay<T>::type> makeFuture(T&& t) {
 468   Promise<typename std::decay<T>::type> p;
 469   p.setValue(std::forward<T>(t));
 470   return p.getFuture();
 471 }
 472
 473 inline // for multiple translation units
 474 Future<void> makeFuture() {
 475   Promise<void> p;
 476   p.setValue();
 477   return p.getFuture();
 478 }
 479
 480 template <class F>
 481 auto makeFutureTry(
 482     F&& func,
 483     typename std::enable_if<!std::is_reference<F>::value, bool>::type sdf)
 484     -> Future<decltype(func())> {
 485   Promise<decltype(func())> p;
 486   p.setWith(
 487     [&func]() {
 488       return (func)();
 489     });
 490   return p.getFuture();
 491 }
 492
 493 template <class F>
 494 auto makeFutureTry(F const& func) -> Future<decltype(func())> {
 495   F copy = func;
 496   return makeFutureTry(std::move(copy));
 497 }
 498
 499 template <class T>
 500 Future<T> makeFuture(std::exception_ptr const& e) {
 501   Promise<T> p;
 502   p.setException(e);
 503   return p.getFuture();
 504 }
 505
 506 template <class T>
 507 Future<T> makeFuture(exception_wrapper ew) {
 508   Promise<T> p;
 509   p.setException(std::move(ew));
 510   return p.getFuture();
 511 }
 512
 513 template <class T, class E>
 514 typename std::enable_if<std::is_base_of<std::exception, E>::value,
 515                         Future<T>>::type
 516 makeFuture(E const& e) {
 517   Promise<T> p;
 518   p.setException(make_exception_wrapper<E>(e));
 519   return p.getFuture();
 520 }
 521
 522 template <class T>
 523 Future<T> makeFuture(Try<T>&& t) {
 524   Promise<typename std::decay<T>::type> p;
 525   p.setTry(std::move(t));
 526   return p.getFuture();
 527 }
 528
 529 template <>
 530 inline Future<void> makeFuture(Try<void>&& t) {
 531   if (t.hasException()) {
 532     return makeFuture<void>(std::move(t.exception()));
 533   } else {
 534     return makeFuture();
 535   }
 536 }
 537
 538 // via
 539 template <typename Executor>
 540 Future<void> via(Executor* executor) {
 541   return makeFuture().via(executor);
 542 }
 543
 544 // when (variadic)
 545
 546 template <typename... Fs>
 547 typename detail::VariadicContext<
 548   typename std::decay<Fs>::type::value_type...>::type
 549 whenAll(Fs&&... fs) {
 550   auto ctx =
 551     new detail::VariadicContext<typename std::decay<Fs>::type::value_type...>();
 552   ctx->total = sizeof...(fs);
 553   auto f_saved = ctx->p.getFuture();
 554   detail::whenAllVariadicHelper(ctx,
 555     std::forward<typename std::decay<Fs>::type>(fs)...);
 556   return f_saved;
 557 }
 558
 559 // when (iterator)
 560
 561 template <class InputIterator>
 562 Future<
 563   std::vector<
 564   Try<typename std::iterator_traits<InputIterator>::value_type::value_type>>>
 565 whenAll(InputIterator first, InputIterator last) {
 566   typedef
 567     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 568
 569   if (first >= last) {
 570     return makeFuture(std::vector<Try<T>>());
 571   }
 572   size_t n = std::distance(first, last);
 573
 574   auto ctx = new detail::WhenAllContext<T>();
 575
 576   ctx->results.resize(n);
 577
 578   auto f_saved = ctx->p.getFuture();
 579
 580   for (size_t i = 0; first != last; ++first, ++i) {
 581      assert(i < n);
 582      auto& f = *first;
 583      f.setCallback_([ctx, i, n](Try<T> t) {
 584        ctx->results[i] = std::move(t);
 585        if (++ctx->count == n) {
 586          ctx->p.setValue(std::move(ctx->results));
 587          delete ctx;
 588        }
 589      });
 590   }
 591
 592   return f_saved;
 593 }
 594
 595 namespace detail {
 596
 597 template <typename T>
 598 struct CollectContext {
 599   explicit CollectContext(int n) : count(0), threw(false) {}
 600
 601   Promise<std::vector<T>> p;
 602   std::vector<T> results;
 603   std::atomic<size_t> count;
 604   std::atomic_bool threw;
 605
 606   typedef std::vector<T> result_type;
 607
 608   static inline Future<std::vector<T>> makeEmptyFuture() {
 609     return makeFuture(std::vector<T>());
 610   }
 611
 612   inline void setValue() {
 613     p.setValue(std::move(results));
 614   }
 615
 616   inline void addResult(int i, Try<T>& t) {
 617     results[i] = std::move(t.value());
 618   }
 619 };
 620
 621 template <>
 622 struct CollectContext<void> {
 623   explicit CollectContext(int n) : count(0), threw(false) {}
 624   Promise<void> p;
 625   std::atomic<size_t> count;
 626   std::atomic_bool threw;
 627
 628   typedef void result_type;
 629
 630   static inline Future<void> makeEmptyFuture() {
 631     return makeFuture();
 632   }
 633
 634   inline void setValue() {
 635     p.setValue();
 636   }
 637
 638   inline void addResult(int i, Try<void>& t) {
 639     // do nothing
 640   }
 641 };
 642
 643 } // detail
 644
 645 template <class InputIterator>
 646 Future<typename detail::CollectContext<
 647   typename std::iterator_traits<InputIterator>::value_type::value_type
 648 >::result_type>
 649 collect(InputIterator first, InputIterator last) {
 650   typedef
 651     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 652
 653   if (first >= last) {
 654     return detail::CollectContext<T>::makeEmptyFuture();
 655   }
 656
 657   size_t n = std::distance(first, last);
 658   auto ctx = new detail::CollectContext<T>(n);
 659   auto f_saved = ctx->p.getFuture();
 660
 661   for (size_t i = 0; first != last; ++first, ++i) {
 662      assert(i < n);
 663      auto& f = *first;
 664      f.setCallback_([ctx, i, n](Try<T> t) {
 665        auto c = ++ctx->count;
 666
 667        if (t.hasException()) {
 668          if (!ctx->threw.exchange(true)) {
 669            ctx->p.setException(std::move(t.exception()));
 670          }
 671        } else if (!ctx->threw) {
 672          ctx->addResult(i, t);
 673          if (c == n) {
 674            ctx->setValue();
 675          }
 676        }
 677
 678        if (c == n) {
 679          delete ctx;
 680        }
 681      });
 682   }
 683
 684   return f_saved;
 685 }
 686
 687 template <class InputIterator>
 688 Future<
 689   std::pair<size_t,
 690             Try<
 691               typename
 692               std::iterator_traits<InputIterator>::value_type::value_type> > >
 693 whenAny(InputIterator first, InputIterator last) {
 694   typedef
 695     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 696
 697   auto ctx = new detail::WhenAnyContext<T>(std::distance(first, last));
 698   auto f_saved = ctx->p.getFuture();
 699
 700   for (size_t i = 0; first != last; first++, i++) {
 701     auto& f = *first;
 702     f.setCallback_([i, ctx](Try<T>&& t) {
 703       if (!ctx->done.exchange(true)) {
 704         ctx->p.setValue(std::make_pair(i, std::move(t)));
 705       }
 706       ctx->decref();
 707     });
 708   }
 709
 710   return f_saved;
 711 }
 712
 713 template <class InputIterator>
 714 Future<std::vector<std::pair<size_t, Try<typename
 715   std::iterator_traits<InputIterator>::value_type::value_type>>>>
 716 whenN(InputIterator first, InputIterator last, size_t n) {
 717   typedef typename
 718     std::iterator_traits<InputIterator>::value_type::value_type T;
 719   typedef std::vector<std::pair<size_t, Try<T>>> V;
 720
 721   struct ctx_t {
 722     V v;
 723     size_t completed;
 724     Promise<V> p;
 725   };
 726   auto ctx = std::make_shared<ctx_t>();
 727   ctx->completed = 0;
 728
 729   // for each completed Future, increase count and add to vector, until we
 730   // have n completed futures at which point we fulfill our Promise with the
 731   // vector
 732   auto it = first;
 733   size_t i = 0;
 734   while (it != last) {
 735     it->then([ctx, n, i](Try<T>&& t) {
 736       auto& v = ctx->v;
 737       auto c = ++ctx->completed;
 738       if (c <= n) {
 739         assert(ctx->v.size() < n);
 740         v.push_back(std::make_pair(i, std::move(t)));
 741         if (c == n) {
 742           ctx->p.setTry(Try<V>(std::move(v)));
 743         }
 744       }
 745     });
 746
 747     it++;
 748     i++;
 749   }
 750
 751   if (i < n) {
 752     ctx->p.setException(std::runtime_error("Not enough futures"));
 753   }
 754
 755   return ctx->p.getFuture();
 756 }
 757
 758 template <class It, class T, class F, class ItT, class Arg>
 759 typename std::enable_if<!isFutureResult<F, T, Arg>::value, Future<T>>::type
 760 reduce(It first, It last, T initial, F func) {
 761   if (first == last) {
 762     return makeFuture(std::move(initial));
 763   }
 764
 765   typedef isTry<Arg> IsTry;
 766
 767   return whenAll(first, last)
 768     .then([initial, func](std::vector<Try<ItT>>& vals) mutable {
 769       for (auto& val : vals) {
 770         initial = func(std::move(initial),
 771                        // Either return a ItT&& or a Try<ItT>&& depending
 772                        // on the type of the argument of func.
 773                        val.template get<IsTry::value, Arg&&>());
 774       }
 775       return initial;
 776     });
 777 }
 778
 779 template <class It, class T, class F, class ItT, class Arg>
 780 typename std::enable_if<isFutureResult<F, T, Arg>::value, Future<T>>::type
 781 reduce(It first, It last, T initial, F func) {
 782   if (first == last) {
 783     return makeFuture(std::move(initial));
 784   }
 785
 786   typedef isTry<Arg> IsTry;
 787
 788   auto f = first->then([initial, func](Try<ItT>& head) mutable {
 789     return func(std::move(initial),
 790                 head.template get<IsTry::value, Arg&&>());
 791   });
 792
 793   for (++first; first != last; ++first) {
 794     f = whenAll(f, *first).then([func](std::tuple<Try<T>, Try<ItT>>& t) {
 795       return func(std::move(std::get<0>(t).value()),
 796                   // Either return a ItT&& or a Try<ItT>&& depending
 797                   // on the type of the argument of func.
 798                   std::get<1>(t).template get<IsTry::value, Arg&&>());
 799     });
 800   }
 801
 802   return f;
 803 }
 804
 805 template <class T>
 806 Future<T> Future<T>::within(Duration dur, Timekeeper* tk) {
 807   return within(dur, TimedOut(), tk);
 808 }
 809
 810 template <class T>
 811 template <class E>
 812 Future<T> Future<T>::within(Duration dur, E e, Timekeeper* tk) {
 813
 814   struct Context {
 815     Context(E ex) : exception(std::move(ex)), promise(), token(false) {}
 816     E exception;
 817     Promise<T> promise;
 818     std::atomic<bool> token;
 819   };
 820   auto ctx = std::make_shared<Context>(std::move(e));
 821
 822   if (!tk) {
 823     tk = folly::detail::getTimekeeperSingleton();
 824   }
 825
 826   tk->after(dur)
 827     .then([ctx](Try<void> const& t) {
 828       if (ctx->token.exchange(true) == false) {
 829         if (t.hasException()) {
 830           ctx->promise.setException(std::move(t.exception()));
 831         } else {
 832           ctx->promise.setException(std::move(ctx->exception));
 833         }
 834       }
 835     });
 836
 837   this->then([ctx](Try<T>&& t) {
 838     if (ctx->token.exchange(true) == false) {
 839       ctx->promise.setTry(std::move(t));
 840     }
 841   });
 842
 843   return ctx->promise.getFuture();
 844 }
 845
 846 template <class T>
 847 Future<T> Future<T>::delayed(Duration dur, Timekeeper* tk) {
 848   return whenAll(*this, futures::sleep(dur, tk))
 849     .then([](std::tuple<Try<T>, Try<void>> tup) {
 850       Try<T>& t = std::get<0>(tup);
 851       return makeFuture<T>(std::move(t));
 852     });
 853 }
 854
 855 namespace detail {
 856
 857 template <class T>
 858 void waitImpl(Future<T>& f) {
 859   // short-circuit if there's nothing to do
 860   if (f.isReady()) return;
 861
 862   Baton<> baton;
 863   f = f.then([&](Try<T> t) {
 864     baton.post();
 865     return makeFuture(std::move(t));
 866   });
 867   baton.wait();
 868
 869   // There's a race here between the return here and the actual finishing of
 870   // the future. f is completed, but the setup may not have finished on done
 871   // after the baton has posted.
 872   while (!f.isReady()) {
 873     std::this_thread::yield();
 874   }
 875 }
 876
 877 template <class T>
 878 void waitImpl(Future<T>& f, Duration dur) {
 879   // short-circuit if there's nothing to do
 880   if (f.isReady()) return;
 881
 882   auto baton = std::make_shared<Baton<>>();
 883   f = f.then([baton](Try<T> t) {
 884     baton->post();
 885     return makeFuture(std::move(t));
 886   });
 887
 888   // Let's preserve the invariant that if we did not timeout (timed_wait returns
 889   // true), then the returned Future is complete when it is returned to the
 890   // caller. We need to wait out the race for that Future to complete.
 891   if (baton->timed_wait(std::chrono::system_clock::now() + dur)) {
 892     while (!f.isReady()) {
 893       std::this_thread::yield();
 894     }
 895   }
 896 }
 897
 898 template <class T>
 899 void waitViaImpl(Future<T>& f, DrivableExecutor* e) {
 900   while (!f.isReady()) {
 901     e->drive();
 902   }
 903 }
 904
 905 } // detail
 906
 907 template <class T>
 908 Future<T>& Future<T>::wait() & {
 909   detail::waitImpl(*this);
 910   return *this;
 911 }
 912
 913 template <class T>
 914 Future<T>&& Future<T>::wait() && {
 915   detail::waitImpl(*this);
 916   return std::move(*this);
 917 }
 918
 919 template <class T>
 920 Future<T>& Future<T>::wait(Duration dur) & {
 921   detail::waitImpl(*this, dur);
 922   return *this;
 923 }
 924
 925 template <class T>
 926 Future<T>&& Future<T>::wait(Duration dur) && {
 927   detail::waitImpl(*this, dur);
 928   return std::move(*this);
 929 }
 930
 931 template <class T>
 932 Future<T>& Future<T>::waitVia(DrivableExecutor* e) & {
 933   detail::waitViaImpl(*this, e);
 934   return *this;
 935 }
 936
 937 template <class T>
 938 Future<T>&& Future<T>::waitVia(DrivableExecutor* e) && {
 939   detail::waitViaImpl(*this, e);
 940   return std::move(*this);
 941 }
 942
 943 template <class T>
 944 T Future<T>::get() {
 945   return std::move(wait().value());
 946 }
 947
 948 template <>
 949 inline void Future<void>::get() {
 950   wait().value();
 951 }
 952
 953 template <class T>
 954 T Future<T>::get(Duration dur) {
 955   wait(dur);
 956   if (isReady()) {
 957     return std::move(value());
 958   } else {
 959     throw TimedOut();
 960   }
 961 }
 962
 963 template <>
 964 inline void Future<void>::get(Duration dur) {
 965   wait(dur);
 966   if (isReady()) {
 967     return;
 968   } else {
 969     throw TimedOut();
 970   }
 971 }
 972
 973 template <class T>
 974 T Future<T>::getVia(DrivableExecutor* e) {
 975   return std::move(waitVia(e).value());
 976 }
 977
 978 template <>
 979 inline void Future<void>::getVia(DrivableExecutor* e) {
 980   waitVia(e).value();
 981 }
 982
 983 template <class T>
 984 Future<bool> Future<T>::willEqual(Future<T>& f) {
 985   return whenAll(*this, f).then([](const std::tuple<Try<T>, Try<T>>& t) {
 986     if (std::get<0>(t).hasValue() && std::get<1>(t).hasValue()) {
 987       return std::get<0>(t).value() == std::get<1>(t).value();
 988     } else {
 989       return false;
 990       }
 991   });
 992 }
 993
 994 template <class T>
 995 template <class F>
 996 Future<T> Future<T>::filter(F predicate) {
 997   auto p = folly::makeMoveWrapper(std::move(predicate));
 998   return this->then([p](T val) {
 999     T const& valConstRef = val;
1000     if (!(*p)(valConstRef)) {
1001       throw PredicateDoesNotObtain();
1002     }
1003     return val;
1004   });
1005 }
1006
1007 namespace futures {
1008   namespace {
1009     template <class Z>
1010     Future<Z> chainHelper(Future<Z> f) {
1011       return f;
1012     }
1013
1014     template <class Z, class F, class Fn, class... Callbacks>
1015     Future<Z> chainHelper(F f, Fn fn, Callbacks... fns) {
1016       return chainHelper<Z>(f.then(fn), fns...);
1017     }
1018   }
1019
1020   template <class A, class Z, class... Callbacks>
1021   std::function<Future<Z>(Try<A>)>
1022   chain(Callbacks... fns) {
1023     MoveWrapper<Promise<A>> pw;
1024     MoveWrapper<Future<Z>> fw(chainHelper<Z>(pw->getFuture(), fns...));
1025     return [=](Try<A> t) mutable {
1026       pw->setTry(std::move(t));
1027       return std::move(*fw);
1028     };
1029   }
1030
1031 }
1032
1033 } // namespace folly
1034
1035 // I haven't included a Future<T&> specialization because I don't forsee us
1036 // using it, however it is not difficult to add when needed. Refer to
1037 // Future<void> for guidance. std::future and boost::future code would also be
1038 // instructive.