Commit fbce2bbe authored by Dominik Charousset's avatar Dominik Charousset

Use attachables for subscriptions, relates #180

parent 8656db86
......@@ -114,6 +114,11 @@ class abstract_actor : public abstract_channel {
*/
void detach(const attachable::token& what);
template <class T>
void detach(const T& what) {
return detach(attachable::token{typeid(T), &what});
}
/**
* Links this actor to `whom`.
*/
......@@ -250,7 +255,7 @@ class abstract_actor : public abstract_channel {
std::atomic<uint32_t> m_exit_reason;
// guards access to m_exit_reason, m_attachables, and m_links
std::mutex m_mtx;
mutable std::mutex m_mtx;
// attached functors that are executed on cleanup (for monitors, links, etc)
std::vector<attachable_ptr> m_attachables;
......
......@@ -23,7 +23,8 @@
#include <string>
#include <memory>
#include "caf/channel.hpp"
#include "caf/actor_addr.hpp"
#include "caf/attachable.hpp"
#include "caf/ref_counted.hpp"
#include "caf/abstract_channel.hpp"
......@@ -54,27 +55,44 @@ class abstract_group : public abstract_channel {
class subscription;
struct subscription_token {
intrusive_ptr<abstract_group> group;
};
class subscription_predicate {
public:
inline subscription_predicate(intrusive_ptr<abstract_group> group)
: m_group(std::move(group)) {
// nop
}
inline bool operator()(const attachable_ptr& ptr) {
return ptr->matches(subscription_token{m_group});
}
private:
intrusive_ptr<abstract_group> m_group;
};
// needs access to unsubscribe()
friend class subscription;
// unsubscribes its channel from the group on destruction
class subscription {
class subscription : public attachable {
public:
friend class abstract_group;
subscription(const subscription&) = delete;
subscription& operator=(const subscription&) = delete;
subscription() = default;
subscription(subscription&&) = default;
subscription(const channel& s, const intrusive_ptr<abstract_group>& g);
subscription(const intrusive_ptr<abstract_group>& g);
~subscription();
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))};
}
inline bool valid() const {
return (m_subscriber) && (m_group);
const intrusive_ptr<abstract_group>& group() const {
return m_group;
}
private:
channel m_subscriber;
intrusive_ptr<abstract_group> m_group;
};
......@@ -126,12 +144,12 @@ class abstract_group : public abstract_channel {
/**
* Subscribes `who` to this group and returns a subscription object.
*/
virtual subscription subscribe(const channel& who) = 0;
virtual attachable_ptr subscribe(const actor_addr& who) = 0;
protected:
abstract_group(module_ptr module, std::string group_id);
// called by subscription objects
virtual void unsubscribe(const channel& who) = 0;
virtual void unsubscribe(const actor_addr& who) = 0;
module_ptr m_module;
std::string m_identifier;
};
......
......@@ -87,11 +87,10 @@ class group : detail::comparable<group>,
/**
* Get a pointer to the group associated with
* `group_identifier` from the module `module_name`.
* `identifier` from the module `mod_name`.
* @threadsafe
*/
static group get(const std::string& module_name,
const std::string& group_identifier);
static group get(const std::string& mod_name, const std::string& identifier);
/**
* Returns an anonymous group.
......@@ -114,12 +113,12 @@ class group : detail::comparable<group>,
static abstract_group::module_ptr
get_module(const std::string& module_name);
private:
inline abstract_group* get() const { return m_ptr.get(); }
inline const abstract_group_ptr& ptr() const {
return m_ptr;
}
private:
abstract_group_ptr m_ptr;
};
} // namespace caf
......
......@@ -538,7 +538,7 @@ class local_actor : public extend<abstract_actor>::with<mixin::memory_cached> {
mailbox_element* m_current_node;
// {group => subscription} map of all joined groups
std::map<group, abstract_group::subscription> m_subscriptions;
// std::map<group, abstract_group::subscription> m_subscriptions;
// set by quit
uint32_t m_planned_exit_reason;
......
......@@ -27,16 +27,22 @@
namespace caf {
abstract_group::subscription::subscription(const channel& s,
const abstract_group_ptr& g)
: m_subscriber(s), m_group(g) {
abstract_group::subscription::subscription(const abstract_group_ptr& g)
: m_group(g) {
// nop
}
abstract_group::subscription::~subscription() {
if (valid()) {
m_group->unsubscribe(m_subscriber);
void abstract_group::subscription::actor_exited(abstract_actor* self,
uint32_t reason) {
m_group->unsubscribe(self->address());
}
bool abstract_group::subscription::matches(const token& what) {
if (what.subtype != typeid(subscription_token)) {
return false;
}
auto& ot = *reinterpret_cast<const subscription_token*>(what.ptr);
return ot.group == m_group;
}
abstract_group::module::module(std::string name) : m_name(std::move(name)) {
......
......@@ -30,8 +30,7 @@ channel::channel(const actor& other)
// nop
}
channel::channel(const group& other)
: m_ptr(actor_cast<abstract_channel_ptr>(other)) {
channel::channel(const group& other) : m_ptr(other.ptr()) {
// nop
}
......
......@@ -56,7 +56,8 @@ class local_group : public abstract_group {
<< CAF_TARG(msg, to_string));
shared_guard guard(m_mtx);
for (auto& s : m_subscribers) {
s->enqueue(sender, message_id::invalid, msg, host);
actor_cast<abstract_actor_ptr>(s)->enqueue(sender, message_id::invalid,
msg, host);
}
}
......@@ -68,8 +69,8 @@ class local_group : public abstract_group {
m_broker->enqueue(sender, message_id::invalid, msg, host);
}
std::pair<bool, size_t> add_subscriber(const channel& who) {
CAF_LOG_TRACE(CAF_TARG(who, to_string));
std::pair<bool, size_t> add_subscriber(const actor_addr& who) {
CAF_LOG_TRACE(""); // serializing who would cause a deadlock
exclusive_guard guard(m_mtx);
if (who && m_subscribers.insert(who).second) {
return {true, m_subscribers.size()};
......@@ -77,23 +78,23 @@ class local_group : public abstract_group {
return {false, m_subscribers.size()};
}
std::pair<bool, size_t> erase_subscriber(const channel& who) {
CAF_LOG_TRACE(CAF_TARG(who, to_string));
std::pair<bool, size_t> erase_subscriber(const actor_addr& who) {
CAF_LOG_TRACE(""); // serializing who would cause a deadlock
exclusive_guard guard(m_mtx);
auto success = m_subscribers.erase(who) > 0;
return {success, m_subscribers.size()};
}
abstract_group::subscription subscribe(const channel& who) {
CAF_LOG_TRACE(CAF_TARG(who, to_string));
attachable_ptr subscribe(const actor_addr& who) override {
CAF_LOG_TRACE(""); // serializing who would cause a deadlock
if (add_subscriber(who).first) {
return {who, this};
return subscription::make(this);
}
return {};
}
void unsubscribe(const channel& who) {
CAF_LOG_TRACE(CAF_TARG(who, to_string));
void unsubscribe(const actor_addr& who) override {
CAF_LOG_TRACE(""); // serializing who would cause a deadlock
erase_subscriber(who);
}
......@@ -107,7 +108,7 @@ class local_group : public abstract_group {
protected:
detail::shared_spinlock m_mtx;
std::set<channel> m_subscribers;
std::set<actor_addr> m_subscribers;
actor m_broker;
};
......@@ -203,22 +204,22 @@ class local_group_proxy : public local_group {
m_proxy_broker = spawn<proxy_broker, hidden>(this);
}
abstract_group::subscription subscribe(const channel& who) {
CAF_LOG_TRACE(CAF_TSARG(who));
attachable_ptr subscribe(const actor_addr& who) override {
CAF_LOG_TRACE(""); // serializing who would cause a deadlock
auto res = add_subscriber(who);
if (res.first) {
if (res.second == 1) {
// join the remote source
anon_send(m_broker, atom("JOIN"), m_proxy_broker);
}
return {who, this};
return subscription::make(this);
}
CAF_LOG_WARNING("channel " << to_string(who) << " already joined");
CAF_LOG_WARNING("actor already joined group");
return {};
}
void unsubscribe(const channel& who) {
CAF_LOG_TRACE(CAF_TSARG(who));
void unsubscribe(const actor_addr& who) override {
CAF_LOG_TRACE(""); // serializing who would cause a deadlock
auto res = erase_subscriber(who);
if (res.first && res.second == 0) {
// leave the remote source,
......
......@@ -63,22 +63,36 @@ void local_actor::on_exit() {
void local_actor::join(const group& what) {
CAF_LOG_TRACE(CAF_TSARG(what));
if (what && m_subscriptions.count(what) == 0) {
CAF_LOG_DEBUG("join group: " << to_string(what));
m_subscriptions.insert(std::make_pair(what, what->subscribe(this)));
if (what == invalid_group) {
return;
}
abstract_group::subscription_predicate pred{what.ptr()};
std::unique_lock<std::mutex> guard{m_mtx};
auto last = m_attachables.end();
auto i = std::find_if(m_attachables.begin(), last, pred);
if (i == last) {
auto ptr = what->subscribe(address());
if (ptr) {
m_attachables.emplace_back(std::move(ptr));
}
}
}
void local_actor::leave(const group& what) {
if (what) {
m_subscriptions.erase(what);
if (what == invalid_group) {
return;
}
detach(abstract_group::subscription_token{what.ptr()});
}
std::vector<group> local_actor::joined_groups() const {
std::vector<group> result;
for (auto& kvp : m_subscriptions) {
result.emplace_back(kvp.first);
std::unique_lock<std::mutex> guard{m_mtx};
for (auto& ptr : m_attachables) {
auto sptr = dynamic_cast<abstract_group::subscription*>(ptr.get());
if (sptr) {
result.emplace_back(sptr->group());
}
}
return result;
}
......@@ -145,7 +159,6 @@ response_promise local_actor::make_response_promise() {
void local_actor::cleanup(uint32_t reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
m_subscriptions.clear();
super::cleanup(reason);
// tell registry we're done
is_registered(false);
......
......@@ -32,6 +32,7 @@
#include "caf/string_algorithms.hpp"
#include "caf/group.hpp"
#include "caf/channel.hpp"
#include "caf/message.hpp"
#include "caf/announce.hpp"
#include "caf/duration.hpp"
......
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