Commit bb46599f authored by Dominik Charousset's avatar Dominik Charousset

Implement pluggable middleman backends

parent 7a3e93a7
...@@ -19,7 +19,9 @@ set(LIBCAF_NET_SRCS ...@@ -19,7 +19,9 @@ set(LIBCAF_NET_SRCS
src/ip.cpp src/ip.cpp
src/message_type.cpp src/message_type.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/net/backend/test.cpp
src/net/middleman.cpp src/net/middleman.cpp
src/net/middleman_backend.cpp
src/network_socket.cpp src/network_socket.cpp
src/pipe_socket.cpp src/pipe_socket.cpp
src/pollset_updater.cpp src/pollset_updater.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include <map>
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/node_id.hpp"
namespace caf {
namespace net {
namespace backend {
/// Minimal backend for unit testing.
/// @warning this backend is *not* thread safe.
class test : public middleman_backend {
public:
// -- member types -----------------------------------------------------------
using peer_entry = std::pair<stream_socket, endpoint_manager_ptr>;
// -- constructors, destructors, and assignment operators --------------------
test(middleman& mm);
~test() override;
// -- interface functions ----------------------------------------------------
error init() override;
endpoint_manager_ptr peer(const node_id& id) override;
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override;
void set_last_hop(node_id*) override;
// -- properties -------------------------------------------------------------
stream_socket socket(const node_id& peer_id) {
return get_peer(peer_id).first;
}
peer_entry& emplace(const node_id& peer_id, stream_socket first,
stream_socket second);
private:
peer_entry& get_peer(const node_id& id);
middleman& mm_;
std::map<node_id, peer_entry> peers_;
proxy_registry proxies_;
};
} // namespace backend
} // namespace net
} // namespace caf
...@@ -56,11 +56,9 @@ public: ...@@ -56,11 +56,9 @@ public:
using write_packet_callback = callback<byte_span, byte_span>; using write_packet_callback = callback<byte_span, byte_span>;
using proxy_registry_ptr = std::shared_ptr<proxy_registry>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit application(proxy_registry_ptr proxies); explicit application(proxy_registry& proxies);
// -- interface functions ---------------------------------------------------- // -- interface functions ----------------------------------------------------
...@@ -225,7 +223,7 @@ private: ...@@ -225,7 +223,7 @@ private:
uint64_t next_request_id_ = 1; uint64_t next_request_id_ = 1;
/// Points to the factory object for generating proxies. /// Points to the factory object for generating proxies.
proxy_registry_ptr proxies_; proxy_registry& proxies_;
}; };
} // namespace basp } // namespace basp
......
...@@ -36,6 +36,8 @@ class transport_worker_dispatcher; ...@@ -36,6 +36,8 @@ class transport_worker_dispatcher;
// -- classes ------------------------------------------------------------------ // -- classes ------------------------------------------------------------------
class endpoint_manager; class endpoint_manager;
class middleman;
class middleman_backend;
class multiplexer; class multiplexer;
class socket_manager; class socket_manager;
...@@ -51,6 +53,7 @@ struct tcp_stream_socket; ...@@ -51,6 +53,7 @@ struct tcp_stream_socket;
// -- smart pointers ----------------------------------------------------------- // -- smart pointers -----------------------------------------------------------
using endpoint_manager_ptr = intrusive_ptr<endpoint_manager>; using endpoint_manager_ptr = intrusive_ptr<endpoint_manager>;
using middleman_backend_ptr = std::unique_ptr<middleman_backend>;
using multiplexer_ptr = std::shared_ptr<multiplexer>; using multiplexer_ptr = std::shared_ptr<multiplexer>;
using socket_manager_ptr = intrusive_ptr<socket_manager>; using socket_manager_ptr = intrusive_ptr<socket_manager>;
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>; using weak_multiplexer_ptr = std::weak_ptr<multiplexer>;
......
...@@ -21,6 +21,8 @@ ...@@ -21,6 +21,8 @@
#include <thread> #include <thread>
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
namespace caf { namespace caf {
...@@ -28,6 +30,14 @@ namespace net { ...@@ -28,6 +30,14 @@ namespace net {
class middleman : public actor_system::module { class middleman : public actor_system::module {
public: public:
// -- member types -----------------------------------------------------------
using module = actor_system::module;
using module_ptr = actor_system::module_ptr;
using middleman_backend_list = std::vector<middleman_backend_ptr>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
~middleman() override; ~middleman() override;
...@@ -46,10 +56,22 @@ public: ...@@ -46,10 +56,22 @@ public:
// -- factory functions ------------------------------------------------------ // -- factory functions ------------------------------------------------------
static actor_system::module* make(actor_system&, detail::type_list<>); template <class... Ts>
static module* make(actor_system& sys, detail::type_list<Ts...> token) {
std::unique_ptr<middleman> result{new middleman(sys)};
if (sizeof...(Ts) > 0) {
result->backends_.reserve(sizeof...(Ts));
create_backends(*result, token);
}
return result.release();
}
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
actor_system& system() {
return sys_;
}
const actor_system_config& config() const noexcept { const actor_system_config& config() const noexcept {
return sys_.config(); return sys_.config();
} }
...@@ -58,11 +80,25 @@ public: ...@@ -58,11 +80,25 @@ public:
return mpx_; return mpx_;
} }
middleman_backend* backend(string_view scheme) const noexcept;
private: private:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit middleman(actor_system& sys); explicit middleman(actor_system& sys);
// -- utility functions ------------------------------------------------------
static void create_backends(middleman&, detail::type_list<>) {
// End of recursion.
}
template <class T, class... Ts>
static void create_backends(middleman& mm, detail::type_list<T, Ts...>) {
mm.backends_.emplace_back(new T(mm));
create_backends(mm, detail::type_list<Ts...>{});
}
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Points to the parent system. /// Points to the parent system.
...@@ -71,6 +107,9 @@ private: ...@@ -71,6 +107,9 @@ private:
/// Stores the global socket I/O multiplexer. /// Stores the global socket I/O multiplexer.
multiplexer_ptr mpx_; multiplexer_ptr mpx_;
/// Stores all available backends for managing peers.
middleman_backend_list backends_;
/// Runs the multiplexer's event loop /// Runs the multiplexer's event loop
std::thread mpx_thread_; std::thread mpx_thread_;
}; };
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include <string>
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/proxy_registry.hpp"
namespace caf {
namespace net {
/// Technology-specific backend for connecting to and managing peer
/// connections.
/// @relates middleman
class middleman_backend : public proxy_registry::backend {
public:
// -- constructors, destructors, and assignment operators --------------------
middleman_backend(std::string id);
virtual ~middleman_backend();
// -- interface functions ----------------------------------------------------
/// Initializes the backend.
virtual error init() = 0;
/// @returns The endpoint manager for `peer` on success, `nullptr` otherwise.
virtual endpoint_manager_ptr peer(const node_id& id) = 0;
// -- properties -------------------------------------------------------------
const std::string& id() const noexcept {
return id_;
}
private:
/// Stores the technology-specific identifier.
std::string id_;
};
/// @relates middleman_backend
using middleman_backend_ptr = std::unique_ptr<middleman_backend>;
} // namespace net
} // namespace caf
...@@ -42,8 +42,7 @@ namespace caf { ...@@ -42,8 +42,7 @@ namespace caf {
namespace net { namespace net {
namespace basp { namespace basp {
application::application(proxy_registry_ptr proxies) application::application(proxy_registry& proxies) : proxies_(proxies) {
: proxies_(std::move(proxies)) {
// nop // nop
} }
...@@ -210,7 +209,7 @@ error application::handle_actor_message(write_packet_callback&, header hdr, ...@@ -210,7 +209,7 @@ error application::handle_actor_message(write_packet_callback&, header hdr,
// Try to fetch the sender. // Try to fetch the sender.
strong_actor_ptr src_hdl; strong_actor_ptr src_hdl;
if (src_node != none && src_id != 0) if (src_node != none && src_id != 0)
src_hdl = proxies_->get_or_put(src_node, src_id); src_hdl = proxies_.get_or_put(src_node, src_id);
// Ship the message. // Ship the message.
auto ptr = make_mailbox_element(std::move(src_hdl), auto ptr = make_mailbox_element(std::move(src_hdl),
make_message_id(hdr.operation_data), make_message_id(hdr.operation_data),
...@@ -269,7 +268,7 @@ error application::handle_resolve_response(write_packet_callback&, header hdr, ...@@ -269,7 +268,7 @@ error application::handle_resolve_response(write_packet_callback&, header hdr,
i->second.deliver(strong_actor_ptr{nullptr}, std::move(ifs)); i->second.deliver(strong_actor_ptr{nullptr}, std::move(ifs));
return none; return none;
} }
i->second.deliver(proxies_->get_or_put(peer_id_, aid), std::move(ifs)); i->second.deliver(proxies_.get_or_put(peer_id_, aid), std::move(ifs));
return none; return none;
} }
...@@ -279,7 +278,7 @@ error application::handle_down_message(write_packet_callback&, header hdr, ...@@ -279,7 +278,7 @@ error application::handle_down_message(write_packet_callback&, header hdr,
binary_deserializer source{system(), payload}; binary_deserializer source{system(), payload};
if (auto err = source(reason)) if (auto err = source(reason))
return err; return err;
proxies_->erase(peer_id_, hdr.operation_data, std::move(reason)); proxies_.erase(peer_id_, hdr.operation_data, std::move(reason));
return none; return none;
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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/net/backend/test.hpp"
#include "caf/expected.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/raise_error.hpp"
namespace caf {
namespace net {
namespace backend {
using transport_type = stream_transport<basp::application>;
test::test(middleman& mm)
: middleman_backend("test"), mm_(mm), proxies_(mm.system(), *this) {
// nop
}
test::~test() {
for (auto& kvp : peers_)
close(kvp.second.first);
}
error test::init() {
return none;
}
endpoint_manager_ptr test::peer(const node_id& id) {
return get_peer(id).second;
}
strong_actor_ptr test::make_proxy(node_id nid, actor_id aid) {
using impl_type = actor_proxy_impl;
using hdl_type = strong_actor_ptr;
actor_config cfg;
return make_actor<impl_type, hdl_type>(aid, nid, &mm_.system(), cfg,
peer(nid));
}
void test::set_last_hop(node_id*) {
// nop
}
test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first,
stream_socket second) {
using transport_type = stream_transport<basp::application>;
nonblocking(first, true);
nonblocking(second, true);
auto mpx = mm_.mpx();
auto mgr = make_endpoint_manager(mpx, mm_.system(),
transport_type{second,
basp::application{proxies_}});
if (auto err = mgr->init())
CAF_RAISE_ERROR("mgr->init() failed");
auto& result = peers_[peer_id];
result = std::make_pair(first, std::move(mgr));
return result;
}
test::peer_entry& test::get_peer(const node_id& id) {
auto i = peers_.find(id);
if (i != peers_.end())
return i->second;
auto sockets = make_stream_socket_pair();
if (!sockets)
CAF_RAISE_ERROR("make_stream_socket_pair failed");
return emplace(id, sockets->first, sockets->second);
}
} // namespace backend
} // namespace net
} // namespace caf
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#include "caf/net/middleman.hpp" #include "caf/net/middleman.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/raise_error.hpp" #include "caf/raise_error.hpp"
#include "caf/uri.hpp" #include "caf/uri.hpp"
...@@ -47,7 +48,7 @@ void middleman::stop() { ...@@ -47,7 +48,7 @@ void middleman::stop() {
mpx_thread_.join(); mpx_thread_.join();
} }
void middleman::init(actor_system_config&cfg) { void middleman::init(actor_system_config& cfg) {
if (auto err = mpx_->init()) if (auto err = mpx_->init())
CAF_RAISE_ERROR("mpx->init failed"); CAF_RAISE_ERROR("mpx->init failed");
if (auto node_uri = get_if<uri>(&cfg, "middleman.this-node")) { if (auto node_uri = get_if<uri>(&cfg, "middleman.this-node")) {
...@@ -56,9 +57,12 @@ void middleman::init(actor_system_config&cfg) { ...@@ -56,9 +57,12 @@ void middleman::init(actor_system_config&cfg) {
} else { } else {
CAF_RAISE_ERROR("no valid entry for middleman.this-node found"); CAF_RAISE_ERROR("no valid entry for middleman.this-node found");
} }
for (auto& backend : backends_)
if (auto err = backend->init())
CAF_RAISE_ERROR("failed to initialize backend");
} }
actor_system::module::id_t middleman::id() const { middleman::module::id_t middleman::id() const {
return module::network_manager; return module::network_manager;
} }
...@@ -66,8 +70,14 @@ void* middleman::subtype_ptr() { ...@@ -66,8 +70,14 @@ void* middleman::subtype_ptr() {
return this; return this;
} }
actor_system::module* middleman::make(actor_system& sys, detail::type_list<>) { middleman_backend* middleman::backend(string_view scheme) const noexcept {
return new middleman(sys); auto predicate = [&](const middleman_backend_ptr& ptr) {
return ptr->id() == scheme;
};
auto i = std::find_if(backends_.begin(), backends_.end(), predicate);
if (i != backends_.end())
return i->get();
return nullptr;
} }
} // namespace net } // namespace net
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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/net/middleman_backend.hpp"
namespace caf {
namespace net {
middleman_backend::middleman_backend(std::string id) : id_(std::move(id)) {
// nop
}
middleman_backend::~middleman_backend() {
// nop
}
} // namespace net
} // namespace caf
...@@ -44,7 +44,7 @@ namespace { ...@@ -44,7 +44,7 @@ namespace {
struct fixture : test_coordinator_fixture<>, proxy_registry::backend { struct fixture : test_coordinator_fixture<>, proxy_registry::backend {
using buffer_type = std::vector<byte>; using buffer_type = std::vector<byte>;
fixture() : app(std::make_shared<proxy_registry>(sys, *this)) { fixture() : proxies(sys, *this), app(proxies) {
REQUIRE_OK(app.init(*this)); REQUIRE_OK(app.init(*this));
uri mars_uri; uri mars_uri;
REQUIRE_OK(parse("tcp://mars", mars_uri)); REQUIRE_OK(parse("tcp://mars", mars_uri));
...@@ -130,6 +130,8 @@ struct fixture : test_coordinator_fixture<>, proxy_registry::backend { ...@@ -130,6 +130,8 @@ struct fixture : test_coordinator_fixture<>, proxy_registry::backend {
node_id mars; node_id mars;
proxy_registry proxies;
basp::application app; basp::application app;
}; };
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#define CAF_SUITE net.backend.test
#include "caf/net/backend/test.hpp"
#include "caf/test/dsl.hpp"
using namespace caf;
namespace {
struct fixture {
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(test_tests, fixture)
CAF_TEST(todo) {
// implement me
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -29,11 +29,13 @@ ...@@ -29,11 +29,13 @@
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/forwarding_actor_proxy.hpp" #include "caf/forwarding_actor_proxy.hpp"
#include "caf/net/actor_proxy_impl.hpp" #include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/backend/test.hpp"
#include "caf/net/basp/connection_state.hpp" #include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp" #include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp" #include "caf/net/basp/ec.hpp"
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp" #include "caf/net/middleman.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp" #include "caf/net/stream_transport.hpp"
...@@ -62,35 +64,18 @@ size_t fetch_size(variant<size_t, sec> x) { ...@@ -62,35 +64,18 @@ size_t fetch_size(variant<size_t, sec> x) {
struct config : actor_system_config { struct config : actor_system_config {
config() { config() {
put(content, "middleman.this-node", unbox(make_uri("test:earth"))); put(content, "middleman.this-node", unbox(make_uri("test:earth")));
load<middleman>(); load<middleman, backend::test>();
} }
}; };
struct fixture : test_coordinator_fixture<config>, struct fixture : test_coordinator_fixture<config>, host_fixture {
host_fixture, fixture() : mars(make_node_id(unbox(make_uri("test:mars")))) {
proxy_registry::backend { auto& mm = sys.network_manager();
fixture() { auto backend = dynamic_cast<backend::test*>(mm.backend("test"));
uri mars_uri; auto mgr = backend->peer(mars);
REQUIRE_OK(parse("test:mars", mars_uri));
mars = make_node_id(mars_uri);
auto proxies = std::make_shared<proxy_registry>(sys, *this);
auto sockets = unbox(make_stream_socket_pair());
sock = sockets.first;
nonblocking(sockets.first, true);
nonblocking(sockets.second, true);
auto mpx = sys.network_manager().mpx();
mgr = make_endpoint_manager(mpx, sys,
transport_type{sockets.second,
basp::application{proxies}});
REQUIRE_OK(mgr->init());
handle_io_event();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
auto& dref = dynamic_cast<endpoint_manager_impl<transport_type>&>(*mgr); auto& dref = dynamic_cast<endpoint_manager_impl<transport_type>&>(*mgr);
app = &dref.application(); app = &dref.application();
} sock = backend->socket(mars);
~fixture() {
close(sock);
} }
bool handle_io_event() override { bool handle_io_event() override {
...@@ -151,21 +136,8 @@ struct fixture : test_coordinator_fixture<config>, ...@@ -151,21 +136,8 @@ struct fixture : test_coordinator_fixture<config>,
return sys; return sys;
} }
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override {
using impl_type = actor_proxy_impl;
using hdl_type = strong_actor_ptr;
actor_config cfg;
return make_actor<impl_type, hdl_type>(aid, nid, &sys, cfg, mgr);
}
void set_last_hop(node_id*) override {
// nop
}
node_id mars; node_id mars;
endpoint_manager_ptr mgr;
stream_socket sock; stream_socket sock;
basp::application* app; basp::application* app;
......
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