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 // when (variadic)
 427
 428 template <typename... Fs>
 429 typename detail::VariadicContext<
 430   typename std::decay<Fs>::type::value_type...>::type
 431 whenAll(Fs&&... fs)
 432 {
 433   auto ctx =
 434     new detail::VariadicContext<typename std::decay<Fs>::type::value_type...>();
 435   ctx->total = sizeof...(fs);
 436   auto f_saved = ctx->p.getFuture();
 437   detail::whenAllVariadicHelper(ctx,
 438     std::forward<typename std::decay<Fs>::type>(fs)...);
 439   return std::move(f_saved);
 440 }
 441
 442 // when (iterator)
 443
 444 template <class InputIterator>
 445 Future<
 446   std::vector<
 447   Try<typename std::iterator_traits<InputIterator>::value_type::value_type>>>
 448 whenAll(InputIterator first, InputIterator last)
 449 {
 450   typedef
 451     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 452
 453   auto n = std::distance(first, last);
 454   if (n == 0) {
 455     return makeFuture(std::vector<Try<T>>());
 456   }
 457
 458   auto ctx = new detail::WhenAllContext<T>();
 459
 460   ctx->total = n;
 461   ctx->results.resize(ctx->total);
 462
 463   auto f_saved = ctx->p.getFuture();
 464
 465   for (size_t i = 0; first != last; ++first, ++i) {
 466      auto& f = *first;
 467      f.setCallback_([ctx, i](Try<T>&& t) {
 468          ctx->results[i] = std::move(t);
 469          if (++ctx->count == ctx->total) {
 470            ctx->p.setValue(std::move(ctx->results));
 471            delete ctx;
 472          }
 473        });
 474   }
 475
 476   return std::move(f_saved);
 477 }
 478
 479 template <class InputIterator>
 480 Future<
 481   std::pair<size_t,
 482             Try<
 483               typename
 484               std::iterator_traits<InputIterator>::value_type::value_type> > >
 485 whenAny(InputIterator first, InputIterator last) {
 486   typedef
 487     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 488
 489   auto ctx = new detail::WhenAnyContext<T>(std::distance(first, last));
 490   auto f_saved = ctx->p.getFuture();
 491
 492   for (size_t i = 0; first != last; first++, i++) {
 493     auto& f = *first;
 494     f.setCallback_([i, ctx](Try<T>&& t) {
 495       if (!ctx->done.exchange(true)) {
 496         ctx->p.setValue(std::make_pair(i, std::move(t)));
 497       }
 498       ctx->decref();
 499     });
 500   }
 501
 502   return std::move(f_saved);
 503 }
 504
 505 template <class InputIterator>
 506 Future<std::vector<std::pair<size_t, Try<typename
 507   std::iterator_traits<InputIterator>::value_type::value_type>>>>
 508 whenN(InputIterator first, InputIterator last, size_t n) {
 509   typedef typename
 510     std::iterator_traits<InputIterator>::value_type::value_type T;
 511   typedef std::vector<std::pair<size_t, Try<T>>> V;
 512
 513   struct ctx_t {
 514     V v;
 515     size_t completed;
 516     Promise<V> p;
 517   };
 518   auto ctx = std::make_shared<ctx_t>();
 519   ctx->completed = 0;
 520
 521   // for each completed Future, increase count and add to vector, until we
 522   // have n completed futures at which point we fulfil our Promise with the
 523   // vector
 524   auto it = first;
 525   size_t i = 0;
 526   while (it != last) {
 527     it->then([ctx, n, i](Try<T>&& t) {
 528       auto& v = ctx->v;
 529       auto c = ++ctx->completed;
 530       if (c <= n) {
 531         assert(ctx->v.size() < n);
 532         v.push_back(std::make_pair(i, std::move(t)));
 533         if (c == n) {
 534           ctx->p.fulfilTry(Try<V>(std::move(v)));
 535         }
 536       }
 537     });
 538
 539     it++;
 540     i++;
 541   }
 542
 543   if (i < n) {
 544     ctx->p.setException(std::runtime_error("Not enough futures"));
 545   }
 546
 547   return ctx->p.getFuture();
 548 }
 549
 550 template <typename T>
 551 Future<T>
 552 waitWithSemaphore(Future<T>&& f) {
 553   Baton<> baton;
 554   auto done = f.then([&](Try<T> &&t) {
 555     baton.post();
 556     return std::move(t.value());
 557   });
 558   baton.wait();
 559   while (!done.isReady()) {
 560     // There's a race here between the return here and the actual finishing of
 561     // the future. f is completed, but the setup may not have finished on done
 562     // after the baton has posted.
 563     std::this_thread::yield();
 564   }
 565   return done;
 566 }
 567
 568 template<>
 569 inline Future<void> waitWithSemaphore<void>(Future<void>&& f) {
 570   Baton<> baton;
 571   auto done = f.then([&](Try<void> &&t) {
 572     baton.post();
 573     t.value();
 574   });
 575   baton.wait();
 576   while (!done.isReady()) {
 577     // There's a race here between the return here and the actual finishing of
 578     // the future. f is completed, but the setup may not have finished on done
 579     // after the baton has posted.
 580     std::this_thread::yield();
 581   }
 582   return done;
 583 }
 584
 585 template <typename T, class Duration>
 586 Future<T>
 587 waitWithSemaphore(Future<T>&& f, Duration timeout) {
 588   auto baton = std::make_shared<Baton<>>();
 589   auto done = f.then([baton](Try<T> &&t) {
 590     baton->post();
 591     return std::move(t.value());
 592   });
 593   baton->timed_wait(std::chrono::system_clock::now() + timeout);
 594   return done;
 595 }
 596
 597 template <class Duration>
 598 Future<void>
 599 waitWithSemaphore(Future<void>&& f, Duration timeout) {
 600   auto baton = std::make_shared<Baton<>>();
 601   auto done = f.then([baton](Try<void> &&t) {
 602     baton->post();
 603     t.value();
 604   });
 605   baton->timed_wait(std::chrono::system_clock::now() + timeout);
 606   return done;
 607 }
 608
 609 }}
 610
 611 // I haven't included a Future<T&> specialization because I don't forsee us
 612 // using it, however it is not difficult to add when needed. Refer to
 613 // Future<void> for guidance. std::future and boost::future code would also be
 614 // instructive.