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 (possibly ongoing, asynchronous) operation in the SimGrid kernel
 239  *
 240  *  As the operation may not be completed yet, the result might be an exception.
 241  *
 242  *  Example of the API (`simgrid::kernel::createProcess` does not exist):
 243  *  <pre>
 244  *  // Create a new process using the Worker code, this process returns
 245  *  // a std::string:
 246  *  simgrid::kernel::Future<std::string> future =
 247  *     simgrid::kernel::createProcess("worker42", host, Worker(42));
 248  *  // At this point, we just created the process so the result is not available.
 249  *  // However, we can attach some work do be done with this result:
 250  *  future.then([](simgrid::kernel::Future<std::string> result) {
 251  *    // This code is called when the operation is completed so the result is
 252  *    // available:
 253  *    try {
 254  *      // Try to get value, this might throw an exception if the operation
 255  *      // failed (such as an exception throwed by the worker process):
 256  *      std::string value = result.get();
 257  *      XBT_INFO("Value: %s", value.c_str());
 258  *    }
 259  *    catch(std::exception& e) {
 260  *      // This is an exception from the asynchronous operation:
 261  *      XBT_INFO("Error: %e", e.what());
 262  *    }
 263  *  );
 264  *  </pre>
 265  *
 266  *  This is based on C++1z @ref std::future but with some differences:
 267  *
 268  *  * there is no thread synchronization (atomic, mutex, condition variable,
 269  *    etc.) because everything happens in the SimGrid event loop;
 270  *
 271  *  * it is purely asynchronous, you are expected to use `.then()`;
 272  *
 273  *  * inside the `.then()`, `.get()` can be used;
 274  *
 275  *  * `.get()` can only be used when `.is_ready()` (as everything happens in
 276  *     a single-thread, the future would be guaranted to deadlock if `.get()`
 277  *     is called when the future is not ready);
 278  *
 279  *  * there is no future chaining support for now (`.then().then()`);
 280  *
 281  *  * there is no sharing (`shared_future`) for now.
 282  */
 283 template<class T>
 284 class Future {
 285 public:
 286   Future() {}
 287   Future(std::shared_ptr<FutureState<T>> state): state_(std::move(state)) {}
 288
 289   // Move type:
 290   Future(Future&) = delete;
 291   Future& operator=(Future&) = delete;
 292   Future(Future&& that) : state_(std::move(that.state_)) {}
 293   Future& operator=(Future&& that)
 294   {
 295     state_ = std::move(that.stat_);
 296   }
 297
 298   /** Whether the future is valid:.
 299    *
 300    *  A future which as been used (`.then` of `.get`) becomes invalid.
 301    *
 302    *  We can use `.then` on a valid future.
 303    */
 304   bool valid() const
 305   {
 306     return state_ != nullptr;
 307   }
 308
 309   /** Whether the future is ready
 310    *
 311    *  A future is ready when it has an associated value or exception.
 312    *
 313    *  We can use `.get()` on ready futures.
 314    **/
 315   bool is_ready() const
 316   {
 317     return state_ != nullptr && state_->is_ready();
 318   }
 319
 320   /** Attach a continuation to this future
 321    *
 322    *  The future must be valid in order to make this call.
 323    *  The continuation is executed when the future becomes ready.
 324    *  The future becomes invalid after this call.
 325    *
 326    *  We don't support future chaining for now (`.then().then()`).
 327    *
 328    * @param continuation This function is called with a ready future
 329    *                     the future is ready
 330    * @exception std::future_error no state is associated with the future
 331    */
 332   template<class F>
 333   void then(F continuation)
 334   {
 335     if (state_ == nullptr)
 336       throw std::future_error(std::future_errc::no_state);
 337     std::unique_ptr<FutureContinuation> ptr =
 338       std::unique_ptr<FutureContinuation>(
 339         new FutureContinuationImpl<T,F>(state_, std::move(continuation)));
 340     state_->set_continuation(std::move(ptr));
 341     state_ = nullptr;
 342   }
 343
 344   /** Get the value from the future
 345    *
 346    *  This is expected to be called
 347    *
 348    *  The future must be valid and ready in order to make this call.
 349    *  @ref std::future blocks when the future is not ready but we are
 350    *  completely single-threaded so blocking would be a deadlock.
 351    *  After the call, the future becomes invalid.
 352    *
 353    *  @return                      value of the future
 354    *  @exception any               Exception from the future
 355    *  @exception std::future_error no state is associated with the future
 356    */
 357   T get()
 358   {
 359     if (state_ == nullptr)
 360       throw std::future_error(std::future_errc::no_state);
 361     std::shared_ptr<FutureState<T>> state = std::move(state_);
 362     return state->get();
 363   }
 364
 365 private:
 366   std::shared_ptr<FutureState<T>> state_;
 367 };
 368
 369 /** Producer side of a @simgrid::kernel::Future
 370  *
 371  *  A @ref Promise is connected to some `Future` and can be used to
 372  *  set its result.
 373  *
 374  *  Similar to @ref std::promise
 375  *
 376  *  <code>
 377  *  // Create a promise and a future:
 378  *  auto promise = std::make_shared<simgrid::kernel::Promise<T>>();
 379  *  auto future = promise->get_future();
 380  *
 381  *  SIMIX_timer_set(date, [promise] {
 382  *    try {
 383  *      int value = compute_the_value();
 384  *      if (value < 0)
 385  *        throw std::logic_error("Bad value");
 386  *      // Whenever the operation is completed, we set the value
 387  *      // for the future:
 388  *      promise.set_value(value);
 389  *    }
 390  *    catch (...) {
 391  *      // If an error occured, we can set an exception which
 392  *      // will be throwed buy future.get():
 393  *      promise.set_exception(std::current_exception());
 394  *    }
 395  *  });
 396  *
 397  *  // Return the future to the caller:
 398  *  return future;
 399  *  </code>
 400  **/
 401 template<class T>
 402 class Promise {
 403 public:
 404   Promise() : state_(std::make_shared<FutureState<T>>()) {}
 405   Promise(std::shared_ptr<FutureState<T>> state) : state_(std::move(state)) {}
 406
 407   // Move type
 408   Promise(Promise&) = delete;
 409   Promise& operator=(Promise&) = delete;
 410   Promise(Promise&& that) :
 411     state_(std::move(that.state_)), future_get_(that.future_set)
 412   {
 413     that.future_get_ = false;
 414   }
 415
 416   Promise& operator=(Promise&& that)
 417   {
 418     this->state_ = std::move(that.state_);
 419     this->future_get_ = that.future_get_;
 420     that.future_get_ = false;
 421     return *this;
 422   }
 423   Future<T> get_future()
 424   {
 425     if (state_ == nullptr)
 426       throw std::future_error(std::future_errc::no_state);
 427     if (future_get_)
 428       throw std::future_error(std::future_errc::future_already_retrieved);
 429     future_get_ = true;
 430     return Future<T>(state_);
 431   }
 432   void set_value(T value)
 433   {
 434     if (state_ == nullptr)
 435       throw std::future_error(std::future_errc::no_state);
 436     state_->set_value(std::move(value));
 437   }
 438   void set_exception(std::exception_ptr exception)
 439   {
 440     if (state_ == nullptr)
 441       throw std::future_error(std::future_errc::no_state);
 442     state_->set_exception(std::move(exception));
 443   }
 444   ~Promise()
 445   {
 446     if (state_ && state_->get_status() == FutureStatus::not_ready)
 447       state_->set_exception(std::make_exception_ptr(
 448         std::future_error(std::future_errc::broken_promise)));
 449   }
 450
 451 private:
 452   std::shared_ptr<FutureState<T>> state_;
 453   bool future_get_ = false;
 454 };
 455
 456 template<>
 457 class Promise<void> {
 458 public:
 459   Promise() : state_(std::make_shared<FutureState<void>>()) {}
 460   Promise(std::shared_ptr<FutureState<void>> state) : state_(std::move(state)) {}
 461   ~Promise()
 462   {
 463     if (state_ && state_->get_status() == FutureStatus::not_ready)
 464       state_->set_exception(std::make_exception_ptr(
 465         std::future_error(std::future_errc::broken_promise)));
 466   }
 467
 468   // Move type
 469   Promise(Promise&) = delete;
 470   Promise& operator=(Promise&) = delete;
 471   Promise(Promise&& that) :
 472     state_(std::move(that.state_)), future_get_(that.future_get_)
 473   {
 474     that.future_get_ = false;
 475   }
 476   Promise& operator=(Promise&& that)
 477   {
 478     this->state_ = std::move(that.state_);
 479     this->future_get_ = that.future_get_;
 480     that.future_get_ = false;
 481     return *this;
 482   }
 483
 484   Future<void> get_future()
 485   {
 486     if (state_ == nullptr)
 487       throw std::future_error(std::future_errc::no_state);
 488     if (future_get_)
 489       throw std::future_error(std::future_errc::future_already_retrieved);
 490     future_get_ = true;
 491     return Future<void>(state_);
 492   }
 493   void set_value()
 494   {
 495     if (state_ == nullptr)
 496       throw std::future_error(std::future_errc::no_state);
 497     state_->set_value();
 498   }
 499   void set_exception(std::exception_ptr exception)
 500   {
 501     if (state_ == nullptr)
 502       throw std::future_error(std::future_errc::no_state);
 503     state_->set_exception(std::move(exception));
 504   }
 505
 506 private:
 507   std::shared_ptr<FutureState<void>> state_;
 508   bool future_get_ = false;
 509 };
 510
 511 }
 512 }
 513
 514 #endif