Commit c0062a1a authored by Jakob Otto's avatar Jakob Otto

Add review feedback

parent ea0ffdef
...@@ -22,8 +22,6 @@ ...@@ -22,8 +22,6 @@
#include <unordered_map> #include <unordered_map>
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/detail/socket_sys_aliases.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/ip_endpoint.hpp" #include "caf/ip_endpoint.hpp"
...@@ -45,6 +43,10 @@ namespace net { ...@@ -45,6 +43,10 @@ namespace net {
template <class Factory> template <class Factory>
class datagram_transport { class datagram_transport {
public: public:
// Maximal UDP-packet size
static constexpr size_t max_datagram_size = std::numeric_limits<
uint16_t>::max();
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
using id_type = ip_endpoint; using id_type = ip_endpoint;
...@@ -66,10 +68,7 @@ public: ...@@ -66,10 +68,7 @@ public:
datagram_transport(udp_datagram_socket handle, factory_type factory) datagram_transport(udp_datagram_socket handle, factory_type factory)
: dispatcher_(*this, std::move(factory)), : dispatcher_(*this, std::move(factory)),
handle_(handle), handle_(handle),
max_consecutive_reads_(0), read_buf_(max_datagram_size),
read_threshold_(1024),
max_(1024),
rd_flag_(receive_policy_flag::exactly),
manager_(nullptr) { manager_(nullptr) {
// nop // nop
} }
...@@ -85,9 +84,7 @@ public: ...@@ -85,9 +84,7 @@ public:
} }
application_type& application() { application_type& application() {
// TODO: This wont work. We need information on which application is wanted return dispatcher_.application();
return application_type{};
// dispatcher_.application();
} }
transport_type& transport() { transport_type& transport() {
...@@ -150,8 +147,8 @@ public: ...@@ -150,8 +147,8 @@ public:
} }
template <class Parent> template <class Parent>
void resolve(Parent&, const uri& locator, const actor& listener) { error resolve(Parent&, const uri& locator, const actor& listener) {
dispatcher_.resolve(*this, locator, listener); return dispatcher_.resolve(*this, locator, listener);
} }
template <class Parent> template <class Parent>
...@@ -179,40 +176,19 @@ public: ...@@ -179,40 +176,19 @@ public:
} }
error add_new_worker(node_id node, id_type id) { error add_new_worker(node_id node, id_type id) {
return dispatcher_.add_new_worker(*this, node, id); auto worker = dispatcher_.add_new_worker(*this, node, id);
if (!worker)
return worker.error();
return none;
} }
void prepare_next_read() { void prepare_next_read() {
read_buf_.clear(); read_buf_.clear();
// This cast does nothing, but prevents a weird compiler error on GCC read_buf_.resize(max_datagram_size);
// <= 4.9.
// TODO: remove cast when dropping support for GCC 4.9.
switch (static_cast<receive_policy_flag>(rd_flag_)) {
case receive_policy_flag::exactly:
if (read_buf_.size() != max_)
read_buf_.resize(max_);
read_threshold_ = max_;
break;
case receive_policy_flag::at_most:
if (read_buf_.size() != max_)
read_buf_.resize(max_);
read_threshold_ = 1;
break;
case receive_policy_flag::at_least: {
// read up to 10% more, but at least allow 100 bytes more
auto max_size = max_ + std::max<size_t>(100, max_ / 10);
if (read_buf_.size() != max_size)
read_buf_.resize(max_size);
read_threshold_ = max_;
break;
}
}
} }
void configure_read(receive_policy::config cfg) { void configure_read(receive_policy::config) {
rd_flag_ = cfg.first; // nop
max_ = cfg.second;
prepare_next_read();
} }
void write_packet(id_type id, span<buffer_type*> buffers) { void write_packet(id_type id, span<buffer_type*> buffers) {
...@@ -314,12 +290,6 @@ private: ...@@ -314,12 +290,6 @@ private:
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 read_threshold_;
size_t max_;
receive_policy_flag rd_flag_;
endpoint_manager* manager_; endpoint_manager* manager_;
}; };
......
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#pragma once #pragma once
#include <caf/logger.hpp> #include <caf/logger.hpp>
#include <caf/sec.hpp>
#include <unordered_map> #include <unordered_map>
#include "caf/byte.hpp" #include "caf/byte.hpp"
...@@ -33,7 +34,7 @@ ...@@ -33,7 +34,7 @@
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 Transport, class IdType> template <class Transport, class IdType>
class transport_worker_dispatcher { class transport_worker_dispatcher {
public: public:
...@@ -68,13 +69,13 @@ public: ...@@ -68,13 +69,13 @@ 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 worker = find_by_id(id); if (auto worker = find_worker(id))
if (!worker) { return worker->handle_data(parent, data);
// TODO: where to get id_type from here? auto worker = add_new_worker(parent, node_id{}, id);
add_new_worker(parent, node_id{}, id); if (worker)
worker = find_by_id(id); return (*worker)->handle_data(parent, data);
} else
return worker->handle_data(parent, data); return std::move(worker.error());
} }
template <class Parent> template <class Parent>
...@@ -84,32 +85,40 @@ public: ...@@ -84,32 +85,40 @@ public:
if (!receiver) if (!receiver)
return; return;
auto nid = receiver->node(); auto nid = receiver->node();
auto worker = find_by_node(nid); if (auto worker = find_worker(nid)) {
if (!worker) { worker->write_message(parent, std::move(msg));
// TODO: where to get id_type from here? return;
add_new_worker(parent, nid, id_type{});
worker = find_by_node(nid);
} }
worker->write_message(parent, std::move(msg)); if (auto worker = add_new_worker(parent, nid, id_type{}))
(*worker)->write_message(parent, std::move(msg));
} }
template <class Parent> template <class Parent>
void resolve(Parent& parent, const uri& locator, const actor& listener) { error resolve(Parent& parent, const uri& locator, const actor& listener) {
if (auto worker = find_by_node(make_node_id(locator))) auto worker = find_worker(make_node_id(locator));
worker->resolve(parent, locator.path(), listener); if (worker == nullptr)
return make_error(sec::runtime_error, "could not find worker");
worker->resolve(parent, locator.path(), listener);
return none;
} }
template <class Parent> template <class Parent>
void new_proxy(Parent& parent, const node_id& nid, actor_id id) { error new_proxy(Parent& parent, const node_id& nid, actor_id id) {
if (auto worker = find_by_node(nid)) auto worker = find_worker(nid);
worker->new_proxy(parent, nid, id); if (worker == nullptr)
return make_error(sec::runtime_error, "could not find worker");
worker->new_proxy(parent, nid, id);
return none;
} }
template <class Parent> template <class Parent>
void local_actor_down(Parent& parent, const node_id& nid, actor_id id, error local_actor_down(Parent& parent, const node_id& nid, actor_id id,
error reason) { error reason) {
if (auto worker = find_by_node(nid)) auto worker = find_worker(nid);
worker->local_actor_down(parent, nid, id, std::move(reason)); if (worker == nullptr)
return make_error(sec::runtime_error, "could not find worker");
worker->local_actor_down(parent, nid, id, std::move(reason));
return none;
} }
template <class... Ts> template <class... Ts>
...@@ -132,40 +141,38 @@ public: ...@@ -132,40 +141,38 @@ 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_by_node(const node_id& nid) { worker_ptr find_worker(const node_id& nid) {
if (workers_by_node_.empty()) return find_worker_impl(workers_by_node_, nid);
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) { worker_ptr find_worker(const id_type& id) {
if (workers_by_id_.empty()) return find_worker_impl(workers_by_id_, id);
return nullptr; }
auto it = workers_by_id_.find(id);
if (it == workers_by_id_.end()) { template <class Key>
CAF_LOG_ERROR("could not find worker by node: " << CAF_ARG(id)); 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 nullptr;
} }
return it->second; 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_;
std::unordered_map<node_id, worker_ptr> workers_by_node_; std::unordered_map<node_id, worker_ptr> workers_by_node_;
...@@ -173,7 +180,7 @@ private: ...@@ -173,7 +180,7 @@ private:
factory_type factory_; factory_type factory_;
transport_type* transport_; transport_type* transport_;
}; }; // namespace net
} // namespace net } // namespace net
} // namespace caf } // namespace caf
...@@ -58,13 +58,12 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x, ...@@ -58,13 +58,12 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
/// Sends the content of `bufs` as a datagram to the endpoint `ep` on socket /// Sends the content of `bufs` as a datagram to the endpoint `ep` on socket
/// `x`. /// `x`.
/// @param x x The UDP socket for sending datagrams. /// @param x The UDP socket for sending datagrams.
/// @param bufs Points to the datagram to send, scattered across up to 10 /// @param bufs Points to the datagram to send, scattered across up to 10
/// buffers. /// buffers.
/// @param ep The enpoint to send the datagram to. /// @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
/// @post either the result is a `sec` or a positive (non-zero) integer
/// @pre `bufs.size() < 10` /// @pre `bufs.size() < 10`
variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs, variant<size_t, sec> write(udp_datagram_socket x, span<std::vector<byte>*> bufs,
ip_endpoint ep); ip_endpoint ep);
......
...@@ -24,19 +24,21 @@ ...@@ -24,19 +24,21 @@
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/detail/socket_sys_includes.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"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/endpoint_manager_impl.hpp" #include "caf/net/endpoint_manager_impl.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
#include "caf/net/socket_guard.hpp"
#include "caf/net/udp_datagram_socket.hpp" #include "caf/net/udp_datagram_socket.hpp"
#include "caf/serializer_impl.hpp" #include "caf/serializer_impl.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
using namespace caf; using namespace caf;
using namespace caf::net; using namespace caf::net;
using namespace caf::net::ip;
namespace { namespace {
...@@ -55,8 +57,9 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -55,8 +57,9 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
CAF_FAIL("mpx->init failed: " << sys.render(err)); CAF_FAIL("mpx->init failed: " << sys.render(err));
mpx->set_thread_id(); mpx->set_thread_id();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
if (auto err = parse("127.0.0.1:0", ep)) auto addresses = local_addresses("localhost");
CAF_FAIL("parse returned an error: " << sys.render(err)); CAF_CHECK(!addresses.empty());
ep = ip_endpoint(*addresses.begin(), 0);
auto send_pair = unbox(make_udp_datagram_socket(ep)); auto send_pair = unbox(make_udp_datagram_socket(ep));
send_socket = send_pair.first; send_socket = send_pair.first;
auto receive_pair = unbox(make_udp_datagram_socket(ep)); auto receive_pair = unbox(make_udp_datagram_socket(ep));
...@@ -106,13 +109,11 @@ class dummy_application { ...@@ -106,13 +109,11 @@ class dummy_application {
using buffer_ptr = std::shared_ptr<buffer_type>; using buffer_ptr = std::shared_ptr<buffer_type>;
public: public:
dummy_application(buffer_ptr rec_buf) explicit dummy_application(buffer_ptr rec_buf)
: rec_buf_(std::move(rec_buf)){ : rec_buf_(std::move(rec_buf)){
// nop // nop
}; };
~dummy_application() = default;
template <class Parent> template <class Parent>
error init(Parent&) { error init(Parent&) {
return none; return none;
......
...@@ -201,7 +201,7 @@ struct fixture : host_fixture { ...@@ -201,7 +201,7 @@ struct fixture : host_fixture {
buffer_type payload(test_span.begin(), test_span.end()); buffer_type payload(test_span.begin(), test_span.end());
auto receiver = actor_cast<strong_actor_ptr>(p); auto receiver = actor_cast<strong_actor_ptr>(p);
if (!receiver) if (!receiver)
CAF_FAIL("receiver cast failed"); 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(nullptr, make_message_id(12345), auto elem = make_mailbox_element(nullptr, make_message_id(12345),
std::move(stack), make_message()); std::move(stack), make_message());
...@@ -216,8 +216,9 @@ struct fixture : host_fixture { ...@@ -216,8 +216,9 @@ 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();
} }
...@@ -272,8 +273,7 @@ CAF_TEST_FIXTURE_SCOPE(transport_worker_dispatcher_test, fixture) ...@@ -272,8 +273,7 @@ CAF_TEST_FIXTURE_SCOPE(transport_worker_dispatcher_test, fixture)
CAF_TEST(init) { CAF_TEST(init) {
dispatcher_type dispatcher{dummy, 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) {
......
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