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

Merge pull request #20

Add transport_worker_dispatcher
parents 1fb4c58c 7f4679ab
...@@ -29,6 +29,8 @@ class multiplexer; ...@@ -29,6 +29,8 @@ class multiplexer;
class socket_manager; class socket_manager;
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>
class transport_worker_dispatcher;
struct network_socket; struct network_socket;
struct pipe_socket; struct pipe_socket;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <unordered_map>
#include "caf/byte.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/net/write_packet_decorator.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
namespace caf {
namespace net {
/// implements a dispatcher that dispatches between transport and workers.
template <class ApplicationFactory, class IdType>
class transport_worker_dispatcher {
public:
// -- member types -----------------------------------------------------------
using id_type = IdType;
using factory_type = ApplicationFactory;
using application_type = typename ApplicationFactory::application_type;
using worker_type = transport_worker<application_type, id_type>;
using worker_ptr = transport_worker_ptr<application_type, id_type>;
// -- constructors, destructors, and assignment operators --------------------
transport_worker_dispatcher(factory_type factory)
: factory_(std::move(factory)) {
// nop
}
// -- member functions -------------------------------------------------------
template <class Parent>
error init(Parent&) {
CAF_ASSERT(workers_by_id_.empty());
return none;
}
template <class Parent>
void handle_data(Parent& parent, span<byte> data, id_type id) {
auto it = workers_by_id_.find(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;
worker->handle_data(parent, data);
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) {
auto sender = msg->msg->sender;
if (!sender)
return;
auto nid = sender->node();
auto it = workers_by_node_.find(nid);
if (it == workers_by_node_.end()) {
// TODO: where to get id_type from here?
add_new_worker(parent, nid, id_type{});
it = workers_by_node_.find(nid);
}
auto worker = it->second;
worker->write_message(parent, std::move(msg));
}
template <class Parent>
void write_packet(Parent& parent, span<const byte> header,
span<const byte> payload, id_type id) {
parent.write_packet(header, payload, id);
}
template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) {
// TODO path should be uri to lookup the corresponding worker
// if enpoint is known -> resolve actor through worker
// if not connect to endpoint?!
if (workers_by_id_.empty())
return;
auto worker = workers_by_id_.begin()->second;
worker->resolve(parent, path, listener);
}
template <class... Ts>
void set_timeout(uint64_t timeout_id, id_type id, Ts&&...) {
workers_by_timeout_id_.emplace(timeout_id, workers_by_id_.at(id));
}
template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) {
auto worker = workers_by_timeout_id_.at(id);
worker->timeout(parent, value, id);
workers_by_timeout_id_.erase(id);
}
void handle_error(sec error) {
for (const auto& p : workers_by_id_) {
auto worker = p.second;
worker->handle_error(error);
}
}
template <class Parent>
error add_new_worker(Parent& parent, node_id node, id_type id) {
auto application = factory_.make();
auto worker = std::make_shared<worker_type>(std::move(application), id);
if (auto err = worker->init(parent))
return err;
workers_by_id_.emplace(std::move(id), worker);
workers_by_node_.emplace(std::move(node), std::move(worker));
return none;
}
private:
// -- worker lookups ---------------------------------------------------------
std::unordered_map<id_type, worker_ptr> workers_by_id_;
std::unordered_map<node_id, worker_ptr> workers_by_node_;
std::unordered_map<uint64_t, worker_ptr> workers_by_timeout_id_;
factory_type factory_;
};
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 transport_worker_dispatcher
#include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/make_actor.hpp"
#include "caf/monitorable_actor.hpp"
#include "caf/node_id.hpp"
#include "caf/uri.hpp"
using namespace caf;
using namespace caf::net;
namespace {
constexpr string_view hello_test = "hello_test";
struct dummy_actor : public monitorable_actor {
dummy_actor(actor_config& cfg) : monitorable_actor(cfg) {
// nop
}
void enqueue(mailbox_element_ptr, execution_unit*) override {
// nop
}
};
class dummy_application {
public:
dummy_application(std::shared_ptr<std::vector<byte>> rec_buf, uint8_t id)
: rec_buf_(std::move(rec_buf)),
id_(id){
// nop
};
~dummy_application() = default;
template <class Parent>
error init(Parent&) {
rec_buf_->push_back(static_cast<byte>(id_));
return none;
}
template <class Transport>
void write_message(Transport& transport,
std::unique_ptr<endpoint_manager::message> msg) {
rec_buf_->push_back(static_cast<byte>(id_));
transport.write_packet(span<byte>{}, make_span(msg->payload));
}
template <class Parent>
void handle_data(Parent&, span<const byte>) {
rec_buf_->push_back(static_cast<byte>(id_));
}
template <class Manager>
void resolve(Manager&, const std::string&, actor) {
rec_buf_->push_back(static_cast<byte>(id_));
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
rec_buf_->push_back(static_cast<byte>(id_));
}
void handle_error(sec) {
rec_buf_->push_back(static_cast<byte>(id_));
}
static expected<std::vector<byte>> serialize(actor_system&,
const type_erased_tuple&) {
return std::vector<byte>{};
}
private:
std::shared_ptr<std::vector<byte>> rec_buf_;
uint8_t id_;
};
struct dummy_application_factory {
public:
using application_type = dummy_application;
dummy_application_factory(std::shared_ptr<std::vector<byte>> buf)
: buf_(buf), application_cnt_(0) {
// nop
}
dummy_application make() {
return dummy_application{buf_, application_cnt_++};
}
private:
std::shared_ptr<std::vector<byte>> buf_;
uint8_t application_cnt_;
};
struct dummy_transport {
using transport_type = dummy_transport;
using application_type = dummy_application;
dummy_transport(std::shared_ptr<std::vector<byte>> buf) : buf_(buf) {
}
template <class IdType>
void write_packet(span<const byte> header, span<const byte> payload, IdType) {
buf_->insert(buf_->end(), header.begin(), header.end());
buf_->insert(buf_->end(), payload.begin(), payload.end());
}
private:
std::shared_ptr<std::vector<byte>> buf_;
};
struct testdata {
testdata(uint8_t worker_id, node_id id, ip_endpoint ep)
: worker_id(worker_id), nid(id), ep(ep) {
// nop
}
uint8_t worker_id;
node_id nid;
ip_endpoint ep;
};
// TODO: switch to std::operator""s when switching to C++14
ip_endpoint operator"" _ep(const char* cstr, size_t cstr_len) {
ip_endpoint ep;
string_view str(cstr, cstr_len);
if (auto err = parse(str, ep))
CAF_FAIL("parse returned error: " << err);
return ep;
}
uri operator"" _u(const char* cstr, size_t cstr_len) {
uri result;
string_view str{cstr, cstr_len};
auto err = parse(str, result);
if (err)
CAF_FAIL("error while parsing " << str << ": " << to_string(err));
return result;
}
struct fixture : host_fixture {
using dispatcher_type = transport_worker_dispatcher<dummy_application_factory,
ip_endpoint>;
fixture()
: buf{std::make_shared<std::vector<byte>>()},
dispatcher{dummy_application_factory{buf}},
dummy{buf} {
add_new_workers();
}
std::unique_ptr<net::endpoint_manager::message>
make_dummy_message(node_id nid) {
actor_id aid = 42;
actor_config cfg;
auto p = make_actor<dummy_actor, strong_actor_ptr>(aid, nid, &sys, cfg);
auto test_span = as_bytes(make_span(hello_test));
std::vector<byte> payload(test_span.begin(), test_span.end());
auto strong_actor = actor_cast<strong_actor_ptr>(p);
mailbox_element::forwarding_stack stack;
auto elem = make_mailbox_element(std::move(strong_actor),
make_message_id(12345), std::move(stack),
make_message());
return detail::make_unique<endpoint_manager::message>(std::move(elem),
payload);
}
bool contains(byte x) {
return std::count(buf->begin(), buf->end(), x) > 0;
}
void add_new_workers() {
for (auto& data : test_data) {
if (auto err = dispatcher.add_new_worker(dummy, data.nid, data.ep))
CAF_FAIL("add_new_worker returned an error: " << err);
}
buf->clear();
}
void test_write_message(testdata& testcase) {
auto msg = make_dummy_message(testcase.nid);
if (!msg->msg->sender)
CAF_FAIL("sender is null");
dispatcher.write_message(dummy, std::move(msg));
}
actor_system_config cfg{};
actor_system sys{cfg};
std::shared_ptr<std::vector<byte>> buf;
dispatcher_type dispatcher;
dummy_transport dummy;
std::vector<testdata> test_data{
{0, make_node_id("http:file"_u), "[::1]:1"_ep},
{1, make_node_id("http:file?a=1&b=2"_u), "[fe80::2:34]:12345"_ep},
{2, make_node_id("http:file#42"_u), "[1234::17]:4444"_ep},
{3, make_node_id("http:file?a=1&b=2#42"_u), "[2332::1]:12"_ep},
};
};
#define CHECK_HANDLE_DATA(testcase) \
dispatcher.handle_data(dummy, span<byte>{}, testcase.ep); \
CAF_CHECK_EQUAL(buf->size(), 1u); \
CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \
buf->clear();
#define CHECK_WRITE_MESSAGE(testcase) \
test_write_message(testcase); \
CAF_CHECK_EQUAL(buf->size(), hello_test.size() + 1u); \
CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \
CAF_CHECK_EQUAL(memcmp(buf->data() + 1, hello_test.data(), \
hello_test.size()), \
0); \
buf->clear();
#define CHECK_TIMEOUT(testcase) \
dispatcher.set_timeout(1u, testcase.ep); \
dispatcher.timeout(dummy, atom("dummy"), 1u); \
CAF_CHECK_EQUAL(buf->size(), 1u); \
CAF_CHECK_EQUAL(static_cast<byte>(testcase.worker_id), buf->at(0)); \
buf->clear();
} // namespace
CAF_TEST_FIXTURE_SCOPE(transport_worker_dispatcher_test, fixture)
CAF_TEST(init) {
dispatcher_type dispatcher{dummy_application_factory{buf}};
if (auto err = dispatcher.init(dummy))
CAF_FAIL("init failed with error: " << err);
}
CAF_TEST(handle_data) {
CHECK_HANDLE_DATA(test_data.at(0));
CHECK_HANDLE_DATA(test_data.at(1));
CHECK_HANDLE_DATA(test_data.at(2));
CHECK_HANDLE_DATA(test_data.at(3));
}
CAF_TEST(write_message write_packet) {
CHECK_WRITE_MESSAGE(test_data.at(0));
CHECK_WRITE_MESSAGE(test_data.at(1));
CHECK_WRITE_MESSAGE(test_data.at(2));
CHECK_WRITE_MESSAGE(test_data.at(3));
}
CAF_TEST(resolve) {
// TODO think of a test for this
}
CAF_TEST(timeout) {
CHECK_TIMEOUT(test_data.at(0));
CHECK_TIMEOUT(test_data.at(1));
CHECK_TIMEOUT(test_data.at(2));
CHECK_TIMEOUT(test_data.at(3));
}
CAF_TEST(handle_error) {
dispatcher.handle_error(sec::unavailable_or_would_block);
CAF_CHECK_EQUAL(buf->size(), 4u);
CAF_CHECK(contains(byte(0)));
CAF_CHECK(contains(byte(1)));
CAF_CHECK(contains(byte(2)));
CAF_CHECK(contains(byte(3)));
}
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