Commit 1c9be4bf authored by Jakob Otto's avatar Jakob Otto

Update branch to new API

parent d7208d90
...@@ -28,6 +28,7 @@ ...@@ -28,6 +28,7 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp" #include "caf/ip_endpoint.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp" #include "caf/net/receive_policy.hpp"
...@@ -46,14 +47,18 @@ class datagram_transport { ...@@ -46,14 +47,18 @@ class datagram_transport {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using buffer_type = std::vector<byte>;
using buffer_cache_type = std::vector<buffer_type>;
using factory_type = Factory; using factory_type = Factory;
using transport_type = datagram_transport; using transport_type = datagram_transport;
using application_type = typename Factory::application_type; using application_type = typename Factory::application_type;
using dispatcher_type = transport_worker_dispatcher<factory_type, using dispatcher_type = transport_worker_dispatcher<
ip_endpoint>; datagram_transport, factory_type, ip_endpoint>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
...@@ -62,16 +67,46 @@ public: ...@@ -62,16 +67,46 @@ public:
handle_(handle), handle_(handle),
max_consecutive_reads_(0), max_consecutive_reads_(0),
read_threshold_(1024), read_threshold_(1024),
collected_(0),
max_(1024), max_(1024),
rd_flag_(receive_policy_flag::exactly) { rd_flag_(receive_policy_flag::exactly),
manager_(nullptr) {
// nop // nop
} }
// -- properties -------------------------------------------------------------
udp_datagram_socket handle() const noexcept {
return handle_;
}
actor_system& system() {
return manager().system();
}
application_type& application() {
// TODO: This wont work. We need information on which application is wanted
return dispatcher_.application();
}
transport_type& transport() {
return *this;
}
endpoint_manager& manager() {
return *manager_;
}
// -- public member functions ------------------------------------------------ // -- public member functions ------------------------------------------------
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& 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(parent)) if (auto err = dispatcher_.init(parent))
return err; return err;
parent.mask_add(operation::read); parent.mask_add(operation::read);
...@@ -85,7 +120,7 @@ public: ...@@ -85,7 +120,7 @@ public:
if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) { if (auto res = get_if<std::pair<size_t, ip_endpoint>>(&ret)) {
auto num_bytes = res->first; auto num_bytes = res->first;
auto ep = res->second; auto ep = res->second;
collected_ += (num_bytes > 0) ? static_cast<size_t>(num_bytes) : 0; read_buf_.resize(num_bytes);
dispatcher_.handle_data(parent, make_span(read_buf_), std::move(ep)); dispatcher_.handle_data(parent, make_span(read_buf_), std::move(ep));
prepare_next_read(); prepare_next_read();
} else { } else {
...@@ -113,18 +148,27 @@ public: ...@@ -113,18 +148,27 @@ public:
} }
template <class Parent> template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) { void resolve(Parent&, const uri& locator, const actor& listener) {
dispatcher_.resolve(parent, path, 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> template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) { void local_actor_down(Parent&, const node_id& peer, actor_id id,
auto decorator = make_write_packet_decorator(*this, parent); error reason) {
dispatcher_.timeout(decorator, value, id); dispatcher_.local_actor_down(*this, peer, id, std::move(reason));
} }
template <class Parent> template <class Parent>
uint64_t set_timeout(uint64_t timeout_id, ip_endpoint ep) { void timeout(Parent&, atom_value value, uint64_t id) {
dispatcher_.timeout(*this, value, id);
}
void set_timeout(uint64_t timeout_id, ip_endpoint ep) {
dispatcher_.set_timeout(timeout_id, ep); dispatcher_.set_timeout(timeout_id, ep);
} }
...@@ -132,13 +176,8 @@ public: ...@@ -132,13 +176,8 @@ public:
dispatcher_.handle_error(code); dispatcher_.handle_error(code);
} }
udp_datagram_socket handle() const noexcept {
return handle_;
}
void prepare_next_read() { void prepare_next_read() {
read_buf_.clear(); read_buf_.clear();
collected_ = 0;
// This cast does nothing, but prevents a weird compiler error on GCC // This cast does nothing, but prevents a weird compiler error on GCC
// <= 4.9. // <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9. // TODO: remove cast when dropping support for GCC 4.9.
...@@ -170,54 +209,84 @@ public: ...@@ -170,54 +209,84 @@ public:
prepare_next_read(); prepare_next_read();
} }
void write_packet(span<const byte> header, span<const byte> payload, void write_packet(ip_endpoint ep, span<buffer_type*> buffers) {
typename dispatcher_type::id_type id) { CAF_ASSERT(!buffers.empty());
std::vector<byte> buf; if (packet_queue_.empty())
buf.reserve(header.size() + payload.size()); manager().register_writing();
buf.insert(buf.end(), header.begin(), header.end()); // By convention, the first buffer is a header buffer. Every other buffer is
buf.insert(buf.end(), payload.begin(), payload.end()); // a payload buffer.
packet_queue_.emplace_back(id, std::move(buf)); packet_queue_.emplace_back(ep, 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 { struct packet {
ip_endpoint destination; ip_endpoint destination;
std::vector<byte> bytes; size_t payload_buf_num;
buffer_cache_type bytes;
packet(ip_endpoint destination, std::vector<byte> bytes) packet(ip_endpoint destination, span<buffer_type*> bufs)
: destination(destination), bytes(std::move(bytes)) { : destination(destination) {
// nop payload_buf_num = bufs.size() - 1;
for (auto buf : bufs)
bytes.emplace_back(false, std::move(*buf));
} }
}; };
private: 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() { bool write_some() {
if (packet_queue_.empty()) while (!packet_queue_.empty()) {
return false; auto& next_packet = packet_queue_.front();
auto& next_packet = packet_queue_.front(); auto send_res = write(handle_, next_packet.bytes,
auto send_res = write(handle_, make_span(next_packet.bytes), next_packet.destination);
next_packet.destination); if (auto num_bytes = get_if<size_t>(&send_res)) {
if (auto num_bytes = get_if<size_t>(&send_res)) { CAF_LOG_DEBUG(CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
CAF_LOG_DEBUG(CAF_ARG(handle_.id) << CAF_ARG(*num_bytes)); packet_queue_.pop_front();
packet_queue_.pop_front(); return true;
return true; }
auto err = get<sec>(send_res);
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
dispatcher_.handle_error(err);
} }
auto err = get<sec>(send_res);
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
dispatcher_.handle_error(err);
return false; return false;
} }
dispatcher_type dispatcher_; dispatcher_type dispatcher_;
udp_datagram_socket handle_; udp_datagram_socket handle_;
buffer_cache_type header_bufs_;
buffer_cache_type payload_bufs_;
std::vector<byte> read_buf_; std::vector<byte> read_buf_;
std::deque<packet> packet_queue_; std::deque<packet> packet_queue_;
size_t max_consecutive_reads_; size_t max_consecutive_reads_;
size_t read_threshold_; size_t read_threshold_;
size_t collected_; // size_t collected_;
size_t max_; size_t max_;
receive_policy_flag rd_flag_; receive_policy_flag rd_flag_;
endpoint_manager* manager_;
}; };
} // namespace net } // namespace net
......
...@@ -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 Application, class IdType = unit_t>
class transport_worker_dispatcher; class transport_worker_dispatcher;
// -- classes ------------------------------------------------------------------ // -- classes ------------------------------------------------------------------
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#pragma once #pragma once
#include <caf/logger.hpp>
#include <unordered_map> #include <unordered_map>
#include "caf/byte.hpp" #include "caf/byte.hpp"
...@@ -33,7 +34,7 @@ namespace caf { ...@@ -33,7 +34,7 @@ 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 ApplicationFactory, class IdType>
class transport_worker_dispatcher { class transport_worker_dispatcher {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -42,6 +43,8 @@ public: ...@@ -42,6 +43,8 @@ public:
using factory_type = ApplicationFactory; using factory_type = ApplicationFactory;
using transport_type = Transport;
using application_type = typename ApplicationFactory::application_type; using application_type = typename ApplicationFactory::application_type;
using worker_type = transport_worker<application_type, id_type>; using worker_type = transport_worker<application_type, id_type>;
...@@ -50,7 +53,7 @@ public: ...@@ -50,7 +53,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
transport_worker_dispatcher(factory_type factory) explicit transport_worker_dispatcher(factory_type factory)
: factory_(std::move(factory)) { : factory_(std::move(factory)) {
// nop // nop
} }
...@@ -65,42 +68,48 @@ public: ...@@ -65,42 +68,48 @@ 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); auto worker = find_by_id(id);
if (it == workers_by_id_.end()) { if (!worker) {
// TODO: where to get node_id from here? // TODO: where to get id_type from here?
add_new_worker(parent, node_id{}, id); add_new_worker(parent, node_id{}, id);
it = workers_by_id_.find(id); worker = find_by_id(id);
} }
auto worker = it->second;
return worker->handle_data(parent, data); return worker->handle_data(parent, data);
} }
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); auto worker = find_by_node(nid);
if (it == workers_by_node_.end()) { if (!worker) {
// TODO: where to get id_type from here? // TODO: where to get id_type from here?
add_new_worker(parent, nid, id_type{}); add_new_worker(parent, nid, id_type{});
it = workers_by_node_.find(nid); worker = find_by_node(nid);
} }
auto worker = it->second;
worker->write_message(parent, std::move(msg)); worker->write_message(parent, std::move(msg));
} }
template <class Parent> template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) { void resolve(Parent& parent, const uri& locator, const actor& listener) {
// TODO path should be uri to lookup the corresponding worker if (auto worker = find_by_node(make_node_id(locator)))
// if enpoint is known -> resolve actor through worker worker->resolve(parent, locator.path(), listener);
// if not connect to endpoint?! }
if (workers_by_id_.empty())
return; template <class Parent>
auto worker = workers_by_id_.begin()->second; void new_proxy(Parent&, const node_id& nid, actor_id id) {
worker->resolve(parent, path, listener); if (auto worker = find_by_node(nid))
worker->new_proxy(*this, nid, id);
}
template <class Parent>
void local_actor_down(Parent&, const node_id& nid, actor_id id,
error reason) {
if (auto worker = find_by_node(nid))
worker->local_actor_down(*this, nid, id, std::move(reason));
} }
template <class... Ts> template <class... Ts>
...@@ -134,6 +143,28 @@ public: ...@@ -134,6 +143,28 @@ public:
} }
private: private:
worker_ptr find_by_node(const node_id& nid) {
if (workers_by_node_.empty())
return nullptr;
auto it = workers_by_node_.find(nid);
if (it == workers_by_node_.end()) {
CAF_LOG_ERROR("could not find worker by node: " << CAF_ARG(nid));
return nullptr;
}
return it->second;
}
worker_ptr find_by_id(const IdType& id) {
if (workers_by_id_.empty())
return nullptr;
auto it = workers_by_id_.find(id);
if (it == workers_by_id_.end()) {
CAF_LOG_ERROR("could not find worker by node: " << CAF_ARG(id));
return nullptr;
}
return it->second;
}
// -- worker lookups --------------------------------------------------------- // -- worker lookups ---------------------------------------------------------
std::unordered_map<id_type, worker_ptr> workers_by_id_; std::unordered_map<id_type, worker_ptr> workers_by_id_;
......
...@@ -56,9 +56,24 @@ error allow_connreset(udp_datagram_socket x, bool new_value); ...@@ -56,9 +56,24 @@ 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 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
/// @post either the result is a `sec` or a positive (non-zero) integer
/// @pre `bufs.size() < 10`
variant<size_t, sec> write(udp_datagram_socket x,
std::initializer_list<span<const 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,
......
...@@ -20,10 +20,9 @@ ...@@ -20,10 +20,9 @@
#include "caf/net/datagram_transport.hpp" #include "caf/net/datagram_transport.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp" #include "caf/net/actor_proxy_impl.hpp"
...@@ -42,6 +41,8 @@ namespace { ...@@ -42,6 +41,8 @@ namespace {
constexpr string_view hello_manager = "hello manager!"; constexpr string_view hello_manager = "hello manager!";
class dummy_application_factory;
struct fixture : test_coordinator_fixture<>, host_fixture { struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() { fixture() {
mpx = std::make_shared<multiplexer>(); mpx = std::make_shared<multiplexer>();
...@@ -50,7 +51,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -50,7 +51,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
} }
bool handle_io_event() override { bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false); return mpx->poll_once(false);
} }
...@@ -59,6 +59,8 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -59,6 +59,8 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
class dummy_application { class dummy_application {
public: public:
using transport_type = datagram_transport<dummy_application_factory>;
dummy_application(std::shared_ptr<std::vector<byte>> rec_buf) dummy_application(std::shared_ptr<std::vector<byte>> rec_buf)
: rec_buf_(std::move(rec_buf)){ : rec_buf_(std::move(rec_buf)){
// nop // nop
...@@ -73,18 +75,19 @@ public: ...@@ -73,18 +75,19 @@ public:
template <class Transport> template <class Transport>
void write_message(Transport& transport, void write_message(Transport& transport,
std::unique_ptr<endpoint_manager::message> msg) { std::unique_ptr<endpoint_manager_queue::message> msg) {
transport.write_packet(span<byte>{}, msg->payload); transport.write_packet(span<byte>{}, msg->payload);
} }
template <class Parent> template <class Parent>
void handle_data(Parent&, span<const byte> data) { error handle_data(Parent&, span<const byte> data) {
rec_buf_->clear(); rec_buf_->clear();
rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end()); rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end());
return none;
} }
template <class Parent> template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) { void resolve(Parent& parent, string_view path, const actor& listener) {
actor_id aid = 42; actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899"; auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(42, hid)); auto nid = unbox(make_node_id(42, hid));
...@@ -94,7 +97,17 @@ public: ...@@ -94,7 +97,17 @@ public:
&parent.system(), &parent.system(),
cfg, cfg,
std::move(ptr)); std::move(ptr));
anon_send(listener, resolve_atom::value, std::move(path), p); anon_send(listener, resolve_atom::value, path, 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 Transport> template <class Transport>
...@@ -119,11 +132,12 @@ private: ...@@ -119,11 +132,12 @@ private:
std::shared_ptr<std::vector<byte>> rec_buf_; std::shared_ptr<std::vector<byte>> rec_buf_;
}; };
struct dummy_application_factory { class dummy_application_factory {
public:
using application_type = dummy_application; using application_type = dummy_application;
dummy_application_factory(std::shared_ptr<std::vector<byte>> buf) explicit dummy_application_factory(std::shared_ptr<std::vector<byte>> buf)
: buf_(buf) { : buf_(std::move(buf)) {
// nop // nop
} }
...@@ -159,7 +173,6 @@ CAF_TEST(receive) { ...@@ -159,7 +173,6 @@ CAF_TEST(receive) {
transport.configure_read(net::receive_policy::exactly(hello_manager.size())); transport.configure_read(net::receive_policy::exactly(hello_manager.size()));
auto mgr = make_endpoint_manager(mpx, sys, transport); auto mgr = make_endpoint_manager(mpx, sys, transport);
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_CHECK_EQUAL(write(sender, as_bytes(make_span(hello_manager)), ep), CAF_CHECK_EQUAL(write(sender, as_bytes(make_span(hello_manager)), ep),
hello_manager.size()); hello_manager.size());
...@@ -172,7 +185,6 @@ CAF_TEST(receive) { ...@@ -172,7 +185,6 @@ CAF_TEST(receive) {
// TODO: test is disabled until resolve in transport_worker_dispatcher is // TODO: test is disabled until resolve in transport_worker_dispatcher is
// implemented correctly. // implemented correctly.
// Idea is to use caf::uri instead of std::string.
/* /*
CAF_TEST(resolve and proxy communication) { CAF_TEST(resolve and proxy communication) {
using transport_type = datagram_transport<dummy_application_factory>; using transport_type = datagram_transport<dummy_application_factory>;
......
...@@ -180,8 +180,8 @@ uri operator"" _u(const char* cstr, size_t cstr_len) { ...@@ -180,8 +180,8 @@ 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<
ip_endpoint>; dummy_transport, dummy_application_factory, ip_endpoint>;
fixture() fixture()
: buf{std::make_shared<buffer_type>()}, : buf{std::make_shared<buffer_type>()},
......
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