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) noexcept : 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     setCallback_([](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>::setCallback_(F&& func) {
  61   throwIfInvalid();
  62   obj_->setCallback_(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   setCallback_(
 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   setCallback_(
 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.setCallback_([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   setCallback_([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 bool Future<T>::isReady() const {
 204   throwIfInvalid();
 205   return obj_->ready();
 206 }
 207
 208 // makeFuture
 209
 210 template <class T>
 211 Future<typename std::decay<T>::type> makeFuture(T&& t) {
 212   Promise<typename std::decay<T>::type> p;
 213   auto f = p.getFuture();
 214   p.setValue(std::forward<T>(t));
 215   return std::move(f);
 216 }
 217
 218 inline // for multiple translation units
 219 Future<void> makeFuture() {
 220   Promise<void> p;
 221   auto f = p.getFuture();
 222   p.setValue();
 223   return std::move(f);
 224 }
 225
 226 template <class F>
 227 auto makeFutureTry(
 228     F&& func,
 229     typename std::enable_if<!std::is_reference<F>::value, bool>::type sdf)
 230     -> Future<decltype(func())> {
 231   Promise<decltype(func())> p;
 232   auto f = p.getFuture();
 233   p.fulfil(
 234     [&func]() {
 235       return (func)();
 236     });
 237   return std::move(f);
 238 }
 239
 240 template <class F>
 241 auto makeFutureTry(F const& func) -> Future<decltype(func())> {
 242   F copy = func;
 243   return makeFutureTry(std::move(copy));
 244 }
 245
 246 template <class T>
 247 Future<T> makeFuture(std::exception_ptr const& e) {
 248   Promise<T> p;
 249   auto f = p.getFuture();
 250   p.setException(e);
 251   return std::move(f);
 252 }
 253
 254 template <class T, class E>
 255 typename std::enable_if<std::is_base_of<std::exception, E>::value, Future<T>>::type
 256 makeFuture(E const& e) {
 257   Promise<T> p;
 258   auto f = p.getFuture();
 259   p.fulfil([&]() -> T { throw e; });
 260   return std::move(f);
 261 }
 262
 263 template <class T>
 264 Future<T> makeFuture(Try<T>&& t) {
 265   try {
 266     return makeFuture<T>(std::move(t.value()));
 267   } catch (...) {
 268     return makeFuture<T>(std::current_exception());
 269   }
 270 }
 271
 272 template <>
 273 inline Future<void> makeFuture(Try<void>&& t) {
 274   try {
 275     t.throwIfFailed();
 276     return makeFuture();
 277   } catch (...) {
 278     return makeFuture<void>(std::current_exception());
 279   }
 280 }
 281
 282 // when (variadic)
 283
 284 template <typename... Fs>
 285 typename detail::VariadicContext<
 286   typename std::decay<Fs>::type::value_type...>::type
 287 whenAll(Fs&&... fs)
 288 {
 289   auto ctx =
 290     new detail::VariadicContext<typename std::decay<Fs>::type::value_type...>();
 291   ctx->total = sizeof...(fs);
 292   auto f_saved = ctx->p.getFuture();
 293   detail::whenAllVariadicHelper(ctx,
 294     std::forward<typename std::decay<Fs>::type>(fs)...);
 295   return std::move(f_saved);
 296 }
 297
 298 // when (iterator)
 299
 300 template <class InputIterator>
 301 Future<
 302   std::vector<
 303   Try<typename std::iterator_traits<InputIterator>::value_type::value_type>>>
 304 whenAll(InputIterator first, InputIterator last)
 305 {
 306   typedef
 307     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 308
 309   auto n = std::distance(first, last);
 310   if (n == 0) {
 311     return makeFuture(std::vector<Try<T>>());
 312   }
 313
 314   auto ctx = new detail::WhenAllContext<T>();
 315
 316   ctx->total = n;
 317   ctx->results.resize(ctx->total);
 318
 319   auto f_saved = ctx->p.getFuture();
 320
 321   for (size_t i = 0; first != last; ++first, ++i) {
 322      auto& f = *first;
 323      f.setCallback_([ctx, i](Try<T>&& t) {
 324          ctx->results[i] = std::move(t);
 325          if (++ctx->count == ctx->total) {
 326            ctx->p.setValue(std::move(ctx->results));
 327            delete ctx;
 328          }
 329        });
 330   }
 331
 332   return std::move(f_saved);
 333 }
 334
 335 template <class InputIterator>
 336 Future<
 337   std::pair<size_t,
 338             Try<
 339               typename
 340               std::iterator_traits<InputIterator>::value_type::value_type> > >
 341 whenAny(InputIterator first, InputIterator last) {
 342   typedef
 343     typename std::iterator_traits<InputIterator>::value_type::value_type T;
 344
 345   auto ctx = new detail::WhenAnyContext<T>(std::distance(first, last));
 346   auto f_saved = ctx->p.getFuture();
 347
 348   for (size_t i = 0; first != last; first++, i++) {
 349     auto& f = *first;
 350     f.setCallback_([i, ctx](Try<T>&& t) {
 351       if (!ctx->done.exchange(true)) {
 352         ctx->p.setValue(std::make_pair(i, std::move(t)));
 353       }
 354       ctx->decref();
 355     });
 356   }
 357
 358   return std::move(f_saved);
 359 }
 360
 361 template <class InputIterator>
 362 Future<std::vector<std::pair<size_t, Try<typename
 363   std::iterator_traits<InputIterator>::value_type::value_type>>>>
 364 whenN(InputIterator first, InputIterator last, size_t n) {
 365   typedef typename
 366     std::iterator_traits<InputIterator>::value_type::value_type T;
 367   typedef std::vector<std::pair<size_t, Try<T>>> V;
 368
 369   struct ctx_t {
 370     V v;
 371     size_t completed;
 372     Promise<V> p;
 373   };
 374   auto ctx = std::make_shared<ctx_t>();
 375   ctx->completed = 0;
 376
 377   // for each completed Future, increase count and add to vector, until we
 378   // have n completed futures at which point we fulfil our Promise with the
 379   // vector
 380   auto it = first;
 381   size_t i = 0;
 382   while (it != last) {
 383     it->then([ctx, n, i](Try<T>&& t) {
 384       auto& v = ctx->v;
 385       auto c = ++ctx->completed;
 386       if (c <= n) {
 387         assert(ctx->v.size() < n);
 388         v.push_back(std::make_pair(i, std::move(t)));
 389         if (c == n) {
 390           ctx->p.fulfilTry(Try<V>(std::move(v)));
 391         }
 392       }
 393     });
 394
 395     it++;
 396     i++;
 397   }
 398
 399   if (i < n) {
 400     ctx->p.setException(std::runtime_error("Not enough futures"));
 401   }
 402
 403   return ctx->p.getFuture();
 404 }
 405
 406 template <typename T>
 407 Future<T>
 408 waitWithSemaphore(Future<T>&& f) {
 409   LifoSem sem;
 410   auto done = f.then([&](Try<T> &&t) {
 411     sem.post();
 412     return std::move(t.value());
 413   });
 414   sem.wait();
 415   return done;
 416 }
 417
 418 template<>
 419 inline Future<void> waitWithSemaphore<void>(Future<void>&& f) {
 420   LifoSem sem;
 421   auto done = f.then([&](Try<void> &&t) {
 422     sem.post();
 423     t.value();
 424   });
 425   sem.wait();
 426   return done;
 427 }
 428
 429 template <typename T, class Duration>
 430 Future<T>
 431 waitWithSemaphore(Future<T>&& f, Duration timeout) {
 432   auto sem = std::make_shared<LifoSem>();
 433   auto done = f.then([sem](Try<T> &&t) {
 434     sem->post();
 435     return std::move(t.value());
 436   });
 437   std::thread t([sem, timeout](){
 438     std::this_thread::sleep_for(timeout);
 439     sem->shutdown();
 440     });
 441   t.detach();
 442   try {
 443     sem->wait();
 444   } catch (ShutdownSemError & ign) { }
 445   return done;
 446 }
 447
 448 template <class Duration>
 449 Future<void>
 450 waitWithSemaphore(Future<void>&& f, Duration timeout) {
 451   auto sem = std::make_shared<LifoSem>();
 452   auto done = f.then([sem](Try<void> &&t) {
 453     sem->post();
 454     t.value();
 455   });
 456   std::thread t([sem, timeout](){
 457     std::this_thread::sleep_for(timeout);
 458     sem->shutdown();
 459     });
 460   t.detach();
 461   try {
 462     sem->wait();
 463   } catch (ShutdownSemError & ign) { }
 464   return done;
 465 }
 466
 467 }}
 468
 469 // I haven't included a Future<T&> specialization because I don't forsee us
 470 // using it, however it is not difficult to add when needed. Refer to
 471 // Future<void> for guidance. std::future and boost::future code would also be
 472 // instructive.