Commit e8ca5421 authored by Jakob Otto's avatar Jakob Otto

Refactor basp::application

parent bddef5ec
......@@ -34,6 +34,7 @@
#include "caf/net/basp/header.hpp"
#include "caf/net/basp/message_type.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/packet_writer.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/node_id.hpp"
#include "caf/proxy_registry.hpp"
......@@ -77,11 +78,14 @@ public:
if (!std::is_base_of<test_tag, Parent>::value)
manager_ = &parent.manager();
// Write handshake.
if (auto err = generate_handshake())
auto header_buf = parent.next_header_buffer();
auto payload_buf = parent.next_buffer();
if (auto err = generate_handshake(payload_buf))
return err;
auto hdr = to_bytes(header{message_type::handshake,
static_cast<uint32_t>(buf_.size()), version});
parent.write_packet(hdr, buf_);
to_bytes(header{message_type::handshake,
static_cast<uint32_t>(payload_buf.size()), version},
header_buf);
parent.write_packet(header_buf, payload_buf);
parent.transport().configure_read(receive_policy::exactly(header_size));
return none;
}
......@@ -89,52 +93,50 @@ public:
template <class Parent>
error write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> ptr) {
auto write_packet = make_callback([&](byte_span hdr, byte_span payload) {
parent.write_packet(hdr, payload);
return none;
});
return write(write_packet, std::move(ptr));
return write(parent, std::move(ptr));
}
template <class Parent>
error handle_data(Parent& parent, byte_span bytes) {
auto write_packet = make_callback([&](byte_span hdr, byte_span payload) {
parent.write_packet(hdr, payload);
return none;
});
static_assert(std::is_base_of<packet_writer, Parent>::value,
"parent must implement packet_writer");
size_t next_read_size = header_size;
if (auto err = handle(next_read_size, write_packet, bytes))
if (auto err = handle(next_read_size, parent, bytes))
return err;
parent.transport().configure_read(receive_policy::exactly(next_read_size));
return none;
}
// TODO: unessecary indirection
template <class Parent>
void resolve(Parent& parent, string_view path, actor listener) {
auto write_packet = make_callback([&](byte_span hdr, byte_span payload) {
parent.write_packet(hdr, payload);
return none;
});
resolve_remote_path(write_packet, path, listener);
void resolve(Parent& parent, string_view path, const actor& listener) {
static_assert(std::is_base_of<packet_writer, Parent>::value,
"parent must implement `packet_writer`");
resolve_remote_path(parent, path, listener);
}
// TODO: can be packet_writer&?
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>{});
auto header_buf = parent.next_header_buffer();
to_bytes(hdr, header_buf);
parent.write_packet(header_buf);
}
// TODO: can be packet_writer&?
template <class Parent>
void local_actor_down(Parent& parent, actor_id id, error reason) {
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
auto header_buf = parent.next_header_buffer();
auto payload_buf = parent.next_buffer();
serializer_impl<buffer_type> sink{system(), payload_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()),
header hdr{message_type::down_message,
static_cast<uint32_t>(payload_buf.size()),
static_cast<uint64_t>(id)};
auto bytes = to_bytes(hdr);
parent.write_packet(make_span(bytes), make_span(buf_));
to_bytes(hdr, header_buf);
parent.write_packet(header_buf, payload_buf);
}
template <class Parent>
......@@ -153,8 +155,8 @@ public:
strong_actor_ptr resolve_local_path(string_view path);
void resolve_remote_path(write_packet_callback& write_packet,
string_view path, actor listener);
void resolve_remote_path(packet_writer& writer, string_view path,
const actor& listener);
// -- properties -------------------------------------------------------------
......@@ -169,37 +171,34 @@ public:
private:
// -- handling of outgoing messages ------------------------------------------
error write(write_packet_callback& write_packet,
error write(packet_writer& writer,
std::unique_ptr<endpoint_manager_queue::message> ptr);
// -- handling of incoming messages ------------------------------------------
error handle(size_t& next_read_size, write_packet_callback& write_packet,
byte_span bytes);
error handle(size_t& next_read_size, packet_writer& writer, byte_span bytes);
error handle(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle(packet_writer& writer, header hdr, byte_span payload);
error handle_handshake(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_handshake(packet_writer& writer, header hdr, byte_span payload);
error handle_actor_message(write_packet_callback& write_packet, header hdr,
error handle_actor_message(packet_writer& writer, header hdr,
byte_span payload);
error handle_resolve_request(write_packet_callback& write_packet, header hdr,
error handle_resolve_request(packet_writer& writer, header hdr,
byte_span payload);
error handle_resolve_response(write_packet_callback& write_packet, header hdr,
error handle_resolve_response(packet_writer& writer, header hdr,
byte_span payload);
error handle_monitor_message(write_packet_callback& write_packet, header hdr,
error handle_monitor_message(packet_writer& writer, header hdr,
byte_span payload);
error handle_down_message(write_packet_callback& write_packet, header hdr,
error handle_down_message(packet_writer& writer, header hdr,
byte_span payload);
/// Writes the handshake payload to `buf_`.
error generate_handshake();
error generate_handshake(std::vector<byte>& buf);
// -- member variables -------------------------------------------------------
......@@ -212,14 +211,11 @@ private:
/// Caches the last header while waiting for the matching payload.
header hdr_;
/// Re-usable buffer for storing payloads.
buffer_type buf_;
/// Stores the ID of our peer.
node_id peer_id_;
/// Tracks which local actors our peer monitors.
std::unordered_set<actor_addr> monitored_actors_;
std::unordered_set<actor_addr> monitored_actors_; // TODO: this is unused
/// Caches actor handles obtained via `resolve`.
std::unordered_map<uint64_t, response_promise> pending_resolves_;
......
......@@ -32,6 +32,7 @@
#include "caf/logger.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/packet_writer.hpp"
#include "caf/no_stages.hpp"
#include "caf/none.hpp"
#include "caf/sec.hpp"
......@@ -78,34 +79,34 @@ strong_actor_ptr application::resolve_local_path(string_view path) {
return nullptr;
}
void application::resolve_remote_path(write_packet_callback& write_packet,
string_view path, actor listener) {
void application::resolve_remote_path(packet_writer& writer, string_view path,
const actor& listener) {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener));
buf_.clear();
serializer_impl<buffer_type> sink{&executor_, buf_};
auto header_buf = writer.next_header_buffer();
auto payload_buf = writer.next_buffer();
serializer_impl<buffer_type> sink{&executor_, payload_buf};
if (auto err = sink(path)) {
CAF_LOG_ERROR("unable to serialize path");
CAF_LOG_ERROR("unable to serialize path" << CAF_ARG(err));
return;
}
auto req_id = next_request_id_++;
auto hdr = to_bytes(header{message_type::resolve_request,
static_cast<uint32_t>(buf_.size()), req_id});
if (auto err = write_packet(hdr, buf_)) {
CAF_LOG_ERROR("unable to write resolve_request header");
return;
}
to_bytes(header{message_type::resolve_request,
static_cast<uint32_t>(payload_buf.size()), req_id},
header_buf);
writer.write_packet(header_buf, payload_buf);
response_promise rp{nullptr, actor_cast<strong_actor_ptr>(listener),
no_stages, make_message_id()};
pending_resolves_.emplace(req_id, std::move(rp));
}
error application::write(write_packet_callback& write_packet,
error application::write(packet_writer& writer,
std::unique_ptr<endpoint_manager_queue::message> ptr) {
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->msg != nullptr);
CAF_LOG_TRACE(CAF_ARG2("content", ptr->msg->content()));
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
auto header_buf = writer.next_header_buffer();
auto payload_elem_buf = writer.next_buffer();
serializer_impl<buffer_type> sink{system(), payload_elem_buf};
const auto& src = ptr->msg->sender;
const auto& dst = ptr->receiver;
if (dst == nullptr) {
......@@ -121,16 +122,18 @@ error application::write(write_packet_callback& write_packet,
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages))
return err;
}
// TODO: avoid extra copy of the payload
buf_.insert(buf_.end(), ptr->payload.begin(), ptr->payload.end());
header hdr{message_type::actor_message, static_cast<uint32_t>(buf_.size()),
// TODO: Is this size correct?
header hdr{message_type::actor_message,
static_cast<uint32_t>(payload_elem_buf.size()
+ ptr->payload.size()),
ptr->msg->mid.integer_value()};
auto bytes = to_bytes(hdr);
return write_packet(make_span(bytes), make_span(buf_));
to_bytes(hdr, header_buf);
// TODO: OK to move payload out of message?
writer.write_packet(header_buf, payload_elem_buf, ptr->payload);
return none;
}
error application::handle(size_t& next_read_size,
write_packet_callback& write_packet,
error application::handle(size_t& next_read_size, packet_writer& writer,
byte_span bytes) {
CAF_LOG_TRACE(CAF_ARG(state_) << CAF_ARG2("bytes.size", bytes.size()));
switch (state_) {
......@@ -149,7 +152,7 @@ error application::handle(size_t& next_read_size,
return none;
}
case connection_state::await_handshake_payload: {
if (auto err = handle_handshake(write_packet, hdr_, bytes))
if (auto err = handle_handshake(writer, hdr_, bytes))
return err;
state_ = connection_state::await_header;
return none;
......@@ -159,7 +162,7 @@ error application::handle(size_t& next_read_size,
return ec::unexpected_number_of_bytes;
hdr_ = header::from_bytes(bytes);
if (hdr_.payload_len == 0)
return handle(write_packet, hdr_, byte_span{});
return handle(writer, hdr_, byte_span{});
next_read_size = hdr_.payload_len;
state_ = connection_state::await_payload;
return none;
......@@ -168,29 +171,29 @@ error application::handle(size_t& next_read_size,
if (bytes.size() != hdr_.payload_len)
return ec::unexpected_number_of_bytes;
state_ = connection_state::await_header;
return handle(write_packet, hdr_, bytes);
return handle(writer, hdr_, bytes);
}
default:
return ec::illegal_state;
}
}
error application::handle(write_packet_callback& write_packet, header hdr,
error application::handle(packet_writer& writer, 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;
case message_type::actor_message:
return handle_actor_message(write_packet, hdr, payload);
return handle_actor_message(writer, hdr, payload);
case message_type::resolve_request:
return handle_resolve_request(write_packet, hdr, payload);
return handle_resolve_request(writer, hdr, payload);
case message_type::resolve_response:
return handle_resolve_response(write_packet, hdr, payload);
return handle_resolve_response(writer, hdr, payload);
case message_type::monitor_message:
return handle_monitor_message(write_packet, hdr, payload);
return handle_monitor_message(writer, hdr, payload);
case message_type::down_message:
return handle_down_message(write_packet, hdr, payload);
return handle_down_message(writer, hdr, payload);
case message_type::heartbeat:
return none;
default:
......@@ -198,7 +201,7 @@ error application::handle(write_packet_callback& write_packet, header hdr,
}
}
error application::handle_handshake(write_packet_callback&, header hdr,
error application::handle_handshake(packet_writer&, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
if (hdr.type != message_type::handshake)
......@@ -224,7 +227,7 @@ error application::handle_handshake(write_packet_callback&, header hdr,
return none;
}
error application::handle_actor_message(write_packet_callback&, header hdr,
error application::handle_actor_message(packet_writer&, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
// Deserialize payload.
......@@ -257,8 +260,8 @@ error application::handle_actor_message(write_packet_callback&, header hdr,
return none;
}
error application::handle_resolve_request(write_packet_callback& write_packet,
header hdr, byte_span payload) {
error application::handle_resolve_request(packet_writer& writer, 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;
......@@ -273,7 +276,8 @@ error application::handle_resolve_request(write_packet_callback& write_packet,
remainder.size()};
// Write result.
auto result = resolve_local_path(path);
buf_.clear();
auto header_buf = writer.next_header_buffer();
auto payload_buf = writer.next_buffer();
actor_id aid;
std::set<std::string> ifs;
if (result) {
......@@ -283,16 +287,18 @@ error application::handle_resolve_request(write_packet_callback& write_packet,
aid = 0;
}
// TODO: figure out how to obtain messaging interface.
serializer_impl<buffer_type> sink{&executor_, buf_};
serializer_impl<buffer_type> sink{&executor_, payload_buf};
if (auto err = sink(aid, ifs))
return err;
auto out_hdr = to_bytes(header{message_type::resolve_response,
static_cast<uint32_t>(buf_.size()),
hdr.operation_data});
return write_packet(out_hdr, buf_);
to_bytes(header{message_type::resolve_response,
static_cast<uint32_t>(payload_buf.size()),
hdr.operation_data},
header_buf);
writer.write_packet(header_buf, payload_buf);
return none;
}
error application::handle_resolve_response(write_packet_callback&, header hdr,
error application::handle_resolve_response(packet_writer&, 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);
......@@ -319,8 +325,8 @@ 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) {
error application::handle_monitor_message(packet_writer& writer, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
if (!payload.empty())
return ec::unexpected_payload;
......@@ -334,19 +340,21 @@ error application::handle_monitor_message(write_packet_callback& write_packet,
});
} else {
error reason = exit_reason::unknown;
buf_.clear();
serializer_impl<buffer_type> sink{&executor_, buf_};
auto header_buf = writer.next_header_buffer();
auto payload_buf = writer.next_buffer();
serializer_impl<buffer_type> sink{&executor_, payload_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_);
to_bytes(header{message_type::down_message,
static_cast<uint32_t>(payload_buf.size()),
hdr.operation_data},
header_buf);
writer.write_packet(header_buf, payload_buf);
}
return none;
}
error application::handle_down_message(write_packet_callback&, header hdr,
error application::handle_down_message(packet_writer&, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
error reason;
......@@ -357,9 +365,9 @@ error application::handle_down_message(write_packet_callback&, header hdr,
return none;
}
error application::generate_handshake() {
buf_.clear();
serializer_impl<buffer_type> sink{&executor_, buf_};
error application::generate_handshake(std::vector<byte>& buf) {
buf.clear(); // TODO: really necessary?
serializer_impl<buffer_type> sink{&executor_, buf};
return sink(system().node(),
get_or(system().config(), "middleman.app-identifiers",
defaults::middleman::app_identifiers));
......
......@@ -29,6 +29,7 @@
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/packet_writer.hpp"
#include "caf/none.hpp"
#include "caf/uri.hpp"
......@@ -43,7 +44,8 @@ namespace {
struct fixture : test_coordinator_fixture<>,
proxy_registry::backend,
basp::application::test_tag {
basp::application::test_tag,
public packet_writer {
using buffer_type = std::vector<byte>;
fixture() : proxies(sys, *this), app(proxies) {
......@@ -66,11 +68,6 @@ struct fixture : test_coordinator_fixture<>,
input = to_buf(xs...);
}
void write_packet(span<const byte> hdr, span<const byte> payload) {
output.insert(output.end(), hdr.begin(), hdr.end());
output.insert(output.end(), payload.begin(), payload.end());
}
void handle_handshake() {
CAF_CHECK_EQUAL(app.state(),
basp::connection_state::await_handshake_header);
......@@ -114,6 +111,14 @@ struct fixture : test_coordinator_fixture<>,
CAF_FAIL("unexpected function call");
}
buffer_type next_buffer() override {
return {};
}
buffer_type next_header_buffer() override {
return {};
}
template <class... Ts>
void configure_read(Ts...) {
// nop
......@@ -130,6 +135,12 @@ struct fixture : test_coordinator_fixture<>,
// nop
}
protected:
void write_impl(span<buffer_type*> buffers) override {
for (auto buf : buffers)
output.insert(output.end(), buf->begin(), buf->end());
}
buffer_type input;
buffer_type output;
......
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