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