Commit e1181789 authored by Dominik Charousset's avatar Dominik Charousset

Move connection caching to MM actor, relates #390

parent 52ab820f
......@@ -14,6 +14,7 @@ set (LIBCAF_IO_SRCS
src/default_multiplexer.cpp
src/doorman.cpp
src/middleman.cpp
src/middleman_actor.cpp
src/hook.cpp
src/interfaces.cpp
src/manager.cpp
......
......@@ -93,7 +93,7 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// connected port
uint16_t remote_port;
// pending operations to be performed after handhsake completed
std::function<void (const maybe<message>&)> callback;
maybe<response_promise> callback;
};
void set_context(connection_handle hdl);
......@@ -130,27 +130,6 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
const node_id& this_node() const {
return instance.this_node();
}
// and endpoint is identified via host and port
using endpoint = std::pair<std::string, uint16_t>;
// stores the result of a `remote_actor()`
using remote_actor_res = std::tuple<node_id, actor_addr,
std::set<std::string>>;
// stores the state of connected endpoints, `i->second.empty()` indicates
// that a connection has been established but the handshake is still
// pending, in which case it is safe to attach to `ctx[i->first].callback`
using endpoint_data = std::pair<connection_handle, maybe<remote_actor_res>>;
using endpoint_map = std::map<endpoint, endpoint_data>;
// caches the result of all `remote_actor()` calls to avoid
// spamming connections for connecting to the same node multiple times
endpoint_map connected_endpoints;
// "reverse lookup" for finding credentials to a connection
endpoint_map::iterator find_endpoint(connection_handle x);
};
/// A broker implementation for the Binary Actor System Protocol (BASP).
......
......@@ -111,6 +111,9 @@ using middleman_actor =
replies_to<spawn_atom, node_id, std::string, message>
::with<ok_atom, actor_addr, std::set<std::string>>>;
/// @relates middleman_actor
middleman_actor make_middleman_actor(actor_system& sys, actor default_broker);
} // namespace io
} // namespace caf
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_REMOTE_ACTOR_HPP
#define CAF_IO_REMOTE_ACTOR_HPP
#include <set>
#include <string>
#include <cstdint>
#include "caf/fwd.hpp"
#include "caf/actor_cast.hpp"
#include "caf/typed_actor.hpp"
namespace caf {
namespace io {
maybe<actor_addr> remote_actor_impl(std::set<std::string> ifs,
std::string host, uint16_t port);
/// Establish a new connection to the actor at `host` on given `port`.
/// @param host Valid hostname or IP address.
/// @param port TCP port.
/// @returns An {@link actor_ptr} to the proxy instance
/// representing a remote actor.
/// @throws network_error Thrown on connection error or
/// when connecting to a typed actor.
inline actor remote_actor(std::string host, uint16_t port) {
auto res = remote_actor_impl(std::set<std::string>{}, std::move(host), port);
return actor_cast<actor>(res);
}
/// Establish a new connection to the typed actor at `host` on given `port`.
/// @param host Valid hostname or IP address.
/// @param port TCP port.
/// @returns An {@link actor_ptr} to the proxy instance
/// representing a typed remote actor.
/// @throws network_error Thrown on connection error or when connecting
/// to an untyped otherwise unexpected actor.
template <class ActorHandle>
ActorHandle typed_remote_actor(std::string host, uint16_t port) {
auto res = remote_actor_impl(ActorHandle::message_types(),
std::move(host), port);
return actor_cast<ActorHandle>(res);
}
} // namespace io
} // namespace caf
#endif // CAF_IO_REMOTE_ACTOR_HPP
......@@ -120,21 +120,18 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
auto& cb = this_context->callback;
if (! cb)
return;
actor_addr addr;
auto cleanup = detail::make_scope_guard([&] {
cb = nullptr;
auto e = connected_endpoints.end();
auto i = find_endpoint(this_context->hdl);
if (i == e)
return;
// store result of this `remote_actor()` call for later retrieval
i->second.second = std::make_tuple(nid, addr, sigs);
cb = none;
});
if (aid == invalid_actor_id) {
// can occur when connecting to the default port of a node
cb(make_message(ok_atom::value, nid, addr, sigs));
cb->deliver(make_message(ok_atom::value,
nid,
actor_addr{invalid_actor_addr},
std::move(sigs)));
return;
}
actor_addr addr;
if (nid == this_node()) {
// connected to self
addr = system().registry().get(aid).first;
......@@ -147,7 +144,8 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
}
if (addr.node() != system().node())
known_remotes.emplace(nid, std::make_pair(this_context->remote_port, addr));
cb(make_message(ok_atom::value, nid, addr, sigs));
cb->deliver(make_message(ok_atom::value, nid, addr, std::move(sigs)));
this_context->callback = none;
}
void basp_broker_state::purge_state(const node_id& nid) {
......@@ -355,7 +353,7 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
// gotcha! send scribe to our BASP broker
// to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
helper->send(bb, connect_atom::value, hdl, addr, port);
helper->send(bb, connect_atom::value, hdl, port);
return;
}
catch (...) {
......@@ -422,7 +420,7 @@ void basp_broker_state::set_context(connection_handle hdl) {
hdl,
invalid_node_id,
0,
nullptr}).first;
none}).first;
}
this_context = &i->second;
}
......@@ -435,21 +433,12 @@ bool basp_broker_state::erase_context(connection_handle hdl) {
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback(sec::disconnect_during_handshake);
ref.callback->deliver(sec::disconnect_during_handshake);
}
ctx.erase(i);
return true;
}
basp_broker_state::endpoint_map::iterator
basp_broker_state::find_endpoint(connection_handle y) {
auto pred = [&](const endpoint_map::value_type& x) {
return x.second.first == y;
};
auto e = connected_endpoints.end();
return std::find_if(connected_endpoints.begin(), e, pred);
}
/******************************************************************************
* basp_broker *
******************************************************************************/
......@@ -593,41 +582,9 @@ behavior basp_broker::make_behavior() {
state.instance.add_published_actor(port, whom, std::move(sigs));
},
// received from middleman actor (delegated)
[=](connect_atom, connection_handle hdl,
std::string& hostname, uint16_t port) {
[=](connect_atom, connection_handle hdl, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hdl.id()));
// store original context via response promise
auto rp = make_response_promise();
// check whether we already know these credentials
auto ep = std::make_pair(std::move(hostname), port);
auto i = state.connected_endpoints.find(ep);
if (i != state.connected_endpoints.end()) {
auto& x = i->second;
if (x.second.invalid()) {
rp.deliver(x.second.error());
} else if (x.second.empty()) {
// in this state, we have a connection but are waiting
// for the handshake to complete and we can attach to the
// respective callback
auto j = state.ctx.find(x.first);
CAF_ASSERT(j != state.ctx.end());
CAF_ASSERT(j->second.callback != nullptr);
auto f = j->second.callback; // store previous callback
j->second.callback = [f, rp](const maybe<message>& x) {
f(x); // run original callback(s) first
if (x)
rp.deliver(*x);
else if (x.invalid())
rp.deliver(x.error());
};
} else {
auto& tup = *(x.second);
rp.deliver(make_message(ok_atom::value, get<0>(tup),
get<1>(tup), get<2>(tup)));
}
return;
}
// initiate handshake in case we aren't connected to these credentials
try {
assign_tcp_scribe(hdl);
}
......@@ -640,15 +597,7 @@ behavior basp_broker::make_behavior() {
ctx.hdl = hdl;
ctx.remote_port = port;
ctx.cstate = basp::await_header;
ctx.callback = [rp](const maybe<message>& res) {
if (res.valid())
rp.deliver(*res);
else if (res.invalid())
rp.deliver(res.error());
};
using endpoint_data = basp_broker_state::endpoint_data;
state.connected_endpoints.emplace(std::move(ep),
endpoint_data{hdl, none});
ctx.callback = rp;
// await server handshake
configure_read(hdl, receive_policy::exactly(basp::header_size));
},
......
......@@ -66,111 +66,6 @@
namespace caf {
namespace io {
namespace {
class middleman_actor_impl : public middleman_actor::base {
public:
middleman_actor_impl(actor_config& cfg, actor default_broker)
: middleman_actor::base(cfg),
broker_(default_broker) {
// nop
}
void on_exit() override {
CAF_LOG_TRACE("");
broker_ = invalid_actor;
}
const char* name() const override {
return "middleman_actor";
}
using put_res = maybe<std::tuple<ok_atom, uint16_t>>;
using get_res = delegated<ok_atom, node_id, actor_addr, std::set<std::string>>;
using del_res = delegated<void>;
behavior_type make_behavior() override {
CAF_LOG_TRACE("");
return {
[=](publish_atom, uint16_t port, actor_addr& whom,
std::set<std::string>& sigs, std::string& addr, bool reuse) {
CAF_LOG_TRACE("");
return put(port, whom, sigs, addr.c_str(), reuse);
},
[=](open_atom, uint16_t port, std::string& addr, bool reuse) -> put_res {
CAF_LOG_TRACE("");
actor_addr whom = invalid_actor_addr;
std::set<std::string> sigs;
return put(port, whom, sigs, addr.c_str(), reuse);
},
[=](connect_atom, std::string& hostname, uint16_t port) -> get_res {
CAF_LOG_TRACE(CAF_ARG(hostname) << CAF_ARG(port));
connection_handle hdl;
try {
hdl = system().middleman().backend().new_tcp_scribe(hostname, port);
} catch(std::exception&) {
// nop
}
if (hdl != invalid_connection_handle) {
delegate(broker_, connect_atom::value, hdl,
std::move(hostname), port);
} else {
auto rp = make_response_promise();
rp.deliver(sec::cannot_connect_to_node);
}
return {};
},
[=](unpublish_atom, const actor_addr&, uint16_t) -> del_res {
CAF_LOG_TRACE("");
forward_current_message(broker_);
return {};
},
[=](close_atom, uint16_t) -> del_res {
CAF_LOG_TRACE("");
forward_current_message(broker_);
return {};
},
[=](spawn_atom, const node_id&, const std::string&, const message&)
-> delegated<ok_atom, actor_addr, std::set<std::string>> {
CAF_LOG_TRACE("");
forward_current_message(broker_);
return {};
}
};
}
private:
put_res put(uint16_t port, actor_addr& whom,
std::set<std::string>& sigs, const char* in = nullptr,
bool reuse_addr = false) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(whom) << CAF_ARG(sigs)
<< CAF_ARG(in) << CAF_ARG(reuse_addr));
accept_handle hdl;
uint16_t actual_port;
// treat empty strings like nullptr
if (in != nullptr && in[0] == '\0')
in = nullptr;
try {
auto res = system().middleman().backend().new_tcp_doorman(port, in,
reuse_addr);
hdl = res.first;
actual_port = res.second;
send(broker_, publish_atom::value, hdl, actual_port,
std::move(whom), std::move(sigs));
return {ok_atom::value, actual_port};
}
catch (std::exception& e) {
return sec::cannot_open_port;
}
}
actor broker_;
};
} // namespace <anonymous>
actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) {
class impl : public middleman {
public:
......@@ -391,7 +286,7 @@ void middleman::start() {
backend().thread_id(thread_.get_id());
}
auto basp = named_broker<basp_broker>(atom("BASP"));
manager_ = system().spawn<middleman_actor_impl, detached + hidden>(basp);
manager_ = make_middleman_actor(system(), basp);
}
void middleman::stop() {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/middleman_actor.hpp"
#include <tuple>
#include <stdexcept>
#include "caf/sec.hpp"
#include "caf/actor.hpp"
#include "caf/logger.hpp"
#include "caf/node_id.hpp"
#include "caf/exception.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/io/basp_broker.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf {
namespace io {
namespace {
class middleman_actor_impl : public middleman_actor::base {
public:
middleman_actor_impl(actor_config& cfg, actor default_broker)
: middleman_actor::base(cfg),
broker_(default_broker) {
// nop
}
void on_exit() override {
CAF_LOG_TRACE("");
broker_ = invalid_actor;
}
const char* name() const override {
return "middleman_actor";
}
using put_res = maybe<std::tuple<ok_atom, uint16_t>>;
using mpi_set = std::set<std::string>;
using get_res = delegated<ok_atom, node_id, actor_addr, mpi_set>;
using del_res = delegated<void>;
using endpoint_data = std::tuple<node_id, actor_addr, mpi_set>;
using endpoint = std::pair<std::string, uint16_t>;
behavior_type make_behavior() override {
CAF_LOG_TRACE("");
return {
[=](publish_atom, uint16_t port, actor_addr& whom,
mpi_set& sigs, std::string& addr, bool reuse) {
CAF_LOG_TRACE("");
return put(port, whom, sigs, addr.c_str(), reuse);
},
[=](open_atom, uint16_t port, std::string& addr, bool reuse) -> put_res {
CAF_LOG_TRACE("");
actor_addr whom = invalid_actor_addr;
mpi_set sigs;
return put(port, whom, sigs, addr.c_str(), reuse);
},
[=](connect_atom, std::string& hostname, uint16_t port) -> get_res {
CAF_LOG_TRACE(CAF_ARG(hostname) << CAF_ARG(port));
auto rp = make_response_promise();
endpoint key{std::move(hostname), port};
// respond immediately if endpoint is cached
auto x = cached(key);
if (x) {
CAF_LOG_DEBUG("found cached entry" << CAF_ARG(*x));
rp.deliver(ok_atom::value, get<0>(*x), get<1>(*x), get<2>(*x));
printf("%s %d\n", __FILE__, __LINE__);
return {};
}
// attach this promise to a pending request if possible
auto rps = pending(key);
if (rps) {
CAF_LOG_DEBUG("attach to pending request");
rps->emplace_back(std::move(rp));
printf("%s %d\n", __FILE__, __LINE__);
return {};
}
// connect to endpoint and initiate handhsake etc.
connection_handle hdl;
try {
hdl = system().middleman().backend().new_tcp_scribe(key.first, port);
} catch(std::exception&) {
rp.deliver(sec::cannot_connect_to_node);
printf("%s %d\n", __FILE__, __LINE__);
return {};
}
std::vector<response_promise> tmp{std::move(rp)};
pending_.emplace(key, std::move(tmp));
request(broker_, connect_atom::value, hdl, port).then(
[=](ok_atom, node_id& nid, actor_addr& addr, mpi_set& sigs) {
printf("%s %d\n", __FILE__, __LINE__);
auto i = pending_.find(key);
if (i == pending_.end())
return;
monitor(addr);
cached_.emplace(key, std::make_tuple(nid, addr, sigs));
auto res = make_message(ok_atom::value, std::move(nid),
std::move(addr), std::move(sigs));
for (auto& x : i->second)
x.deliver(res);
pending_.erase(i);
},
[=](error& err) {
printf("%s %d\n", __FILE__, __LINE__);
auto i = pending_.find(key);
if (i == pending_.end())
return;
for (auto& x : i->second)
x.deliver(err);
pending_.erase(i);
}
);
return {};
},
[=](unpublish_atom, const actor_addr&, uint16_t) -> del_res {
printf("%s %d\n", __FILE__, __LINE__);
CAF_LOG_TRACE("");
forward_current_message(broker_);
return {};
},
[=](close_atom, uint16_t) -> del_res {
CAF_LOG_TRACE("");
forward_current_message(broker_);
return {};
},
[=](spawn_atom, const node_id&, const std::string&, const message&)
-> delegated<ok_atom, actor_addr, mpi_set> {
CAF_LOG_TRACE("");
forward_current_message(broker_);
return {};
},
[=](const down_msg& dm) {
printf("%s %d\n", __FILE__, __LINE__);
auto i = cached_.begin();
auto e = cached_.end();
while (i != e) {
if (get<1>(i->second) == dm.source)
i = cached_.erase(i);
else
++i;
}
}
};
}
private:
put_res put(uint16_t port, actor_addr& whom,
mpi_set& sigs, const char* in = nullptr,
bool reuse_addr = false) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(whom) << CAF_ARG(sigs)
<< CAF_ARG(in) << CAF_ARG(reuse_addr));
accept_handle hdl;
uint16_t actual_port;
// treat empty strings like nullptr
if (in != nullptr && in[0] == '\0')
in = nullptr;
try {
auto res = system().middleman().backend().new_tcp_doorman(port, in,
reuse_addr);
hdl = res.first;
actual_port = res.second;
send(broker_, publish_atom::value, hdl, actual_port,
std::move(whom), std::move(sigs));
return {ok_atom::value, actual_port};
}
catch (std::exception&) {
return sec::cannot_open_port;
}
}
maybe<endpoint_data&> cached(const endpoint& ep) {
auto i = cached_.find(ep);
if (i != cached_.end())
return i->second;
return none;
}
maybe<std::vector<response_promise>&> pending(const endpoint& ep) {
auto i = pending_.find(ep);
if (i != pending_.end())
return i->second;
return none;
}
actor broker_;
std::map<endpoint, endpoint_data> cached_;
std::map<endpoint, std::vector<response_promise>> pending_;
};
} // namespace <anonymous>
middleman_actor make_middleman_actor(actor_system& sys, actor db) {
return sys.spawn<middleman_actor_impl, detached + hidden>(std::move(db));
}
} // namespace io
} // namespace caf
......@@ -22,6 +22,7 @@
#define CAF_SUITE io_unpublish
#include "caf/test/unit_test.hpp"
#include <new>
#include <thread>
#include <atomic>
......@@ -53,41 +54,67 @@ public:
}
};
void test_invalid_unpublish(actor_system& system, const actor& published,
uint16_t port) {
auto d = system.spawn<dummy>();
system.middleman().unpublish(d, port);
auto ra = system.middleman().remote_actor("127.0.0.1", port);
CAF_REQUIRE(ra);
CAF_CHECK(ra != d);
CAF_CHECK(ra == published);
anon_send_exit(d, exit_reason::user_shutdown);
}
struct fixture {
fixture() {
new (&system) actor_system(actor_system_config{test::engine::argc(),
test::engine::argv()}
.load<io::middleman>());
testee = system.spawn<dummy>();
}
CAF_TEST(unpublishing) {
auto argc = test::engine::argc();
auto argv = test::engine::argv();
{ // scope for local variables
actor_system system{actor_system_config{argc, argv}.load<io::middleman>()};
auto d = system.spawn<dummy>();
auto port = system.middleman().publish(d, 0);
CAF_REQUIRE(port);
CAF_MESSAGE("published actor on port " << *port);
test_invalid_unpublish(system, d, *port);
CAF_MESSAGE("finished `invalid_unpublish`");
system.middleman().unpublish(d, *port);
// must fail now
CAF_MESSAGE("expect error...");
try {
auto res = system.middleman().remote_actor("127.0.0.1", *port);
CAF_TEST_ERROR("unexpected: remote actor succeeded!");
} catch(std::exception&) {
CAF_MESSAGE("unpublish succeeded");
}
anon_send_exit(d, exit_reason::user_shutdown);
~fixture() {
anon_send_exit(testee, exit_reason::user_shutdown);
testee = invalid_actor;
system.~actor_system();
CAF_CHECK_EQUAL(s_dtor_called.load(), 2);
}
maybe<actor> remote_actor(const char* hostname, uint16_t port) {
maybe<actor> result;
scoped_actor self{system, true};
self->request(system.middleman().actor_handle(),
connect_atom::value, hostname, port).receive(
[&](ok_atom, node_id&, actor_addr& res, std::set<std::string>& xs) {
CAF_REQUIRE(xs.empty());
result = actor_cast<actor>(std::move(res));
},
[&](error& err) {
result = std::move(err);
}
);
return result;
}
// check after dtor of system was called
CAF_CHECK_EQUAL(s_dtor_called.load(), 2);
}
union { actor_system system; }; // manually control ctor/dtor
actor testee;
};
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(unpublish_tests, fixture)
CAF_TEST(unpublishing) {
auto port = system.middleman().publish(testee, 0);
CAF_REQUIRE(port);
CAF_MESSAGE("published actor on port " << *port);
CAF_MESSAGE("test invalid unpublish");
auto testee2 = system.spawn<dummy>();
system.middleman().unpublish(testee2, *port);
auto x0 = remote_actor("127.0.0.1", *port);
CAF_CHECK(x0 != testee2);
CAF_CHECK(x0 == testee);
anon_send_exit(testee2, exit_reason::kill);
CAF_MESSAGE("unpublish testee");
system.middleman().unpublish(testee, *port);
CAF_MESSAGE("check whether testee is still available via cache");
auto x1 = remote_actor("127.0.0.1", *port);
CAF_CHECK(x1 && *x1 == testee);
CAF_MESSAGE("fake dead of testee and check if testee becomes unavailable");
anon_send(system.middleman().actor_handle(), down_msg{testee.address(),
exit_reason::normal});
// must fail now
auto x2 = remote_actor("127.0.0.1", *port);
CAF_CHECK(! x2);
}
CAF_TEST_FIXTURE_SCOPE_END()
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