Commit 1c0e6196 authored by Dominik Charousset's avatar Dominik Charousset

Fix CPU spikes for joining many groups, close #371

parent 585cd8b5
...@@ -23,79 +23,22 @@ ...@@ -23,79 +23,22 @@
#include <string> #include <string>
#include <memory> #include <memory>
#include "caf/fwd.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/attachable.hpp" #include "caf/attachable.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/abstract_channel.hpp" #include "caf/abstract_channel.hpp"
namespace caf { namespace caf {
namespace detail {
class group_manager;
class peer_connection;
} // namespace detail
} // namespace caf
namespace caf {
class group;
class serializer;
class local_actor;
class deserializer;
/// A multicast group. /// A multicast group.
class abstract_group : public abstract_channel { class abstract_group : public abstract_channel {
public: public:
friend class detail::group_manager;
friend class detail::peer_connection;
friend class local_actor; friend class local_actor;
~abstract_group();
class subscription;
struct subscription_token {
intrusive_ptr<abstract_group> group;
static constexpr size_t token_type = attachable::token::subscription;
};
class subscription_predicate {
public:
inline subscription_predicate(intrusive_ptr<abstract_group> group)
: group_(std::move(group)) {
// nop
}
inline bool operator()(const attachable_ptr& ptr) {
return ptr->matches(subscription_token{group_});
}
private:
intrusive_ptr<abstract_group> group_;
};
// needs access to unsubscribe()
friend class subscription; friend class subscription;
friend class detail::group_manager;
// unsubscribes its channel from the group on destruction ~abstract_group();
class subscription : public attachable {
public:
subscription(const intrusive_ptr<abstract_group>& g);
void actor_exited(abstract_actor* self, uint32_t reason) override;
bool matches(const token& what) override;
static inline attachable_ptr make(intrusive_ptr<abstract_group> ptr) {
return attachable_ptr{new subscription(std::move(ptr))};
}
const intrusive_ptr<abstract_group>& group() const {
return group_;
}
private:
intrusive_ptr<abstract_group> group_;
};
/// Interface for user-defined multicast implementations. /// Interface for user-defined multicast implementations.
class module { class module {
...@@ -135,16 +78,19 @@ public: ...@@ -135,16 +78,19 @@ public:
/// Returns the name of the module. /// Returns the name of the module.
const std::string& module_name() const; const std::string& module_name() const;
/// Subscribes `who` to this group and returns a subscription object. /// Subscribes `who` to this group and returns `true` on success
virtual attachable_ptr subscribe(const actor_addr& who) = 0; /// or `false` if `who` is already subscribed.
virtual bool subscribe(const actor_addr& who) = 0;
/// Stops any background actors or threads and IO handles. /// Stops any background actors or threads and IO handles.
virtual void stop() = 0; virtual void stop() = 0;
protected: protected:
abstract_group(module_ptr module, std::string group_id, const node_id& nid); abstract_group(module_ptr module, std::string group_id, const node_id& nid);
// called by subscription objects
// called by local_actor
virtual void unsubscribe(const actor_addr& who) = 0; virtual void unsubscribe(const actor_addr& who) = 0;
module_ptr module_; module_ptr module_;
std::string identifier_; std::string identifier_;
}; };
......
...@@ -667,6 +667,9 @@ protected: ...@@ -667,6 +667,9 @@ protected:
// used by functor-based actors to implemented make_behavior() or act() // used by functor-based actors to implemented make_behavior() or act()
std::function<behavior (local_actor*)> initial_behavior_fac_; std::function<behavior (local_actor*)> initial_behavior_fac_;
// used for group management
std::set<group> subscriptions_;
/// @endcond /// @endcond
private: private:
......
...@@ -27,26 +27,6 @@ ...@@ -27,26 +27,6 @@
namespace caf { namespace caf {
abstract_group::subscription::subscription(const abstract_group_ptr& g)
: group_(g) {
// nop
}
void abstract_group::subscription::actor_exited(abstract_actor* ptr, uint32_t) {
group_->unsubscribe(ptr->address());
}
bool abstract_group::subscription::matches(const token& what) {
if (what.subtype != attachable::token::subscription) {
return false;
}
if (what.ptr) {
auto& ot = *reinterpret_cast<const subscription_token*>(what.ptr);
return ot.group == group_;
}
return true;
}
abstract_group::module::module(std::string mname) : name_(std::move(mname)) { abstract_group::module::module(std::string mname) : name_(std::move(mname)) {
// nop // nop
} }
......
...@@ -108,12 +108,11 @@ public: ...@@ -108,12 +108,11 @@ public:
return {success, subscribers_.size()}; return {success, subscribers_.size()};
} }
attachable_ptr subscribe(const actor_addr& who) override { bool subscribe(const actor_addr& who) override {
CAF_LOG_TRACE(""); // serializing who would cause a deadlock CAF_LOG_TRACE(""); // serializing who would cause a deadlock
if (add_subscriber(who).first) { if (add_subscriber(who).first)
return subscription::make(this); return true;
} return false;
return {};
} }
void unsubscribe(const actor_addr& who) override { void unsubscribe(const actor_addr& who) override {
...@@ -254,18 +253,17 @@ public: ...@@ -254,18 +253,17 @@ public:
monitor_ = spawn(broker_monitor_actor, this); monitor_ = spawn(broker_monitor_actor, this);
} }
attachable_ptr subscribe(const actor_addr& who) override { bool subscribe(const actor_addr& who) override {
CAF_LOG_TRACE(CAF_TSARG(who)); CAF_LOG_TRACE(CAF_TSARG(who));
auto res = add_subscriber(who); auto res = add_subscriber(who);
if (res.first) { if (res.first) {
if (res.second == 1) { // join remote source
// join the remote source if (res.second == 1)
anon_send(broker_, join_atom::value, proxy_broker_); anon_send(broker_, join_atom::value, proxy_broker_);
} return true;
return subscription::make(this);
} }
CAF_LOG_WARNING("actor already joined group"); CAF_LOG_WARNING("actor already joined group");
return {}; return false;
} }
void unsubscribe(const actor_addr& who) override { void unsubscribe(const actor_addr& who) override {
......
...@@ -67,27 +67,16 @@ void local_actor::demonitor(const actor_addr& whom) { ...@@ -67,27 +67,16 @@ void local_actor::demonitor(const actor_addr& whom) {
void local_actor::join(const group& what) { void local_actor::join(const group& what) {
CAF_LOG_TRACE(CAF_TSARG(what)); CAF_LOG_TRACE(CAF_TSARG(what));
if (what == invalid_group) { if (what == invalid_group)
return; return;
} if (what->subscribe(address()))
abstract_group::subscription_token tk{what.ptr()}; subscriptions_.emplace(what);
std::unique_lock<std::mutex> guard{mtx_};
if (detach_impl(tk, attachables_head_, true, true) == 0) {
auto ptr = what->subscribe(address());
if (ptr) {
attach_impl(ptr);
}
}
} }
void local_actor::leave(const group& what) { void local_actor::leave(const group& what) {
CAF_LOG_TRACE(CAF_TSARG(what)); CAF_LOG_TRACE(CAF_TSARG(what));
if (what == invalid_group) { if (subscriptions_.erase(what) > 0)
return;
}
if (detach(abstract_group::subscription_token{what.ptr()}) > 0) {
what->unsubscribe(address()); what->unsubscribe(address());
}
} }
void local_actor::on_exit() { void local_actor::on_exit() {
...@@ -96,15 +85,8 @@ void local_actor::on_exit() { ...@@ -96,15 +85,8 @@ void local_actor::on_exit() {
std::vector<group> local_actor::joined_groups() const { std::vector<group> local_actor::joined_groups() const {
std::vector<group> result; std::vector<group> result;
result.reserve(20); for (auto& x : subscriptions_)
attachable::token stk{attachable::token::subscription, nullptr}; result.emplace_back(x);
std::unique_lock<std::mutex> guard{mtx_};
for (attachable* i = attachables_head_.get(); i != 0; i = i->next.get()) {
if (i->matches(stk)) {
auto ptr = static_cast<abstract_group::subscription*>(i);
result.emplace_back(ptr->group());
}
}
return result; return result;
} }
...@@ -984,6 +966,12 @@ void local_actor::cleanup(uint32_t reason) { ...@@ -984,6 +966,12 @@ void local_actor::cleanup(uint32_t reason) {
detail::sync_request_bouncer f{reason}; detail::sync_request_bouncer f{reason};
mailbox_.close(f); mailbox_.close(f);
pending_responses_.clear(); pending_responses_.clear();
{ // lifetime scope of temporary
actor_addr me = address();
for (auto& subscription : subscriptions_)
subscription->unsubscribe(me);
subscriptions_.clear();
}
abstract_actor::cleanup(reason); abstract_actor::cleanup(reason);
// tell registry we're done // tell registry we're done
is_registered(false); is_registered(false);
......
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