Commit 9e05460c authored by Matthias Vallentin's avatar Matthias Vallentin Committed by Dominik Charousset

Allow actor publishing at random high-level ports.

Using 0 as port parameter to io::publish instructs the OS to choose a
random high-level port available for binding. The user gets the actual
bound port as return value from io::publish, which used to return void.
If the user enters a non-zero port, the passed port shall equal the
return value.
parent 6221cc05
......@@ -313,19 +313,19 @@ class default_multiplexer : public multiplexer {
event_handler* ptr;
};
connection_handle add_tcp_scribe(broker*, default_socket&& sock);
connection_handle add_tcp_scribe(broker*, default_socket_acceptor&& sock);
connection_handle add_tcp_scribe(broker*, native_socket fd) override;
connection_handle add_tcp_scribe(broker*, const std::string& h,
uint16_t port) override;
accept_handle add_tcp_doorman(broker*, default_socket&& sock);
accept_handle add_tcp_doorman(broker*, default_socket_acceptor&& sock);
accept_handle add_tcp_doorman(broker*, native_socket fd) override;
accept_handle add_tcp_doorman(broker*, uint16_t p, const char* h,
bool reuse_addr) override;
std::pair<accept_handle, uint16_t>
add_tcp_doorman(broker*, uint16_t p, const char* h, bool reuse_addr) override;
void dispatch_runnable(runnable_ptr ptr) override;
......@@ -776,16 +776,18 @@ void ipv4_connect(Socket& sock, const std::string& host, uint16_t port) {
sock = new_ipv4_connection(host, port);
}
default_socket_acceptor new_ipv4_acceptor(uint16_t port,
const char* addr = nullptr,
bool reuse_addr = false);
std::pair<default_socket_acceptor, uint16_t>
new_ipv4_acceptor(uint16_t port, const char* addr = nullptr,
bool reuse_addr = false);
template <class SocketAcceptor>
void ipv4_bind(SocketAcceptor& sock,
uint16_t ipv4_bind(SocketAcceptor& sock,
uint16_t port,
const char* addr = nullptr) {
CAF_LOGF_TRACE(CAF_ARG(port));
sock = new_ipv4_acceptor(port, addr);
auto acceptor = new_ipv4_acceptor(port, addr);
sock = std::move(acceptor.first);
return acceptor.second;
}
} // namespace network
......
......@@ -74,9 +74,9 @@ class multiplexer {
* Tries to create a new TCP doorman running on port `p`, optionally
* accepting only connections from IP address `in`.
*/
virtual accept_handle add_tcp_doorman(broker* ptr, uint16_t port,
const char* in = nullptr,
bool reuse_addr = false) = 0;
virtual std::pair<accept_handle, uint16_t>
add_tcp_doorman(broker* ptr, uint16_t port, const char* in = nullptr,
bool reuse_addr = false) = 0;
/**
* Simple wrapper for runnables
......
......@@ -29,34 +29,38 @@
namespace caf {
namespace io {
void publish_impl(abstract_actor_ptr whom, uint16_t port,
const char* in, bool reuse_addr);
uint16_t publish_impl(abstract_actor_ptr whom, uint16_t port,
const char* in, bool reuse_addr);
/**
* Publishes `whom` at `port`. The connection is managed by the middleman.
* @param whom Actor that should be published at `port`.
* @param port Unused TCP port.
* @param in The IP address to listen to or `INADDR_ANY` if `in == nullptr`.
* @returns The actual port the OS uses after `bind()`. If `port == 0` the OS
* chooses a random high-level port.
* @throws bind_failure
*/
inline void publish(caf::actor whom, uint16_t port, const char* in = nullptr,
bool reuse_addr = false) {
inline uint16_t publish(caf::actor whom, uint16_t port,
const char* in = nullptr, bool reuse_addr = false) {
if (!whom) {
return;
return 0;
}
publish_impl(actor_cast<abstract_actor_ptr>(whom), port, in, reuse_addr);
return publish_impl(actor_cast<abstract_actor_ptr>(whom), port, in,
reuse_addr);
}
/**
* @copydoc publish(actor,uint16_t,const char*)
*/
template <class... Rs>
void typed_publish(typed_actor<Rs...> whom, uint16_t port,
const char* in = nullptr, bool reuse_addr = false) {
uint16_t typed_publish(typed_actor<Rs...> whom, uint16_t port,
const char* in = nullptr, bool reuse_addr = false) {
if (!whom) {
return;
return 0;
}
publish_impl(actor_cast<abstract_actor_ptr>(whom), port, in, reuse_addr);
return publish_impl(actor_cast<abstract_actor_ptr>(whom), port, in,
reuse_addr);
}
} // namespace io
......
......@@ -27,10 +27,12 @@ namespace io {
/**
* Makes *all* local groups accessible via network on address `addr` and `port`.
* @returns The actual port the OS uses after `bind()`. If `port == 0` the OS
* chooses a random high-level port.
* @throws bind_failure
* @throws network_error
*/
void publish_local_groups(uint16_t port, const char* addr = nullptr);
uint16_t publish_local_groups(uint16_t port, const char* addr = nullptr);
} // namespace io
} // namespace caf
......
......@@ -767,10 +767,12 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self,
return add_tcp_doorman(self, default_socket_acceptor{*this, fd});
}
accept_handle default_multiplexer::add_tcp_doorman(broker* self, uint16_t port,
const char* host,
bool reuse_addr) {
return add_tcp_doorman(self, new_ipv4_acceptor(port, host, reuse_addr));
std::pair<accept_handle, uint16_t>
default_multiplexer::add_tcp_doorman(broker* self, uint16_t port,
const char* host, bool reuse_addr) {
auto acceptor = new_ipv4_acceptor(port, host, reuse_addr);
auto bound_port = acceptor.second;
return {add_tcp_doorman(self, std::move(acceptor.first)), bound_port};
}
/******************************************************************************
......@@ -949,8 +951,8 @@ default_socket new_ipv4_connection(const std::string& host, uint16_t port) {
return default_socket{backend, new_ipv4_connection_impl(host, port)};
}
native_socket new_ipv4_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr) {
std::pair<native_socket, uint16_t>
new_ipv4_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr) {
CAF_LOGF_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
# ifdef CAF_WINDOWS
// make sure TCP has been initialized via WSAStartup
......@@ -983,20 +985,29 @@ native_socket new_ipv4_acceptor_impl(uint16_t port, const char* addr,
throw bind_failure(last_socket_error_as_string());
}
if (listen(fd, SOMAXCONN) != 0) {
throw network_error("listen() failed");
throw network_error("listen() failed: " + last_socket_error_as_string());
}
if (port == 0) {
socklen_t len = sizeof(serv_addr);
if (getsockname(fd, reinterpret_cast<sockaddr*>(&serv_addr), &len) < 0) {
throw network_error("getsockname(): " + last_socket_error_as_string());
}
}
// ok, no exceptions so far
sguard.release();
CAF_LOGF_DEBUG("sockfd = " << fd);
return fd;
CAF_LOGF_DEBUG("sockfd = " << fd << ", port = " << ntohs(serv_addr.sin_port));
return {fd, ntohs(serv_addr.sin_port)};
}
default_socket_acceptor new_ipv4_acceptor(uint16_t port, const char* addr,
bool reuse) {
std::pair<default_socket_acceptor, uint16_t>
new_ipv4_acceptor(uint16_t port, const char* addr, bool reuse) {
CAF_LOGF_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
auto& backend = get_multiplexer_singleton();
return default_socket_acceptor{backend,
new_ipv4_acceptor_impl(port, addr, reuse)};
auto acceptor = new_ipv4_acceptor_impl(port, addr, reuse);
auto bound_port = acceptor.second;
CAF_REQUIRE(bound_port == port);
return {default_socket_acceptor{backend, std::move(acceptor.first)},
bound_port};
}
} // namespace network
......
......@@ -33,8 +33,8 @@
namespace caf {
namespace io {
void publish_impl(abstract_actor_ptr whom, uint16_t port,
const char* in, bool reuse_addr) {
uint16_t publish_impl(abstract_actor_ptr whom, uint16_t port,
const char* in, bool reuse_addr) {
using namespace detail;
auto mm = middleman::instance();
scoped_actor self;
......@@ -43,9 +43,9 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port,
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
try {
auto hdl = mm->backend().add_tcp_doorman(bro.get(), port, in, reuse_addr);
bro->add_published_actor(hdl, whom, port);
mm->notify<hook::actor_published>(whom->address(), port);
anon_send(selfhdl, atom("OK"));
bro->add_published_actor(std::move(hdl.first), whom, hdl.second);
mm->notify<hook::actor_published>(whom->address(), hdl.second);
anon_send(selfhdl, atom("OK"), hdl.second);
}
catch (bind_failure& e) {
anon_send(selfhdl, atom("BIND_FAIL"), e.what());
......@@ -54,10 +54,11 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port,
anon_send(selfhdl, atom("ERROR"), e.what());
}
});
uint16_t bound_port = 0;
// block caller and re-throw exception here in case of an error
self->receive(
on(atom("OK")) >> [] {
// success
on(atom("OK"), arg_match) >> [&](uint16_t p) {
bound_port = p;
},
on(atom("BIND_FAIL"), arg_match) >> [](std::string& str) {
throw bind_failure(std::move(str));
......@@ -66,6 +67,7 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port,
throw network_error(std::move(str));
}
);
return bound_port;
}
} // namespace io
......
......@@ -39,10 +39,10 @@ struct group_nameserver : event_based_actor {
} // namespace <anonymous>
void publish_local_groups(uint16_t port, const char* addr) {
uint16_t publish_local_groups(uint16_t port, const char* addr) {
auto gn = spawn<group_nameserver, hidden>();
try {
publish(gn, port, addr);
return publish(gn, port, addr);
}
catch (std::exception&) {
anon_send_exit(gn, exit_reason::user_shutdown);
......
......@@ -327,49 +327,31 @@ class server : public event_based_actor {
};
template <class F>
uint16_t at_some_port(uint16_t first_port, F fun) {
auto port = first_port;
for (;;) {
try {
fun(port);
return port;
}
catch (bind_failure&) {
// try next port
++port;
}
}
}
void test_remote_actor(const char* app_path, bool run_remote_actor) {
scoped_actor self;
auto serv = self->spawn<server, monitored>();
auto publish_serv = [=](uint16_t p) {
io::publish(serv, p, "127.0.0.1");
};
auto publish_groups = [](uint16_t p) {
io::publish_local_groups(p);
};
// publish on two distinct ports and use the latter one afterwards
auto port0 = at_some_port(4242, publish_serv);
CAF_LOGF_INFO("first publish succeeded on port " << port0);
auto port = at_some_port(port0 + 1, publish_serv);
CAF_PRINT("running on port " << port);
CAF_LOGF_INFO("running on port " << port);
auto port1 = io::publish(serv, 0, "127.0.0.1");
CAF_CHECK(port1 > 0);
CAF_PRINT("first publish succeeded on port " << port1);
auto port2 = io::publish(serv, 0, "127.0.0.1");
CAF_CHECK(port2 > 0);
CAF_PRINT("second publish succeeded on port " << port1);
CAF_LOGF_INFO("running on port " << port2);
// publish local groups as well
auto gport = at_some_port(port + 1, publish_groups);
auto gport = io::publish_local_groups(0);
CAF_CHECK(gport > 0);
// check whether accessing local actors via io::remote_actors works correctly,
// i.e., does not return a proxy instance
auto serv2 = io::remote_actor("127.0.0.1", port);
auto serv2 = io::remote_actor("127.0.0.1", port2);
CAF_CHECK(serv2 != invalid_actor && !serv2->is_remote());
CAF_CHECK(serv == serv2);
thread child;
if (run_remote_actor) {
child = run_program(app_path, "-c", port, port0, gport);
child = run_program(app_path, "-c", port2, port1, gport);
} else {
CAF_PRINT("please run client with: "
<< "-c " << port << " " << port0 << " " << gport);
<< "-c " << port2 << " " << port1 << " " << gport);
}
CAF_CHECKPOINT();
self->receive(
......
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