Commit 425aa491 authored by Dominik Charousset's avatar Dominik Charousset

fixed event-based message processing for context-switching and thread-mapped actors

parent d04af15c
...@@ -111,6 +111,7 @@ class behavior_stack ...@@ -111,6 +111,7 @@ class behavior_stack
void exec(Policy& policy, Client* client) { void exec(Policy& policy, Client* client) {
while (!empty()) { while (!empty()) {
invoke(policy, client, client->receive_node()); invoke(policy, client, client->receive_node());
cleanup();
} }
} }
......
...@@ -282,8 +282,9 @@ class receive_policy { ...@@ -282,8 +282,9 @@ class receive_policy {
} }
template<class Client> template<class Client>
static inline void hm_cleanup(Client* client, nestable) { static inline void hm_cleanup(Client* client, pointer previous, nestable) {
client->m_current_node = &(client->m_dummy_node); client->m_current_node->marked = false;
client->m_current_node = previous;
} }
template<class Client> template<class Client>
...@@ -308,7 +309,7 @@ class receive_policy { ...@@ -308,7 +309,7 @@ class receive_policy {
} }
template<class Client> template<class Client>
static inline void hm_cleanup(Client* client, sequential) { static inline void hm_cleanup(Client* client, pointer /*previous*/, sequential) {
client->m_current_node = &(client->m_dummy_node); client->m_current_node = &(client->m_dummy_node);
if (client->has_behavior()) { if (client->has_behavior()) {
client->request_timeout(client->get_behavior().timeout()); client->request_timeout(client->get_behavior().timeout());
...@@ -347,12 +348,11 @@ class receive_policy { ...@@ -347,12 +348,11 @@ class receive_policy {
client->mark_arrived(awaited_response); client->mark_arrived(awaited_response);
client->remove_handler(awaited_response); client->remove_handler(awaited_response);
} }
hm_cleanup(client, policy);
return hm_msg_handled; return hm_msg_handled;
} }
case sync_response: { case sync_response: {
if (awaited_response.valid() && node->mid == awaited_response) { if (awaited_response.valid() && node->mid == awaited_response) {
hm_begin(client, node, policy); auto previous_node = hm_begin(client, node, policy);
# ifdef CPPA_DEBUG # ifdef CPPA_DEBUG
if (!fun(node->msg)) { if (!fun(node->msg)) {
std::cerr << "WARNING: actor didn't handle a " std::cerr << "WARNING: actor didn't handle a "
...@@ -363,7 +363,7 @@ class receive_policy { ...@@ -363,7 +363,7 @@ class receive_policy {
# endif # endif
client->mark_arrived(awaited_response); client->mark_arrived(awaited_response);
client->remove_handler(awaited_response); client->remove_handler(awaited_response);
hm_cleanup(client, policy); hm_cleanup(client, previous_node, policy);
return hm_msg_handled; return hm_msg_handled;
} }
return hm_cache_msg; return hm_cache_msg;
...@@ -381,7 +381,7 @@ class receive_policy { ...@@ -381,7 +381,7 @@ class receive_policy {
id.response_id(), id.response_id(),
any_tuple()); any_tuple());
} }
hm_cleanup(client, policy); hm_cleanup(client, previous_node, policy);
return hm_msg_handled; return hm_msg_handled;
} }
// no match (restore client members) // no match (restore client members)
......
...@@ -109,6 +109,8 @@ class stacked_actor_mixin : public Base { ...@@ -109,6 +109,8 @@ class stacked_actor_mixin : public Base {
} }
void become_impl(behavior&& bhvr, bool discard_old, message_id_t mid) { void become_impl(behavior&& bhvr, bool discard_old, message_id_t mid) {
dthis()->reset_timeout();
dthis()->request_timeout(bhvr.timeout());
if (m_bhvr_stack_ptr) { if (m_bhvr_stack_ptr) {
if (discard_old) m_bhvr_stack_ptr->pop_async_back(); if (discard_old) m_bhvr_stack_ptr->pop_async_back();
m_bhvr_stack_ptr->push_back(std::move(bhvr), mid); m_bhvr_stack_ptr->push_back(std::move(bhvr), mid);
......
...@@ -111,6 +111,10 @@ class thread_mapped_actor : public detail::stacked_actor_mixin< ...@@ -111,6 +111,10 @@ class thread_mapped_actor : public detail::stacked_actor_mixin<
inline void initialized(bool value) { m_initialized = value; } inline void initialized(bool value) { m_initialized = value; }
// required by stacked_actor_mixin
inline void reset_timeout() { }
inline void request_timeout(const util::duration&) { }
protected: protected:
bool initialized(); bool initialized();
......
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