Commit 86ddf627 authored by Dominik Charousset's avatar Dominik Charousset

Implement proxy actors

parent 8d4eaa87
......@@ -7,6 +7,7 @@ file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp")
# list cpp files excluding platform-dependent files
set(LIBCAF_NET_SRCS
src/actor_proxy_impl.cpp
src/endpoint_manager.cpp
src/host.cpp
src/multiplexer.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 "caf/actor_proxy.hpp"
#include "caf/net/endpoint_manager.hpp"
namespace caf {
namespace net {
/// Implements a simple proxy forwarding all operations to a manager.
class actor_proxy_impl : public actor_proxy {
public:
using super = actor_proxy;
using serialize_fun = std::vector<char> (*)(actor_system&,
const type_erased_tuple&);
actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst);
~actor_proxy_impl() override;
void enqueue(mailbox_element_ptr what, execution_unit* context) override;
bool add_backlink(abstract_actor* x) override;
bool remove_backlink(abstract_actor* x) override;
void kill_proxy(execution_unit* ctx, error rsn) override;
private:
endpoint_manager::serialize_fun_type sf_;
endpoint_manager_ptr dst_;
};
} // namespace net
} // namespace caf
......@@ -41,6 +41,13 @@ public:
using super = socket_manager;
/// Represents either an error or a serialized payload.
using maybe_buffer = expected<std::vector<char>>;
/// A function type for serializing message payloads.
using serialize_fun_type = maybe_buffer (*)(actor_system&,
const type_erased_tuple&);
struct event : intrusive::singly_linked<event> {
struct resolve_request {
std::string path;
......@@ -84,6 +91,8 @@ public:
/// Serialized representation of of `msg->content()`.
std::vector<char> payload;
message(mailbox_element_ptr msg, std::vector<char> payload);
};
struct message_policy {
......@@ -113,24 +122,28 @@ public:
// -- properties -------------------------------------------------------------
event_queue_type& event_queue() {
return events_;
actor_system& system() {
return sys_;
}
message_queue_type& message_queue() {
return messages_;
}
std::unique_ptr<message> next_message();
// -- event management -------------------------------------------------------
/// Resolves a path to a remote actor
/// Resolves a path to a remote actor.
void resolve(std::string path, actor listener);
/// Enqueues a message to the endpoint.
void enqueue(mailbox_element_ptr msg, std::vector<char> payload);
// -- pure virtual member functions ------------------------------------------
/// Initializes the manager before adding it to the multiplexer's event loop.
virtual error init() = 0;
/// @returns the protocol-specific function for serializing payloads.
virtual serialize_fun_type serialize_fun() const noexcept = 0;
protected:
/// Points to the hosting actor system.
actor_system& sys_;
......
......@@ -69,7 +69,8 @@ public:
}
bool handle_write_event() override {
if (!this->events_.empty()) {
if (!this->events_.blocked() && !this->events_.empty()) {
do {
this->events_.fetch_more();
auto& q = this->events_.queue();
q.inc_deficit(q.total_task_size());
......@@ -84,6 +85,7 @@ public:
transport_.timeout(*this, t.type, t.id);
}
}
} while (!this->events_.try_block());
}
return transport_.handle_write_event(*this);
}
......@@ -92,9 +94,13 @@ public:
transport_.handle_error(application_, code);
}
serialize_fun_type serialize_fun() const noexcept override {
return application_type::serialize;
}
private:
Transport transport_;
Application application_;
transport_type transport_;
application_type application_;
};
} // namespace net
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/actor_system.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp"
namespace caf {
namespace net {
actor_proxy_impl::actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst)
: super(cfg), sf_(dst->serialize_fun()), dst_(std::move(dst)) {
// anon_send(broker_, monitor_atom::value, ctrl());
}
actor_proxy_impl::~actor_proxy_impl() {
// anon_send(broker_, make_message(delete_atom::value, node(), id()));
}
void actor_proxy_impl::enqueue(mailbox_element_ptr what, execution_unit*) {
CAF_PUSH_AID(0);
CAF_ASSERT(what != nullptr);
if (auto payload = sf_(home_system(), what->content()))
dst_->enqueue(std::move(what), std::move(*payload));
else
CAF_LOG_ERROR(
"unable to serialize payload: " << home_system().render(payload.error()));
}
bool actor_proxy_impl::add_backlink(abstract_actor* x) {
if (monitorable_actor::add_backlink(x)) {
enqueue(make_mailbox_element(ctrl(), make_message_id(), {},
link_atom::value, x->ctrl()),
nullptr);
return true;
}
return false;
}
bool actor_proxy_impl::remove_backlink(abstract_actor* x) {
if (monitorable_actor::remove_backlink(x)) {
enqueue(make_mailbox_element(ctrl(), make_message_id(), {},
unlink_atom::value, x->ctrl()),
nullptr);
return true;
}
return false;
}
void actor_proxy_impl::kill_proxy(execution_unit* ctx, error rsn) {
cleanup(std::move(rsn), ctx);
}
} // namespace net
} // namespace caf
......@@ -35,19 +35,40 @@ endpoint_manager::event::event(atom_value type, uint64_t id)
// nop
}
endpoint_manager::message::message(mailbox_element_ptr msg,
std::vector<char> payload)
: msg(std::move(msg)), payload(std::move(payload)) {
// nop
}
endpoint_manager::endpoint_manager(socket handle, const multiplexer_ptr& parent,
actor_system& sys)
: super(handle, parent),
sys_(sys),
events_(event_policy{}),
messages_(message_policy{}) {
// nop
events_.try_block();
messages_.try_block();
}
endpoint_manager::~endpoint_manager() {
// nop
}
std::unique_ptr<endpoint_manager::message> endpoint_manager::next_message() {
if (messages_.blocked())
return nullptr;
messages_.fetch_more();
auto& q = messages_.queue();
auto ts = q.next_task_size();
if (ts == 0)
return nullptr;
q.inc_deficit(ts);
auto result = q.next();
messages_.try_block();
return result;
}
void endpoint_manager::resolve(std::string path, actor listener) {
using intrusive::inbox_result;
auto ptr = new event(std::move(path), std::move(listener));
......@@ -63,5 +84,12 @@ void endpoint_manager::resolve(std::string path, actor listener) {
}
}
void endpoint_manager::enqueue(mailbox_element_ptr msg,
std::vector<char> payload) {
auto ptr = new message(std::move(msg), std::move(payload));
if (messages_.push_back(ptr) == intrusive::inbox_result::unblocked_reader)
mask_add(operation::write);
}
} // namespace net
} // namespace caf
......@@ -24,7 +24,10 @@
#include "host_fixture.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp"
......@@ -45,10 +48,25 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
CAF_FAIL("mpx->init failed: " << sys.render(err));
}
bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false);
}
multiplexer_ptr mpx;
};
class dummy_application {};
class dummy_application {
public:
static expected<std::vector<char>> serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<char> result;
binary_serializer sink{sys, result};
if (auto err = message::save(sink, x))
return err;
return result;
}
};
class dummy_transport {
public:
......@@ -80,7 +98,11 @@ public:
}
template <class Manager>
bool handle_write_event(Manager&) {
bool handle_write_event(Manager& mgr) {
for (auto x = mgr.next_message(); x != nullptr; x = mgr.next_message()) {
auto& payload = x->payload;
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end());
}
auto res = write(handle_, write_buf_.data(), write_buf_.size());
if (auto num_bytes = get_if<size_t>(&res)) {
write_buf_.erase(write_buf_.begin(), write_buf_.begin() + *num_bytes);
......@@ -95,9 +117,14 @@ public:
}
template <class Manager>
void resolve(Manager&, std::string path, actor listener) {
anon_send(listener, resolve_atom::value, std::move(path),
make_error(sec::feature_disabled));
void resolve(Manager& mgr, std::string path, actor listener) {
actor_id aid = 42;
node_id nid{42, "00112233445566778899aa00112233445566778899aa"};
actor_config cfg;
auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid,
&mgr.system(), cfg,
&mgr);
anon_send(listener, resolve_atom::value, std::move(path), p);
}
template <class Manager>
......@@ -137,12 +164,51 @@ CAF_TEST(send and receive) {
CAF_CHECK_EQUAL(write(sockets.second, hello_manager.data(),
hello_manager.size()),
hello_manager.size());
while (mpx->poll_once(false))
; // Repeat.
run();
CAF_CHECK_EQUAL(string_view(buf->data(), buf->size()), hello_manager);
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()),
hello_test.size());
CAF_CHECK_EQUAL(string_view(read_buf.data(), hello_test.size()), hello_test);
}
CAF_TEST(resolve and proxy communication) {
std::vector<char> read_buf(1024);
auto buf = std::make_shared<std::vector<char>>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf},
dummy_application{});
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
run();
CAF_CHECK_EQUAL(read(sockets.second, read_buf.data(), read_buf.size()),
hello_test.size());
mgr->resolve("/id/42", 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();
auto read_res = read(sockets.second, read_buf.data(), read_buf.size());
if (!holds_alternative<size_t>(read_res)) {
CAF_ERROR("read() returned an error: " << sys.render(get<sec>(read_res)));
return;
}
read_buf.resize(get<size_t>(read_res));
CAF_MESSAGE("receive buffer contains " << read_buf.size() << " bytes");
message msg;
binary_deserializer source{sys, read_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()
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