diff --git a/misc/experimental/promise/api/celix/Deferred.h b/misc/experimental/promise/api/celix/Deferred.h index fcbcd2933..85ef15cc0 100644 --- a/misc/experimental/promise/api/celix/Deferred.h +++ b/misc/experimental/promise/api/celix/Deferred.h @@ -20,6 +20,7 @@ #pragma once #include +#include #include "celix/impl/SharedPromiseState.h" #include "celix/Promise.h" @@ -113,9 +114,6 @@ namespace celix { void resolve(const T& value); //NOTE not part of the spec.. update to resolveWith with a return celix::Promise ?? - template - void resolveWith(celix::Promise with); - /** * Resolve the Promise associated with this Deferred with the specified Promise. *

@@ -136,12 +134,81 @@ namespace celix { * the specified Promise. The returned Promise will be resolved with a failure of IllegalStateException if the * associated Promise was already resolved when the specified Promise was resolved. */ - //TODO support void promises - //Promise resolveWith(Promise with); + template + void resolveWith(celix::Promise with); private: std::shared_ptr> state; }; + + template<> + class Deferred { + public: + using type = void; + + Deferred(); + + explicit Deferred(std::shared_ptr> state); + + //TODO deferred ctor with factory + + /** + * Fail the Promise associated with this Deferred. + *

+ * After the associated Promise is resolved with the specified failure, all registered callbacks are called and any + * chained Promises are resolved. + *

+ * Resolving the associated Promise happens-before any registered callback is called. That is, in a registered + * callback, Promise.isDone() must return true and Promise.getValue() and Promise.getFailure() must not block. + * + * @param failure The failure in the form of an exception pointer. + * @throws PromiseInvocationException If the associated Promise was already resolved. + */ + void fail(std::exception_ptr failure); + + /** + * Fail the Promise associated with this Deferred. + *

+ * After the associated Promise is resolved with the specified failure, all registered callbacks are called and any + * chained Promises are resolved. + *

+ * Resolving the associated Promise happens-before any registered callback is called. That is, in a registered + * callback, Promise.isDone() must return true and Promise.getValue() and Promise.getFailure() must not block. + * + * @param failure The failure in the form of an const std::exception reference. + * @throws PromiseInvocationException If the associated Promise was already resolved. + */ + void fail(const std::exception& failure); + + /** + * Returns the Promise associated with this Deferred. + *

+ * All Promise objects created by the associated Promise will use the + * executors of the associated Promise. + * + * @return The Promise associated with this Deferred. + */ + Promise getPromise(); + + /** + * Successfully resolve the Promise associated with this Deferred. + *

+ * After the associated Promise is resolved with the specified value, all registered callbacks are called and any + * chained Promises are resolved. + *

+ * Resolving the associated Promise happens-before any registered callback is called. That is, in a registered + * callback, Promise.isDone() must return true and Promise.getValue() and Promise.getFailure() must not block. + * + * @param value The value of the resolved Promise. + * @throws PromiseInvocationException If the associated Promise was already resolved. + */ + void resolve(); + + void resolveWith(celix::Promise with); + + private: + std::shared_ptr> state; + }; } @@ -153,17 +220,29 @@ namespace celix { template inline celix::Deferred::Deferred() : state{std::make_shared>()} {} +inline celix::Deferred::Deferred() : state{std::make_shared>()} {} + template inline celix::Deferred::Deferred(std::shared_ptr> _state) : state{std::move(_state)} {} +inline celix::Deferred::Deferred(std::shared_ptr> _state) : state{std::move(_state)} {} + template -inline void celix::Deferred::fail(std::exception_ptr p) { - state->fail(p); +inline void celix::Deferred::fail(std::exception_ptr failure) { + state->fail(std::move(failure)); +} + +inline void celix::Deferred::fail(std::exception_ptr failure) { + state->fail(std::move(failure)); } template -inline void celix::Deferred::fail(const std::exception& e) { - state->fail(e); +inline void celix::Deferred::fail(const std::exception& failure) { + state->fail(failure); +} + +inline void celix::Deferred::fail(const std::exception& failure) { + state->fail(failure); } template @@ -171,6 +250,10 @@ inline celix::Promise celix::Deferred::getPromise() { return celix::Promise{state}; } +inline celix::Promise celix::Deferred::getPromise() { + return celix::Promise{state}; +} + template template inline void celix::Deferred::resolveWith(celix::Promise with) { @@ -185,6 +268,18 @@ inline void celix::Deferred::resolveWith(celix::Promise with) { }); } +inline void celix::Deferred::resolveWith(celix::Promise with) { + auto s = state; + with.onResolve([s, with]{ + if (with.isSuccessfullyResolved()) { + with.getValue(); + s->resolve(); + } else { + s->fail(with.getFailure()); + } + }); +} + template inline void celix::Deferred::resolve(T&& value) { state->resolve(std::forward(value)); @@ -194,3 +289,7 @@ template inline void celix::Deferred::resolve(const T& value) { state->resolve(value); } + +inline void celix::Deferred::resolve() { + state->resolve(); +} diff --git a/misc/experimental/promise/api/celix/Promise.h b/misc/experimental/promise/api/celix/Promise.h index 79d7f9c45..8b58f26ed 100644 --- a/misc/experimental/promise/api/celix/Promise.h +++ b/misc/experimental/promise/api/celix/Promise.h @@ -241,7 +241,7 @@ namespace celix { * @param the consumer callback * @returns A new Promise which is chained to this Promise. The returned Promise must be resolved when this Promise is resolved after the specified Consumer is executed. */ - Promise thenAccept(std::function consumer); + Promise thenAccept(std::function consumer); /** * Fall back to the value of the specified Promise if this Promise fails. @@ -384,6 +384,52 @@ namespace celix { celix::Promise then(std::function(celix::Promise)> success, std::function)> failure = {}); private: const std::shared_ptr> state; + + friend class Promise; + }; + + template<> + class Promise { + public: + using type = void; + + explicit Promise(std::shared_ptr> s); + + bool isDone() const; + + bool isSuccessfullyResolved() const; + + std::exception_ptr getFailure() const; + + bool getValue() const; // NOLINT(modernize-use-nodiscard) + + void wait() const; //NOTE not part of the OSGI promise, wait till resolved (used in testing) + + Promise& onSuccess(std::function success); + + Promise& onFailure(std::function failure); + + Promise& onResolve(std::function callback); + + Promise recover(std::function recover); + + Promise thenAccept(std::function consumer); + + Promise fallbackTo(celix::Promise fallback); + + template + celix::Promise map(std::function mapper); + + template + Promise timeout(std::chrono::duration duration); + + template + Promise delay(std::chrono::duration duration); + + template + celix::Promise then(std::function(celix::Promise)> success, std::function)> failure = {}); + private: + const std::shared_ptr> state; }; } @@ -396,12 +442,18 @@ namespace celix { template inline celix::Promise::Promise(std::shared_ptr> s) : state{std::move(s)} { } +inline celix::Promise::Promise(std::shared_ptr> s) : state{std::move(s)} { +} template inline const T& celix::Promise::getValue() const { return state->getValue(); } +inline bool celix::Promise::getValue() const { + return state->getValue(); +} + template inline T celix::Promise::moveValue() { return state->moveValue(); @@ -412,50 +464,91 @@ inline bool celix::Promise::isDone() const { return state->isDone(); } +inline bool celix::Promise::isDone() const { + return state->isDone(); +} + template inline bool celix::Promise::isSuccessfullyResolved() const { return state->isSuccessfullyResolved(); } +inline bool celix::Promise::isSuccessfullyResolved() const { + return state->isSuccessfullyResolved(); +} + template inline std::exception_ptr celix::Promise::getFailure() const { return state->getFailure(); } +inline std::exception_ptr celix::Promise::getFailure() const { + return state->getFailure(); +} + template inline celix::Promise& celix::Promise::onSuccess(std::function success) { state->addOnSuccessConsumeCallback(std::move(success)); return *this; } +inline celix::Promise& celix::Promise::onSuccess(std::function success) { + state->addOnSuccessConsumeCallback(std::move(success)); + return *this; +} + template inline celix::Promise& celix::Promise::onFailure(std::function failure) { state->addOnFailureConsumeCallback(std::move(failure)); return *this; } +inline celix::Promise& celix::Promise::onFailure(std::function failure) { + state->addOnFailureConsumeCallback(std::move(failure)); + return *this; +} + template inline celix::Promise& celix::Promise::onResolve(std::function callback) { state->addChain(std::move(callback)); return *this; } +inline celix::Promise& celix::Promise::onResolve(std::function callback) { + state->addChain(std::move(callback)); + return *this; +} + template template inline celix::Promise celix::Promise::timeout(std::chrono::duration duration) { return celix::Promise{celix::impl::SharedPromiseState::timeout(state, duration)}; } +template +inline celix::Promise celix::Promise::timeout(std::chrono::duration duration) { + return celix::Promise{celix::impl::SharedPromiseState::timeout(state, duration)}; +} + template template inline celix::Promise celix::Promise::delay(std::chrono::duration duration) { return celix::Promise{state->delay(duration)}; } +template +inline celix::Promise celix::Promise::delay(std::chrono::duration duration) { + return celix::Promise{state->delay(duration)}; +} + template inline celix::Promise celix::Promise::recover(std::function recover) { return celix::Promise{state->recover(std::move(recover))}; -}; +} + +inline celix::Promise celix::Promise::recover(std::function recover) { + return celix::Promise{state->recover(std::move(recover))}; +} template template @@ -463,17 +556,30 @@ inline celix::Promise celix::Promise::map(std::function mapper) { return celix::Promise{state->map(std::move(mapper))}; } +template +inline celix::Promise celix::Promise::map(std::function mapper) { + return celix::Promise{state->map(std::move(mapper))}; +} + template inline celix::Promise celix::Promise::thenAccept(std::function consumer) { return celix::Promise{state->thenAccept(std::move(consumer))}; } +inline celix::Promise celix::Promise::thenAccept(std::function consumer) { + return celix::Promise{state->thenAccept(std::move(consumer))}; +} + template inline celix::Promise celix::Promise::fallbackTo(celix::Promise fallback) { return celix::Promise{state->fallbackTo(fallback.state)}; } +inline celix::Promise celix::Promise::fallbackTo(celix::Promise fallback) { + return celix::Promise{state->fallbackTo(fallback.state)}; +} + template inline celix::Promise celix::Promise::filter(std::function predicate) { return celix::Promise{state->filter(std::move(predicate))}; @@ -484,6 +590,10 @@ inline void celix::Promise::wait() const { state->wait(); } +inline void celix::Promise::wait() const { + state->wait(); +} + template template inline celix::Promise celix::Promise::then(std::function(celix::Promise)> success, std::function)> failure) { @@ -509,4 +619,30 @@ inline celix::Promise celix::Promise::then(std::function }; state->addChain(std::move(chain)); return celix::Promise{p}; +} + +template +inline celix::Promise celix::Promise::then(std::function(celix::Promise)> success, std::function)> failure) { + auto s = state; + auto p = std::make_shared>(state->getExecutor()); + + auto chain = [s, p, success, failure]() { + //chain is called when s is resolved + if (s->isSuccessfullyResolved()) { + try { + auto tmpPromise = success(celix::Promise{s}); + p->resolveWith(tmpPromise.state); + } catch (...) { + //failure(); TODO not sure if this needs to be called + p->fail(std::current_exception()); + } + } else { + if (failure) { + failure(celix::Promise{s}); + } + p->fail(s->getFailure()); + } + }; + state->addChain(std::move(chain)); + return celix::Promise{p}; } \ No newline at end of file diff --git a/misc/experimental/promise/api/celix/PromiseFactory.h b/misc/experimental/promise/api/celix/PromiseFactory.h index 6f430441d..a9b81a0ad 100644 --- a/misc/experimental/promise/api/celix/PromiseFactory.h +++ b/misc/experimental/promise/api/celix/PromiseFactory.h @@ -41,6 +41,8 @@ namespace celix { template celix::Promise resolved(T&& value); + celix::Promise resolved(); + //TODO rest private: tbb::task_arena executor; //TODO look into different thread pool libraries @@ -78,4 +80,10 @@ inline celix::Promise celix::PromiseFactory::resolved(T &&value) { auto p = std::make_shared>(executor); p->resolve(std::forward(value)); return celix::Promise{p}; +} + +inline celix::Promise celix::PromiseFactory::resolved() { + auto p = std::make_shared>(executor); + p->resolve(); + return celix::Promise{p}; } \ No newline at end of file diff --git a/misc/experimental/promise/api/celix/impl/SharedPromiseState.h b/misc/experimental/promise/api/celix/impl/SharedPromiseState.h index a673f4d8f..7c6ab1c55 100644 --- a/misc/experimental/promise/api/celix/impl/SharedPromiseState.h +++ b/misc/experimental/promise/api/celix/impl/SharedPromiseState.h @@ -23,6 +23,7 @@ #include #include #include +#include #include #include @@ -94,9 +95,6 @@ namespace celix { void addChain(std::function chainFunction); tbb::task_arena getExecutor() const; - -// template -// std::shared_ptr> then(std::function success, std::function failure); private: /** * Complete the resolving and call the registered tasks @@ -120,6 +118,82 @@ namespace celix { std::exception_ptr exp{nullptr}; DataType data{}; }; + + template<> + class SharedPromiseState { + public: + explicit SharedPromiseState(const tbb::task_arena &executor = {}); + + ~SharedPromiseState() = default; + + void resolve(); + + void fail(std::exception_ptr p); + + void fail(const std::exception &e); + + void tryResolve(); + + void tryFail(std::exception_ptr p); + + bool getValue() const; //copy + std::exception_ptr getFailure() const; + + void wait() const; + + bool isDone() const; + + bool isSuccessfullyResolved() const; + + void addOnSuccessConsumeCallback(std::function callback); + + void addOnFailureConsumeCallback(std::function callback); + + void addOnResolve(std::function callback); + + template + std::shared_ptr> delay(std::chrono::duration duration); + + std::shared_ptr> recover(std::function recover); + + std::shared_ptr> fallbackTo(std::shared_ptr> fallbackTo); + + void resolveWith(std::shared_ptr> with); + + template + std::shared_ptr> map(std::function mapper); + + std::shared_ptr> thenAccept(std::function consumer); + + template + static std::shared_ptr> + timeout(std::shared_ptr> state, std::chrono::duration duration); + + void addChain(std::function chainFunction); + + tbb::task_arena getExecutor() const; + private: + /** + * Complete the resolving and call the registered tasks + * A reference to the possible locked unique_lock. + */ + void complete(std::unique_lock &lck); + + /** + * Wait for data and check if it resolved as expected (expects mutex locked) + */ + void waitForAndCheckData(std::unique_lock &lck, bool expectValid) const; + + tbb::task_arena executor; //TODO look into different thread pool libraries + //TODO add ScheduledExecutorService like object + + mutable std::mutex mutex{}; //protects below + mutable std::condition_variable cond{}; + bool done = false; + bool dataMoved = false; + std::vector> chain{}; //chain tasks are executed on thread pool. + std::exception_ptr exp{nullptr}; + }; } } @@ -131,6 +205,8 @@ namespace celix { template inline celix::impl::SharedPromiseState::SharedPromiseState(const tbb::task_arena& _executor) : executor{_executor} {} +inline celix::impl::SharedPromiseState::SharedPromiseState(const tbb::task_arena& _executor) : executor{_executor} {} + template inline celix::impl::SharedPromiseState::~SharedPromiseState() { std::unique_lock lck{mutex}; @@ -167,13 +243,31 @@ inline void celix::impl::SharedPromiseState::resolve(const T& value) { complete(lck); } +inline void celix::impl::SharedPromiseState::resolve() { + std::unique_lock lck{mutex}; + if (done) { + throw celix::PromiseInvocationException("Cannot resolve Promise. Promise is already done"); + } + exp = nullptr; + complete(lck); +} + template inline void celix::impl::SharedPromiseState::fail(std::exception_ptr e) { std::unique_lock lck{mutex}; if (done) { throw celix::PromiseInvocationException("Cannot fail Promise. Promise is already done"); } - exp = e; + exp = std::move(e); + complete(lck); +} + +inline void celix::impl::SharedPromiseState::fail(std::exception_ptr e) { + std::unique_lock lck{mutex}; + if (done) { + throw celix::PromiseInvocationException("Cannot fail Promise. Promise is already done"); + } + exp = std::move(e); complete(lck); } @@ -182,6 +276,10 @@ inline void celix::impl::SharedPromiseState::fail(const std::exception& e) { fail(std::make_exception_ptr(e)); } +inline void celix::impl::SharedPromiseState::fail(const std::exception& e) { + fail(std::make_exception_ptr(e)); +} + template inline void celix::impl::SharedPromiseState::tryResolve(T&& value) { std::unique_lock lck{mutex}; @@ -192,11 +290,27 @@ inline void celix::impl::SharedPromiseState::tryResolve(T&& value) { } } +inline void celix::impl::SharedPromiseState::tryResolve() { + std::unique_lock lck{mutex}; + if (!done) { + exp = nullptr; + complete(lck); + } +} + template inline void celix::impl::SharedPromiseState::tryFail(std::exception_ptr e) { std::unique_lock lck{mutex}; if (!done) { - exp = e; + exp = std::move(e); + complete(lck); + } +} + +inline void celix::impl::SharedPromiseState::tryFail(std::exception_ptr e) { + std::unique_lock lck{mutex}; + if (!done) { + exp = std::move(e); complete(lck); } } @@ -207,12 +321,22 @@ inline bool celix::impl::SharedPromiseState::isDone() const { return done; } +inline bool celix::impl::SharedPromiseState::isDone() const { + std::lock_guard lck{mutex}; + return done; +} + template inline bool celix::impl::SharedPromiseState::isSuccessfullyResolved() const { std::lock_guard lck{mutex}; return done && !exp; } +inline bool celix::impl::SharedPromiseState::isSuccessfullyResolved() const { + std::lock_guard lck{mutex}; + return done && !exp; +} + template inline void celix::impl::SharedPromiseState::waitForAndCheckData(std::unique_lock& lck, bool expectValid) const { @@ -229,6 +353,20 @@ inline void celix::impl::SharedPromiseState::waitForAndCheckData(std::unique_ } } +inline void celix::impl::SharedPromiseState::waitForAndCheckData(std::unique_lock& lck, bool expectValid) const { + if (!lck.owns_lock()) { + lck.lock(); + } + cond.wait(lck, [this]{return done;}); + if (expectValid && exp) { + throw celix::PromiseInvocationException{"Expected a succeeded promise, but promise failed"}; + } else if(!expectValid && !exp && !dataMoved) { + throw celix::PromiseInvocationException{"Expected a failed promise, but promise succeeded"}; + } else if (dataMoved) { + throw celix::PromiseInvocationException{"Invalid use of promise, data is moved and not available anymore!"}; + } +} + template inline const T& celix::impl::SharedPromiseState::getValue() const { std::unique_lock lck{mutex}; @@ -237,11 +375,21 @@ inline const T& celix::impl::SharedPromiseState::getValue() const { return *ptr; } +inline bool celix::impl::SharedPromiseState::getValue() const { + std::unique_lock lck{mutex}; + waitForAndCheckData(lck, true); + return true; +} + template inline tbb::task_arena celix::impl::SharedPromiseState::getExecutor() const { return executor; } +inline tbb::task_arena celix::impl::SharedPromiseState::getExecutor() const { + return executor; +} + template inline T celix::impl::SharedPromiseState::moveValue() { std::unique_lock lck{mutex}; @@ -257,6 +405,11 @@ inline void celix::impl::SharedPromiseState::wait() const { cond.wait(lck, [this]{return done;}); } +inline void celix::impl::SharedPromiseState::wait() const { + std::unique_lock lck{mutex}; + cond.wait(lck, [this]{return done;}); +} + template inline std::exception_ptr celix::impl::SharedPromiseState::getFailure() const { std::unique_lock lck{mutex}; @@ -264,13 +417,29 @@ inline std::exception_ptr celix::impl::SharedPromiseState::getFailure() const return exp; } +inline std::exception_ptr celix::impl::SharedPromiseState::getFailure() const { + std::unique_lock lck{mutex}; + waitForAndCheckData(lck, false); + return exp; +} + template inline void celix::impl::SharedPromiseState::resolveWith(std::shared_ptr> with) { with->addOnResolve([this](bool succeeded, T* v, std::exception_ptr e) { if (succeeded) { tryResolve(std::forward(*v)); } else { - tryFail(e); + tryFail(std::move(e)); + } + }); +} + +inline void celix::impl::SharedPromiseState::resolveWith(std::shared_ptr> with) { + with->addOnResolve([this](bool succeeded, std::exception_ptr e) { + if (succeeded) { + tryResolve(); + } else { + tryFail(std::move(e)); } }); } @@ -288,6 +457,18 @@ inline std::shared_ptr> celix::impl::SharedPr return p; } +template +inline std::shared_ptr> celix::impl::SharedPromiseState::timeout(std::shared_ptr> state, std::chrono::duration duration) { + auto p = std::make_shared>(state->executor); + p->resolveWith(state); + state->executor.execute([duration, p]{ + std::this_thread::sleep_for(duration); //TODO use scheduler instead of sleep on thread (using unnecessary resources) + p->tryFail(std::make_exception_ptr(celix::PromiseTimeoutException{})); + //TODO is a callback to deferred needed to abort ? + }); + return p; +} + template template inline std::shared_ptr> celix::impl::SharedPromiseState::delay(std::chrono::duration duration) { @@ -299,7 +480,29 @@ inline std::shared_ptr> celix::impl::SharedPr if (succeeded) { p->resolve(std::forward(*v)); } else { - p->fail(e); + p->fail(std::move(e)); + } + } catch (celix::PromiseInvocationException&) { + //somebody already resolved p? + } catch (...) { + p->fail(std::current_exception()); + } + }); + + return p; +} + +template +inline std::shared_ptr> celix::impl::SharedPromiseState::delay(std::chrono::duration duration) { + auto p = std::make_shared>(executor); + + addOnResolve([p, duration](bool succeeded, std::exception_ptr e) { + std::this_thread::sleep_for(duration); //TODO use scheduler instead of sleep on thread (using unnecessary resources) + try { + if (succeeded) { + p->resolve(); + } else { + p->fail(std::move(e)); } } catch (celix::PromiseInvocationException&) { //somebody already resolved p? @@ -319,15 +522,36 @@ inline std::shared_ptr> celix::impl::SharedPr auto p = std::make_shared>(executor); addOnResolve([p, recover](bool succeeded, T *v, const std::exception_ptr& /*e*/) { - if (succeeded) { - p->resolve(std::forward(*v)); - } else { - try { - p->resolve(recover()); - } catch (...) { - p->fail(std::current_exception()); //or state->failure(); - } - } + if (succeeded) { + p->resolve(std::forward(*v)); + } else { + try { + p->resolve(recover()); + } catch (...) { + p->fail(std::current_exception()); //or state->failure(); + } + } + }); + return p; +} + +inline std::shared_ptr> celix::impl::SharedPromiseState::recover(std::function recover) { + if (!recover) { + throw celix::PromiseInvocationException{"provided recover callback is not valid"}; + } + auto p = std::make_shared>(executor); + + addOnResolve([p, recover](bool succeeded, const std::exception_ptr& /*e*/) { + if (succeeded) { + p->resolve(); + } else { + try { + recover(); + p->resolve(); + } catch (...) { + p->fail(std::current_exception()); //or state->failure(); + } + } }); return p; } @@ -379,6 +603,25 @@ inline std::shared_ptr> celix::impl::SharedPr return p; } +inline std::shared_ptr> celix::impl::SharedPromiseState::fallbackTo(std::shared_ptr> fallbackTo) { + auto p = std::make_shared>(executor); + auto chainFunction = [this, p, fallbackTo] { + if (isSuccessfullyResolved()) { + getValue(); + p->resolve(); + } else { + if (fallbackTo->isSuccessfullyResolved()) { + fallbackTo->getValue(); + p->resolve(); + } else { + p->fail(getFailure()); + } + } + }; + addChain(std::move(chainFunction)); + return p; +} + template inline void celix::impl::SharedPromiseState::addChain(std::function chainFunction) { std::function localChain{}; @@ -396,6 +639,22 @@ inline void celix::impl::SharedPromiseState::addChain(std::function c } } +inline void celix::impl::SharedPromiseState::addChain(std::function chainFunction) { + std::function localChain{}; + { + + std::lock_guard lck{mutex}; + if (!done) { + chain.push_back(std::move(chainFunction)); + } else { + localChain = std::move(chainFunction); + } + } + if (localChain) { + localChain(); + } +} + template template inline std::shared_ptr> celix::impl::SharedPromiseState::map(std::function mapper) { @@ -419,6 +678,29 @@ inline std::shared_ptr> celix::impl::SharedPr return p; } +template +inline std::shared_ptr> celix::impl::SharedPromiseState::map(std::function mapper) { + if (!mapper) { + throw celix::PromiseInvocationException("provided mapper is not valid"); + } + auto p = std::make_shared>(executor); + auto chainFunction = [this, p, mapper] { + try { + if (isSuccessfullyResolved()) { + getValue(); + R val = mapper(); + p->resolve(std::forward(val)); + } else { + p->fail(getFailure()); + } + } catch (...) { + p->fail(std::current_exception()); + } + }; + addChain(std::move(chainFunction)); + return p; +} + template inline std::shared_ptr> celix::impl::SharedPromiseState::thenAccept(std::function consumer) { if (!consumer) { @@ -442,6 +724,28 @@ inline std::shared_ptr> celix::impl::SharedPr return p; } +inline std::shared_ptr> celix::impl::SharedPromiseState::thenAccept(std::function consumer) { + if (!consumer) { + throw celix::PromiseInvocationException("provided consumer is not valid"); + } + auto p = std::make_shared>(executor); + auto chainFunction = [this, p, consumer] { + if (isSuccessfullyResolved()) { + try { + getValue(); + consumer(); + p->resolve(); + } catch (...) { + p->fail(std::current_exception()); + } + } else { + p->fail(getFailure()); + } + }; + addChain(std::move(chainFunction)); + return p; +} + template inline void celix::impl::SharedPromiseState::addOnResolve(std::function callback) { std::function task = [this, callback] { @@ -457,6 +761,18 @@ inline void celix::impl::SharedPromiseState::addOnResolve(std::function::addOnResolve(std::function callback) { + std::function task = [this, callback] { + std::exception_ptr e = nullptr; + { + std::lock_guard lck{mutex}; + e = exp; + } + callback(!e, e); + }; + addChain(task); +} + template inline void celix::impl::SharedPromiseState::addOnSuccessConsumeCallback(std::function callback) { std::function task = [this, callback] { @@ -467,6 +783,16 @@ inline void celix::impl::SharedPromiseState::addOnSuccessConsumeCallback(std: addChain(task); } +inline void celix::impl::SharedPromiseState::addOnSuccessConsumeCallback(std::function callback) { + std::function task = [this, callback] { + if (isSuccessfullyResolved()) { + getValue(); + callback(); + } + }; + addChain(task); +} + template inline void celix::impl::SharedPromiseState::addOnFailureConsumeCallback(std::function callback) { std::function task = [this, callback] { @@ -485,6 +811,23 @@ inline void celix::impl::SharedPromiseState::addOnFailureConsumeCallback(std: addChain(task); } +inline void celix::impl::SharedPromiseState::addOnFailureConsumeCallback(std::function callback) { + std::function task = [this, callback] { + if (!isSuccessfullyResolved()) { + try { + std::rethrow_exception(getFailure()); + } catch (const std::exception &e) { + callback(e); + } catch (...) { + //NOTE not a exception based on std::exception, "repacking" it to logical error + std::logic_error logicError{"Unknown exception throw for the failure of A celix::Promise"}; + callback(logicError); + } + } + }; + addChain(task); +} + template inline void celix::impl::SharedPromiseState::complete(std::unique_lock& lck) { if (!lck.owns_lock()) { @@ -496,6 +839,27 @@ inline void celix::impl::SharedPromiseState::complete(std::unique_lock> localChain{}; + localChain.swap(chain); + lck.unlock(); + for (auto &chainTask : localChain) { + executor.execute(chainTask); //TODO maybe use std::move? //TODO optimize if complete is already executor on executor? + } + lck.lock(); + } +} + +inline void celix::impl::SharedPromiseState::complete(std::unique_lock& lck) { + if (!lck.owns_lock()) { + lck.lock(); + } + if (done) { + throw celix::PromiseInvocationException("Promise is already resolved"); + } + done = true; + cond.notify_all(); + while (!chain.empty()) { std::vector> localChain{}; localChain.swap(chain); diff --git a/misc/experimental/promise/gtest/CMakeLists.txt b/misc/experimental/promise/gtest/CMakeLists.txt index d7de9ea6e..fdc3a2184 100644 --- a/misc/experimental/promise/gtest/CMakeLists.txt +++ b/misc/experimental/promise/gtest/CMakeLists.txt @@ -17,6 +17,7 @@ add_executable(test_promise src/PromiseTestSuite.cc + src/VoidPromiseTestSuite.cc ) target_link_libraries(test_promise PRIVATE GTest::gtest GTest::gtest_main Celix::Promise) diff --git a/misc/experimental/promise/gtest/src/PromiseTestSuite.cc b/misc/experimental/promise/gtest/src/PromiseTestSuite.cc index f5d19328d..1b99c0ded 100644 --- a/misc/experimental/promise/gtest/src/PromiseTestSuite.cc +++ b/misc/experimental/promise/gtest/src/PromiseTestSuite.cc @@ -96,7 +96,7 @@ TEST_F(PromiseTestSuite, onSuccessHandling) { EXPECT_EQ(true, resolveCalled); } -TEST_F(PromiseTestSuite, onFailreHandling) { +TEST_F(PromiseTestSuite, onFailureHandling) { auto deferred = factory.deferred(); bool successCalled = false; bool failureCalled = false; diff --git a/misc/experimental/promise/gtest/src/VoidPromiseTestSuite.cc b/misc/experimental/promise/gtest/src/VoidPromiseTestSuite.cc new file mode 100644 index 000000000..679aaa55f --- /dev/null +++ b/misc/experimental/promise/gtest/src/VoidPromiseTestSuite.cc @@ -0,0 +1,352 @@ +/** + *Licensed to the Apache Software Foundation (ASF) under one + *or more contributor license agreements. See the NOTICE file + *distributed with this work for additional information + *regarding copyright ownership. The ASF licenses this file + *to you under the Apache License, Version 2.0 (the + *"License"); you may not use this file except in compliance + *with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + *Unless required by applicable law or agreed to in writing, + *software distributed under the License is distributed on an + *"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + *specific language governing permissions and limitations + *under the License. + */ + +#include + +#include + +#include "celix/PromiseFactory.h" + +class VoidPromiseTestSuite : public ::testing::Test { +public: + ~VoidPromiseTestSuite() override = default; + + celix::PromiseFactory factory{ tbb::task_arena{5, 1} }; +}; + + + +TEST_F(VoidPromiseTestSuite, simplePromise) { + auto deferred = factory.deferred(); + std::thread t{[deferred] () mutable { //TODO TBD make deferred a shared_ptr to prevent need for mutable? + std::this_thread::sleep_for(std::chrono::milliseconds{50}); + deferred.resolve(); + }}; + auto promise = deferred.getPromise(); + EXPECT_TRUE(promise.getValue()); //block until ready + EXPECT_TRUE(promise.isDone()); //got value, so promise is done + EXPECT_ANY_THROW(promise.getFailure()); //succeeded, so no exception available + + EXPECT_TRUE(promise.getValue()); //note multiple call are valid; + t.join(); +} + +TEST_F(VoidPromiseTestSuite, failingPromise) { + auto deferred = factory.deferred(); + auto cpy = deferred; + std::thread t{[deferred] () mutable { + deferred.fail(std::logic_error{"failing"}); + }}; + auto promise = deferred.getPromise(); + EXPECT_THROW(promise.getValue(), celix::PromiseInvocationException); + EXPECT_TRUE(promise.getFailure() != nullptr); + t.join(); +} + +TEST_F(VoidPromiseTestSuite, failingPromiseWithExceptionPtr) { + auto deferred = factory.deferred(); + std::thread t{[&deferred]{ + try { + std::string{}.at(1); // this generates an std::out_of_range + //or use std::make_exception_ptr + } catch (...) { + deferred.fail(std::current_exception()); + } + }}; + auto promise = deferred.getPromise(); + EXPECT_THROW(promise.getValue(), celix::PromiseInvocationException); + EXPECT_TRUE(promise.getFailure() != nullptr); + t.join(); +} + +TEST_F(VoidPromiseTestSuite, onSuccessHandling) { + auto deferred = factory.deferred(); + bool called = false; + bool resolveCalled = false; + auto p = deferred.getPromise() + .onSuccess([&called]() { + called = true; + }) + .onResolve([&resolveCalled]() { + resolveCalled = true; + }); + deferred.resolve(); + p.wait(); + EXPECT_EQ(true, called); + EXPECT_EQ(true, resolveCalled); +} + +TEST_F(VoidPromiseTestSuite, onFailureHandling) { + auto deferred = factory.deferred(); + bool successCalled = false; + bool failureCalled = false; + bool resolveCalled = false; + auto p = deferred.getPromise() + .onSuccess([&]() { + successCalled = true; + }) + .onFailure([&](const std::exception &e) { + failureCalled = true; + std::cout << "got error: " << e.what() << std::endl; + }) + .onResolve([&resolveCalled]() { + resolveCalled = true; + }); + try { + std::string{}.at(1); // this generates an std::out_of_range + //or use std::make_exception_ptr + } catch (...) { + deferred.fail(std::current_exception()); + } + p.wait(); + EXPECT_EQ(false, successCalled); + EXPECT_EQ(true, failureCalled); + EXPECT_EQ(true, resolveCalled); +} + +TEST_F(VoidPromiseTestSuite, resolveSuccessWith) { + auto deferred1 = factory.deferred(); + auto deferred2 = factory.deferred(); + + bool called = false; + deferred1.getPromise() + .onSuccess([&called]() { + called = true; + }); + + //currently deferred1 will be resolved in thread, and onSuccess is trigger on the promise of deferred2 + //now resolving deferred2 with the promise of deferred1 + deferred2.resolveWith(deferred1.getPromise()); + auto p = deferred2.getPromise(); + deferred1.resolve(); + p.wait(); + EXPECT_EQ(true, called); +} + +TEST_F(VoidPromiseTestSuite, resolveFailureWith) { + auto deferred1 = factory.deferred(); + auto deferred2 = factory.deferred(); + bool failureCalled = false; + bool successCalled = false; + deferred2.getPromise() + .onSuccess([&]() { + successCalled = true; + }) + .onFailure([&](const std::exception &e) { + failureCalled = true; + std::cout << "got error: " << e.what() << std::endl; + }); + + //currently deferred1 will be resolved in thread, and onSuccess is trigger on the promise of deferred2 + //now resolving deferred2 with the promise of deferred1 + deferred2.resolveWith(deferred1.getPromise()); + auto p = deferred2.getPromise(); + try { + std::string().at(1); // this generates an std::out_of_range + //or use std::make_exception_ptr + } catch (...) { + deferred1.fail(std::current_exception()); + } + p.wait(); + EXPECT_EQ(false, successCalled); + EXPECT_EQ(true, failureCalled); +} + +TEST_F(VoidPromiseTestSuite, resolveWithTimeout) { + auto deferred1 = factory.deferred(); + std::thread t{[&deferred1]{ + std::this_thread::sleep_for(std::chrono::milliseconds{50}); + try { + deferred1.resolve(); + } catch(...) { + //note resolve with throws an exception if promise is already resolved + } + }}; + + bool firstSuccessCalled = false; + bool secondSuccessCalled = false; + bool secondFailedCalled = false; + auto p = deferred1.getPromise() + .onSuccess([&firstSuccessCalled]() { + firstSuccessCalled = true; + }) + .timeout(std::chrono::milliseconds{10}) + .onSuccess([&secondSuccessCalled]() { + secondSuccessCalled = true; + }) + .onFailure([&secondFailedCalled](const std::exception&) { + secondFailedCalled = true; + }); + t.join(); + p.wait(); + EXPECT_EQ(true, firstSuccessCalled); + EXPECT_EQ(false, secondSuccessCalled); + EXPECT_EQ(true, secondFailedCalled); + + firstSuccessCalled = false; + secondSuccessCalled = false; + secondFailedCalled = false; + auto p2 = deferred1.getPromise() + .onSuccess([&firstSuccessCalled]() { + firstSuccessCalled = true; + }) + .timeout(std::chrono::milliseconds{50}) + .onSuccess([&secondSuccessCalled]() { + secondSuccessCalled = true; + }) + .onFailure([&secondFailedCalled](const std::exception&) { + secondFailedCalled = true; + }); + p2.wait(); + EXPECT_EQ(true, firstSuccessCalled); + EXPECT_EQ(true, secondSuccessCalled); + EXPECT_EQ(false, secondFailedCalled); +} + +TEST_F(VoidPromiseTestSuite, resolveWithDelay) { + auto deferred1 = factory.deferred(); + bool successCalled = false; + bool failedCalled = false; + auto t1 = std::chrono::system_clock::now(); + std::chrono::system_clock::time_point t2; + auto p = deferred1.getPromise() + .delay(std::chrono::milliseconds{50}) + .onSuccess([&successCalled, &t2]() { + successCalled = true; + t2 = std::chrono::system_clock::now(); + }) + .onFailure([&failedCalled](const std::exception&) { + failedCalled = true; + }); + deferred1.resolve(); + p.wait(); + EXPECT_EQ(true, successCalled); + EXPECT_EQ(false, failedCalled); + auto durationInMs = std::chrono::duration_cast(t2 - t1); + EXPECT_GE(durationInMs, std::chrono::milliseconds{10}); +} + + +TEST_F(VoidPromiseTestSuite, resolveWithRecover) { + auto deferred1 = factory.deferred(); + bool successCalled = false; + deferred1.getPromise() + .recover([]{ return 42; }) + .onSuccess([&successCalled]() { + successCalled = true; + }); + try { + throw std::logic_error("failure"); + } catch (...) { + deferred1.fail(std::current_exception()); + } + EXPECT_EQ(true, successCalled); +} + +TEST_F(VoidPromiseTestSuite, chainAndMapResult) { + auto deferred1 = factory.deferred(); + std::thread t{[&deferred1]{ + deferred1.resolve(); + }}; + int two = deferred1.getPromise() + .map([]() { + return 2; + }).getValue(); + t.join(); + EXPECT_EQ(2, two); +} + +TEST_F(VoidPromiseTestSuite, chainWithThenAccept) { + auto deferred1 = factory.deferred(); + bool called = false; + deferred1.getPromise() + .thenAccept([&called](){ + called = true; + }); + deferred1.resolve(); + EXPECT_TRUE(called); +} + +TEST_F(VoidPromiseTestSuite, promiseWithFallbackTo) { + auto deferred1 = factory.deferred(); + try { + throw std::logic_error("failure"); + } catch (...) { + deferred1.fail(std::current_exception()); + } + + auto deferred2 = factory.deferred(); + deferred2.resolve(); + + + long val = deferred1.getPromise().fallbackTo(deferred2.getPromise()).getValue(); + EXPECT_TRUE(val); +} + +TEST_F(VoidPromiseTestSuite, outOfScopeUnresolvedPromises) { + bool called = false; + { + auto deferred1 = factory.deferred(); + deferred1.getPromise().onResolve([&]{ + called = true; + }); + //promise and deferred out of scope + } + EXPECT_FALSE(called); +} + +TEST_F(VoidPromiseTestSuite, chainPromises) { + auto success = [](celix::Promise p) -> celix::Promise { + //TODO Promises::resolved(p.getValue() + p.getValue()) + celix::Deferred result; + p.getValue(); + result.resolve(42); + return result.getPromise(); + }; + celix::Deferred initial; + initial.resolve(); + long result = initial.getPromise().then(success).getValue(); + EXPECT_EQ(42, result); +} + +TEST_F(VoidPromiseTestSuite, chainFailedPromises) { + bool called = false; + auto success = [](celix::Promise p) -> celix::Promise { + //nop + return p; + }; + auto failed = [&called](const celix::Promise& /*p*/) -> void { + called = true; + }; + celix::Deferred deferred; + deferred.fail(std::logic_error{"fail"}); + deferred.getPromise().then(success, failed).wait(); + EXPECT_TRUE(called); +} + +TEST_F(VoidPromiseTestSuite, failedResolvedWithPromiseFactory) { + auto factory = celix::PromiseFactory{}; + auto p1 = factory.failed(std::logic_error{"test"}); + EXPECT_TRUE(p1.isDone()); + EXPECT_NE(nullptr, p1.getFailure()); + + auto p2 = factory.resolved(); + EXPECT_TRUE(p2.isDone()); + EXPECT_TRUE(p2.getValue()); +} \ No newline at end of file