Commit 55ab269d authored by Jakob Otto's avatar Jakob Otto

Include review feedback

parent 60c7e61e
...@@ -76,14 +76,14 @@ public: ...@@ -76,14 +76,14 @@ public:
if (!std::is_base_of<test_tag, Parent>::value) if (!std::is_base_of<test_tag, Parent>::value)
manager_ = &parent.manager(); manager_ = &parent.manager();
// Write handshake. // Write handshake.
auto header_buf = parent.next_header_buffer(); auto hdr = parent.next_header_buffer();
auto payload_buf = parent.next_buffer(); auto payload = parent.next_buffer();
if (auto err = generate_handshake(payload_buf)) if (auto err = generate_handshake(payload))
return err; return err;
to_bytes(header{message_type::handshake, to_bytes(header{message_type::handshake,
static_cast<uint32_t>(payload_buf.size()), version}, static_cast<uint32_t>(payload.size()), version},
header_buf); hdr);
parent.write_packet(header_buf, payload_buf); parent.write_packet(hdr, payload);
parent.transport().configure_read(receive_policy::exactly(header_size)); parent.transport().configure_read(receive_policy::exactly(header_size));
return none; return none;
} }
...@@ -146,17 +146,17 @@ private: ...@@ -146,17 +146,17 @@ private:
error handle_actor_message(packet_writer& writer, header hdr, error handle_actor_message(packet_writer& writer, header hdr,
byte_span payload); byte_span payload);
error handle_resolve_request(packet_writer& writer, header hdr, error handle_resolve_request(packet_writer& writer, header rec_hdr,
byte_span payload); byte_span received);
error handle_resolve_response(packet_writer& writer, header hdr, error handle_resolve_response(packet_writer& writer, header received_hdr,
byte_span payload); byte_span received);
error handle_monitor_message(packet_writer& writer, header hdr, error handle_monitor_message(packet_writer& writer, header received_hdr,
byte_span payload); byte_span received);
error handle_down_message(packet_writer& writer, header hdr, error handle_down_message(packet_writer& writer, header received_hdr,
byte_span payload); byte_span received);
/// Writes the handshake payload to `buf_`. /// Writes the handshake payload to `buf_`.
error generate_handshake(buffer_type& buf); error generate_handshake(buffer_type& buf);
......
...@@ -29,7 +29,7 @@ namespace net { ...@@ -29,7 +29,7 @@ namespace net {
/// Implements the interface for transport and application policies and /// Implements the interface for transport and application policies and
/// dispatches member functions either to `object` or `parent`. /// dispatches member functions either to `object` or `parent`.
template <class Object, class Parent> template <class Object, class Parent>
class packet_writer_impl final : public packet_writer { class packet_writer_decorator final : public packet_writer {
public: public:
// -- member types ----------------------------------------------------------- // -- member types -----------------------------------------------------------
...@@ -39,7 +39,7 @@ public: ...@@ -39,7 +39,7 @@ public:
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
packet_writer_impl(Object& object, Parent& parent) packet_writer_decorator(Object& object, Parent& parent)
: object_(object), parent_(parent) { : object_(object), parent_(parent) {
// nop // nop
} }
...@@ -88,8 +88,8 @@ private: ...@@ -88,8 +88,8 @@ private:
}; };
template <class Object, class Parent> template <class Object, class Parent>
packet_writer_impl<Object, Parent> make_packet_writer_impl(Object& object, packet_writer_decorator<Object, Parent>
Parent& parent) { make_packet_writer_decorator(Object& object, Parent& parent) {
return {object, parent}; return {object, parent};
} }
......
...@@ -238,40 +238,41 @@ private: ...@@ -238,40 +238,41 @@ private:
// -- private member functions ----------------------------------------------- // -- private member functions -----------------------------------------------
bool write_some() { bool write_some() {
auto begin = [&]() { return write_queue_.begin(); }; CAF_LOG_TRACE(CAF_ARG(handle_.id));
// helper to sort empty buffers back into the right queues // helper to sort empty buffers back into the right queues
auto recycle = [&]() { auto recycle = [&]() {
auto is_header = [](std::pair<bool, buffer_type>& p) { return p.first; }; auto& front = write_queue_.front();
begin()->second.clear(); auto& is_header = front.first;
if (is_header(*begin())) auto& buf = front.second;
free_header_bufs_.emplace_back(std::move(begin()->second)); written_ = 0;
buf.clear();
if (is_header)
free_header_bufs_.emplace_back(std::move(buf));
else else
free_bufs_.emplace_back(std::move(begin()->second)); free_bufs_.emplace_back(std::move(buf));
write_queue_.pop_front(); write_queue_.pop_front();
}; };
// nothing to write // nothing to write
if (write_queue_.empty()) if (write_queue_.empty())
return false; return false;
do { do {
if (begin()->second.empty()) { auto& buf = write_queue_.front().second;
if (buf.empty()) {
recycle(); recycle();
continue; continue;
} }
// get size of send buffer auto data = buf.data() + written_;
auto ret = send_buffer_size(handle_); auto len = buf.size() - written_;
if (!ret) { auto write_ret = write(handle_, make_span(data, len));
CAF_LOG_ERROR("send_buffer_size returned an error" << CAF_ARG(ret));
return false;
}
// is send buffer of socket full?
if (begin()->second.size() > *ret)
return true;
CAF_LOG_TRACE(CAF_ARG(handle_.id));
auto write_ret = write(handle_, make_span(begin()->second));
if (auto num_bytes = get_if<size_t>(&write_ret)) { if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(handle_.id) << CAF_ARG(*num_bytes)); CAF_LOG_DEBUG(CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
if (*num_bytes >= begin()->second.size()) if (*num_bytes + written_ >= buf.size()) {
recycle(); recycle();
written_ = 0;
} else {
written_ = *num_bytes;
return false;
}
} else { } else {
auto err = get<sec>(write_ret); auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) { if (err != sec::unavailable_or_would_block) {
...@@ -279,6 +280,7 @@ private: ...@@ -279,6 +280,7 @@ private:
worker_.handle_error(err); worker_.handle_error(err);
return false; return false;
} }
return true;
} }
} while (!write_queue_.empty()); } while (!write_queue_.empty());
return false; return false;
......
...@@ -22,7 +22,7 @@ ...@@ -22,7 +22,7 @@
#include "caf/ip_endpoint.hpp" #include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_impl.hpp" #include "caf/net/packet_writer_decorator.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
#include "caf/unit.hpp" #include "caf/unit.hpp"
...@@ -64,45 +64,45 @@ public: ...@@ -64,45 +64,45 @@ public:
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
auto writer = make_packet_writer_impl(*this, parent); auto writer = make_packet_writer_decorator(*this, parent);
return application_.init(writer); return application_.init(writer);
} }
template <class Parent> template <class Parent>
error handle_data(Parent& parent, span<const byte> data) { error handle_data(Parent& parent, span<const byte> data) {
auto writer = make_packet_writer_impl(*this, parent); auto writer = make_packet_writer_decorator(*this, parent);
return application_.handle_data(writer, data); return application_.handle_data(writer, data);
} }
template <class Parent> template <class Parent>
void write_message(Parent& parent, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) { std::unique_ptr<endpoint_manager_queue::message> msg) {
auto writer = make_packet_writer_impl(*this, parent); auto writer = make_packet_writer_decorator(*this, parent);
application_.write_message(writer, std::move(msg)); application_.write_message(writer, std::move(msg));
} }
template <class Parent> template <class Parent>
void resolve(Parent& parent, string_view path, const actor& listener) { void resolve(Parent& parent, string_view path, const actor& listener) {
auto writer = make_packet_writer_impl(*this, parent); auto writer = make_packet_writer_decorator(*this, parent);
application_.resolve(writer, path, listener); application_.resolve(writer, path, listener);
} }
template <class Parent> template <class Parent>
void new_proxy(Parent& parent, const node_id&, actor_id id) { void new_proxy(Parent& parent, const node_id&, actor_id id) {
auto writer = make_packet_writer_impl(*this, parent); auto writer = make_packet_writer_decorator(*this, parent);
application_.new_proxy(writer, id); application_.new_proxy(writer, id);
} }
template <class Parent> template <class Parent>
void local_actor_down(Parent& parent, const node_id&, actor_id id, void local_actor_down(Parent& parent, const node_id&, actor_id id,
error reason) { error reason) {
auto writer = make_packet_writer_impl(*this, parent); auto writer = make_packet_writer_decorator(*this, parent);
application_.local_actor_down(writer, id, std::move(reason)); application_.local_actor_down(writer, id, std::move(reason));
} }
template <class Parent> template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) { void timeout(Parent& parent, atom_value value, uint64_t id) {
auto writer = make_packet_writer_impl(*this, parent); auto writer = make_packet_writer_decorator(*this, parent);
application_.timeout(writer, value, id); application_.timeout(writer, value, id);
} }
......
...@@ -24,7 +24,7 @@ ...@@ -24,7 +24,7 @@
#include "caf/ip_endpoint.hpp" #include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_impl.hpp" #include "caf/net/packet_writer_decorator.hpp"
#include "caf/net/transport_worker.hpp" #include "caf/net/transport_worker.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
#include "caf/unit.hpp" #include "caf/unit.hpp"
......
...@@ -53,8 +53,8 @@ error application::write_message( ...@@ -53,8 +53,8 @@ error application::write_message(
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->msg != nullptr); CAF_ASSERT(ptr->msg != nullptr);
CAF_LOG_TRACE(CAF_ARG2("content", ptr->msg->content())); CAF_LOG_TRACE(CAF_ARG2("content", ptr->msg->content()));
auto payload_elem_buf = writer.next_buffer(); auto payload_prefix = writer.next_buffer();
serializer_impl<buffer_type> sink{system(), payload_elem_buf}; serializer_impl<buffer_type> sink{system(), payload_prefix};
const auto& src = ptr->msg->sender; const auto& src = ptr->msg->sender;
const auto& dst = ptr->receiver; const auto& dst = ptr->receiver;
if (dst == nullptr) { if (dst == nullptr) {
...@@ -70,57 +70,55 @@ error application::write_message( ...@@ -70,57 +70,55 @@ error application::write_message(
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages)) if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages))
return err; return err;
} }
// TODO: Is this size correct? auto hdr = writer.next_header_buffer();
auto header_buf = writer.next_header_buffer();
to_bytes(header{message_type::actor_message, to_bytes(header{message_type::actor_message,
static_cast<uint32_t>(payload_elem_buf.size() static_cast<uint32_t>(payload_prefix.size()
+ ptr->payload.size()), + ptr->payload.size()),
ptr->msg->mid.integer_value()}, ptr->msg->mid.integer_value()},
header_buf); hdr);
// TODO: OK to move payload out of message? writer.write_packet(hdr, payload_prefix, ptr->payload);
writer.write_packet(header_buf, payload_elem_buf, ptr->payload);
return none; return none;
} }
void application::resolve(packet_writer& writer, string_view path, void application::resolve(packet_writer& writer, string_view path,
const actor& listener) { const actor& listener) {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener)); CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener));
auto payload_buf = writer.next_buffer(); auto payload = writer.next_buffer();
serializer_impl<buffer_type> sink{&executor_, payload_buf}; serializer_impl<buffer_type> sink{&executor_, payload};
if (auto err = sink(path)) { if (auto err = sink(path)) {
CAF_LOG_ERROR("unable to serialize path" << CAF_ARG(err)); CAF_LOG_ERROR("unable to serialize path" << CAF_ARG(err));
return; return;
} }
auto req_id = next_request_id_++; auto req_id = next_request_id_++;
auto header_buf = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::resolve_request, to_bytes(header{message_type::resolve_request,
static_cast<uint32_t>(payload_buf.size()), req_id}, static_cast<uint32_t>(payload.size()), req_id},
header_buf); hdr);
writer.write_packet(header_buf, payload_buf); writer.write_packet(hdr, payload);
response_promise rp{nullptr, actor_cast<strong_actor_ptr>(listener), response_promise rp{nullptr, actor_cast<strong_actor_ptr>(listener),
no_stages, make_message_id()}; no_stages, make_message_id()};
pending_resolves_.emplace(req_id, std::move(rp)); pending_resolves_.emplace(req_id, std::move(rp));
} }
void application::new_proxy(packet_writer& writer, actor_id id) { void application::new_proxy(packet_writer& writer, actor_id id) {
auto header_buf = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::monitor_message, 0, static_cast<uint64_t>(id)}, to_bytes(header{message_type::monitor_message, 0, static_cast<uint64_t>(id)},
header_buf); hdr);
writer.write_packet(header_buf); writer.write_packet(hdr);
} }
void application::local_actor_down(packet_writer& writer, actor_id id, void application::local_actor_down(packet_writer& writer, actor_id id,
error reason) { error reason) {
auto payload_buf = writer.next_buffer(); auto payload = writer.next_buffer();
serializer_impl<buffer_type> sink{system(), payload_buf}; serializer_impl<buffer_type> sink{system(), payload};
if (auto err = sink(reason)) if (auto err = sink(reason))
CAF_RAISE_ERROR("unable to serialize an error"); CAF_RAISE_ERROR("unable to serialize an error");
auto header_buf = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::down_message, to_bytes(header{message_type::down_message,
static_cast<uint32_t>(payload_buf.size()), static_cast<uint32_t>(payload.size()),
static_cast<uint64_t>(id)}, static_cast<uint64_t>(id)},
header_buf); hdr);
writer.write_packet(header_buf, payload_buf); writer.write_packet(hdr, payload);
} }
expected<std::vector<byte>> application::serialize(actor_system& sys, expected<std::vector<byte>> application::serialize(actor_system& sys,
...@@ -281,15 +279,15 @@ error application::handle_actor_message(packet_writer&, header hdr, ...@@ -281,15 +279,15 @@ error application::handle_actor_message(packet_writer&, header hdr,
return none; return none;
} }
error application::handle_resolve_request(packet_writer& writer, header hdr, error application::handle_resolve_request(packet_writer& writer, header rec_hdr,
byte_span payload) { byte_span received) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size())); CAF_LOG_TRACE(CAF_ARG(rec_hdr) << CAF_ARG2("received.size", received.size()));
CAF_ASSERT(hdr.type == message_type::resolve_request); CAF_ASSERT(rec_hdr.type == message_type::resolve_request);
size_t path_size = 0; size_t path_size = 0;
binary_deserializer source{&executor_, payload}; binary_deserializer source{&executor_, received};
if (auto err = source.begin_sequence(path_size)) if (auto err = source.begin_sequence(path_size))
return err; return err;
// We expect the payload to consist only of the path. // We expect the received to consist only of the path.
if (path_size != source.remaining()) if (path_size != source.remaining())
return ec::invalid_payload; return ec::invalid_payload;
auto remainder = source.remainder(); auto remainder = source.remainder();
...@@ -306,24 +304,25 @@ error application::handle_resolve_request(packet_writer& writer, header hdr, ...@@ -306,24 +304,25 @@ error application::handle_resolve_request(packet_writer& writer, header hdr,
aid = 0; aid = 0;
} }
// TODO: figure out how to obtain messaging interface. // TODO: figure out how to obtain messaging interface.
auto payload_buf = writer.next_buffer(); auto payload = writer.next_buffer();
serializer_impl<buffer_type> sink{&executor_, payload_buf}; serializer_impl<buffer_type> sink{&executor_, payload};
if (auto err = sink(aid, ifs)) if (auto err = sink(aid, ifs))
return err; return err;
auto header_buf = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::resolve_response, to_bytes(header{message_type::resolve_response,
static_cast<uint32_t>(payload_buf.size()), static_cast<uint32_t>(payload.size()),
hdr.operation_data}, rec_hdr.operation_data},
header_buf); hdr);
writer.write_packet(header_buf, payload_buf); writer.write_packet(hdr, payload);
return none; return none;
} }
error application::handle_resolve_response(packet_writer&, header hdr, error application::handle_resolve_response(packet_writer&, header received_hdr,
byte_span payload) { byte_span received) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size())); CAF_LOG_TRACE(CAF_ARG(received_hdr)
CAF_ASSERT(hdr.type == message_type::resolve_response); << CAF_ARG2("received.size", received.size()));
auto i = pending_resolves_.find(hdr.operation_data); CAF_ASSERT(received_hdr.type == message_type::resolve_response);
auto i = pending_resolves_.find(received_hdr.operation_data);
if (i == pending_resolves_.end()) { if (i == pending_resolves_.end()) {
CAF_LOG_ERROR("received unknown ID in resolve_response message"); CAF_LOG_ERROR("received unknown ID in resolve_response message");
return none; return none;
...@@ -335,7 +334,7 @@ error application::handle_resolve_response(packet_writer&, header hdr, ...@@ -335,7 +334,7 @@ error application::handle_resolve_response(packet_writer&, header hdr,
}); });
actor_id aid; actor_id aid;
std::set<std::string> ifs; std::set<std::string> ifs;
binary_deserializer source{&executor_, payload}; binary_deserializer source{&executor_, received};
if (auto err = source(aid, ifs)) if (auto err = source(aid, ifs))
return err; return err;
if (aid == 0) { if (aid == 0) {
...@@ -346,12 +345,14 @@ error application::handle_resolve_response(packet_writer&, header hdr, ...@@ -346,12 +345,14 @@ error application::handle_resolve_response(packet_writer&, header hdr,
return none; return none;
} }
error application::handle_monitor_message(packet_writer& writer, header hdr, error application::handle_monitor_message(packet_writer& writer,
byte_span payload) { header received_hdr,
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size())); byte_span received) {
if (!payload.empty()) CAF_LOG_TRACE(CAF_ARG(received_hdr)
<< CAF_ARG2("received.size", received.size()));
if (!received.empty())
return ec::unexpected_payload; 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); auto hdl = system().registry().get(aid);
if (hdl != nullptr) { if (hdl != nullptr) {
endpoint_manager_ptr mgr = manager_; endpoint_manager_ptr mgr = manager_;
...@@ -361,28 +362,29 @@ error application::handle_monitor_message(packet_writer& writer, header hdr, ...@@ -361,28 +362,29 @@ error application::handle_monitor_message(packet_writer& writer, header hdr,
}); });
} else { } else {
error reason = exit_reason::unknown; error reason = exit_reason::unknown;
auto payload_buf = writer.next_buffer(); auto payload = writer.next_buffer();
serializer_impl<buffer_type> sink{&executor_, payload_buf}; serializer_impl<buffer_type> sink{&executor_, payload};
if (auto err = sink(reason)) if (auto err = sink(reason))
return err; return err;
auto header_buf = writer.next_header_buffer(); auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::down_message, to_bytes(header{message_type::down_message,
static_cast<uint32_t>(payload_buf.size()), static_cast<uint32_t>(payload.size()),
hdr.operation_data}, received_hdr.operation_data},
header_buf); hdr);
writer.write_packet(header_buf, payload_buf); writer.write_packet(hdr, payload);
} }
return none; return none;
} }
error application::handle_down_message(packet_writer&, header hdr, error application::handle_down_message(packet_writer&, header received_hdr,
byte_span payload) { byte_span received) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size())); CAF_LOG_TRACE(CAF_ARG(received_hdr)
<< CAF_ARG2("received.size", received.size()));
error reason; error reason;
binary_deserializer source{&executor_, payload}; binary_deserializer source{&executor_, received};
if (auto err = source(reason)) if (auto err = source(reason))
return err; 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; return none;
} }
......
...@@ -28,6 +28,18 @@ namespace caf { ...@@ -28,6 +28,18 @@ namespace caf {
namespace net { namespace net {
namespace basp { 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 { int header::compare(header other) const noexcept {
auto x = to_bytes(*this); auto x = to_bytes(*this);
auto y = to_bytes(other); auto y = to_bytes(other);
...@@ -46,14 +58,6 @@ header header::from_bytes(span<const byte> bytes) { ...@@ -46,14 +58,6 @@ header header::from_bytes(span<const byte> bytes) {
return result; return result;
} }
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));
}
std::array<byte, header_size> to_bytes(header x) { std::array<byte, header_size> to_bytes(header x) {
std::array<byte, header_size> result{}; std::array<byte, header_size> result{};
to_bytes_impl(x, result.data()); to_bytes_impl(x, result.data());
......
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