folly/wangle/Future-inl.h

   1 /*
   2  * Copyright 2014 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/wangle/detail/Core.h>
  23 #include <folly/Baton.h>
  24
  25 namespace folly { namespace wangle {
  26
  27 template <typename T>
  28 struct isFuture {
  29   static const bool value = false;
  30 };
  31
  32 template <typename T>
  33 struct isFuture<Future<T> > {
  34   static const bool value = true;
  35 };
  36
  37 template <class T>
  38 Future<T>::Future(Future<T>&& other) noexcept : core_(nullptr) {
  39   *this = std::move(other);
  40 }
  41
  42 template <class T>
  43 Future<T>& Future<T>::operator=(Future<T>&& other) {
  44   std::swap(core_, other.core_);
  45   return *this;
  46 }
  47
  48 template <class T>
  49 Future<T>::~Future() {
  50   detach();
  51 }
  52
  53 template <class T>
  54 void Future<T>::detach() {
  55   if (core_) {
  56     core_->detachFuture();
  57     core_ = nullptr;
  58   }
  59 }
  60
  61 template <class T>
  62 void Future<T>::throwIfInvalid() const {
  63   if (!core_)
  64     throw NoState();
  65 }
  66
  67 template <class T>
  68 template <class F>
  69 void Future<T>::setCallback_(F&& func) {
  70   throwIfInvalid();
  71   core_->setCallback(std::move(func));
  72 }
  73
  74 // Variant: f.then([](Try<T>&& t){ return t.value(); });
  75 template <class T>
  76 template <class F>
  77 typename std::enable_if<
  78   !isFuture<typename std::result_of<F(Try<T>&&)>::type>::value,
  79   Future<typename std::result_of<F(Try<T>&&)>::type> >::type
  80 Future<T>::then(F&& func) {
  81   typedef typename std::result_of<F(Try<T>&&)>::type B;
  82
  83   throwIfInvalid();
  84
  85   // wrap these so we can move them into the lambda
  86   folly::MoveWrapper<Promise<B>> p;
  87   folly::MoveWrapper<F> funcm(std::forward<F>(func));
  88
  89   // grab the Future now before we lose our handle on the Promise
  90   auto f = p->getFuture();
  91
  92   /* This is a bit tricky.
  93
  94      We can't just close over *this in case this Future gets moved. So we
  95      make a new dummy Future. We could figure out something more
  96      sophisticated that avoids making a new Future object when it can, as an
  97      optimization. But this is correct.
  98
  99      core_ can't be moved, it is explicitly disallowed (as is copying). But
 100      if there's ever a reason to allow it, this is one place that makes that
 101      assumption and would need to be fixed. We use a standard shared pointer
 102      for core_ (by copying it in), which means in essence obj holds a shared
 103      pointer to itself.  But this shouldn't leak because Promise will not
 104      outlive the continuation, because Promise will setException() with a
 105      broken Promise if it is destructed before completed. We could use a
 106      weak pointer but it would have to be converted to a shared pointer when
 107      func is executed (because the Future returned by func may possibly
 108      persist beyond the callback, if it gets moved), and so it is an
 109      optimization to just make it shared from the get-go.
 110
 111      We have to move in the Promise and func using the MoveWrapper
 112      hack. (func could be copied but it's a big drag on perf).
 113
 114      Two subtle but important points about this design. detail::Core has no
 115      back pointers to Future or Promise, so if Future or Promise get moved
 116      (and they will be moved in performant code) we don't have to do
 117      anything fancy. And because we store the continuation in the
 118      detail::Core, not in the Future, we can execute the continuation even
 119      after the Future has gone out of scope. This is an intentional design
 120      decision. It is likely we will want to be able to cancel a continuation
 121      in some circumstances, but I think it should be explicit not implicit
 122      in the destruction of the Future used to create it.
 123      */
 124   setCallback_(
 125     [p, funcm](Try<T>&& t) mutable {
 126       p->fulfil([&]() {
 127           return (*funcm)(std::move(t));
 128         });
 129     });
 130
 131   return std::move(f);
 132 }
 133
 134 // Variant: f.then([](T&& t){ return t; });
 135 template <class T>
 136 template <class F>
 137 typename std::enable_if<
 138   !std::is_same<T, void>::value &&
 139   !isFuture<typename std::result_of<
 140     F(typename detail::AliasIfVoid<T>::type&&)>::type>::value,
 141   Future<typename std::result_of<
 142     F(typename detail::AliasIfVoid<T>::type&&)>::type> >::type
 143 Future<T>::then(F&& func) {
 144   typedef typename std::result_of<F(T&&)>::type B;
 145
 146   throwIfInvalid();
 147
 148   folly::MoveWrapper<Promise<B>> p;
 149   folly::MoveWrapper<F> funcm(std::forward<F>(func));
 150   auto f = p->getFuture();
 151
 152   setCallback_(
 153     [p, funcm](Try<T>&& t) mutable {
 154       if (t.hasException()) {
 155         p->setException(t.getException());
 156       } else {
 157         p->fulfil([&]() {
 158           return (*funcm)(std::move(t.value()));
 159         });
 160       }
 161     });
 162
 163   return std::move(f);
 164 }
 165
 166 // Variant: f.then([](){ return; });
 167 template <class T>
 168 template <class F>
 169 typename std::enable_if<
 170   std::is_same<T, void>::value &&
 171   !isFuture<typename std::result_of<F()>::type>::value,
 172   Future<typename std::result_of<F()>::type> >::type
 173 Future<T>::then(F&& func) {
 174   typedef typename std::result_of<F()>::type B;
 175
 176   throwIfInvalid();
 177
 178   folly::MoveWrapper<Promise<B>> p;
 179   folly::MoveWrapper<F> funcm(std::forward<F>(func));
 180   auto f = p->getFuture();
 181
 182   setCallback_(
 183     [p, funcm](Try<T>&& t) mutable {
 184       if (t.hasException()) {
 185         p->setException(t.getException());
 186       } else {
 187         p->fulfil([&]() {
 188           return (*funcm)();
 189         });
 190       }
 191     });
 192
 193   return std::move(f);
 194 }
 195
 196 // Variant: f.then([](Try<T>&& t){ return makeFuture<T>(t.value()); });
 197 template <class T>
 198 template <class F>
 199 typename std::enable_if<
 200   isFuture<typename std::result_of<F(Try<T>&&)>::type>::value,
 201   Future<typename std::result_of<F(Try<T>&&)>::type::value_type> >::type
 202 Future<T>::then(F&& func) {
 203   typedef typename std::result_of<F(Try<T>&&)>::type::value_type B;
 204
 205   throwIfInvalid();
 206
 207   // wrap these so we can move them into the lambda
 208   folly::MoveWrapper<Promise<B>> p;
 209   folly::MoveWrapper<F> funcm(std::forward<F>(func));
 210
 211   // grab the Future now before we lose our handle on the Promise
 212   auto f = p->getFuture();
 213
 214   setCallback_(
 215     [p, funcm](Try<T>&& t) mutable {
 216       try {
 217         auto f2 = (*funcm)(std::move(t));
 218         // that didn't throw, now we can steal p
 219         f2.setCallback_([p](Try<B>&& b) mutable {
 220             p->fulfilTry(std::move(b));
 221           });
 222       } catch (...) {
 223         p->setException(std::current_exception());
 224       }
 225     });
 226
 227   return std::move(f);
 228 }
 229
 230 // Variant: f.then([](T&& t){ return makeFuture<T>(t); });
 231 template <class T>
 232 template <class F>
 233 typename std::enable_if<
 234   !std::is_same<T, void>::value &&
 235   isFuture<typename std::result_of<
 236     F(typename detail::AliasIfVoid<T>::type&&)>::type>::value,
 237   Future<typename std::result_of<
 238     F(typename detail::AliasIfVoid<T>::type&&)>::type::value_type> >::type
 239 Future<T>::then(F&& func) {
 240   typedef typename std::result_of<F(T&&)>::type::value_type B;
 241
 242   throwIfInvalid();
 243
 244   folly::MoveWrapper<Promise<B>> p;
 245   folly::MoveWrapper<F> funcm(std::forward<F>(func));
 246   auto f = p->getFuture();
 247
 248   setCallback_(
 249     [p, funcm](Try<T>&& t) mutable {
 250       if (t.hasException()) {
 251         p->setException(t.getException());
 252       } else {
 253         try {
 254           auto f2 = (*funcm)(std::move(t.value()));
 255           f2.setCallback_([p](Try<B>&& b) mutable {
 256               p->fulfilTry(std::move(b));
 257             });
 258         } catch (...) {
 259           p->setException(std::current_exception());
 260         }
 261       }
 262     });
 263
 264   return std::move(f);
 265 }
 266
 267 // Variant: f.then([](){ return makeFuture(); });
 268 template <class T>
 269 template <class F>
 270 typename std::enable_if<
 271   std::is_same<T, void>::value &&
 272   isFuture<typename std::result_of<F()>::type>::value,
 273   Future<typename std::result_of<F()>::type::value_type> >::type
 274 Future<T>::then(F&& func) {
 275   typedef typename std::result_of<F()>::type::value_type B;
 276
 277   throwIfInvalid();
 278
 279   folly::MoveWrapper<Promise<B>> p;
 280   folly::MoveWrapper<F> funcm(std::forward<F>(func));
 281
 282   auto f = p->getFuture();
 283
 284   setCallback_(
 285     [p, funcm](Try<T>&& t) mutable {
 286       if (t.hasException()) {
 287         p->setException(t.getException());
 288       } else {
 289         try {
 290           auto f2 = (*funcm)();
 291           f2.setCallback_([p](Try<B>&& b) mutable {
 292               p->fulfilTry(std::move(b));
 293             });
 294         } catch (...) {
 295           p->setException(std::current_exception());
 296         }
 297       }
 298     });
 299
 300   return std::move(f);
 301 }
 302
 303 template <class T>
 304 Future<void> Future<T>::then() {
 305   return then([] (Try<T>&& t) {});
 306 }
 307
 308 template <class T>
 309 typename std::add_lvalue_reference<T>::type Future<T>::value() {
 310   throwIfInvalid();
 311
 312   return core_->getTry().value();
 313 }
 314
 315 template <class T>
 316 typename std::add_lvalue_reference<const T>::type Future<T>::value() const {
 317   throwIfInvalid();
 318
 319   return core_->getTry().value();
 320 }
 321
 322 template <class T>
 323 Try<T>& Future<T>::getTry() {
 324   throwIfInvalid();
 325
 326   return core_->getTry();
 327 }
 328
 329 template <class T>
 330 template <typename Executor>
 331 inline Future<T> Future<T>::via(Executor* executor) {
 332   throwIfInvalid();
 333
 334   this->deactivate();
 335   core_->setExecutor(executor);
 336
 337   return std::move(*this);
 338 }
 339
 340 template <class T>
 341 bool Future<T>::isReady() const {
 342   throwIfInvalid();
 343   return core_->ready();
 344 }
 345
 346 template <class T>
 347 void Future<T>::raise(std::exception_ptr exception) {
 348   core_->raise(exception);
 349 }
 350
 351 // makeFuture
 352
 353 template <class T>
 354 Future<typename std::decay<T>::type> makeFuture(T&& t) {
 355   Promise<typename std::decay<T>::type> p;
 356   auto f = p.getFuture();
 357   p.setValue(std::forward<T>(t));
 358   return std::move(f);
 359 }
 360
 361 inline // for multiple translation units
 362 Future<void> makeFuture() {
 363   Promise<void> p;
 364   auto f = p.getFuture();
 365   p.setValue();
 366   return std::move(f);
 367 }
 368
 369 template <class F>
 370 auto makeFutureTry(
 371     F&& func,
 372     typename std::enable_if<!std::is_reference<F>::value, bool>::type sdf)
 373     -> Future<decltype(func())> {
 374   Promise<decltype(func())> p;
 375   auto f = p.getFuture();
 376   p.fulfil(
 377     [&func]() {
 378       return (func)();
 379     });
 380   return std::move(f);
 381 }
 382
 383 template <class F>
 384 auto makeFutureTry(F const& func) -> Future<decltype(func())> {
 385   F copy = func;
 386   return makeFutureTry(std::move(copy));
 387 }
 388
 389 template <class T>
 390 Future<T> makeFuture(std::exception_ptr const& e) {
 391   Promise<T> p;
 392   auto f = p.getFuture();
 393   p.setException(e);
 394   return std::move(f);
 395 }
 396
 397 template <class T, class E>
 398 typename std::enable_if<std::is_base_of<std::exception, E>::value,
 399                         Future<T>>::type
 400 makeFuture(E const& e) {
 401   Promise<T> p;
 402   auto f = p.getFuture();
 403   p.fulfil([&]() -> T { throw e; });
 404   return std::move(f);
 405 }
 406
 407 template <class T>
 408 Future<T> makeFuture(Try<T>&& t) {
 409   try {
 410     return makeFuture<T>(std::move(t.value()));
 411   } catch (...) {
 412     return makeFuture<T>(std::current_exception());
 413   }
 414 }
 415
 416 template <>
 417 inline Future<void> makeFuture(Try<void>&& t) {
 418   try {
 419     t.throwIfFailed();
 420     return makeFuture();
 421   } catch (...) {
 422     return makeFuture<void>(std::current_exception());
 423   }
 424 }
 425
 426 // via
 427 template <typename Executor>
 428 Future<void> via(Executor* executor) {
 429   return makeFuture().via(executor);
 430 }
 431
 432 // when (variadic)
 433
 434 template <typename... Fs>
 435 typename detail::VariadicContext<
 436   typename std::decay<Fs>::type::value_type...>::type
 437 whenAll(Fs&&... fs)
 438 {
 439   auto ctx =
 440     new detail::VariadicContext<typename std::decay<Fs>::type::value_type...>();
 441   ctx->total = sizeof...(fs);
 442   auto f_saved = ctx->p.getFuture();
 443   detail::whenAllVariadicHelper(ctx,
 444     std::forward<typename std::decay<Fs>::type>(fs)...);
 445   return std::move(f_saved);
 446 }
 447
 448 // when (iterator)
 449
 450 template <class InputIterator>
 451 Future<
 452   std::vector<
 453   Try<typename std::iterator_traits<InputIterator>::value_type::value_type>>>
 454 whenAll(InputIterator first, InputIterator last)
 455 {
 456   typedef
 457     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 458
 459   auto n = std::distance(first, last);
 460   if (n == 0) {
 461     return makeFuture(std::vector<Try<T>>());
 462   }
 463
 464   auto ctx = new detail::WhenAllContext<T>();
 465
 466   ctx->results.resize(n);
 467
 468   auto f_saved = ctx->p.getFuture();
 469
 470   for (size_t i = 0; first != last; ++first, ++i) {
 471      assert(i < n);
 472      auto& f = *first;
 473      f.setCallback_([ctx, i, n](Try<T>&& t) {
 474          ctx->results[i] = std::move(t);
 475          if (++ctx->count == n) {
 476            ctx->p.setValue(std::move(ctx->results));
 477            delete ctx;
 478          }
 479        });
 480   }
 481
 482   return std::move(f_saved);
 483 }
 484
 485 template <class InputIterator>
 486 Future<
 487   std::pair<size_t,
 488             Try<
 489               typename
 490               std::iterator_traits<InputIterator>::value_type::value_type> > >
 491 whenAny(InputIterator first, InputIterator last) {
 492   typedef
 493     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 494
 495   auto ctx = new detail::WhenAnyContext<T>(std::distance(first, last));
 496   auto f_saved = ctx->p.getFuture();
 497
 498   for (size_t i = 0; first != last; first++, i++) {
 499     auto& f = *first;
 500     f.setCallback_([i, ctx](Try<T>&& t) {
 501       if (!ctx->done.exchange(true)) {
 502         ctx->p.setValue(std::make_pair(i, std::move(t)));
 503       }
 504       ctx->decref();
 505     });
 506   }
 507
 508   return std::move(f_saved);
 509 }
 510
 511 template <class InputIterator>
 512 Future<std::vector<std::pair<size_t, Try<typename
 513   std::iterator_traits<InputIterator>::value_type::value_type>>>>
 514 whenN(InputIterator first, InputIterator last, size_t n) {
 515   typedef typename
 516     std::iterator_traits<InputIterator>::value_type::value_type T;
 517   typedef std::vector<std::pair<size_t, Try<T>>> V;
 518
 519   struct ctx_t {
 520     V v;
 521     size_t completed;
 522     Promise<V> p;
 523   };
 524   auto ctx = std::make_shared<ctx_t>();
 525   ctx->completed = 0;
 526
 527   // for each completed Future, increase count and add to vector, until we
 528   // have n completed futures at which point we fulfil our Promise with the
 529   // vector
 530   auto it = first;
 531   size_t i = 0;
 532   while (it != last) {
 533     it->then([ctx, n, i](Try<T>&& t) {
 534       auto& v = ctx->v;
 535       auto c = ++ctx->completed;
 536       if (c <= n) {
 537         assert(ctx->v.size() < n);
 538         v.push_back(std::make_pair(i, std::move(t)));
 539         if (c == n) {
 540           ctx->p.fulfilTry(Try<V>(std::move(v)));
 541         }
 542       }
 543     });
 544
 545     it++;
 546     i++;
 547   }
 548
 549   if (i < n) {
 550     ctx->p.setException(std::runtime_error("Not enough futures"));
 551   }
 552
 553   return ctx->p.getFuture();
 554 }
 555
 556 template <typename T>
 557 Future<T>
 558 waitWithSemaphore(Future<T>&& f) {
 559   Baton<> baton;
 560   auto done = f.then([&](Try<T> &&t) {
 561     baton.post();
 562     return std::move(t.value());
 563   });
 564   baton.wait();
 565   while (!done.isReady()) {
 566     // There's a race here between the return here and the actual finishing of
 567     // the future. f is completed, but the setup may not have finished on done
 568     // after the baton has posted.
 569     std::this_thread::yield();
 570   }
 571   return done;
 572 }
 573
 574 template<>
 575 inline Future<void> waitWithSemaphore<void>(Future<void>&& f) {
 576   Baton<> baton;
 577   auto done = f.then([&](Try<void> &&t) {
 578     baton.post();
 579     t.value();
 580   });
 581   baton.wait();
 582   while (!done.isReady()) {
 583     // There's a race here between the return here and the actual finishing of
 584     // the future. f is completed, but the setup may not have finished on done
 585     // after the baton has posted.
 586     std::this_thread::yield();
 587   }
 588   return done;
 589 }
 590
 591 template <typename T, class Duration>
 592 Future<T>
 593 waitWithSemaphore(Future<T>&& f, Duration timeout) {
 594   auto baton = std::make_shared<Baton<>>();
 595   auto done = f.then([baton](Try<T> &&t) {
 596     baton->post();
 597     return std::move(t.value());
 598   });
 599   baton->timed_wait(std::chrono::system_clock::now() + timeout);
 600   return done;
 601 }
 602
 603 template <class Duration>
 604 Future<void>
 605 waitWithSemaphore(Future<void>&& f, Duration timeout) {
 606   auto baton = std::make_shared<Baton<>>();
 607   auto done = f.then([baton](Try<void> &&t) {
 608     baton->post();
 609     t.value();
 610   });
 611   baton->timed_wait(std::chrono::system_clock::now() + timeout);
 612   return done;
 613 }
 614
 615 }}
 616
 617 // I haven't included a Future<T&> specialization because I don't forsee us
 618 // using it, however it is not difficult to add when needed. Refer to
 619 // Future<void> for guidance. std::future and boost::future code would also be
 620 // instructive.