Commit 488e19bc authored by Dominik Charousset's avatar Dominik Charousset

Re-implement container-based binary serializers

The stream serializers are flexible. However, they are also complex and
using them as default implementation leaves performance on the table.
parent 89bf26d1
...@@ -30,6 +30,8 @@ set(LIBCAF_CORE_SRCS ...@@ -30,6 +30,8 @@ set(LIBCAF_CORE_SRCS
src/behavior.cpp src/behavior.cpp
src/behavior_impl.cpp src/behavior_impl.cpp
src/behavior_stack.cpp src/behavior_stack.cpp
src/binary_deserializer.cpp
src/binary_serializer.cpp
src/blocking_actor.cpp src/blocking_actor.cpp
src/blocking_behavior.cpp src/blocking_behavior.cpp
src/chars.cpp src/chars.cpp
......
...@@ -93,12 +93,14 @@ ...@@ -93,12 +93,14 @@
#include "caf/composed_behavior.hpp" #include "caf/composed_behavior.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/primitive_variant.hpp" #include "caf/primitive_variant.hpp"
#include "caf/stream_serializer.hpp"
#include "caf/make_config_option.hpp" #include "caf/make_config_option.hpp"
#include "caf/timeout_definition.hpp" #include "caf/timeout_definition.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/composable_behavior.hpp" #include "caf/composable_behavior.hpp"
#include "caf/config_option_adder.hpp" #include "caf/config_option_adder.hpp"
#include "caf/stream_deserializer.hpp"
#include "caf/typed_actor_pointer.hpp" #include "caf/typed_actor_pointer.hpp"
#include "caf/scoped_execution_unit.hpp" #include "caf/scoped_execution_unit.hpp"
#include "caf/typed_response_promise.hpp" #include "caf/typed_response_promise.hpp"
......
...@@ -23,9 +23,74 @@ ...@@ -23,9 +23,74 @@
namespace caf { namespace caf {
/// A stream serializer that writes into an unbounded contiguous character /// Implements the deserializer interface with a binary serialization protocol.
/// sequence. class binary_deserializer final : public deserializer {
using binary_deserializer = stream_deserializer<charbuf>; public:
// -- member types -----------------------------------------------------------
using super = deserializer;
using buffer = std::vector<char>;
// -- constructors, destructors, and assignment operators --------------------
binary_deserializer(actor_system& sys, const char* buf, size_t buf_size);
binary_deserializer(execution_unit* ctx, const char* buf, size_t buf_size);
binary_deserializer(actor_system& sys, const buffer& buf);
binary_deserializer(execution_unit* ctx, const buffer& buf);
// -- overridden member functions --------------------------------------------
error begin_object(uint16_t& typenr, std::string& name) override;
error end_object() override;
error begin_sequence(size_t& list_size) override;
error end_sequence() override;
error apply_raw(size_t num_bytes, void* data) override;
protected:
error apply_impl(int8_t&) override;
error apply_impl(uint8_t&) override;
error apply_impl(int16_t&) override;
error apply_impl(uint16_t&) override;
error apply_impl(int32_t&) override;
error apply_impl(uint32_t&) override;
error apply_impl(int64_t&) override;
error apply_impl(uint64_t&) override;
error apply_impl(float&) override;
error apply_impl(double&) override;
error apply_impl(long double&) override;
error apply_impl(std::string&) override;
error apply_impl(std::u16string&) override;
error apply_impl(std::u32string&) override;
private:
bool range_check(size_t read_size) {
return pos_ + read_size <= end_;
}
const char* pos_;
const char* end_;
};
} // namespace caf } // namespace caf
...@@ -18,14 +18,83 @@ ...@@ -18,14 +18,83 @@
#pragma once #pragma once
#include "caf/stream_serializer.hpp" #include <cstddef>
#include "caf/streambuf.hpp" #include <cstdint>
#include <vector>
#include "caf/serializer.hpp"
namespace caf { namespace caf {
/// A stream serializer that writes into an unbounded contiguous character /// Implements the serializer interface with a binary serialization protocol.
/// sequence. class binary_serializer final : public serializer {
using binary_serializer = stream_serializer<vectorbuf>; public:
// -- member types -----------------------------------------------------------
} // namespace caf using super = serializer;
using buffer = std::vector<char>;
// -- constructors, destructors, and assignment operators --------------------
binary_serializer(actor_system& sys, buffer& buf);
binary_serializer(execution_unit* ctx, buffer& buf);
// -- overridden member functions --------------------------------------------
/// Sets the write position to given offset.
/// @pre `offset <= buf.size()`
void seek(size_t offset);
/// Jumps `num_bytes` forward. Resizes the buffer (filling it with zeros)
/// when skipping past the end.
void skip(size_t num_bytes);
// -- overridden member functions --------------------------------------------
error begin_object(uint16_t& typenr, std::string& name) override;
error end_object() override;
error begin_sequence(size_t& list_size) override;
error end_sequence() override;
error apply_raw(size_t num_bytes, void* data) override;
protected:
error apply_impl(int8_t&) override;
error apply_impl(uint8_t&) override;
error apply_impl(int16_t&) override;
error apply_impl(uint16_t&) override;
error apply_impl(int32_t&) override;
error apply_impl(uint32_t&) override;
error apply_impl(int64_t&) override;
error apply_impl(uint64_t&) override;
error apply_impl(float&) override;
error apply_impl(double&) override;
error apply_impl(long double&) override;
error apply_impl(std::string&) override;
error apply_impl(std::u16string&) override;
error apply_impl(std::u32string&) override;
private:
buffer& buf_;
buffer::iterator write_pos_;
};
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/binary_deserializer.hpp"
#include <iomanip>
#include <sstream>
#include <type_traits>
#include "caf/detail/ieee_754.hpp"
#include "caf/detail/network_order.hpp"
namespace caf {
namespace {
template <class T>
error apply_int(binary_deserializer& bs, T& x) {
typename std::make_unsigned<T>::type tmp;
if (auto err = bs.apply_raw(sizeof(T), &tmp))
return err;
x = static_cast<T>(detail::from_network_order(tmp));
return none;
}
template <class T>
error apply_float(binary_deserializer& bs, T& x) {
typename detail::ieee_754_trait<T>::packed_type tmp;
if (auto err = apply_int(bs, tmp))
return err;
x = detail::unpack754(tmp);
return none;
}
} // namespace <anonmyous>
binary_deserializer::binary_deserializer(actor_system& sys, const char* buf,
size_t buf_size)
: super(sys), pos_(buf), end_(buf + buf_size) {
// nop
}
binary_deserializer::binary_deserializer(execution_unit* ctx, const char* buf,
size_t buf_size)
: super(ctx), pos_(buf), end_(buf + buf_size) {
// nop
}
binary_deserializer::binary_deserializer(actor_system& sys, const buffer& buf)
: binary_deserializer(sys, buf.data(), buf.size()) {
// nop
}
binary_deserializer::binary_deserializer(execution_unit* ctx, const buffer& buf)
: binary_deserializer(ctx, buf.data(), buf.size()) {
// nop
}
error binary_deserializer::begin_object(uint16_t& nr, std::string& name) {
if (nr != 0)
return apply(nr);
if (auto err = apply(nr))
return err;
return apply(name);
}
error binary_deserializer::end_object() {
return none;
}
error binary_deserializer::begin_sequence(size_t& list_size) {
auto s = static_cast<uint32_t>(list_size);
if (auto err = apply(s))
return err;
list_size = s;
return none;
}
error binary_deserializer::end_sequence() {
return none;
}
error binary_deserializer::apply_raw(size_t num_bytes, void* storage) {
if (!range_check(num_bytes))
return sec::end_of_stream;
memcpy(storage, pos_, num_bytes);
pos_ += num_bytes;
return none;
}
error binary_deserializer::apply_impl(int8_t& x) {
return apply_raw(sizeof(int8_t), &x);
}
error binary_deserializer::apply_impl(uint8_t& x) {
return apply_raw(sizeof(uint8_t), &x);
}
error binary_deserializer::apply_impl(int16_t& x) {
return apply_int(*this, x);
}
error binary_deserializer::apply_impl(uint16_t& x) {
return apply_int(*this, x);
}
error binary_deserializer::apply_impl(int32_t& x) {
return apply_int(*this, x);
}
error binary_deserializer::apply_impl(uint32_t& x) {
return apply_int(*this, x);
}
error binary_deserializer::apply_impl(int64_t& x) {
return apply_int(*this, x);
}
error binary_deserializer::apply_impl(uint64_t& x) {
return apply_int(*this, x);
}
error binary_deserializer::apply_impl(float& x) {
return apply_float(*this, x);
}
error binary_deserializer::apply_impl(double& x) {
return apply_float(*this, x);
}
error binary_deserializer::apply_impl(long double& x) {
// The IEEE-754 conversion does not work for long double
// => fall back to string serialization (even though it sucks).
std::string tmp;
if (auto err = apply(tmp))
return err;
std::istringstream iss{std::move(tmp)};
iss >> x;
return none;
}
error binary_deserializer::apply_impl(std::string& x) {
size_t str_size;
if (auto err = begin_sequence(str_size))
return err;
if (!range_check(str_size))
return sec::end_of_stream;
x.assign(pos_, pos_ + str_size);
pos_ += str_size;
return end_sequence();
}
error binary_deserializer::apply_impl(std::u16string& x) {
auto str_size = x.size();
if (auto err = begin_sequence(str_size))
return err;
for (size_t i = 0; i < str_size; ++i) {
// The standard does not guarantee that char16_t is exactly 16 bits.
uint16_t tmp;
if (auto err = apply_int(*this, tmp))
return err;
x.push_back(static_cast<char16_t>(tmp));
}
return none;
}
error binary_deserializer::apply_impl(std::u32string& x) {
auto str_size = x.size();
if (auto err = begin_sequence(str_size))
return err;
for (size_t i = 0; i < str_size; ++i) {
// The standard does not guarantee that char32_t is exactly 32 bits.
uint32_t tmp;
if (auto err = apply_int(*this, tmp))
return err;
x.push_back(static_cast<char32_t>(tmp));
}
return none;
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/binary_serializer.hpp"
#include <iomanip>
#include <sstream>
#include "caf/detail/ieee_754.hpp"
#include "caf/detail/network_order.hpp"
namespace caf {
namespace {
template <class T>
error apply_int(binary_serializer& bs, T x) {
auto y = detail::to_network_order(x);
return bs.apply_raw(sizeof(T), &y);
}
} // namespace <anonmyous>
binary_serializer::binary_serializer(actor_system& sys, buffer& buf)
: super(sys), buf_(buf), write_pos_(buf.end()) {
// nop
}
binary_serializer::binary_serializer(execution_unit* ctx, buffer& buf)
: super(ctx), buf_(buf), write_pos_(buf.end()) {
// nop
}
void binary_serializer::seek(size_t offset) {
write_pos_ = buf_.begin() + offset;
}
void binary_serializer::skip(size_t num_bytes) {
auto last = buf_.end();
if (write_pos_ + num_bytes <= last) {
write_pos_ += num_bytes;
} else {
buf_.insert(last, num_bytes - (last - write_pos_), 0);
write_pos_ = buf_.end();
}
}
error binary_serializer::begin_object(uint16_t& nr, std::string& name) {
if (nr != 0)
return apply(nr);
if (auto err = apply(nr))
return err;
return apply(name);
}
error binary_serializer::end_object() {
return none;
}
error binary_serializer::begin_sequence(size_t& list_size) {
auto s = static_cast<uint32_t>(list_size);
return apply(s);
}
error binary_serializer::end_sequence() {
return none;
}
error binary_serializer::apply_raw(size_t num_bytes, void* data) {
auto ptr = reinterpret_cast<char*>(data);
auto last = buf_.end();
if (write_pos_ == last) {
buf_.insert(last, ptr, ptr + num_bytes);
write_pos_ = buf_.end();
} else if (write_pos_ + num_bytes < last) {
std::copy(ptr, ptr + num_bytes, write_pos_);
write_pos_ += num_bytes;
} else {
auto remaining = static_cast<size_t>(last - write_pos_);
std::copy(ptr, ptr + remaining, write_pos_);
buf_.insert(last, ptr + remaining, ptr + num_bytes);
write_pos_ = buf_.end();
}
return none;
}
error binary_serializer::apply_impl(int8_t& x) {
return apply_raw(sizeof(int8_t), &x);
}
error binary_serializer::apply_impl(uint8_t& x) {
return apply_raw(sizeof(uint8_t), &x);
}
error binary_serializer::apply_impl(int16_t& x) {
return apply_int(*this, static_cast<uint16_t>(x));
}
error binary_serializer::apply_impl(uint16_t& x) {
return apply_int(*this, x);
}
error binary_serializer::apply_impl(int32_t& x) {
return apply_int(*this, static_cast<uint32_t>(x));
}
error binary_serializer::apply_impl(uint32_t& x) {
return apply_int(*this, x);
}
error binary_serializer::apply_impl(int64_t& x) {
return apply_int(*this, static_cast<uint64_t>(x));
}
error binary_serializer::apply_impl(uint64_t& x) {
return apply_int(*this, x);
}
error binary_serializer::apply_impl(float& x) {
return apply_int(*this, detail::pack754(x));
}
error binary_serializer::apply_impl(double& x) {
return apply_int(*this, detail::pack754(x));
}
error binary_serializer::apply_impl(long double& x) {
// The IEEE-754 conversion does not work for long double
// => fall back to string serialization (event though it sucks).
std::ostringstream oss;
oss << std::setprecision(std::numeric_limits<long double>::digits) << x;
auto tmp = oss.str();
return apply_impl(tmp);
}
error binary_serializer::apply_impl(std::string& x) {
auto str_size = x.size();
if (str_size == 0)
return error::eval([&] { return begin_sequence(str_size); },
[&] { return end_sequence(); });
auto data = const_cast<char*>(x.c_str());
return error::eval([&] { return begin_sequence(str_size); },
[&] { return apply_raw(str_size, data); },
[&] { return end_sequence(); });
}
error binary_serializer::apply_impl(std::u16string& x) {
auto str_size = x.size();
if (auto err = begin_sequence(str_size))
return err;
for (auto c : x) {
// The standard does not guarantee that char16_t is exactly 16 bits.
if (auto err = apply_int(*this, static_cast<uint16_t>(c)))
return err;
}
return none;
}
error binary_serializer::apply_impl(std::u32string& x) {
auto str_size = x.size();
if (auto err = begin_sequence(str_size))
return err;
for (auto c : x) {
// The standard does not guarantee that char16_t is exactly 16 bits.
if (auto err = apply_int(*this, static_cast<uint32_t>(c)))
return err;
}
return none;
}
} // namespace caf
...@@ -21,51 +21,52 @@ ...@@ -21,51 +21,52 @@
#define CAF_SUITE serialization #define CAF_SUITE serialization
#include "caf/test/unit_test.hpp" #include "caf/test/unit_test.hpp"
#include <new> #include <algorithm>
#include <set> #include <cassert>
#include <cstdint>
#include <cstring>
#include <functional>
#include <iostream>
#include <iterator>
#include <limits>
#include <list> #include <list>
#include <stack>
#include <tuple>
#include <locale> #include <locale>
#include <memory> #include <memory>
#include <string> #include <new>
#include <limits> #include <set>
#include <vector>
#include <cstring>
#include <sstream> #include <sstream>
#include <cstdint> #include <stack>
#include <cstring>
#include <cassert>
#include <iterator>
#include <typeinfo>
#include <iostream>
#include <stdexcept> #include <stdexcept>
#include <algorithm> #include <string>
#include <functional> #include <tuple>
#include <type_traits> #include <type_traits>
#include <typeinfo>
#include <vector>
#include "caf/message.hpp"
#include "caf/streambuf.hpp"
#include "caf/serializer.hpp"
#include "caf/ref_counted.hpp"
#include "caf/deserializer.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/message_handler.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/primitive_variant.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/make_type_erased_view.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/deserializer.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/make_type_erased_tuple_view.hpp" #include "caf/make_type_erased_tuple_view.hpp"
#include "caf/make_type_erased_view.hpp"
#include "caf/message.hpp"
#include "caf/message_handler.hpp"
#include "caf/primitive_variant.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/ref_counted.hpp"
#include "caf/serializer.hpp"
#include "caf/stream_deserializer.hpp"
#include "caf/stream_serializer.hpp"
#include "caf/streambuf.hpp"
#include "caf/detail/enum_to_string.hpp"
#include "caf/detail/get_mac_addresses.hpp"
#include "caf/detail/ieee_754.hpp" #include "caf/detail/ieee_754.hpp"
#include "caf/detail/int_list.hpp" #include "caf/detail/int_list.hpp"
#include "caf/detail/safe_equal.hpp" #include "caf/detail/safe_equal.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
#include "caf/detail/enum_to_string.hpp"
#include "caf/detail/get_mac_addresses.hpp"
using namespace std; using namespace std;
using namespace caf; using namespace caf;
...@@ -295,6 +296,7 @@ CAF_TEST(enum_classes) { ...@@ -295,6 +296,7 @@ CAF_TEST(enum_classes) {
CAF_TEST(strings) { CAF_TEST(strings) {
auto buf = serialize(str); auto buf = serialize(str);
std::cout << "buf: " << deep_to_string(buf) << std::endl;
string x; string x;
deserialize(buf, x); deserialize(buf, x);
CAF_CHECK_EQUAL(str, x); CAF_CHECK_EQUAL(str, x);
...@@ -388,7 +390,7 @@ CAF_TEST(multiple_messages) { ...@@ -388,7 +390,7 @@ CAF_TEST(multiple_messages) {
CAF_TEST(type_erased_value) { CAF_TEST(type_erased_value) {
auto buf = serialize(str); auto buf = serialize(str);
type_erased_value_ptr ptr{new type_erased_value_impl<std::string>}; type_erased_value_ptr ptr{new type_erased_value_impl<std::string>};
binary_deserializer bd{&context, buf.data(), buf.size()}; binary_deserializer bd{&context, buf};
ptr->load(bd); ptr->load(bd);
CAF_CHECK_EQUAL(str, *reinterpret_cast<const std::string*>(ptr->get())); CAF_CHECK_EQUAL(str, *reinterpret_cast<const std::string*>(ptr->get()));
} }
......
...@@ -198,24 +198,18 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender, ...@@ -198,24 +198,18 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
void instance::write(execution_unit* ctx, buffer_type& buf, void instance::write(execution_unit* ctx, buffer_type& buf,
header& hdr, payload_writer* pw) { header& hdr, payload_writer* pw) {
CAF_LOG_TRACE(CAF_ARG(hdr)); CAF_LOG_TRACE(CAF_ARG(hdr));
error err; binary_serializer sink{ctx, buf};
if (pw != nullptr) { if (pw != nullptr) {
auto pos = buf.size(); // Write the BASP header after the payload.
// write payload first (skip first 72 bytes and write header later) auto header_offset = buf.size();
char placeholder[basp::header_size]; sink.skip(header_size);
buf.insert(buf.end(), std::begin(placeholder), std::end(placeholder)); if (auto err = (*pw)(sink))
binary_serializer bs{ctx, buf}; CAF_LOG_ERROR(CAF_ARG(err));
(*pw)(bs); sink.seek(header_offset);
auto plen = buf.size() - pos - basp::header_size; auto payload_len = buf.size() - (header_offset + basp::header_size);
CAF_ASSERT(plen <= std::numeric_limits<uint32_t>::max()); hdr.payload_len = static_cast<uint32_t>(payload_len);
hdr.payload_len = static_cast<uint32_t>(plen);
stream_serializer<charbuf> out{ctx, buf.data() + pos, basp::header_size};
err = out(hdr);
} else {
binary_serializer bs{ctx, buf};
err = bs(hdr);
} }
if (err) if (auto err = sink(hdr))
CAF_LOG_ERROR(CAF_ARG(err)); CAF_LOG_ERROR(CAF_ARG(err));
} }
......
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