Commit 6812f901 authored by Dominik Charousset's avatar Dominik Charousset

Implement optional timeout for TCP connections

parent a642ccd6
...@@ -74,6 +74,13 @@ bool CAF_NET_EXPORT last_socket_error_is_temporary(); ...@@ -74,6 +74,13 @@ bool CAF_NET_EXPORT last_socket_error_is_temporary();
/// @relates socket /// @relates socket
std::string CAF_NET_EXPORT last_socket_error_as_string(); std::string CAF_NET_EXPORT last_socket_error_as_string();
/// Queries whether `x` is a valid and connected socket by reading the socket
/// option `SO_ERROR`. Sets the last socket error in case the socket entered an
/// error state.
/// @returns `true` if no error is pending on `x`, `false` otherwise.
/// @relates socket
bool CAF_NET_EXPORT probe(socket x);
/// Sets x to be inherited by child processes if `new_value == true` /// Sets x to be inherited by child processes if `new_value == true`
/// or not if `new_value == false`. Not implemented on Windows. /// or not if `new_value == false`. Not implemented on Windows.
/// @relates socket /// @relates socket
......
...@@ -44,6 +44,19 @@ public: ...@@ -44,6 +44,19 @@ public:
: stream_transport_error::permanent; : stream_transport_error::permanent;
} }
/// Checks whether connecting a non-blocking socket was successful.
static ptrdiff_t connect(stream_socket x) {
// A connection is established if the OS reports a socket as ready for read
// or write and if there is no error on the socket.
return net::probe(x) ? 1 : -1;
}
/// Convenience function that always returns 1. Exists to make writing code
/// against multiple policies easier by providing the same interface.
static ptrdiff_t accept(stream_socket) {
return 1;
}
/// Returns the number of bytes that are buffered internally and that /// Returns the number of bytes that are buffered internally and that
/// available for immediate read. /// available for immediate read.
static constexpr size_t buffered() { static constexpr size_t buffered() {
......
...@@ -8,6 +8,7 @@ ...@@ -8,6 +8,7 @@
#include "caf/ip_endpoint.hpp" #include "caf/ip_endpoint.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/timespan.hpp"
#include "caf/uri.hpp" #include "caf/uri.hpp"
namespace caf::net { namespace caf::net {
...@@ -21,16 +22,41 @@ struct CAF_NET_EXPORT tcp_stream_socket : stream_socket { ...@@ -21,16 +22,41 @@ struct CAF_NET_EXPORT tcp_stream_socket : stream_socket {
/// Creates a `tcp_stream_socket` connected to given remote node. /// Creates a `tcp_stream_socket` connected to given remote node.
/// @param node Host and port of the remote node. /// @param node Host and port of the remote node.
/// @param timeout Maximum waiting time on the connection before canceling it.
/// @returns The connected socket or an error. /// @returns The connected socket or an error.
/// @relates tcp_stream_socket /// @relates tcp_stream_socket
expected<tcp_stream_socket> expected<tcp_stream_socket> CAF_NET_EXPORT //
CAF_NET_EXPORT make_connected_tcp_stream_socket(ip_endpoint node); make_connected_tcp_stream_socket(ip_endpoint node, timespan timeout = infinite);
/// Creates a `tcp_stream_socket` connected to @p auth. /// Creates a `tcp_stream_socket` connected to @p auth.
/// @param auth Host and port of the remote node. /// @param auth Host and port of the remote node.
/// @param timeout Maximum waiting time on the connection before canceling it.
/// @returns The connected socket or an error. /// @returns The connected socket or an error.
/// @note The timeout applies to a *single* connection attempt. If the DNS
/// lookup for @p auth returns more than one possible IP address then the
/// @p timeout applies to each connection attempt individually. For
/// example, passing a timeout of one second with a DNS result of five
/// entries would mean this function can block up to five seconds if all
/// attempts time out.
/// @relates tcp_stream_socket /// @relates tcp_stream_socket
expected<tcp_stream_socket> CAF_NET_EXPORT expected<tcp_stream_socket> CAF_NET_EXPORT //
make_connected_tcp_stream_socket(const uri::authority_type& auth); make_connected_tcp_stream_socket(const uri::authority_type& auth,
timespan timeout = infinite);
/// Creates a `tcp_stream_socket` connected to given @p host and @p port.
/// @param host TCP endpoint for connecting to.
/// @param port TCP port of the server.
/// @param timeout Maximum waiting time on the connection before canceling it.
/// @returns The connected socket or an error.
/// @note The timeout applies to a *single* connection attempt. If the DNS
/// lookup for @p auth returns more than one possible IP address then the
/// @p timeout applies to each connection attempt individually. For
/// example, passing a timeout of one second with a DNS result of five
/// entries would mean this function can block up to five seconds if all
/// attempts time out.
/// @relates tcp_stream_socket
expected<tcp_stream_socket> CAF_NET_EXPORT //
make_connected_tcp_stream_socket(std::string host, uint16_t port,
timespan timeout = infinite);
} // namespace caf::net } // namespace caf::net
...@@ -91,8 +91,14 @@ std::errc last_socket_error() { ...@@ -91,8 +91,14 @@ std::errc last_socket_error() {
} }
bool last_socket_error_is_temporary() { bool last_socket_error_is_temporary() {
int wsa_code = WSAGetLastError(); switch (WSAGetLastError()) {
return wsa_code == WSAEWOULDBLOCK || wsa_code == WSATRY_AGAIN; case WSATRY_AGAIN:
case WSAEINPROGRESS:
case WSAEWOULDBLOCK:
return true;
default:
return false;
}
} }
std::string last_socket_error_as_string() { std::string last_socket_error_as_string() {
...@@ -115,6 +121,17 @@ bool would_block_or_temporarily_unavailable(int errcode) { ...@@ -115,6 +121,17 @@ bool would_block_or_temporarily_unavailable(int errcode) {
return errcode == WSAEWOULDBLOCK || errcode == WSATRY_AGAIN; return errcode == WSAEWOULDBLOCK || errcode == WSATRY_AGAIN;
} }
bool probe(socket x) {
auto err = 0;
auto len = static_cast<socklen_t>(sizeof(err));
if (getsockopt(x.id, SOL_SOCKET, SO_ERROR, &err, &len) == 0) {
WSASetLastError(err);
return err == 0;
} else {
return false;
}
}
error child_process_inherit(socket x, bool) { error child_process_inherit(socket x, bool) {
// TODO: possible to implement via SetHandleInformation? // TODO: possible to implement via SetHandleInformation?
if (x == invalid_socket) if (x == invalid_socket)
...@@ -143,12 +160,16 @@ std::errc last_socket_error() { ...@@ -143,12 +160,16 @@ std::errc last_socket_error() {
} }
bool last_socket_error_is_temporary() { bool last_socket_error_is_temporary() {
auto code = errno; switch (errno) {
# if EAGAIN == EWOULDBLOCK case EAGAIN:
return code == EAGAIN; case EINPROGRESS:
# else # if EAGAIN != EWOULDBLOCK
return code == EAGAIN || code == EWOULDBLOCK; case EWOULDBLOCK:
# endif # endif
return true;
default:
return false;
}
} }
std::string last_socket_error_as_string() { std::string last_socket_error_as_string() {
...@@ -159,6 +180,17 @@ bool would_block_or_temporarily_unavailable(int errcode) { ...@@ -159,6 +180,17 @@ bool would_block_or_temporarily_unavailable(int errcode) {
return errcode == EAGAIN || errcode == EWOULDBLOCK; return errcode == EAGAIN || errcode == EWOULDBLOCK;
} }
bool probe(socket x) {
auto err = 0;
auto len = static_cast<socklen_t>(sizeof(err));
if (getsockopt(x.id, SOL_SOCKET, SO_ERROR, &err, &len) == 0) {
errno = err;
return err == 0;
} else {
return false;
}
}
error child_process_inherit(socket x, bool new_value) { error child_process_inherit(socket x, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(new_value)); CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(new_value));
// read flags for x // read flags for x
......
...@@ -15,31 +15,115 @@ ...@@ -15,31 +15,115 @@
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
#include <algorithm>
#ifndef CAF_WINDOWS
# include <poll.h>
#endif
#ifdef CAF_WINDOWS
# define POLL_FN ::WSAPoll
#else
# define POLL_FN ::poll
#endif
namespace caf::net { namespace caf::net {
namespace { namespace {
bool connect_with_timeout(stream_socket fd, const sockaddr* addr,
socklen_t addrlen, timespan timeout) {
namespace sc = std::chrono;
CAF_LOG_TRACE(CAF_ARG(fd.id) << CAF_ARG(timeout));
// Set to non-blocking or fail.
if (auto err = nonblocking(fd, true))
return false;
// Calculate deadline and define a lambda for getting the relative time in ms.
auto deadline = sc::steady_clock::now() + timeout;
auto ms_until_deadline = [deadline] {
auto t = sc::steady_clock::now();
auto ms_count = sc::duration_cast<sc::milliseconds>(deadline - t).count();
return std::max(static_cast<int>(ms_count), 0);
};
// Call connect() once and see if it succeeds. Otherwise enter a poll()-loop.
if (connect(fd.id, addr, addrlen) == 0) {
// Done! Try restoring the socket to blocking and return.
if (auto err = nonblocking(fd, false))
return false;
else
return true;
} else if (!last_socket_error_is_temporary()) {
// Hard error. No need to restore the socket to blocking since we are going
// to close it.
return false;
} else {
// Loop until the reaching the deadline.
pollfd pollset[1];
pollset[0].fd = fd.id;
pollset[0].events = POLLOUT;
auto ms = ms_until_deadline();
do {
auto pres = POLL_FN(pollset, 1, ms);
if (pres > 0) {
// Check that the socket really is ready to go by reading SO_ERROR.
if (probe(fd)) {
// Done! Try restoring the socket to blocking and return.
if (auto err = nonblocking(fd, false))
return false;
else
return true;
} else {
return false;
}
} else if (pres < 0 && !last_socket_error_is_temporary()) {
return false;
}
// Else: timeout or EINTR. Try-again.
ms = ms_until_deadline();
} while (ms > 0);
}
// No need to restore the socket to blocking since we are going to close it.
return false;
}
template <int Family> template <int Family>
bool ip_connect(stream_socket fd, std::string host, uint16_t port) { bool ip_connect(stream_socket fd, std::string host, uint16_t port,
timespan timeout) {
CAF_LOG_TRACE("Family =" << (Family == AF_INET ? "AF_INET" : "AF_INET6") CAF_LOG_TRACE("Family =" << (Family == AF_INET ? "AF_INET" : "AF_INET6")
<< CAF_ARG(fd.id) << CAF_ARG(host) << CAF_ARG(port)); << CAF_ARG(fd.id) << CAF_ARG(host) << CAF_ARG(port)
<< CAF_ARG(timeout));
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family"); static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
using sockaddr_type = using sockaddr_type =
typename std::conditional<Family == AF_INET, sockaddr_in, typename std::conditional<Family == AF_INET, sockaddr_in,
sockaddr_in6>::type; sockaddr_in6>::type;
sockaddr_type sa; sockaddr_type sa;
memset(&sa, 0, sizeof(sockaddr_type)); memset(&sa, 0, sizeof(sockaddr_type));
inet_pton(Family, host.c_str(), &detail::addr_of(sa)); if (inet_pton(Family, host.c_str(), &detail::addr_of(sa)) == 1) {
detail::family_of(sa) = Family; detail::family_of(sa) = Family;
detail::port_of(sa) = htons(port); detail::port_of(sa) = htons(port);
using sa_ptr = const sockaddr*; using sa_ptr = const sockaddr*;
if (timeout == infinite) {
return ::connect(fd.id, reinterpret_cast<sa_ptr>(&sa), sizeof(sa)) == 0; return ::connect(fd.id, reinterpret_cast<sa_ptr>(&sa), sizeof(sa)) == 0;
} else {
return connect_with_timeout(fd, reinterpret_cast<sa_ptr>(&sa), sizeof(sa),
timeout);
}
} else {
CAF_LOG_DEBUG("inet_pton failed to parse"
<< host << "for family"
<< (Family == AF_INET ? "AF_INET" : "AF_INET6"));
return false;
}
} }
} // namespace } // namespace
expected<tcp_stream_socket> make_connected_tcp_stream_socket(ip_endpoint node) { expected<tcp_stream_socket> make_connected_tcp_stream_socket(ip_endpoint node,
CAF_LOG_DEBUG("tcp connect to: " << to_string(node)); timespan timeout) {
CAF_LOG_TRACE(CAF_ARG(node) << CAF_ARG(timeout));
CAF_LOG_DEBUG_IF(timeout == infinite, "try to connect to TCP node" << node);
CAF_LOG_DEBUG_IF(timeout != infinite, "try to connect to TCP node"
<< node << "with timeout" << timeout);
auto proto = node.address().embeds_v4() ? AF_INET : AF_INET6; auto proto = node.address().embeds_v4() ? AF_INET : AF_INET6;
int socktype = SOCK_STREAM; int socktype = SOCK_STREAM;
#ifdef SOCK_CLOEXEC #ifdef SOCK_CLOEXEC
...@@ -51,21 +135,25 @@ expected<tcp_stream_socket> make_connected_tcp_stream_socket(ip_endpoint node) { ...@@ -51,21 +135,25 @@ expected<tcp_stream_socket> make_connected_tcp_stream_socket(ip_endpoint node) {
return err; return err;
auto sguard = make_socket_guard(sock); auto sguard = make_socket_guard(sock);
if (proto == AF_INET6) { if (proto == AF_INET6) {
if (ip_connect<AF_INET6>(sock, to_string(node.address()), node.port())) { if (ip_connect<AF_INET6>(sock, to_string(node.address()), node.port(),
CAF_LOG_INFO("successfully connected to (IPv6):" << to_string(node)); timeout)) {
CAF_LOG_INFO("established TCP connection to IPv6 node"
<< to_string(node));
return sguard.release(); return sguard.release();
} }
} else if (ip_connect<AF_INET>(sock, to_string(node.address().embedded_v4()), } else if (ip_connect<AF_INET>(sock, to_string(node.address().embedded_v4()),
node.port())) { node.port(), timeout)) {
CAF_LOG_INFO("successfully connected to (IPv4):" << to_string(node)); CAF_LOG_INFO("established TCP connection to IPv4 node" << to_string(node));
return sguard.release(); return sguard.release();
} }
CAF_LOG_WARNING("could not connect to: " << to_string(node)); CAF_LOG_INFO("failed to connect to" << node);
return make_error(sec::cannot_connect_to_node); return make_error(sec::cannot_connect_to_node);
} }
expected<tcp_stream_socket> expected<tcp_stream_socket>
make_connected_tcp_stream_socket(const uri::authority_type& node) { make_connected_tcp_stream_socket(const uri::authority_type& node,
timespan timeout) {
CAF_LOG_TRACE(CAF_ARG(node) << CAF_ARG(timeout));
auto port = node.port; auto port = node.port;
if (port == 0) if (port == 0)
return make_error(sec::cannot_connect_to_node, "port is zero"); return make_error(sec::cannot_connect_to_node, "port is zero");
...@@ -77,18 +165,21 @@ make_connected_tcp_stream_socket(const uri::authority_type& node) { ...@@ -77,18 +165,21 @@ make_connected_tcp_stream_socket(const uri::authority_type& node) {
if (addrs.empty()) if (addrs.empty())
return make_error(sec::cannot_connect_to_node, "empty authority"); return make_error(sec::cannot_connect_to_node, "empty authority");
for (auto& addr : addrs) { for (auto& addr : addrs) {
if (auto sock = make_connected_tcp_stream_socket(ip_endpoint{addr, port})) auto ep = ip_endpoint{addr, port};
if (auto sock = make_connected_tcp_stream_socket(ep, timeout))
return *sock; return *sock;
} }
return make_error(sec::cannot_connect_to_node, to_string(node)); return make_error(sec::cannot_connect_to_node, to_string(node));
} }
expected<tcp_stream_socket> make_connected_tcp_stream_socket(std::string host, expected<tcp_stream_socket> make_connected_tcp_stream_socket(std::string host,
uint16_t port) { uint16_t port,
timespan timeout) {
CAF_LOG_TRACE(CAF_ARG(host) << CAF_ARG(port) << CAF_ARG(timeout));
uri::authority_type auth; uri::authority_type auth;
auth.host = std::move(host); auth.host = std::move(host);
auth.port = port; auth.port = port;
return make_connected_tcp_stream_socket(auth); return make_connected_tcp_stream_socket(auth, timeout);
} }
} // namespace caf::net } // namespace caf::net
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