Commit 98ad5a3e authored by Joseph Noir's avatar Joseph Noir

Add separate implementation for multi timeouts

parent ef96bb17
...@@ -59,7 +59,7 @@ public: ...@@ -59,7 +59,7 @@ public:
atom_value type, uint64_t id) = 0; atom_value type, uint64_t id) = 0;
/// Schedules a `timeout_msg` for `self` at time point `t`. /// Schedules a `timeout_msg` for `self` at time point `t`.
virtual void add_ordinary_timeout(time_point t, abstract_actor* self, virtual void set_multi_timeout(time_point t, abstract_actor* self,
atom_value type, uint64_t id) = 0; atom_value type, uint64_t id) = 0;
/// Schedules a `sec::request_timeout` for `self` at time point `t`. /// Schedules a `sec::request_timeout` for `self` at time point `t`.
......
...@@ -42,6 +42,12 @@ public: ...@@ -42,6 +42,12 @@ public:
uint64_t id; uint64_t id;
}; };
struct multi_timeout {
strong_actor_ptr self;
atom_value type;
uint64_t id;
};
/// Request for a `sec::request_timeout` error. /// Request for a `sec::request_timeout` error.
struct request_timeout { struct request_timeout {
strong_actor_ptr self; strong_actor_ptr self;
...@@ -61,7 +67,7 @@ public: ...@@ -61,7 +67,7 @@ public:
message content; message content;
}; };
using value_type = variant<ordinary_timeout, request_timeout, using value_type = variant<ordinary_timeout, multi_timeout, request_timeout,
actor_msg, group_msg>; actor_msg, group_msg>;
using map_type = std::multimap<time_point, value_type>; using map_type = std::multimap<time_point, value_type>;
...@@ -73,6 +79,11 @@ public: ...@@ -73,6 +79,11 @@ public:
bool operator()(const secondary_map::value_type& x) const noexcept; bool operator()(const secondary_map::value_type& x) const noexcept;
}; };
struct multi_predicate {
atom_value type;
bool operator()(const secondary_map::value_type& x) const noexcept;
};
struct request_predicate { struct request_predicate {
message_id id; message_id id;
bool operator()(const secondary_map::value_type& x) const noexcept; bool operator()(const secondary_map::value_type& x) const noexcept;
...@@ -83,6 +94,8 @@ public: ...@@ -83,6 +94,8 @@ public:
void operator()(ordinary_timeout& x); void operator()(ordinary_timeout& x);
void operator()(multi_timeout& x);
void operator()(request_timeout& x); void operator()(request_timeout& x);
void operator()(actor_msg& x); void operator()(actor_msg& x);
...@@ -93,7 +106,7 @@ public: ...@@ -93,7 +106,7 @@ public:
void set_ordinary_timeout(time_point t, abstract_actor* self, void set_ordinary_timeout(time_point t, abstract_actor* self,
atom_value type, uint64_t id) override; atom_value type, uint64_t id) override;
void add_ordinary_timeout(time_point t, abstract_actor* self, void set_multi_timeout(time_point t, abstract_actor* self,
atom_value type, uint64_t id) override; atom_value type, uint64_t id) override;
void set_request_timeout(time_point t, abstract_actor* self, void set_request_timeout(time_point t, abstract_actor* self,
......
...@@ -31,6 +31,12 @@ operator()(const secondary_map::value_type& x) const noexcept { ...@@ -31,6 +31,12 @@ operator()(const secondary_map::value_type& x) const noexcept {
return ptr != nullptr ? ptr->type == type : false; return ptr != nullptr ? ptr->type == type : false;
} }
bool simple_actor_clock::multi_predicate::
operator()(const secondary_map::value_type& x) const noexcept {
auto ptr = get_if<multi_timeout>(&x.second->second);
return ptr != nullptr ? ptr->type == type : false;
}
bool simple_actor_clock::request_predicate:: bool simple_actor_clock::request_predicate::
operator()(const secondary_map::value_type& x) const noexcept { operator()(const secondary_map::value_type& x) const noexcept {
auto ptr = get_if<request_timeout>(&x.second->second); auto ptr = get_if<request_timeout>(&x.second->second);
...@@ -45,6 +51,14 @@ void simple_actor_clock::visitor::operator()(ordinary_timeout& x) { ...@@ -45,6 +51,14 @@ void simple_actor_clock::visitor::operator()(ordinary_timeout& x) {
thisptr->drop_lookup(x.self->get(), pred); thisptr->drop_lookup(x.self->get(), pred);
} }
void simple_actor_clock::visitor::operator()(multi_timeout& x) {
CAF_ASSERT(x.self != nullptr);
x.self->get()->eq_impl(make_message_id(), x.self, nullptr,
timeout_msg{x.type, x.id});
multi_predicate pred{x.type};
thisptr->drop_lookup(x.self->get(), pred);
}
void simple_actor_clock::visitor::operator()(request_timeout& x) { void simple_actor_clock::visitor::operator()(request_timeout& x) {
CAF_ASSERT(x.self != nullptr); CAF_ASSERT(x.self != nullptr);
x.self->get()->eq_impl(x.id, x.self, nullptr, sec::request_timeout); x.self->get()->eq_impl(x.id, x.self, nullptr, sec::request_timeout);
...@@ -76,10 +90,10 @@ void simple_actor_clock::set_ordinary_timeout(time_point t, abstract_actor* self ...@@ -76,10 +90,10 @@ void simple_actor_clock::set_ordinary_timeout(time_point t, abstract_actor* self
} }
} }
void simple_actor_clock::add_ordinary_timeout(time_point t, abstract_actor* self, void simple_actor_clock::set_multi_timeout(time_point t, abstract_actor* self,
atom_value type, uint64_t id) { atom_value type, uint64_t id) {
auto sptr = actor_cast<strong_actor_ptr>(self); auto sptr = actor_cast<strong_actor_ptr>(self);
ordinary_timeout tmp{std::move(sptr), type, id}; multi_timeout tmp{std::move(sptr), type, id};
auto j = schedule_.emplace(t, std::move(tmp)); auto j = schedule_.emplace(t, std::move(tmp));
actor_lookup_.emplace(self, j); actor_lookup_.emplace(self, j);
} }
......
...@@ -49,7 +49,7 @@ behavior testee(stateful_actor<testee_state, raw_event_based_actor>* self, ...@@ -49,7 +49,7 @@ behavior testee(stateful_actor<testee_state, raw_event_based_actor>* self,
[=](add_atom) { [=](add_atom) {
auto n = t->now() + seconds(10); auto n = t->now() + seconds(10);
self->state.timeout_id += 1; self->state.timeout_id += 1;
t->add_ordinary_timeout(n, self, atom(""), self->state.timeout_id); t->set_multi_timeout(n, self, atom(""), self->state.timeout_id);
}, },
[=](put_atom) { [=](put_atom) {
auto n = t->now() + seconds(10); auto n = t->now() + seconds(10);
...@@ -129,17 +129,44 @@ CAF_TEST(override_receive_timeout) { ...@@ -129,17 +129,44 @@ CAF_TEST(override_receive_timeout) {
expect((timeout_msg), from(aut).to(aut).with(tid{43})); expect((timeout_msg), from(aut).to(aut).with(tid{43}));
} }
CAF_TEST(add_receive_timeout) { CAF_TEST(multi_timeout) {
// Have AUT call t.set_receive_timeout(). // Have AUT call t.set_multi_timeout().
self->send(aut, add_atom::value); self->send(aut, add_atom::value);
expect((add_atom), from(self).to(aut).with(_)); expect((add_atom), from(self).to(aut).with(_));
CAF_CHECK_EQUAL(t.schedule().size(), 1u); CAF_CHECK_EQUAL(t.schedule().size(), 1u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 1u); CAF_CHECK_EQUAL(t.actor_lookup().size(), 1u);
// Advance time just a little bit. // Advance time just a little bit.
t.advance_time(seconds(5)); t.advance_time(seconds(5));
// Have AUT call t.set_timeout() again. // Have AUT call t.set_multi_timeout() again.
self->send(aut, add_atom::value);
expect((add_atom), from(self).to(aut).with(_));
CAF_CHECK_EQUAL(t.schedule().size(), 2u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 2u);
// Advance time to send timeout message.
t.advance_time(seconds(5));
CAF_CHECK_EQUAL(t.schedule().size(), 1u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 1u);
// Have AUT receive the timeout.
expect((timeout_msg), from(aut).to(aut).with(tid{42}));
// Advance time to send second timeout message.
t.advance_time(seconds(5));
CAF_CHECK_EQUAL(t.schedule().size(), 0u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 0u);
// Have AUT receive the timeout.
expect((timeout_msg), from(aut).to(aut).with(tid{43}));
}
CAF_TEST(mixed_receive_and_multi_timeouts) {
// Have AUT call t.set_receive_timeout().
self->send(aut, add_atom::value); self->send(aut, add_atom::value);
expect((add_atom), from(self).to(aut).with(_)); expect((add_atom), from(self).to(aut).with(_));
CAF_CHECK_EQUAL(t.schedule().size(), 1u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 1u);
// Advance time just a little bit.
t.advance_time(seconds(5));
// Have AUT call t.set_multi_timeout() again.
self->send(aut, ok_atom::value);
expect((ok_atom), from(self).to(aut).with(_));
CAF_CHECK_EQUAL(t.schedule().size(), 2u); CAF_CHECK_EQUAL(t.schedule().size(), 2u);
CAF_CHECK_EQUAL(t.actor_lookup().size(), 2u); CAF_CHECK_EQUAL(t.actor_lookup().size(), 2u);
// Advance time to send timeout message. // Advance time to send timeout message.
...@@ -154,6 +181,7 @@ CAF_TEST(add_receive_timeout) { ...@@ -154,6 +181,7 @@ CAF_TEST(add_receive_timeout) {
CAF_CHECK_EQUAL(t.actor_lookup().size(), 0u); CAF_CHECK_EQUAL(t.actor_lookup().size(), 0u);
// Have AUT receive the timeout. // Have AUT receive the timeout.
expect((timeout_msg), from(aut).to(aut).with(tid{43})); expect((timeout_msg), from(aut).to(aut).with(tid{43}));
} }
CAF_TEST(single_request_timeout) { CAF_TEST(single_request_timeout) {
......
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