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