Commit 98280d5c authored by Jakob Otto's avatar Jakob Otto

Add tcp backend

parent 61549251
...@@ -50,6 +50,7 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS} ...@@ -50,6 +50,7 @@ add_library(libcaf_net_obj OBJECT ${CAF_NET_HEADERS}
src/ip.cpp src/ip.cpp
src/message_queue.cpp src/message_queue.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/net/backend/tcp.cpp
src/net/backend/test.cpp src/net/backend/test.cpp
src/net/endpoint_manager_queue.cpp src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp src/net/middleman.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/detail/net_export.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/node_id.hpp"
namespace caf::net::backend {
/// Minimal backend for tcp communication.
/// @warning this backend is *not* thread safe.
class CAF_NET_EXPORT tcp : public middleman_backend {
public:
// -- constructors, destructors, and assignment operators --------------------
tcp(middleman& mm);
~tcp() override;
// -- interface functions ----------------------------------------------------
error init() override;
void stop() override;
endpoint_manager_ptr connect(const uri& locator);
endpoint_manager_ptr peer(const node_id& id) override;
void publish(actor handle, const uri& locator) override;
void resolve(const uri& locator, const actor& listener) override;
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override;
void set_last_hop(node_id*) override;
void publish();
// -- properties -------------------------------------------------------------
tcp_stream_socket socket(const node_id& peer_id) {
return socket_cast<tcp_stream_socket>(peers_[peer_id]->handle());
}
uint16_t port() const noexcept override {
return listening_port_;
}
private:
class basp_application_factory {
public:
using application_type = basp::application;
basp_application_factory(proxy_registry& proxies);
~basp_application_factory();
template <class Parent>
error init(Parent&) {
return none;
}
application_type make() const {
return application_type{proxies_};
}
private:
proxy_registry& proxies_;
};
template <class Handle>
endpoint_manager_ptr& emplace(const node_id& peer_id, Handle socket_handle) {
using transport_type = stream_transport<basp::application>;
nonblocking(socket_handle, true);
auto mpx = mm_.mpx();
basp::application app{proxies_};
auto mgr = make_endpoint_manager(
mpx, mm_.system(), transport_type{socket_handle, std::move(app)});
if (auto err = mgr->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << err);
CAF_RAISE_ERROR("mgr->init() failed");
}
mpx->register_reading(mgr);
return peers_[peer_id];
}
endpoint_manager_ptr get_peer(const node_id& id);
middleman& mm_;
std::map<node_id, endpoint_manager_ptr> peers_;
proxy_registry proxies_;
uint16_t listening_port_;
};
} // namespace caf::net::backend
...@@ -51,6 +51,8 @@ public: ...@@ -51,6 +51,8 @@ public:
endpoint_manager_ptr peer(const node_id& id) override; endpoint_manager_ptr peer(const node_id& id) override;
void publish(actor handle, const uri& locator) override;
void resolve(const uri& locator, const actor& listener) override; void resolve(const uri& locator, const actor& listener) override;
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override; strong_actor_ptr make_proxy(node_id nid, actor_id aid) override;
...@@ -63,6 +65,10 @@ public: ...@@ -63,6 +65,10 @@ public:
return get_peer(peer_id).first; return get_peer(peer_id).first;
} }
uint16_t port() const noexcept override {
return 0;
}
peer_entry& emplace(const node_id& peer_id, stream_socket first, peer_entry& emplace(const node_id& peer_id, stream_socket first,
stream_socket second); stream_socket second);
......
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#pragma once #pragma once
#include <cstddef> #include <cstddef>
#include <cstdint>
// -- hard-coded default values for various CAF options ------------------------ // -- hard-coded default values for various CAF options ------------------------
...@@ -30,4 +31,7 @@ extern const size_t max_payload_buffers; ...@@ -30,4 +31,7 @@ extern const size_t max_payload_buffers;
/// Maximum number of cached buffers for sending headers. /// Maximum number of cached buffers for sending headers.
extern const size_t max_header_buffers; extern const size_t max_header_buffers;
/// Port to listen on for tcp.
extern const uint16_t tcp_port;
} // namespace caf::defaults::middleman } // namespace caf::defaults::middleman
...@@ -46,6 +46,9 @@ public: ...@@ -46,6 +46,9 @@ public:
/// @returns The endpoint manager for `peer` on success, `nullptr` otherwise. /// @returns The endpoint manager for `peer` on success, `nullptr` otherwise.
virtual endpoint_manager_ptr peer(const node_id& id) = 0; virtual endpoint_manager_ptr peer(const node_id& id) = 0;
/// Publishes an actor.
virtual void publish(actor handle, const uri& locator) = 0;
/// Resolves a path to a remote actor. /// Resolves a path to a remote actor.
virtual void resolve(const uri& locator, const actor& listener) = 0; virtual void resolve(const uri& locator, const actor& listener) = 0;
...@@ -57,6 +60,8 @@ public: ...@@ -57,6 +60,8 @@ public:
return id_; return id_;
} }
virtual uint16_t port() const = 0;
private: private:
/// Stores the technology-specific identifier. /// Stores the technology-specific identifier.
std::string id_; std::string id_;
......
...@@ -24,4 +24,6 @@ const size_t max_payload_buffers = 100; ...@@ -24,4 +24,6 @@ const size_t max_payload_buffers = 100;
const size_t max_header_buffers = 10; const size_t max_header_buffers = 10;
const uint16_t tcp_port = 0;
} // namespace caf::defaults::middleman } // namespace caf::defaults::middleman
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/tcp.hpp"
#include <string>
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/doorman.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/send.hpp"
namespace caf::net::backend {
tcp::tcp(middleman& mm)
: middleman_backend("tcp"), mm_(mm), proxies_(mm.system(), *this) {
// nop
}
tcp::~tcp() {
// nop
}
error tcp::init() {
uint16_t conf_port = get_or<uint16_t>(
mm_.system().config(), "middleman.tcp_port", defaults::middleman::tcp_port);
ip_endpoint ep;
auto local_address = std::string("[::]:") + std::to_string(conf_port);
if (auto err = detail::parse(local_address, ep))
return err;
auto acceptor = make_tcp_accept_socket(ep);
if (!acceptor)
return acceptor.error();
auto acc_guard = make_socket_guard(*acceptor);
nonblocking(acc_guard.socket(), true);
auto port = local_port(*acceptor);
if (!port)
return port.error();
listening_port_ = *port;
auto doorman_uri = make_uri("tcp://doorman");
if (!doorman_uri)
return doorman_uri.error();
;
auto& mpx = mm_.mpx();
auto mgr = make_endpoint_manager(
mpx, mm_.system(),
doorman{acc_guard.release(), tcp::basp_application_factory{proxies_}});
if (auto err = mgr->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << err);
return err;
}
mpx->register_reading(mgr);
return none;
}
void tcp::stop() {
for (const auto& p : peers_)
proxies_.erase(p.first);
peers_.clear();
}
endpoint_manager_ptr tcp::connect(const uri& locator) {
auto auth = locator.authority();
auto host = auth.host;
if (auto hostname = get_if<std::string>(&host)) {
for (const auto& addr : ip::resolve(*hostname)) {
ip_endpoint ep{addr, auth.port};
auto sock = make_connected_tcp_stream_socket(ep);
if (!sock)
continue;
else
return emplace(make_node_id(*locator.authority_only()), *sock);
}
}
return nullptr;
}
endpoint_manager_ptr tcp::peer(const node_id& id) {
return get_peer(id);
}
void tcp::publish(actor handle, const uri& locator) {
auto path = locator.path();
mm_.system().registry().put(std::string(path.begin(), path.end()), handle);
}
void tcp::resolve(const uri& locator, const actor& listener) {
auto id = locator.authority_only();
if (id)
peer(make_node_id(*id))->resolve(locator, listener);
else
anon_send(listener, error(basp::ec::invalid_locator));
}
strong_actor_ptr tcp::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 tcp::set_last_hop(node_id*) {
// nop
}
endpoint_manager_ptr tcp::get_peer(const node_id& id) {
auto i = peers_.find(id);
if (i != peers_.end())
return i->second;
return nullptr;
}
} // namespace caf::net::backend
...@@ -54,6 +54,10 @@ endpoint_manager_ptr test::peer(const node_id& id) { ...@@ -54,6 +54,10 @@ endpoint_manager_ptr test::peer(const node_id& id) {
return get_peer(id).second; return get_peer(id).second;
} }
void test::publish(actor, const uri&) {
// nop
}
void test::resolve(const uri& locator, const actor& listener) { void test::resolve(const uri& locator, const actor& listener) {
auto id = locator.authority_only(); auto id = locator.authority_only();
if (id) if (id)
......
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