Commit d17f4394 authored by Joseph Noir's avatar Joseph Noir

Rework TCP sockets and ip address lookup

parent 1616d88e
...@@ -11,17 +11,17 @@ set(LIBCAF_NET_SRCS ...@@ -11,17 +11,17 @@ set(LIBCAF_NET_SRCS
src/datagram_socket.cpp src/datagram_socket.cpp
src/endpoint_manager.cpp src/endpoint_manager.cpp
src/host.cpp src/host.cpp
src/interfaces.cpp src/ip.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/network_socket.cpp src/network_socket.cpp
src/pipe_socket.cpp src/pipe_socket.cpp
src/pollset_updater.cpp src/pollset_updater.cpp
src/socket.cpp src/socket.cpp
src/socket_guard.cpp
src/socket_manager.cpp src/socket_manager.cpp
src/stream_socket.cpp src/stream_socket.cpp
src/scribe.cpp src/scribe.cpp
src/tcp.cpp src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp
) )
add_custom_target(libcaf_net) add_custom_target(libcaf_net)
......
...@@ -18,27 +18,34 @@ ...@@ -18,27 +18,34 @@
#pragma once #pragma once
#include <cstdint> #include "caf/detail/socket_sys_includes.hpp"
#include <vector>
#include "caf/net/ip.hpp"
#include "caf/optional.hpp"
namespace caf { namespace caf {
namespace net { namespace detail {
/// Utility class bundling access to network interface names and addresses. inline auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
class interfaces { return what.sin_addr;
public: }
/// Returns a native IPv4 or IPv6 translation of `host`.
static optional<std::pair<std::string, ip>> inline auto family_of(sockaddr_in& what) -> decltype(what.sin_family)& {
native_address(const std::string& host, optional<ip> preferred = none); return what.sin_family;
}
/// Returns the host and protocol available for a local server socket
static std::vector<std::pair<std::string, ip>> inline auto port_of(sockaddr_in& what) -> decltype(what.sin_port)& {
server_address(uint16_t port, const char* host, return what.sin_port;
optional<ip> preferred = none); }
};
inline auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)& {
} // namespace net return what.sin6_addr;
}
inline auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)& {
return what.sin6_family;
}
inline auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
} // namespace detail
} // namespace caf } // namespace caf
...@@ -18,19 +18,24 @@ ...@@ -18,19 +18,24 @@
#pragma once #pragma once
#include <cstddef>
#include <string> #include <string>
#include <vector>
#include "caf/fwd.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
namespace ip {
/// Returns the ip addresses assigned to `host`.
std::vector<ip_address> resolve(const std::string& host);
// IP version tag. /// Returns the ip addresses for the local hostname.
enum class ip { v4, v6 }; std::vector<ip_address> local_addrs(const std::string& host);
/// @relates ip /// Returns the hostname of this device.
inline std::string to_string(ip x) { std::string hostname();
return x == ip::v4 ? "IPv4" : "IPv6";
}
} // namespace ip
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -24,19 +24,39 @@ namespace caf { ...@@ -24,19 +24,39 @@ namespace caf {
namespace net { namespace net {
/// Closes the guarded socket when destroyed. /// Closes the guarded socket when destroyed.
template <class Socket>
class socket_guard { class socket_guard {
public: public:
explicit socket_guard(net::socket fd); explicit socket_guard(Socket fd) : fd_(fd) {
// nop
~socket_guard(); }
net::socket release(); ~socket_guard() {
close();
void close(); }
Socket release() {
auto fd = fd_;
fd_ = Socket{};
return fd;
}
void close() {
if (fd_.id != net::invalid_socket) {
CAF_LOG_DEBUG("close socket" << CAF_ARG(fd_));
net::close(fd_);
fd_ = Socket{};
}
}
private: private:
net::socket fd_; Socket fd_;
}; };
template <class Socket>
socket_guard<Socket> make_socket_guard(Socket sock) {
return socket_guard<Socket>{sock};
}
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -18,11 +18,8 @@ ...@@ -18,11 +18,8 @@
#pragma once #pragma once
#include "caf/expected.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/network_socket.hpp" #include "caf/net/network_socket.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
......
...@@ -5,7 +5,7 @@ ...@@ -5,7 +5,7 @@
* | |___ / ___ \| _| Framework * * | |___ / ___ \| _| Framework *
* \____/_/ \_|_| * * \____/_/ \_|_| *
* * * *
* Copyright 2011-2018 Dominik Charousset * * Copyright 2011-2019 Dominik Charousset *
* * * *
* Distributed under the terms and conditions of the BSD 3-Clause License or * * Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software * * (at your option) under the terms and conditions of the Boost Software *
...@@ -16,34 +16,60 @@ ...@@ -16,34 +16,60 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/net/socket_guard.hpp" #pragma once
#include "caf/logger.hpp" #include "caf/net/abstract_socket.hpp"
#include "caf/net/network_socket.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/uri.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
socket_guard::socket_guard(net::socket fd) : fd_(fd) { /// Represents an open TCP endpoint. Can be implicitly converted to a
// nop /// `stream_socket` for sending and receiving, or `network_socket`
} /// for inspection.
struct tcp_accept_socket : abstract_socket<tcp_accept_socket> {
socket_guard::~socket_guard() { using super = abstract_socket<tcp_accept_socket>;
close();
} using super::super;
net::socket socket_guard::release() { constexpr operator socket() const noexcept {
auto fd = fd_; return socket{id};
fd_ = net::invalid_socket;
return fd;
}
void socket_guard::close() {
if (fd_ != net::invalid_socket) {
CAF_LOG_DEBUG("close socket" << CAF_ARG(fd_));
net::close(fd_);
fd_ = net::invalid_socket;
} }
}
constexpr operator network_socket() const noexcept {
return network_socket{id};
}
constexpr operator stream_socket() const noexcept {
return stream_socket{id};
}
};
/// Creates a new TCP socket to accept connections on a given port.
/// @param port The port to listen on.
/// @param addr Only accepts connections originating from this address.
/// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket.
/// @relates stream_socket
expected<tcp_accept_socket> make_accept_socket(ip_address host, uint16_t port,
bool reuse_addr);
/// Creates a new TCP socket to accept connections on a given port.
/// @param port The port to listen on.
/// @param addr Only accepts connections originating from this address.
/// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket.
/// @relates stream_socket
expected<tcp_accept_socket> make_accept_socket(const uri::authority_type& auth,
bool reuse_addr = false);
/// Accept a connection on `x`.
/// @param x Listening endpoint.
/// @returns The socket that handles the accepted connection.
/// @relates stream_socket
expected<tcp_stream_socket> accept(tcp_accept_socket x);
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -18,41 +18,52 @@ ...@@ -18,41 +18,52 @@
#pragma once #pragma once
#include <cstdint> #include "caf/net/abstract_socket.hpp"
#include "caf/net/network_socket.hpp"
#include "caf/expected.hpp" #include "caf/net/socket.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/uri.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
struct tcp { /// Represents a TCP connection. Can be implicitly converted to a
/// Creates a new TCP socket to accept connections on a given port. /// `stream_socket` for sending and receiving, or `network_socket`
/// @param port The port to listen on. /// for inspection.
/// @param addr Only accepts connections originating from this address. struct tcp_stream_socket : abstract_socket<tcp_stream_socket> {
/// @param reuse_addr Optionally sets the SO_REUSEADDR option on the socket. using super = abstract_socket<tcp_stream_socket>;
/// @relates stream_socket
static expected<stream_socket> make_accept_socket(uint16_t port, using super::super;
const char* addr = nullptr,
bool reuse_addr = false); constexpr operator socket() const noexcept {
/// Create a `stream_socket` connected to `host`:`port` via the return socket{id};
/// `preferred` IP version. }
/// @param host The remote host to connecto to.
/// @param port The port on the remote host to connect to. constexpr operator network_socket() const noexcept {
/// @preferred Preferred IP version. return network_socket{id};
/// @returns The connected socket or an error. }
/// @relates stream_socket
static expected<stream_socket> constexpr operator stream_socket() const noexcept {
make_connected_socket(std::string host, uint16_t port, return stream_socket{id};
optional<ip> preferred = none); }
/// Accept a connection on `x`.
/// @param x Listening endpoint.
/// @returns The socket that handles the accepted connection.
/// @relates stream_socket
static expected<stream_socket> accept(stream_socket x);
}; };
/// Create a `stream_socket` connected to `host`:`port`.
/// @param host The remote host to connecto to.
/// @param port The port on the remote host to connect to.
/// @preferred Preferred IP version.
/// @returns The connected socket or an error.
/// @relates stream_socket
expected<stream_socket>
make_connected_socket(ip_address host, uint16_t port);
/// Create a `stream_socket` connected to `auth`.
/// @param authority The remote host to connecto to.
/// @preferred Preferred IP version.
/// @returns The connected socket or an error.
/// @relates stream_socket
expected<stream_socket>
make_connected_socket(const uri::authority_type& auth);
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -20,11 +20,11 @@ ...@@ -20,11 +20,11 @@
#include <vector> #include <vector>
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/socket.hpp" #include "caf/net/socket.hpp"
#include "caf/net/stream_socket.hpp" #include "caf/net/stream_socket.hpp"
#include "caf/net/tcp.hpp" #include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/policy/scribe.hpp" #include "caf/policy/scribe.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
...@@ -34,11 +34,11 @@ namespace policy { ...@@ -34,11 +34,11 @@ namespace policy {
/// A doorman accepts TCP connections and creates scribes to handle them. /// A doorman accepts TCP connections and creates scribes to handle them.
class doorman { class doorman {
public: public:
doorman(net::stream_socket acceptor) : acceptor_(acceptor) { doorman(net::tcp_accept_socket acceptor) : acceptor_(acceptor) {
// nop // nop
} }
net::socket acceptor_; net::tcp_accept_socket acceptor_;
net::socket handle() { net::socket handle() {
return acceptor_; return acceptor_;
...@@ -54,23 +54,20 @@ public: ...@@ -54,23 +54,20 @@ public:
template <class Parent> template <class Parent>
bool handle_read_event(Parent& parent) { bool handle_read_event(Parent& parent) {
auto sck = net::tcp::accept( auto sck = net::accept(acceptor_);
net::socket_cast<net::stream_socket>(acceptor_));
if (!sck) { if (!sck) {
CAF_LOG_ERROR("accept failed:" << sck.error()); CAF_LOG_ERROR("accept failed:" << sck.error());
return false; return false;
} }
auto mpx = parent.multiplexer(); auto mpx = parent.multiplexer();
if (!mpx) { if (!mpx) {
CAF_LOG_DEBUG( CAF_LOG_DEBUG("unable to get multiplexer from parent");
"could not acquire multiplexer to create a new endpoint manager");
return false; return false;
} }
auto child = make_endpoint_manager(mpx, parent.system(), scribe{*sck}, auto child = make_endpoint_manager(mpx, parent.system(), scribe{*sck},
parent.application().make()); parent.application().make());
if (auto err = child->init()) { if (auto err = child->init())
return false; return false;
}
return true; return true;
} }
......
...@@ -16,12 +16,10 @@ ...@@ -16,12 +16,10 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/net/interfaces.hpp" #include <cstddef>
#include <string>
#include <algorithm> #include <utility>
#include <cerrno> #include <vector>
#include <cstdlib>
#include <cstring>
// clang-format off // clang-format off
#ifdef CAF_WINDOWS #ifdef CAF_WINDOWS
...@@ -29,36 +27,41 @@ ...@@ -29,36 +27,41 @@
# define _WIN32_WINNT 0x0600 # define _WIN32_WINNT 0x0600
# endif # endif
# include <iostream> # include <iostream>
# include <winsock2.h>
# include <ws2tcpip.h>
# include <iphlpapi.h> # include <iphlpapi.h>
#else #else
# include <sys/socket.h>
# include <netinet/in.h>
# include <net/if.h> # include <net/if.h>
# include <unistd.h>
# include <netdb.h> # include <netdb.h>
# include <ifaddrs.h> # include <ifaddrs.h>
# include <sys/ioctl.h> # include <sys/ioctl.h>
# include <arpa/inet.h> # include <sys/types.h>
#endif #endif
// clang-format on // clang-format on
#include <map>
#include <memory>
#include <string>
#include <utility>
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/error.hpp"
#include "caf/ip_address.hpp"
#include "caf/ipv4_address.hpp"
#include "caf/logger.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/string_view.hpp"
#ifndef HOST_NAME_MAX
#define HOST_NAME_MAX 255
#endif
using std::string;
using std::vector;
using std::pair;
namespace caf { namespace caf {
namespace net { namespace net {
namespace ip {
namespace { namespace {
// {interface_name => {protocol => address}} constexpr auto v4_any_addr = "0.0.0.0";
using interfaces_map = std::map<std::string, constexpr auto v6_any_addr = "::";
std::map<ip, std::vector<std::string>>>; constexpr auto localhost = "localhost";
template <class T> template <class T>
void* vptr(T* ptr) { void* vptr(T* ptr) {
...@@ -71,6 +74,7 @@ void* fetch_in_addr(int family, sockaddr* addr) { ...@@ -71,6 +74,7 @@ void* fetch_in_addr(int family, sockaddr* addr) {
return vptr(&reinterpret_cast<sockaddr_in6*>(addr)->sin6_addr); return vptr(&reinterpret_cast<sockaddr_in6*>(addr)->sin6_addr);
} }
// TODO: Use getnameinfo instead?
int fetch_addr_str(bool get_ipv4, bool get_ipv6, char (&buf)[INET6_ADDRSTRLEN], int fetch_addr_str(bool get_ipv4, bool get_ipv6, char (&buf)[INET6_ADDRSTRLEN],
sockaddr* addr) { sockaddr* addr) {
if (addr == nullptr) if (addr == nullptr)
...@@ -83,62 +87,104 @@ int fetch_addr_str(bool get_ipv4, bool get_ipv6, char (&buf)[INET6_ADDRSTRLEN], ...@@ -83,62 +87,104 @@ int fetch_addr_str(bool get_ipv4, bool get_ipv6, char (&buf)[INET6_ADDRSTRLEN],
: AF_UNSPEC; : AF_UNSPEC;
} }
void find_by_name(const vector<pair<string, ip_address>>& interfaces,
const string& name, vector<ip_address>& results) {
for (auto& p : interfaces)
if (p.first == name)
results.push_back(p.second);
}
void find_by_addr(const vector<pair<string, ip_address>>& interfaces,
ip_address addr, vector<ip_address>& results) {
for (auto& p : interfaces)
if (p.second == addr)
results.push_back(p.second);
}
} // namespace } // namespace
optional<std::pair<std::string, ip>> std::vector<ip_address> resolve(const std::string& host) {
interfaces::native_address(const std::string& host, optional<ip> preferred) {
addrinfo hint; addrinfo hint;
memset(&hint, 0, sizeof(hint)); memset(&hint, 0, sizeof(hint));
hint.ai_socktype = SOCK_STREAM; hint.ai_socktype = SOCK_STREAM;
if (preferred) hint.ai_family = AF_UNSPEC;
hint.ai_family = *preferred == ip::v4 ? AF_INET : AF_INET6; if (host.empty())
hint.ai_flags = AI_PASSIVE;
addrinfo* tmp = nullptr; addrinfo* tmp = nullptr;
if (getaddrinfo(host.c_str(), nullptr, &hint, &tmp) != 0) if (getaddrinfo(host.c_str(), nullptr, &hint, &tmp) != 0)
return none; return {};
std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo}; std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo};
char buffer[INET6_ADDRSTRLEN]; char buffer[INET6_ADDRSTRLEN];
std::vector<ip_address> results;
for (auto i = addrs.get(); i != nullptr; i = i->ai_next) { for (auto i = addrs.get(); i != nullptr; i = i->ai_next) {
auto family = fetch_addr_str(true, true, buffer, i->ai_addr); auto family = fetch_addr_str(true, true, buffer, i->ai_addr);
if (family != AF_UNSPEC) if (family != AF_UNSPEC) {
return std::make_pair(buffer, family == AF_INET ? ip::v4 : ip::v6); ip_address ip;
if (auto err = parse(buffer, ip)) {
CAF_LOG_ERROR("could not parse into ip address " << buffer);
continue;
} }
return none; results.emplace_back(ip);
}
}
// TODO: Should we just prefer ipv6 or use a config option?
//std::stable_sort(std::begin(results), std::end(results),
// [](const ip_address& lhs, const ip_address& rhs) {
// return !lhs.embeds_v4() && rhs.embeds_v4();
// });
return results;
} }
std::vector<std::pair<std::string, ip>> std::vector<ip_address> local_addrs(const std::string& host) {
interfaces::server_address(uint16_t port, const char* host, std::vector<ip_address> results;
optional<ip> preferred) { // The string is not returned by getifaddrs, let's just do that ourselves.
using addr_pair = std::pair<std::string, ip>; if (host == localhost)
addrinfo hint; return {ip_address{{0}, {0x1}},
memset(&hint, 0, sizeof(hint)); ipv6_address{make_ipv4_address(127, 0, 0, 1)}};
hint.ai_socktype = SOCK_STREAM; if (host == v4_any_addr || host == v6_any_addr)
if (preferred) return {ip_address{}};
hint.ai_family = *preferred == ip::v4 ? AF_INET : AF_INET6; ifaddrs* tmp = nullptr;
else if (getifaddrs(&tmp) != 0)
hint.ai_family = AF_UNSPEC;
if (host == nullptr)
hint.ai_flags = AI_PASSIVE;
auto port_str = std::to_string(port);
addrinfo* tmp = nullptr;
if (getaddrinfo(host, port_str.c_str(), &hint, &tmp) != 0)
return {}; return {};
std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo}; std::unique_ptr<ifaddrs, decltype(freeifaddrs)*> addrs{tmp, freeifaddrs};
char buffer[INET6_ADDRSTRLEN]; char buffer[INET6_ADDRSTRLEN];
// Take the first ipv6 address or the first available address otherwise // Unless explicitly specified we are going to skip link-local addresses.
std::vector<addr_pair> results; auto is_link_local = starts_with(host, "fe80:");
for (auto i = addrs.get(); i != nullptr; i = i->ai_next) { std::vector<std::pair<std::string, ip_address>> interfaces;
auto family = fetch_addr_str(true, true, buffer, i->ai_addr); for (auto i = addrs.get(); i != nullptr; i = i->ifa_next) {
auto family = fetch_addr_str(true, true, buffer, i->ifa_addr);
if (family != AF_UNSPEC) { if (family != AF_UNSPEC) {
results.emplace_back(std::string{buffer}, ip_address ip;
family == AF_INET ? ip::v4 : ip::v6); if (!is_link_local && starts_with(buffer, "fe80:")) {
CAF_LOG_DEBUG("skipping link-local address: " << buffer);
continue;
} else if (auto err = parse(buffer, ip)) {
CAF_LOG_ERROR("could not parse into ip address " << buffer);
continue;
} }
interfaces.emplace_back(std::string{i->ifa_name}, ip);
} }
std::stable_sort(std::begin(results), std::end(results), }
[](const addr_pair& lhs, const addr_pair& rhs) { ip_address host_addr;
return lhs.second > rhs.second; if (host.empty() || host == v4_any_addr || host == v6_any_addr)
}); for (auto& p : interfaces)
results.push_back(std::move(p.second));
else if (auto err = parse(host, host_addr))
find_by_name(interfaces, host, results);
else
find_by_addr(interfaces, host_addr, results);
return results; return results;
} }
std::string hostname() {
char buf[HOST_NAME_MAX + 1];
buf[HOST_NAME_MAX] = '\0';
gethostname(buf, HOST_NAME_MAX);
struct hostent* h;
h = gethostbyname(buf);
return std::string{buf};
}
} // namespace ip
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -10,6 +10,7 @@ ...@@ -10,6 +10,7 @@
* Distributed under the terms and conditions of the BSD 3-Clause License or * * Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software * * (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. * * License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see * * If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and * * http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
...@@ -21,6 +22,7 @@ ...@@ -21,6 +22,7 @@
#include "caf/detail/net_syscall.hpp" #include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_aliases.hpp" #include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/socket.hpp" #include "caf/net/socket.hpp"
#include "caf/net/socket_guard.hpp" #include "caf/net/socket_guard.hpp"
......
...@@ -16,44 +16,26 @@ ...@@ -16,44 +16,26 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/net/tcp.hpp" #include "caf/net/tcp_accept_socket.hpp"
#include "caf/detail/net_syscall.hpp" #include "caf/detail/net_syscall.hpp"
#include "caf/detail/sockaddr_members.hpp"
#include "caf/detail/socket_sys_aliases.hpp" #include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp" #include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp"
#include "caf/ip_address.hpp"
#include "caf/ipv4_address.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/interfaces.hpp" #include "caf/net/ip.hpp"
#include "caf/net/socket_guard.hpp" #include "caf/net/socket_guard.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/sec.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
namespace { namespace {
auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
return what.sin_addr;
}
auto family_of(sockaddr_in& what) -> decltype(what.sin_family)& {
return what.sin_family;
}
auto port_of(sockaddr_in& what) -> decltype(what.sin_port)& {
return what.sin_port;
}
auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)& {
return what.sin6_addr;
}
auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)& {
return what.sin6_family;
}
auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
expected<void> set_inaddr_any(socket, sockaddr_in& sa) { expected<void> set_inaddr_any(socket, sockaddr_in& sa) {
sa.sin_addr.s_addr = INADDR_ANY; sa.sin_addr.s_addr = INADDR_ANY;
return unit; return unit;
...@@ -82,7 +64,7 @@ caf::expected<socket> new_ip_acceptor_impl(uint16_t port, const char* addr, ...@@ -82,7 +64,7 @@ caf::expected<socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(Family, socktype, 0)); CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(Family, socktype, 0));
child_process_inherit(fd, false); child_process_inherit(fd, false);
// sguard closes the socket in case of exception // sguard closes the socket in case of exception
socket_guard sguard{fd}; auto sguard = make_socket_guard(socket{fd});
if (reuse_addr) { if (reuse_addr) {
int on = 1; int on = 1;
CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0, CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0,
...@@ -94,113 +76,59 @@ caf::expected<socket> new_ip_acceptor_impl(uint16_t port, const char* addr, ...@@ -94,113 +76,59 @@ caf::expected<socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
Family == AF_INET, sockaddr_in, sockaddr_in6>::type; Family == AF_INET, sockaddr_in, sockaddr_in6>::type;
sockaddr_type sa; sockaddr_type sa;
memset(&sa, 0, sizeof(sockaddr_type)); memset(&sa, 0, sizeof(sockaddr_type));
family_of(sa) = Family; detail::family_of(sa) = Family;
if (any) if (any)
set_inaddr_any(fd, sa); set_inaddr_any(fd, sa);
CAF_NET_SYSCALL("inet_pton", tmp, !=, 1, CAF_NET_SYSCALL("inet_pton", tmp, !=, 1,
inet_pton(Family, addr, &addr_of(sa))); inet_pton(Family, addr, &detail::addr_of(sa)));
port_of(sa) = htons(port); detail::port_of(sa) = htons(port);
CAF_NET_SYSCALL("bind", res, !=, 0, CAF_NET_SYSCALL("bind", res, !=, 0,
bind(fd, reinterpret_cast<sockaddr*>(&sa), bind(fd, reinterpret_cast<sockaddr*>(&sa),
static_cast<socket_size_type>(sizeof(sa)))); static_cast<socket_size_type>(sizeof(sa))));
return sguard.release(); return sguard.release();
} }
template <int Family>
bool ip_connect(socket fd, const std::string& host, uint16_t port) {
CAF_LOG_TRACE("Family =" << (Family == AF_INET ? "AF_INET" : "AF_INET6")
<< CAF_ARG(fd.id) << CAF_ARG(host));
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);
using sa_ptr = const sockaddr*;
return ::connect(fd.id, reinterpret_cast<sa_ptr>(&sa), sizeof(sa)) == 0;
}
} // namespace } // namespace
expected<stream_socket> tcp::make_accept_socket(uint16_t port, const char* addr, expected<tcp_accept_socket> make_accept_socket(ip_address addr, uint16_t port,
bool reuse_addr) { bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr")); CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << to_string(addr));
auto addrs = interfaces::server_address(port, addr); auto h = addr.embeds_v4() ? to_string(addr.embedded_v4()) : to_string(addr);
auto addr_str = std::string{addr == nullptr ? "" : addr}; bool any = addr.zero();
if (addrs.empty()) auto p = addr.embeds_v4()
return make_error(sec::cannot_open_port, "No local interface available", ? new_ip_acceptor_impl<AF_INET>(port, h.c_str(), reuse_addr, any)
addr_str); : new_ip_acceptor_impl<AF_INET6>(port, h.c_str(), reuse_addr, any);
bool any = addr_str.empty() || addr_str == "::" || addr_str == "0.0.0.0";
socket fd = invalid_socket;
for (auto& elem : addrs) {
auto hostname = elem.first.c_str();
auto p = elem.second == ip::v4
? new_ip_acceptor_impl<AF_INET>(port, hostname, reuse_addr, any)
: new_ip_acceptor_impl<AF_INET6>(port, hostname, reuse_addr,
any);
if (!p) { if (!p) {
CAF_LOG_DEBUG(p.error());
continue;
}
fd = *p;
break;
}
if (fd == invalid_socket_id) {
CAF_LOG_WARNING("could not open tcp socket on:" << CAF_ARG(port) CAF_LOG_WARNING("could not open tcp socket on:" << CAF_ARG(port)
<< CAF_ARG(addr_str)); << CAF_ARG(h));
return make_error(sec::cannot_open_port, "tcp socket creation failed", port, return make_error(sec::cannot_open_port, "tcp socket creation failed", port,
addr_str); h);
} }
socket_guard sguard{fd}; auto fd = socket_cast<tcp_accept_socket>(*p);
auto sguard = make_socket_guard(fd);
CAF_NET_SYSCALL("listen", tmp, !=, 0, listen(fd.id, SOMAXCONN)); CAF_NET_SYSCALL("listen", tmp, !=, 0, listen(fd.id, SOMAXCONN));
// ok, no errors so far
CAF_LOG_DEBUG(CAF_ARG(fd.id)); CAF_LOG_DEBUG(CAF_ARG(fd.id));
return socket_cast<stream_socket>(sguard.release()); return sguard.release();
} }
expected<stream_socket> tcp::make_connected_socket(std::string host, expected<tcp_accept_socket> make_accept_socket(const uri::authority_type& auth,
uint16_t port, bool reuse_addr) {
optional<ip> preferred) { if (auto ip = get_if<ip_address>(&auth.host))
CAF_LOG_TRACE(CAF_ARG(host) << CAF_ARG(port) << CAF_ARG(preferred)); return make_accept_socket(*ip, auth.port, reuse_addr);
CAF_LOG_DEBUG("try to connect to:" << CAF_ARG(host) << CAF_ARG(port)); auto host = get<std::string>(auth.host);
auto res = interfaces::native_address(host, std::move(preferred)); auto addrs = ip::local_addrs(host);
if (!res) { if (addrs.empty())
CAF_LOG_DEBUG("no such host"); return make_error(sec::cannot_open_port, "No local interface available",
return make_error(sec::cannot_connect_to_node, "no such host", host, port); to_string(auth));
} for (auto& addr : addrs) {
auto proto = res->second; if (auto sock = make_accept_socket(addr, auth.port, reuse_addr))
CAF_ASSERT(proto == ip::v4 || proto == ip::v6); return *sock;
int socktype = SOCK_STREAM;
#ifdef SOCK_CLOEXEC
socktype |= SOCK_CLOEXEC;
#endif
CAF_NET_SYSCALL("socket", fd, ==, -1,
::socket(proto == ip::v4 ? AF_INET : AF_INET6, socktype, 0));
child_process_inherit(fd, false);
socket_guard sguard(fd);
if (proto == ip::v6) {
if (ip_connect<AF_INET6>(fd, res->first, port)) {
CAF_LOG_INFO("successfully connected to (IPv6):" << CAF_ARG(host)
<< CAF_ARG(port));
return socket_cast<stream_socket>(sguard.release());
}
sguard.close();
// IPv4 fallback
return make_connected_socket(host, port, ip::v4);
}
if (!ip_connect<AF_INET>(fd, res->first, port)) {
CAF_LOG_WARNING("could not connect to:" << CAF_ARG(host) << CAF_ARG(port));
return make_error(sec::cannot_connect_to_node, "ip_connect failed", host,
port);
} }
CAF_LOG_INFO("successfully connected to (IPv4):" << CAF_ARG(host) return make_error(sec::cannot_open_port, "tcp socket creation failed",
<< CAF_ARG(port)); to_string(auth));
return socket_cast<stream_socket>(sguard.release());
} }
expected<stream_socket> tcp::accept(stream_socket x) { expected<tcp_stream_socket> accept(tcp_accept_socket x) {
auto sck = ::accept(x.id, nullptr, nullptr); auto sck = ::accept(x.id, nullptr, nullptr);
if (sck == net::invalid_socket_id) { if (sck == net::invalid_socket_id) {
auto err = net::last_socket_error(); auto err = net::last_socket_error();
...@@ -210,7 +138,7 @@ expected<stream_socket> tcp::accept(stream_socket x) { ...@@ -210,7 +138,7 @@ expected<stream_socket> tcp::accept(stream_socket x) {
} }
return caf::make_error(sec::socket_operation_failed); return caf::make_error(sec::socket_operation_failed);
} }
return stream_socket{sck}; return tcp_stream_socket{sck};
} }
} // namespace net } // namespace net
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/sockaddr_members.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/expected.hpp"
#include "caf/ipv4_address.hpp"
#include "caf/logger.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/sec.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace net {
namespace {
template <int Family>
bool ip_connect(stream_socket fd, std::string host, uint16_t port) {
CAF_LOG_TRACE("Family =" << (Family == AF_INET ? "AF_INET" : "AF_INET6")
<< CAF_ARG(fd.id) << CAF_ARG(host) << CAF_ARG(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(), &detail::addr_of(sa));
detail::family_of(sa) = Family;
detail::port_of(sa) = htons(port);
using sa_ptr = const sockaddr*;
return ::connect(fd.id, reinterpret_cast<sa_ptr>(&sa), sizeof(sa)) == 0;
}
} // namespace
expected<stream_socket> make_connected_socket(ip_address host, uint16_t port) {
CAF_LOG_DEBUG("try to connect to:" << CAF_ARG(to_string(host))
<< CAF_ARG(port));
auto proto = host.embeds_v4() ? AF_INET : AF_INET6;
auto socktype = SOCK_STREAM;
#ifdef SOCK_CLOEXEC
socktype |= SOCK_CLOEXEC;
#endif
CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(proto, socktype, 0));
child_process_inherit(fd, false);
auto sguard = make_socket_guard(stream_socket{fd});
if (proto == AF_INET6) {
if (ip_connect<AF_INET6>(fd, to_string(host), port)) {
CAF_LOG_INFO("successfully connected to (IPv6):" << CAF_ARG(host)
<< CAF_ARG(port));
return sguard.release();
}
} else if (ip_connect<AF_INET>(fd, to_string(host.embedded_v4()), port)) {
return sguard.release();
}
CAF_LOG_WARNING("could not connect to:" << CAF_ARG(host) << CAF_ARG(port));
sguard.close();
return make_error(sec::cannot_connect_to_node);
}
expected<stream_socket> make_connected_socket(const uri::authority_type& auth) {
auto port = auth.port;
if (port == 0)
return make_error(sec::cannot_connect_to_node, "port is zero");
std::vector<ip_address> addrs;
if (auto str = get_if<std::string>(&auth.host))
addrs = ip::resolve(std::move(*str));
else if (auto addr = get_if<ip_address>(&auth.host))
addrs.push_back(*addr);
if (addrs.empty())
return make_error(sec::cannot_connect_to_node, "empty authority");
for (auto& addr : addrs) {
if (auto sock = make_connected_socket(addr, port))
return *sock;
}
return make_error(sec::cannot_connect_to_node, to_string(auth));
}
} // namespace net
} // namespace caf
...@@ -19,10 +19,13 @@ ...@@ -19,10 +19,13 @@
#define CAF_SUITE doorman #define CAF_SUITE doorman
#include "caf/policy/doorman.hpp" #include "caf/policy/doorman.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/tcp.hpp" #include "caf/net/tcp_accept_socket.hpp"
#include "caf/uri.hpp"
#include "caf/net/ip.hpp"
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
...@@ -38,6 +41,8 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -38,6 +41,8 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
mpx = std::make_shared<multiplexer>(); mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init()) if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err)); CAF_FAIL("mpx->init failed: " << sys.render(err));
auth.port = 0;
auth.host = std::string{"0.0.0.0"};
} }
bool handle_io_event() override { bool handle_io_event() override {
...@@ -46,6 +51,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -46,6 +51,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
} }
multiplexer_ptr mpx; multiplexer_ptr mpx;
uri::authority_type auth;
}; };
class dummy_application { class dummy_application {
...@@ -113,11 +119,14 @@ public: ...@@ -113,11 +119,14 @@ public:
CAF_TEST_FIXTURE_SCOPE(doorman_tests, fixture) CAF_TEST_FIXTURE_SCOPE(doorman_tests, fixture)
CAF_TEST(tcp connect) { CAF_TEST(tcp connect) {
auto acceptor = unbox(tcp::make_accept_socket(0, nullptr, false)); auto acceptor = unbox(make_accept_socket(auth, false));
auto port = unbox(local_port(socket_cast<network_socket>(acceptor))); auto port = unbox(local_port(socket_cast<network_socket>(acceptor)));
CAF_MESSAGE("opened acceptor on port " << port); CAF_MESSAGE("opened acceptor on port " << port);
auto con_fd = unbox(tcp::make_connected_socket("localhost", port)); uri::authority_type dst;
auto acc_fd = unbox(tcp::accept(acceptor)); dst.port = port;
dst.host = std::string{"localhost"};
auto con_fd = unbox(make_connected_socket(dst));
auto acc_fd = unbox(accept(acceptor));
CAF_MESSAGE("accepted connection"); CAF_MESSAGE("accepted connection");
close(con_fd); close(con_fd);
close(acc_fd); close(acc_fd);
...@@ -125,7 +134,7 @@ CAF_TEST(tcp connect) { ...@@ -125,7 +134,7 @@ CAF_TEST(tcp connect) {
} }
CAF_TEST(doorman accept) { CAF_TEST(doorman accept) {
auto acceptor = unbox(tcp::make_accept_socket(0, nullptr, false)); auto acceptor = unbox(make_accept_socket(auth, false));
auto port = unbox(local_port(socket_cast<network_socket>(acceptor))); auto port = unbox(local_port(socket_cast<network_socket>(acceptor)));
CAF_MESSAGE("opened acceptor on port " << port); CAF_MESSAGE("opened acceptor on port " << port);
auto mgr = make_endpoint_manager(mpx, sys, policy::doorman{acceptor}, auto mgr = make_endpoint_manager(mpx, sys, policy::doorman{acceptor},
...@@ -134,7 +143,10 @@ CAF_TEST(doorman accept) { ...@@ -134,7 +143,10 @@ CAF_TEST(doorman accept) {
handle_io_event(); handle_io_event();
auto before = mpx->num_socket_managers(); auto before = mpx->num_socket_managers();
CAF_MESSAGE("connecting to doorman"); CAF_MESSAGE("connecting to doorman");
auto fd = unbox(tcp::make_connected_socket("localhost", port)); uri::authority_type dst;
dst.port = port;
dst.host = std::string{"localhost"};
auto fd = unbox(make_connected_socket(dst));
CAF_MESSAGE("waiting for connection"); CAF_MESSAGE("waiting for connection");
while (mpx->num_socket_managers() != before + 1) while (mpx->num_socket_managers() != before + 1)
handle_io_event(); handle_io_event();
......
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