Commit 7b51d55e authored by Dominik Charousset's avatar Dominik Charousset

Make proxy_registry thread safe

parent 39db309e
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#pragma once #pragma once
#include <mutex>
#include <utility> #include <utility>
#include <functional> #include <functional>
#include <unordered_map> #include <unordered_map>
...@@ -70,17 +71,17 @@ public: ...@@ -70,17 +71,17 @@ public:
using proxy_map = std::map<actor_id, strong_actor_ptr>; using proxy_map = std::map<actor_id, strong_actor_ptr>;
/// Returns the number of proxies for `node`. /// Returns the number of proxies for `node`.
size_t count_proxies(const node_id& node); size_t count_proxies(const node_id& node) const;
/// Returns the proxy instance identified by `node` and `aid`. /// Returns the proxy instance identified by `node` and `aid`.
strong_actor_ptr get(const node_id& node, actor_id aid); strong_actor_ptr get(const node_id& node, actor_id aid) const;
/// Returns the proxy instance identified by `node` and `aid` /// Returns the proxy instance identified by `node` and `aid`
/// or creates a new (default) proxy instance. /// or creates a new (default) proxy instance.
strong_actor_ptr get_or_put(const node_id& nid, actor_id aid); strong_actor_ptr get_or_put(const node_id& nid, actor_id aid);
/// Returns all known proxies. /// Returns all known proxies.
std::vector<strong_actor_ptr> get_all(const node_id& node); std::vector<strong_actor_ptr> get_all(const node_id& node) const;
/// Deletes all proxies for `node`. /// Deletes all proxies for `node`.
void erase(const node_id& nid); void erase(const node_id& nid);
...@@ -105,21 +106,19 @@ public: ...@@ -105,21 +106,19 @@ public:
return system_; return system_;
} }
size_t size() const { /// Sets the thread-local last hop variable on the backend.
return proxies_.size();
}
void set_last_hop(node_id* ptr) { void set_last_hop(node_id* ptr) {
backend_.set_last_hop(ptr); backend_.set_last_hop(ptr);
} }
private: private:
/// @pre mtx_ is locked
void kill_proxy(strong_actor_ptr&, error); void kill_proxy(strong_actor_ptr&, error);
actor_system& system_; actor_system& system_;
backend& backend_; backend& backend_;
mutable std::mutex mtx_;
std::unordered_map<node_id, proxy_map> proxies_; std::unordered_map<node_id, proxy_map> proxies_;
}; };
} // namespace caf } // namespace caf
...@@ -45,42 +45,51 @@ proxy_registry::~proxy_registry() { ...@@ -45,42 +45,51 @@ proxy_registry::~proxy_registry() {
clear(); clear();
} }
size_t proxy_registry::count_proxies(const node_id& node) { size_t proxy_registry::count_proxies(const node_id& node) const {
std::unique_lock<std::mutex> guard {mtx_};
auto i = proxies_.find(node); auto i = proxies_.find(node);
return (i != proxies_.end()) ? i->second.size() : 0; return i != proxies_.end() ? i->second.size() : 0;
} }
strong_actor_ptr proxy_registry::get(const node_id& node, actor_id aid) { strong_actor_ptr proxy_registry::get(const node_id& node, actor_id aid) const {
auto& submap = proxies_[node]; std::unique_lock<std::mutex> guard{mtx_};
auto i = submap.find(aid); auto i = proxies_.find(node);
if (i != submap.end()) if (i == proxies_.end())
return i->second; return nullptr;
return nullptr; auto j = i->second.find(aid);
return j != i->second.end() ? j->second : nullptr;
} }
strong_actor_ptr proxy_registry::get_or_put(const node_id& nid, actor_id aid) { strong_actor_ptr proxy_registry::get_or_put(const node_id& nid, actor_id aid) {
CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid)); CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid));
std::unique_lock<std::mutex> guard{mtx_};
auto& result = proxies_[nid][aid]; auto& result = proxies_[nid][aid];
if (!result) if (!result)
result = backend_.make_proxy(nid, aid); result = backend_.make_proxy(nid, aid);
return result; return result;
} }
std::vector<strong_actor_ptr> proxy_registry::get_all(const node_id& node) { std::vector<strong_actor_ptr>
proxy_registry::get_all(const node_id& node) const {
// Reserve at least some memory outside of the critical section.
std::vector<strong_actor_ptr> result; std::vector<strong_actor_ptr> result;
result.reserve(128);
std::unique_lock<std::mutex> guard{mtx_};
auto i = proxies_.find(node); auto i = proxies_.find(node);
if (i != proxies_.end()) if (i != proxies_.end())
for (auto& kvp : i->second) for (auto& kvp : i->second)
result.push_back(kvp.second); result.emplace_back(kvp.second);
return result; return result;
} }
bool proxy_registry::empty() const { bool proxy_registry::empty() const {
std::unique_lock<std::mutex> guard{mtx_};
return proxies_.empty(); return proxies_.empty();
} }
void proxy_registry::erase(const node_id& nid) { void proxy_registry::erase(const node_id& nid) {
CAF_LOG_TRACE(CAF_ARG(nid)); CAF_LOG_TRACE(CAF_ARG(nid));
std::unique_lock<std::mutex> guard{mtx_};
auto i = proxies_.find(nid); auto i = proxies_.find(nid);
if (i == proxies_.end()) if (i == proxies_.end())
return; return;
...@@ -91,6 +100,7 @@ void proxy_registry::erase(const node_id& nid) { ...@@ -91,6 +100,7 @@ void proxy_registry::erase(const node_id& nid) {
void proxy_registry::erase(const node_id& nid, actor_id aid, error rsn) { void proxy_registry::erase(const node_id& nid, actor_id aid, error rsn) {
CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid)); CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid));
std::unique_lock<std::mutex> guard{mtx_};
auto i = proxies_.find(nid); auto i = proxies_.find(nid);
if (i != proxies_.end()) { if (i != proxies_.end()) {
auto& submap = i->second; auto& submap = i->second;
...@@ -105,6 +115,7 @@ void proxy_registry::erase(const node_id& nid, actor_id aid, error rsn) { ...@@ -105,6 +115,7 @@ void proxy_registry::erase(const node_id& nid, actor_id aid, error rsn) {
} }
void proxy_registry::clear() { void proxy_registry::clear() {
std::unique_lock<std::mutex> guard{mtx_};
for (auto& kvp : proxies_) for (auto& kvp : proxies_)
for (auto& sub_kvp : kvp.second) for (auto& sub_kvp : kvp.second)
kill_proxy(sub_kvp.second, exit_reason::remote_link_unreachable); kill_proxy(sub_kvp.second, exit_reason::remote_link_unreachable);
......
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