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