Commit 95a30652 authored by Dominik Charousset's avatar Dominik Charousset

New `ccall` utility and coding style nitpicks

parent 41957aa3
......@@ -31,9 +31,7 @@ namespace network {
* callbacks for incoming connections as well as for error handling.
*/
class acceptor_manager : public manager {
public:
~acceptor_manager();
/**
......@@ -41,7 +39,6 @@ class acceptor_manager : public manager {
* a new connection is awaiting acceptance.
*/
virtual void new_connection() = 0;
};
} // namespace network
......
......@@ -153,12 +153,6 @@ void nonblocking(native_socket fd, bool new_value);
*/
std::pair<native_socket, native_socket> create_pipe();
/**
* Throws network_error with given error message and
* the platform-specific error code if `add_errno` is `true`.
*/
void throw_io_failure(const char* what, bool add_errno = true);
/**
* Returns true if `fd` is configured as nodelay socket.
* @throws network_error
......@@ -222,6 +216,11 @@ class event_handler {
*/
virtual void removed_from_loop(operation op) = 0;
/**
* Returns the native socket handle for this handler.
*/
virtual native_socket fd() const = 0;
/**
* Returns the `multiplexer` this acceptor belongs to.
*/
......@@ -243,11 +242,6 @@ class event_handler {
m_eventbf = value;
}
/**
* Returns the native socket handle for this handler.
*/
virtual native_socket fd() const = 0;
protected:
default_multiplexer& m_backend;
int m_eventbf;
......@@ -294,12 +288,10 @@ class default_socket {
using default_socket_acceptor = default_socket;
class default_multiplexer : public multiplexer {
public:
friend class io::middleman; // disambiguate reference
friend class supervisor;
public:
struct event {
native_socket fd;
int mask;
......@@ -356,7 +348,6 @@ class default_multiplexer : public multiplexer {
void del(operation op, native_socket fd, event_handler* ptr);
private:
// platform-dependent additional initialization code
void init();
......@@ -417,20 +408,18 @@ class default_multiplexer : public multiplexer {
std::vector<event> m_events; // always sorted by .fd
multiplexer_poll_shadow_data m_shadow;
std::pair<native_socket, native_socket> m_pipe;
std::thread::id m_tid;
};
default_multiplexer& get_multiplexer_singleton();
template <class T>
inline connection_handle conn_hdl_from_socket(const T& sock) {
connection_handle conn_hdl_from_socket(const T& sock) {
return connection_handle::from_int(
int64_from_native_socket(sock.native_handle()));
}
template <class T>
inline accept_handle accept_hdl_from_socket(const T& sock) {
accept_handle accept_hdl_from_socket(const T& sock) {
return accept_handle::from_int(
int64_from_native_socket(sock.native_handle()));
}
......@@ -461,17 +450,10 @@ class stream : public event_handler {
configure_read(receive_policy::at_most(1024));
}
/**
* Returns the `multiplexer` this stream belongs to.
*/
inline default_multiplexer& backend() {
return static_cast<default_multiplexer&>(m_sock.backend());
}
/**
* Returns the IO socket.
*/
inline Socket& socket_handle() {
Socket& socket_handle() {
return m_sock;
}
......@@ -501,8 +483,7 @@ class stream : public event_handler {
}
/**
* Configures how much data will be provided
* for the next `consume` callback.
* Configures how much data will be provided for the next `consume` callback.
* @warning Must not be called outside the IO multiplexers event loop
* once the stream has been started.
*/
......@@ -513,7 +494,7 @@ class stream : public event_handler {
/**
* Copies data to the write buffer.
* @note Not thread safe.
* @warning Not thread safe.
*/
void write(const void* buf, size_t num_bytes) {
CAF_LOG_TRACE("num_bytes: " << num_bytes);
......@@ -560,7 +541,7 @@ class stream : public event_handler {
backend().del(operation::read, m_sock.fd(), this);
}
void handle_event(operation op) {
void handle_event(operation op) override {
CAF_LOG_TRACE("op = " << static_cast<int>(op));
switch (op) {
case operation::read: {
......@@ -609,7 +590,6 @@ class stream : public event_handler {
}
}
protected:
native_socket fd() const override {
return m_sock.fd();
}
......@@ -678,9 +658,7 @@ class stream : public event_handler {
*/
template <class SocketAcceptor>
class acceptor : public event_handler {
public:
using socket_type = typename SocketAcceptor::socket_type;
/**
......@@ -703,7 +681,7 @@ class acceptor : public event_handler {
/**
* Returns the IO socket.
*/
inline SocketAcceptor& socket_handle() {
SocketAcceptor& socket_handle() {
return m_accept_sock;
}
......@@ -711,7 +689,7 @@ class acceptor : public event_handler {
* Returns the accepted socket. This member function should
* be called only from the `new_connection` callback.
*/
inline socket_type& accepted_socket() {
socket_type& accepted_socket() {
return m_sock;
}
......@@ -767,18 +745,14 @@ class acceptor : public event_handler {
}
}
protected:
native_socket fd() const override {
return m_accept_sock.fd();
}
private:
manager_ptr m_mgr;
SocketAcceptor m_accept_sock;
socket_type m_sock;
};
native_socket new_tcp_connection_impl(const std::string&, uint16_t,
......
......@@ -17,8 +17,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_NETWORK_ADDRESSES_HPP
#define CAF_IO_NETWORK_ADDRESSES_HPP
#ifndef CAF_IO_NETWORK_INTERFACES_HPP
#define CAF_IO_NETWORK_INTERFACES_HPP
#include <map>
#include <vector>
......@@ -61,14 +61,14 @@ class interfaces {
bool include_localhost = true);
/**
* Returns `pair<string, protocol>` for given host address.
* Returns a native IPv4 or IPv6 translation of `host`.
**/
static optional<std::pair<std::string, protocol>>
get_addrinfo_of_host(const std::string& host, optional<protocol> preferred = none);
native_address(const std::string& host, optional<protocol> preferred = none);
};
} // namespace network
} // namespace io
} // namespace caf
#endif // CAF_IO_NETWORK_ADDRESSES_HPP
#endif // CAF_IO_NETWORK_INTERFACES_HPP
......@@ -35,7 +35,7 @@ namespace network {
*/
class manager : public ref_counted {
public:
virtual ~manager();
~manager();
/**
* Causes the manager to stop read operations on its IO device.
......@@ -49,11 +49,6 @@ class manager : public ref_counted {
virtual void io_failure(operation op) = 0;
};
/**
* @relates manager
*/
using manager_ptr = intrusive_ptr<manager>;
} // namespace network
} // namespace io
} // namespace caf
......
......@@ -33,16 +33,13 @@ namespace network {
* for incoming data as well as for error handling.
*/
class stream_manager : public manager {
public:
virtual ~stream_manager();
~stream_manager();
/**
* Called by the underlying IO device whenever it received data.
*/
virtual void consume(const void* data, size_t num_bytes) = 0;
};
} // namespace network
......
......@@ -48,6 +48,7 @@
using std::string;
namespace {
#ifdef CAF_MACOS
constexpr int no_sigpipe_flag = SO_NOSIGPIPE;
#elif defined(CAF_WINDOWS)
......@@ -55,8 +56,46 @@ namespace {
#else // BSD or Linux
constexpr int no_sigpipe_flag = MSG_NOSIGNAL;
#endif
constexpr auto ipv4 = caf::io::network::protocol::ipv4;
constexpr auto ipv6 = caf::io::network::protocol::ipv6;
// safe ourselves some typing
constexpr auto ipv4 = caf::io::network::protocol::ipv4;
constexpr auto ipv6 = caf::io::network::protocol::ipv6;
// predicate for `ccall` meaning "expected result of f is 0"
bool cc_zero(int value) {
return value == 0;
}
// predicate for `ccall` meaning "expected result of f is 1"
bool cc_one(int value) {
return value == 1;
}
// predicate for `ccall` meaning "expected result of f is not -1"
bool cc_not_minus1(int value) {
return value != -1;
}
// predicate for `ccall` meaning "expected result of f is a valid socket"
bool cc_valid_socket(caf::io::network::native_socket fd) {
return fd != caf::io::network::invalid_native_socket;
}
// calls a C function and throws a `network_error` if `p` returns false
template <class Predicate, class F, class... Ts>
auto ccall(Predicate p, const char* errmsg, F f, Ts&&... args)
-> decltype(f(std::forward<Ts>(args)...)) {
using namespace caf::io::network;
auto result = f(std::forward<Ts>(args)...);
if (!p(result)) {
std::ostringstream oss;
oss << errmsg << ": " << last_socket_error_as_string()
<< " [errno: " << last_socket_error() << "]";
throw caf::network_error(oss.str());
}
return result;
}
} // namespace <anonymous>
namespace caf {
......@@ -75,15 +114,10 @@ namespace network {
void nonblocking(native_socket fd, bool new_value) {
// read flags for fd
auto rf = fcntl(fd, F_GETFL, 0);
if (rf == -1) {
throw_io_failure("unable to read socket flags");
}
auto rf = ccall(cc_not_minus1, "cannot read flags", fcntl, fd, F_GETFL, 0);
// calculate and set new flags
auto wf = new_value ? (rf | O_NONBLOCK) : (rf & (~(O_NONBLOCK)));
if (fcntl(fd, F_SETFL, wf) < 0) {
throw_io_failure("unable to set file descriptor flags");
}
ccall(cc_not_minus1, "cannot set flags", fcntl, fd, F_SETFL, wf);
}
std::pair<native_socket, native_socket> create_pipe() {
......@@ -123,17 +157,7 @@ namespace network {
void nonblocking(native_socket fd, bool new_value) {
u_long mode = new_value ? 1 : 0;
if (ioctlsocket(fd, FIONBIO, &mode) < 0) {
throw_io_failure("unable to set FIONBIO");
}
}
// calls a C function and converts its returned error code to an exception
template <class F, class... Ts>
inline void ccall(const char* errmsg, F f, Ts&&... args) {
if (f(std::forward<Ts>(args)...) != 0) {
throw_io_failure(errmsg);
}
ccall(cc_zero, "unable to set FIONBIO", ioctlsocket, fd, FIONBIO, &mode);
}
/**************************************************************************\
......@@ -170,10 +194,8 @@ namespace network {
std::pair<native_socket, native_socket> create_pipe() {
socklen_t addrlen = sizeof(sockaddr_in);
native_socket socks[2] = {invalid_native_socket, invalid_native_socket};
auto listener = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (listener == invalid_native_socket) {
throw_io_failure("socket() failed");
}
auto listener = ccall(cc_valid_socket, "socket() failed", socket, AF_INET,
SOCK_STREAM, IPPROTO_TCP);
union {
sockaddr_in inaddr;
sockaddr addr;
......@@ -196,30 +218,29 @@ namespace network {
});
// bind listener to a local port
int reuse = 1;
ccall("setsockopt() failed", ::setsockopt, listener, SOL_SOCKET,
ccall(cc_zero, "setsockopt() failed", setsockopt, listener, SOL_SOCKET,
SO_REUSEADDR, reinterpret_cast<char*>(&reuse),
static_cast<socklen_t>(sizeof(reuse)));
ccall("bind() failed", ::bind, listener, &a.addr, sizeof(a.inaddr));
int{sizeof(reuse)});
ccall(cc_zero, "bind() failed", bind, listener,
&a.addr, int{sizeof(a.inaddr)});
// read the port in use: win32 getsockname may only set the port number
// (http://msdn.microsoft.com/library/ms738543.aspx):
memset(&a, 0, sizeof(a));
ccall("getsockname() failed", ::getsockname, listener, &a.addr, &addrlen);
ccall(cc_zero, "getsockname() failed", getsockname,
listener, &a.addr, &addrlen);
a.inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
a.inaddr.sin_family = AF_INET;
// set listener to listen mode
ccall("listen() failed", ::listen, listener, 1);
ccall(cc_zero, "listen() failed", listen, listener, 1);
// create read-only end of the pipe
DWORD flags = 0;
auto read_fd = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, flags);
if (read_fd == invalid_native_socket) {
throw_io_failure("cannot create read handle: WSASocket() failed");
}
ccall("connect() failed", ::connect, read_fd, &a.addr, sizeof(a.inaddr));
auto read_fd = ccall(cc_valid_socket, WSASocket, AF_INET, SOCK_STREAM,
0, NULL, 0, flags);
ccall(cc_zero, "connect() failed", connect, read_fd,
&a.addr, int{sizeof(a.inaddr)});
// get write-only end of the pipe
auto write_fd = accept(listener, NULL, NULL);
if (write_fd == invalid_native_socket) {
throw_io_failure("cannot create write handle: accept() failed");
}
auto write_fd = ccall(cc_valid_socket, "accept() failed",
accept, listener, NULL, NULL);
closesocket(listener);
success = true;
return {read_fd, write_fd};
......@@ -541,7 +562,8 @@ int del_flag(operation op, int bf) {
return 0;
}
void default_multiplexer::add(operation op, native_socket fd, event_handler* ptr) {
void default_multiplexer::add(operation op, native_socket fd,
event_handler* ptr) {
CAF_REQUIRE(fd != invalid_native_socket);
// ptr == nullptr is only allowed to store our pipe read handle
// and the pipe read handle is added in the ctor (not allowed here)
......@@ -551,7 +573,8 @@ void default_multiplexer::add(operation op, native_socket fd, event_handler* ptr
new_event(add_flag, op, fd, ptr);
}
void default_multiplexer::del(operation op, native_socket fd, event_handler* ptr) {
void default_multiplexer::del(operation op, native_socket fd,
event_handler* ptr) {
CAF_REQUIRE(fd != invalid_native_socket);
// ptr == nullptr is only allowed when removing our pipe read handle
CAF_REQUIRE(ptr != nullptr || fd == m_pipe.first);
......@@ -678,14 +701,14 @@ void default_multiplexer::init() {
}
default_multiplexer::~default_multiplexer() {
# ifdef CAF_WINDOWS
WSACleanup();
# endif
if (m_epollfd != invalid_native_socket) {
closesocket(m_epollfd);
}
closesocket(m_pipe.first);
closesocket(m_pipe.second);
# ifdef CAF_WINDOWS
WSACleanup();
# endif
}
void default_multiplexer::dispatch_runnable(runnable_ptr ptr) {
......@@ -738,7 +761,8 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
return ptr->hdl();
}
accept_handle default_multiplexer::add_tcp_doorman(broker* self,
accept_handle
default_multiplexer::add_tcp_doorman(broker* self,
default_socket_acceptor&& sock) {
CAF_LOG_TRACE("sock.fd = " << sock.fd());
CAF_REQUIRE(sock.fd() != network::invalid_native_socket);
......@@ -751,10 +775,9 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self,
}
void new_connection() override {
auto& dm = m_acceptor.backend();
accept_msg().handle = dm.add_tcp_scribe(parent(),
std::move(m_acceptor.accepted_socket()));
parent()->invoke_message(invalid_actor_addr,
invalid_message_id,
accept_msg().handle
= dm.add_tcp_scribe(parent(), std::move(m_acceptor.accepted_socket()));
parent()->invoke_message(invalid_actor_addr, invalid_message_id,
m_accept_msg);
}
void stop_reading() override {
......@@ -794,7 +817,6 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
return add_tcp_scribe(self, new_tcp_connection(host, port));
}
std::pair<accept_handle, uint16_t>
default_multiplexer::new_tcp_doorman(uint16_t port, const char* in,
bool reuse_addr) {
......@@ -824,22 +846,11 @@ default_multiplexer::add_tcp_doorman(broker* self, uint16_t port,
* platform-independent implementations (finally) *
******************************************************************************/
void throw_io_failure(const char* what, bool add_errno) {
if (add_errno) {
std::ostringstream oss;
oss << what << ": " << last_socket_error_as_string()
<< " [errno: " << last_socket_error() << "]";
throw network_error(oss.str());
}
throw network_error(what);
}
void tcp_nodelay(native_socket fd, bool new_value) {
int flag = new_value ? 1 : 0;
if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY,
reinterpret_cast<setsockopt_ptr>(&flag), sizeof(flag)) < 0) {
throw_io_failure("unable to set TCP_NODELAY");
}
ccall(cc_zero, "unable to set TCP_NODELAY", setsockopt, fd, IPPROTO_TCP,
TCP_NODELAY, reinterpret_cast<setsockopt_ptr>(&flag),
socklen_t{sizeof(flag)});
}
bool is_error(ssize_t res, bool is_nonblock) {
......@@ -918,7 +929,8 @@ default_socket::default_socket(default_multiplexer& ref, native_socket sockfd)
}
default_socket::default_socket(default_socket&& other)
: m_parent(other.m_parent), m_fd(other.m_fd) {
: m_parent(other.m_parent),
m_fd(other.m_fd) {
other.m_fd = invalid_native_socket;
}
......@@ -940,20 +952,23 @@ void default_socket::close_read() {
}
}
struct socket_guard {
class socket_guard {
public:
socket_guard(native_socket fd) : m_fd(fd) {
// nop
}
~socket_guard() {
if (m_fd != invalid_native_socket)
closesocket(m_fd);
}
native_socket release() {
auto fd = m_fd;
m_fd = invalid_native_socket;
return fd;
}
private:
native_socket m_fd;
};
......@@ -1019,18 +1034,14 @@ native_socket new_tcp_connection_impl(const std::string& host, uint16_t port,
// make sure TCP has been initialized via WSAStartup
get_multiplexer_singleton();
# endif
auto res = interfaces::get_addrinfo_of_host(host, preferred);
auto res = interfaces::native_address(host, preferred);
if (!res) {
std::string errstr = "no such host: ";
errstr += host;
throw network_error(std::move(errstr));
throw network_error("no such host: " + host);
}
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) {
throw network_error("socket creation failed");
}
auto fd = ccall(cc_valid_socket, "socket creation failed", socket,
proto == ipv4 ? AF_INET : AF_INET6, SOCK_STREAM, 0);
socket_guard sguard(fd);
if (proto == ipv6) {
if (ip_connect<AF_INET6>(fd, res->first, port)) {
......@@ -1041,7 +1052,7 @@ native_socket new_tcp_connection_impl(const std::string& host, uint16_t port,
}
if (!ip_connect<AF_INET>(fd, res->first, 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);
}
return sguard.release();
}
......@@ -1054,9 +1065,8 @@ default_socket new_tcp_connection(const std::string& host, uint16_t 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());
}
ccall(cc_zero, "read_port failed", getsockname, fd,
reinterpret_cast<sockaddr*>(&sa), &len);
}
void set_inaddr_any(native_socket, sockaddr_in& sa) {
......@@ -1067,10 +1077,9 @@ 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");
}
ccall(cc_zero, "unable to unset IPV6_V6ONLY", setsockopt, fd, IPPROTO_IPV6,
IPV6_V6ONLY, reinterpret_cast<setsockopt_ptr>(&off),
socklen_t{sizeof(off)});
}
template <int Family>
......@@ -1093,15 +1102,12 @@ uint16_t new_ip_acceptor_impl(native_socket fd, uint16_t port,
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));
} else {
ccall(cc_one, "invalid IP address", inet_pton, Family, addr, &addr_of(sa));
}
port_of(sa) = htons(port);
if (bind(fd, reinterpret_cast<sockaddr*>(&sa), sizeof(sa)) != 0) {
throw bind_failure(last_socket_error_as_string());
}
ccall(cc_zero, "cannot bind socket", bind, fd,
reinterpret_cast<sockaddr*>(&sa), socklen_t{sizeof(sa)});
read_port(fd, sa);
return ntohs(port_of(sa));
}
......@@ -1115,33 +1121,28 @@ new_tcp_acceptor_impl(uint16_t port, const char* addr, bool reuse_addr) {
# endif
protocol proto = ipv6;
if (addr) {
auto addrs = interfaces::get_addrinfo_of_host(addr);
auto addrs = interfaces::native_address(addr);
if (!addrs) {
std::string errstr = "no such host: ";
errstr += addr;
throw network_error(std::move(errstr));
std::string errmsg = "invalid IP address: ";
errmsg += addr;
throw network_error(errmsg);
}
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");
}
auto fd = ccall(cc_valid_socket, "could not create server socket", socket,
proto == ipv4 ? AF_INET : AF_INET6, SOCK_STREAM, 0);
// sguard closes the socket in case of exception
socket_guard sguard(fd);
if (reuse_addr) {
int on = 1;
if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<setsockopt_ptr>(&on), sizeof(on)) < 0) {
throw_io_failure("unable to set SO_REUSEADDR");
}
ccall(cc_zero, "unable to set SO_REUSEADDR", setsockopt, fd, SOL_SOCKET,
SO_REUSEADDR, reinterpret_cast<setsockopt_ptr>(&on),
socklen_t{sizeof(on)});
}
auto p = proto == ipv4 ? new_ip_acceptor_impl<AF_INET>(fd, port, addr)
: new_ip_acceptor_impl<AF_INET6>(fd, port, addr);
if (listen(fd, SOMAXCONN) != 0) {
throw network_error("listen() failed: " + last_socket_error_as_string());
}
ccall(cc_zero, "listen() failed", listen, fd, SOMAXCONN);
// ok, no exceptions so far
CAF_LOGF_DEBUG("sockfd = " << fd << ", port = " << p);
return {sguard.release(), p};
......
......@@ -145,7 +145,7 @@ std::vector<std::string> interfaces::list_addresses(protocol proc,
}
optional<std::pair<std::string, protocol>>
interfaces::get_addrinfo_of_host(const std::string& host,
interfaces::native_address(const std::string& host,
optional<protocol> preferred) {
addrinfo hint;
memset(&hint, 0, sizeof(hint));
......
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