Unverified Commit 493ec8b2 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #32

No copy refactoring
parents 4da5fec8 3efaa0c0
......@@ -20,6 +20,7 @@ set(LIBCAF_NET_SRCS
src/application.cpp
src/convert_ip_endpoint.cpp
src/datagram_socket.cpp
src/defaults.cpp
src/ec.cpp
src/endpoint_manager.cpp
src/header.cpp
......@@ -30,6 +31,7 @@ set(LIBCAF_NET_SRCS
src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp
src/net/middleman_backend.cpp
src/net/packet_writer.cpp
src/network_socket.cpp
src/pipe_socket.cpp
src/pollset_updater.cpp
......
......@@ -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"
......@@ -56,8 +57,6 @@ public:
using byte_span = span<const byte>;
using write_packet_callback = callback<byte_span, byte_span>;
struct test_tag {};
// -- constructors, destructors, and assignment operators --------------------
......@@ -77,65 +76,37 @@ public:
if (!std::is_base_of<test_tag, Parent>::value)
manager_ = &parent.manager();
// Write handshake.
if (auto err = generate_handshake())
auto hdr = parent.next_header_buffer();
auto payload = parent.next_payload_buffer();
if (auto err = generate_handshake(payload))
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.size()), version},
hdr);
parent.write_packet(hdr, payload);
parent.transport().configure_read(receive_policy::exactly(header_size));
return none;
}
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));
}
error write_message(packet_writer& writer,
std::unique_ptr<endpoint_manager_queue::message> 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;
}
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(packet_writer& writer, string_view path, const actor& listener);
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>{});
}
static void new_proxy(packet_writer& writer, actor_id id);
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_));
}
void local_actor_down(packet_writer& writer, actor_id id, error reason);
template <class Parent>
void timeout(Parent&, atom_value, uint64_t) {
......@@ -146,16 +117,13 @@ public:
// nop
}
static expected<std::vector<byte>> serialize(actor_system& sys,
static expected<buffer_type> serialize(actor_system& sys,
const type_erased_tuple& x);
// -- utility functions ------------------------------------------------------
strong_actor_ptr resolve_local_path(string_view path);
void resolve_remote_path(write_packet_callback& write_packet,
string_view path, actor listener);
// -- properties -------------------------------------------------------------
connection_state state() const noexcept {
......@@ -167,39 +135,31 @@ public:
}
private:
// -- handling of outgoing messages ------------------------------------------
error write(write_packet_callback& write_packet,
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,
byte_span payload);
error handle_resolve_request(packet_writer& writer, header rec_hdr,
byte_span received);
error handle_resolve_response(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_resolve_response(packet_writer& writer, header received_hdr,
byte_span received);
error handle_monitor_message(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_monitor_message(packet_writer& writer, header received_hdr,
byte_span received);
error handle_down_message(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_down_message(packet_writer& writer, header received_hdr,
byte_span received);
/// Writes the handshake payload to `buf_`.
error generate_handshake();
error generate_handshake(buffer_type& buf);
// -- member variables -------------------------------------------------------
......@@ -212,14 +172,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_;
......
......@@ -78,6 +78,10 @@ struct header : detail::comparable<header> {
/// @relates header
std::array<byte, header_size> to_bytes(header x);
/// Serializes a header to a byte representation.
/// @relates header
void to_bytes(header x, std::vector<byte>& buf);
/// @relates header
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, header& x) {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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 <cstddef>
// -- hard-coded default values for various CAF options ------------------------
namespace caf {
namespace defaults {
namespace middleman {
/// Maximum number of cached buffers for sending payloads.
extern const size_t max_payload_buffers;
/// Maximum number of cached buffers for sending headers.
extern const size_t max_header_buffers;
} // namespace middleman
} // namespace defaults
} // 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. *
******************************************************************************/
#pragma once
#include <vector>
#include "caf/byte.hpp"
#include "caf/net/fwd.hpp"
#include "caf/span.hpp"
namespace caf {
namespace net {
/// Implements an interface for packet writing in application-layers.
class packet_writer {
public:
using buffer_type = std::vector<byte>;
virtual ~packet_writer();
/// Returns a buffer for writing header information.
virtual buffer_type next_header_buffer() = 0;
/// Returns a buffer for writing payload content.
virtual buffer_type next_payload_buffer() = 0;
/// Convenience function to write a packet consisting of multiple buffers.
/// @param buffers all buffers for the packet. The first buffer is a header
/// buffer, the other buffers are payload buffer.
/// @warning this function takes ownership of `buffers`.
template <class... Ts>
void write_packet(Ts&... buffers) {
buffer_type* bufs[] = {&buffers...};
write_impl(make_span(bufs, sizeof...(Ts)));
}
protected:
/// Implementing function for `write_packet`.
/// @param buffers a `span` containing all buffers of a packet.
virtual void write_impl(span<buffer_type*> buffers) = 0;
};
} // namespace net
} // namespace caf
......@@ -18,6 +18,8 @@
#pragma once
#include "caf/net/packet_writer.hpp"
#include "caf/byte.hpp"
#include "caf/span.hpp"
......@@ -25,9 +27,9 @@ namespace caf {
namespace net {
/// Implements the interface for transport and application policies and
/// dispatches member functions either to `decorator` or `parent`.
/// dispatches member functions either to `object` or `parent`.
template <class Object, class Parent>
class write_packet_decorator {
class packet_writer_decorator final : public packet_writer {
public:
// -- member types -----------------------------------------------------------
......@@ -37,7 +39,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
write_packet_decorator(Object& object, Parent& parent)
packet_writer_decorator(Object& object, Parent& parent)
: object_(object), parent_(parent) {
// nop
}
......@@ -56,15 +58,16 @@ public:
return parent_.manager();
}
// -- member functions -------------------------------------------------------
buffer_type next_header_buffer() override {
return transport().next_header_buffer();
}
template <class... Ts>
void write_packet(span<const byte> header, span<const byte> payload,
Ts&&... xs) {
parent_.write_packet(header, payload, std::forward<Ts>(xs)...,
object_.id());
buffer_type next_payload_buffer() override {
return transport().next_payload_buffer();
}
// -- member functions -------------------------------------------------------
void cancel_timeout(atom_value type, uint64_t id) {
parent_.cancel_timeout(type, id);
}
......@@ -74,14 +77,19 @@ public:
return parent_.set_timeout(tout, type, std::forward<Ts>(xs)...);
}
protected:
void write_impl(span<buffer_type*> buffers) override {
parent_.write_packet(object_.id(), buffers);
}
private:
Object& object_;
Parent& parent_;
};
template <class Object, class Parent>
write_packet_decorator<Object, Parent>
make_write_packet_decorator(Object& object, Parent& parent) {
packet_writer_decorator<Object, Parent>
make_packet_writer_decorator(Object& object, Parent& parent) {
return {object, parent};
}
......
......@@ -18,10 +18,13 @@
#pragma once
#include "caf/actor_system_config.hpp"
#include "caf/byte.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_socket.hpp"
......@@ -45,6 +48,10 @@ public:
using worker_type = transport_worker<application_type>;
using buffer_type = std::vector<byte>;
using buffer_cache_type = std::vector<buffer_type>;
// -- constructors, destructors, and assignment operators --------------------
stream_transport(stream_socket handle, application_type application)
......@@ -87,6 +94,13 @@ public:
template <class Parent>
error init(Parent& parent) {
manager_ = &parent;
auto& cfg = system().config();
auto max_header_bufs = get_or(cfg, "middleman.max-header-buffers",
defaults::middleman::max_header_buffers);
header_bufs_.reserve(max_header_bufs);
auto max_payload_bufs = get_or(cfg, "middleman.max-payload-buffers",
defaults::middleman::max_payload_buffers);
payload_bufs_.reserve(max_payload_bufs);
if (auto err = worker_.init(*this))
return err;
return none;
......@@ -198,49 +212,96 @@ public:
prepare_next_read();
}
void write_packet(span<const byte> header, span<const byte> payload,
typename worker_type::id_type) {
if (write_buf_.empty())
void write_packet(unit_t, span<buffer_type*> buffers) {
CAF_ASSERT(!buffers.empty());
if (write_queue_.empty())
manager().register_writing();
write_buf_.insert(write_buf_.end(), header.begin(), header.end());
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end());
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
auto i = buffers.begin();
write_queue_.emplace_back(true, std::move(*(*i++)));
while (i != buffers.end())
write_queue_.emplace_back(false, std::move(*(*i++)));
}
// -- buffer management ------------------------------------------------------
buffer_type next_header_buffer() {
return next_buffer_impl(header_bufs_);
}
buffer_type next_payload_buffer() {
return next_buffer_impl(payload_bufs_);
}
private:
// -- private member functions -----------------------------------------------
// -- utility functions ------------------------------------------------------
static buffer_type next_buffer_impl(buffer_cache_type cache) {
if (cache.empty()) {
return {};
}
auto buf = std::move(cache.back());
cache.pop_back();
return buf;
}
bool write_some() {
if (write_buf_.empty())
CAF_LOG_TRACE(CAF_ARG(handle_.id));
if (write_queue_.empty())
return false;
auto len = write_buf_.size() - written_;
auto buf = write_buf_.data() + written_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len));
auto ret = write(handle_, make_span(buf, len));
if (auto num_bytes = get_if<size_t>(&ret)) {
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
// Update state.
written_ += *num_bytes;
if (written_ >= write_buf_.size()) {
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = write_queue_.front();
auto& is_header = front.first;
auto& buf = front.second;
written_ = 0;
write_buf_.clear();
buf.clear();
if (is_header) {
if (header_bufs_.size() < header_bufs_.capacity())
header_bufs_.emplace_back(std::move(buf));
} else if (payload_bufs_.size() < payload_bufs_.capacity()) {
payload_bufs_.emplace_back(std::move(buf));
}
write_queue_.pop_front();
};
// Write buffers from the write_queue_ for as long as possible.
do {
auto& buf = write_queue_.front().second;
CAF_ASSERT(!buf.empty());
auto data = buf.data() + written_;
auto len = buf.size() - written_;
auto write_ret = write(handle_, make_span(data, len));
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
if (*num_bytes + written_ >= buf.size()) {
recycle();
written_ = 0;
} else {
written_ += *num_bytes;
return false;
}
} else {
auto err = get<sec>(ret);
auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
worker_.handle_error(err);
return false;
}
}
return true;
}
} while (!write_queue_.empty());
return false;
}
worker_type worker_;
stream_socket handle_;
std::vector<byte> read_buf_;
std::vector<byte> write_buf_;
buffer_cache_type header_bufs_;
buffer_cache_type payload_bufs_;
buffer_type read_buf_;
std::deque<std::pair<bool, buffer_type>> write_queue_;
// TODO implement retries using this member!
// size_t max_consecutive_reads_;
......
......@@ -22,7 +22,7 @@
#include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/write_packet_decorator.hpp"
#include "caf/net/packet_writer_decorator.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
......@@ -64,46 +64,46 @@ public:
template <class Parent>
error init(Parent& parent) {
auto decorator = make_write_packet_decorator(*this, parent);
return application_.init(decorator);
auto writer = make_packet_writer_decorator(*this, parent);
return application_.init(writer);
}
template <class Parent>
error handle_data(Parent& parent, span<const byte> data) {
auto decorator = make_write_packet_decorator(*this, parent);
return application_.handle_data(decorator, data);
auto writer = make_packet_writer_decorator(*this, parent);
return application_.handle_data(writer, data);
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.write_message(decorator, std::move(msg));
auto writer = make_packet_writer_decorator(*this, parent);
application_.write_message(writer, std::move(msg));
}
template <class Parent>
void resolve(Parent& parent, string_view path, const actor& listener) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.resolve(decorator, path, listener);
auto writer = make_packet_writer_decorator(*this, parent);
application_.resolve(writer, 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);
auto writer = make_packet_writer_decorator(*this, parent);
application_.new_proxy(writer, 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));
auto writer = make_packet_writer_decorator(*this, parent);
application_.local_actor_down(writer, 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);
application_.timeout(decorator, value, id);
auto writer = make_packet_writer_decorator(*this, parent);
application_.timeout(writer, value, id);
}
void handle_error(sec error) {
......
......@@ -24,8 +24,8 @@
#include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_decorator.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/net/write_packet_decorator.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
......
......@@ -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"
......@@ -47,6 +48,79 @@ application::application(proxy_registry& proxies) : proxies_(proxies) {
// nop
}
error application::write_message(
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()));
auto payload_prefix = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{system(), payload_prefix};
const auto& src = ptr->msg->sender;
const auto& dst = ptr->receiver;
if (dst == nullptr) {
// TODO: valid?
return none;
}
if (src != nullptr) {
auto src_id = src->id();
system().registry().put(src_id, src);
if (auto err = sink(src->node(), src_id, dst->id(), ptr->msg->stages))
return err;
} else {
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages))
return err;
}
auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::actor_message,
static_cast<uint32_t>(payload_prefix.size()
+ ptr->payload.size()),
ptr->msg->mid.integer_value()},
hdr);
writer.write_packet(hdr, payload_prefix, ptr->payload);
return none;
}
void application::resolve(packet_writer& writer, string_view path,
const actor& listener) {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener));
auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload};
if (auto err = sink(path)) {
CAF_LOG_ERROR("unable to serialize path" << CAF_ARG(err));
return;
}
auto req_id = next_request_id_++;
auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::resolve_request,
static_cast<uint32_t>(payload.size()), req_id},
hdr);
writer.write_packet(hdr, payload);
response_promise rp{nullptr, actor_cast<strong_actor_ptr>(listener),
no_stages, make_message_id()};
pending_resolves_.emplace(req_id, std::move(rp));
}
void application::new_proxy(packet_writer& writer, actor_id id) {
auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::monitor_message, 0, static_cast<uint64_t>(id)},
hdr);
writer.write_packet(hdr);
}
void application::local_actor_down(packet_writer& writer, actor_id id,
error reason) {
auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{system(), payload};
if (auto err = sink(reason))
CAF_RAISE_ERROR("unable to serialize an error");
auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::down_message,
static_cast<uint32_t>(payload.size()),
static_cast<uint64_t>(id)},
hdr);
writer.write_packet(hdr, payload);
}
expected<std::vector<byte>> application::serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<byte> result;
......@@ -78,59 +152,7 @@ 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) {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener));
buf_.clear();
serializer_impl<buffer_type> sink{&executor_, buf_};
if (auto err = sink(path)) {
CAF_LOG_ERROR("unable to serialize path");
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;
}
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,
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_};
const auto& src = ptr->msg->sender;
const auto& dst = ptr->receiver;
if (dst == nullptr) {
// TODO: valid?
return none;
}
if (src != nullptr) {
auto src_id = src->id();
system().registry().put(src_id, src);
if (auto err = sink(src->node(), src_id, dst->id(), ptr->msg->stages))
return err;
} else {
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()),
ptr->msg->mid.integer_value()};
auto bytes = to_bytes(hdr);
return write_packet(make_span(bytes), make_span(buf_));
}
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 +171,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 +181,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 +190,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 +220,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 +246,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,15 +279,15 @@ 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) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
CAF_ASSERT(hdr.type == message_type::resolve_request);
error application::handle_resolve_request(packet_writer& writer, header rec_hdr,
byte_span received) {
CAF_LOG_TRACE(CAF_ARG(rec_hdr) << CAF_ARG2("received.size", received.size()));
CAF_ASSERT(rec_hdr.type == message_type::resolve_request);
size_t path_size = 0;
binary_deserializer source{&executor_, payload};
binary_deserializer source{&executor_, received};
if (auto err = source.begin_sequence(path_size))
return err;
// We expect the payload to consist only of the path.
// We expect the received buffer to contain the path only.
if (path_size != source.remaining())
return ec::invalid_payload;
auto remainder = source.remainder();
......@@ -273,7 +295,6 @@ error application::handle_resolve_request(write_packet_callback& write_packet,
remainder.size()};
// Write result.
auto result = resolve_local_path(path);
buf_.clear();
actor_id aid;
std::set<std::string> ifs;
if (result) {
......@@ -283,20 +304,25 @@ 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_};
auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload};
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_);
auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::resolve_response,
static_cast<uint32_t>(payload.size()),
rec_hdr.operation_data},
hdr);
writer.write_packet(hdr, payload);
return none;
}
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);
error application::handle_resolve_response(packet_writer&, header received_hdr,
byte_span received) {
CAF_LOG_TRACE(CAF_ARG(received_hdr)
<< CAF_ARG2("received.size", received.size()));
CAF_ASSERT(received_hdr.type == message_type::resolve_response);
auto i = pending_resolves_.find(received_hdr.operation_data);
if (i == pending_resolves_.end()) {
CAF_LOG_ERROR("received unknown ID in resolve_response message");
return none;
......@@ -308,7 +334,7 @@ error application::handle_resolve_response(write_packet_callback&, header hdr,
});
actor_id aid;
std::set<std::string> ifs;
binary_deserializer source{&executor_, payload};
binary_deserializer source{&executor_, received};
if (auto err = source(aid, ifs))
return err;
if (aid == 0) {
......@@ -319,12 +345,14 @@ 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())
error application::handle_monitor_message(packet_writer& writer,
header received_hdr,
byte_span received) {
CAF_LOG_TRACE(CAF_ARG(received_hdr)
<< CAF_ARG2("received.size", received.size()));
if (!received.empty())
return ec::unexpected_payload;
auto aid = static_cast<actor_id>(hdr.operation_data);
auto aid = static_cast<actor_id>(received_hdr.operation_data);
auto hdl = system().registry().get(aid);
if (hdl != nullptr) {
endpoint_manager_ptr mgr = manager_;
......@@ -334,32 +362,34 @@ 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 payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload};
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_);
auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::down_message,
static_cast<uint32_t>(payload.size()),
received_hdr.operation_data},
hdr);
writer.write_packet(hdr, payload);
}
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 application::handle_down_message(packet_writer&, header received_hdr,
byte_span received) {
CAF_LOG_TRACE(CAF_ARG(received_hdr)
<< CAF_ARG2("received.size", received.size()));
error reason;
binary_deserializer source{&executor_, payload};
binary_deserializer source{&executor_, received};
if (auto err = source(reason))
return err;
proxies_.erase(peer_id_, hdr.operation_data, std::move(reason));
proxies_.erase(peer_id_, received_hdr.operation_data, std::move(reason));
return none;
}
error application::generate_handshake() {
buf_.clear();
serializer_impl<buffer_type> sink{&executor_, buf_};
error application::generate_handshake(std::vector<byte>& buf) {
serializer_impl<buffer_type> sink{&executor_, buf};
return sink(system().node(),
get_or(system().config(), "middleman.app-identifiers",
defaults::middleman::app_identifiers));
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 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/defaults.hpp"
namespace caf {
namespace defaults {
namespace middleman {
const size_t max_payload_buffers = 100;
const size_t max_header_buffers = 10;
} // namespace middleman
} // namespace defaults
} // namespace caf
......@@ -28,6 +28,18 @@ namespace caf {
namespace net {
namespace basp {
namespace {
void to_bytes_impl(const header& x, byte* ptr) {
*ptr = static_cast<byte>(x.type);
auto payload_len = detail::to_network_order(x.payload_len);
memcpy(ptr + 1, &payload_len, sizeof(payload_len));
auto operation_data = detail::to_network_order(x.operation_data);
memcpy(ptr + 5, &operation_data, sizeof(operation_data));
}
} // namespace
int header::compare(header other) const noexcept {
auto x = to_bytes(*this);
auto y = to_bytes(other);
......@@ -47,16 +59,16 @@ header header::from_bytes(span<const byte> bytes) {
}
std::array<byte, header_size> to_bytes(header x) {
std::array<byte, header_size> result;
auto ptr = result.data();
*ptr = static_cast<byte>(x.type);
auto payload_len = detail::to_network_order(x.payload_len);
memcpy(ptr + 1, &payload_len, sizeof(payload_len));
auto operation_data = detail::to_network_order(x.operation_data);
memcpy(ptr + 5, &operation_data, sizeof(operation_data));
std::array<byte, header_size> result{};
to_bytes_impl(x, result.data());
return result;
}
void to_bytes(header x, std::vector<byte>& buf) {
buf.resize(header_size);
to_bytes_impl(x, buf.data());
}
} // namespace basp
} // 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. *
******************************************************************************/
#include "caf/net/packet_writer.hpp"
namespace caf {
namespace net {
packet_writer::~packet_writer() {
// nop
}
} // namespace net
} // namespace caf
......@@ -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_payload_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;
......
......@@ -71,10 +71,10 @@ public:
return none;
}
template <class Transport>
void write_message(Transport& transport,
std::unique_ptr<endpoint_manager_queue::message> msg) {
transport.write_packet(span<byte>{}, msg->payload);
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> ptr) {
parent.write_packet(ptr->payload);
}
template <class Parent>
......
......@@ -43,6 +43,8 @@ using namespace caf::policy;
namespace {
using buffer_type = std::vector<byte>;
struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() {
mpx = std::make_shared<multiplexer>();
......@@ -98,13 +100,15 @@ public:
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
std::unique_ptr<endpoint_manager_queue::message> ptr) {
// Ignore proxy announcement messages.
if (msg->msg == nullptr)
if (ptr->msg == nullptr)
return;
header_type header{static_cast<uint32_t>(msg->payload.size())};
std::vector<byte> payload(msg->payload.begin(), msg->payload.end());
parent.write_packet(as_bytes(make_span(&header, 1)), make_span(payload));
auto header_buf = parent.next_header_buffer();
serializer_impl<buffer_type> sink{sys_, header_buf};
header_type header{static_cast<uint32_t>(ptr->payload.size())};
sink(header);
parent.write_packet(header_buf, ptr->payload);
}
static expected<std::vector<byte>> serialize(actor_system& sys,
......@@ -147,7 +151,8 @@ public:
} else {
if (data.size() != sizeof(header_type))
CAF_FAIL("");
memcpy(&header_, data.data(), sizeof(header_type));
binary_deserializer source{nullptr, data};
source(header_);
if (header_.payload == 0)
Base::handle_packet(parent, header_, span<const byte>{});
else
......@@ -187,8 +192,8 @@ public:
// nop
}
void handle_error(sec) {
// nop
void handle_error(sec sec) {
CAF_FAIL("handle_error called: " << CAF_ARG(sec));
}
private:
......
......@@ -37,6 +37,8 @@ using namespace caf::net;
namespace {
using buffer_type = std::vector<byte>;
constexpr string_view hello_test = "hello test!";
struct application_result {
......@@ -72,7 +74,8 @@ public:
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
parent.write_packet(span<const byte>{}, msg->payload);
auto header_buffer = parent.next_header_buffer();
parent.write_packet(header_buffer, msg->payload);
}
template <class Parent>
......@@ -118,16 +121,29 @@ public:
using application_type = dummy_application;
dummy_transport(std::shared_ptr<transport_result> res) : res_(res) {
dummy_transport(std::shared_ptr<transport_result> res)
: res_(std::move(res)) {
// nop
}
void write_packet(span<const byte> header, span<const byte> payload,
ip_endpoint ep) {
auto& buf = res_->packet_buffer;
buf.insert(buf.begin(), header.begin(), header.end());
buf.insert(buf.begin(), payload.begin(), payload.end());
void write_packet(ip_endpoint ep, span<buffer_type*> buffers) {
res_->ep = ep;
auto& packet_buf = res_->packet_buffer;
packet_buf.clear();
for (auto buf : buffers)
packet_buf.insert(packet_buf.end(), buf->begin(), buf->end());
}
transport_type& transport() {
return *this;
}
std::vector<byte> next_header_buffer() {
return {};
}
std::vector<byte> next_payload_buffer() {
return {};
}
private:
......
......@@ -33,6 +33,8 @@ using namespace caf::net;
namespace {
using buffer_type = std::vector<byte>;
constexpr string_view hello_test = "hello_test";
struct dummy_actor : public monitorable_actor {
......@@ -47,7 +49,7 @@ struct dummy_actor : public monitorable_actor {
class dummy_application {
public:
dummy_application(std::shared_ptr<std::vector<byte>> rec_buf, uint8_t id)
dummy_application(std::shared_ptr<buffer_type> rec_buf, uint8_t id)
: rec_buf_(std::move(rec_buf)),
id_(id){
// nop
......@@ -61,11 +63,12 @@ public:
return none;
}
template <class Transport>
void write_message(Transport& transport,
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
rec_buf_->push_back(static_cast<byte>(id_));
transport.write_packet(span<byte>{}, make_span(msg->payload));
auto header_buf = parent.next_header_buffer();
parent.write_packet(header_buf, msg->payload);
}
template <class Parent>
......@@ -88,13 +91,13 @@ public:
rec_buf_->push_back(static_cast<byte>(id_));
}
static expected<std::vector<byte>> serialize(actor_system&,
static expected<buffer_type> serialize(actor_system&,
const type_erased_tuple&) {
return std::vector<byte>{};
return buffer_type{};
}
private:
std::shared_ptr<std::vector<byte>> rec_buf_;
std::shared_ptr<buffer_type> rec_buf_;
uint8_t id_;
};
......@@ -102,8 +105,8 @@ 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) {
dummy_application_factory(std::shared_ptr<buffer_type> buf)
: buf_(std::move(buf)), application_cnt_(0) {
// nop
}
......@@ -112,7 +115,7 @@ public:
}
private:
std::shared_ptr<std::vector<byte>> buf_;
std::shared_ptr<buffer_type> buf_;
uint8_t application_cnt_;
};
......@@ -121,22 +124,35 @@ struct dummy_transport {
using application_type = dummy_application;
dummy_transport(std::shared_ptr<std::vector<byte>> buf) : buf_(buf) {
dummy_transport(std::shared_ptr<buffer_type> buf) : buf_(std::move(buf)) {
// nop
}
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());
void write_packet(IdType, span<buffer_type*> buffers) {
for (auto buf : buffers)
buf_->insert(buf_->end(), buf->begin(), buf->end());
}
transport_type& transport() {
return *this;
}
buffer_type next_header_buffer() {
return {};
}
buffer_type next_payload_buffer() {
return {};
}
private:
std::shared_ptr<std::vector<byte>> buf_;
std::shared_ptr<buffer_type> buf_;
};
struct testdata {
testdata(uint8_t worker_id, node_id id, ip_endpoint ep)
: worker_id(worker_id), nid(id), ep(ep) {
: worker_id(worker_id), nid(std::move(id)), ep(ep) {
// nop
}
......@@ -168,7 +184,7 @@ struct fixture : host_fixture {
ip_endpoint>;
fixture()
: buf{std::make_shared<std::vector<byte>>()},
: buf{std::make_shared<buffer_type>()},
dispatcher{dummy_application_factory{buf}},
dummy{buf} {
add_new_workers();
......@@ -180,7 +196,7 @@ struct fixture : host_fixture {
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());
buffer_type 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),
......@@ -213,7 +229,7 @@ struct fixture : host_fixture {
actor_system_config cfg{};
actor_system sys{cfg};
std::shared_ptr<std::vector<byte>> buf;
std::shared_ptr<buffer_type> buf;
dispatcher_type dispatcher;
dummy_transport dummy;
......
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