Commit 6048c9b8 authored by neverlord's avatar neverlord

timeout handling

parent a7b756a2
...@@ -100,6 +100,13 @@ auto abstract_event_based_actor::handle_message(mailbox_element& node) -> handle ...@@ -100,6 +100,13 @@ auto abstract_event_based_actor::handle_message(mailbox_element& node) -> handle
void abstract_event_based_actor::resume(util::fiber*, scheduler::callback* cb) void abstract_event_based_actor::resume(util::fiber*, scheduler::callback* cb)
{ {
auto done_cb = [&]()
{
m_state.store(abstract_scheduled_actor::done);
m_loop_stack.clear();
on_exit();
cb->exec_done();
};
self.set(this); self.set(this);
try try
{ {
...@@ -137,7 +144,12 @@ void abstract_event_based_actor::resume(util::fiber*, scheduler::callback* cb) ...@@ -137,7 +144,12 @@ void abstract_event_based_actor::resume(util::fiber*, scheduler::callback* cb)
} }
case msg_handled: case msg_handled:
{ {
// try to match cached messages if (m_loop_stack.empty())
{
done_cb();
return;
}
// try to match cached messages before receiving new ones
auto i = m_cache.begin(); auto i = m_cache.begin();
while (i != m_cache.end() && !m_loop_stack.empty()) while (i != m_cache.end() && !m_loop_stack.empty())
{ {
...@@ -151,6 +163,11 @@ void abstract_event_based_actor::resume(util::fiber*, scheduler::callback* cb) ...@@ -151,6 +163,11 @@ void abstract_event_based_actor::resume(util::fiber*, scheduler::callback* cb)
case msg_handled: case msg_handled:
{ {
m_cache.erase(i); m_cache.erase(i);
if (m_loop_stack.empty())
{
done_cb();
return;
}
i = m_cache.begin(); i = m_cache.begin();
break; break;
} }
...@@ -181,10 +198,7 @@ void abstract_event_based_actor::resume(util::fiber*, scheduler::callback* cb) ...@@ -181,10 +198,7 @@ void abstract_event_based_actor::resume(util::fiber*, scheduler::callback* cb)
{ {
cleanup(exit_reason::unhandled_exception); cleanup(exit_reason::unhandled_exception);
} }
m_state.store(abstract_scheduled_actor::done); done_cb();
m_loop_stack.clear();
on_exit();
cb->exec_done();
} }
void abstract_event_based_actor::on_exit() void abstract_event_based_actor::on_exit()
......
...@@ -109,11 +109,12 @@ void yielding_actor::dequeue(behavior& bhvr) ...@@ -109,11 +109,12 @@ void yielding_actor::dequeue(behavior& bhvr)
auto& fun = bhvr.get_partial_function(); auto& fun = bhvr.get_partial_function();
if (bhvr.timeout().valid() == false) if (bhvr.timeout().valid() == false)
{ {
dequeue(bhvr.get_partial_function()); dequeue(fun);
return; return;
} }
if (m_invoke.invoke_from_cache(fun) == false) if (m_invoke.invoke_from_cache(fun) == false)
{ {
request_timeout(bhvr.timeout());
bool timeout_occured = false; bool timeout_occured = false;
for (;;) for (;;)
{ {
......
#define CPPA_VERBOSE_CHECK
#include <stack> #include <stack>
#include <chrono> #include <chrono>
#include <iostream> #include <iostream>
......
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