Commit 6d0030bb authored by Dominik Charousset's avatar Dominik Charousset

fixed timeout handling related issues

parent fdba22eb
......@@ -111,6 +111,13 @@ match_expr_convert(const Arg0& arg0, const Args&... args) {
return {match_expr_concat(arg0, args...)};
}
template<typename... Lhs, typename F>
inline behavior operator,(const match_expr<Lhs...>& lhs,
const timeout_definition<F>& rhs) {
return match_expr_convert(lhs, rhs);
}
} // namespace cppa
#endif // CPPA_BEHAVIOR_HPP
......@@ -70,9 +70,10 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> {
else {
get_scheduler()->delayed_send(this, d, atom("TIMEOUT"),
++m_active_timeout_id);
m_has_pending_timeout_request = true;
}
m_has_pending_timeout_request = true;
}
else m_has_pending_timeout_request = false;
}
void reset_timeout() {
......@@ -96,7 +97,8 @@ class abstract_scheduled_actor : public abstract_actor<scheduled_actor> {
}
inline bool waits_for_timeout(std::uint32_t timeout_id) {
return m_active_timeout_id == timeout_id;
return m_has_pending_timeout_request
&& m_active_timeout_id == timeout_id;
}
bool m_has_pending_timeout_request;
......
......@@ -68,6 +68,9 @@ class behavior_stack
// erases the last asynchronous message handler
void pop_async_back();
// erases the synchronous response handler associated with @p response_id
void erase(message_id_t response_id);
void push_back(behavior&& what,
message_id_t expected_response = message_id_t());
......
......@@ -306,8 +306,10 @@ class receive_policy {
template<class Client>
static inline void hm_cleanup(Client* client, sequential) {
client->m_current_node = &(client->m_dummy_node);
// we definitely don't have a pending timeout now
client->m_has_pending_timeout_request = false;
if (client->has_behavior()) {
client->request_timeout(client->get_behavior().timeout());
}
else client->m_has_pending_timeout_request = false;
}
template<class Client>
......@@ -355,6 +357,7 @@ class receive_policy {
# endif
hm_cleanup(client, policy);
client->mark_arrived(awaited_response);
client->remove_handler(awaited_response);
return hm_msg_handled;
}
return hm_cache_msg;
......
......@@ -46,6 +46,8 @@ namespace cppa { namespace detail {
template<class Derived, class Base>
class stacked_actor_mixin : public Base {
friend class receive_policy;
public:
virtual void unbecome() {
......@@ -121,6 +123,12 @@ class stacked_actor_mixin : public Base {
}
}
inline void remove_handler(message_id_t id) {
if (m_bhvr_stack_ptr) {
m_bhvr_stack_ptr->erase(id);
}
}
};
} } // namespace cppa::detail
......
......@@ -138,12 +138,16 @@ class event_based_actor : public detail::abstract_scheduled_actor {
private:
inline bool has_behavior() const {
return m_bhvr_stack.empty() == false;
}
inline behavior& get_behavior() {
CPPA_REQUIRE(m_bhvr_stack.empty() == false);
return m_bhvr_stack.back();
}
// required by detail::nestable_receive_policy
// required by detail::nestable_receive_policy
static const detail::receive_policy_flag receive_flag = detail::rp_sequential;
inline void handle_timeout(behavior& bhvr) {
CPPA_REQUIRE(bhvr.timeout().valid());
......@@ -153,6 +157,9 @@ class event_based_actor : public detail::abstract_scheduled_actor {
request_timeout(get_behavior().timeout());
}
}
inline void remove_handler(message_id_t id) {
m_bhvr_stack.erase(id);
}
// stack elements are moved to m_erased_stack_elements and erased later
// to prevent possible segfaults that can occur if a currently executed
......
......@@ -81,10 +81,16 @@ void behavior_stack::pop_async_back() {
m_elements.erase(i);
}
}
}
if (m_elements.empty() == false) {
m_erased_elements.emplace_back(std::move(m_elements.back().first));
m_elements.pop_back();
void behavior_stack::erase(message_id_t response_id) {
auto last = m_elements.end();
auto i = std::find_if(m_elements.begin(), last, [=](element_type& e) {
return e.second == response_id;
});
if (i != last) {
m_erased_elements.emplace_back(std::move(i->first));
m_elements.erase(i);
}
}
......
......@@ -103,6 +103,7 @@ void event_based_actor::do_become(behavior&& bhvr, bool discard_old) {
}
void event_based_actor::become_waiting_for(behavior&& bhvr, message_future mf) {
CPPA_REQUIRE(bhvr.timeout().valid());
reset_timeout();
request_timeout(bhvr.timeout());
m_bhvr_stack.push_back(std::move(bhvr), mf);
......
......@@ -627,6 +627,12 @@ size_t test__spawn() {
);
await_all_others_done();
receive_response(sync_send(sync_testee, "!?")) (
others() >> [&]() {
CPPA_ERROR("'sync_testee' still alive?");
},
after(std::chrono::milliseconds(5)) >> []() { }
);
auto inflater = factory::event_based(
[](std::string*, actor_ptr* receiver) {
......
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