Commit dc423f59 authored by Dominik Charousset's avatar Dominik Charousset

Have response promises hold a strong reference

By having the response promises hold onto a strong reference, we allow
actors to wait for events such as other actors going down before
responding to a promise without having to somehow ensure at least one
strong reference to the monitoring actor.
parent 063022f1
...@@ -35,6 +35,8 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -35,6 +35,8 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- Remote spawning of actors is no longer considered experimental. - Remote spawning of actors is no longer considered experimental.
- The output of `--dump-config` now prints valid config file syntax. - The output of `--dump-config` now prints valid config file syntax.
- Response promises now hold a strong reference to their parent actor to avoid
`broken_promise` errors in some (legitimate) edge cases (#1361).
### Deprecated ### Deprecated
......
...@@ -197,7 +197,7 @@ private: ...@@ -197,7 +197,7 @@ private:
void delegate_impl(abstract_actor* receiver, message msg); void delegate_impl(abstract_actor* receiver, message msg);
mutable size_t ref_count = 1; mutable size_t ref_count = 1;
weak_actor_ptr weak_self; strong_actor_ptr self;
strong_actor_ptr source; strong_actor_ptr source;
forwarding_stack stages; forwarding_stack stages;
message_id id; message_id id;
......
...@@ -40,7 +40,7 @@ response_promise::response_promise(local_actor* self, strong_actor_ptr source, ...@@ -40,7 +40,7 @@ response_promise::response_promise(local_actor* self, strong_actor_ptr source,
// anonymous messages since there's nowhere to send the message to anyway. // anonymous messages since there's nowhere to send the message to anyway.
if (requires_response(mid)) { if (requires_response(mid)) {
state_ = make_counted<state>(); state_ = make_counted<state>();
state_->weak_self = self->ctrl(); state_->self = self->ctrl();
state_->source.swap(source); state_->source.swap(source);
state_->stages.swap(stages); state_->stages.swap(stages);
state_->id = mid; state_->id = mid;
...@@ -60,7 +60,7 @@ bool response_promise::async() const noexcept { ...@@ -60,7 +60,7 @@ bool response_promise::async() const noexcept {
} }
bool response_promise::pending() const noexcept { bool response_promise::pending() const noexcept {
return state_ != nullptr && state_->weak_self != nullptr; return state_ != nullptr && state_->self != nullptr;
} }
strong_actor_ptr response_promise::source() const noexcept { strong_actor_ptr response_promise::source() const noexcept {
...@@ -122,7 +122,7 @@ void response_promise::respond_to(local_actor* self, mailbox_element* request, ...@@ -122,7 +122,7 @@ void response_promise::respond_to(local_actor* self, mailbox_element* request,
if (request && requires_response(*request) if (request && requires_response(*request)
&& has_response_receiver(*request)) { && has_response_receiver(*request)) {
state tmp; state tmp;
tmp.weak_self = self->ctrl(); tmp.self = self->ctrl();
tmp.source.swap(request->sender); tmp.source.swap(request->sender);
tmp.stages.swap(request->stages); tmp.stages.swap(request->stages);
tmp.id = request->mid; tmp.id = request->mid;
...@@ -136,7 +136,7 @@ void response_promise::respond_to(local_actor* self, mailbox_element* request, ...@@ -136,7 +136,7 @@ void response_promise::respond_to(local_actor* self, mailbox_element* request,
if (request && requires_response(*request) if (request && requires_response(*request)
&& has_response_receiver(*request)) { && has_response_receiver(*request)) {
state tmp; state tmp;
tmp.weak_self = self->ctrl(); tmp.self = self->ctrl();
tmp.source.swap(request->sender); tmp.source.swap(request->sender);
tmp.stages.swap(request->stages); tmp.stages.swap(request->stages);
tmp.id = request->mid; tmp.id = request->mid;
...@@ -151,17 +151,16 @@ response_promise::state::~state() { ...@@ -151,17 +151,16 @@ response_promise::state::~state() {
// Note: the state may get destroyed outside of the actor. For example, when // Note: the state may get destroyed outside of the actor. For example, when
// storing the promise in a run-later continuation. Hence, we can't call // storing the promise in a run-later continuation. Hence, we can't call
// deliver_impl here since it calls self->context(). // deliver_impl here since it calls self->context().
if (weak_self && source) { if (self && source) {
CAF_LOG_DEBUG("broken promise!"); CAF_LOG_DEBUG("broken promise!");
auto element = make_mailbox_element(weak_self.lock(), id.response_id(), auto element = make_mailbox_element(self, id.response_id(), no_stages,
no_stages,
make_error(sec::broken_promise)); make_error(sec::broken_promise));
source->enqueue(std::move(element), nullptr); source->enqueue(std::move(element), nullptr);
} }
} }
void response_promise::state::cancel() { void response_promise::state::cancel() {
weak_self = nullptr; self = nullptr;
} }
void response_promise::state::deliver_impl(message msg) { void response_promise::state::deliver_impl(message msg) {
...@@ -169,17 +168,19 @@ void response_promise::state::deliver_impl(message msg) { ...@@ -169,17 +168,19 @@ void response_promise::state::deliver_impl(message msg) {
// Even though we are holding a weak pointer, we can access the pointer // Even though we are holding a weak pointer, we can access the pointer
// without any additional check here because only the actor itself is allowed // without any additional check here because only the actor itself is allowed
// to call this function. // to call this function.
auto self = static_cast<local_actor*>(weak_self.get()->get()); auto selfptr = static_cast<local_actor*>(self->get());
if (msg.empty() && id.is_async()) { if (msg.empty() && id.is_async()) {
CAF_LOG_DEBUG("drop response: empty response to asynchronous input"); CAF_LOG_DEBUG("drop response: empty response to asynchronous input");
} else if (!stages.empty()) { } else if (!stages.empty()) {
auto next = std::move(stages.back()); auto next = std::move(stages.back());
stages.pop_back(); stages.pop_back();
detail::profiled_send(self, std::move(source), next, id, std::move(stages), detail::profiled_send(selfptr, std::move(source), next, id,
self->context(), std::move(msg)); std::move(stages), selfptr->context(),
std::move(msg));
} else if (source != nullptr) { } else if (source != nullptr) {
detail::profiled_send(self, self->ctrl(), source, id.response_id(), detail::profiled_send(selfptr, self, source, id.response_id(),
forwarding_stack{}, self->context(), std::move(msg)); forwarding_stack{}, selfptr->context(),
std::move(msg));
} }
cancel(); cancel();
} }
...@@ -188,9 +189,10 @@ void response_promise::state::delegate_impl(abstract_actor* receiver, ...@@ -188,9 +189,10 @@ void response_promise::state::delegate_impl(abstract_actor* receiver,
message msg) { message msg) {
CAF_LOG_TRACE(CAF_ARG(msg)); CAF_LOG_TRACE(CAF_ARG(msg));
if (receiver != nullptr) { if (receiver != nullptr) {
auto self = static_cast<local_actor*>(weak_self.get()->get()); auto selfptr = static_cast<local_actor*>(self->get());
detail::profiled_send(self, std::move(source), receiver, id, detail::profiled_send(selfptr, std::move(source), receiver, id,
std::move(stages), self->context(), std::move(msg)); std::move(stages), selfptr->context(),
std::move(msg));
} else { } else {
CAF_LOG_DEBUG("drop response: invalid delegation target"); CAF_LOG_DEBUG("drop response: invalid delegation target");
} }
......
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