Commit 61fbafb9 authored by Dominik Charousset's avatar Dominik Charousset

Properly implement BASP monitor and down messages

parent c3d7504e
......@@ -20,6 +20,7 @@
#include <cstdint>
#include <memory>
#include <type_traits>
#include <unordered_map>
#include <unordered_set>
#include <vector>
......@@ -56,6 +57,8 @@ public:
using write_packet_callback = callback<byte_span, byte_span>;
struct test_tag {};
// -- constructors, destructors, and assignment operators --------------------
explicit application(proxy_registry& proxies);
......@@ -66,6 +69,10 @@ public:
error init(Parent& parent) {
// Initialize member variables.
system_ = &parent.system();
// TODO: use `if constexpr` when switching to C++17.
// Allow unit tests to run the application without endpoint manager.
if (!std::is_base_of<test_tag, Parent>::value)
manager_ = &parent.manager();
// Write handshake.
if (auto err = generate_handshake())
return err;
......@@ -79,18 +86,8 @@ public:
template <class Parent>
error write_message(Parent& parent,
std::unique_ptr<endpoint_manager::message> ptr) {
// There is one special case: actor_proxy_impl sends a message without
// mailbox element on construction to trigger monitoring messages.
if (ptr->msg == nullptr) {
CAF_ASSERT(ptr->payload.empty());
CAF_ASSERT(ptr->receiver != nullptr);
CAF_ASSERT(ptr->receiver->node() == peer_id_);
header hdr{message_type::monitor_message, 0,
static_cast<uint64_t>(ptr->receiver->id())};
auto bytes = to_bytes(hdr);
parent.write_packet(make_span(bytes), span<const byte>{});
return none;
}
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->msg != nullptr);
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
const auto& src = ptr->msg->sender;
......@@ -137,8 +134,27 @@ public:
resolve_remote_path(write_packet, path, listener);
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
template <class Parent>
void new_proxy(Parent& parent, actor_id id) {
header hdr{message_type::monitor_message, 0, static_cast<uint64_t>(id)};
auto bytes = to_bytes(hdr);
parent.write_packet(make_span(bytes), span<const byte>{});
}
template <class Parent>
void local_actor_down(Parent& parent, actor_id id, error reason) {
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
if (auto err = sink(reason))
CAF_RAISE_ERROR("unable to serialize an error");
header hdr{message_type::down_message, static_cast<uint32_t>(buf_.size()),
static_cast<uint64_t>(id)};
auto bytes = to_bytes(hdr);
parent.write_packet(make_span(bytes), make_span(buf_));
}
template <class Parent>
void timeout(Parent&, atom_value, uint64_t) {
// nop
}
......@@ -187,6 +203,9 @@ private:
error handle_resolve_response(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_monitor_message(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_down_message(write_packet_callback& write_packet, header hdr,
byte_span payload);
......@@ -224,6 +243,9 @@ private:
/// Points to the factory object for generating proxies.
proxy_registry& proxies_;
/// Points to the endpoint manager that owns this applications.
endpoint_manager* manager_ = nullptr;
};
} // namespace basp
......
......@@ -56,6 +56,17 @@ public:
actor listener;
};
struct new_proxy {
node_id peer;
actor_id id;
};
struct local_actor_down {
node_id observing_peer;
actor_id id;
error reason;
};
struct timeout {
atom_value type;
uint64_t id;
......@@ -63,10 +74,14 @@ public:
event(uri locator, actor listener);
event(node_id peer, actor_id proxy_id);
event(node_id observing_peer, actor_id local_actor_id, error reason);
event(atom_value type, uint64_t id);
/// Either contains a string for `resolve` requests or an `atom_value`
variant<resolve_request, timeout> value;
/// Holds the event data.
variant<resolve_request, new_proxy, local_actor_down, timeout> value;
};
struct event_policy {
......@@ -144,8 +159,11 @@ public:
void enqueue(mailbox_element_ptr msg, strong_actor_ptr receiver,
std::vector<byte> payload);
/// Enqueues a timeout to the endpoint.
void enqueue(timeout_msg msg);
/// Enqueues an event to the endpoint.
template <class... Ts>
void enqueue_event(Ts&&... xs) {
enqueue(new event(std::forward<Ts>(xs)...));
}
// -- pure virtual member functions ------------------------------------------
......@@ -156,6 +174,8 @@ public:
virtual serialize_fun_type serialize_fun() const noexcept = 0;
protected:
void enqueue(event* ptr);
/// Points to the hosting actor system.
actor_system& sys_;
......
......@@ -22,6 +22,7 @@
#include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp"
#include "caf/atom.hpp"
#include "caf/detail/overload.hpp"
#include "caf/net/endpoint_manager.hpp"
namespace caf {
......@@ -93,14 +94,21 @@ public:
auto& q = this->events_.queue();
q.inc_deficit(q.total_task_size());
for (auto ptr = q.next(); ptr != nullptr; ptr = q.next()) {
using timeout = endpoint_manager::event::timeout;
using resolve_request = endpoint_manager::event::resolve_request;
if (auto rr = get_if<resolve_request>(&ptr->value)) {
transport_.resolve(*this, rr->locator, rr->listener);
} else {
auto& t = get<timeout>(ptr->value);
transport_.timeout(*this, t.type, t.id);
}
auto f = detail::make_overload(
[&](endpoint_manager::event::resolve_request& x) {
transport_.resolve(*this, x.locator, x.listener);
},
[&](endpoint_manager::event::new_proxy& x) {
transport_.new_proxy(*this, x.peer, x.id);
},
[&](endpoint_manager::event::local_actor_down& x) {
transport_.local_actor_down(*this, x.observing_peer, x.id,
std::move(x.reason));
},
[&](endpoint_manager::event::timeout& x) {
transport_.timeout(*this, x.type, x.id);
});
visit(f, ptr->value);
}
} while (!this->events_.try_block());
}
......
......@@ -104,7 +104,10 @@ public:
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
collected_ += *num_bytes;
if (collected_ >= read_threshold_) {
worker_.handle_data(*this, read_buf_);
if (auto err = worker_.handle_data(*this, read_buf_)) {
CAF_LOG_WARNING("handle_data failed:" << CAF_ARG(err));
return false;
}
prepare_next_read();
}
return true;
......@@ -135,9 +138,15 @@ public:
worker_.resolve(*this, locator.path(), listener);
}
template <class... Ts>
void set_timeout(uint64_t, Ts&&...) {
// nop
template <class Parent>
void new_proxy(Parent&, const node_id& peer, actor_id id) {
worker_.new_proxy(*this, peer, id);
}
template <class Parent>
void local_actor_down(Parent&, const node_id& peer, actor_id id,
error reason) {
worker_.local_actor_down(*this, peer, id, std::move(reason));
}
template <class Parent>
......@@ -145,6 +154,11 @@ public:
worker_.timeout(*this, value, id);
}
template <class... Ts>
void set_timeout(uint64_t, Ts&&...) {
// nop
}
void handle_error(sec code) {
worker_.handle_error(code);
}
......
......@@ -69,9 +69,9 @@ public:
}
template <class Parent>
void handle_data(Parent& parent, span<const byte> data) {
error handle_data(Parent& parent, span<const byte> data) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.handle_data(decorator, data);
return application_.handle_data(decorator, data);
}
template <class Parent>
......@@ -87,6 +87,19 @@ public:
application_.resolve(decorator, path, listener);
}
template <class Parent>
void new_proxy(Parent& parent, const node_id&, actor_id id) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.new_proxy(decorator, id);
}
template <class Parent>
void local_actor_down(Parent& parent, const node_id&, actor_id id,
error reason) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.local_actor_down(decorator, id, std::move(reason));
}
template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) {
auto decorator = make_write_packet_decorator(*this, parent);
......
......@@ -64,7 +64,7 @@ public:
}
template <class Parent>
void 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 (it == workers_by_id_.end()) {
// TODO: where to get node_id from here?
......@@ -72,7 +72,7 @@ public:
it = workers_by_id_.find(id);
}
auto worker = it->second;
worker->handle_data(parent, data);
return worker->handle_data(parent, data);
}
template <class Parent>
......
......@@ -28,7 +28,7 @@ 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)) {
CAF_ASSERT(dst_ != nullptr);
dst_->enqueue(nullptr, ctrl(), {});
dst_->enqueue_event(node(), id());
}
actor_proxy_impl::~actor_proxy_impl() {
......
......@@ -29,6 +29,7 @@
#include "caf/detail/parse.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/no_stages.hpp"
......@@ -56,6 +57,7 @@ expected<std::vector<byte>> application::serialize(actor_system& sys,
}
strong_actor_ptr application::resolve_local_path(string_view path) {
CAF_LOG_TRACE(CAF_ARG(path));
// We currently support two path formats: `id/<actor_id>` and `name/<atom>`.
static constexpr string_view id_prefix = "id/";
if (starts_with(path, id_prefix)) {
......@@ -78,6 +80,7 @@ strong_actor_ptr application::resolve_local_path(string_view path) {
void application::resolve_remote_path(write_packet_callback& write_packet,
string_view path, actor listener) {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener));
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
if (auto err = sink(path)) {
......@@ -99,6 +102,7 @@ void application::resolve_remote_path(write_packet_callback& write_packet,
error application::handle(size_t& next_read_size,
write_packet_callback& write_packet,
byte_span bytes) {
CAF_LOG_TRACE(CAF_ARG(state_) << CAF_ARG2("bytes.size", bytes.size()));
switch (state_) {
case connection_state::await_handshake_header: {
if (bytes.size() != header_size)
......@@ -143,6 +147,7 @@ error application::handle(size_t& next_read_size,
error application::handle(write_packet_callback& write_packet, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
switch (hdr.type) {
case message_type::handshake:
return ec::unexpected_handshake;
......@@ -152,10 +157,12 @@ error application::handle(write_packet_callback& write_packet, header hdr,
return handle_resolve_request(write_packet, hdr, payload);
case message_type::resolve_response:
return handle_resolve_response(write_packet, hdr, payload);
case message_type::heartbeat:
return none;
case message_type::monitor_message:
return handle_monitor_message(write_packet, hdr, payload);
case message_type::down_message:
return handle_down_message(write_packet, hdr, payload);
case message_type::heartbeat:
return none;
default:
return ec::unimplemented;
}
......@@ -163,6 +170,7 @@ error application::handle(write_packet_callback& write_packet, header hdr,
error application::handle_handshake(write_packet_callback&, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
if (hdr.type != message_type::handshake)
return ec::missing_handshake;
if (hdr.operation_data != version)
......@@ -188,6 +196,7 @@ error application::handle_handshake(write_packet_callback&, header hdr,
error application::handle_actor_message(write_packet_callback&, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
// Deserialize payload.
actor_id src_id;
node_id src_node;
......@@ -220,6 +229,7 @@ error application::handle_actor_message(write_packet_callback&, header hdr,
error application::handle_resolve_request(write_packet_callback& write_packet,
header hdr, byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
CAF_ASSERT(hdr.type == message_type::resolve_request);
size_t path_size = 0;
binary_deserializer source{system(), payload};
......@@ -248,6 +258,7 @@ error application::handle_resolve_request(write_packet_callback& write_packet,
error application::handle_resolve_response(write_packet_callback&, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
CAF_ASSERT(hdr.type == message_type::resolve_response);
auto i = pending_resolves_.find(hdr.operation_data);
if (i == pending_resolves_.end()) {
......@@ -272,8 +283,36 @@ error application::handle_resolve_response(write_packet_callback&, header hdr,
return none;
}
error application::handle_monitor_message(write_packet_callback& write_packet,
header hdr, byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
if (!payload.empty())
return ec::unexpected_payload;
auto aid = static_cast<actor_id>(hdr.operation_data);
auto hdl = system().registry().get(aid);
if (hdl != nullptr) {
endpoint_manager_ptr mgr = manager_;
auto nid = peer_id_;
hdl->get()->attach_functor([mgr, nid, aid](error reason) mutable {
mgr->enqueue_event(std::move(nid), aid, std::move(reason));
});
} else {
error reason = exit_reason::unknown;
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
if (auto err = sink(reason))
return err;
auto out_hdr = to_bytes(header{message_type::down_message,
static_cast<uint32_t>(buf_.size()),
hdr.operation_data});
return write_packet(out_hdr, buf_);
}
return none;
}
error application::handle_down_message(write_packet_callback&, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
error reason;
binary_deserializer source{system(), payload};
if (auto err = source(reason))
......
......@@ -31,6 +31,17 @@ endpoint_manager::event::event(uri locator, actor listener)
// nop
}
endpoint_manager::event::event(node_id peer, actor_id proxy_id)
: value(new_proxy{peer, proxy_id}) {
// nop
}
endpoint_manager::event::event(node_id observing_peer, actor_id local_actor_id,
error reason)
: value(local_actor_down{observing_peer, local_actor_id, std::move(reason)}) {
// nop
}
endpoint_manager::event::event(atom_value type, uint64_t id)
: value(timeout{type, id}) {
// nop
......@@ -98,5 +109,10 @@ void endpoint_manager::enqueue(mailbox_element_ptr msg,
mask_add(operation::write);
}
void endpoint_manager::enqueue(event* ptr) {
if (events_.push_back(ptr) == intrusive::inbox_result::unblocked_reader)
mask_add(operation::write);
}
} // namespace net
} // namespace caf
......@@ -41,7 +41,9 @@ using namespace caf::net;
namespace {
struct fixture : test_coordinator_fixture<>, proxy_registry::backend {
struct fixture : test_coordinator_fixture<>,
proxy_registry::backend,
basp::application::test_tag {
using buffer_type = std::vector<byte>;
fixture() : proxies(sys, *this), app(proxies) {
......@@ -108,6 +110,10 @@ struct fixture : test_coordinator_fixture<>, proxy_registry::backend {
return *this;
}
endpoint_manager& manager() {
CAF_FAIL("unexpected function call");
}
template <class... Ts>
void configure_read(Ts...) {
// nop
......
......@@ -138,6 +138,16 @@ public:
// nop
}
template <class Parent>
void new_proxy(Parent&, const node_id&, actor_id) {
// nop
}
template <class Parent>
void local_actor_down(Parent&, const node_id&, actor_id, error) {
// nop
}
private:
stream_socket handle_;
......
......@@ -79,9 +79,10 @@ public:
}
template <class Parent>
void handle_data(Parent&, span<const byte> data) {
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>
......@@ -99,8 +100,18 @@ public:
std::string{path.begin(), path.end()}, p);
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
template <class Parent>
void timeout(Parent&, atom_value, uint64_t) {
// nop
}
template <class Parent>
void new_proxy(Parent&, actor_id) {
// nop
}
template <class Parent>
void local_actor_down(Parent&, actor_id, error) {
// nop
}
......
......@@ -141,7 +141,7 @@ public:
}
template <class Parent>
void handle_data(Parent& parent, span<const byte> data) {
error handle_data(Parent& parent, span<const byte> data) {
if (await_payload_) {
Base::handle_packet(parent, header_, data);
await_payload_ = false;
......@@ -156,6 +156,7 @@ public:
net::receive_policy::exactly(header_.payload));
await_payload_ = true;
}
return none;
}
template <class Parent>
......@@ -172,8 +173,18 @@ public:
std::string{path.begin(), path.end()}, p);
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
template <class Parent>
void timeout(Parent&, atom_value, uint64_t) {
// nop
}
template <class Parent>
void new_proxy(Parent&, actor_id) {
// nop
}
template <class Parent>
void local_actor_down(Parent&, actor_id, error) {
// nop
}
......
......@@ -77,10 +77,11 @@ public:
}
template <class Parent>
void handle_data(Parent&, span<const byte> data) {
error handle_data(Parent&, span<const byte> data) {
auto& buf = res_->data_buffer;
buf.clear();
buf.insert(buf.begin(), data.begin(), data.end());
return none;
}
template <class Parent>
......
......@@ -70,8 +70,9 @@ public:
}
template <class Parent>
void handle_data(Parent&, span<const byte>) {
error handle_data(Parent&, span<const byte>) {
rec_buf_->push_back(static_cast<byte>(id_));
return none;
}
template <class Manager>
......
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