Commit 19b24c67 authored by Joseph Noir's avatar Joseph Noir

Create file for TCP-specific functions

parent c1363b86
......@@ -21,6 +21,7 @@ set(LIBCAF_NET_SRCS
src/socket_manager.cpp
src/stream_socket.cpp
src/scribe.cpp
src/tcp.cpp
)
add_custom_target(libcaf_net)
......
......@@ -30,6 +30,11 @@ namespace net {
/// Utility class bundling access to network interface names and addresses.
class interfaces {
public:
/// Returns a native IPv4 or IPv6 translation of `host`.
static optional<std::pair<std::string, ip>>
native_address(const std::string& host,
optional<ip> preferred = none);
/// Returns the host and protocol available for a local server socket
static std::vector<std::pair<std::string, ip>>
server_address(uint16_t port, const char* host,
......
......@@ -18,7 +18,6 @@
#pragma once
#include <cstdint>
#include "caf/expected.hpp"
#include "caf/variant.hpp"
......@@ -63,7 +62,7 @@ error nodelay(stream_socket x, bool new_value);
/// @param buf Points to destination buffer.
/// @param buf_size Specifies the maximum size of the buffer in bytes.
/// @returns The number of received bytes on success, an error code otherwise.
/// @relates pipe_socket
/// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
variant<size_t, sec> read(stream_socket x, span<byte> buf);
......@@ -72,7 +71,7 @@ variant<size_t, sec> read(stream_socket x, span<byte> buf);
/// @param buf Points to the message to send.
/// @param buf_size Specifies the size of the buffer in bytes.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates pipe_socket
/// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
variant<size_t, sec> write(stream_socket x, span<const byte> buf);
......@@ -82,13 +81,5 @@ variant<size_t, sec> write(stream_socket x, span<const byte> buf);
variant<size_t, sec>
check_stream_socket_io_res(std::make_signed<size_t>::type res);
/// Creates a new TCP socket to accept connections on a given `port`
/// optionally limiting connections initiated from `addr`. Setting
/// `reuse_addr` to true allows rebinding to a port already in use.
/// @relates stream_socket
expected<stream_socket> make_accept_socket(uint16_t port,
const char* addr = nullptr,
bool reuse_addr = false);
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#pragma once
#include <cstdint>
#include "caf/expected.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/stream_socket.hpp"
namespace caf {
namespace net {
struct tcp {
/// 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
static expected<stream_socket> make_accept_socket(uint16_t port,
const char* addr = nullptr,
bool reuse_addr = false);
/// Create a `stream_socket` connected to `host`:`port` via the
/// `preferred` IP version.
/// @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
static expected<stream_socket> make_connected_socket(std::string host,
uint16_t port,
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);
};
} // namespace net
} // namespace caf
......@@ -84,6 +84,28 @@ int fetch_addr_str(bool get_ipv4, bool get_ipv6,
} // namespace
optional<std::pair<std::string, ip>>
interfaces::native_address(const std::string& host,
optional<ip> preferred) {
addrinfo hint;
memset(&hint, 0, sizeof(hint));
hint.ai_socktype = SOCK_STREAM;
if (preferred)
hint.ai_family = *preferred == ip::v4 ? AF_INET : AF_INET6;
addrinfo* tmp = nullptr;
if (getaddrinfo(host.c_str(), nullptr, &hint, &tmp) != 0)
return none;
std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo};
char buffer[INET6_ADDRSTRLEN];
for (auto i = addrs.get(); i != nullptr; i = i->ai_next) {
auto family = fetch_addr_str(true, true, buffer, i->ai_addr);
if (family != AF_UNSPEC)
return std::make_pair(buffer, family == AF_INET ? ip::v4
: ip::v6);
}
return none;
}
std::vector<std::pair<std::string, ip>>
interfaces::server_address(uint16_t port, const char* host,
optional<ip> preferred) {
......
......@@ -9,8 +9,7 @@
* *
* 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. *
* *
* 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. *
......@@ -23,8 +22,6 @@
#include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/logger.hpp"
#include "caf/net/interfaces.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/socket.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/span.hpp"
......@@ -33,93 +30,6 @@
namespace caf {
namespace net {
namespace {
// Save ourselves some typing.
constexpr auto ipv4 = caf::net::ip::v4;
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) {
sa.sin_addr.s_addr = INADDR_ANY;
return unit;
}
expected<void> set_inaddr_any(socket x, sockaddr_in6& sa) {
sa.sin6_addr = in6addr_any;
// also accept ipv4 requests on this socket
int off = 0;
CAF_NET_SYSCALL("setsockopt", res, !=, 0,
setsockopt(x.id, IPPROTO_IPV6, IPV6_V6ONLY,
reinterpret_cast<setsockopt_ptr>(&off),
static_cast<socket_size_type>(sizeof(off))));
return unit;
}
template <int Family, int SockType = SOCK_STREAM>
caf::expected<socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr, bool any) {
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
int socktype = SockType;
#ifdef SOCK_CLOEXEC
socktype |= SOCK_CLOEXEC;
#endif
CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(Family, socktype, 0));
child_process_inherit(fd, false);
// sguard closes the socket in case of exception
socket_guard sguard{fd};
if (reuse_addr) {
int on = 1;
CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0,
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<setsockopt_ptr>(&on),
static_cast<socket_size_type>(sizeof(on))));
}
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 (any)
set_inaddr_any(fd, sa);
CAF_NET_SYSCALL("inet_pton", tmp, !=, 1,
inet_pton(Family, addr, &addr_of(sa)));
port_of(sa) = htons(port);
CAF_NET_SYSCALL("bind", res, !=, 0,
bind(fd, reinterpret_cast<sockaddr*>(&sa),
static_cast<socket_size_type>(sizeof(sa))));
return sguard.release();
}
} // namespace
#ifdef CAF_WINDOWS
constexpr int no_sigpipe_io_flag = 0;
......@@ -279,40 +189,5 @@ check_stream_socket_io_res(std::make_signed<size_t>::type res) {
return static_cast<size_t>(res);
}
expected<stream_socket> make_accept_socket(uint16_t port, const char* addr,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
auto addrs = interfaces::server_address(port, addr);
auto addr_str = std::string{addr == nullptr ? "" : addr};
if (addrs.empty())
return make_error(sec::cannot_open_port, "No local interface available",
addr_str);
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 == ipv4
? new_ip_acceptor_impl<AF_INET>(port, hostname, reuse_addr, any)
: new_ip_acceptor_impl<AF_INET6>(port, hostname, reuse_addr, any);
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_ARG(addr_str));
return make_error(sec::cannot_open_port, "tcp socket creation failed",
port, addr_str);
}
socket_guard sguard{fd.id};
CAF_NET_SYSCALL("listen", tmp2, !=, 0, listen(fd.id, SOMAXCONN));
// ok, no errors so far
CAF_LOG_DEBUG(CAF_ARG(fd.id));
return socket_cast<stream_socket>(sguard.release());
}
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/logger.hpp"
#include "caf/net/interfaces.hpp"
#include "caf/net/socket_guard.hpp"
namespace caf {
namespace net {
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) {
sa.sin_addr.s_addr = INADDR_ANY;
return unit;
}
expected<void> set_inaddr_any(socket x, sockaddr_in6& sa) {
sa.sin6_addr = in6addr_any;
// also accept ipv4 requests on this socket
int off = 0;
CAF_NET_SYSCALL("setsockopt", res, !=, 0,
setsockopt(x.id, IPPROTO_IPV6, IPV6_V6ONLY,
reinterpret_cast<setsockopt_ptr>(&off),
static_cast<socket_size_type>(sizeof(off))));
return unit;
}
template <int Family, int SockType = SOCK_STREAM>
caf::expected<socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr, bool any) {
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
int socktype = SockType;
#ifdef SOCK_CLOEXEC
socktype |= SOCK_CLOEXEC;
#endif
CAF_NET_SYSCALL("socket", fd, ==, -1, ::socket(Family, socktype, 0));
child_process_inherit(fd, false);
// sguard closes the socket in case of exception
socket_guard sguard{fd};
if (reuse_addr) {
int on = 1;
CAF_NET_SYSCALL("setsockopt", tmp1, !=, 0,
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<setsockopt_ptr>(&on),
static_cast<socket_size_type>(sizeof(on))));
}
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 (any)
set_inaddr_any(fd, sa);
CAF_NET_SYSCALL("inet_pton", tmp, !=, 1,
inet_pton(Family, addr, &addr_of(sa)));
port_of(sa) = htons(port);
CAF_NET_SYSCALL("bind", res, !=, 0,
bind(fd, reinterpret_cast<sockaddr*>(&sa),
static_cast<socket_size_type>(sizeof(sa))));
return sguard.release();
}
template <int Family>
bool ip_connect(socket fd, const std::string& host, uint16_t port,
bool nonblock = true) {
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
|| (nonblock && errno == EINPROGRESS);
}
} // namespace
expected<stream_socket> tcp::make_accept_socket(uint16_t port, const char* addr,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
auto addrs = interfaces::server_address(port, addr);
auto addr_str = std::string{addr == nullptr ? "" : addr};
if (addrs.empty())
return make_error(sec::cannot_open_port, "No local interface available",
addr_str);
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) {
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_ARG(addr_str));
return make_error(sec::cannot_open_port, "tcp socket creation failed",
port, addr_str);
}
socket_guard sguard{fd};
CAF_NET_SYSCALL("listen", tmp, !=, 0, listen(fd.id, SOMAXCONN));
// ok, no errors so far
CAF_LOG_DEBUG(CAF_ARG(fd.id));
return socket_cast<stream_socket>(sguard.release());
}
expected<stream_socket>
tcp::make_connected_socket(std::string host, uint16_t port,
optional<ip> preferred) {
CAF_LOG_TRACE(CAF_ARG(host) << CAF_ARG(port) << CAF_ARG(preferred));
CAF_LOG_DEBUG("try to connect to:" << CAF_ARG(host) << CAF_ARG(port));
auto res = interfaces::native_address(host, std::move(preferred));
if (!res) {
CAF_LOG_DEBUG("no such host");
return make_error(sec::cannot_connect_to_node, "no such host", host, port);
}
auto proto = res->second;
CAF_ASSERT(proto == ip::v4 || proto == ip::v6);
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);
nonblocking(fd, true);
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) << CAF_ARG(port));
return socket_cast<stream_socket>(sguard.release());
}
expected<stream_socket> tcp::accept(stream_socket x) {
auto sck = ::accept(x.id, nullptr, nullptr);
if (sck == net::invalid_socket_id) {
auto err = net::last_socket_error();
if (err != std::errc::operation_would_block
&& err != std::errc::resource_unavailable_try_again) {
return caf::make_error(sec::unavailable_or_would_block);
}
return caf::make_error(sec::socket_operation_failed);
}
return stream_socket{sck};
}
} // namespace net
} // namespace caf
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