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 Future<void> Future<T>::then() {
 240   return then([] (Try<T>&& t) {});
 241 }
 242
 243 // onError where the callback returns T
 244 template <class T>
 245 template <class F>
 246 typename std::enable_if<
 247   !detail::callableWith<F, exception_wrapper>::value &&
 248   !detail::Extract<F>::ReturnsFuture::value,
 249   Future<T>>::type
 250 Future<T>::onError(F&& func) {
 251   typedef typename detail::Extract<F>::FirstArg Exn;
 252   static_assert(
 253       std::is_same<typename detail::Extract<F>::RawReturn, T>::value,
 254       "Return type of onError callback must be T or Future<T>");
 255
 256   Promise<T> p;
 257   auto f = p.getFuture();
 258   auto pm = folly::makeMoveWrapper(std::move(p));
 259   auto funcm = folly::makeMoveWrapper(std::move(func));
 260   setCallback_([pm, funcm](Try<T>&& t) mutable {
 261     if (!t.template withException<Exn>([&] (Exn& e) {
 262           pm->setWith([&]{
 263             return (*funcm)(e);
 264           });
 265         })) {
 266       pm->setTry(std::move(t));
 267     }
 268   });
 269
 270   return f;
 271 }
 272
 273 // onError where the callback returns Future<T>
 274 template <class T>
 275 template <class F>
 276 typename std::enable_if<
 277   !detail::callableWith<F, exception_wrapper>::value &&
 278   detail::Extract<F>::ReturnsFuture::value,
 279   Future<T>>::type
 280 Future<T>::onError(F&& func) {
 281   static_assert(
 282       std::is_same<typename detail::Extract<F>::Return, Future<T>>::value,
 283       "Return type of onError callback must be T or Future<T>");
 284   typedef typename detail::Extract<F>::FirstArg Exn;
 285
 286   Promise<T> p;
 287   auto f = p.getFuture();
 288   auto pm = folly::makeMoveWrapper(std::move(p));
 289   auto funcm = folly::makeMoveWrapper(std::move(func));
 290   setCallback_([pm, funcm](Try<T>&& t) mutable {
 291     if (!t.template withException<Exn>([&] (Exn& e) {
 292           try {
 293             auto f2 = (*funcm)(e);
 294             f2.setCallback_([pm](Try<T>&& t2) mutable {
 295               pm->setTry(std::move(t2));
 296             });
 297           } catch (const std::exception& e2) {
 298             pm->setException(exception_wrapper(std::current_exception(), e2));
 299           } catch (...) {
 300             pm->setException(exception_wrapper(std::current_exception()));
 301           }
 302         })) {
 303       pm->setTry(std::move(t));
 304     }
 305   });
 306
 307   return f;
 308 }
 309
 310 template <class T>
 311 template <class F>
 312 Future<T> Future<T>::ensure(F func) {
 313   MoveWrapper<F> funcw(std::move(func));
 314   return this->then([funcw](Try<T>&& t) {
 315     (*funcw)();
 316     return makeFuture(std::move(t));
 317   });
 318 }
 319
 320 template <class T>
 321 template <class F>
 322 Future<T> Future<T>::onTimeout(Duration dur, F&& func, Timekeeper* tk) {
 323   auto funcw = folly::makeMoveWrapper(std::forward<F>(func));
 324   return within(dur, tk)
 325     .onError([funcw](TimedOut const&) { return (*funcw)(); });
 326 }
 327
 328 template <class T>
 329 template <class F>
 330 typename std::enable_if<
 331   detail::callableWith<F, exception_wrapper>::value &&
 332   detail::Extract<F>::ReturnsFuture::value,
 333   Future<T>>::type
 334 Future<T>::onError(F&& func) {
 335   static_assert(
 336       std::is_same<typename detail::Extract<F>::Return, Future<T>>::value,
 337       "Return type of onError callback must be T or Future<T>");
 338
 339   Promise<T> p;
 340   auto f = p.getFuture();
 341   auto pm = folly::makeMoveWrapper(std::move(p));
 342   auto funcm = folly::makeMoveWrapper(std::move(func));
 343   setCallback_([pm, funcm](Try<T> t) mutable {
 344     if (t.hasException()) {
 345       try {
 346         auto f2 = (*funcm)(std::move(t.exception()));
 347         f2.setCallback_([pm](Try<T> t2) mutable {
 348           pm->setTry(std::move(t2));
 349         });
 350       } catch (const std::exception& e2) {
 351         pm->setException(exception_wrapper(std::current_exception(), e2));
 352       } catch (...) {
 353         pm->setException(exception_wrapper(std::current_exception()));
 354       }
 355     } else {
 356       pm->setTry(std::move(t));
 357     }
 358   });
 359
 360   return f;
 361 }
 362
 363 // onError(exception_wrapper) that returns T
 364 template <class T>
 365 template <class F>
 366 typename std::enable_if<
 367   detail::callableWith<F, exception_wrapper>::value &&
 368   !detail::Extract<F>::ReturnsFuture::value,
 369   Future<T>>::type
 370 Future<T>::onError(F&& func) {
 371   static_assert(
 372       std::is_same<typename detail::Extract<F>::Return, Future<T>>::value,
 373       "Return type of onError callback must be T or Future<T>");
 374
 375   Promise<T> p;
 376   auto f = p.getFuture();
 377   auto pm = folly::makeMoveWrapper(std::move(p));
 378   auto funcm = folly::makeMoveWrapper(std::move(func));
 379   setCallback_([pm, funcm](Try<T> t) mutable {
 380     if (t.hasException()) {
 381       pm->setWith([&]{
 382         return (*funcm)(std::move(t.exception()));
 383       });
 384     } else {
 385       pm->setTry(std::move(t));
 386     }
 387   });
 388
 389   return f;
 390 }
 391
 392 template <class T>
 393 typename std::add_lvalue_reference<T>::type Future<T>::value() {
 394   throwIfInvalid();
 395
 396   return core_->getTry().value();
 397 }
 398
 399 template <class T>
 400 typename std::add_lvalue_reference<const T>::type Future<T>::value() const {
 401   throwIfInvalid();
 402
 403   return core_->getTry().value();
 404 }
 405
 406 template <class T>
 407 Try<T>& Future<T>::getTry() {
 408   throwIfInvalid();
 409
 410   return core_->getTry();
 411 }
 412
 413 template <class T>
 414 Optional<Try<T>> Future<T>::poll() {
 415   Optional<Try<T>> o;
 416   if (core_->ready()) {
 417     o = std::move(core_->getTry());
 418   }
 419   return o;
 420 }
 421
 422 template <class T>
 423 template <typename Executor>
 424 inline Future<T> Future<T>::via(Executor* executor) && {
 425   throwIfInvalid();
 426
 427   setExecutor(executor);
 428
 429   return std::move(*this);
 430 }
 431
 432 template <class T>
 433 template <typename Executor>
 434 inline Future<T> Future<T>::via(Executor* executor) & {
 435   throwIfInvalid();
 436
 437   MoveWrapper<Promise<T>> p;
 438   auto f = p->getFuture();
 439   then([p](Try<T>&& t) mutable { p->setTry(std::move(t)); });
 440   return std::move(f).via(executor);
 441 }
 442
 443 template <class T>
 444 bool Future<T>::isReady() const {
 445   throwIfInvalid();
 446   return core_->ready();
 447 }
 448
 449 template <class T>
 450 void Future<T>::raise(exception_wrapper exception) {
 451   core_->raise(std::move(exception));
 452 }
 453
 454 // makeFuture
 455
 456 template <class T>
 457 Future<typename std::decay<T>::type> makeFuture(T&& t) {
 458   Promise<typename std::decay<T>::type> p;
 459   p.setValue(std::forward<T>(t));
 460   return p.getFuture();
 461 }
 462
 463 inline // for multiple translation units
 464 Future<void> makeFuture() {
 465   Promise<void> p;
 466   p.setValue();
 467   return p.getFuture();
 468 }
 469
 470 template <class F>
 471 auto makeFutureTry(
 472     F&& func,
 473     typename std::enable_if<!std::is_reference<F>::value, bool>::type sdf)
 474     -> Future<decltype(func())> {
 475   Promise<decltype(func())> p;
 476   p.setWith(
 477     [&func]() {
 478       return (func)();
 479     });
 480   return p.getFuture();
 481 }
 482
 483 template <class F>
 484 auto makeFutureTry(F const& func) -> Future<decltype(func())> {
 485   F copy = func;
 486   return makeFutureTry(std::move(copy));
 487 }
 488
 489 template <class T>
 490 Future<T> makeFuture(std::exception_ptr const& e) {
 491   Promise<T> p;
 492   p.setException(e);
 493   return p.getFuture();
 494 }
 495
 496 template <class T>
 497 Future<T> makeFuture(exception_wrapper ew) {
 498   Promise<T> p;
 499   p.setException(std::move(ew));
 500   return p.getFuture();
 501 }
 502
 503 template <class T, class E>
 504 typename std::enable_if<std::is_base_of<std::exception, E>::value,
 505                         Future<T>>::type
 506 makeFuture(E const& e) {
 507   Promise<T> p;
 508   p.setException(make_exception_wrapper<E>(e));
 509   return p.getFuture();
 510 }
 511
 512 template <class T>
 513 Future<T> makeFuture(Try<T>&& t) {
 514   Promise<typename std::decay<T>::type> p;
 515   p.setTry(std::move(t));
 516   return p.getFuture();
 517 }
 518
 519 template <>
 520 inline Future<void> makeFuture(Try<void>&& t) {
 521   if (t.hasException()) {
 522     return makeFuture<void>(std::move(t.exception()));
 523   } else {
 524     return makeFuture();
 525   }
 526 }
 527
 528 // via
 529 template <typename Executor>
 530 Future<void> via(Executor* executor) {
 531   return makeFuture().via(executor);
 532 }
 533
 534 // when (variadic)
 535
 536 template <typename... Fs>
 537 typename detail::VariadicContext<
 538   typename std::decay<Fs>::type::value_type...>::type
 539 whenAll(Fs&&... fs) {
 540   auto ctx =
 541     new detail::VariadicContext<typename std::decay<Fs>::type::value_type...>();
 542   ctx->total = sizeof...(fs);
 543   auto f_saved = ctx->p.getFuture();
 544   detail::whenAllVariadicHelper(ctx,
 545     std::forward<typename std::decay<Fs>::type>(fs)...);
 546   return f_saved;
 547 }
 548
 549 // when (iterator)
 550
 551 template <class InputIterator>
 552 Future<
 553   std::vector<
 554   Try<typename std::iterator_traits<InputIterator>::value_type::value_type>>>
 555 whenAll(InputIterator first, InputIterator last) {
 556   typedef
 557     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 558
 559   if (first >= last) {
 560     return makeFuture(std::vector<Try<T>>());
 561   }
 562   size_t n = std::distance(first, last);
 563
 564   auto ctx = new detail::WhenAllContext<T>();
 565
 566   ctx->results.resize(n);
 567
 568   auto f_saved = ctx->p.getFuture();
 569
 570   for (size_t i = 0; first != last; ++first, ++i) {
 571      assert(i < n);
 572      auto& f = *first;
 573      f.setCallback_([ctx, i, n](Try<T> t) {
 574        ctx->results[i] = std::move(t);
 575        if (++ctx->count == n) {
 576          ctx->p.setValue(std::move(ctx->results));
 577          delete ctx;
 578        }
 579      });
 580   }
 581
 582   return f_saved;
 583 }
 584
 585 namespace detail {
 586
 587 template <typename T>
 588 struct CollectContext {
 589   explicit CollectContext(int n) : count(0), threw(false) {}
 590
 591   Promise<std::vector<T>> p;
 592   std::vector<T> results;
 593   std::atomic<size_t> count;
 594   std::atomic_bool threw;
 595
 596   typedef std::vector<T> result_type;
 597
 598   static inline Future<std::vector<T>> makeEmptyFuture() {
 599     return makeFuture(std::vector<T>());
 600   }
 601
 602   inline void setValue() {
 603     p.setValue(std::move(results));
 604   }
 605
 606   inline void addResult(int i, Try<T>& t) {
 607     results[i] = std::move(t.value());
 608   }
 609 };
 610
 611 template <>
 612 struct CollectContext<void> {
 613   explicit CollectContext(int n) : count(0), threw(false) {}
 614   Promise<void> p;
 615   std::atomic<size_t> count;
 616   std::atomic_bool threw;
 617
 618   typedef void result_type;
 619
 620   static inline Future<void> makeEmptyFuture() {
 621     return makeFuture();
 622   }
 623
 624   inline void setValue() {
 625     p.setValue();
 626   }
 627
 628   inline void addResult(int i, Try<void>& t) {
 629     // do nothing
 630   }
 631 };
 632
 633 } // detail
 634
 635 template <class InputIterator>
 636 Future<typename detail::CollectContext<
 637   typename std::iterator_traits<InputIterator>::value_type::value_type
 638 >::result_type>
 639 collect(InputIterator first, InputIterator last) {
 640   typedef
 641     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 642
 643   if (first >= last) {
 644     return detail::CollectContext<T>::makeEmptyFuture();
 645   }
 646
 647   size_t n = std::distance(first, last);
 648   auto ctx = new detail::CollectContext<T>(n);
 649   auto f_saved = ctx->p.getFuture();
 650
 651   for (size_t i = 0; first != last; ++first, ++i) {
 652      assert(i < n);
 653      auto& f = *first;
 654      f.setCallback_([ctx, i, n](Try<T> t) {
 655        auto c = ++ctx->count;
 656
 657        if (t.hasException()) {
 658          if (!ctx->threw.exchange(true)) {
 659            ctx->p.setException(std::move(t.exception()));
 660          }
 661        } else if (!ctx->threw) {
 662          ctx->addResult(i, t);
 663          if (c == n) {
 664            ctx->setValue();
 665          }
 666        }
 667
 668        if (c == n) {
 669          delete ctx;
 670        }
 671      });
 672   }
 673
 674   return f_saved;
 675 }
 676
 677 template <class InputIterator>
 678 Future<
 679   std::pair<size_t,
 680             Try<
 681               typename
 682               std::iterator_traits<InputIterator>::value_type::value_type> > >
 683 whenAny(InputIterator first, InputIterator last) {
 684   typedef
 685     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 686
 687   auto ctx = new detail::WhenAnyContext<T>(std::distance(first, last));
 688   auto f_saved = ctx->p.getFuture();
 689
 690   for (size_t i = 0; first != last; first++, i++) {
 691     auto& f = *first;
 692     f.setCallback_([i, ctx](Try<T>&& t) {
 693       if (!ctx->done.exchange(true)) {
 694         ctx->p.setValue(std::make_pair(i, std::move(t)));
 695       }
 696       ctx->decref();
 697     });
 698   }
 699
 700   return f_saved;
 701 }
 702
 703 template <class InputIterator>
 704 Future<std::vector<std::pair<size_t, Try<typename
 705   std::iterator_traits<InputIterator>::value_type::value_type>>>>
 706 whenN(InputIterator first, InputIterator last, size_t n) {
 707   typedef typename
 708     std::iterator_traits<InputIterator>::value_type::value_type T;
 709   typedef std::vector<std::pair<size_t, Try<T>>> V;
 710
 711   struct ctx_t {
 712     V v;
 713     size_t completed;
 714     Promise<V> p;
 715   };
 716   auto ctx = std::make_shared<ctx_t>();
 717   ctx->completed = 0;
 718
 719   // for each completed Future, increase count and add to vector, until we
 720   // have n completed futures at which point we fulfill our Promise with the
 721   // vector
 722   auto it = first;
 723   size_t i = 0;
 724   while (it != last) {
 725     it->then([ctx, n, i](Try<T>&& t) {
 726       auto& v = ctx->v;
 727       auto c = ++ctx->completed;
 728       if (c <= n) {
 729         assert(ctx->v.size() < n);
 730         v.push_back(std::make_pair(i, std::move(t)));
 731         if (c == n) {
 732           ctx->p.setTry(Try<V>(std::move(v)));
 733         }
 734       }
 735     });
 736
 737     it++;
 738     i++;
 739   }
 740
 741   if (i < n) {
 742     ctx->p.setException(std::runtime_error("Not enough futures"));
 743   }
 744
 745   return ctx->p.getFuture();
 746 }
 747
 748 template <class It, class T, class F, class ItT, class Arg>
 749 typename std::enable_if<!isFutureResult<F, T, Arg>::value, Future<T>>::type
 750 reduce(It first, It last, T initial, F func) {
 751   if (first == last) {
 752     return makeFuture(std::move(initial));
 753   }
 754
 755   typedef isTry<Arg> IsTry;
 756
 757   return whenAll(first, last)
 758     .then([initial, func](std::vector<Try<ItT>>& vals) mutable {
 759       for (auto& val : vals) {
 760         initial = func(std::move(initial),
 761                        // Either return a ItT&& or a Try<ItT>&& depending
 762                        // on the type of the argument of func.
 763                        val.template get<IsTry::value, Arg&&>());
 764       }
 765       return initial;
 766     });
 767 }
 768
 769 template <class It, class T, class F, class ItT, class Arg>
 770 typename std::enable_if<isFutureResult<F, T, Arg>::value, Future<T>>::type
 771 reduce(It first, It last, T initial, F func) {
 772   if (first == last) {
 773     return makeFuture(std::move(initial));
 774   }
 775
 776   typedef isTry<Arg> IsTry;
 777
 778   auto f = first->then([initial, func](Try<ItT>& head) mutable {
 779     return func(std::move(initial),
 780                 head.template get<IsTry::value, Arg&&>());
 781   });
 782
 783   for (++first; first != last; ++first) {
 784     f = whenAll(f, *first).then([func](std::tuple<Try<T>, Try<ItT>>& t) {
 785       return func(std::move(std::get<0>(t).value()),
 786                   // Either return a ItT&& or a Try<ItT>&& depending
 787                   // on the type of the argument of func.
 788                   std::get<1>(t).template get<IsTry::value, Arg&&>());
 789     });
 790   }
 791
 792   return f;
 793 }
 794
 795 template <class T>
 796 Future<T> Future<T>::within(Duration dur, Timekeeper* tk) {
 797   return within(dur, TimedOut(), tk);
 798 }
 799
 800 template <class T>
 801 template <class E>
 802 Future<T> Future<T>::within(Duration dur, E e, Timekeeper* tk) {
 803
 804   struct Context {
 805     Context(E ex) : exception(std::move(ex)), promise(), token(false) {}
 806     E exception;
 807     Promise<T> promise;
 808     std::atomic<bool> token;
 809   };
 810   auto ctx = std::make_shared<Context>(std::move(e));
 811
 812   if (!tk) {
 813     tk = folly::detail::getTimekeeperSingleton();
 814   }
 815
 816   tk->after(dur)
 817     .then([ctx](Try<void> const& t) {
 818       if (ctx->token.exchange(true) == false) {
 819         if (t.hasException()) {
 820           ctx->promise.setException(std::move(t.exception()));
 821         } else {
 822           ctx->promise.setException(std::move(ctx->exception));
 823         }
 824       }
 825     });
 826
 827   this->then([ctx](Try<T>&& t) {
 828     if (ctx->token.exchange(true) == false) {
 829       ctx->promise.setTry(std::move(t));
 830     }
 831   });
 832
 833   return ctx->promise.getFuture();
 834 }
 835
 836 template <class T>
 837 Future<T> Future<T>::delayed(Duration dur, Timekeeper* tk) {
 838   return whenAll(*this, futures::sleep(dur, tk))
 839     .then([](std::tuple<Try<T>, Try<void>> tup) {
 840       Try<T>& t = std::get<0>(tup);
 841       return makeFuture<T>(std::move(t));
 842     });
 843 }
 844
 845 namespace detail {
 846
 847 template <class T>
 848 void waitImpl(Future<T>& f) {
 849   // short-circuit if there's nothing to do
 850   if (f.isReady()) return;
 851
 852   Baton<> baton;
 853   f = f.then([&](Try<T> t) {
 854     baton.post();
 855     return makeFuture(std::move(t));
 856   });
 857   baton.wait();
 858
 859   // There's a race here between the return here and the actual finishing of
 860   // the future. f is completed, but the setup may not have finished on done
 861   // after the baton has posted.
 862   while (!f.isReady()) {
 863     std::this_thread::yield();
 864   }
 865 }
 866
 867 template <class T>
 868 void waitImpl(Future<T>& f, Duration dur) {
 869   // short-circuit if there's nothing to do
 870   if (f.isReady()) return;
 871
 872   auto baton = std::make_shared<Baton<>>();
 873   f = f.then([baton](Try<T> t) {
 874     baton->post();
 875     return makeFuture(std::move(t));
 876   });
 877
 878   // Let's preserve the invariant that if we did not timeout (timed_wait returns
 879   // true), then the returned Future is complete when it is returned to the
 880   // caller. We need to wait out the race for that Future to complete.
 881   if (baton->timed_wait(std::chrono::system_clock::now() + dur)) {
 882     while (!f.isReady()) {
 883       std::this_thread::yield();
 884     }
 885   }
 886 }
 887
 888 template <class T>
 889 void waitViaImpl(Future<T>& f, DrivableExecutor* e) {
 890   while (!f.isReady()) {
 891     e->drive();
 892   }
 893 }
 894
 895 } // detail
 896
 897 template <class T>
 898 Future<T>& Future<T>::wait() & {
 899   detail::waitImpl(*this);
 900   return *this;
 901 }
 902
 903 template <class T>
 904 Future<T>&& Future<T>::wait() && {
 905   detail::waitImpl(*this);
 906   return std::move(*this);
 907 }
 908
 909 template <class T>
 910 Future<T>& Future<T>::wait(Duration dur) & {
 911   detail::waitImpl(*this, dur);
 912   return *this;
 913 }
 914
 915 template <class T>
 916 Future<T>&& Future<T>::wait(Duration dur) && {
 917   detail::waitImpl(*this, dur);
 918   return std::move(*this);
 919 }
 920
 921 template <class T>
 922 Future<T>& Future<T>::waitVia(DrivableExecutor* e) & {
 923   detail::waitViaImpl(*this, e);
 924   return *this;
 925 }
 926
 927 template <class T>
 928 Future<T>&& Future<T>::waitVia(DrivableExecutor* e) && {
 929   detail::waitViaImpl(*this, e);
 930   return std::move(*this);
 931 }
 932
 933 template <class T>
 934 T Future<T>::get() {
 935   return std::move(wait().value());
 936 }
 937
 938 template <>
 939 inline void Future<void>::get() {
 940   wait().value();
 941 }
 942
 943 template <class T>
 944 T Future<T>::get(Duration dur) {
 945   wait(dur);
 946   if (isReady()) {
 947     return std::move(value());
 948   } else {
 949     throw TimedOut();
 950   }
 951 }
 952
 953 template <>
 954 inline void Future<void>::get(Duration dur) {
 955   wait(dur);
 956   if (isReady()) {
 957     return;
 958   } else {
 959     throw TimedOut();
 960   }
 961 }
 962
 963 template <class T>
 964 T Future<T>::getVia(DrivableExecutor* e) {
 965   return std::move(waitVia(e).value());
 966 }
 967
 968 template <>
 969 inline void Future<void>::getVia(DrivableExecutor* e) {
 970   waitVia(e).value();
 971 }
 972
 973 template <class T>
 974 Future<bool> Future<T>::willEqual(Future<T>& f) {
 975   return whenAll(*this, f).then([](const std::tuple<Try<T>, Try<T>>& t) {
 976     if (std::get<0>(t).hasValue() && std::get<1>(t).hasValue()) {
 977       return std::get<0>(t).value() == std::get<1>(t).value();
 978     } else {
 979       return false;
 980       }
 981   });
 982 }
 983
 984 template <class T>
 985 template <class F>
 986 Future<T> Future<T>::filter(F predicate) {
 987   auto p = folly::makeMoveWrapper(std::move(predicate));
 988   return this->then([p](T val) {
 989     T const& valConstRef = val;
 990     if (!(*p)(valConstRef)) {
 991       throw PredicateDoesNotObtain();
 992     }
 993     return val;
 994   });
 995 }
 996
 997 namespace futures {
 998   namespace {
 999     template <class Z>
1000     Future<Z> chainHelper(Future<Z> f) {
1001       return f;
1002     }
1003
1004     template <class Z, class F, class Fn, class... Callbacks>
1005     Future<Z> chainHelper(F f, Fn fn, Callbacks... fns) {
1006       return chainHelper<Z>(f.then(fn), fns...);
1007     }
1008   }
1009
1010   template <class A, class Z, class... Callbacks>
1011   std::function<Future<Z>(Try<A>)>
1012   chain(Callbacks... fns) {
1013     MoveWrapper<Promise<A>> pw;
1014     MoveWrapper<Future<Z>> fw(chainHelper<Z>(pw->getFuture(), fns...));
1015     return [=](Try<A> t) mutable {
1016       pw->setTry(std::move(t));
1017       return std::move(*fw);
1018     };
1019   }
1020
1021   template <class It, class F, class ItT, class Result>
1022   std::vector<Future<Result>> map(It first, It last, F func) {
1023     std::vector<Future<Result>> results;
1024     for (auto it = first; it != last; it++) {
1025       results.push_back(it->then(func));
1026     }
1027     return results;
1028   }
1029 }
1030
1031 } // namespace folly
1032
1033 // I haven't included a Future<T&> specialization because I don't forsee us
1034 // using it, however it is not difficult to add when needed. Refer to
1035 // Future<void> for guidance. std::future and boost::future code would also be
1036 // instructive.