Commit 35c394b0 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/publish-at-random-port' into develop

parents 6221cc05 7611fb6b
...@@ -313,19 +313,19 @@ class default_multiplexer : public multiplexer { ...@@ -313,19 +313,19 @@ class default_multiplexer : public multiplexer {
event_handler* ptr; 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*, native_socket fd) override;
connection_handle add_tcp_scribe(broker*, const std::string& h, connection_handle add_tcp_scribe(broker*, const std::string& h,
uint16_t port) override; 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*, native_socket fd) override;
accept_handle add_tcp_doorman(broker*, uint16_t p, const char* h, std::pair<accept_handle, uint16_t>
bool reuse_addr) override; add_tcp_doorman(broker*, uint16_t p, const char* h, bool reuse_addr) override;
void dispatch_runnable(runnable_ptr ptr) override; void dispatch_runnable(runnable_ptr ptr) override;
...@@ -776,16 +776,18 @@ void ipv4_connect(Socket& sock, const std::string& host, uint16_t port) { ...@@ -776,16 +776,18 @@ void ipv4_connect(Socket& sock, const std::string& host, uint16_t port) {
sock = new_ipv4_connection(host, port); sock = new_ipv4_connection(host, port);
} }
default_socket_acceptor new_ipv4_acceptor(uint16_t port, std::pair<default_socket_acceptor, uint16_t>
const char* addr = nullptr, new_ipv4_acceptor(uint16_t port, const char* addr = nullptr,
bool reuse_addr = false); bool reuse_addr = false);
template <class SocketAcceptor> template <class SocketAcceptor>
void ipv4_bind(SocketAcceptor& sock, uint16_t ipv4_bind(SocketAcceptor& sock,
uint16_t port, uint16_t port,
const char* addr = nullptr) { const char* addr = nullptr) {
CAF_LOGF_TRACE(CAF_ARG(port)); 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 } // namespace network
......
...@@ -74,8 +74,8 @@ class multiplexer { ...@@ -74,8 +74,8 @@ class multiplexer {
* Tries to create a new TCP doorman running on port `p`, optionally * Tries to create a new TCP doorman running on port `p`, optionally
* accepting only connections from IP address `in`. * accepting only connections from IP address `in`.
*/ */
virtual accept_handle add_tcp_doorman(broker* ptr, uint16_t port, virtual std::pair<accept_handle, uint16_t>
const char* in = nullptr, add_tcp_doorman(broker* ptr, uint16_t port, const char* in = nullptr,
bool reuse_addr = false) = 0; bool reuse_addr = false) = 0;
/** /**
......
...@@ -29,7 +29,7 @@ ...@@ -29,7 +29,7 @@
namespace caf { namespace caf {
namespace io { namespace io {
void publish_impl(abstract_actor_ptr whom, uint16_t port, uint16_t publish_impl(abstract_actor_ptr whom, uint16_t port,
const char* in, bool reuse_addr); const char* in, bool reuse_addr);
/** /**
...@@ -37,26 +37,30 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port, ...@@ -37,26 +37,30 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port,
* @param whom Actor that should be published at `port`. * @param whom Actor that should be published at `port`.
* @param port Unused TCP port. * @param port Unused TCP port.
* @param in The IP address to listen to or `INADDR_ANY` if `in == nullptr`. * @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 * @throws bind_failure
*/ */
inline void publish(caf::actor whom, uint16_t port, const char* in = nullptr, inline uint16_t publish(caf::actor whom, uint16_t port,
bool reuse_addr = false) { const char* in = nullptr, bool reuse_addr = false) {
if (!whom) { 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*) * @copydoc publish(actor,uint16_t,const char*)
*/ */
template <class... Rs> template <class... Rs>
void typed_publish(typed_actor<Rs...> whom, uint16_t port, uint16_t typed_publish(typed_actor<Rs...> whom, uint16_t port,
const char* in = nullptr, bool reuse_addr = false) { const char* in = nullptr, bool reuse_addr = false) {
if (!whom) { 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 } // namespace io
......
...@@ -27,10 +27,12 @@ namespace io { ...@@ -27,10 +27,12 @@ namespace io {
/** /**
* Makes *all* local groups accessible via network on address `addr` and `port`. * 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 bind_failure
* @throws network_error * @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 io
} // namespace caf } // namespace caf
......
...@@ -767,10 +767,12 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self, ...@@ -767,10 +767,12 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self,
return add_tcp_doorman(self, default_socket_acceptor{*this, fd}); return add_tcp_doorman(self, default_socket_acceptor{*this, fd});
} }
accept_handle default_multiplexer::add_tcp_doorman(broker* self, uint16_t port, std::pair<accept_handle, uint16_t>
const char* host, default_multiplexer::add_tcp_doorman(broker* self, uint16_t port,
bool reuse_addr) { const char* host, bool reuse_addr) {
return add_tcp_doorman(self, new_ipv4_acceptor(port, host, 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) { ...@@ -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)}; return default_socket{backend, new_ipv4_connection_impl(host, port)};
} }
native_socket new_ipv4_acceptor_impl(uint16_t port, const char* addr, std::pair<native_socket, uint16_t>
bool reuse_addr) { new_ipv4_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr) {
CAF_LOGF_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr")); CAF_LOGF_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
# ifdef CAF_WINDOWS # ifdef CAF_WINDOWS
// make sure TCP has been initialized via WSAStartup // make sure TCP has been initialized via WSAStartup
...@@ -983,20 +985,29 @@ native_socket new_ipv4_acceptor_impl(uint16_t port, const char* addr, ...@@ -983,20 +985,29 @@ native_socket new_ipv4_acceptor_impl(uint16_t port, const char* addr,
throw bind_failure(last_socket_error_as_string()); throw bind_failure(last_socket_error_as_string());
} }
if (listen(fd, SOMAXCONN) != 0) { 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 // ok, no exceptions so far
sguard.release(); sguard.release();
CAF_LOGF_DEBUG("sockfd = " << fd); CAF_LOGF_DEBUG("sockfd = " << fd << ", port = " << ntohs(serv_addr.sin_port));
return fd; return {fd, ntohs(serv_addr.sin_port)};
} }
default_socket_acceptor new_ipv4_acceptor(uint16_t port, const char* addr, std::pair<default_socket_acceptor, uint16_t>
bool reuse) { new_ipv4_acceptor(uint16_t port, const char* addr, bool reuse) {
CAF_LOGF_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr")); CAF_LOGF_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
auto& backend = get_multiplexer_singleton(); auto& backend = get_multiplexer_singleton();
return default_socket_acceptor{backend, auto acceptor = new_ipv4_acceptor_impl(port, addr, reuse);
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 } // namespace network
......
...@@ -33,7 +33,7 @@ ...@@ -33,7 +33,7 @@
namespace caf { namespace caf {
namespace io { namespace io {
void publish_impl(abstract_actor_ptr whom, uint16_t port, uint16_t publish_impl(abstract_actor_ptr whom, uint16_t port,
const char* in, bool reuse_addr) { const char* in, bool reuse_addr) {
using namespace detail; using namespace detail;
auto mm = middleman::instance(); auto mm = middleman::instance();
...@@ -43,9 +43,9 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port, ...@@ -43,9 +43,9 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port,
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP")); auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
try { try {
auto hdl = mm->backend().add_tcp_doorman(bro.get(), port, in, reuse_addr); auto hdl = mm->backend().add_tcp_doorman(bro.get(), port, in, reuse_addr);
bro->add_published_actor(hdl, whom, port); bro->add_published_actor(std::move(hdl.first), whom, hdl.second);
mm->notify<hook::actor_published>(whom->address(), port); mm->notify<hook::actor_published>(whom->address(), hdl.second);
anon_send(selfhdl, atom("OK")); anon_send(selfhdl, atom("OK"), hdl.second);
} }
catch (bind_failure& e) { catch (bind_failure& e) {
anon_send(selfhdl, atom("BIND_FAIL"), e.what()); anon_send(selfhdl, atom("BIND_FAIL"), e.what());
...@@ -54,10 +54,11 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port, ...@@ -54,10 +54,11 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port,
anon_send(selfhdl, atom("ERROR"), e.what()); anon_send(selfhdl, atom("ERROR"), e.what());
} }
}); });
uint16_t bound_port = 0;
// block caller and re-throw exception here in case of an error // block caller and re-throw exception here in case of an error
self->receive( self->receive(
on(atom("OK")) >> [] { on(atom("OK"), arg_match) >> [&](uint16_t p) {
// success bound_port = p;
}, },
on(atom("BIND_FAIL"), arg_match) >> [](std::string& str) { on(atom("BIND_FAIL"), arg_match) >> [](std::string& str) {
throw bind_failure(std::move(str)); throw bind_failure(std::move(str));
...@@ -66,6 +67,7 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port, ...@@ -66,6 +67,7 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port,
throw network_error(std::move(str)); throw network_error(std::move(str));
} }
); );
return bound_port;
} }
} // namespace io } // namespace io
......
...@@ -39,10 +39,10 @@ struct group_nameserver : event_based_actor { ...@@ -39,10 +39,10 @@ struct group_nameserver : event_based_actor {
} // namespace <anonymous> } // 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>(); auto gn = spawn<group_nameserver, hidden>();
try { try {
publish(gn, port, addr); return publish(gn, port, addr);
} }
catch (std::exception&) { catch (std::exception&) {
anon_send_exit(gn, exit_reason::user_shutdown); anon_send_exit(gn, exit_reason::user_shutdown);
......
...@@ -7,13 +7,21 @@ All functions shown in this section can be accessed by including the header \lst ...@@ -7,13 +7,21 @@ All functions shown in this section can be accessed by including the header \lst
\subsection{Publishing of Actors} \subsection{Publishing of Actors}
\begin{lstlisting} \begin{lstlisting}
void publish(actor whom, std::uint16_t port, const char* addr = 0) uint16_t publish(actor whom, uint16_t port, const char* addr = 0, bool reuse_addr = false)
\end{lstlisting} \end{lstlisting}
The function \lstinline^publish^ binds an actor to a given port. The function \lstinline^publish^ binds an actor to a given port.
It throws \lstinline^network_error^ if socket related errors occur or \lstinline^bind_failure^ if the specified port is already in use. To choose the next high-level port available available for binding, one can specify \lstinline^port == 0^ and retrieves the bound port as return value.
The optional \lstinline^addr^ parameter can be used to listen only to the given IP address. The return value is equal to \lstinline^port^ if \lstinline^port != 0^.
The function throws \lstinline^network_error^ if socket related errors occur or \lstinline^bind_failure^ if the specified port is already in use.
The optional \lstinline^addr^ parameter can be used to listen only to the given address.
Otherwise, the actor accepts all incoming connections (\lstinline^INADDR_ANY^). Otherwise, the actor accepts all incoming connections (\lstinline^INADDR_ANY^).
The flag \lstinline^reuse_addr^ controls the behavior when binding an IP
address to a port, with the same semantics as the BSD socket flag \lstinline^SO_REUSEADDR^.
For example, if \lstinline^reuse_addr = false^, binding two sockets to 0.0.0.0:42 and 10.0.0.1:42 will fail with \texttt{EADDRINUSE} since 0.0.0.0 includes 10.0.0.1.
With \lstinline^reuse_addr = true^ binding would succeed because 10.0.0.1 and
0.0.0.0 are not literally equal addresses.
\begin{lstlisting} \begin{lstlisting}
publish(self, 4242); publish(self, 4242);
......
...@@ -327,49 +327,31 @@ class server : public event_based_actor { ...@@ -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) { void test_remote_actor(const char* app_path, bool run_remote_actor) {
scoped_actor self; scoped_actor self;
auto serv = self->spawn<server, monitored>(); 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 // publish on two distinct ports and use the latter one afterwards
auto port0 = at_some_port(4242, publish_serv); auto port1 = io::publish(serv, 0, "127.0.0.1");
CAF_LOGF_INFO("first publish succeeded on port " << port0); CAF_CHECK(port1 > 0);
auto port = at_some_port(port0 + 1, publish_serv); CAF_PRINT("first publish succeeded on port " << port1);
CAF_PRINT("running on port " << port); auto port2 = io::publish(serv, 0, "127.0.0.1");
CAF_LOGF_INFO("running on port " << port); CAF_CHECK(port2 > 0);
CAF_PRINT("second publish succeeded on port " << port1);
CAF_LOGF_INFO("running on port " << port2);
// publish local groups as well // 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, // check whether accessing local actors via io::remote_actors works correctly,
// i.e., does not return a proxy instance // 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(serv2 != invalid_actor && !serv2->is_remote());
CAF_CHECK(serv == serv2); CAF_CHECK(serv == serv2);
thread child; thread child;
if (run_remote_actor) { if (run_remote_actor) {
child = run_program(app_path, "-c", port, port0, gport); child = run_program(app_path, "-c", port2, port1, gport);
} else { } else {
CAF_PRINT("please run client with: " CAF_PRINT("please run client with: "
<< "-c " << port << " " << port0 << " " << gport); << "-c " << port2 << " " << port1 << " " << gport);
} }
CAF_CHECKPOINT(); CAF_CHECKPOINT();
self->receive( 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