Commit e12717da authored by Dominik Charousset's avatar Dominik Charousset

Fix handling of sync/async timeouts, close #283

parent eebfb6de
...@@ -62,14 +62,21 @@ void blocking_actor::dequeue(behavior& bhvr, message_id mid) { ...@@ -62,14 +62,21 @@ void blocking_actor::dequeue(behavior& bhvr, message_id mid) {
return; return;
} }
// requesting an invalid timeout will reset our active timeout // requesting an invalid timeout will reset our active timeout
auto timeout_id = request_timeout(bhvr.timeout()); uint32_t timeout_id = 0;
if (mid == invalid_message_id) {
timeout_id = request_timeout(bhvr.timeout());
} else {
request_sync_timeout_msg(bhvr.timeout(), mid);
}
// read incoming messages // read incoming messages
for (;;) { for (;;) {
await_data(); await_data();
auto msg = next_message(); auto msg = next_message();
switch (invoke_message(msg, bhvr, mid)) { switch (invoke_message(msg, bhvr, mid)) {
case im_success: case im_success:
if (mid == invalid_message_id) {
reset_timeout(timeout_id); reset_timeout(timeout_id);
}
return; return;
case im_skipped: case im_skipped:
if (msg) { if (msg) {
......
...@@ -30,7 +30,6 @@ ...@@ -30,7 +30,6 @@
#include "caf/detail/logging.hpp" #include "caf/detail/logging.hpp"
#include "caf/detail/sync_request_bouncer.hpp" #include "caf/detail/sync_request_bouncer.hpp"
namespace caf { namespace caf {
// local actors are created with a reference count of one that is adjusted // local actors are created with a reference count of one that is adjusted
...@@ -152,6 +151,15 @@ uint32_t local_actor::request_timeout(const duration& d) { ...@@ -152,6 +151,15 @@ uint32_t local_actor::request_timeout(const duration& d) {
return result; return result;
} }
void local_actor::request_sync_timeout_msg(const duration& d, message_id mid) {
if (!d.valid()) {
return;
}
auto sched_cd = detail::singletons::get_scheduling_coordinator();
sched_cd->delayed_send(d, address(), this, mid,
make_message(sync_timeout_msg{}));
}
void local_actor::handle_timeout(behavior& bhvr, uint32_t timeout_id) { void local_actor::handle_timeout(behavior& bhvr, uint32_t timeout_id) {
if (!is_active_timeout(timeout_id)) { if (!is_active_timeout(timeout_id)) {
return; return;
...@@ -166,12 +174,6 @@ void local_actor::handle_timeout(behavior& bhvr, uint32_t timeout_id) { ...@@ -166,12 +174,6 @@ void local_actor::handle_timeout(behavior& bhvr, uint32_t timeout_id) {
m_bhvr_stack.pop_back(); m_bhvr_stack.pop_back();
return; return;
} }
// request next timeout for non-blocking (i.e. event-based) actors
// if behavior stack was not modified by calling become()/unbecome()
if (m_bhvr_stack.back() == bhvr) {
CAF_ASSERT(bhvr.timeout().valid());
request_timeout(bhvr.timeout());
}
} }
void local_actor::reset_timeout(uint32_t timeout_id) { void local_actor::reset_timeout(uint32_t timeout_id) {
...@@ -337,6 +339,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -337,6 +339,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
// by calling quit(...) // by calling quit(...)
return im_success; return im_success;
case msg_type::timeout: { case msg_type::timeout: {
if (awaited_id == invalid_message_id) {
CAF_LOG_DEBUG("handle timeout message"); CAF_LOG_DEBUG("handle timeout message");
auto& tm = ptr->msg.get_as<timeout_msg>(0); auto& tm = ptr->msg.get_as<timeout_msg>(0);
handle_timeout(fun, tm.timeout_id); handle_timeout(fun, tm.timeout_id);
...@@ -345,6 +348,10 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -345,6 +348,10 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
} }
return im_success; return im_success;
} }
// ignore "async" timeout
CAF_LOG_DEBUG("async timeout ignored while in sync mode");
return im_dropped;
}
case msg_type::sync_response: case msg_type::sync_response:
CAF_LOG_DEBUG("handle as synchronous response: " CAF_LOG_DEBUG("handle as synchronous response: "
<< CAF_TARG(ptr->msg, to_string) << ", " << CAF_TARG(ptr->msg, to_string) << ", "
...@@ -633,15 +640,12 @@ resumable::resume_result local_actor::resume(execution_unit* eu, ...@@ -633,15 +640,12 @@ resumable::resume_result local_actor::resume(execution_unit* eu,
return resumable::resume_result::done; return resumable::resume_result::done;
} }
} }
auto had_tout = has_timeout();
auto tout = active_timeout_id();
int handled_msgs = 0; int handled_msgs = 0;
auto reset_timeout_if_needed = [&] { auto reset_timeout_if_needed = [&] {
if (had_tout && handled_msgs > 0 && tout == active_timeout_id()) { if (handled_msgs > 0) {
request_timeout(get_behavior().timeout()); request_timeout(get_behavior().timeout());
} }
}; };
// max_throughput = 0 means infinite
for (size_t i = 0; i < max_throughput; ++i) { for (size_t i = 0; i < max_throughput; ++i) {
auto ptr = next_message(); auto ptr = next_message();
if (ptr) { if (ptr) {
...@@ -857,12 +861,6 @@ void local_actor::quit(uint32_t reason) { ...@@ -857,12 +861,6 @@ void local_actor::quit(uint32_t reason) {
} }
} }
void local_actor::request_sync_timeout_msg(const duration& dr, message_id mid) {
auto sched_cd = detail::singletons::get_scheduling_coordinator();
sched_cd->delayed_send(dr, address(), this, mid,
make_message(sync_timeout_msg{}));
}
// <backward_compatibility version="0.12"> // <backward_compatibility version="0.12">
message& local_actor::last_dequeued() { message& local_actor::last_dequeued() {
if (!m_current_element) { if (!m_current_element) {
......
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