Commit 41957aa3 authored by Marian Triebe's avatar Marian Triebe

Add support for dualstack IPv6

Closes #174, closes #156
parent 57ca0479
...@@ -419,7 +419,6 @@ class default_multiplexer : public multiplexer { ...@@ -419,7 +419,6 @@ class default_multiplexer : public multiplexer {
std::pair<native_socket, native_socket> m_pipe; std::pair<native_socket, native_socket> m_pipe;
std::thread::id m_tid; std::thread::id m_tid;
}; };
default_multiplexer& get_multiplexer_singleton(); default_multiplexer& get_multiplexer_singleton();
...@@ -658,20 +657,20 @@ class stream : public event_handler { ...@@ -658,20 +657,20 @@ class stream : public event_handler {
} }
// reading & writing // reading & writing
Socket m_sock; Socket m_sock;
// reading // reading
manager_ptr m_reader; manager_ptr m_reader;
size_t m_threshold; size_t m_threshold;
size_t m_collected; size_t m_collected;
size_t m_max; size_t m_max;
receive_policy_flag m_rd_flag; receive_policy_flag m_rd_flag;
buffer_type m_rd_buf; buffer_type m_rd_buf;
// writing // writing
manager_ptr m_writer; manager_ptr m_writer;
bool m_writing; bool m_writing;
size_t m_written; size_t m_written;
buffer_type m_wr_buf; buffer_type m_wr_buf;
buffer_type m_wr_offline_buf; buffer_type m_wr_offline_buf;
}; };
/** /**
...@@ -763,7 +762,9 @@ class acceptor : public event_handler { ...@@ -763,7 +762,9 @@ class acceptor : public event_handler {
void removed_from_loop(operation op) override { void removed_from_loop(operation op) override {
CAF_LOG_TRACE("m_accept_sock.fd = " << m_accept_sock.fd() CAF_LOG_TRACE("m_accept_sock.fd = " << m_accept_sock.fd()
<< "op = " << static_cast<int>(op)); << "op = " << static_cast<int>(op));
if (op == operation::read) m_mgr.reset(); if (op == operation::read) {
m_mgr.reset();
}
} }
protected: protected:
...@@ -780,31 +781,17 @@ class acceptor : public event_handler { ...@@ -780,31 +781,17 @@ class acceptor : public event_handler {
}; };
native_socket new_ipv4_connection_impl(const std::string&, uint16_t); native_socket new_tcp_connection_impl(const std::string&, uint16_t,
optional<protocol> preferred = none);
default_socket new_ipv4_connection(const std::string& host, uint16_t port); default_socket new_tcp_connection(const std::string& host, uint16_t port);
template <class Socket>
void ipv4_connect(Socket& sock, const std::string& host, uint16_t port) {
sock = new_ipv4_connection(host, port);
}
std::pair<native_socket, uint16_t> std::pair<native_socket, uint16_t>
new_ipv4_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr); new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr);
std::pair<default_socket_acceptor, uint16_t> std::pair<default_socket_acceptor, uint16_t>
new_ipv4_acceptor(uint16_t port, const char* addr = nullptr, new_tcp_acceptor(uint16_t port, const char* addr = nullptr,
bool reuse_addr = false); bool reuse_addr = false);
template <class SocketAcceptor>
uint16_t ipv4_bind(SocketAcceptor& sock,
uint16_t port,
const char* addr = nullptr) {
CAF_LOGF_TRACE(CAF_ARG(port));
auto acceptor = new_ipv4_acceptor(port, addr);
sock = std::move(acceptor.first);
return acceptor.second;
}
} // namespace network } // namespace network
} // namespace io } // namespace io
......
...@@ -25,6 +25,8 @@ ...@@ -25,6 +25,8 @@
#include <string> #include <string>
#include <utility> #include <utility>
#include "caf/optional.hpp"
#include "caf/io/network/protocol.hpp" #include "caf/io/network/protocol.hpp"
namespace caf { namespace caf {
...@@ -57,6 +59,12 @@ class interfaces { ...@@ -57,6 +59,12 @@ class interfaces {
*/ */
static std::vector<std::string> list_addresses(protocol proc, static std::vector<std::string> list_addresses(protocol proc,
bool include_localhost = true); bool include_localhost = true);
/**
* Returns `pair<string, protocol>` for given host address.
**/
static optional<std::pair<std::string, protocol>>
get_addrinfo_of_host(const std::string& host, optional<protocol> preferred = none);
}; };
} // namespace network } // namespace network
......
...@@ -20,11 +20,15 @@ ...@@ -20,11 +20,15 @@
#include "caf/io/network/default_multiplexer.hpp" #include "caf/io/network/default_multiplexer.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/optional.hpp"
#include "caf/exception.hpp" #include "caf/exception.hpp"
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/network/protocol.hpp"
#include "caf/io/network/interfaces.hpp"
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
# include <winsock2.h> # include <winsock2.h>
# include <ws2tcpip.h> /* socklen_t, et al (MSVC20xx) */ # include <ws2tcpip.h> /* socklen_t, et al (MSVC20xx) */
...@@ -51,6 +55,8 @@ namespace { ...@@ -51,6 +55,8 @@ namespace {
#else // BSD or Linux #else // BSD or Linux
constexpr int no_sigpipe_flag = MSG_NOSIGNAL; constexpr int no_sigpipe_flag = MSG_NOSIGNAL;
#endif #endif
constexpr auto ipv4 = caf::io::network::protocol::ipv4;
constexpr auto ipv6 = caf::io::network::protocol::ipv6;
} // namespace <anonymous> } // namespace <anonymous>
namespace caf { namespace caf {
...@@ -768,7 +774,7 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self, ...@@ -768,7 +774,7 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self,
connection_handle default_multiplexer::new_tcp_scribe(const std::string& host, connection_handle default_multiplexer::new_tcp_scribe(const std::string& host,
uint16_t port) { uint16_t port) {
auto fd = new_ipv4_connection_impl(host, port); auto fd = new_tcp_connection_impl(host, port);
return connection_handle::from_int(int64_from_native_socket(fd)); return connection_handle::from_int(int64_from_native_socket(fd));
} }
...@@ -785,14 +791,14 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self, ...@@ -785,14 +791,14 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
connection_handle default_multiplexer::add_tcp_scribe(broker* self, connection_handle default_multiplexer::add_tcp_scribe(broker* self,
const std::string& host, const std::string& host,
uint16_t port) { uint16_t port) {
return add_tcp_scribe(self, new_ipv4_connection(host, port)); return add_tcp_scribe(self, new_tcp_connection(host, port));
} }
std::pair<accept_handle, uint16_t> std::pair<accept_handle, uint16_t>
default_multiplexer::new_tcp_doorman(uint16_t port, const char* in, default_multiplexer::new_tcp_doorman(uint16_t port, const char* in,
bool reuse_addr) { bool reuse_addr) {
auto res = new_ipv4_acceptor_impl(port, in, reuse_addr); auto res = new_tcp_acceptor_impl(port, in, reuse_addr);
return {accept_handle::from_int(int64_from_native_socket(res.first)), return {accept_handle::from_int(int64_from_native_socket(res.first)),
res.second}; res.second};
} }
...@@ -809,7 +815,7 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self, ...@@ -809,7 +815,7 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self,
std::pair<accept_handle, uint16_t> std::pair<accept_handle, uint16_t>
default_multiplexer::add_tcp_doorman(broker* self, uint16_t port, default_multiplexer::add_tcp_doorman(broker* self, uint16_t port,
const char* host, bool reuse_addr) { const char* host, bool reuse_addr) {
auto acceptor = new_ipv4_acceptor(port, host, reuse_addr); auto acceptor = new_tcp_acceptor(port, host, reuse_addr);
auto bound_port = acceptor.second; auto bound_port = acceptor.second;
return {add_tcp_doorman(self, std::move(acceptor.first)), bound_port}; return {add_tcp_doorman(self, std::move(acceptor.first)), bound_port};
} }
...@@ -943,62 +949,183 @@ struct socket_guard { ...@@ -943,62 +949,183 @@ struct socket_guard {
if (m_fd != invalid_native_socket) if (m_fd != invalid_native_socket)
closesocket(m_fd); closesocket(m_fd);
} }
void release() { native_socket release() {
auto fd = m_fd;
m_fd = invalid_native_socket; m_fd = invalid_native_socket;
return fd;
} }
private: private:
native_socket m_fd; native_socket m_fd;
}; };
native_socket new_ipv4_connection_impl(const std::string& host, uint16_t port) { in_addr& addr_of(sockaddr_in& what) {
return what.sin_addr;
}
sa_family_t& family_of(sockaddr_in& what) {
return what.sin_family;
}
in_port_t& port_of(sockaddr_in& what) {
return what.sin_port;
}
in6_addr& addr_of(sockaddr_in6& what) {
return what.sin6_addr;
}
sa_family_t& family_of(sockaddr_in6& what) {
return what.sin6_family;
}
in_port_t& port_of(sockaddr_in6& what) {
return what.sin6_port;
}
in_port_t& port_of(sockaddr& what) {
switch (what.sa_family) {
case AF_INET:
return port_of(reinterpret_cast<sockaddr_in&>(what));
case AF_INET6:
return port_of(reinterpret_cast<sockaddr_in6&>(what));
default:
break;
}
throw std::invalid_argument("invalid protocol family");
}
template <int Family>
bool ip_connect(native_socket fd, const std::string& host, uint16_t port) {
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
using sockaddr_type =
typename std::conditional<
Family == AF_INET,
sockaddr_in,
sockaddr_in6
>::type;
sockaddr_type sa;
memset(&sa, 0, sizeof(sockaddr_type));
inet_pton(Family, host.c_str(), &addr_of(sa));
family_of(sa) = Family;
port_of(sa) = htons(port);
return connect(fd, reinterpret_cast<const sockaddr*>(&sa), sizeof(sa)) == 0;
}
native_socket new_tcp_connection_impl(const std::string& host, uint16_t port,
optional<protocol> preferred) {
CAF_LOGF_TRACE(CAF_ARG(host) << ", " << CAF_ARG(port)); CAF_LOGF_TRACE(CAF_ARG(host) << ", " << CAF_ARG(port));
CAF_LOGF_INFO("try to connect to " << host << " on port " << port); CAF_LOGF_INFO("try to connect to " << host << " on port " << port);
# 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();
# endif # endif
sockaddr_in serv_addr; auto res = interfaces::get_addrinfo_of_host(host, preferred);
hostent* server; if (!res) {
native_socket fd = socket(AF_INET, SOCK_STREAM, 0); std::string errstr = "no such host: ";
errstr += host;
throw network_error(std::move(errstr));
}
auto proto = res->second;
CAF_REQUIRE(proto == ipv4 || proto == ipv6);
auto fd = socket(proto == ipv4 ? AF_INET : AF_INET6, SOCK_STREAM, 0);
if (fd == invalid_native_socket) { if (fd == invalid_native_socket) {
throw network_error("socket creation failed"); throw network_error("socket creation failed");
} }
socket_guard sguard(fd); socket_guard sguard(fd);
server = gethostbyname(host.c_str()); if (proto == ipv6) {
if (!server) { if (ip_connect<AF_INET6>(fd, res->first, port)) {
std::string errstr = "no such host: "; return fd;
errstr += host; }
throw network_error(std::move(errstr)); // IPv4 fallback
return new_tcp_connection_impl(host, port, ipv4);
} }
memset(&serv_addr, 0, sizeof(serv_addr)); if (!ip_connect<AF_INET>(fd, res->first, port)) {
serv_addr.sin_family = AF_INET;
memmove(&serv_addr.sin_addr.s_addr, server->h_addr,
static_cast<size_t>(server->h_length));
serv_addr.sin_port = htons(port);
CAF_LOGF_DEBUG("call connect()");
if (connect(fd, reinterpret_cast<const sockaddr*>(&serv_addr),
sizeof(serv_addr)) != 0) {
CAF_LOGF_ERROR("could not connect to to " << host << " on port " << port); CAF_LOGF_ERROR("could not connect to to " << host << " on port " << port);
throw network_error("could not connect to host"); throw network_error("could not connect to host");
} }
sguard.release(); return sguard.release();
return fd;
} }
default_socket new_ipv4_connection(const std::string& host, uint16_t port) { default_socket new_tcp_connection(const std::string& host, uint16_t port) {
auto& backend = get_multiplexer_singleton(); auto& backend = get_multiplexer_singleton();
return default_socket{backend, new_ipv4_connection_impl(host, port)}; return default_socket{backend, new_tcp_connection_impl(host, port)};
}
template <class SockAddrType>
void read_port(native_socket fd, SockAddrType& sa) {
socklen_t len = sizeof(SockAddrType);
if (getsockname(fd, reinterpret_cast<sockaddr*>(&sa) , &len) != 0) {
throw network_error("getsockname(): " + last_socket_error_as_string());
}
}
void set_inaddr_any(native_socket, sockaddr_in& sa) {
sa.sin_addr.s_addr = INADDR_ANY;
}
void set_inaddr_any(native_socket fd, sockaddr_in6& sa) {
sa.sin6_addr = in6addr_any;
// also accept ipv4 requests on this socket
int off = 0;
if (setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY,
reinterpret_cast<setsockopt_ptr>(&off), sizeof(off)) != 0) {
throw network_error("unable to unset IPV6_V6ONLY");
}
}
template <int Family>
uint16_t new_ip_acceptor_impl(native_socket fd, uint16_t port,
const char* addr) {
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
CAF_LOGF_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
# ifdef CAF_WINDOWS
// make sure TCP has been initialized via WSAStartup
get_multiplexer_singleton();
# endif
using sockaddr_type =
typename std::conditional<
Family == AF_INET,
sockaddr_in,
sockaddr_in6
>::type;
sockaddr_type sa;
memset(&sa, 0, sizeof(sockaddr_type));
family_of(sa) = Family;
if (!addr) {
set_inaddr_any(fd, sa);
} else if (::inet_pton(Family, addr, &addr_of(sa)) <= 0) {
std::string err("invalid IP address: ");
err += addr;
throw network_error(std::move(err));
}
port_of(sa) = htons(port);
if (bind(fd, reinterpret_cast<sockaddr*>(&sa), sizeof(sa)) != 0) {
throw bind_failure(last_socket_error_as_string());
}
read_port(fd, sa);
return ntohs(port_of(sa));
} }
std::pair<native_socket, uint16_t> std::pair<native_socket, uint16_t>
new_ipv4_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr) { new_tcp_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
get_multiplexer_singleton(); get_multiplexer_singleton();
# endif # endif
native_socket fd = socket(AF_INET, SOCK_STREAM, 0); protocol proto = ipv6;
if (fd == invalid_native_socket) { if (addr) {
auto addrs = interfaces::get_addrinfo_of_host(addr);
if (!addrs) {
std::string errstr = "no such host: ";
errstr += addr;
throw network_error(std::move(errstr));
}
proto = addrs->second;
CAF_REQUIRE(proto == ipv4 || proto == ipv6);
}
native_socket fd = socket(proto == ipv4 ? AF_INET : AF_INET6, SOCK_STREAM, 0);
if(fd == invalid_native_socket) {
throw network_error("could not create server socket"); throw network_error("could not create server socket");
} }
// sguard closes the socket in case of exception // sguard closes the socket in case of exception
...@@ -1010,39 +1137,21 @@ new_ipv4_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr) { ...@@ -1010,39 +1137,21 @@ new_ipv4_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr) {
throw_io_failure("unable to set SO_REUSEADDR"); throw_io_failure("unable to set SO_REUSEADDR");
} }
} }
struct sockaddr_in serv_addr; auto p = proto == ipv4 ? new_ip_acceptor_impl<AF_INET>(fd, port, addr)
memset(&serv_addr, 0, sizeof(serv_addr)); : new_ip_acceptor_impl<AF_INET6>(fd, port, addr);
serv_addr.sin_family = AF_INET;
if (!addr) {
serv_addr.sin_addr.s_addr = INADDR_ANY;
} else if (::inet_pton(AF_INET, addr, &serv_addr.sin_addr) <= 0) {
throw network_error("invalid IPv4 address");
}
serv_addr.sin_port = htons(port);
if (bind(fd, reinterpret_cast<sockaddr*>(&serv_addr),
sizeof(serv_addr)) < 0) {
throw bind_failure(last_socket_error_as_string());
}
if (listen(fd, SOMAXCONN) != 0) { if (listen(fd, SOMAXCONN) != 0) {
throw network_error("listen() failed: " + last_socket_error_as_string()); 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(); CAF_LOGF_DEBUG("sockfd = " << fd << ", port = " << p);
CAF_LOGF_DEBUG("sockfd = " << fd << ", port = " << ntohs(serv_addr.sin_port)); return {sguard.release(), p};
return {fd, ntohs(serv_addr.sin_port)};
} }
std::pair<default_socket_acceptor, uint16_t> std::pair<default_socket_acceptor, uint16_t>
new_ipv4_acceptor(uint16_t port, const char* addr, bool reuse) { new_tcp_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();
auto acceptor = new_ipv4_acceptor_impl(port, addr, reuse); auto acceptor = new_tcp_acceptor_impl(port, addr, reuse);
auto bound_port = acceptor.second; auto bound_port = acceptor.second;
CAF_REQUIRE(port == 0 || bound_port == port); CAF_REQUIRE(port == 0 || bound_port == port);
return {default_socket_acceptor{backend, std::move(acceptor.first)}, return {default_socket_acceptor{backend, std::move(acceptor.first)},
......
...@@ -19,8 +19,8 @@ ...@@ -19,8 +19,8 @@
#include "caf/io/network/interfaces.hpp" #include "caf/io/network/interfaces.hpp"
#include <errno.h>
#include <netdb.h> #include <netdb.h>
#include <errno.h>
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <net/if.h> #include <net/if.h>
...@@ -144,6 +144,39 @@ std::vector<std::string> interfaces::list_addresses(protocol proc, ...@@ -144,6 +144,39 @@ std::vector<std::string> interfaces::list_addresses(protocol proc,
return result; return result;
} }
optional<std::pair<std::string, protocol>>
interfaces::get_addrinfo_of_host(const std::string& host,
optional<protocol> preferred) {
addrinfo hint;
memset(&hint, 0, sizeof(hint));
hint.ai_socktype = SOCK_STREAM;
if (preferred) {
hint.ai_family = *preferred == protocol::ipv4 ? AF_INET : AF_INET6;
}
addrinfo* tmp = nullptr;
if (getaddrinfo(host.c_str(), nullptr, &hint, &tmp)) {
return none;
}
std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo};
for (auto i = addrs.get(); i != nullptr; i = i->ai_next) {
auto family = i->ai_family;
if (family == AF_INET || family == AF_INET6) {
char buffer[INET6_ADDRSTRLEN];
auto res = family == AF_INET
? inet_ntop(family,
&reinterpret_cast<sockaddr_in*>(i->ai_addr)->sin_addr,
buffer, sizeof(buffer))
: inet_ntop(family,
&reinterpret_cast<sockaddr_in6*>(i->ai_addr)->sin6_addr,
buffer, sizeof(buffer));
if (res != nullptr) {
return {{res, family == AF_INET ? protocol::ipv4 : protocol::ipv6}};
}
}
}
return none;
}
} // namespace network } // namespace network
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -65,6 +65,7 @@ int main() { ...@@ -65,6 +65,7 @@ int main() {
auto port = publish_at_some_port(4242, d); auto port = publish_at_some_port(4242, d);
std::this_thread::sleep_for(std::chrono::milliseconds(50)); std::this_thread::sleep_for(std::chrono::milliseconds(50));
io::unpublish(d, port); io::unpublish(d, port);
CAF_CHECKPOINT();
// must fail now // must fail now
try { try {
auto oops = io::remote_actor("127.0.0.1", port); auto oops = io::remote_actor("127.0.0.1", port);
......
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