Commit 09e62f81 authored by Dominik Charousset's avatar Dominik Charousset

Bring the UDP socket API in line with the others

parent b2d9a553
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include <deque>
#include <vector>
#include "caf/byte_buffer.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/transport_base.hpp"
#include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/sec.hpp"
namespace caf::net {
template <class Factory>
using datagram_transport_base
= transport_base<datagram_transport<Factory>,
transport_worker_dispatcher<Factory, ip_endpoint>,
udp_datagram_socket, Factory, ip_endpoint>;
/// Implements a udp_transport policy that manages a datagram socket.
template <class Factory>
class datagram_transport : public datagram_transport_base<Factory> {
public:
// Maximal UDP-packet size
static constexpr size_t max_datagram_size
= std::numeric_limits<uint16_t>::max();
// -- member types -----------------------------------------------------------
using factory_type = Factory;
using id_type = ip_endpoint;
using application_type = typename factory_type::application_type;
using super = datagram_transport_base<factory_type>;
using buffer_cache_type = typename super::buffer_cache_type;
// -- constructors, destructors, and assignment operators --------------------
datagram_transport(udp_datagram_socket handle, factory_type factory)
: super(handle, std::move(factory)) {
// nop
}
// -- public member functions ------------------------------------------------
error start(endpoint_manager& manager) override {
CAF_LOG_TRACE("");
if (auto err = super::start(manager))
return err;
prepare_next_read();
return none;
}
bool handle_read_event(endpoint_manager&) override {
CAF_LOG_TRACE(CAF_ARG(this->handle_.id));
for (size_t reads = 0; reads < this->max_consecutive_reads_; ++reads) {
auto ret = read(this->handle_, this->read_buf_);
if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) {
auto& [num_bytes, ep] = *res;
CAF_LOG_DEBUG("received " << num_bytes << " bytes");
this->read_buf_.resize(num_bytes);
if (auto err = this->next_layer_.handle_data(*this, this->read_buf_,
std::move(ep))) {
CAF_LOG_ERROR("handle_data failed: " << err);
return false;
}
prepare_next_read();
} else {
auto err = get<sec>(ret);
if (err == sec::unavailable_or_would_block) {
break;
} else {
CAF_LOG_DEBUG("read failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
return false;
}
}
}
return true;
}
bool handle_write_event(endpoint_manager& manager) override {
CAF_LOG_TRACE(CAF_ARG2("socket", this->handle_.id)
<< CAF_ARG2("queue-size", packet_queue_.size()));
auto fetch_next_message = [&] {
if (auto msg = manager.next_message()) {
this->next_layer_.write_message(*this, std::move(msg));
return true;
}
return false;
};
do {
if (auto err = write_some())
return err == sec::unavailable_or_would_block;
} while (fetch_next_message());
return !packet_queue_.empty();
}
// TODO: remove this function. `resolve` should add workers when needed.
error add_new_worker(node_id node, id_type id) {
auto worker = this->next_layer_.add_new_worker(*this, node, id);
if (!worker)
return worker.error();
return none;
}
void write_packet(id_type id, span<byte_buffer*> buffers) override {
CAF_LOG_TRACE("");
CAF_ASSERT(!buffers.empty());
if (packet_queue_.empty())
this->manager().register_writing();
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
packet_queue_.emplace_back(id, buffers);
}
/// Helper struct for managing outgoing packets
struct packet {
id_type id;
buffer_cache_type bytes;
size_t size;
packet(id_type id, span<byte_buffer*> bufs) : id(id) {
size = 0;
for (auto buf : bufs) {
size += buf->size();
bytes.emplace_back(std::move(*buf));
}
}
std::vector<byte_buffer*> get_buffer_ptrs() {
std::vector<byte_buffer*> ptrs;
for (auto& buf : bytes)
ptrs.emplace_back(&buf);
return ptrs;
}
};
private:
// -- utility functions ------------------------------------------------------
void prepare_next_read() {
this->read_buf_.resize(max_datagram_size);
}
error write_some() {
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = packet_queue_.front();
auto& bufs = front.bytes;
auto it = bufs.begin();
if (this->header_bufs_.size() < this->header_bufs_.capacity()) {
it->clear();
this->header_bufs_.emplace_back(std::move(*it++));
}
for (; it != bufs.end()
&& this->payload_bufs_.size() < this->payload_bufs_.capacity();
++it) {
it->clear();
this->payload_bufs_.emplace_back(std::move(*it));
}
packet_queue_.pop_front();
};
// Write as many bytes as possible.
while (!packet_queue_.empty()) {
auto& packet = packet_queue_.front();
auto ptrs = packet.get_buffer_ptrs();
auto write_ret = write(this->handle_, ptrs, packet.id);
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(this->handle_.id) << CAF_ARG(*num_bytes));
CAF_LOG_WARNING_IF(*num_bytes < packet.size,
"packet was not sent completely");
recycle();
} else {
auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_ERROR("write failed" << CAF_ARG(err));
this->next_layer_.handle_error(err);
}
return err;
}
}
return none;
}
std::deque<packet> packet_queue_;
};
} // namespace caf::net
......@@ -49,7 +49,7 @@ error CAF_NET_EXPORT keepalive(stream_socket x, bool new_value);
error CAF_NET_EXPORT nodelay(stream_socket x, bool new_value);
/// Receives data from `x`.
/// @param x A connected endpoint.
/// @param x A connected socket.
/// @param buf Points to destination buffer.
/// @returns The number of received bytes on success, 0 if the socket is closed,
/// or -1 in case of an error.
......
......@@ -4,6 +4,7 @@
#pragma once
#include "caf//net/datagram_socket.hpp"
#include "caf/byte_span.hpp"
#include "caf/detail/net_export.hpp"
#include "caf/fwd.hpp"
......@@ -15,8 +16,8 @@ namespace caf::net {
/// A datagram-oriented network communication endpoint for bidirectional
/// byte transmission.
struct CAF_NET_EXPORT udp_datagram_socket : network_socket {
using super = network_socket;
struct CAF_NET_EXPORT udp_datagram_socket : datagram_socket {
using super = datagram_socket;
using super::super;
};
......@@ -27,37 +28,22 @@ struct CAF_NET_EXPORT udp_datagram_socket : network_socket {
/// specific port that was bound.
/// @returns The connected socket or an error.
/// @relates udp_datagram_socket
expected<std::pair<udp_datagram_socket, uint16_t>>
expected<udp_datagram_socket>
CAF_NET_EXPORT make_udp_datagram_socket(ip_endpoint ep,
bool reuse_addr = false);
/// Enables or disables `SIO_UDP_CONNRESET` error on `x`.
/// @relates udp_datagram_socket
error CAF_NET_EXPORT allow_connreset(udp_datagram_socket x, bool new_value);
bool reuse_addr = true);
/// Receives the next datagram on socket `x`.
/// @param x The UDP socket for receiving datagrams.
/// @param buf Writable output buffer.
/// @returns The number of received bytes and the sender as `ip_endpoint` on
/// success, an error code otherwise.
/// @param src If non-null, CAF optionally stores the address of the sender for
/// the datagram.
/// @returns The number of received bytes on success, 0 if the socket is closed,
/// or -1 in case of an error.
/// @relates udp_datagram_socket
/// @post buf was modified and the resulting integer represents the length of
/// the received datagram, even if it did not fit into the given buffer.
std::variant<std::pair<size_t, ip_endpoint>, sec>
CAF_NET_EXPORT read(udp_datagram_socket x, byte_span buf);
/// Sends the content of `bufs` as a datagram to the endpoint `ep` on socket
/// `x`.
/// @param x The UDP socket for sending datagrams.
/// @param bufs Points to the datagram to send, scattered across up to 10
/// buffers.
/// @param ep The endpoint to send the datagram to.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates udp_datagram_socket
/// @pre `bufs.size() < 10`
std::variant<size_t, sec> CAF_NET_EXPORT write(udp_datagram_socket x,
span<byte_buffer*> bufs,
ip_endpoint ep);
ptrdiff_t CAF_NET_EXPORT read(udp_datagram_socket x, byte_span buf,
ip_endpoint* src = nullptr);
/// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`.
/// @param x The UDP socket for sending datagrams.
......@@ -65,14 +51,7 @@ std::variant<size_t, sec> CAF_NET_EXPORT write(udp_datagram_socket x,
/// @param ep The endpoint to send the datagram to.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates udp_datagram_socket
std::variant<size_t, sec> CAF_NET_EXPORT write(udp_datagram_socket x,
const_byte_span buf,
ip_endpoint ep);
/// Converts the result from I/O operation on a ::udp_datagram_socket to either
/// an error code or a non-zero positive integer.
/// @relates udp_datagram_socket
std::variant<size_t, sec> CAF_NET_EXPORT
check_udp_datagram_socket_io_res(std::make_signed<size_t>::type res);
ptrdiff_t CAF_NET_EXPORT write(udp_datagram_socket x, const_byte_span buf,
ip_endpoint ep);
} // namespace caf::net
......@@ -16,33 +16,21 @@
#include "caf/net/socket_guard.hpp"
#include "caf/span.hpp"
namespace caf::net {
#ifdef CAF_WINDOWS
error allow_connreset(udp_datagram_socket x, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(new_value));
DWORD bytes_returned = 0;
CAF_NET_SYSCALL("WSAIoctl", res, !=, 0,
WSAIoctl(x.id, _WSAIOW(IOC_VENDOR, 12), &new_value,
sizeof(new_value), NULL, 0, &bytes_returned, NULL,
NULL));
return none;
}
namespace {
#else // CAF_WINDOWS
#if defined(CAF_WINDOWS) || defined(CAF_MACOS) || defined(CAF_IOS) \
|| defined(CAF_BSD)
constexpr int no_sigpipe_io_flag = 0;
#else
constexpr int no_sigpipe_io_flag = MSG_NOSIGNAL;
#endif
error allow_connreset(udp_datagram_socket x, bool) {
if (socket_cast<net::socket>(x) == invalid_socket)
return sec::socket_invalid;
// nop; SIO_UDP_CONNRESET only exists on Windows
return none;
}
} // namespace
#endif // CAF_WINDOWS
namespace caf::net {
expected<std::pair<udp_datagram_socket, uint16_t>>
make_udp_datagram_socket(ip_endpoint ep, bool reuse_addr) {
expected<udp_datagram_socket> make_udp_datagram_socket(ip_endpoint ep,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(ep));
sockaddr_storage addr = {};
detail::convert(ep, addr);
......@@ -61,120 +49,29 @@ make_udp_datagram_socket(ip_endpoint ep, bool reuse_addr) {
}
CAF_NET_SYSCALL("bind", err1, !=, 0,
::bind(sock.id, reinterpret_cast<sockaddr*>(&addr), len));
CAF_NET_SYSCALL("getsockname", err2, !=, 0,
getsockname(sock.id, reinterpret_cast<sockaddr*>(&addr),
&len));
CAF_LOG_DEBUG(CAF_ARG(sock.id));
auto port = addr.ss_family == AF_INET
? reinterpret_cast<sockaddr_in*>(&addr)->sin_port
: reinterpret_cast<sockaddr_in6*>(&addr)->sin6_port;
return std::make_pair(sguard.release(), ntohs(port));
return sguard.release();
}
std::variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
byte_span buf) {
ptrdiff_t read(udp_datagram_socket x, byte_span buf, ip_endpoint* src) {
sockaddr_storage addr = {};
socklen_t len = sizeof(sockaddr_storage);
auto res = ::recvfrom(x.id, reinterpret_cast<socket_recv_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr),
&len);
auto ret = check_udp_datagram_socket_io_res(res);
if (auto num_bytes = std::get_if<size_t>(&ret)) {
CAF_LOG_INFO_IF(*num_bytes == 0, "Received empty datagram");
CAF_LOG_WARNING_IF(*num_bytes > buf.size(),
"recvfrom cut of message, only received "
<< CAF_ARG(buf.size()) << " of " << CAF_ARG(num_bytes)
<< " bytes");
ip_endpoint ep;
if (auto err = detail::convert(addr, ep)) {
CAF_ASSERT(err.category() == type_id_v<sec>);
return static_cast<sec>(err.code());
}
return std::pair<size_t, ip_endpoint>(*num_bytes, ep);
} else {
return std::get<sec>(ret);
}
}
std::variant<size_t, sec> write(udp_datagram_socket x, const_byte_span buf,
ip_endpoint ep) {
sockaddr_storage addr = {};
detail::convert(ep, addr);
auto len = static_cast<socklen_t>(
ep.address().embeds_v4() ? sizeof(sockaddr_in) : sizeof(sockaddr_in6));
auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), len);
auto ret = check_udp_datagram_socket_io_res(res);
if (auto num_bytes = std::get_if<size_t>(&ret))
return *num_bytes;
else
return std::get<sec>(ret);
buf.size(), no_sigpipe_io_flag,
reinterpret_cast<sockaddr*>(&addr), &len);
if (src)
std::ignore = detail::convert(addr, *src);
return res;
}
#ifdef CAF_WINDOWS
std::variant<size_t, sec> write(udp_datagram_socket x, span<byte_buffer*> bufs,
ip_endpoint ep) {
CAF_ASSERT(bufs.size() < 10);
WSABUF buf_array[10];
auto convert = [](byte_buffer* buf) {
return WSABUF{static_cast<ULONG>(buf->size()),
reinterpret_cast<CHAR*>(buf->data())};
};
std::transform(bufs.begin(), bufs.end(), std::begin(buf_array), convert);
ptrdiff_t write(udp_datagram_socket x, const_byte_span buf, ip_endpoint ep) {
sockaddr_storage addr = {};
detail::convert(ep, addr);
auto len = ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6);
DWORD bytes_sent = 0;
auto res = WSASendTo(x.id, buf_array, static_cast<DWORD>(bufs.size()),
&bytes_sent, 0, reinterpret_cast<sockaddr*>(&addr), len,
nullptr, nullptr);
if (res != 0) {
auto code = last_socket_error();
if (code == std::errc::operation_would_block
|| code == std::errc::resource_unavailable_try_again)
return sec::unavailable_or_would_block;
return sec::socket_operation_failed;
}
return static_cast<size_t>(bytes_sent);
}
#else // CAF_WINDOWS
std::variant<size_t, sec> write(udp_datagram_socket x, span<byte_buffer*> bufs,
ip_endpoint ep) {
CAF_ASSERT(bufs.size() < 10);
auto convert = [](byte_buffer* buf) {
return iovec{buf->data(), buf->size()};
};
sockaddr_storage addr = {};
detail::convert(ep, addr);
iovec buf_array[10];
std::transform(bufs.begin(), bufs.end(), std::begin(buf_array), convert);
msghdr message = {};
memset(&message, 0, sizeof(msghdr));
message.msg_name = &addr;
message.msg_namelen = ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6);
message.msg_iov = buf_array;
message.msg_iovlen = static_cast<int>(bufs.size());
auto res = sendmsg(x.id, &message, 0);
return check_udp_datagram_socket_io_res(res);
}
#endif // CAF_WINDOWS
std::variant<size_t, sec>
check_udp_datagram_socket_io_res(std::make_signed<size_t>::type res) {
if (res < 0) {
auto code = last_socket_error();
if (code == std::errc::operation_would_block
|| code == std::errc::resource_unavailable_try_again)
return sec::unavailable_or_would_block;
return sec::socket_operation_failed;
}
return static_cast<size_t>(res);
return ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr),
static_cast<socklen_t>(len));
}
} // namespace caf::net
......@@ -6,15 +6,11 @@
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/test/dsl.hpp"
#include "net-test.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/byte_buffer.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/ip_address.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/ipv4_address.hpp"
#include "caf/net/ip.hpp"
using namespace caf;
......@@ -27,14 +23,12 @@ constexpr std::string_view hello_test = "Hello test!";
struct fixture {
fixture() : buf(1024) {
addresses = local_addresses("localhost");
auto addresses = local_addresses("localhost");
CAF_CHECK(!addresses.empty());
ep = ip_endpoint(*addresses.begin(), 0);
auto send_pair = unbox(make_udp_datagram_socket(ep));
send_socket = send_pair.first;
auto receive_pair = unbox(make_udp_datagram_socket(ep));
receive_socket = receive_pair.first;
ep.port(receive_pair.second);
ep = ip_endpoint(addresses.front(), 0);
send_socket = unbox(make_udp_datagram_socket(ep));
receive_socket = unbox(make_udp_datagram_socket(ep));
ep.port(unbox(local_port(receive_socket)));
}
~fixture() {
......@@ -42,9 +36,6 @@ struct fixture {
close(receive_socket);
}
std::vector<ip_address> addresses;
actor_system_config cfg;
actor_system sys{cfg};
ip_endpoint ep;
udp_datagram_socket send_socket;
udp_datagram_socket receive_socket;
......@@ -52,38 +43,20 @@ struct fixture {
};
error read_from_socket(udp_datagram_socket sock, byte_buffer& buf) {
uint8_t receive_attempts = 0;
std::variant<std::pair<size_t, ip_endpoint>, sec> read_ret;
do {
read_ret = read(sock, buf);
if (auto read_res = get_if<std::pair<size_t, ip_endpoint>>(&read_ret)) {
buf.resize(read_res->first);
} else if (get<sec>(read_ret) != sec::unavailable_or_would_block) {
return make_error(get<sec>(read_ret), "read failed");
auto receive_attempts = 0;
while (receive_attempts < 100) {
auto read_res = read(sock, buf);
if (read_res > 0) {
buf.resize(static_cast<size_t>(read_res));
return {};
}
if (++receive_attempts > 100)
return make_error(sec::runtime_error,
"too many unavailable_or_would_blocks");
} while (read_ret.index() != 0);
return none;
}
struct header {
header(size_t payload_size) : payload_size(payload_size) {
// nop
}
header() : header(0) {
// nop
}
template <class Inspector>
friend bool inspect(Inspector& f, header& x) {
return f.object(x).fields(f.field("payload_size", x.payload_size));
if (!last_socket_error_is_temporary())
return make_error(sec::socket_operation_failed,
last_socket_error_as_string());
++receive_attempts;
}
size_t payload_size;
};
return make_error(sec::runtime_error, "too many read attempts");
}
} // namespace
......@@ -91,56 +64,24 @@ CAF_TEST_FIXTURE_SCOPE(udp_datagram_socket_test, fixture)
CAF_TEST(socket creation) {
ip_endpoint ep;
CAF_CHECK_EQUAL(parse("0.0.0.0:0", ep), none);
auto ret = make_udp_datagram_socket(ep);
if (!ret)
CAF_FAIL("socket creation failed: " << ret.error());
CAF_CHECK_EQUAL(local_port(ret->first), ret->second);
CHECK_EQ(parse("0.0.0.0:0", ep), none);
CHECK(make_udp_datagram_socket(ep));
}
CAF_TEST(read / write using span<byte>) {
CAF_TEST(read and write) {
if (auto err = nonblocking(socket_cast<net::socket>(receive_socket), true))
CAF_FAIL("setting socket to nonblocking failed: " << err);
// Our first read must fail (nothing to receive yet).
auto read_res = read(receive_socket, buf);
if (CAF_CHECK(holds_alternative<sec>(read_res)))
CAF_CHECK(get<sec>(read_res) == sec::unavailable_or_would_block);
CHECK(read_res < 0);
CHECK(last_socket_error_is_temporary());
CAF_MESSAGE("sending data to " << to_string(ep));
auto write_res = write(send_socket, as_bytes(make_span(hello_test)), ep);
if (CAF_CHECK(holds_alternative<size_t>(write_res)))
CAF_CHECK_EQUAL(get<size_t>(write_res), hello_test.size());
CAF_CHECK_EQUAL(read_from_socket(receive_socket, buf), none);
CHECK_EQ(write_res, static_cast<ptrdiff_t>(hello_test.size()));
CHECK_EQ(read_from_socket(receive_socket, buf), none);
std::string_view received{reinterpret_cast<const char*>(buf.data()),
buf.size()};
CAF_CHECK_EQUAL(received, hello_test);
}
CAF_TEST(read / write using span<byte_buffer*>) {
// generate header and payload in separate buffers
header hdr{hello_test.size()};
byte_buffer hdr_buf;
binary_serializer sink(sys, hdr_buf);
if (!sink.apply(hdr))
CAF_FAIL("failed to serialize payload: " << sink.get_error());
auto bytes = as_bytes(make_span(hello_test));
byte_buffer payload_buf(bytes.begin(), bytes.end());
auto packet_size = hdr_buf.size() + payload_buf.size();
std::vector<byte_buffer*> bufs{&hdr_buf, &payload_buf};
auto write_res = write(send_socket, bufs, ep);
if (CAF_CHECK(holds_alternative<size_t>(write_res)))
CAF_CHECK_EQUAL(get<size_t>(write_res), packet_size);
// receive both as one single packet.
buf.resize(packet_size);
CAF_CHECK_EQUAL(read_from_socket(receive_socket, buf), none);
CAF_CHECK_EQUAL(buf.size(), packet_size);
binary_deserializer source(nullptr, buf);
header recv_hdr;
if (!source.apply(recv_hdr))
CAF_FAIL("failed to deserialize header: " << source.get_error());
CAF_CHECK_EQUAL(hdr.payload_size, recv_hdr.payload_size);
std::string_view received{reinterpret_cast<const char*>(buf.data())
+ sizeof(header),
buf.size() - sizeof(header)};
CAF_CHECK_EQUAL(received, hello_test);
CHECK_EQ(received, hello_test);
}
CAF_TEST_FIXTURE_SCOPE_END()
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