X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/b9a15f078a62e397f321dfb570254652785cc377..c3afdfa7a603897d1fbca56e7d1705983038405e:/include/simgrid/kernel/future.hpp diff --git a/include/simgrid/kernel/future.hpp b/include/simgrid/kernel/future.hpp index c58f46b164..cedfd39c96 100644 --- a/include/simgrid/kernel/future.hpp +++ b/include/simgrid/kernel/future.hpp @@ -7,9 +7,13 @@ #ifndef SIMGRID_KERNEL_FUTURE_HPP #define SIMGRID_KERNEL_FUTURE_HPP +#include +#include + #include #include +#include #include #include @@ -27,8 +31,6 @@ template class Promise; // Those are implementation details: enum class FutureStatus; template class FutureState; -class FutureContinuation; -template class FutureContinuationImpl; enum class FutureStatus { not_ready, @@ -36,46 +38,10 @@ enum class FutureStatus { done, }; -/** A continuation attached to a future to be executed when it is ready */ -XBT_PUBLIC_CLASS FutureContinuation { -public: - FutureContinuation() {} - - // No copy: - FutureContinuation(FutureContinuation&) = delete; - FutureContinuation& operator=(FutureContinuation&) = delete; - - virtual ~FutureContinuation() {} - virtual void operator()() = 0; -}; - -/** Default implementation of `FutureContinuation` - * - * @param T value type of the future - * @param F type of the wrapped code/callback/continuation - */ -template -class FutureContinuationImpl : public FutureContinuation { -public: - FutureContinuationImpl(std::shared_ptr> ptr, F callback) - : ptr_(std::move(ptr)), callback_(std::move(callback)) {} - ~FutureContinuationImpl() override {} - void operator()() override - { - try { - callback_(Future(ptr_)); - } - // Those exceptions are lost. - // If we want to implement callback chaining, we'll have to catch them and - // foward them to the next future. - catch (...) { - // We could log this. - } - } -private: - std::shared_ptr> ptr_; - F callback_; -}; +template +struct is_future : public std::integral_constant {}; +template +struct is_future> : public std::integral_constant {}; /** Bases stuff for all @ref simgrid::kernel::FutureState */ class FutureStateBase { @@ -93,7 +59,7 @@ public: this->set_ready(); } - void set_continuation(std::unique_ptr continuation) + void set_continuation(simgrid::xbt::Task continuation) { xbt_assert(!continuation_); switch (status_) { @@ -105,13 +71,15 @@ public: case FutureStatus::ready: // The future is ready, execute the continuation directly. // We might execute it from the event loop instead: - (*continuation)(); + continuation(); break; case FutureStatus::not_ready: // The future is not ready so we mast keep the continuation for // executing it later: continuation_ = std::move(continuation); break; + default: + DIE_IMPOSSIBLE; } } @@ -126,8 +94,8 @@ public: } protected: - FutureStateBase() {} - ~FutureStateBase() {}; + FutureStateBase() = default; + ~FutureStateBase() = default; /** Set the future as ready and trigger the continuation */ void set_ready() @@ -138,7 +106,7 @@ protected: // We need to do this becase the current implementation of the // continuation has a shared_ptr to the FutureState. auto continuation = std::move(continuation_); - (*continuation)(); + continuation(); } } @@ -160,7 +128,7 @@ protected: private: FutureStatus status_ = FutureStatus::not_ready; std::exception_ptr exception_; - std::unique_ptr continuation_; + simgrid::xbt::Task continuation_; }; /** Shared state for future and promises @@ -235,6 +203,24 @@ public: } }; +template +void bindPromise(Promise promise, Future future) +{ + struct PromiseBinder { + public: + PromiseBinder(Promise promise) : promise_(std::move(promise)) {} + void operator()(Future future) + { + simgrid::xbt::setPromise(promise_, future); + } + private: + Promise promise_; + }; + future.then_(PromiseBinder(std::move(promise))); +} + +template Future unwrapFuture(Future> future); + /** Result of some (probably) asynchronous operation in the SimGrid kernel * * @ref simgrid::simix::Future and @ref simgrid::simix::Future provide an @@ -294,7 +280,7 @@ public: template class Future { public: - Future() {} + Future() = default; Future(std::shared_ptr> state): state_(std::move(state)) {} // Move type: @@ -303,7 +289,8 @@ public: Future(Future&& that) : state_(std::move(that.state_)) {} Future& operator=(Future&& that) { - state_ = std::move(that.stat_); + state_ = std::move(that.state_); + return *this; } /** Whether the future is valid:. @@ -328,28 +315,78 @@ public: return state_ != nullptr && state_->is_ready(); } + /** Attach a continuation to this future + * + * This is like .then() but avoid the creation of a new future. + */ + template + void then_(F continuation) + { + if (state_ == nullptr) + throw std::future_error(std::future_errc::no_state); + // Give shared-ownership to the continuation: + auto state = std::move(state_); + state->set_continuation(simgrid::xbt::makeTask( + std::move(continuation), state)); + } + + /** Attach a continuation to this future + * + * This version never does future unwrapping. + */ + template + auto thenNoUnwrap(F continuation) + -> Future + { + typedef decltype(continuation(std::move(*this))) R; + if (state_ == nullptr) + throw std::future_error(std::future_errc::no_state); + auto state = std::move(state_); + // Create a new future... + Promise promise; + Future future = promise.get_future(); + // ...and when the current future is ready... + state->set_continuation(simgrid::xbt::makeTask( + [](Promise promise, std::shared_ptr> state, F continuation) { + // ...set the new future value by running the continuation. + Future future(std::move(state)); + simgrid::xbt::fulfillPromise(promise,[&]{ + return continuation(std::move(future)); + }); + }, + std::move(promise), state, std::move(continuation))); + return std::move(future); + } + /** Attach a continuation to this future * * The future must be valid in order to make this call. * The continuation is executed when the future becomes ready. * The future becomes invalid after this call. * - * We don't support future chaining for now (`.then().then()`). - * * @param continuation This function is called with a ready future * the future is ready * @exception std::future_error no state is associated with the future */ template - void then(F continuation) + auto then(F continuation) + -> typename std::enable_if< + !is_future::value, + Future + >::type { - if (state_ == nullptr) - throw std::future_error(std::future_errc::no_state); - std::unique_ptr ptr = - std::unique_ptr( - new FutureContinuationImpl(state_, std::move(continuation))); - state_->set_continuation(std::move(ptr)); - state_ = nullptr; + return this->thenNoUnwrap(std::move(continuation)); + } + + /** Attach a continuation to this future (future chaining) */ + template + auto then(F continuation) + -> typename std::enable_if< + is_future::value, + decltype(continuation(std::move(*this))) + >::type + { + return unwrapFuture(this->thenNoUnwap(std::move(continuation))); } /** Get the value from the future @@ -377,6 +414,15 @@ private: std::shared_ptr> state_; }; +template +Future unwrapFuture(Future> future) +{ + Promise promise; + Future result = promise.get_future(); + bindPromise(std::move(promise), std::move(future)); + return std::move(result); +} + /** Producer side of a @simgrid::kernel::Future * * A @ref Promise is connected to some `Future` and can be used to @@ -416,10 +462,10 @@ public: Promise(std::shared_ptr> state) : state_(std::move(state)) {} // Move type - Promise(Promise&) = delete; - Promise& operator=(Promise&) = delete; + Promise(Promise const&) = delete; + Promise& operator=(Promise const&) = delete; Promise(Promise&& that) : - state_(std::move(that.state_)), future_get_(that.future_set) + state_(std::move(that.state_)), future_get_(that.future_get_) { that.future_get_ = false; } @@ -477,8 +523,8 @@ public: } // Move type - Promise(Promise&) = delete; - Promise& operator=(Promise&) = delete; + Promise(Promise const&) = delete; + Promise& operator=(Promise const&) = delete; Promise(Promise&& that) : state_(std::move(that.state_)), future_get_(that.future_get_) {