Commit b068df71 authored by Jakob Otto's avatar Jakob Otto

Reuse buffers for serializing

parent 3cdf5f18
......@@ -136,7 +136,8 @@ public:
// nop
}
static expected<buffer_type> serialize(actor_system& sys, const message& x);
static error_code<sec> serialize(actor_system& sys, const message& x,
std::vector<byte>& buf);
// -- utility functions ------------------------------------------------------
......
......@@ -44,11 +44,9 @@ public:
using super = socket_manager;
/// Represents either an error or a serialized payload.
using maybe_buffer = expected<std::vector<byte>>;
/// A function type for serializing message payloads.
using serialize_fun_type = maybe_buffer (*)(actor_system&, const message&);
using serialize_fun_type = error_code<sec> (*)(actor_system&, const message&,
std::vector<byte>& buf);
// -- constructors, destructors, and assignment operators --------------------
......
......@@ -60,9 +60,9 @@ error application::write_message(
// TODO: valid?
return none;
}
auto payload_buf = serialize(system(), ptr->msg->content());
if (!payload_buf)
return payload_buf.error();
auto payload_buf = writer.next_payload_buffer();
if (auto err = serialize(system(), ptr->msg->content(), payload_buf))
return err;
if (src != nullptr) {
auto src_id = src->id();
system().registry().put(src_id, src);
......@@ -75,10 +75,10 @@ error application::write_message(
auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::actor_message,
static_cast<uint32_t>(payload_prefix_buf.size()
+ payload_buf->size()),
+ payload_buf.size()),
ptr->msg->mid.integer_value()},
hdr);
writer.write_packet(hdr, payload_prefix_buf, *payload_buf);
writer.write_packet(hdr, payload_prefix_buf, payload_buf);
return none;
}
......@@ -121,13 +121,10 @@ void application::local_actor_down(packet_writer& writer, actor_id id,
writer.write_packet(hdr, payload);
}
expected<std::vector<byte>> application::serialize(actor_system& sys,
const message& x) {
std::vector<byte> result;
binary_serializer sink{sys, result};
if (auto err = x.save(sink))
return err.value();
return result;
error_code<sec> application::serialize(actor_system& sys, const message& x,
std::vector<byte>& buf) {
binary_serializer sink{sys, buf};
return x.save(sink);
}
strong_actor_ptr application::resolve_local_path(string_view path) {
......
......@@ -119,13 +119,13 @@ 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) {
if (auto payload = serialize(transport.system(), msg->msg->payload))
transport.write_packet(*payload);
else
CAF_FAIL("serializing failed: " << payload.error());
auto payload_buf = parent.next_payload_buffer();
if (auto err = serialize(parent.system(), msg->msg->payload, payload_buf))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(payload_buf);
}
template <class Parent>
......@@ -167,12 +167,10 @@ public:
CAF_FAIL("handle_error called: " << to_string(sec));
}
static expected<buffer_type> serialize(actor_system& sys, const message& x) {
buffer_type result;
binary_serializer sink{sys, result};
if (auto err = x.save(sink))
return err.value();
return result;
static error_code<sec> serialize(actor_system& sys, const message& x,
std::vector<byte>& buf) {
binary_serializer sink{sys, buf};
return x.save(sink);
}
private:
......
......@@ -60,13 +60,10 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
class dummy_application {
public:
static expected<std::vector<byte>> serialize(actor_system& sys,
const message& x) {
std::vector<byte> result;
binary_serializer sink{sys, result};
if (auto err = x.save(sink))
return err.value();
return result;
static error_code<sec> serialize(actor_system& sys, const message& x,
std::vector<byte>& buf) {
binary_serializer sink{sys, buf};
return x.save(sink);
}
template <class Parent>
......@@ -74,13 +71,13 @@ 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) {
if (auto payload = serialize(transport.system(), msg->msg->payload))
transport.write_packet(*payload);
else
CAF_FAIL("serializing failed: " << payload.error());
auto payload_buf = parent.next_payload_buffer();
if (auto err = serialize(parent.system(), msg->msg->payload, payload_buf))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(payload_buf);
}
template <class Parent>
......@@ -119,9 +116,9 @@ class dummy_application_factory {
public:
using application_type = dummy_application;
static expected<std::vector<byte>> serialize(actor_system& sys,
const message& x) {
return dummy_application::serialize(sys, x);
static error_code<sec> serialize(actor_system& sys, const message& x,
std::vector<byte>& buf) {
return dummy_application::serialize(sys, x, buf);
}
template <class Parent>
......
......@@ -62,13 +62,10 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
class dummy_application {
public:
static expected<std::vector<byte>> serialize(actor_system& sys,
const message& x) {
std::vector<byte> result;
binary_serializer sink{sys, result};
if (auto err = x.save(sink))
return err.value();
return result;
static error_code<sec> serialize(actor_system& sys, const message& x,
std::vector<byte>& buf) {
binary_serializer sink{sys, buf};
return x.save(sink);
}
};
......@@ -106,12 +103,12 @@ public:
template <class Manager>
bool handle_write_event(Manager& mgr) {
auto sf = mgr.serialize_fun();
for (auto x = mgr.next_message(); x != nullptr; x = mgr.next_message()) {
if (auto payload = application_type::serialize(mgr.system(),
x->msg->payload))
buf_.insert(buf_.end(), payload->begin(), payload->end());
else
CAF_FAIL("serializing failed: " << payload.error());
std::vector<byte> payload_buf;
if (auto err = sf(mgr.system(), x->msg->payload, payload_buf))
CAF_FAIL("serializing failed: " << err);
buf_.insert(buf_.end(), payload_buf.begin(), payload_buf.end());
}
auto res = write(handle_, buf_);
if (auto num_bytes = get_if<size_t>(&res)) {
......
......@@ -90,13 +90,13 @@ 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) {
if (auto payload = serialize(transport.system(), msg->msg->payload))
transport.write_packet(*payload);
else
CAF_FAIL("serializing failed: " << payload.error());
auto payload_buf = parent.next_payload_buffer();
if (auto err = serialize(parent.system(), msg->msg->payload, payload_buf))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(payload_buf);
}
template <class Parent>
......@@ -138,12 +138,10 @@ public:
// nop
}
static expected<buffer_type> serialize(actor_system& sys, const message& x) {
buffer_type result;
binary_serializer sink{sys, result};
if (auto err = x.save(sink))
return err.value();
return result;
static error_code<sec> serialize(actor_system& sys, const message& x,
std::vector<byte>& buf) {
binary_serializer sink{sys, buf};
return x.save(sink);
}
private:
......
......@@ -105,23 +105,20 @@ public:
if (ptr->msg == nullptr)
return;
auto header_buf = parent.next_header_buffer();
auto payload_buf = serialize(parent.system(), ptr->msg->payload);
if (!payload_buf)
CAF_FAIL("serializing failed: " << payload_buf.error());
auto payload_buf = parent.next_payload_buffer();
if (auto err = serialize(parent.system(), ptr->msg->payload, payload_buf))
CAF_FAIL("serializing failed: " << err);
binary_serializer sink{sys_, header_buf};
header_type header{static_cast<uint32_t>(payload_buf->size())};
header_type header{static_cast<uint32_t>(payload_buf.size())};
if (auto err = sink(header))
CAF_FAIL("serializing failed: " << err);
parent.write_packet(header_buf, *payload_buf);
parent.write_packet(header_buf, payload_buf);
}
static expected<std::vector<byte>> serialize(actor_system& sys,
const message& x) {
std::vector<byte> result;
binary_serializer sink{sys, result};
if (auto err = x.save(sink))
return err.value();
return result;
static error_code<sec> serialize(actor_system& sys, const message& x,
std::vector<byte>& buf) {
binary_serializer sink{sys, buf};
return x.save(sink);
}
private:
......
......@@ -289,7 +289,8 @@ CAF_TEST(handle_data) {
CHECK_HANDLE_DATA(test_data.at(3));
}
/*CAF_TEST(write_message write_packet) {
/*
CAF_TEST(write_message write_packet) {
CHECK_WRITE_MESSAGE(test_data.at(0));
CHECK_WRITE_MESSAGE(test_data.at(1));
CHECK_WRITE_MESSAGE(test_data.at(2));
......
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