Commit bd1abea1 authored by Dominik Charousset's avatar Dominik Charousset

Add actor_clock::cancel_all function

parent 188be50c
...@@ -55,15 +55,16 @@ public: ...@@ -55,15 +55,16 @@ public:
/// Schedules a `timeout_msg` for `self` at time point `t`, overriding any /// Schedules a `timeout_msg` for `self` at time point `t`, overriding any
/// previous receive timeout. /// previous receive timeout.
virtual void set_receive_timeout(time_point t, abstract_actor* self, virtual void set_ordinary_timeout(time_point t, abstract_actor* self,
uint32_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`.
virtual void set_request_timeout(time_point t, abstract_actor* self, virtual void set_request_timeout(time_point t, abstract_actor* self,
message_id id) = 0; message_id id) = 0;
/// Cancels a pending receive timeout. /// Cancels a pending receive timeout.
virtual void cancel_receive_timeout(abstract_actor* self) = 0; virtual void cancel_ordinary_timeout(abstract_actor* self,
atom_value type) = 0;
/// Cancels the pending request timeout for `id`. /// Cancels the pending request timeout for `id`.
virtual void cancel_request_timeout(abstract_actor* self, message_id id) = 0; virtual void cancel_request_timeout(abstract_actor* self, message_id id) = 0;
...@@ -78,6 +79,9 @@ public: ...@@ -78,6 +79,9 @@ public:
/// Schedules an arbitrary message to `target` for time point `t`. /// Schedules an arbitrary message to `target` for time point `t`.
virtual void schedule_message(time_point t, group target, virtual void schedule_message(time_point t, group target,
strong_actor_ptr sender, message content) = 0; strong_actor_ptr sender, message content) = 0;
/// Cancels all timeouts and scheduled messages.
virtual void cancel_all() = 0;
}; };
} // namespace caf } // namespace caf
......
...@@ -37,9 +37,10 @@ public: ...@@ -37,9 +37,10 @@ public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
/// Request for a `timeout_msg`. /// Request for a `timeout_msg`.
struct receive_timeout { struct ordinary_timeout {
strong_actor_ptr self; strong_actor_ptr self;
uint32_t id; atom_value type;
uint64_t id;
}; };
/// Request for a `sec::request_timeout` error. /// Request for a `sec::request_timeout` error.
...@@ -61,14 +62,15 @@ public: ...@@ -61,14 +62,15 @@ public:
message content; message content;
}; };
using value_type = variant<receive_timeout, request_timeout, using value_type = variant<ordinary_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>;
using secondary_map = std::multimap<abstract_actor*, map_type::iterator>; using secondary_map = std::multimap<abstract_actor*, map_type::iterator>;
struct receive_predicate { struct ordinary_predicate {
atom_value type;
bool operator()(const secondary_map::value_type& x) const noexcept; bool operator()(const secondary_map::value_type& x) const noexcept;
}; };
...@@ -80,7 +82,7 @@ public: ...@@ -80,7 +82,7 @@ public:
struct visitor { struct visitor {
simple_actor_clock* thisptr; simple_actor_clock* thisptr;
void operator()(receive_timeout& x); void operator()(ordinary_timeout& x);
void operator()(request_timeout& x); void operator()(request_timeout& x);
...@@ -89,13 +91,13 @@ public: ...@@ -89,13 +91,13 @@ public:
void operator()(group_msg& x); void operator()(group_msg& x);
}; };
void set_receive_timeout(time_point t, abstract_actor* self, void set_ordinary_timeout(time_point t, abstract_actor* self,
uint32_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,
message_id id) override; message_id id) override;
void cancel_receive_timeout(abstract_actor* self) override; void cancel_ordinary_timeout(abstract_actor* self, atom_value type) override;
void cancel_request_timeout(abstract_actor* self, message_id id) override; void cancel_request_timeout(abstract_actor* self, message_id id) override;
...@@ -107,6 +109,8 @@ public: ...@@ -107,6 +109,8 @@ public:
void schedule_message(time_point t, group target, strong_actor_ptr sender, void schedule_message(time_point t, group target, strong_actor_ptr sender,
message content) override; message content) override;
void cancel_all() override;
inline const map_type& schedule() const { inline const map_type& schedule() const {
return schedule_; return schedule_;
} }
...@@ -117,7 +121,8 @@ public: ...@@ -117,7 +121,8 @@ public:
protected: protected:
template <class Predicate> template <class Predicate>
secondary_map::iterator lookup(abstract_actor* self, Predicate pred) { secondary_map::iterator lookup(abstract_actor* self,
Predicate pred) {
auto e = actor_lookup_.end(); auto e = actor_lookup_.end();
auto range = actor_lookup_.equal_range(self); auto range = actor_lookup_.equal_range(self);
if (range.first == range.second) if (range.first == range.second)
...@@ -142,8 +147,10 @@ protected: ...@@ -142,8 +147,10 @@ protected:
actor_lookup_.erase(i); actor_lookup_.erase(i);
} }
/// Timeout schedule.
map_type schedule_; map_type schedule_;
/// Secondary index for accessing timeouts by actor.
secondary_map actor_lookup_; secondary_map actor_lookup_;
}; };
......
...@@ -34,13 +34,13 @@ public: ...@@ -34,13 +34,13 @@ public:
thread_safe_actor_clock(); thread_safe_actor_clock();
void set_receive_timeout(time_point t, abstract_actor* self, void set_ordinary_timeout(time_point t, abstract_actor* self,
uint32_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,
message_id id) override; message_id id) override;
void cancel_receive_timeout(abstract_actor* self) override; void cancel_ordinary_timeout(abstract_actor* self, atom_value type) override;
void cancel_request_timeout(abstract_actor* self, message_id id) override; void cancel_request_timeout(abstract_actor* self, message_id id) override;
...@@ -52,6 +52,8 @@ public: ...@@ -52,6 +52,8 @@ public:
void schedule_message(time_point t, group target, strong_actor_ptr sender, void schedule_message(time_point t, group target, strong_actor_ptr sender,
message content) override; message content) override;
void cancel_all() override;
void run_dispatch_loop(); void run_dispatch_loop();
void cancel_dispatch_loop(); void cancel_dispatch_loop();
......
...@@ -25,25 +25,23 @@ ...@@ -25,25 +25,23 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
bool simple_actor_clock::receive_predicate:: bool simple_actor_clock::ordinary_predicate::
operator()(const secondary_map::value_type& x) const noexcept { operator()(const secondary_map::value_type& x) const noexcept {
return holds_alternative<receive_timeout>(x.second->second); auto ptr = get_if<ordinary_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 {
if (holds_alternative<request_timeout>(x.second->second)) { auto ptr = get_if<request_timeout>(&x.second->second);
auto& rt = get<request_timeout>(x.second->second); return ptr != nullptr ? ptr->id == id : false;
return rt.id == id;
}
return false;
} }
void simple_actor_clock::visitor::operator()(receive_timeout& x) { void simple_actor_clock::visitor::operator()(ordinary_timeout& x) {
CAF_ASSERT(x.self != nullptr); CAF_ASSERT(x.self != nullptr);
x.self->get()->eq_impl(make_message_id(), x.self, nullptr, x.self->get()->eq_impl(make_message_id(), x.self, nullptr,
timeout_msg{x.id}); timeout_msg{x.type, x.id});
receive_predicate pred; ordinary_predicate pred{x.type};
thisptr->drop_lookup(x.self->get(), pred); thisptr->drop_lookup(x.self->get(), pred);
} }
...@@ -63,16 +61,17 @@ void simple_actor_clock::visitor::operator()(group_msg& x) { ...@@ -63,16 +61,17 @@ void simple_actor_clock::visitor::operator()(group_msg& x) {
std::move(x.content)); std::move(x.content));
} }
void simple_actor_clock::set_receive_timeout(time_point t, abstract_actor* self, void simple_actor_clock::set_ordinary_timeout(time_point t, abstract_actor* self,
uint32_t id) { atom_value type, uint64_t id) {
receive_predicate pred; ordinary_predicate pred{type};
auto i = lookup(self, pred); auto i = lookup(self, pred);
auto sptr = actor_cast<strong_actor_ptr>(self); auto sptr = actor_cast<strong_actor_ptr>(self);
ordinary_timeout tmp{std::move(sptr), type, id};
if (i != actor_lookup_.end()) { if (i != actor_lookup_.end()) {
schedule_.erase(i->second); schedule_.erase(i->second);
i->second = schedule_.emplace(t, receive_timeout{std::move(sptr), id}); i->second = schedule_.emplace(t, std::move(tmp));
} else { } else {
auto j = schedule_.emplace(t, receive_timeout{std::move(sptr), id}); auto j = schedule_.emplace(t, std::move(tmp));
actor_lookup_.emplace(self, j); actor_lookup_.emplace(self, j);
} }
} }
...@@ -82,17 +81,19 @@ void simple_actor_clock::set_request_timeout(time_point t, abstract_actor* self, ...@@ -82,17 +81,19 @@ void simple_actor_clock::set_request_timeout(time_point t, abstract_actor* self,
request_predicate pred{id}; request_predicate pred{id};
auto i = lookup(self, pred); auto i = lookup(self, pred);
auto sptr = actor_cast<strong_actor_ptr>(self); auto sptr = actor_cast<strong_actor_ptr>(self);
request_timeout tmp{std::move(sptr), id};
if (i != actor_lookup_.end()) { if (i != actor_lookup_.end()) {
schedule_.erase(i->second); schedule_.erase(i->second);
i->second = schedule_.emplace(t, request_timeout{std::move(sptr), id}); i->second = schedule_.emplace(t, std::move(tmp));
} else { } else {
auto j = schedule_.emplace(t, request_timeout{std::move(sptr), id}); auto j = schedule_.emplace(t, std::move(tmp));
actor_lookup_.emplace(self, j); actor_lookup_.emplace(self, j);
} }
} }
void simple_actor_clock::cancel_receive_timeout(abstract_actor* self) { void simple_actor_clock::cancel_ordinary_timeout(abstract_actor* self,
receive_predicate pred; atom_value type) {
ordinary_predicate pred{type};
cancel(self, pred); cancel(self, pred);
} }
...@@ -124,5 +125,10 @@ void simple_actor_clock::schedule_message(time_point t, group target, ...@@ -124,5 +125,10 @@ void simple_actor_clock::schedule_message(time_point t, group target,
t, group_msg{std::move(target), std::move(sender), std::move(content)}); t, group_msg{std::move(target), std::move(sender), std::move(content)});
} }
void simple_actor_clock::cancel_all() {
actor_lookup_.clear();
schedule_.clear();
}
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
...@@ -31,11 +31,12 @@ thread_safe_actor_clock::thread_safe_actor_clock() : done_(false) { ...@@ -31,11 +31,12 @@ thread_safe_actor_clock::thread_safe_actor_clock() : done_(false) {
// nop // nop
} }
void thread_safe_actor_clock::set_receive_timeout(time_point t, void thread_safe_actor_clock::set_ordinary_timeout(time_point t,
abstract_actor* self, abstract_actor* self,
uint32_t id) { atom_value type,
uint64_t id) {
guard_type guard{mx_}; guard_type guard{mx_};
super::set_receive_timeout(t, self, id); super::set_ordinary_timeout(t, self, type, id);
cv_.notify_all(); cv_.notify_all();
} }
...@@ -47,9 +48,10 @@ void thread_safe_actor_clock::set_request_timeout(time_point t, ...@@ -47,9 +48,10 @@ void thread_safe_actor_clock::set_request_timeout(time_point t,
cv_.notify_all(); cv_.notify_all();
} }
void thread_safe_actor_clock::cancel_receive_timeout(abstract_actor* self) { void thread_safe_actor_clock::cancel_ordinary_timeout(abstract_actor* self,
atom_value type) {
guard_type guard{mx_}; guard_type guard{mx_};
super::cancel_receive_timeout(self); super::cancel_ordinary_timeout(self, type);
cv_.notify_all(); cv_.notify_all();
} }
...@@ -83,6 +85,12 @@ void thread_safe_actor_clock::schedule_message(time_point t, group target, ...@@ -83,6 +85,12 @@ void thread_safe_actor_clock::schedule_message(time_point t, group target,
cv_.notify_all(); cv_.notify_all();
} }
void thread_safe_actor_clock::cancel_all() {
guard_type guard{mx_};
super::cancel_all();
cv_.notify_all();
}
void thread_safe_actor_clock::run_dispatch_loop() { void thread_safe_actor_clock::run_dispatch_loop() {
visitor f{this}; visitor f{this};
guard_type guard{mx_}; guard_type guard{mx_};
......
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