Commit 827a1952 authored by Dominik Charousset's avatar Dominik Charousset

Port length-prefix framing to latest layering API

parent 7b67c79e
...@@ -12,6 +12,7 @@ ...@@ -12,6 +12,7 @@
#include "caf/byte_span.hpp" #include "caf/byte_span.hpp"
#include "caf/detail/network_order.hpp" #include "caf/detail/network_order.hpp"
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/net/message_oriented_layer_ptr.hpp"
#include "caf/net/receive_policy.hpp" #include "caf/net/receive_policy.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/span.hpp" #include "caf/span.hpp"
...@@ -34,7 +35,9 @@ public: ...@@ -34,7 +35,9 @@ public:
using length_prefix_type = uint32_t; using length_prefix_type = uint32_t;
static constexpr size_t max_message_length = INT32_MAX; static constexpr size_t max_message_length = INT32_MAX - sizeof(uint32_t);
static constexpr uint32_t default_receive_size = 4 * 1024; // 4kb.
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
...@@ -44,107 +47,140 @@ public: ...@@ -44,107 +47,140 @@ public:
// nop // nop
} }
// -- interface for the upper layer ------------------------------------------ // -- initialization ---------------------------------------------------------
template <class LowerLayer> template <class LowerLayerPtr>
class access { error init(socket_manager* owner, LowerLayerPtr down, const settings& cfg) {
public: down->configure_read(
access(LowerLayer* lower_layer, length_prefix_framing* this_layer) receive_policy::between(sizeof(uint32_t), default_receive_size));
: lower_layer_(lower_layer), this_layer_(this_layer) { return upper_layer_.init(owner, this_layer_ptr(down), cfg);
// nop }
// -- properties -------------------------------------------------------------
auto& upper_layer() noexcept {
return upper_layer_;
} }
void begin_message() { const auto& upper_layer() const noexcept {
lower_layer_->begin_output(); return upper_layer_;
auto& buf = message_buffer(); }
// -- interface for the upper layer ------------------------------------------
template <class LowerLayerPtr>
void begin_message(LowerLayerPtr down) {
down->begin_output();
auto& buf = down->output_buffer();
message_offset_ = buf.size(); message_offset_ = buf.size();
buf.insert(buf.end(), 4, byte{0}); buf.insert(buf.end(), 4, byte{0});
} }
byte_buffer& message_buffer() { template <class LowerLayerPtr>
return lower_layer_->output_buffer(); byte_buffer& message_buffer(LowerLayerPtr down) {
return down->output_buffer();
} }
bool end_message() { template <class LowerLayerPtr>
[[nodiscard]] bool end_message(LowerLayerPtr down) {
using detail::to_network_order; using detail::to_network_order;
auto& buf = message_buffer(); auto& buf = down->output_buffer();
CAF_ASSERT(message_offset_ < buf.size());
auto msg_begin = buf.begin() + message_offset_; auto msg_begin = buf.begin() + message_offset_;
auto msg_size = std::distance(msg_begin + 4, buf.end()); auto msg_size = std::distance(msg_begin + 4, buf.end());
if (msg_size > 0 && msg_size < max_message_length) { if (msg_size > 0 && static_cast<size_t>(msg_size) < max_message_length) {
auto u32_size = to_network_order(static_cast<uint32_t>(msg_size)); auto u32_size = to_network_order(static_cast<uint32_t>(msg_size));
memcpy(std::addressof(*msg_begin), &u32_size, 4); memcpy(std::addressof(*msg_begin), &u32_size, 4);
return true; return true;
} else { } else {
abort_reason(make_error( auto err = make_error(sec::runtime_error,
sec::runtime_error, msg_size == 0 ? "logic error: message of size 0" msg_size == 0 ? "logic error: message of size 0"
: "maximum message size exceeded")); : "maximum message size exceeded");
down->abort_reason(std::move(err));
return false; return false;
} }
} }
bool can_send_more() const noexcept { // -- interface for the lower layer ------------------------------------------
return lower_layer_->can_send_more();
}
void abort_reason(error reason) { template <class LowerLayerPtr>
return lower_layer_->abort_reason(std::move(reason)); bool prepare_send(LowerLayerPtr down) {
return upper_layer_.prepare_send(this_layer_ptr(down));
} }
void configure_read(receive_policy policy) { template <class LowerLayerPtr>
lower_layer_->configure_read(policy); bool done_sending(LowerLayerPtr down) {
return upper_layer_.done_sending(this_layer_ptr(down));
} }
private: template <class LowerLayerPtr>
LowerLayer* lower_layer_; void abort(LowerLayerPtr down, const error& reason) {
length_prefix_framing* this_layer_; upper_layer_.abort(this_layer_ptr(down), reason);
size_t message_offset_ = 0;
};
// -- properties -------------------------------------------------------------
auto& upper_layer() noexcept {
return upper_layer_;
} }
const auto& upper_layer() const noexcept { template <class LowerLayerPtr>
return upper_layer_; ptrdiff_t consume(LowerLayerPtr down, byte_span input, byte_span) {
auto buffer = input;
auto consumed = ptrdiff_t{0};
auto this_layer = this_layer_ptr(down);
for (;;) {
if (input.size() < sizeof(uint32_t)) {
return consumed;
} else {
auto [msg_size, sub_buffer] = split(input);
if (msg_size == 0) {
consumed += static_cast<ptrdiff_t>(sizeof(uint32_t));
input = sub_buffer;
} else if (msg_size > max_message_length) {
auto err = make_error(sec::runtime_error,
"maximum message size exceeded");
down->abort_reason(std::move(err));
return -1;
} else if (msg_size > sub_buffer.size()) {
if (msg_size + sizeof(uint32_t) > receive_buf_upper_bound_) {
auto min_read_size = static_cast<uint32_t>(sizeof(uint32_t));
receive_buf_upper_bound_
= static_cast<uint32_t>(msg_size + sizeof(uint32_t));
down->configure_read(
receive_policy::between(min_read_size, receive_buf_upper_bound_));
}
return consumed;
} else {
auto msg = sub_buffer.subspan(0, msg_size);
if (auto res = upper_layer_.consume(this_layer, msg); res >= 0) {
consumed += static_cast<ptrdiff_t>(msg.size()) + sizeof(uint32_t);
input = sub_buffer.subspan(msg_size);
} else {
return -1;
}
} }
// -- role: upper layer ------------------------------------------------------
template <class LowerLayer>
bool prepare_send(LowerLayer& down) {
access<LowerLayer> this_layer{&down, this};
return upper_layer_.prepare_send(this_layer);
} }
template <class LowerLayer>
bool done_sending(LowerLayer& down) {
access<LowerLayer> this_layer{&down, this};
return upper_layer_.done_sending(this_layer);
} }
template <class LowerLayer>
void abort(LowerLayer& down, const error& reason) {
access<LowerLayer> this_layer{&down, this};
return upper_layer_.abort(this_layer, reason);
} }
template <class LowerLayer> // -- convenience functions --------------------------------------------------
ptrdiff_t consume(LowerLayer& down, byte_span buffer, byte_span) {
using detail::from_network_order; static std::pair<size_t, byte_span> split(byte_span buffer) noexcept {
if (buffer.size() < 4) CAF_ASSERT(buffer.size() >= sizeof(uint32_t));
return 0; auto u32_size = uint32_t{0};
uint32_t u32_size = 0; memcpy(&u32_size, buffer.data(), sizeof(uint32_t));
memcpy(&u32_size, buffer.data(), 4); auto msg_size = static_cast<size_t>(detail::from_network_order(u32_size));
auto msg_size = static_cast<size_t>(from_network_order(u32_size)); return std::make_pair(msg_size, buffer.subspan(sizeof(uint32_t)));
if (buffer.size() < msg_size + 4)
return 0;
upper_layer_.consume(down, make_span(buffer.data() + 4, msg_size));
return msg_size + 4;
} }
private: private:
// -- implementation details -------------------------------------------------
template <class LowerLayerPtr>
auto this_layer_ptr(LowerLayerPtr down) {
return make_message_oriented_layer_ptr(this, down);
}
// -- member variables -------------------------------------------------------
UpperLayer upper_layer_; UpperLayer upper_layer_;
size_t message_offset_ = 0;
uint32_t receive_buf_upper_bound_ = default_receive_size;
}; };
} // namespace caf::net } // namespace caf::net
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#pragma once
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf::net {
/// Wraps a pointer to a mixed-message-oriented layer with a pointer to its
/// lower layer. Both pointers are then used to implement the interface required
/// for a mixed-message-oriented layer when calling into its upper layer.
template <class Layer, class LowerLayerPtr>
class message_oriented_layer_ptr {
public:
class access {
public:
access(Layer* layer, LowerLayerPtr down) : lptr_(layer), llptr_(down) {
// nop
}
bool can_send_more() const noexcept {
return lptr_->can_send_more(llptr_);
}
auto handle() const noexcept {
return lptr_->handle(llptr_);
}
void begin_message() {
lptr_->begin_message(llptr_);
}
[[nodiscard]] auto& message_buffer() {
return lptr_->message_buffer(llptr_);
}
[[nodiscard]] bool end_message() {
return lptr_->end_message(llptr_);
}
void abort_reason(error reason) {
return lptr_->abort_reason(llptr_, std::move(reason));
}
const error& abort_reason() {
return lptr_->abort_reason(llptr_);
}
private:
Layer* lptr_;
LowerLayerPtr llptr_;
};
message_oriented_layer_ptr(Layer* layer, LowerLayerPtr down)
: access_(layer, down) {
// nop
}
message_oriented_layer_ptr(const message_oriented_layer_ptr&) = default;
explicit operator bool() const noexcept {
return true;
}
access* operator->() const noexcept {
return &access_;
}
access& operator*() const noexcept {
return access_;
}
private:
mutable access access_;
};
template <class Layer, class LowerLayerPtr>
auto make_message_oriented_layer_ptr(Layer* this_layer, LowerLayerPtr down) {
using result_t = message_oriented_layer_ptr<Layer, LowerLayerPtr>;
return result_t{this_layer, down};
}
} // namespace caf::net
...@@ -7,7 +7,6 @@ ...@@ -7,7 +7,6 @@
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
#include "caf/net/receive_policy.hpp"
namespace caf::net { namespace caf::net {
......
...@@ -6,9 +6,9 @@ ...@@ -6,9 +6,9 @@
#include "caf/net/length_prefix_framing.hpp" #include "caf/net/length_prefix_framing.hpp"
#include "caf/test/dsl.hpp" #include "net-test.hpp"
#include <deque> #include <cctype>
#include <numeric> #include <numeric>
#include <vector> #include <vector>
...@@ -21,179 +21,106 @@ ...@@ -21,179 +21,106 @@
#include "caf/tag/message_oriented.hpp" #include "caf/tag/message_oriented.hpp"
using namespace caf; using namespace caf;
using namespace std::literals;
namespace { namespace {
/// upper layer: expect messages using string_list = std::vector<std::string>;
/// Needs to be initilized by the layer two steps down.
struct ul_expect_messages { struct app {
using input_tag = tag::message_oriented; using input_tag = tag::message_oriented;
void set_expected_messages(std::vector<byte_buffer> messages) { template <class LowerLayerPtr>
expected_messages.clear(); caf::error init(net::socket_manager*, LowerLayerPtr, const settings&) {
for (auto& msg : messages) return none;
expected_messages.emplace_back(std::move(msg));
} }
void add_expected_messages(std::vector<byte_buffer> messages) { template <class LowerLayerPtr>
for (auto& msg : messages) bool prepare_send(LowerLayerPtr) {
expected_messages.emplace_back(std::move(msg)); return true;
} }
template <class LowerLayer> template <class LowerLayerPtr>
ptrdiff_t consume(LowerLayer&, byte_span buffer) { bool done_sending(LowerLayerPtr) {
CAF_REQUIRE(expected_messages.size() > 0); return true;
auto& next = expected_messages.front();
CAF_CHECK_EQUAL(next.size(), buffer.size());
CAF_CHECK(std::equal(next.begin(), next.end(), buffer.begin()));
expected_messages.pop_front();
return buffer.size();
} }
std::deque<byte_buffer> expected_messages; template <class LowerLayerPtr>
}; void abort(LowerLayerPtr, const error&) {
// nop
/// lower layer: offer stream for message parsing
template <class UpperLayer>
struct ll_provide_stream_for_messages {
using output_tag = tag::stream_oriented;
void set_expectations(std::vector<byte> data,
std::vector<byte_buffer> messages) {
data_stream = std::move(data);
auto& checking_layer = upper_layer.upper_layer();
checking_layer.set_expected_messages(messages);
} }
void add_expectations(const std::vector<byte>& data, template <class LowerLayerPtr>
std::vector<byte_buffer> messages) { ptrdiff_t consume(LowerLayerPtr down, byte_span buf) {
data_stream.insert(data_stream.end(), data.begin(), data.end()); auto printable = [](byte x) { return ::isprint(static_cast<uint8_t>(x)); };
auto& checking_layer = upper_layer.upper_layer(); if (CHECK(std::all_of(buf.begin(), buf.end(), printable))) {
checking_layer.add_expect_messages(messages); auto str_buf = reinterpret_cast<char*>(buf.data());
} inputs.emplace_back(std::string{str_buf, buf.size()});
std::string response = "ok ";
void run() { response += std::to_string(inputs.size());
CAF_CHECK(!data_stream.empty()); auto response_bytes = as_bytes(make_span(response));
while (processed != data_stream.size()) { down->begin_message();
auto all_data = make_span(data_stream.data() + processed, auto& buf = down->message_buffer();
data_stream.size() - processed); buf.insert(buf.end(), response_bytes.begin(), response_bytes.end());
auto new_data = make_span(data_stream.data() + offered, CHECK(down->end_message());
data_stream.size() - offered); return static_cast<ptrdiff_t>(buf.size());
auto newly_offered = new_data.size(); } else {
auto consumed = upper_layer.consume(*this, all_data, new_data); return -1;
CAF_CHECK(consumed >= 0);
CAF_CHECK(static_cast<size_t>(consumed) <= data_stream.size());
offered += newly_offered;
processed += consumed;
if (consumed > 0) {
data_stream.erase(data_stream.begin(), data_stream.begin() + consumed);
offered -= processed;
processed = 0;
}
if (consumed == 0 || data_stream.empty())
return;
} }
} }
size_t processed = 0; std::vector<std::string> inputs;
size_t offered = 0;
std::vector<byte> data_stream;
UpperLayer upper_layer;
}; };
template <class... Ts> void encode(byte_buffer& buf, string_view msg) {
byte_buffer to_buf(const Ts&... xs) {
byte_buffer buf;
binary_serializer sink{nullptr, buf};
if (!(sink.apply(xs) && ...))
CAF_FAIL("to_buf failed: " << sink.get_error());
return buf;
}
void encode_message(std::vector<byte>& data, const byte_buffer& message) {
using detail::to_network_order; using detail::to_network_order;
auto current_size = data.size(); auto prefix = to_network_order(static_cast<uint32_t>(msg.size()));
data.insert(data.end(), 4, byte{0}); auto prefix_bytes = as_bytes(make_span(&prefix, 1));
auto msg_begin = data.begin() + current_size; buf.insert(buf.end(), prefix_bytes.begin(), prefix_bytes.end());
auto msg_size = message.size(); auto bytes = as_bytes(make_span(msg));
auto u32_size = to_network_order(static_cast<uint32_t>(msg_size)); buf.insert(buf.end(), bytes.begin(), bytes.end());
memcpy(std::addressof(*msg_begin), &u32_size, 4);
data.insert(data.end(), message.begin(), message.end());
} }
struct fixture { auto decode(byte_buffer& buf) {
using test_layers = ll_provide_stream_for_messages< auto printable = [](byte x) { return ::isprint(static_cast<uint8_t>(x)); };
net::length_prefix_framing<ul_expect_messages>>; string_list result;
auto input = make_span(buf);
void generate_messages(size_t num, size_t factor = 10) { while (!input.empty()) {
for (size_t n = 1; n <= num; n += 1) { auto [msg_size, msg] = net::length_prefix_framing<app>::split(input);
std::vector<int> buf(n * factor); if (msg_size > msg.size()) {
std::iota(buf.begin(), buf.end(), n); CAF_FAIL("cannot decode buffer: invalid message size");
messages.emplace_back(to_buf(buf)); } else if (!std::all_of(msg.begin(), msg.begin() + msg_size, printable)) {
CAF_FAIL("cannot decode buffer: unprintable characters found in message");
} else {
auto str = std::string{reinterpret_cast<char*>(msg.data()), msg_size};
result.emplace_back(std::move(str));
input = msg.subspan(msg_size);
} }
for (auto& msg : messages)
encode_message(data, msg);
} }
return result;
}
void set_expectations() { } // namespace
layers.set_expectations(data, messages);
}
void test_receive_data() { SCENARIO("length-prefix framing reads data with 32-bit size headers") {
layers.run(); GIVEN("a length_prefix_framing with an app that consumed strings") {
mock_stream_transport<net::length_prefix_framing<app>> uut;
CHECK_EQ(uut.init(), error{});
WHEN("pushing data into the unit-under-test") {
encode(uut.input, "hello");
encode(uut.input, "world");
auto input_size = static_cast<ptrdiff_t>(uut.input.size());
CHECK_EQ(uut.handle_input(), input_size);
THEN("the app receives all strings as individual messages") {
auto& state = uut.upper_layer.upper_layer();
if (CHECK_EQ(state.inputs.size(), 2u)) {
CHECK_EQ(state.inputs[0], "hello");
CHECK_EQ(state.inputs[1], "world");
}
auto response = string_view{reinterpret_cast<char*>(uut.output.data()),
uut.output.size()};
CHECK_EQ(decode(uut.output), string_list({"ok 1", "ok 2"}));
}
} }
void clear() {
data.clear();
messages.clear();
} }
test_layers layers;
std::vector<byte> data;
std::vector<byte_buffer> messages;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(length_prefix_framing_tests, fixture)
CAF_TEST(process messages) {
// Single message.
generate_messages(1);
set_expectations();
test_receive_data();
clear();
// Multiple messages.
generate_messages(10);
set_expectations();
test_receive_data();
}
CAF_TEST(incomplete message) {
generate_messages(1, 1000);
CAF_MESSAGE("data.size() = " << data.size());
auto initial_size = data.size();
auto data_copy = data;
auto mid = data.size() / 2;
data.resize(mid);
CAF_MESSAGE("data.size() = " << data.size());
data_copy.erase(data_copy.begin(), data_copy.begin() + mid);
CAF_MESSAGE("data_copy.size() = " << data_copy.size());
CAF_REQUIRE(data.size() + data_copy.size() == initial_size);
// Don't set expectations because there shouldn't be a complete message
// in the bytes.
auto messages_copy = messages;
messages.clear();
CAF_REQUIRE(messages.empty());
set_expectations();
test_receive_data();
data.insert(data.end(), data_copy.begin(), data_copy.end());
messages = messages_copy;
set_expectations();
test_receive_data();
} }
CAF_TEST_FIXTURE_SCOPE_END()
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