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