Commit 65f675e4 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Make basp_broker::make_proxy thread-safe

parent e4fb896c
......@@ -42,6 +42,9 @@ public:
/// Creates a new proxy instance.
virtual strong_actor_ptr make_proxy(node_id, actor_id) = 0;
/// Sets the thread-local last-hop pointer to detect indirect connections.
virtual void set_last_hop(node_id* ptr) = 0;
};
proxy_registry(actor_system& sys, backend& be);
......@@ -93,19 +96,23 @@ public:
void clear();
/// Returns the hosting actor system.
inline actor_system& system() {
actor_system& system() {
return system_;
}
/// Returns the hosting actor system.
inline const actor_system& system() const {
const actor_system& system() const {
return system_;
}
inline size_t size() const {
size_t size() const {
return proxies_.size();
}
void set_last_hop(node_id* ptr) {
backend_.set_last_hop(ptr);
}
private:
void kill_proxy(strong_actor_ptr&, error);
......
......@@ -50,6 +50,8 @@ public:
message msg;
auto mid = make_message_id(dref.hdr_.operation_data);
binary_deserializer source{ctx, dref.payload_};
// Registry setup.
dref.proxies_->set_last_hop(&dref.last_hop_);
// Get the local receiver.
if (dref.hdr_.has(basp::header::named_receiver_flag))
dst = sys.registry().get(static_cast<atom_value>(dref.hdr_.dest_actor));
......
......@@ -48,6 +48,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from proxy_registry::backend
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override;
// inherited from proxy_registry::backend
void set_last_hop(node_id* ptr) override;
// inherited from basp::instance::callee
void finalize_handshake(const node_id& nid, actor_id aid,
std::set<std::string>& sigs) override;
......
......@@ -36,6 +36,21 @@
#include "caf/sec.hpp"
#include "caf/send.hpp"
namespace {
#ifdef CAF_MSVC
# define THREAD_LOCAL thread_local
#else
# define THREAD_LOCAL __thread
#endif
// Used by make_proxy to detect indirect connections.
THREAD_LOCAL caf::node_id* t_last_hop = nullptr;
#undef THREAD_LOCAL
} // namespace
namespace caf {
namespace io {
......@@ -64,17 +79,17 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
CAF_ASSERT(nid != this_node());
if (nid == none || aid == invalid_actor_id)
return nullptr;
auto mm = &system().middleman();
// this member function is being called whenever we deserialize a
// payload received from a remote node; if a remote node A sends
// us a handle to a third node B, then we assume that A offers a route to B
if (nid != this_context->id
&& instance.tbl().add_indirect(this_context->id, nid))
learned_new_node_indirectly(nid);
if (t_last_hop != nullptr && nid != *t_last_hop
&& instance.tbl().add_indirect(*t_last_hop, nid))
mm->backend().dispatch([=] { learned_new_node_indirectly(nid); });
// we need to tell remote side we are watching this actor now;
// use a direct route if possible, i.e., when talking to a third node
// create proxy and add functor that will be called if we
// receive a basp::down_message
auto mm = &system().middleman();
actor_config cfg;
auto res = make_actor<forwarding_actor_proxy, strong_actor_ptr>(
aid, nid, &(self->home_system()), cfg, self);
......@@ -92,6 +107,10 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
return res;
}
void basp_broker_state::set_last_hop(node_id* ptr) {
t_last_hop = ptr;
}
void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
std::set<std::string>& sigs) {
CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid) << CAF_ARG(sigs));
......@@ -269,6 +288,7 @@ void basp_broker_state::set_context(connection_handle hdl) {
.first;
}
this_context = &i->second;
t_last_hop = &i->second.id;
}
void basp_broker_state::cleanup(connection_handle hdl) {
......
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