Commit 999361a4 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'issue/1361'

Close #1361.
parents 0498f5c3 dc423f59
...@@ -38,6 +38,8 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -38,6 +38,8 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- When starting a new thread via CAF, the thread hooks API now receives an - When starting a new thread via CAF, the thread hooks API now receives an
additional tag that identifies the component responsible for launching the new additional tag that identifies the component responsible for launching the new
thread. thread.
- 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