Commit 180bab00 authored by Joseph Noir's avatar Joseph Noir

Cleanup and move sockt guard to its own files

parent f0ff057b
......@@ -45,8 +45,9 @@ set(LIBCAF_IO_SRCS
src/stream.cpp
src/tcp.cpp
src/udp.cpp
src/socket_utils.cpp
src/native_socket.cpp
src/call_cfun.cpp
src/socket_guard.cpp
)
add_custom_target(libcaf_io)
......
......@@ -18,6 +18,9 @@
#pragma once
#include <cstdio>
#include <cstdlib>
#include "caf/io/network/native_socket.hpp"
namespace caf {
......@@ -42,7 +45,6 @@ bool cc_valid_socket(caf::io::network::native_socket fd);
return make_error(sec::network_syscall_failed, \
fun_name, last_socket_error_as_string())
#ifdef CAF_WINDOWS
// Calls a C functions and calls exit() if `predicate(var)` returns false.
#define CALL_CRITICAL_CFUN(var, predicate, funname, expr) \
auto var = expr; \
......@@ -52,10 +54,5 @@ bool cc_valid_socket(caf::io::network::native_socket fd);
abort(); \
} static_cast<void>(0)
#ifndef SIO_UDP_CONNRESET
#define SIO_UDP_CONNRESET _WSAIOW(IOC_VENDOR, 12)
#endif
#endif // CAF_WINDOWS
} // namespace detail
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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 "caf/io/network/native_socket.hpp"
namespace caf {
namespace detail {
class socket_guard {
public:
explicit socket_guard(io::network::native_socket fd);
~socket_guard();
io::network::native_socket release();
void close();
private:
io::network::native_socket fd_;
};
} // namespace detail
} // namespace detail
......@@ -29,7 +29,7 @@
#include "caf/io/network/ip_endpoint.hpp"
#include "caf/io/network/event_handler.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/datagram_manager.hpp"
namespace caf {
......@@ -101,6 +101,7 @@ public:
const manager_ptr mgr);
std::unordered_map<datagram_handle, ip_endpoint>& endpoints();
const std::unordered_map<datagram_handle, ip_endpoint>& endpoints() const;
void remove_endpoint(datagram_handle hdl);
......
......@@ -42,15 +42,13 @@
#include "caf/io/network/ip_endpoint.hpp"
#include "caf/io/network/multiplexer.hpp"
#include "caf/io/network/pipe_reader.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/io/network/event_handler.hpp"
#include "caf/io/network/receive_buffer.hpp"
#include "caf/io/network/stream_manager.hpp"
#include "caf/io/network/acceptor_manager.hpp"
#include "caf/io/network/datagram_manager.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/logger.hpp"
// poll xs epoll backend
......
......@@ -18,14 +18,40 @@
#pragma once
#include <string>
#include <cstdint>
#include "caf/config.hpp"
#include "caf/expected.hpp"
namespace caf {
namespace io {
namespace network {
// Annoying platform-dependent bootstrapping.
#ifdef CAF_WINDOWS
using setsockopt_ptr = const char*;
using getsockopt_ptr = char*;
using socket_send_ptr = const char*;
using socket_recv_ptr = char*;
//using socket_size_type = int;
#else
using setsockopt_ptr = const void*;
using getsockopt_ptr = void*;
using socket_send_ptr = const void*;
using socket_recv_ptr = void*;
using socket_size_type = unsigned;
void closesocket(int fd);
#endif
using signed_size_type = std::make_signed<size_t>::type;
// More bootstrapping.
extern const int ec_out_of_memory;
extern const int ec_interrupted_syscall;
extern const int no_sigpipe_socket_flag;
extern const int no_sigpipe_io_flag;
#ifdef CAF_WINDOWS
using native_socket = size_t;
constexpr native_socket invalid_native_socket = static_cast<native_socket>(-1);
......@@ -40,6 +66,57 @@ namespace network {
}
#endif
/// Returns the last socket error as an integer.
int last_socket_error();
/// Returns true if `errcode` indicates that an operation would
/// block or return nothing at the moment and can be tried again
/// at a later point.
bool would_block_or_temporarily_unavailable(int errcode);
/// Returns the last socket error as human-readable string.
std::string last_socket_error_as_string();
/// Creates two connected sockets. The former is the read handle
/// and the latter is the write handle.
std::pair<native_socket, native_socket> create_pipe();
/// Sets fd to nonblocking if `set_nonblocking == true`
/// or to blocking if `set_nonblocking == false`
/// throws `network_error` on error
expected<void> nonblocking(native_socket fd, bool new_value);
/// Enables or disables Nagle's algorithm on `fd`.
/// @throws network_error
expected<void> tcp_nodelay(native_socket fd, bool new_value);
/// Enables or disables `SIGPIPE` events from `fd`.
expected<void> allow_sigpipe(native_socket fd, bool new_value);
/// Enables or disables `SIO_UDP_CONNRESET`error on `fd`.
expected<void> allow_udp_connreset(native_socket fd, bool new_value);
/// Get the socket buffer size for `fd`.
expected<int> send_buffer_size(native_socket fd);
/// Set the socket buffer size for `fd`.
expected<void> send_buffer_size(native_socket fd, int new_value);
/// Convenience functions for checking the result of `recv` or `send`.
bool is_error(signed_size_type res, bool is_nonblock);
/// Returns the locally assigned port of `fd`.
expected<uint16_t> local_port_of_fd(native_socket fd);
/// Returns the locally assigned address of `fd`.
expected<std::string> local_addr_of_fd(native_socket fd);
/// Returns the port used by the remote host of `fd`.
expected<uint16_t> remote_port_of_fd(native_socket fd);
/// Returns the remote host address of `fd`.
expected<std::string> remote_addr_of_fd(native_socket fd);
} // namespace network
} // namespace io
} // namespace caf
......
......@@ -32,9 +32,13 @@ namespace network {
class pipe_reader : public event_handler {
public:
pipe_reader(default_multiplexer& dm);
void removed_from_loop(operation op) override;
void handle_event(operation op) override;
void init(native_socket sock_fd);
resumable* try_read_next();
};
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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 <string>
#include "caf/expected.hpp"
#include "caf/io/network/native_socket.hpp"
#ifdef CAF_WINDOWS
# ifndef WIN32_LEAN_AND_MEAN
# define WIN32_LEAN_AND_MEAN
# endif // WIN32_LEAN_AND_MEAN
# ifndef NOMINMAX
# define NOMINMAX
# endif
# ifdef CAF_MINGW
# undef _WIN32_WINNT
# undef WINVER
# define _WIN32_WINNT WindowsVista
# define WINVER WindowsVista
# include <w32api.h>
# endif
# include <winsock2.h>
# include <windows.h>
# include <ws2tcpip.h>
# include <ws2ipdef.h>
#else
# include <unistd.h>
# include <cerrno>
# include <sys/socket.h>
# include <netinet/in.h>
# include <netinet/ip.h>
#endif
namespace caf {
namespace io {
namespace network {
// annoying platform-dependent bootstrapping
#ifdef CAF_WINDOWS
using setsockopt_ptr = const char*;
using getsockopt_ptr = char*;
using socket_send_ptr = const char*;
using socket_recv_ptr = char*;
using socklen_t = int;
using ssize_t = std::make_signed<size_t>::type;
inline int last_socket_error() { return WSAGetLastError(); }
inline bool would_block_or_temporarily_unavailable(int errcode) {
return errcode == WSAEWOULDBLOCK || errcode == WSATRY_AGAIN;
}
constexpr int ec_out_of_memory = WSAENOBUFS;
constexpr int ec_interrupted_syscall = WSAEINTR;
#else
using setsockopt_ptr = const void*;
using getsockopt_ptr = void*;
using socket_send_ptr = const void*;
using socket_recv_ptr = void*;
inline void closesocket(int fd) { close(fd); }
inline int last_socket_error() { return errno; }
inline bool would_block_or_temporarily_unavailable(int errcode) {
return errcode == EAGAIN || errcode == EWOULDBLOCK;
}
constexpr int ec_out_of_memory = ENOMEM;
constexpr int ec_interrupted_syscall = EINTR;
#endif
// platform-dependent SIGPIPE setup
#if defined(CAF_MACOS) || defined(CAF_IOS) || defined(CAF_BSD)
// Use the socket option but no flags to recv/send on macOS/iOS/BSD.
constexpr int no_sigpipe_socket_flag = SO_NOSIGPIPE;
constexpr int no_sigpipe_io_flag = 0;
#elif defined(CAF_WINDOWS)
// Do nothing on Windows (SIGPIPE does not exist).
constexpr int no_sigpipe_socket_flag = 0;
constexpr int no_sigpipe_io_flag = 0;
#else
// Use flags to recv/send on Linux/Android but no socket option.
constexpr int no_sigpipe_socket_flag = 0;
constexpr int no_sigpipe_io_flag = MSG_NOSIGNAL;
#endif
/// Returns the last socket error as human-readable string.
std::string last_socket_error_as_string();
/// Creates two connected sockets. The former is the read handle
/// and the latter is the write handle.
std::pair<native_socket, native_socket> create_pipe();
/// Sets fd to nonblocking if `set_nonblocking == true`
/// or to blocking if `set_nonblocking == false`
/// throws `network_error` on error
expected<void> nonblocking(native_socket fd, bool new_value);
/// Enables or disables Nagle's algorithm on `fd`.
/// @throws network_error
expected<void> tcp_nodelay(native_socket fd, bool new_value);
/// Enables or disables `SIGPIPE` events from `fd`.
expected<void> allow_sigpipe(native_socket fd, bool new_value);
/// Enables or disables `SIO_UDP_CONNRESET`error on `fd`.
expected<void> allow_udp_connreset(native_socket fd, bool new_value);
/// Get the socket buffer size for `fd`.
expected<int> send_buffer_size(native_socket fd);
/// Set the socket buffer size for `fd`.
expected<void> send_buffer_size(native_socket fd, int new_value);
/// Convenience functions for checking the result of `recv` or `send`.
bool is_error(ssize_t res, bool is_nonblock);
/// Returns the locally assigned port of `fd`.
expected<uint16_t> local_port_of_fd(native_socket fd);
/// Returns the locally assigned address of `fd`.
expected<std::string> local_addr_of_fd(native_socket fd);
/// Returns the port used by the remote host of `fd`.
expected<uint16_t> remote_port_of_fd(native_socket fd);
/// Returns the remote host address of `fd`.
expected<std::string> remote_addr_of_fd(native_socket fd);
/// @cond PRIVTE
// Utility functions to privde access to sockaddr fields.
auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)&;
auto family_of(sockaddr_in& what) -> decltype(what.sin_family)&;
auto port_of(sockaddr_in& what) -> decltype(what.sin_port)&;
auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)&;
auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)&;
auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)&;
auto port_of(sockaddr& what) -> decltype(port_of(std::declval<sockaddr_in&>()));
/// @endcond
} // namespace network
} // namespace io
} // namespace caf
......@@ -24,7 +24,7 @@
namespace caf {
namespace policy {
/// Policy object for wrapping default UDP operations
/// Policy object for wrapping default UDP operations.
struct udp {
/// Write a datagram containing `buf_len` bytes to `fd` addressed
/// at the endpoint in `sa` with size `sa_len`. Returns true as long
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
......@@ -25,7 +24,6 @@
#include "caf/defaults.hpp"
#include "caf/config_value.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace {
......@@ -75,7 +73,6 @@ void datagram_handler::ack_writes(bool x) {
ack_writes_ = x;
}
void datagram_handler::write(datagram_handle hdl, const void* buf,
size_t num_bytes) {
wr_offline_buf_.emplace_back();
......@@ -96,7 +93,6 @@ void datagram_handler::flush(const manager_ptr& mgr) {
}
}
std::unordered_map<datagram_handle, ip_endpoint>& datagram_handler::endpoints() {
return ep_by_hdl_;
}
......@@ -141,7 +137,7 @@ void datagram_handler::removed_from_loop(operation op) {
switch (op) {
case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break;
case operation::propagate_error: break;
case operation::propagate_error: ; // nop
};
}
......
......@@ -24,7 +24,6 @@
#include "caf/logger.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf {
......
......@@ -33,18 +33,21 @@
#include "caf/io/network/datagram_servant_impl.hpp"
#include "caf/detail/call_cfun.hpp"
#include "caf/detail/socket_guard.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
# include <ws2tcpip.h> // socklen_t, etc. (MSVC20xx)
# include <ws2tcpip.h> // socket_size_type, etc. (MSVC20xx)
# include <windows.h>
# include <io.h>
# include <unistd.h>
#else
# include <cerrno>
# include <netdb.h>
# include <fcntl.h>
# include <unistd.h>
# include <sys/types.h>
# include <arpa/inet.h>
# include <sys/socket.h>
......@@ -57,10 +60,34 @@ using std::string;
namespace {
// safe ourselves some typing
// Save ourselves some typing.
constexpr auto ipv4 = caf::io::network::protocol::ipv4;
constexpr auto ipv6 = caf::io::network::protocol::ipv6;
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;
}
} // namespace <anonymous>
namespace caf {
......@@ -665,34 +692,6 @@ int64_t default_multiplexer::next_endpoint_id() {
// -- Related helper functions -------------------------------------------------
class socket_guard {
public:
explicit socket_guard(native_socket fd) : fd_(fd) {
// nop
}
~socket_guard() {
close();
}
native_socket release() {
auto fd = fd_;
fd_ = invalid_native_socket;
return fd;
}
void close() {
if (fd_ != invalid_native_socket) {
CAF_LOG_DEBUG("close socket" << CAF_ARG(fd_));
closesocket(fd_);
fd_ = invalid_native_socket;
}
}
private:
native_socket fd_;
};
template <int Family>
bool ip_connect(native_socket fd, const std::string& host, uint16_t port) {
CAF_LOG_TRACE("Family =" << (Family == AF_INET ? "AF_INET" : "AF_INET6")
......@@ -726,7 +725,7 @@ new_tcp_connection(const std::string& host, uint16_t port,
CAF_ASSERT(proto == ipv4 || proto == ipv6);
CALL_CFUN(fd, detail::cc_valid_socket, "socket",
socket(proto == ipv4 ? AF_INET : AF_INET6, SOCK_STREAM, 0));
socket_guard sguard(fd);
detail::socket_guard sguard(fd);
if (proto == ipv6) {
if (ip_connect<AF_INET6>(fd, res->first, port)) {
CAF_LOG_INFO("successfully connected to host via IPv6");
......@@ -747,7 +746,7 @@ new_tcp_connection(const std::string& host, uint16_t port,
template <class SockAddrType>
expected<void> read_port(native_socket fd, SockAddrType& sa) {
socklen_t len = sizeof(SockAddrType);
socket_size_type len = sizeof(SockAddrType);
CALL_CFUN(res, detail::cc_zero, "getsockname",
getsockname(fd, reinterpret_cast<sockaddr*>(&sa), &len));
return unit;
......@@ -765,7 +764,7 @@ expected<void> set_inaddr_any(native_socket fd, sockaddr_in6& sa) {
CALL_CFUN(res, detail::cc_zero, "setsockopt",
setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY,
reinterpret_cast<setsockopt_ptr>(&off),
static_cast<socklen_t>(sizeof(off))));
static_cast<socket_size_type>(sizeof(off))));
return unit;
}
......@@ -776,13 +775,13 @@ expected<native_socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
CALL_CFUN(fd, detail::cc_valid_socket, "socket", socket(Family, SockType, 0));
// sguard closes the socket in case of exception
socket_guard sguard{fd};
detail::socket_guard sguard{fd};
if (reuse_addr) {
int on = 1;
CALL_CFUN(tmp1, detail::cc_zero, "setsockopt",
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<setsockopt_ptr>(&on),
static_cast<socklen_t>(sizeof(on))));
static_cast<socket_size_type>(sizeof(on))));
}
using sockaddr_type =
typename std::conditional<
......@@ -800,7 +799,7 @@ expected<native_socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
port_of(sa) = htons(port);
CALL_CFUN(res, detail::cc_zero, "bind",
bind(fd, reinterpret_cast<sockaddr*>(&sa),
static_cast<socklen_t>(sizeof(sa))));
static_cast<socket_size_type>(sizeof(sa))));
return sguard.release();
}
......@@ -832,7 +831,7 @@ expected<native_socket> new_tcp_acceptor_impl(uint16_t port, const char* addr,
return make_error(sec::cannot_open_port, "tcp socket creation failed",
port, addr_str);
}
socket_guard sguard{fd};
detail::socket_guard sguard{fd};
CALL_CFUN(tmp2, detail::cc_zero, "listen", listen(fd, SOMAXCONN));
// ok, no errors so far
CAF_LOG_DEBUG(CAF_ARG(fd));
......@@ -846,7 +845,7 @@ new_remote_udp_endpoint_impl(const std::string& host, uint16_t port,
auto lep = new_local_udp_endpoint_impl(0, nullptr, false, preferred);
if (!lep)
return std::move(lep.error());
socket_guard sguard{(*lep).first};
detail::socket_guard sguard{(*lep).first};
std::pair<native_socket, ip_endpoint> info;
memset(std::get<1>(info).address(), 0, sizeof(sockaddr_storage));
if (!interfaces::get_endpoint(host, port, std::get<1>(info), (*lep).second))
......
......@@ -23,7 +23,6 @@
#include "caf/logger.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf {
......
......@@ -20,9 +20,14 @@
#include "caf/logger.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
#else
# include <sys/socket.h>
#endif
namespace caf {
namespace io {
namespace network {
......@@ -69,4 +74,3 @@ void event_handler::set_fd_flags() {
} // namespace network
} // namespace io
} // namespace caf
......@@ -21,6 +21,8 @@
#include "caf/sec.hpp"
#include "caf/logger.hpp"
#include "caf/io/network/native_socket.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
# include <windows.h>
......@@ -209,12 +211,12 @@ std::string host(const ip_endpoint& ep) {
case AF_INET:
inet_ntop(AF_INET,
&const_cast<sockaddr_in*>(reinterpret_cast<const sockaddr_in*>(ep.caddress()))->sin_addr,
addr, static_cast<socklen_t>(*ep.clength()));
addr, static_cast<socket_size_type>(*ep.clength()));
break;
case AF_INET6:
inet_ntop(AF_INET6,
&const_cast<sockaddr_in6*>(reinterpret_cast<const sockaddr_in6*>(ep.caddress()))->sin6_addr,
addr, static_cast<socklen_t>(*ep.clength()));
addr, static_cast<socket_size_type>(*ep.clength()));
break;
default:
addr[0] = '\0';
......
......@@ -16,47 +16,107 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/network/native_socket.hpp"
#include "caf/sec.hpp"
#include "caf/logger.hpp"
#include "caf/detail/call_cfun.hpp"
#include "caf/io/network/protocol.hpp"
#include "caf/io/network/socket_utils.hpp"
#ifdef CAF_WINDOWS
# ifndef WIN32_LEAN_AND_MEAN
# define WIN32_LEAN_AND_MEAN
# endif
# ifndef NOMINMAX
# define NOMINMAX
# endif
# ifdef CAF_MINGW
# undef _WIN32_WINNT
# undef WINVER
# define _WIN32_WINNT WindowsVista
# define WINVER WindowsVista
# include <w32api.h>
# endif
# include <winsock2.h>
# include <ws2tcpip.h> // socklen_t, etc. (MSVC20xx)
# include <windows.h>
# include <io.h>
# include <ws2tcpip.h>
# include <ws2ipdef.h>
#else
# include <cerrno>
# include <netdb.h>
# include <fcntl.h>
# include <sys/types.h>
# include <unistd.h>
# include <arpa/inet.h>
# include <sys/socket.h>
# include <netinet/in.h>
# include <netinet/ip.h>
# include <netinet/tcp.h>
# include <utility>
#endif
using std::string;
namespace {
#ifdef CAF_WINDOWS
#ifndef SIO_UDP_CONNRESET
#define SIO_UDP_CONNRESET _WSAIOW(IOC_VENDOR, 12)
#endif
#endif // CAF_WINDOWS
auto port_of(sockaddr_in& what) -> decltype(what.sin_port)& {
return what.sin_port;
}
} // namespace <anonymous>
auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
auto port_of(sockaddr& what) -> decltype(port_of(std::declval<sockaddr_in&>())) {
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;
}
CAF_CRITICAL("invalid protocol family");
}
}
namespace caf {
namespace io {
namespace network {
#ifdef CAF_WINDOWS
int last_socket_error() { return WSAGetLastError(); }
bool would_block_or_temporarily_unavailable(int errcode) {
return errcode == WSAEWOULDBLOCK || errcode == WSATRY_AGAIN;
}
const int ec_out_of_memory = WSAENOBUFS;
const int ec_interrupted_syscall = WSAEINTR;
#else
void closesocket(int fd) { close(fd); }
int last_socket_error() { return errno; }
bool would_block_or_temporarily_unavailable(int errcode) {
return errcode == EAGAIN || errcode == EWOULDBLOCK;
}
const int ec_out_of_memory = ENOMEM;
const int ec_interrupted_syscall = EINTR;
#endif
// platform-dependent SIGPIPE setup
#if defined(CAF_MACOS) || defined(CAF_IOS) || defined(CAF_BSD)
// Use the socket option but no flags to recv/send on macOS/iOS/BSD.
const int no_sigpipe_socket_flag = SO_NOSIGPIPE;
const int no_sigpipe_io_flag = 0;
#elif defined(CAF_WINDOWS)
// Do nothing on Windows (SIGPIPE does not exist).
const int no_sigpipe_socket_flag = 0;
const int no_sigpipe_io_flag = 0;
#else
// Use flags to recv/send on Linux/Android but no socket option.
const int no_sigpipe_socket_flag = 0;
const int no_sigpipe_io_flag = MSG_NOSIGNAL;
#endif
#ifndef CAF_WINDOWS
string last_socket_error_as_string() {
......@@ -114,7 +174,7 @@ namespace network {
(LPTSTR) & errorText, // output
0, // minimum size for output buffer
nullptr); // arguments - see note
std::string result;
string result;
if (errorText != nullptr) {
result = errorText;
// release memory allocated by FormatMessage()
......@@ -138,7 +198,7 @@ namespace network {
expected<void> allow_udp_connreset(native_socket fd, bool new_value) {
DWORD bytes_returned = 0;
CALL_CFUN(res, detail::cc_zero, "WSAIoctl",
WSAIoctl(fd, SIO_UDP_CONNRESET, &new_value, sizeof(new_value),
WSAIoctl(fd, _WSAIOW(IOC_VENDOR, 12), &new_value, sizeof(new_value),
NULL, 0, &bytes_returned, NULL, NULL));
return unit;
}
......@@ -175,7 +235,7 @@ namespace network {
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *
\**************************************************************************/
std::pair<native_socket, native_socket> create_pipe() {
socklen_t addrlen = sizeof(sockaddr_in);
socket_size_type addrlen = sizeof(sockaddr_in);
native_socket socks[2] = {invalid_native_socket, invalid_native_socket};
CALL_CRITICAL_CFUN(listener, detail::cc_valid_socket, "socket",
socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
......@@ -232,7 +292,7 @@ namespace network {
expected<int> send_buffer_size(native_socket fd) {
int size;
socklen_t ret_size = sizeof(size);
socket_size_type ret_size = sizeof(size);
CALL_CFUN(res, detail::cc_zero, "getsockopt",
getsockopt(fd, SOL_SOCKET, SO_SNDBUF,
reinterpret_cast<getsockopt_ptr>(&size), &ret_size));
......@@ -243,7 +303,7 @@ expected<void> send_buffer_size(native_socket fd, int new_value) {
CALL_CFUN(res, detail::cc_zero, "setsockopt",
setsockopt(fd, SOL_SOCKET, SO_SNDBUF,
reinterpret_cast<setsockopt_ptr>(&new_value),
static_cast<socklen_t>(sizeof(int))));
static_cast<socket_size_type>(sizeof(int))));
return unit;
}
......@@ -253,11 +313,11 @@ expected<void> tcp_nodelay(native_socket fd, bool new_value) {
CALL_CFUN(res, detail::cc_zero, "setsockopt",
setsockopt(fd, IPPROTO_TCP, TCP_NODELAY,
reinterpret_cast<setsockopt_ptr>(&flag),
static_cast<socklen_t>(sizeof(flag))));
static_cast<socket_size_type>(sizeof(flag))));
return unit;
}
bool is_error(ssize_t res, bool is_nonblock) {
bool is_error(signed_size_type res, bool is_nonblock) {
if (res < 0) {
auto err = last_socket_error();
if (!is_nonblock || !would_block_or_temporarily_unavailable(err)) {
......@@ -269,9 +329,9 @@ bool is_error(ssize_t res, bool is_nonblock) {
return false;
}
expected<std::string> local_addr_of_fd(native_socket fd) {
expected<string> local_addr_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
socket_size_type st_len = sizeof(st);
sockaddr* sa = reinterpret_cast<sockaddr*>(&st);
CALL_CFUN(tmp1, detail::cc_zero, "getsockname", getsockname(fd, sa, &st_len));
char addr[INET6_ADDRSTRLEN] {0};
......@@ -292,15 +352,15 @@ expected<std::string> local_addr_of_fd(native_socket fd) {
expected<uint16_t> local_port_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
socket_size_type st_len = sizeof(st);
CALL_CFUN(tmp, detail::cc_zero, "getsockname",
getsockname(fd, reinterpret_cast<sockaddr*>(&st), &st_len));
return ntohs(port_of(reinterpret_cast<sockaddr&>(st)));
}
expected<std::string> remote_addr_of_fd(native_socket fd) {
expected<string> remote_addr_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
socket_size_type st_len = sizeof(st);
sockaddr* sa = reinterpret_cast<sockaddr*>(&st);
CALL_CFUN(tmp, detail::cc_zero, "getpeername", getpeername(fd, sa, &st_len));
char addr[INET6_ADDRSTRLEN] {0};
......@@ -321,49 +381,12 @@ expected<std::string> remote_addr_of_fd(native_socket fd) {
expected<uint16_t> remote_port_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
socket_size_type st_len = sizeof(st);
CALL_CFUN(tmp, detail::cc_zero, "getpeername",
getpeername(fd, reinterpret_cast<sockaddr*>(&st), &st_len));
return ntohs(port_of(reinterpret_cast<sockaddr&>(st)));
}
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;
}
auto port_of(sockaddr& what) -> decltype(port_of(std::declval<sockaddr_in&>())) {
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;
}
CAF_CRITICAL("invalid protocol family");
}
} // namespace network
} // namespace io
} // namespace caf
......@@ -18,10 +18,18 @@
#include "caf/logger.hpp"
#include <cstdint>
#include "caf/io/network/pipe_reader.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/default_multiplexer.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
#else
# include <unistd.h>
# include <sys/socket.h>
#endif
namespace caf {
namespace io {
namespace network {
......@@ -36,7 +44,7 @@ void pipe_reader::removed_from_loop(operation) {
}
resumable* pipe_reader::try_read_next() {
intptr_t ptrval;
std::intptr_t ptrval;
// on windows, we actually have sockets, otherwise we have file handles
# ifdef CAF_WINDOWS
auto res = recv(fd(), reinterpret_cast<socket_recv_ptr>(&ptrval),
......
......@@ -24,7 +24,6 @@
#include "caf/logger.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/detail/socket_guard.hpp"
#include <unistd.h>
#include "caf/logger.hpp"
namespace caf {
namespace detail {
socket_guard::socket_guard(io::network::native_socket fd) : fd_(fd) {
// nop
}
socket_guard::~socket_guard() {
close();
}
io::network::native_socket socket_guard::release() {
auto fd = fd_;
fd_ = io::network::invalid_native_socket;
return fd;
}
void socket_guard::close() {
if (fd_ != io::network::invalid_native_socket) {
CAF_LOG_DEBUG("close socket" << CAF_ARG(fd_));
io::network::closesocket(fd_);
fd_ = io::network::invalid_native_socket;
}
}
} // namespace detail
} // namespace detail
......@@ -25,7 +25,6 @@
#include "caf/defaults.hpp"
#include "caf/config_value.hpp"
#include "caf/io/network/socket_utils.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf {
......@@ -96,7 +95,7 @@ void stream::removed_from_loop(operation op) {
switch (op) {
case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break;
case operation::propagate_error: break;
case operation::propagate_error: ; // nop
}
}
......
......@@ -18,12 +18,21 @@
#include "caf/policy/tcp.hpp"
#include <cstring>
#include "caf/logger.hpp"
#include "caf/io/network/socket_utils.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
#else
# include <sys/types.h>
# include <sys/socket.h>
#endif
using caf::io::network::is_error;
using caf::io::network::rw_state;
using caf::io::network::native_socket;
using caf::io::network::socket_size_type;
using caf::io::network::no_sigpipe_io_flag;
namespace caf {
......@@ -59,8 +68,8 @@ bool tcp::try_accept(native_socket& result, native_socket fd) {
using namespace io::network;
CAF_LOG_TRACE(CAF_ARG(fd));
sockaddr_storage addr;
memset(&addr, 0, sizeof(addr));
socklen_t addrlen = sizeof(addr);
std::memset(&addr, 0, sizeof(addr));
socket_size_type addrlen = sizeof(addr);
result = ::accept(fd, reinterpret_cast<sockaddr*>(&addr), &addrlen);
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(result));
if (result == invalid_native_socket) {
......
......@@ -19,17 +19,19 @@
#include "caf/policy/udp.hpp"
#include "caf/logger.hpp"
#include "caf/io/network/socket_utils.hpp"
#ifdef CAF_WINDOWS
using caf::io::network::ssize_t;
# include <winsock2.h>
#else
# include <sys/types.h>
# include <sys/socket.h>
#endif
using caf::io::network::is_error;
using caf::io::network::ip_endpoint;
using caf::io::network::native_socket;
using caf::io::network::signed_size_type;
using caf::io::network::socket_size_type;
namespace caf {
namespace policy {
......@@ -38,7 +40,7 @@ bool udp::read_datagram(size_t& result, native_socket fd, void* buf,
size_t buf_len, ip_endpoint& ep) {
CAF_LOG_TRACE(CAF_ARG(fd));
memset(ep.address(), 0, sizeof(sockaddr_storage));
socklen_t len = sizeof(sockaddr_storage);
socket_size_type len = sizeof(sockaddr_storage);
auto sres = ::recvfrom(fd, static_cast<io::network::socket_recv_ptr>(buf),
buf_len, 0, ep.address(), &len);
if (is_error(sres, true)) {
......@@ -47,7 +49,7 @@ bool udp::read_datagram(size_t& result, native_socket fd, void* buf,
}
if (sres == 0)
CAF_LOG_INFO("Received empty datagram");
else if (sres > static_cast<ssize_t>(buf_len))
else if (sres > static_cast<signed_size_type>(buf_len))
CAF_LOG_WARNING("recvfrom cut of message, only received "
<< CAF_ARG(buf_len) << " of " << CAF_ARG(sres) << " bytes");
result = (sres > 0) ? static_cast<size_t>(sres) : 0;
......@@ -58,7 +60,7 @@ bool udp::read_datagram(size_t& result, native_socket fd, void* buf,
bool udp::write_datagram(size_t& result, native_socket fd, void* buf,
size_t buf_len, const ip_endpoint& ep) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(buf_len));
socklen_t len = static_cast<socklen_t>(*ep.clength());
socket_size_type len = static_cast<socket_size_type>(*ep.clength());
auto sres = ::sendto(fd, reinterpret_cast<io::network::socket_send_ptr>(buf),
buf_len, 0, ep.caddress(), len);
if (is_error(sres, true)) {
......
......@@ -42,6 +42,14 @@
#include "caf/openssl/session.hpp"
#ifdef CAF_WINDOWS
# include <winsock2.h>
# include <ws2tcpip.h> // socket_size_type, etc. (MSVC20xx)
#else
# include <sys/types.h>
# include <sys/socket.h>
#endif
namespace caf {
namespace openssl {
......@@ -70,7 +78,7 @@ struct ssl_policy {
CAF_LOG_TRACE(CAF_ARG(fd));
sockaddr_storage addr;
memset(&addr, 0, sizeof(addr));
socklen_t addrlen = sizeof(addr);
caf::io::network::socket_size_type addrlen = sizeof(addr);
result = accept(fd, reinterpret_cast<sockaddr*>(&addr), &addrlen);
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(result));
if (result == io::network::invalid_native_socket) {
......
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