Commit 7d702528 authored by Dominik Charousset's avatar Dominik Charousset

Fix copy-constructors of async::promise

parent f30ebfdd
......@@ -121,7 +121,7 @@ private:
} // namespace caf
namespace caf::detail {
template <class F>
template <class F, bool IsSingleShot>
struct default_action_impl : detail::atomic_ref_counted, action::impl {
std::atomic<action::state> state_;
F f_;
......@@ -149,6 +149,8 @@ struct default_action_impl : detail::atomic_ref_counted, action::impl {
// to implement time-based loops.
if (state_.load() == action::state::scheduled) {
f_();
if constexpr (IsSingleShot)
state_ = action::state::disposed;
}
}
......@@ -177,7 +179,15 @@ namespace caf {
/// @param f The body for the action.
template <class F>
action make_action(F f) {
using impl_t = detail::default_action_impl<F>;
using impl_t = detail::default_action_impl<F, false>;
return action{make_counted<impl_t>(std::move(f))};
}
/// Convenience function for creating an @ref action from a function object.
/// @param f The body for the action.
template <class F>
action make_single_shot_action(F f) {
using impl_t = detail::default_action_impl<F, true>;
return action{make_counted<impl_t>(std::move(f))};
}
......
......@@ -47,7 +47,7 @@ public:
g(static_cast<const error&>(std::get<error>(cp->value)));
}
};
auto cb_action = make_action(std::move(cb));
auto cb_action = make_single_shot_action(std::move(cb));
auto event = typename cell_type::event{ctx_, cb_action};
bool fire_immediately = false;
{ // Critical section.
......@@ -115,6 +115,16 @@ public:
return {ctx, cell_};
}
/// Queries whether the result of the asynchronous computation is still
/// pending, i.e., neither `set_value` nor `set_error` has been called on the
/// @ref promise.
/// @pre `valid()`
bool pending() const {
CAF_ASSERT(valid());
std::unique_lock guard{cell_->mtx};
return std::holds_alternative<none_t>(cell_->value);
}
private:
using cell_ptr = std::shared_ptr<detail::async_cell<T>>;
......
......@@ -19,16 +19,25 @@ template <class T>
class promise {
public:
promise(promise&&) noexcept = default;
promise(const promise&) noexcept = default;
promise& operator=(promise&&) noexcept = default;
promise& operator=(const promise&) noexcept = default;
promise(const promise& other) noexcept : promise(other.cell_) {
// nop
}
promise& operator=(const promise& other) noexcept {
promise copy{other};
cell_.swap(copy.cell_);
return *this;
}
promise() : cell_(std::make_shared<cell_type>()) {
// nop
}
~promise() {
if (cell_) {
if (valid()) {
auto& cnt = cell_->promises;
if (cnt == 1 || cnt.fetch_sub(1, std::memory_order_acq_rel) == 1) {
typename cell_type::event_list events;
......@@ -59,7 +68,7 @@ public:
/// @pre `valid()`
void set_value(T value) {
if (cell_) {
if (valid()) {
do_set(value);
cell_ = nullptr;
}
......@@ -67,7 +76,7 @@ public:
/// @pre `valid()`
void set_error(error reason) {
if (cell_) {
if (valid()) {
do_set(reason);
cell_ = nullptr;
}
......@@ -82,7 +91,7 @@ private:
using cell_type = detail::async_cell<T>;
using cell_ptr = std::shared_ptr<cell_type>;
explicit promise(cell_type cell) : cell_(std::move(cell)) {
explicit promise(cell_ptr cell) noexcept : cell_(std::move(cell)) {
CAF_ASSERT(cell_ != nullptr);
cell_->promises.fetch_add(1, std::memory_order_relaxed);
}
......
......@@ -8,6 +8,7 @@
#include "core-test.hpp"
#include "caf/flow/scoped_coordinator.hpp"
#include "caf/scheduled_actor/flow.hpp"
using namespace caf;
......@@ -71,4 +72,55 @@ SCENARIO("actors can observe futures") {
}
}
SCENARIO("never setting a value or an error breaks the promises") {
GIVEN("multiple promises that point to the same cell") {
WHEN("the last promise goes out of scope") {
THEN("the future reports a broken promise") {
using promise_t = async::promise<int32_t>;
using future_t = async::future<int32_t>;
future_t fut;
{
auto uut = promise_t{};
fut = uut.get_future();
CHECK(fut.pending());
{
// copy ctor
promise_t cpy{uut};
CHECK(fut.pending());
// move ctor
promise_t mv{std::move(cpy)};
CHECK(fut.pending());
{
// copy assign
promise_t cpy2;
cpy2 = mv;
CHECK(fut.pending());
// move assign
promise_t mv2;
mv2 = std::move(mv);
CHECK(fut.pending());
}
CHECK(fut.pending());
}
CHECK(fut.pending());
}
CHECK(!fut.pending());
auto ctx = flow::scoped_coordinator::make();
size_t observed_events = 0;
fut.bind_to(ctx.get()).then(
[&observed_events](int32_t) {
++observed_events;
FAIL("unexpected value");
},
[&observed_events](const error& err) {
++observed_events;
CHECK_EQ(err, make_error(sec::broken_promise));
});
ctx->run();
CHECK_EQ(observed_events, 1u);
}
}
}
}
END_FIXTURE_SCOPE()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment