Commit f930ce71 authored by Dominik Charousset's avatar Dominik Charousset

unified handling of timeout messages

parent 1a6cbb9a
...@@ -38,7 +38,7 @@ if (CMAKE_CXX_FLAGS) ...@@ -38,7 +38,7 @@ if (CMAKE_CXX_FLAGS)
set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "") set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "")
else (CMAKE_CXX_FLAGS) else (CMAKE_CXX_FLAGS)
set(CXXFLAGS_PROVIDED false) set(CXXFLAGS_PROVIDED false)
set(CMAKE_CXX_FLAGS "-std=c++11 -Wextra -Wall -pedantic -fno-omit-frame-pointer -fsanitize=address") set(CMAKE_CXX_FLAGS "-std=c++11 -Wextra -Wall -pedantic")
set(CMAKE_CXX_FLAGS_DEBUG "-O0 -g") set(CMAKE_CXX_FLAGS_DEBUG "-O0 -g")
set(CMAKE_CXX_FLAGS_MINSIZEREL "-Os") set(CMAKE_CXX_FLAGS_MINSIZEREL "-Os")
set(CMAKE_CXX_FLAGS_RELEASE "-O3 -DNDEBUG") set(CMAKE_CXX_FLAGS_RELEASE "-O3 -DNDEBUG")
......
...@@ -32,7 +32,9 @@ class proper_actor_base : public Policies::resume_policy::template mixin<Base, D ...@@ -32,7 +32,9 @@ class proper_actor_base : public Policies::resume_policy::template mixin<Base, D
public: public:
template <typename... Ts> template <typename... Ts>
proper_actor_base(Ts&&... args) : super(std::forward<Ts>(args)...) { } proper_actor_base(Ts&&... args)
: super(std::forward<Ts>(args)...), m_has_pending_tout(false)
, m_pending_tout(0) { }
// grant access to the actor's mailbox // grant access to the actor's mailbox
typename Base::mailbox_type& mailbox() { typename Base::mailbox_type& mailbox() {
...@@ -113,12 +115,50 @@ class proper_actor_base : public Policies::resume_policy::template mixin<Base, D ...@@ -113,12 +115,50 @@ class proper_actor_base : public Policies::resume_policy::template mixin<Base, D
awaited_response); awaited_response);
} }
// timeout handling
inline void reset_timeout() { inline void reset_timeout() {
invoke_policy().reset_timeout(); if (m_has_pending_tout) {
++m_pending_tout;
m_has_pending_tout = false;
}
}
void request_timeout(const util::duration& d) {
if (!d.valid()) m_has_pending_tout = false;
else {
auto msg = make_any_tuple(atom("SYNC_TOUT"), ++m_pending_tout);
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
dptr()->enqueue({this->address(), this}, std::move(msg));
//auto e = this->new_mailbox_element(this, std::move(msg));
//this->m_mailbox.enqueue(e);
}
else dptr()->delayed_send_tuple(this, d, std::move(msg));
m_has_pending_tout = true;
}
}
inline void handle_timeout(behavior& bhvr) {
bhvr.handle_timeout();
reset_timeout();
}
inline void pop_timeout() {
CPPA_REQUIRE(m_pending_tout > 0);
--m_pending_tout;
}
inline void push_timeout() {
++m_pending_tout;
}
inline bool waits_for_timeout(std::uint32_t timeout_id) const {
return m_has_pending_tout && m_pending_tout == timeout_id;
} }
inline void request_timeout(const util::duration& d) { inline bool has_pending_timeout() const {
invoke_policy().request_timeout(dptr(), d); return m_has_pending_tout;
} }
protected: protected:
...@@ -146,6 +186,8 @@ class proper_actor_base : public Policies::resume_policy::template mixin<Base, D ...@@ -146,6 +186,8 @@ class proper_actor_base : public Policies::resume_policy::template mixin<Base, D
private: private:
Policies m_policies; Policies m_policies;
bool m_has_pending_tout;
std::uint32_t m_pending_tout;
}; };
......
...@@ -70,43 +70,47 @@ class event_based_resume { ...@@ -70,43 +70,47 @@ class event_based_resume {
CPPA_PUSH_AID(d->id()); CPPA_PUSH_AID(d->id());
auto done_cb = [&]() -> bool { auto done_cb = [&]() -> bool {
CPPA_LOG_TRACE(""); CPPA_LOG_TRACE("");
if ( d->exit_reason() == exit_reason::not_exited d->bhvr_stack().clear();
&& d->planned_exit_reason() == exit_reason::not_exited) { d->bhvr_stack().cleanup();
if (d->planned_exit_reason() == exit_reason::not_exited) {
d->planned_exit_reason(exit_reason::normal); d->planned_exit_reason(exit_reason::normal);
} }
d->on_exit(); d->on_exit();
if (!d->bhvr_stack().empty()) { if (!d->bhvr_stack().empty()) {
CPPA_LOG_DEBUG("on_exit did set a new behavior");
d->planned_exit_reason(exit_reason::not_exited); d->planned_exit_reason(exit_reason::not_exited);
return false; // on_exit did set a new behavior return false; // on_exit did set a new behavior
} }
d->set_state(actor_state::done); d->set_state(actor_state::done);
d->bhvr_stack().clear();
d->bhvr_stack().cleanup();
d->cleanup(d->planned_exit_reason()); d->cleanup(d->planned_exit_reason());
return true; return true;
}; };
auto actor_done = [&] {
return d->bhvr_stack().empty()
|| d->planned_exit_reason() != exit_reason::not_exited;
};
try { try {
for (;;) { for (;;) {
auto ptr = dptr()->next_message(); auto ptr = d->next_message();
if (ptr) { if (ptr) {
CPPA_REQUIRE(!dptr()->bhvr_stack().empty()); CPPA_REQUIRE(!d->bhvr_stack().empty());
auto bhvr = dptr()->bhvr_stack().back(); auto bhvr = d->bhvr_stack().back();
auto mid = dptr()->bhvr_stack().back_id(); auto mid = d->bhvr_stack().back_id();
if (dptr()->invoke_message(ptr, bhvr, mid)) { if (d->invoke_message(ptr, bhvr, mid)) {
if (dptr()->bhvr_stack().empty() && done_cb()) { if (actor_done() && done_cb()) {
CPPA_LOG_DEBUG("behavior stack empty"); CPPA_LOG_DEBUG("actor exited");
return resume_result::done; return resume_result::done;
} }
} }
// add ptr to cache if invoke_message did not // add ptr to cache if invoke_message
// reset it // did not reset it
if (ptr) dptr()->push_to_cache(std::move(ptr)); if (ptr) d->push_to_cache(std::move(ptr));
} }
else { else {
CPPA_LOG_DEBUG("no more element in mailbox; going to block"); CPPA_LOG_DEBUG("no more element in mailbox; going to block");
d->set_state(actor_state::about_to_block); d->set_state(actor_state::about_to_block);
std::atomic_thread_fence(std::memory_order_seq_cst); std::atomic_thread_fence(std::memory_order_seq_cst);
if (!dptr()->has_next_message()) { if (!d->has_next_message()) {
switch (d->cas_state(actor_state::about_to_block, switch (d->cas_state(actor_state::about_to_block,
actor_state::blocked)) { actor_state::blocked)) {
case actor_state::ready: case actor_state::ready:
......
...@@ -127,8 +127,7 @@ class invoke_policy { ...@@ -127,8 +127,7 @@ class invoke_policy {
std::list<std::unique_ptr<mailbox_element, detail::disposer> > m_cache; std::list<std::unique_ptr<mailbox_element, detail::disposer> > m_cache;
template<class Actor> inline void handle_timeout(partial_function&) {
inline void handle_timeout(Actor*, partial_function&) {
CPPA_CRITICAL("handle_timeout(partial_function&)"); CPPA_CRITICAL("handle_timeout(partial_function&)");
} }
...@@ -170,7 +169,7 @@ class invoke_policy { ...@@ -170,7 +169,7 @@ class invoke_policy {
} }
else if (v0 == atom("SYNC_TOUT")) { else if (v0 == atom("SYNC_TOUT")) {
CPPA_REQUIRE(!mid.valid()); CPPA_REQUIRE(!mid.valid());
return dptr()->waits_for_timeout(v1) ? timeout_message return self->waits_for_timeout(v1) ? timeout_message
: expired_timeout_message; : expired_timeout_message;
} }
} }
...@@ -317,7 +316,7 @@ class invoke_policy { ...@@ -317,7 +316,7 @@ class invoke_policy {
} }
case timeout_message: { case timeout_message: {
CPPA_LOG_DEBUG("handle timeout message"); CPPA_LOG_DEBUG("handle timeout message");
dptr()->handle_timeout(self, fun); self->handle_timeout(fun);
if (awaited_response.valid()) { if (awaited_response.valid()) {
self->mark_arrived(awaited_response); self->mark_arrived(awaited_response);
self->remove_handler(awaited_response); self->remove_handler(awaited_response);
......
...@@ -64,7 +64,7 @@ class nestable_invoke : public invoke_policy<nestable_invoke> { ...@@ -64,7 +64,7 @@ class nestable_invoke : public invoke_policy<nestable_invoke> {
inline mailbox_element* hm_begin(Actor* self, mailbox_element* node) { inline mailbox_element* hm_begin(Actor* self, mailbox_element* node) {
auto previous = self->current_node(); auto previous = self->current_node();
self->current_node(node); self->current_node(node);
push_timeout(); self->push_timeout();
node->marked = true; node->marked = true;
return previous; return previous;
} }
...@@ -79,27 +79,11 @@ class nestable_invoke : public invoke_policy<nestable_invoke> { ...@@ -79,27 +79,11 @@ class nestable_invoke : public invoke_policy<nestable_invoke> {
inline void hm_revert(Actor* self, mailbox_element* previous) { inline void hm_revert(Actor* self, mailbox_element* previous) {
self->current_node()->marked = false; self->current_node()->marked = false;
self->current_node(previous); self->current_node(previous);
pop_timeout(); self->pop_timeout();
} }
typedef std::chrono::high_resolution_clock::time_point timeout_type; typedef std::chrono::high_resolution_clock::time_point timeout_type;
inline void reset_timeout() { }
template<class Actor>
inline void request_timeout(Actor*, const util::duration&) { }
template<class Actor>
inline void handle_timeout(Actor*, behavior& bhvr) {
bhvr.handle_timeout();
}
inline void pop_timeout() { }
inline void push_timeout() { }
inline bool waits_for_timeout(std::uint32_t) { return false; }
}; };
} } // namespace cppa::policy } } // namespace cppa::policy
......
...@@ -118,13 +118,6 @@ class no_scheduling { ...@@ -118,13 +118,6 @@ class no_scheduling {
await_data(self); await_data(self);
self->set_state(actor_state::ready); self->set_state(actor_state::ready);
if (self->resume(&fself) == resumable::done) { if (self->resume(&fself) == resumable::done) {
CPPA_LOG_DEBUG("resume returned resumable::done");
self->planned_exit_reason(exit_reason::normal);
}
auto per = self->planned_exit_reason();
if (per != exit_reason::not_exited) {
CPPA_LOG_DEBUG("planned exit reason: " << per);
self->cleanup(per);
return; return;
} }
} }
......
...@@ -47,8 +47,6 @@ class sequential_invoke : public invoke_policy<sequential_invoke> { ...@@ -47,8 +47,6 @@ class sequential_invoke : public invoke_policy<sequential_invoke> {
public: public:
sequential_invoke() : m_has_pending_tout(false), m_pending_tout(0) { }
static inline bool hm_should_skip(mailbox_element*) { static inline bool hm_should_skip(mailbox_element*) {
return false; return false;
} }
...@@ -70,57 +68,6 @@ class sequential_invoke : public invoke_policy<sequential_invoke> { ...@@ -70,57 +68,6 @@ class sequential_invoke : public invoke_policy<sequential_invoke> {
self->current_node(previous); self->current_node(previous);
} }
inline void reset_timeout() {
if (m_has_pending_tout) {
++m_pending_tout;
m_has_pending_tout = false;
}
}
template<class Actor>
void request_timeout(Actor* self, const util::duration& d) {
if (!d.valid()) m_has_pending_tout = false;
else {
auto msg = make_any_tuple(atom("SYNC_TOUT"), ++m_pending_tout);
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
self->enqueue({self->address(), self}, std::move(msg));
//auto e = this->new_mailbox_element(this, std::move(msg));
//this->m_mailbox.enqueue(e);
}
else self->delayed_send_tuple(self, d, std::move(msg));
m_has_pending_tout = true;
}
}
template<class Actor>
inline void handle_timeout(Actor*, behavior& bhvr) {
bhvr.handle_timeout();
reset_timeout();
}
inline void pop_timeout() {
CPPA_REQUIRE(m_pending_tout > 0);
--m_pending_tout;
}
inline void push_timeout() {
++m_pending_tout;
}
inline bool waits_for_timeout(std::uint32_t timeout_id) const {
return m_has_pending_tout && m_pending_tout == timeout_id;
}
inline bool has_pending_timeout() const {
return m_has_pending_tout;
}
private:
bool m_has_pending_tout;
std::uint32_t m_pending_tout;
}; };
} } // namespace cppa::policy } } // namespace cppa::policy
......
...@@ -56,19 +56,19 @@ class event_testee : public sb_actor<event_testee> { ...@@ -56,19 +56,19 @@ class event_testee : public sb_actor<event_testee> {
// quits after 5 timeouts // quits after 5 timeouts
actor spawn_event_testee2() { actor spawn_event_testee2() {
struct impl : sb_actor<impl> { struct impl : untyped_actor {
behavior wait4timeout(int remaining) { behavior wait4timeout(int remaining) {
CPPA_LOG_TRACE(CPPA_ARG(remaining));
return ( return (
after(chrono::milliseconds(50)) >> [=]() { after(chrono::milliseconds(50)) >> [=] {
if (remaining == 1) quit(); if (remaining == 1) quit();
else become(wait4timeout(remaining - 1)); else become(wait4timeout(remaining - 1));
} }
); );
} }
behavior make_behavior() override {
behavior init_state; return wait4timeout(5);
}
impl() : init_state(wait4timeout(5)) { }
}; };
return spawn<impl>(); return spawn<impl>();
} }
......
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