Commit bdd63e7b authored by Dominik Charousset's avatar Dominik Charousset

Add new flag to `publish` to enable `SO_REUSEADDR`

parent b056bc9f
...@@ -324,7 +324,8 @@ class default_multiplexer : public multiplexer { ...@@ -324,7 +324,8 @@ class default_multiplexer : public multiplexer {
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) override; accept_handle 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;
...@@ -775,10 +776,9 @@ void ipv4_connect(Socket& sock, const std::string& host, uint16_t port) { ...@@ -775,10 +776,9 @@ void ipv4_connect(Socket& sock, const std::string& host, uint16_t port) {
sock = new_ipv4_connection(host, port); sock = new_ipv4_connection(host, port);
} }
native_socket new_ipv4_acceptor_impl(uint16_t port, const char* addr);
default_socket_acceptor new_ipv4_acceptor(uint16_t port, default_socket_acceptor new_ipv4_acceptor(uint16_t port,
const char* addr = nullptr); const char* addr = nullptr,
bool reuse_addr = false);
template <class SocketAcceptor> template <class SocketAcceptor>
void ipv4_bind(SocketAcceptor& sock, void ipv4_bind(SocketAcceptor& sock,
......
...@@ -75,7 +75,8 @@ class multiplexer { ...@@ -75,7 +75,8 @@ class multiplexer {
* 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 accept_handle add_tcp_doorman(broker* ptr, uint16_t port,
const char* in = nullptr) = 0; const char* in = nullptr,
bool reuse_addr = false) = 0;
/** /**
* Simple wrapper for runnables * Simple wrapper for runnables
......
...@@ -29,7 +29,8 @@ ...@@ -29,7 +29,8 @@
namespace caf { namespace caf {
namespace io { namespace io {
void publish_impl(abstract_actor_ptr whom, uint16_t port, const char* in); void 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. * Publishes `whom` at `port`. The connection is managed by the middleman.
...@@ -38,11 +39,12 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port, const char* in); ...@@ -38,11 +39,12 @@ void publish_impl(abstract_actor_ptr whom, uint16_t port, const char* in);
* @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`.
* @throws bind_failure * @throws bind_failure
*/ */
inline void publish(caf::actor whom, uint16_t port, const char* in = nullptr) { inline void publish(caf::actor whom, uint16_t port, const char* in = nullptr,
bool reuse_addr = false) {
if (!whom) { if (!whom) {
return; return;
} }
publish_impl(actor_cast<abstract_actor_ptr>(whom), port, in); publish_impl(actor_cast<abstract_actor_ptr>(whom), port, in, reuse_addr);
} }
/** /**
...@@ -50,11 +52,11 @@ inline void publish(caf::actor whom, uint16_t port, const char* in = nullptr) { ...@@ -50,11 +52,11 @@ inline void publish(caf::actor whom, uint16_t port, const char* in = nullptr) {
*/ */
template <class... Rs> template <class... Rs>
void typed_publish(typed_actor<Rs...> whom, uint16_t port, void typed_publish(typed_actor<Rs...> whom, uint16_t port,
const char* in = nullptr) { const char* in = nullptr, bool reuse_addr = false) {
if (!whom) { if (!whom) {
return; return;
} }
publish_impl(actor_cast<abstract_actor_ptr>(whom), port, in); publish_impl(actor_cast<abstract_actor_ptr>(whom), port, in, reuse_addr);
} }
} // namespace io } // namespace io
......
...@@ -767,10 +767,10 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self, ...@@ -767,10 +767,10 @@ 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, accept_handle default_multiplexer::add_tcp_doorman(broker* self, uint16_t port,
uint16_t port, const char* host,
const char* host) { bool reuse_addr) {
return add_tcp_doorman(self, new_ipv4_acceptor(port, host)); return add_tcp_doorman(self, new_ipv4_acceptor(port, host, reuse_addr));
} }
/****************************************************************************** /******************************************************************************
...@@ -949,9 +949,9 @@ default_socket new_ipv4_connection(const std::string& host, uint16_t port) { ...@@ -949,9 +949,9 @@ 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) { native_socket new_ipv4_acceptor_impl(uint16_t port, const char* addr,
CAF_LOGF_TRACE(CAF_ARG(port) bool reuse_addr) {
<< ", 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
get_multiplexer_singleton(); get_multiplexer_singleton();
...@@ -962,10 +962,12 @@ native_socket new_ipv4_acceptor_impl(uint16_t port, const char* addr) { ...@@ -962,10 +962,12 @@ native_socket new_ipv4_acceptor_impl(uint16_t port, const char* addr) {
} }
// sguard closes the socket in case of exception // sguard closes the socket in case of exception
socket_guard sguard(fd); socket_guard sguard(fd);
int on = 1; if (reuse_addr) {
if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, int on = 1;
reinterpret_cast<setsockopt_ptr>(&on), sizeof(on)) < 0) { if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
throw_io_failure("unable to set SO_REUSEADDR"); reinterpret_cast<setsockopt_ptr>(&on), sizeof(on)) < 0) {
throw_io_failure("unable to set SO_REUSEADDR");
}
} }
struct sockaddr_in serv_addr; struct sockaddr_in serv_addr;
memset(&serv_addr, 0, sizeof(serv_addr)); memset(&serv_addr, 0, sizeof(serv_addr));
...@@ -989,10 +991,12 @@ native_socket new_ipv4_acceptor_impl(uint16_t port, const char* addr) { ...@@ -989,10 +991,12 @@ native_socket new_ipv4_acceptor_impl(uint16_t port, const char* addr) {
return fd; return fd;
} }
default_socket_acceptor new_ipv4_acceptor(uint16_t port, const char* addr) { default_socket_acceptor 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, new_ipv4_acceptor_impl(port, addr)}; return default_socket_acceptor{backend,
new_ipv4_acceptor_impl(port, addr, reuse)};
} }
} // namespace network } // namespace network
......
...@@ -32,14 +32,15 @@ ...@@ -32,14 +32,15 @@
namespace caf { namespace caf {
namespace io { namespace io {
void publish_impl(abstract_actor_ptr whom, uint16_t port, const char* in) { void publish_impl(abstract_actor_ptr whom, uint16_t port,
const char* in, bool reuse_addr) {
using namespace detail; using namespace detail;
auto mm = middleman::instance(); auto mm = middleman::instance();
std::promise<bool> res; std::promise<bool> res;
mm->run_later([&] { mm->run_later([&] {
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); auto hdl = mm->backend().add_tcp_doorman(bro.get(), port, in, reuse_addr);
bro->add_published_actor(hdl, whom, port); bro->add_published_actor(hdl, whom, port);
mm->notify<hook::actor_published>(whom->address(), port); mm->notify<hook::actor_published>(whom->address(), port);
res.set_value(true); res.set_value(true);
......
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