Unverified Commit cdea6c3b authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #34

Add `datagram-transport`
parents 71254c55 17def710
...@@ -52,5 +52,6 @@ ...@@ -52,5 +52,6 @@
# include <netinet/tcp.h> # include <netinet/tcp.h>
# include <sys/socket.h> # include <sys/socket.h>
# include <unistd.h> # include <unistd.h>
# include <sys/uio.h>
#endif #endif
// clang-format on // clang-format on
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <deque>
#include <unordered_map>
#include "caf/byte.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/sec.hpp"
#include "caf/span.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace net {
/// Implements a udp_transport policy that manages a datagram socket.
template <class Factory>
class datagram_transport {
public:
// Maximal UDP-packet size
static constexpr size_t max_datagram_size = std::numeric_limits<
uint16_t>::max();
// -- member types -----------------------------------------------------------
using id_type = ip_endpoint;
using buffer_type = std::vector<byte>;
using buffer_cache_type = std::vector<buffer_type>;
using factory_type = Factory;
using transport_type = datagram_transport<Factory>;
using dispatcher_type = transport_worker_dispatcher<transport_type, id_type>;
using application_type = typename Factory::application_type;
// -- constructors, destructors, and assignment operators --------------------
datagram_transport(udp_datagram_socket handle, factory_type factory)
: dispatcher_(*this, std::move(factory)),
handle_(handle),
read_buf_(max_datagram_size),
manager_(nullptr) {
// nop
}
// -- properties -------------------------------------------------------------
udp_datagram_socket handle() const noexcept {
return handle_;
}
actor_system& system() {
return manager().system();
}
transport_type& transport() {
return *this;
}
endpoint_manager& manager() {
return *manager_;
}
// -- public member functions ------------------------------------------------
template <class Parent>
error init(Parent& parent) {
manager_ = &parent;
auto& cfg = system().config();
auto max_header_bufs = get_or(cfg, "middleman.max-header-buffers",
defaults::middleman::max_header_buffers);
header_bufs_.reserve(max_header_bufs);
auto max_payload_bufs = get_or(cfg, "middleman.max-payload-buffers",
defaults::middleman::max_payload_buffers);
payload_bufs_.reserve(max_payload_bufs);
if (auto err = dispatcher_.init(*this))
return err;
return none;
}
template <class Parent>
bool handle_read_event(Parent&) {
CAF_LOG_TRACE(CAF_ARG(handle_.id));
auto ret = read(handle_, make_span(read_buf_));
if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) {
auto num_bytes = res->first;
auto ep = res->second;
read_buf_.resize(num_bytes);
dispatcher_.handle_data(*this, make_span(read_buf_), std::move(ep));
prepare_next_read();
} else {
auto err = get<sec>(ret);
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
dispatcher_.handle_error(err);
return false;
}
return true;
}
template <class Parent>
bool handle_write_event(Parent& parent) {
CAF_LOG_TRACE(CAF_ARG(handle_.id)
<< CAF_ARG2("queue-size", packet_queue_.size()));
// Try to write leftover data.
write_some();
// Get new data from parent.
for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) {
dispatcher_.write_message(*this, std::move(msg));
}
// Write prepared data.
return write_some();
}
template <class Parent>
void resolve(Parent&, const uri& locator, const actor& listener) {
dispatcher_.resolve(*this, locator, listener);
}
template <class Parent>
void new_proxy(Parent&, const node_id& peer, actor_id id) {
dispatcher_.new_proxy(*this, peer, id);
}
template <class Parent>
void local_actor_down(Parent&, const node_id& peer, actor_id id,
error reason) {
dispatcher_.local_actor_down(*this, peer, id, std::move(reason));
}
template <class Parent>
void timeout(Parent&, atom_value value, uint64_t id) {
dispatcher_.timeout(*this, value, id);
}
void set_timeout(uint64_t timeout_id, id_type id) {
dispatcher_.set_timeout(timeout_id, id);
}
void handle_error(sec code) {
dispatcher_.handle_error(code);
}
error add_new_worker(node_id node, id_type id) {
auto worker = dispatcher_.add_new_worker(*this, node, id);
if (!worker)
return worker.error();
return none;
}
void prepare_next_read() {
read_buf_.clear();
read_buf_.resize(max_datagram_size);
}
void configure_read(receive_policy::config) {
// nop
}
void write_packet(id_type id, span<buffer_type*> buffers) {
CAF_ASSERT(!buffers.empty());
if (packet_queue_.empty())
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);
}
// -- buffer management ------------------------------------------------------
buffer_type next_header_buffer() {
return next_buffer_impl(header_bufs_);
}
buffer_type next_payload_buffer() {
return next_buffer_impl(payload_bufs_);
}
/// Helper struct for managing outgoing packets
struct packet {
id_type id;
buffer_cache_type bytes;
size_t size;
packet(id_type id, span<buffer_type*> bufs) : id(id) {
size = 0;
for (auto buf : bufs) {
size += buf->size();
bytes.emplace_back(std::move(*buf));
}
}
};
private:
// -- utility functions ------------------------------------------------------
static buffer_type next_buffer_impl(buffer_cache_type cache) {
if (cache.empty()) {
return {};
}
auto buf = std::move(cache.back());
cache.pop_back();
return buf;
}
bool write_some() {
CAF_LOG_TRACE(CAF_ARG(handle_.id));
// 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 (header_bufs_.size() < header_bufs_.capacity()) {
it->clear();
header_bufs_.emplace_back(std::move(*it++));
}
for (;
it != bufs.end() && payload_bufs_.size() < payload_bufs_.capacity();
++it) {
it->clear();
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();
std::vector<std::vector<byte>*> ptrs;
for (auto& buf : packet.bytes)
ptrs.emplace_back(&buf);
auto write_ret = write(handle_, make_span(ptrs), packet.id);
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(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_DEBUG("send failed" << CAF_ARG(err));
dispatcher_.handle_error(err);
return false;
}
return true;
}
}
return false;
}
dispatcher_type dispatcher_;
udp_datagram_socket handle_;
buffer_cache_type header_bufs_;
buffer_cache_type payload_bufs_;
std::vector<byte> read_buf_;
std::deque<packet> packet_queue_;
endpoint_manager* manager_;
};
} // namespace net
} // namespace caf
...@@ -53,10 +53,6 @@ public: ...@@ -53,10 +53,6 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
application_type& application() {
return transport_.application();
}
transport_type& transport() { transport_type& transport() {
return transport_; return transport_;
} }
......
...@@ -30,7 +30,7 @@ namespace net { ...@@ -30,7 +30,7 @@ namespace net {
template <class Application, class IdType = unit_t> template <class Application, class IdType = unit_t>
class transport_worker; class transport_worker;
template <class Application, class IdType = unit_t> template <class Transport, class IdType = unit_t>
class transport_worker_dispatcher; class transport_worker_dispatcher;
// -- classes ------------------------------------------------------------------ // -- classes ------------------------------------------------------------------
......
...@@ -248,8 +248,6 @@ private: ...@@ -248,8 +248,6 @@ private:
bool write_some() { bool write_some() {
CAF_LOG_TRACE(CAF_ARG(handle_.id)); CAF_LOG_TRACE(CAF_ARG(handle_.id));
if (write_queue_.empty())
return false;
// Helper function to sort empty buffers back into the right caches. // Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() { auto recycle = [&]() {
auto& front = write_queue_.front(); auto& front = write_queue_.front();
...@@ -266,7 +264,7 @@ private: ...@@ -266,7 +264,7 @@ private:
write_queue_.pop_front(); write_queue_.pop_front();
}; };
// Write buffers from the write_queue_ for as long as possible. // Write buffers from the write_queue_ for as long as possible.
do { while (!write_queue_.empty()) {
auto& buf = write_queue_.front().second; auto& buf = write_queue_.front().second;
CAF_ASSERT(!buf.empty()); CAF_ASSERT(!buf.empty());
auto data = buf.data() + written_; auto data = buf.data() + written_;
...@@ -290,7 +288,7 @@ private: ...@@ -290,7 +288,7 @@ private:
} }
return true; return true;
} }
} while (!write_queue_.empty()); }
return false; return false;
} }
......
...@@ -18,6 +18,8 @@ ...@@ -18,6 +18,8 @@
#pragma once #pragma once
#include <caf/logger.hpp>
#include <caf/sec.hpp>
#include <unordered_map> #include <unordered_map>
#include "caf/byte.hpp" #include "caf/byte.hpp"
...@@ -26,23 +28,26 @@ ...@@ -26,23 +28,26 @@
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_decorator.hpp" #include "caf/net/packet_writer_decorator.hpp"
#include "caf/net/transport_worker.hpp" #include "caf/net/transport_worker.hpp"
#include "caf/send.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
#include "caf/unit.hpp" #include "caf/unit.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
/// implements a dispatcher that dispatches between transport and workers. /// Implements a dispatcher that dispatches between transport and workers.
template <class ApplicationFactory, class IdType> template <class Transport, class IdType>
class transport_worker_dispatcher { class transport_worker_dispatcher {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using id_type = IdType; using id_type = IdType;
using factory_type = ApplicationFactory; using transport_type = Transport;
using application_type = typename ApplicationFactory::application_type; using factory_type = typename transport_type::factory_type;
using application_type = typename factory_type::application_type;
using worker_type = transport_worker<application_type, id_type>; using worker_type = transport_worker<application_type, id_type>;
...@@ -50,8 +55,8 @@ public: ...@@ -50,8 +55,8 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
transport_worker_dispatcher(factory_type factory) transport_worker_dispatcher(transport_type& transport, factory_type factory)
: factory_(std::move(factory)) { : factory_(std::move(factory)), transport_(&transport) {
// nop // nop
} }
...@@ -65,48 +70,52 @@ public: ...@@ -65,48 +70,52 @@ public:
template <class Parent> template <class Parent>
error handle_data(Parent& parent, span<byte> data, id_type id) { error handle_data(Parent& parent, span<byte> data, id_type id) {
auto it = workers_by_id_.find(id); if (auto worker = find_worker(id))
if (it == workers_by_id_.end()) {
// TODO: where to get node_id from here?
add_new_worker(parent, node_id{}, id);
it = workers_by_id_.find(id);
}
auto worker = it->second;
return worker->handle_data(parent, data); return worker->handle_data(parent, data);
// TODO: Where to get node_id from here?
auto worker = add_new_worker(parent, node_id{}, id);
if (worker)
return (*worker)->handle_data(parent, data);
else
return std::move(worker.error());
} }
template <class Parent> template <class Parent>
void write_message(Parent& parent, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) { std::unique_ptr<endpoint_manager_queue::message> msg) {
auto sender = msg->msg->sender; auto receiver = msg->receiver;
if (!sender) if (!receiver)
return; return;
auto nid = sender->node(); auto nid = receiver->node();
auto it = workers_by_node_.find(nid); if (auto worker = find_worker(nid)) {
if (it == workers_by_node_.end()) { worker->write_message(parent, std::move(msg));
return;
}
// TODO: where to get id_type from here? // TODO: where to get id_type from here?
add_new_worker(parent, nid, id_type{}); if (auto worker = add_new_worker(parent, nid, id_type{}))
it = workers_by_node_.find(nid); (*worker)->write_message(parent, std::move(msg));
} }
auto worker = it->second;
worker->write_message(parent, std::move(msg)); template <class Parent>
void resolve(Parent& parent, const uri& locator, const actor& listener) {
if (auto worker = find_worker(make_node_id(locator)))
worker->resolve(parent, locator.path(), listener);
else
anon_send(listener,
make_error(sec::runtime_error, "could not resolve node"));
} }
template <class Parent> template <class Parent>
void write_packet(Parent& parent, span<const byte> header, void new_proxy(Parent& parent, const node_id& nid, actor_id id) {
span<const byte> payload, id_type id) { if (auto worker = find_worker(nid))
parent.write_packet(header, payload, id); worker->new_proxy(parent, nid, id);
} }
template <class Parent> template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) { void local_actor_down(Parent& parent, const node_id& nid, actor_id id,
// TODO path should be uri to lookup the corresponding worker error reason) {
// if enpoint is known -> resolve actor through worker if (auto worker = find_worker(nid))
// if not connect to endpoint?! worker->local_actor_down(parent, nid, id, std::move(reason));
if (workers_by_id_.empty())
return;
auto worker = workers_by_id_.begin()->second;
worker->resolve(parent, path, listener);
} }
template <class... Ts> template <class... Ts>
...@@ -116,10 +125,11 @@ public: ...@@ -116,10 +125,11 @@ public:
template <class Parent> template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) { void timeout(Parent& parent, atom_value value, uint64_t id) {
auto worker = workers_by_timeout_id_.at(id); if (auto worker = workers_by_timeout_id_.at(id)) {
worker->timeout(parent, value, id); worker->timeout(parent, value, id);
workers_by_timeout_id_.erase(id); workers_by_timeout_id_.erase(id);
} }
}
void handle_error(sec error) { void handle_error(sec error) {
for (const auto& p : workers_by_id_) { for (const auto& p : workers_by_id_) {
...@@ -129,17 +139,36 @@ public: ...@@ -129,17 +139,36 @@ public:
} }
template <class Parent> template <class Parent>
error add_new_worker(Parent& parent, node_id node, id_type id) { expected<worker_ptr> add_new_worker(Parent& parent, node_id node,
id_type id) {
auto application = factory_.make(); auto application = factory_.make();
auto worker = std::make_shared<worker_type>(std::move(application), id); auto worker = std::make_shared<worker_type>(std::move(application), id);
if (auto err = worker->init(parent)) if (auto err = worker->init(parent))
return err; return err;
workers_by_id_.emplace(std::move(id), worker); workers_by_id_.emplace(std::move(id), worker);
workers_by_node_.emplace(std::move(node), std::move(worker)); workers_by_node_.emplace(std::move(node), worker);
return none; return worker;
} }
private: private:
worker_ptr find_worker(const node_id& nid) {
return find_worker_impl(workers_by_node_, nid);
}
worker_ptr find_worker(const id_type& id) {
return find_worker_impl(workers_by_id_, id);
}
template <class Key>
worker_ptr find_worker_impl(const std::unordered_map<Key, worker_ptr>& map,
const Key& key) {
if (map.count(key) == 0) {
CAF_LOG_DEBUG("could not find worker: " << CAF_ARG(key));
return nullptr;
}
return map.at(key);
}
// -- worker lookups --------------------------------------------------------- // -- worker lookups ---------------------------------------------------------
std::unordered_map<id_type, worker_ptr> workers_by_id_; std::unordered_map<id_type, worker_ptr> workers_by_id_;
...@@ -147,6 +176,7 @@ private: ...@@ -147,6 +176,7 @@ private:
std::unordered_map<uint64_t, worker_ptr> workers_by_timeout_id_; std::unordered_map<uint64_t, worker_ptr> workers_by_timeout_id_;
factory_type factory_; factory_type factory_;
transport_type* transport_;
}; };
} // namespace net } // namespace net
......
...@@ -56,9 +56,22 @@ error allow_connreset(udp_datagram_socket x, bool new_value); ...@@ -56,9 +56,22 @@ error allow_connreset(udp_datagram_socket x, bool new_value);
variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x, variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
span<byte> buf); span<byte> 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 enpoint 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`
variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
ip_endpoint ep);
/// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`. /// Sends the content of `buf` as a datagram to the endpoint `ep` on socket `x`.
/// @param x The UDP socket for sending datagrams. /// @param x The UDP socket for sending datagrams.
/// @param buf The buffer to send. /// @param buf The buffer to send.
/// @param ep The enpoint to send the datagram to.
/// @returns The number of written bytes on success, otherwise an error code. /// @returns The number of written bytes on success, otherwise an error code.
/// @relates udp_datagram_socket /// @relates udp_datagram_socket
variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf, variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
......
...@@ -58,7 +58,7 @@ error allow_connreset(udp_datagram_socket x, bool) { ...@@ -58,7 +58,7 @@ error allow_connreset(udp_datagram_socket x, bool) {
expected<std::pair<udp_datagram_socket, uint16_t>> expected<std::pair<udp_datagram_socket, uint16_t>>
make_udp_datagram_socket(ip_endpoint ep, bool reuse_addr) { make_udp_datagram_socket(ip_endpoint ep, bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(ep)); CAF_LOG_TRACE(CAF_ARG(ep));
sockaddr_storage addr; sockaddr_storage addr = {};
detail::convert(ep, addr); detail::convert(ep, addr);
CAF_NET_SYSCALL("socket", fd, ==, invalid_socket_id, CAF_NET_SYSCALL("socket", fd, ==, invalid_socket_id,
::socket(addr.ss_family, SOCK_DGRAM, 0)); ::socket(addr.ss_family, SOCK_DGRAM, 0));
...@@ -112,12 +112,12 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x, ...@@ -112,12 +112,12 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf, variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
ip_endpoint ep) { ip_endpoint ep) {
sockaddr_storage addr; sockaddr_storage addr = {};
detail::convert(ep, addr); detail::convert(ep, addr);
auto len = ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6);
auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()), auto res = ::sendto(x.id, reinterpret_cast<socket_send_ptr>(buf.data()),
buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), buf.size(), 0, reinterpret_cast<sockaddr*>(&addr), len);
ep.address().embeds_v4() ? sizeof(sockaddr_in)
: sizeof(sockaddr_in6));
auto ret = check_udp_datagram_socket_io_res(res); auto ret = check_udp_datagram_socket_io_res(res);
if (auto num_bytes = get_if<size_t>(&ret)) if (auto num_bytes = get_if<size_t>(&ret))
return *num_bytes; return *num_bytes;
...@@ -125,6 +125,60 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf, ...@@ -125,6 +125,60 @@ variant<size_t, sec> write(udp_datagram_socket x, span<const byte> buf,
return get<sec>(ret); return get<sec>(ret);
} }
#ifdef CAF_WINDOWS
variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
ip_endpoint ep) {
CAF_ASSERT(bufs.size() < 10);
WSABUF buf_array[10];
auto convert = [](std::vector<byte>* buf) {
return WSABUF{static_cast<ULONG>(buf->size()),
reinterpret_cast<CHAR*>(buf->data())};
};
std::transform(bufs.begin(), bufs.end(), std::begin(buf_array), convert);
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
variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
ip_endpoint ep) {
CAF_ASSERT(bufs.size() < 10);
auto convert = [](std::vector<byte>* 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 = bufs.size();
auto res = sendmsg(x.id, &message, 0);
return check_udp_datagram_socket_io_res(res);
}
#endif // CAF_WINDOWS
variant<size_t, sec> variant<size_t, sec>
check_udp_datagram_socket_io_res(std::make_signed<size_t>::type res) { check_udp_datagram_socket_io_res(std::make_signed<size_t>::type res) {
if (res < 0) { if (res < 0) {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#define CAF_SUITE datagram_transport
#include "caf/net/datagram_transport.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "caf/byte.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_impl.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
using namespace caf;
using namespace caf::net;
using namespace caf::net::ip;
namespace {
constexpr string_view hello_manager = "hello manager!";
class dummy_application_factory;
struct fixture : test_coordinator_fixture<>, host_fixture {
using buffer_type = std::vector<byte>;
using buffer_ptr = std::shared_ptr<buffer_type>;
fixture() : shared_buf(std::make_shared<buffer_type>(1024)) {
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
mpx->set_thread_id();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
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));
recv_socket = receive_pair.first;
ep.port(htons(receive_pair.second));
CAF_MESSAGE("sending message to " << CAF_ARG(ep));
if (auto err = nonblocking(recv_socket, true))
CAF_FAIL("nonblocking() returned an error: " << err);
}
~fixture() {
close(send_socket);
close(recv_socket);
}
bool handle_io_event() override {
return mpx->poll_once(false);
}
error read_from_socket(udp_datagram_socket sock, buffer_type& buf) {
uint8_t receive_attempts = 0;
variant<std::pair<size_t, ip_endpoint>, sec> read_ret;
do {
read_ret = read(sock, make_span(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");
}
if (++receive_attempts > 100)
return make_error(sec::runtime_error,
"too many unavailable_or_would_blocks");
} while (read_ret.index() != 0);
return none;
}
multiplexer_ptr mpx;
buffer_ptr shared_buf;
ip_endpoint ep;
udp_datagram_socket send_socket;
udp_datagram_socket recv_socket;
};
class dummy_application {
using buffer_type = std::vector<byte>;
using buffer_ptr = std::shared_ptr<buffer_type>;
public:
explicit dummy_application(buffer_ptr rec_buf)
: rec_buf_(std::move(rec_buf)){
// nop
};
template <class Parent>
error init(Parent&) {
return none;
}
template <class Transport>
void write_message(Transport& transport,
std::unique_ptr<endpoint_manager_queue::message> msg) {
transport.write_packet(msg->payload);
}
template <class Parent>
error handle_data(Parent&, span<const byte> data) {
rec_buf_->clear();
rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end());
return none;
}
template <class Parent>
void resolve(Parent& parent, string_view path, const actor& listener) {
actor_id aid = 42;
auto uri = unbox(make_uri("test:/id/42"));
auto nid = make_node_id(uri);
actor_config cfg;
endpoint_manager_ptr ptr{&parent.manager()};
auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid,
&parent.system(),
cfg,
std::move(ptr));
anon_send(listener, resolve_atom::value,
std::string{path.begin(), path.end()}, p);
}
template <class Parent>
void new_proxy(Parent&, actor_id) {
// nop
}
template <class Parent>
void local_actor_down(Parent&, actor_id, error) {
// nop
}
template <class Parent>
void timeout(Parent&, atom_value, uint64_t) {
// nop
}
void handle_error(sec sec) {
CAF_FAIL("handle_error called: " << to_string(sec));
}
static expected<buffer_type> serialize(actor_system& sys,
const type_erased_tuple& x) {
buffer_type result;
serializer_impl<buffer_type> sink{sys, result};
if (auto err = message::save(sink, x))
return err;
return result;
}
private:
buffer_ptr rec_buf_;
};
class dummy_application_factory {
using buffer_ptr = std::shared_ptr<std::vector<byte>>;
public:
using application_type = dummy_application;
explicit dummy_application_factory(buffer_ptr buf) : buf_(std::move(buf)) {
// nop
}
dummy_application make() {
return dummy_application{buf_};
}
private:
buffer_ptr buf_;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(datagram_transport_tests, fixture)
CAF_TEST(receive) {
using transport_type = datagram_transport<dummy_application_factory>;
if (auto err = nonblocking(recv_socket, true))
CAF_FAIL("nonblocking() returned an error: " << sys.render(err));
auto mgr = make_endpoint_manager(mpx, sys,
transport_type{recv_socket,
dummy_application_factory{
shared_buf}});
CAF_CHECK_EQUAL(mgr->init(), none);
auto mgr_impl = mgr.downcast<endpoint_manager_impl<transport_type>>();
CAF_CHECK(mgr_impl != nullptr);
auto& transport = mgr_impl->transport();
transport.configure_read(net::receive_policy::exactly(hello_manager.size()));
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_CHECK_EQUAL(write(send_socket, as_bytes(make_span(hello_manager)), ep),
hello_manager.size());
CAF_MESSAGE("wrote " << hello_manager.size() << " bytes.");
run();
CAF_CHECK_EQUAL(string_view(reinterpret_cast<char*>(shared_buf->data()),
shared_buf->size()),
hello_manager);
}
CAF_TEST(resolve and proxy communication) {
using transport_type = datagram_transport<dummy_application_factory>;
buffer_type recv_buf(1024);
auto uri = unbox(make_uri("test:/id/42"));
auto mgr = make_endpoint_manager(mpx, sys,
transport_type{send_socket,
dummy_application_factory{
shared_buf}});
CAF_CHECK_EQUAL(mgr->init(), none);
auto mgr_impl = mgr.downcast<endpoint_manager_impl<transport_type>>();
CAF_CHECK(mgr_impl != nullptr);
auto& transport = mgr_impl->transport();
transport.add_new_worker(make_node_id(uri), ep);
run();
mgr->resolve(uri, self);
run();
self->receive(
[&](resolve_atom, const std::string&, const strong_actor_ptr& p) {
CAF_MESSAGE("got a proxy, send a message to it");
self->send(actor_cast<actor>(p), "hello proxy!");
},
after(std::chrono::seconds(0)) >>
[&] { CAF_FAIL("manager did not respond with a proxy."); });
run();
CAF_CHECK_EQUAL(read_from_socket(recv_socket, recv_buf), none);
CAF_MESSAGE("receive buffer contains " << recv_buf.size() << " bytes");
message msg;
binary_deserializer source{sys, recv_buf};
CAF_CHECK_EQUAL(source(msg), none);
if (msg.match_elements<std::string>())
CAF_CHECK_EQUAL(msg.get_as<std::string>(0), "hello proxy!");
else
CAF_ERROR("expected a string, got: " << to_string(msg));
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -70,7 +70,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> { ...@@ -70,7 +70,7 @@ struct fixture : host_fixture, test_coordinator_fixture<config> {
auto backend = dynamic_cast<backend::test*>(mm.backend("test")); auto backend = dynamic_cast<backend::test*>(mm.backend("test"));
auto mgr = backend->peer(mars); auto mgr = backend->peer(mars);
auto& dref = dynamic_cast<endpoint_manager_impl<transport_type>&>(*mgr); auto& dref = dynamic_cast<endpoint_manager_impl<transport_type>&>(*mgr);
app = &dref.application(); app = &dref.transport().application();
sock = backend->socket(mars); sock = backend->socket(mars);
} }
......
...@@ -122,6 +122,8 @@ private: ...@@ -122,6 +122,8 @@ private:
struct dummy_transport { struct dummy_transport {
using transport_type = dummy_transport; using transport_type = dummy_transport;
using factory_type = dummy_application_factory;
using application_type = dummy_application; using application_type = dummy_application;
dummy_transport(std::shared_ptr<buffer_type> buf) : buf_(std::move(buf)) { dummy_transport(std::shared_ptr<buffer_type> buf) : buf_(std::move(buf)) {
...@@ -180,12 +182,12 @@ uri operator"" _u(const char* cstr, size_t cstr_len) { ...@@ -180,12 +182,12 @@ uri operator"" _u(const char* cstr, size_t cstr_len) {
} }
struct fixture : host_fixture { struct fixture : host_fixture {
using dispatcher_type = transport_worker_dispatcher<dummy_application_factory, using dispatcher_type = transport_worker_dispatcher<dummy_transport,
ip_endpoint>; ip_endpoint>;
fixture() fixture()
: buf{std::make_shared<buffer_type>()}, : buf{std::make_shared<buffer_type>()},
dispatcher{dummy_application_factory{buf}}, dispatcher{dummy, dummy_application_factory{buf}},
dummy{buf} { dummy{buf} {
add_new_workers(); add_new_workers();
} }
...@@ -197,13 +199,14 @@ struct fixture : host_fixture { ...@@ -197,13 +199,14 @@ struct fixture : host_fixture {
auto p = make_actor<dummy_actor, strong_actor_ptr>(aid, nid, &sys, cfg); auto p = make_actor<dummy_actor, strong_actor_ptr>(aid, nid, &sys, cfg);
auto test_span = as_bytes(make_span(hello_test)); auto test_span = as_bytes(make_span(hello_test));
buffer_type payload(test_span.begin(), test_span.end()); buffer_type payload(test_span.begin(), test_span.end());
auto strong_actor = actor_cast<strong_actor_ptr>(p); auto receiver = actor_cast<strong_actor_ptr>(p);
if (!receiver)
CAF_FAIL("failed to cast receiver to a strong_actor_ptr");
mailbox_element::forwarding_stack stack; mailbox_element::forwarding_stack stack;
auto elem = make_mailbox_element(std::move(strong_actor), auto elem = make_mailbox_element(nullptr, make_message_id(12345),
make_message_id(12345), std::move(stack), std::move(stack), make_message());
make_message());
return detail::make_unique<endpoint_manager_queue::message>(std::move(elem), return detail::make_unique<endpoint_manager_queue::message>(std::move(elem),
strong_actor, receiver,
payload); payload);
} }
...@@ -213,16 +216,17 @@ struct fixture : host_fixture { ...@@ -213,16 +216,17 @@ struct fixture : host_fixture {
void add_new_workers() { void add_new_workers() {
for (auto& data : test_data) { for (auto& data : test_data) {
if (auto err = dispatcher.add_new_worker(dummy, data.nid, data.ep)) auto worker = dispatcher.add_new_worker(dummy, data.nid, data.ep);
CAF_FAIL("add_new_worker returned an error: " << err); if (!worker)
CAF_FAIL("add_new_worker returned an error: " << worker.error());
} }
buf->clear(); buf->clear();
} }
void test_write_message(testdata& testcase) { void test_write_message(testdata& testcase) {
auto msg = make_dummy_message(testcase.nid); auto msg = make_dummy_message(testcase.nid);
if (!msg->msg->sender) if (!msg->receiver)
CAF_FAIL("sender is null"); CAF_FAIL("receiver is null");
dispatcher.write_message(dummy, std::move(msg)); dispatcher.write_message(dummy, std::move(msg));
} }
...@@ -268,9 +272,8 @@ struct fixture : host_fixture { ...@@ -268,9 +272,8 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE(transport_worker_dispatcher_test, fixture) CAF_TEST_FIXTURE_SCOPE(transport_worker_dispatcher_test, fixture)
CAF_TEST(init) { CAF_TEST(init) {
dispatcher_type dispatcher{dummy_application_factory{buf}}; dispatcher_type dispatcher{dummy, dummy_application_factory{buf}};
if (auto err = dispatcher.init(dummy)) CAF_CHECK_EQUAL(dispatcher.init(dummy), none);
CAF_FAIL("init failed with error: " << err);
} }
CAF_TEST(handle_data) { CAF_TEST(handle_data) {
......
...@@ -39,7 +39,7 @@ namespace { ...@@ -39,7 +39,7 @@ namespace {
constexpr string_view hello_test = "Hello test!"; constexpr string_view hello_test = "Hello test!";
struct fixture : host_fixture { struct fixture : host_fixture {
fixture() : host_fixture() { fixture() : host_fixture(), buf(1024) {
addresses = local_addresses("localhost"); addresses = local_addresses("localhost");
CAF_CHECK(!addresses.empty()); CAF_CHECK(!addresses.empty());
ep = ip_endpoint(*addresses.begin(), 0); ep = ip_endpoint(*addresses.begin(), 0);
...@@ -61,14 +61,48 @@ struct fixture : host_fixture { ...@@ -61,14 +61,48 @@ struct fixture : host_fixture {
ip_endpoint ep; ip_endpoint ep;
udp_datagram_socket send_socket; udp_datagram_socket send_socket;
udp_datagram_socket receive_socket; udp_datagram_socket receive_socket;
std::vector<byte> buf;
};
error read_from_socket(udp_datagram_socket sock, std::vector<byte>& buf) {
uint8_t receive_attempts = 0;
variant<std::pair<size_t, ip_endpoint>, sec> read_ret;
do {
read_ret = read(sock, make_span(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");
}
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 typename Inspector::result_type inspect(Inspector& f, header& x) {
return f(meta::type_name("header"), x.payload_size);
}
size_t payload_size;
}; };
} // namespace } // namespace
CAF_TEST_FIXTURE_SCOPE(udp_datagram_socket_test, fixture) CAF_TEST_FIXTURE_SCOPE(udp_datagram_socket_test, fixture)
CAF_TEST(send and receive) { CAF_TEST(read / write using span<byte>) {
std::vector<byte> buf(1024);
if (auto err = nonblocking(socket_cast<net::socket>(receive_socket), true)) if (auto err = nonblocking(socket_cast<net::socket>(receive_socket), true))
CAF_FAIL("setting socket to nonblocking failed: " << err); CAF_FAIL("setting socket to nonblocking failed: " << err);
CAF_CHECK_EQUAL(read(receive_socket, make_span(buf)), CAF_CHECK_EQUAL(read(receive_socket, make_span(buf)),
...@@ -76,21 +110,35 @@ CAF_TEST(send and receive) { ...@@ -76,21 +110,35 @@ CAF_TEST(send and receive) {
CAF_MESSAGE("sending data to " << to_string(ep)); CAF_MESSAGE("sending data to " << to_string(ep));
CAF_CHECK_EQUAL(write(send_socket, as_bytes(make_span(hello_test)), ep), CAF_CHECK_EQUAL(write(send_socket, as_bytes(make_span(hello_test)), ep),
hello_test.size()); hello_test.size());
int receive_attempts = 0; CAF_CHECK_EQUAL(read_from_socket(receive_socket, buf), none);
variant<std::pair<size_t, ip_endpoint>, sec> read_ret;
do {
read_ret = read(receive_socket, make_span(buf));
if (auto read_res = get_if<std::pair<size_t, ip_endpoint>>(&read_ret)) {
CAF_CHECK_EQUAL(read_res->first, hello_test.size());
buf.resize(read_res->first);
} else if (get<sec>(read_ret) != sec::unavailable_or_would_block) {
CAF_FAIL("read returned an error: " << sys.render(get<sec>(read_ret)));
}
if (++receive_attempts > 100)
CAF_FAIL("unable to receive data from the UDP socket");
} while (read_ret.index() != 0);
string_view received{reinterpret_cast<const char*>(buf.data()), buf.size()}; string_view received{reinterpret_cast<const char*>(buf.data()), buf.size()};
CAF_CHECK_EQUAL(received, hello_test); CAF_CHECK_EQUAL(received, hello_test);
} }
CAF_TEST(read / write using span<std::vector<byte>*>) {
// generate header and payload in separate buffers
header hdr{hello_test.size()};
std::vector<byte> hdr_buf;
serializer_impl<std::vector<byte>> sink(sys, hdr_buf);
if (auto err = sink(hdr))
CAF_FAIL("serializing payload failed" << sys.render(err));
auto bytes = as_bytes(make_span(hello_test));
std::vector<byte> payload_buf(bytes.begin(), bytes.end());
auto packet_size = hdr_buf.size() + payload_buf.size();
std::vector<std::vector<byte>*> bufs{&hdr_buf, &payload_buf};
CAF_CHECK_EQUAL(write(send_socket, make_span(bufs), ep), 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;
source(recv_hdr);
CAF_CHECK_EQUAL(hdr.payload_size, recv_hdr.payload_size);
string_view received{reinterpret_cast<const char*>(buf.data())
+ sizeof(header),
buf.size() - sizeof(header)};
CAF_CHECK_EQUAL(received, hello_test);
}
CAF_TEST_FIXTURE_SCOPE_END() 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