Commit c78d473d authored by Jakob Otto's avatar Jakob Otto

Include review feedback

parent f061df04
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#pragma once #pragma once
#include <map> #include <map>
#include <mutex>
#include "caf/detail/net_export.hpp" #include "caf/detail/net_export.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
...@@ -39,6 +40,10 @@ namespace caf::net::backend { ...@@ -39,6 +40,10 @@ namespace caf::net::backend {
/// @warning this backend is *not* thread safe. /// @warning this backend is *not* thread safe.
class CAF_NET_EXPORT tcp : public middleman_backend { class CAF_NET_EXPORT tcp : public middleman_backend {
public: public:
using peer_map = std::map<node_id, endpoint_manager_ptr>;
using emplace_return_type = std::pair<peer_map::iterator, bool>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
tcp(middleman& mm); tcp(middleman& mm);
...@@ -51,7 +56,7 @@ public: ...@@ -51,7 +56,7 @@ public:
void stop() override; void stop() override;
expected<endpoint_manager_ptr> connect(const uri& locator) override; expected<endpoint_manager_ptr> get_or_connect(const uri& locator) override;
endpoint_manager_ptr peer(const node_id& id) override; endpoint_manager_ptr peer(const node_id& id) override;
...@@ -63,19 +68,14 @@ public: ...@@ -63,19 +68,14 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
tcp_stream_socket socket(const node_id& peer_id) { uint16_t port() const noexcept override;
return socket_cast<tcp_stream_socket>(peers_[peer_id]->handle());
}
uint16_t port() const noexcept override {
return listening_port_;
}
template <class Handle> template <class Handle>
expected<endpoint_manager_ptr> expected<endpoint_manager_ptr>
emplace(const node_id& peer_id, Handle socket_handle) { emplace(const node_id& peer_id, Handle socket_handle) {
using transport_type = stream_transport<basp::application>; using transport_type = stream_transport<basp::application>;
nonblocking(socket_handle, true); if (auto err = nonblocking(socket_handle, true))
return err;
auto mpx = mm_.mpx(); auto mpx = mm_.mpx();
basp::application app{proxies_}; basp::application app{proxies_};
auto mgr = make_endpoint_manager( auto mgr = make_endpoint_manager(
...@@ -85,8 +85,15 @@ public: ...@@ -85,8 +85,15 @@ public:
return err; return err;
} }
mpx->register_reading(mgr); mpx->register_reading(mgr);
peers_[peer_id] = std::move(mgr); emplace_return_type res;
return peers_[peer_id]; {
const std::lock_guard<std::mutex> lock(lock_);
res = peers_.emplace(peer_id, mgr);
}
if (res.second)
return std::move(mgr);
else
return make_error(sec::runtime_error, "peer_id already exists");
} }
private: private:
...@@ -94,11 +101,13 @@ private: ...@@ -94,11 +101,13 @@ private:
middleman& mm_; middleman& mm_;
std::map<node_id, endpoint_manager_ptr> peers_; peer_map peers_;
proxy_registry proxies_; proxy_registry proxies_;
uint16_t listening_port_; uint16_t listening_port_;
std::mutex lock_;
}; };
} // namespace caf::net::backend } // namespace caf::net::backend
...@@ -51,7 +51,7 @@ public: ...@@ -51,7 +51,7 @@ public:
endpoint_manager_ptr peer(const node_id& id) override; endpoint_manager_ptr peer(const node_id& id) override;
expected<endpoint_manager_ptr> connect(const uri& locator) override; expected<endpoint_manager_ptr> get_or_connect(const uri& locator) override;
void resolve(const uri& locator, const actor& listener) override; void resolve(const uri& locator, const actor& listener) override;
...@@ -65,9 +65,7 @@ public: ...@@ -65,9 +65,7 @@ public:
return get_peer(peer_id).first; return get_peer(peer_id).first;
} }
uint16_t port() const noexcept override { 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);
......
...@@ -82,6 +82,8 @@ public: ...@@ -82,6 +82,8 @@ public:
// Publishes an actor. // Publishes an actor.
template <class Handle> template <class Handle>
void publish(Handle whom, const std::string& path) { void publish(Handle whom, const std::string& path) {
// TODO: Currently, we can't get the interface from the registry. Either we
// change that, or we need to associate the handle with the interface.
system().registry().put(path, whom); system().registry().put(path, whom);
} }
...@@ -89,18 +91,19 @@ public: ...@@ -89,18 +91,19 @@ public:
void resolve(const uri& locator, const actor& listener); void resolve(const uri& locator, const actor& listener);
template <class Handle = actor, class Duration = std::chrono::seconds> template <class Handle = actor, class Duration = std::chrono::seconds>
expected<Handle> remote_actor(const uri& locator, Duration timeout_duration expected<Handle>
= std::chrono::seconds(5)) { remote_actor(const uri& locator, Duration timeout = std::chrono::seconds(5)) {
scoped_actor self{sys_}; scoped_actor self{sys_};
resolve(locator, self); resolve(locator, self);
Handle handle; Handle handle;
error err; error err;
self->receive( self->receive(
[&handle](strong_actor_ptr& ptr, const std::set<std::string>&) { [&handle](strong_actor_ptr& ptr, const std::set<std::string>&) {
// TODO: This cast is not type-safe.
handle = actor_cast<Handle>(std::move(ptr)); handle = actor_cast<Handle>(std::move(ptr));
}, },
[&err](const error& e) { err = e; }, [&err](const error& e) { err = e; },
after(timeout_duration) >> after(timeout) >>
[&err] { [&err] {
err = make_error(sec::runtime_error, err = make_error(sec::runtime_error,
"manager did not respond with a proxy."); "manager did not respond with a proxy.");
......
...@@ -47,7 +47,7 @@ public: ...@@ -47,7 +47,7 @@ public:
virtual endpoint_manager_ptr peer(const node_id& id) = 0; virtual endpoint_manager_ptr peer(const node_id& id) = 0;
/// Establishes a connection to a remote node. /// Establishes a connection to a remote node.
virtual expected<endpoint_manager_ptr> connect(const uri& locator) = 0; virtual expected<endpoint_manager_ptr> get_or_connect(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;
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#include "caf/net/backend/tcp.hpp" #include "caf/net/backend/tcp.hpp"
#include <mutex>
#include <string> #include <string>
#include "caf/net/actor_proxy_impl.hpp" #include "caf/net/actor_proxy_impl.hpp"
...@@ -56,7 +57,8 @@ error tcp::init() { ...@@ -56,7 +57,8 @@ error tcp::init() {
if (!acceptor) if (!acceptor)
return acceptor.error(); return acceptor.error();
auto acc_guard = make_socket_guard(*acceptor); auto acc_guard = make_socket_guard(*acceptor);
nonblocking(acc_guard.socket(), true); if (auto err = nonblocking(acc_guard.socket(), true))
return err;
auto port = local_port(*acceptor); auto port = local_port(*acceptor);
if (!port) if (!port)
return port.error(); return port.error();
...@@ -82,17 +84,21 @@ void tcp::stop() { ...@@ -82,17 +84,21 @@ void tcp::stop() {
peers_.clear(); peers_.clear();
} }
expected<endpoint_manager_ptr> tcp::connect(const uri& locator) { expected<endpoint_manager_ptr> tcp::get_or_connect(const uri& locator) {
auto auth = locator.authority(); if (auto auth = locator.authority_only()) {
auto host = auth.host; auto id = make_node_id(*auth);
if (auto hostname = get_if<std::string>(&host)) { if (auto ptr = peer(id))
for (const auto& addr : ip::resolve(*hostname)) { return ptr;
ip_endpoint ep{addr, auth.port}; auto host = locator.authority().host;
auto sock = make_connected_tcp_stream_socket(ep); if (auto hostname = get_if<std::string>(&host)) {
if (!sock) for (const auto& addr : ip::resolve(*hostname)) {
continue; ip_endpoint ep{addr, locator.authority().port};
else auto sock = make_connected_tcp_stream_socket(ep);
return emplace(make_node_id(*locator.authority_only()), *sock); if (!sock)
continue;
else
return emplace(id, *sock);
}
} }
} }
return sec::cannot_connect_to_node; return sec::cannot_connect_to_node;
...@@ -103,12 +109,11 @@ endpoint_manager_ptr tcp::peer(const node_id& id) { ...@@ -103,12 +109,11 @@ endpoint_manager_ptr tcp::peer(const node_id& id) {
} }
void tcp::resolve(const uri& locator, const actor& listener) { void tcp::resolve(const uri& locator, const actor& listener) {
auto id = locator.authority_only(); if (auto id = locator.authority_only()) {
if (id) {
auto p = peer(make_node_id(*id)); auto p = peer(make_node_id(*id));
if (p == nullptr) { if (p == nullptr) {
CAF_LOG_INFO("connecting to " << CAF_ARG(locator)); CAF_LOG_INFO("connecting to " << CAF_ARG(locator));
auto res = connect(locator); auto res = get_or_connect(locator);
if (!res) if (!res)
anon_send(listener, error(sec::cannot_connect_to_node)); anon_send(listener, error(sec::cannot_connect_to_node));
else else
...@@ -132,7 +137,12 @@ void tcp::set_last_hop(node_id*) { ...@@ -132,7 +137,12 @@ void tcp::set_last_hop(node_id*) {
// nop // nop
} }
uint16_t tcp::port() const noexcept {
return listening_port_;
}
endpoint_manager_ptr tcp::get_peer(const node_id& id) { endpoint_manager_ptr tcp::get_peer(const node_id& id) {
const std::lock_guard<std::mutex> lock(lock_);
auto i = peers_.find(id); auto i = peers_.find(id);
if (i != peers_.end()) if (i != peers_.end())
return i->second; return i->second;
......
...@@ -55,9 +55,11 @@ endpoint_manager_ptr test::peer(const node_id& id) { ...@@ -55,9 +55,11 @@ endpoint_manager_ptr test::peer(const node_id& id) {
return get_peer(id).second; return get_peer(id).second;
} }
expected<endpoint_manager_ptr> test::connect(const uri&) { expected<endpoint_manager_ptr> test::get_or_connect(const uri& locator) {
if (auto ptr = peer(make_node_id(*locator.authority_only())))
return ptr;
return make_error(sec::runtime_error, return make_error(sec::runtime_error,
"function not implemented in test_backend"); "connecting not implemented in test backend");
} }
void test::resolve(const uri& locator, const actor& listener) { void test::resolve(const uri& locator, const actor& listener) {
...@@ -80,6 +82,10 @@ void test::set_last_hop(node_id*) { ...@@ -80,6 +82,10 @@ void test::set_last_hop(node_id*) {
// nop // nop
} }
uint16_t test::port() const noexcept {
return 0;
}
test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first, test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first,
stream_socket second) { stream_socket second) {
using transport_type = stream_transport<basp::application>; using transport_type = stream_transport<basp::application>;
......
...@@ -97,9 +97,8 @@ void* middleman::subtype_ptr() { ...@@ -97,9 +97,8 @@ void* middleman::subtype_ptr() {
} }
expected<endpoint_manager_ptr> middleman::connect(const uri& locator) { expected<endpoint_manager_ptr> middleman::connect(const uri& locator) {
auto ptr = backend(locator.scheme()); if (auto ptr = backend(locator.scheme()))
if (auto ret = ptr->connect(locator)) return ptr->get_or_connect(locator);
return ret;
else else
return basp::ec::invalid_scheme; return basp::ec::invalid_scheme;
} }
...@@ -123,8 +122,7 @@ middleman_backend* middleman::backend(string_view scheme) const noexcept { ...@@ -123,8 +122,7 @@ middleman_backend* middleman::backend(string_view scheme) const noexcept {
} }
expected<uint16_t> middleman::port(string_view scheme) const { expected<uint16_t> middleman::port(string_view scheme) const {
auto ptr = backend(scheme); if (auto ptr = backend(scheme))
if (ptr != nullptr)
return ptr->port(); return ptr->port();
else else
return basp::ec::invalid_scheme; return basp::ec::invalid_scheme;
......
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