include/simgrid/kernel/future.hpp

   1 /* Copyright (c) 2016. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 #ifndef SIMGRID_KERNEL_FUTURE_HPP
   8 #define SIMGRID_KERNEL_FUTURE_HPP
   9
  10 #include <boost/optional.hpp>
  11
  12 #include <xbt/base.h>
  13
  14 #include <functional>
  15 #include <future>
  16 #include <memory>
  17 #include <utility>
  18 #include <type_traits>
  19
  20 namespace simgrid {
  21 namespace kernel {
  22
  23 // There are the public classes:
  24 template<class T> class Future;
  25 template<class T> class Promise;
  26
  27 // Those are implementation details:
  28 enum class FutureStatus;
  29 template<class T> class FutureState;
  30 class FutureContinuation;
  31 template<class T, class F> class FutureContinuationImpl;
  32
  33 enum class FutureStatus {
  34   not_ready,
  35   ready,
  36   done,
  37 };
  38
  39 /** A continuation attached to a future to be executed when it is ready */
  40 XBT_PUBLIC_CLASS FutureContinuation {
  41 public:
  42   FutureContinuation() {}
  43
  44   // No copy:
  45   FutureContinuation(FutureContinuation&) = delete;
  46   FutureContinuation& operator=(FutureContinuation&) = delete;
  47
  48   virtual ~FutureContinuation() {}
  49   virtual void operator()() = 0;
  50 };
  51
  52 /** Default implementation of `FutureContinuation`
  53  *
  54  *  @param T   value type of the future
  55  *  @param F   type of the wrapped code/callback/continuation
  56  */
  57 template<class T, class F>
  58 class FutureContinuationImpl : public FutureContinuation {
  59 public:
  60   FutureContinuationImpl(std::shared_ptr<FutureState<T>> ptr, F callback)
  61     : ptr_(std::move(ptr)), callback_(std::move(callback)) {}
  62   ~FutureContinuationImpl() override {}
  63   void operator()() override
  64   {
  65     try {
  66       callback_(Future<T>(ptr_));
  67     }
  68     // Those exceptions are lost.
  69     // If we want to implement callback chaining, we'll have to catch them and
  70     // foward them to the next future.
  71     catch (...) {
  72       // We could log this.
  73     }
  74   }
  75 private:
  76   std::shared_ptr<FutureState<T>> ptr_;
  77   F callback_;
  78 };
  79
  80 /** Bases stuff for all @ref simgrid::kernel::FutureState<T> */
  81 class FutureStateBase {
  82 public:
  83   // No copy/move:
  84   FutureStateBase(FutureStateBase const&) = delete;
  85   FutureStateBase& operator=(FutureStateBase const&) = delete;
  86
  87   void set_exception(std::exception_ptr exception)
  88   {
  89     xbt_assert(exception_ == nullptr);
  90     if (status_ != FutureStatus::not_ready)
  91       throw std::future_error(std::future_errc::promise_already_satisfied);
  92     exception_ = std::move(exception);
  93     this->set_ready();
  94   }
  95
  96   void set_continuation(std::unique_ptr<FutureContinuation> continuation)
  97   {
  98     xbt_assert(!continuation_);
  99     switch (status_) {
 100     case FutureStatus::done:
 101       // This is not supposed to happen if continuation is set
 102       // via the Promise:
 103       xbt_die("Set continuation on finished future");
 104       break;
 105     case FutureStatus::ready:
 106       // The future is ready, execute the continuation directly.
 107       // We might execute it from the event loop instead:
 108       (*continuation)();
 109       break;
 110     case FutureStatus::not_ready:
 111       // The future is not ready so we mast keep the continuation for
 112       // executing it later:
 113       continuation_ = std::move(continuation);
 114       break;
 115     }
 116   }
 117
 118   FutureStatus get_status() const
 119   {
 120     return status_;
 121   }
 122
 123   bool is_ready() const
 124   {
 125     return status_ == FutureStatus::ready;
 126   }
 127
 128 protected:
 129   FutureStateBase() {}
 130   ~FutureStateBase() {};
 131
 132   /** Set the future as ready and trigger the continuation */
 133   void set_ready()
 134   {
 135     status_ = FutureStatus::ready;
 136     if (continuation_) {
 137       // We unregister the continuation before executing it.
 138       // We need to do this becase the current implementation of the
 139       // continuation has a shared_ptr to the FutureState.
 140       auto continuation = std::move(continuation_);
 141       (*continuation)();
 142     }
 143   }
 144
 145   /** Set the future as done and raise an exception if any
 146    *
 147    *  This does half the job of `.get()`.
 148    **/
 149   void resolve()
 150   {
 151     if (status_ != FutureStatus::ready)
 152       xbt_die("Deadlock: this future is not ready");
 153     status_ = FutureStatus::done;
 154     if (exception_) {
 155       std::exception_ptr exception = std::move(exception_);
 156       std::rethrow_exception(std::move(exception));
 157     }
 158   }
 159
 160 private:
 161   FutureStatus status_ = FutureStatus::not_ready;
 162   std::exception_ptr exception_;
 163   std::unique_ptr<FutureContinuation> continuation_;
 164 };
 165
 166 /** Shared state for future and promises
 167  *
 168  *  You are not expected to use them directly but to create them
 169  *  implicitely through a @ref simgrid::kernel::Promise.
 170  *  Alternatively kernel operations could inherit or contain FutureState
 171  *  if they are managed with @ref std::shared_ptr.
 172  **/
 173 template<class T>
 174 class FutureState : public FutureStateBase {
 175 public:
 176
 177   void set_value(T value)
 178   {
 179     if (this->get_status() != FutureStatus::not_ready)
 180       throw std::future_error(std::future_errc::promise_already_satisfied);
 181     value_ = std::move(value);
 182     this->set_ready();
 183   }
 184
 185   T get()
 186   {
 187     this->resolve();
 188     xbt_assert(this->value_);
 189     auto result = std::move(this->value_.get());
 190     this->value_ = boost::optional<T>();
 191     return std::move(result);
 192   }
 193
 194 private:
 195   boost::optional<T> value_;
 196 };
 197
 198 template<class T>
 199 class FutureState<T&> : public FutureStateBase {
 200 public:
 201   void set_value(T& value)
 202   {
 203     if (this->get_status() != FutureStatus::not_ready)
 204       throw std::future_error(std::future_errc::promise_already_satisfied);
 205     value_ = &value;
 206     this->set_ready();
 207   }
 208
 209   T& get()
 210   {
 211     this->resolve();
 212     xbt_assert(this->value_);
 213     T* result = value_;
 214     value_ = nullptr;
 215     return *value_;
 216   }
 217
 218 private:
 219   T* value_ = nullptr;
 220 };
 221
 222 template<>
 223 class FutureState<void> : public FutureStateBase {
 224 public:
 225   void set_value()
 226   {
 227     if (this->get_status() != FutureStatus::not_ready)
 228       throw std::future_error(std::future_errc::promise_already_satisfied);
 229     this->set_ready();
 230   }
 231
 232   void get()
 233   {
 234     this->resolve();
 235   }
 236 };
 237
 238 /** Result of some (probably) asynchronous operation in the SimGrid kernel
 239  *
 240  * @ref simgrid::simix::Future and @ref simgrid::simix::Future provide an
 241  * abstration for asynchronous stuff happening in the SimGrid kernel. They
 242  * are based on C++1z futures.
 243  *
 244  * The future represents a value which will be available at some point when this
 245  * asynchronous operaiont is finished. Alternatively, if this operations fails,
 246  * the result of the operation might be an exception.
 247  *
 248  *  As the operation is possibly no terminated yet, we cannot get the result
 249  *  yet. Moreover, as we cannot block in the SimGrid kernel we cannot wait for
 250  *  it. However, we can attach some code/callback/continuation which will be
 251  *  executed when the operation terminates.
 252  *
 253  *  Example of the API (`simgrid::kernel::createProcess` does not exist):
 254  *  <pre>
 255  *  // Create a new process using the Worker code, this process returns
 256  *  // a std::string:
 257  *  simgrid::kernel::Future<std::string> future =
 258  *     simgrid::kernel::createProcess("worker42", host, Worker(42));
 259  *  // At this point, we just created the process so the result is not available.
 260  *  // However, we can attach some work do be done with this result:
 261  *  future.then([](simgrid::kernel::Future<std::string> result) {
 262  *    // This code is called when the operation is completed so the result is
 263  *    // available:
 264  *    try {
 265  *      // Try to get value, this might throw an exception if the operation
 266  *      // failed (such as an exception throwed by the worker process):
 267  *      std::string value = result.get();
 268  *      XBT_INFO("Value: %s", value.c_str());
 269  *    }
 270  *    catch(std::exception& e) {
 271  *      // This is an exception from the asynchronous operation:
 272  *      XBT_INFO("Error: %e", e.what());
 273  *    }
 274  *  );
 275  *  </pre>
 276  *
 277  *  This is based on C++1z @ref std::future but with some differences:
 278  *
 279  *  * there is no thread synchronization (atomic, mutex, condition variable,
 280  *    etc.) because everything happens in the SimGrid event loop;
 281  *
 282  *  * it is purely asynchronous, you are expected to use `.then()`;
 283  *
 284  *  * inside the `.then()`, `.get()` can be used;
 285  *
 286  *  * `.get()` can only be used when `.is_ready()` (as everything happens in
 287  *     a single-thread, the future would be guaranted to deadlock if `.get()`
 288  *     is called when the future is not ready);
 289  *
 290  *  * there is no future chaining support for now (`.then().then()`);
 291  *
 292  *  * there is no sharing (`shared_future`) for now.
 293  */
 294 template<class T>
 295 class Future {
 296 public:
 297   Future() {}
 298   Future(std::shared_ptr<FutureState<T>> state): state_(std::move(state)) {}
 299
 300   // Move type:
 301   Future(Future&) = delete;
 302   Future& operator=(Future&) = delete;
 303   Future(Future&& that) : state_(std::move(that.state_)) {}
 304   Future& operator=(Future&& that)
 305   {
 306     state_ = std::move(that.stat_);
 307   }
 308
 309   /** Whether the future is valid:.
 310    *
 311    *  A future which as been used (`.then` of `.get`) becomes invalid.
 312    *
 313    *  We can use `.then` on a valid future.
 314    */
 315   bool valid() const
 316   {
 317     return state_ != nullptr;
 318   }
 319
 320   /** Whether the future is ready
 321    *
 322    *  A future is ready when it has an associated value or exception.
 323    *
 324    *  We can use `.get()` on ready futures.
 325    **/
 326   bool is_ready() const
 327   {
 328     return state_ != nullptr && state_->is_ready();
 329   }
 330
 331   /** Attach a continuation to this future
 332    *
 333    *  The future must be valid in order to make this call.
 334    *  The continuation is executed when the future becomes ready.
 335    *  The future becomes invalid after this call.
 336    *
 337    *  We don't support future chaining for now (`.then().then()`).
 338    *
 339    * @param continuation This function is called with a ready future
 340    *                     the future is ready
 341    * @exception std::future_error no state is associated with the future
 342    */
 343   template<class F>
 344   void then(F continuation)
 345   {
 346     if (state_ == nullptr)
 347       throw std::future_error(std::future_errc::no_state);
 348     std::unique_ptr<FutureContinuation> ptr =
 349       std::unique_ptr<FutureContinuation>(
 350         new FutureContinuationImpl<T,F>(state_, std::move(continuation)));
 351     state_->set_continuation(std::move(ptr));
 352     state_ = nullptr;
 353   }
 354
 355   /** Get the value from the future
 356    *
 357    *  This is expected to be called
 358    *
 359    *  The future must be valid and ready in order to make this call.
 360    *  @ref std::future blocks when the future is not ready but we are
 361    *  completely single-threaded so blocking would be a deadlock.
 362    *  After the call, the future becomes invalid.
 363    *
 364    *  @return                      value of the future
 365    *  @exception any               Exception from the future
 366    *  @exception std::future_error no state is associated with the future
 367    */
 368   T get()
 369   {
 370     if (state_ == nullptr)
 371       throw std::future_error(std::future_errc::no_state);
 372     std::shared_ptr<FutureState<T>> state = std::move(state_);
 373     return state->get();
 374   }
 375
 376 private:
 377   std::shared_ptr<FutureState<T>> state_;
 378 };
 379
 380 /** Producer side of a @simgrid::kernel::Future
 381  *
 382  *  A @ref Promise is connected to some `Future` and can be used to
 383  *  set its result.
 384  *
 385  *  Similar to @ref std::promise
 386  *
 387  *  <code>
 388  *  // Create a promise and a future:
 389  *  auto promise = std::make_shared<simgrid::kernel::Promise<T>>();
 390  *  auto future = promise->get_future();
 391  *
 392  *  SIMIX_timer_set(date, [promise] {
 393  *    try {
 394  *      int value = compute_the_value();
 395  *      if (value < 0)
 396  *        throw std::logic_error("Bad value");
 397  *      // Whenever the operation is completed, we set the value
 398  *      // for the future:
 399  *      promise.set_value(value);
 400  *    }
 401  *    catch (...) {
 402  *      // If an error occured, we can set an exception which
 403  *      // will be throwed buy future.get():
 404  *      promise.set_exception(std::current_exception());
 405  *    }
 406  *  });
 407  *
 408  *  // Return the future to the caller:
 409  *  return future;
 410  *  </code>
 411  **/
 412 template<class T>
 413 class Promise {
 414 public:
 415   Promise() : state_(std::make_shared<FutureState<T>>()) {}
 416   Promise(std::shared_ptr<FutureState<T>> state) : state_(std::move(state)) {}
 417
 418   // Move type
 419   Promise(Promise&) = delete;
 420   Promise& operator=(Promise&) = delete;
 421   Promise(Promise&& that) :
 422     state_(std::move(that.state_)), future_get_(that.future_set)
 423   {
 424     that.future_get_ = false;
 425   }
 426
 427   Promise& operator=(Promise&& that)
 428   {
 429     this->state_ = std::move(that.state_);
 430     this->future_get_ = that.future_get_;
 431     that.future_get_ = false;
 432     return *this;
 433   }
 434   Future<T> get_future()
 435   {
 436     if (state_ == nullptr)
 437       throw std::future_error(std::future_errc::no_state);
 438     if (future_get_)
 439       throw std::future_error(std::future_errc::future_already_retrieved);
 440     future_get_ = true;
 441     return Future<T>(state_);
 442   }
 443   void set_value(T value)
 444   {
 445     if (state_ == nullptr)
 446       throw std::future_error(std::future_errc::no_state);
 447     state_->set_value(std::move(value));
 448   }
 449   void set_exception(std::exception_ptr exception)
 450   {
 451     if (state_ == nullptr)
 452       throw std::future_error(std::future_errc::no_state);
 453     state_->set_exception(std::move(exception));
 454   }
 455   ~Promise()
 456   {
 457     if (state_ && state_->get_status() == FutureStatus::not_ready)
 458       state_->set_exception(std::make_exception_ptr(
 459         std::future_error(std::future_errc::broken_promise)));
 460   }
 461
 462 private:
 463   std::shared_ptr<FutureState<T>> state_;
 464   bool future_get_ = false;
 465 };
 466
 467 template<>
 468 class Promise<void> {
 469 public:
 470   Promise() : state_(std::make_shared<FutureState<void>>()) {}
 471   Promise(std::shared_ptr<FutureState<void>> state) : state_(std::move(state)) {}
 472   ~Promise()
 473   {
 474     if (state_ && state_->get_status() == FutureStatus::not_ready)
 475       state_->set_exception(std::make_exception_ptr(
 476         std::future_error(std::future_errc::broken_promise)));
 477   }
 478
 479   // Move type
 480   Promise(Promise&) = delete;
 481   Promise& operator=(Promise&) = delete;
 482   Promise(Promise&& that) :
 483     state_(std::move(that.state_)), future_get_(that.future_get_)
 484   {
 485     that.future_get_ = false;
 486   }
 487   Promise& operator=(Promise&& that)
 488   {
 489     this->state_ = std::move(that.state_);
 490     this->future_get_ = that.future_get_;
 491     that.future_get_ = false;
 492     return *this;
 493   }
 494
 495   Future<void> get_future()
 496   {
 497     if (state_ == nullptr)
 498       throw std::future_error(std::future_errc::no_state);
 499     if (future_get_)
 500       throw std::future_error(std::future_errc::future_already_retrieved);
 501     future_get_ = true;
 502     return Future<void>(state_);
 503   }
 504   void set_value()
 505   {
 506     if (state_ == nullptr)
 507       throw std::future_error(std::future_errc::no_state);
 508     state_->set_value();
 509   }
 510   void set_exception(std::exception_ptr exception)
 511   {
 512     if (state_ == nullptr)
 513       throw std::future_error(std::future_errc::no_state);
 514     state_->set_exception(std::move(exception));
 515   }
 516
 517 private:
 518   std::shared_ptr<FutureState<void>> state_;
 519   bool future_get_ = false;
 520 };
 521
 522 }
 523 }
 524
 525 #endif