Commit 553f714c authored by Dominik Charousset's avatar Dominik Charousset

Merge pull request #453 from actor-framework/topic/serialization-tweaks

Make (de)serializer more flexible
parents d5671fa3 024f037a
......@@ -8,8 +8,6 @@
#include <iostream>
#include "caf/all.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp"
using std::cout;
using std::cerr;
......
......@@ -33,8 +33,6 @@ set (LIBCAF_CORE_SRCS
src/behavior.cpp
src/behavior_stack.cpp
src/behavior_impl.cpp
src/binary_deserializer.cpp
src/binary_serializer.cpp
src/blocking_actor.cpp
src/concatenated_tuple.cpp
src/continue_helper.cpp
......
......@@ -39,6 +39,7 @@
#include "caf/duration.hpp"
#include "caf/exception.hpp"
#include "caf/resumable.hpp"
#include "caf/streambuf.hpp"
#include "caf/actor_addr.hpp"
#include "caf/actor_pool.hpp"
#include "caf/attachable.hpp"
......
......@@ -20,57 +20,14 @@
#ifndef CAF_BINARY_DESERIALIZER_HPP
#define CAF_BINARY_DESERIALIZER_HPP
#include <string>
#include <cstddef>
#include <utility>
#include "caf/deserializer.hpp"
#include "caf/stream_deserializer.hpp"
#include "caf/streambuf.hpp"
namespace caf {
/// Implements the deserializer interface with a binary serialization protocol.
class binary_deserializer : public deserializer {
public:
binary_deserializer(actor_system& sys, const void* buf, size_t buf_size);
binary_deserializer(execution_unit* ctx, const void* buf, size_t buf_size);
binary_deserializer(actor_system& ctx, const void* first, const void* last);
binary_deserializer(execution_unit* ctx, const void* first, const void* last);
template <class C, class T>
binary_deserializer(C&& ctx, const T& xs)
: binary_deserializer(std::forward<C>(ctx), xs.data(), xs.size()) {
// nop
}
/// Replaces the current read buffer.
void set_rdbuf(const void* buf, size_t buf_size);
/// Replaces the current read buffer.
void set_rdbuf(const void* first, const void* last);
/// Returns whether this deserializer has reached the end of its buffer.
bool at_end() const;
/// Compares the next `num_bytes` from the underlying buffer to `buf`
/// with same semantics as `strncmp(this->pos_, buf, num_bytes) == 0`.
bool buf_equals(const void* buf, size_t num_bytes);
/// Moves the current read position in the buffer by `num_bytes`.
binary_deserializer& advance(ptrdiff_t num_bytes);
void begin_object(uint16_t& nr, std::string& name) override;
void end_object() override;
void begin_sequence(size_t& num_elements) override;
void end_sequence() override;
void apply_raw(size_t num_bytes, void* data) override;
protected:
void apply_builtin(builtin in_out_type, void* in_out) override;
private:
const void* pos_;
const void* end_;
};
/// A stream serializer that writes into an unbounded contiguous character
/// sequence.
using binary_deserializer = stream_deserializer<charbuf>;
} // namespace caf
......
......@@ -20,98 +20,14 @@
#ifndef CAF_BINARY_SERIALIZER_HPP
#define CAF_BINARY_SERIALIZER_HPP
#include <utility>
#include <sstream>
#include <iomanip>
#include <iterator>
#include <functional>
#include <type_traits>
#include "caf/fwd.hpp"
#include "caf/serializer.hpp"
#include "caf/primitive_variant.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/stream_serializer.hpp"
#include "caf/streambuf.hpp"
namespace caf {
/// Implements the serializer interface with a binary serialization protocol.
class binary_serializer : public serializer {
public:
/// Creates a binary serializer writing to given iterator position.
template <class C,
class Out,
class E =
typename std::enable_if<
std::is_same<
typename std::iterator_traits<Out>::iterator_category,
std::output_iterator_tag
>::value
|| std::is_same<
typename std::iterator_traits<Out>::iterator_category,
std::random_access_iterator_tag
>::value
>::type>
binary_serializer(C&& ctx, Out it) : serializer(std::forward<C>(ctx)) {
reset_iter(it);
}
template <class C,
class T,
class E =
typename std::enable_if<
detail::has_char_insert<T>::value
>::type>
binary_serializer(C&& ctx, T& out) : serializer(std::forward<C>(ctx)) {
reset_container(out);
}
using write_fun = std::function<void(const char*, size_t)>;
void begin_object(uint16_t& typenr, std::string& name) override;
void end_object() override;
void begin_sequence(size_t& list_size) override;
void end_sequence() override;
void apply_raw(size_t num_bytes, void* data) override;
void apply_builtin(builtin in_out_type, void* in_out) override;
template <class OutIter>
void reset_iter(OutIter iter) {
struct fun {
fun(OutIter pos) : pos_(pos) {
// nop
}
void operator()(const char* first, size_t num_bytes) {
pos_ = std::copy(first, first + num_bytes, pos_);
}
OutIter pos_;
};
out_ = fun{iter};
}
template <class T>
void reset_container(T& x) {
struct fun {
fun(T& container) : x_(container) {
// nop
}
void operator()(const char* first, size_t num_bytes) {
auto last = first + num_bytes;
x_.insert(x_.end(), first, last);
}
T& x_;
};
out_ = fun{x};
}
private:
write_fun out_;
};
/// A stream serializer that writes into an unbounded contiguous character
/// sequence.
using binary_serializer = stream_serializer<vectorbuf>;
} // namespace caf
......
......@@ -44,7 +44,7 @@ public:
data_processor(const data_processor&) = delete;
data_processor& operator=(const data_processor&) = delete;
data_processor(execution_unit* ctx) : context_(ctx) {
data_processor(execution_unit* ctx = nullptr) : context_(ctx) {
// nop
}
......
......@@ -42,7 +42,7 @@ public:
explicit deserializer(actor_system& sys);
explicit deserializer(execution_unit* ctx);
explicit deserializer(execution_unit* ctx = nullptr);
};
/// Reads `x` from `source`.
......
......@@ -17,6 +17,9 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_DETAIL_NETWORK_ORDER_HPP
#define CAF_DETAIL_NETWORK_ORDER_HPP
#include "caf/config.hpp"
namespace caf {
......@@ -67,3 +70,5 @@ inline T from_network_order(T value) {
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_NETWORK_ORDER_HPP
......@@ -613,6 +613,30 @@ public:
static constexpr bool value = false;
};
template <class T>
struct has_data_member {
template <class U>
static auto test(U* x) -> decltype(x->data(), std::true_type());
template <class U>
static auto test(...) -> std::false_type;
using type = decltype(test<T>(nullptr));
static constexpr bool value = type::value;
};
template <class T>
struct has_size_member {
template <class U>
static auto test(U* x) -> decltype(x->size(), std::true_type());
template <class U>
static auto test(...) -> std::false_type;
using type = decltype(test<T>(nullptr));
static constexpr bool value = type::value;
};
/// Checks whether T is convertible to either `std::function<void (T&)>`
/// or `std::function<void (const T&)>`.
template <class F, class T>
......
......@@ -74,12 +74,10 @@ class continue_helper;
class mailbox_element;
class message_handler;
class response_promise;
class binary_serializer;
class event_based_actor;
class type_erased_tuple;
class type_erased_value;
class actor_control_block;
class binary_deserializer;
class actor_system_config;
class uniform_type_info_map;
class forwarding_actor_proxy;
......
......@@ -41,7 +41,7 @@ public:
explicit serializer(actor_system& sys);
explicit serializer(execution_unit* ctx);
explicit serializer(execution_unit* ctx = nullptr);
virtual ~serializer();
};
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_STREAM_DESERIALIZER_HPP
#define CAF_STREAM_DESERIALIZER_HPP
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <exception>
#include <iomanip>
#include <limits>
#include <sstream>
#include <stdexcept>
#include <string>
#include <type_traits>
#include "caf/deserializer.hpp"
#include "caf/logger.hpp"
#include "caf/detail/ieee_754.hpp"
#include "caf/detail/network_order.hpp"
namespace caf {
/// Implements the deserializer interface with a binary serialization protocol.
template <class Streambuf>
class stream_deserializer : public deserializer {
static_assert(
std::is_base_of<
std::streambuf,
typename std::remove_reference<Streambuf>::type
>::value,
"Streambuf must inherit from std::streambuf"
);
public:
template <class... Ts>
explicit stream_deserializer(actor_system& sys, Ts&&... xs)
: deserializer{sys},
streambuf_{std::forward<Ts>(xs)...} {
}
template <class... Ts>
explicit stream_deserializer(execution_unit* ctx, Ts&&... xs)
: deserializer{ctx},
streambuf_{std::forward<Ts>(xs)...} {
}
template <
class S,
class = typename std::enable_if<
std::is_same<
typename std::remove_reference<S>::type,
typename std::remove_reference<Streambuf>::type
>::value
>::type
>
explicit stream_deserializer(S&& sb)
: deserializer{nullptr},
streambuf_{std::forward<S>(sb)} {
}
void begin_object(uint16_t& typenr, std::string& name) override {
apply_int(typenr);
if (typenr == 0)
apply(name);
}
void end_object() override {
// nop
}
void begin_sequence(size_t& num_elements) override {
static_assert(sizeof(size_t) >= sizeof(uint32_t),
"sizeof(size_t) < sizeof(uint32_t)");
uint32_t tmp;
apply_int(tmp);
num_elements = static_cast<size_t>(tmp);
}
void end_sequence() override {
// nop
}
void apply_raw(size_t num_bytes, void* data) override {
range_check(num_bytes);
streambuf_.sgetn(reinterpret_cast<char*>(data), num_bytes);
}
protected:
void apply_builtin(builtin type, void* val) override {
CAF_ASSERT(val != nullptr);
switch (type) {
case i8_v:
case u8_v:
apply_raw(sizeof(uint8_t), val);
break;
case i16_v:
case u16_v:
apply_int(*reinterpret_cast<uint16_t*>(val));
break;
case i32_v:
case u32_v:
apply_int(*reinterpret_cast<uint32_t*>(val));
break;
case i64_v:
case u64_v:
apply_int(*reinterpret_cast<uint64_t*>(val));
break;
case float_v:
apply_float(*reinterpret_cast<float*>(val));
break;
case double_v:
apply_float(*reinterpret_cast<double*>(val));
break;
case ldouble_v: {
// the IEEE-754 conversion does not work for long double
// => fall back to string serialization (even though it sucks)
std::string tmp;
apply(tmp);
std::istringstream iss{std::move(tmp)};
iss >> *reinterpret_cast<long double*>(val);
break;
}
case string8_v: {
auto& str = *reinterpret_cast<std::string*>(val);
uint32_t str_size;
apply_int(str_size);
range_check(str_size);
str.resize(str_size);
// TODO: When using C++14, switch to str.data(), which then has a
// non-const overload.
streambuf_.sgetn(reinterpret_cast<char*>(&str[0]), str_size);
break;
}
case string16_v: {
auto& str = *reinterpret_cast<std::u16string*>(val);
uint32_t str_size;
apply_int(str_size);
auto bytes = str_size * sizeof(std::u16string::value_type);
range_check(bytes);
str.resize(str_size);
// TODO: When using C++14, switch to str.data(), which then has a
// non-const overload.
streambuf_.sgetn(reinterpret_cast<char*>(&str[0]), bytes);
break;
}
case string32_v: {
auto& str = *reinterpret_cast<std::u32string*>(val);
uint32_t str_size;
apply_int(str_size);
auto bytes = str_size * sizeof(std::u32string::value_type);
range_check(bytes);
str.resize(str_size);
// TODO: When using C++14, switch to str.data(), which then has a
// non-const overload.
streambuf_.sgetn(reinterpret_cast<char*>(&str[0]), bytes);
break;
}
}
}
private:
void range_check(size_t read_size) {
static_assert(sizeof(std::streamsize) <= sizeof(size_t),
"std::streamsize > std::size_t");
if (static_cast<size_t>(streambuf_.in_avail()) < read_size) {
CAF_LOG_ERROR("range_check failed");
throw std::out_of_range("stream_deserializer<T>::read_range()");
}
}
template <class T>
void apply_int(T& x) {
T tmp;
apply_raw(sizeof(T), &tmp);
x = detail::from_network_order(tmp);
}
template <class T>
void apply_float(T& x) {
typename detail::ieee_754_trait<T>::packed_type tmp;
apply_int(tmp);
x = detail::unpack754(tmp);
}
Streambuf streambuf_;
};
} // namespace caf
#endif // CAF_STREAM_DESERIALIZER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_STREAM_SERIALIZER_HPP
#define CAF_STREAM_SERIALIZER_HPP
#include <cstddef>
#include <cstdint>
#include <iomanip>
#include <limits>
#include <streambuf>
#include <sstream>
#include <string>
#include <type_traits>
#include "caf/serializer.hpp"
#include "caf/streambuf.hpp"
#include "caf/detail/ieee_754.hpp"
#include "caf/detail/network_order.hpp"
namespace caf {
/// Implements the serializer interface with a binary serialization protocol.
template <class Streambuf>
class stream_serializer : public serializer {
static_assert(
std::is_base_of<
std::streambuf,
typename std::remove_reference<Streambuf>::type
>::value,
"Streambuf must inherit from std::streambuf"
);
public:
template <class... Ts>
explicit stream_serializer(actor_system& sys, Ts&&... xs)
: serializer{sys},
streambuf_{std::forward<Ts>(xs)...} {
}
template <class... Ts>
explicit stream_serializer(execution_unit* ctx, Ts&&... xs)
: serializer{ctx},
streambuf_{std::forward<Ts>(xs)...} {
}
template <
class S,
class = typename std::enable_if<
std::is_same<
typename std::remove_reference<S>::type,
typename std::remove_reference<Streambuf>::type
>::value
>::type
>
explicit stream_serializer(S&& sb)
: serializer{nullptr},
streambuf_{std::forward<S>(sb)} {
}
void begin_object(uint16_t& typenr, std::string& name) override {
apply(typenr);
if (typenr == 0)
apply(name);
}
void end_object() override {
// nop
}
void begin_sequence(size_t& list_size) override {
auto s = static_cast<uint32_t>(list_size);
apply(s);
}
void end_sequence() override {
// nop
}
void apply_raw(size_t num_bytes, void* data) override {
streambuf_.sputn(reinterpret_cast<char*>(data), num_bytes);
}
protected:
void apply_builtin(builtin type, void* val) override {
CAF_ASSERT(val != nullptr);
switch (type) {
case i8_v:
case u8_v:
apply_raw(sizeof(uint8_t), val);
break;
case i16_v:
case u16_v:
apply_int(*reinterpret_cast<uint16_t*>(val));
break;
case i32_v:
case u32_v:
apply_int(*reinterpret_cast<uint32_t*>(val));
break;
case i64_v:
case u64_v:
apply_int(*reinterpret_cast<uint64_t*>(val));
break;
case float_v:
apply_int(detail::pack754(*reinterpret_cast<float*>(val)));
break;
case double_v:
apply_int(detail::pack754(*reinterpret_cast<double*>(val)));
break;
case ldouble_v: {
// 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)
<< *reinterpret_cast<long double*>(val);
auto tmp = oss.str();
auto s = static_cast<uint32_t>(tmp.size());
apply(s);
apply_raw(tmp.size(), const_cast<char*>(tmp.c_str()));
break;
}
case string8_v: {
auto str = reinterpret_cast<std::string*>(val);
auto s = static_cast<uint32_t>(str->size());
apply(s);
apply_raw(str->size(), const_cast<char*>(str->c_str()));
break;
}
case string16_v: {
auto& str = *reinterpret_cast<std::u16string*>(val);
auto s = static_cast<uint32_t>(str.size());
apply(s);
for (auto c : str) {
// the standard does not guarantee that char16_t is exactly 16 bits...
auto tmp = static_cast<uint16_t>(c);
apply_raw(sizeof(tmp), &tmp);
}
break;
}
case string32_v: {
auto& str = *reinterpret_cast<std::u32string*>(val);
auto s = static_cast<uint32_t>(str.size());
apply(s);
for (auto c : str) {
// the standard does not guarantee that char32_t is exactly 32 bits...
auto tmp = static_cast<uint32_t>(c);
apply_raw(sizeof(tmp), &tmp);
}
break;
}
}
}
private:
template <class T>
void apply_int(T x) {
auto y = detail::to_network_order(x);
apply_raw(sizeof(T), &y);
}
Streambuf streambuf_;
};
} // namespace caf
#endif // CAF_STREAM_SERIALIZER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#ifndef CAF_STREAMBUF_HPP
#define CAF_STREAMBUF_HPP
#include <algorithm>
#include <cstddef>
#include <cstring>
#include <streambuf>
#include <vector>
#include "caf/detail/type_traits.hpp"
namespace caf {
/// A streambuffer abstraction over a fixed array of bytes. This streambuffer
/// cannot overflow/underflow. Once it has reached its end, attempts to read
/// characters will return `trait_type::eof`.
template <class CharT = char, class Traits = std::char_traits<CharT>>
class arraybuf : public std::basic_streambuf<CharT, Traits> {
public:
using base = std::basic_streambuf<CharT, Traits>;
using char_type = typename base::char_type;
using traits_type = typename base::traits_type;
using int_type = typename base::int_type;
using pos_type = typename base::pos_type;
using off_type = typename base::off_type;
/// Constructs an array streambuffer from a container.
/// @param c A contiguous container.
/// @pre `c.data()` must point to a contiguous sequence of characters having
/// length `c.size()`.
template <
class Container,
class = typename std::enable_if<
detail::has_data_member<Container>::value
&& detail::has_size_member<Container>::value
>::type
>
arraybuf(Container& c)
: arraybuf{const_cast<CharT*>(c.data()), c.size()} {
}
/// Constructs an array streambuffer from a raw character sequence.
/// @param data A pointer to the first character.
/// @param size The length of the character sequence.
arraybuf(CharT* data, size_t size) {
setbuf(data, static_cast<std::streamsize>(size));
}
// There exists a bug in libstdc++ version < 5: the implementation does not
// provide the necessary move constructors, so we have to roll our own :-/.
// See https://gcc.gnu.org/bugzilla/show_bug.cgi?id=54316 for details.
// TODO: remove after having raised the minimum GCC version to 5.
arraybuf(arraybuf&& other) {
this->setg(other.eback(), other.gptr(), other.egptr());
this->setp(other.pptr(), other.epptr());
other.setg(nullptr, nullptr, nullptr);
other.setp(nullptr, nullptr);
}
// TODO: remove after having raised the minimum GCC version to 5.
arraybuf& operator=(arraybuf&& other) {
this->setg(other.eback(), other.gptr(), other.egptr());
this->setp(other.pptr(), other.epptr());
other.setg(nullptr, nullptr, nullptr);
other.setp(nullptr, nullptr);
return *this;
}
/// Replaces the internal character sequence with a new one.
/// @param s A pointer to the first character.
/// @param n The length of the character sequence.
std::basic_streambuf<CharT, Traits>*
setbuf(CharT* s, std::streamsize n) override {
this->setg(s, s, s + n);
this->setp(s, s + n);
return this;
}
protected:
std::streamsize xsputn(const char_type* s, std::streamsize n) override {
auto available = this->epptr() - this->pptr();
auto actual = std::min(n, available);
std::memcpy(this->pptr(), s, actual * sizeof(char_type));
this->pbump(actual);
return actual;
}
std::streamsize xsgetn(char_type* s, std::streamsize n) override {
auto available = this->egptr() - this->gptr();
auto actual = std::min(n, available);
std::memcpy(s, this->gptr(), actual * sizeof(char_type));
this->gbump(actual);
return actual;
}
};
/// A streambuffer abstraction over a contiguous container. It supports
/// reading in the same style as arraybuf, but is unbounded for output.
template <class Container>
class containerbuf
: public std::basic_streambuf<
typename Container::value_type,
std::char_traits<typename Container::value_type>
> {
public:
using base = std::basic_streambuf<
typename Container::value_type,
std::char_traits<typename Container::value_type>
>;
using char_type = typename base::char_type;
using traits_type = typename base::traits_type;
using int_type = typename base::int_type;
using pos_type = typename base::pos_type;
using off_type = typename base::off_type;
/// Constructs a container streambuf.
/// @param c A contiguous container.
template <
class C = Container,
class = typename std::enable_if<
detail::has_data_member<C>::value && detail::has_size_member<C>::value
>::type
>
containerbuf(Container& c) : container_{c} {
this->setg(const_cast<char_type*>(c.data()),
const_cast<char_type*>(c.data()),
const_cast<char_type*>(c.data()) + c.size());
}
// See note in arraybuf(arraybuf&&).
// TODO: remove after having raised the minimum GCC version to 5.
containerbuf(containerbuf&& other)
: container_{other.container_} {
this->setg(other.eback(), other.gptr(), other.egptr());
other.setg(nullptr, nullptr, nullptr);
}
// TODO: remove after having raised the minimum GCC version to 5.
containerbuf& operator=(containerbuf&& other) {
this->setg(other.eback(), other.gptr(), other.egptr());
other.setg(nullptr, nullptr, nullptr);
return *this;
}
// Hides base-class implementation to simplify single-character lookup.
int_type sgetc() {
if (this->gptr() == this->egptr())
return traits_type::eof();
return traits_type::to_int_type(*this->gptr());
}
// Hides base-class implementation to simplify single-character insert.
int_type sputc(char_type c) {
container_.push_back(c);
return c;
}
protected:
// We can't get obtain more characters on underflow, so we only optimize
// multi-character sequential reads.
std::streamsize xsgetn(char_type* s, std::streamsize n) override {
auto available = this->egptr() - this->gptr();
auto actual = std::min(n, available);
std::memcpy(s, this->gptr(), actual * sizeof(char_type));
this->gbump(actual);
return actual;
}
// Should never get called, because there is always space in the buffer.
// (But just in case, it does the same thing as sputc.)
int_type overflow(int_type c = traits_type::eof()) final {
if (!traits_type::eq_int_type(c, traits_type::eof()))
container_.push_back(traits_type::to_char_type(c));
return c;
}
std::streamsize xsputn(const char_type* s, std::streamsize n) override {
// TODO: Do a performance analysis whether the current implementation based
// on copy_n is faster than these two statements:
// (1) container_.resize(container_.size() + n);
// (2) std::memcpy(this->pptr(), s, n * sizeof(char_type));
std::copy_n(s, n, std::back_inserter(container_));
return n;
}
private:
Container& container_;
};
using charbuf = arraybuf<char>;
using vectorbuf = containerbuf<std::vector<char>>;
} // namespace caf
#endif // CAF_STREAMBUF_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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 <limits>
#include <string>
#include <cstdint>
#include <cstring>
#include <sstream>
#include <iomanip>
#include <iterator>
#include <exception>
#include <stdexcept>
#include <type_traits>
#include "caf/logger.hpp"
#include "caf/actor_system.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/detail/ieee_754.hpp"
#include "caf/detail/network_order.hpp"
namespace caf {
namespace {
using pointer = const void*;
const char* as_char_pointer(pointer ptr) {
return reinterpret_cast<const char*>(ptr);
}
template <class Distance>
pointer advanced(pointer ptr, Distance num_bytes) {
return reinterpret_cast<const char*>(ptr) + num_bytes;
}
inline void range_check(pointer begin, pointer end, size_t read_size) {
if (advanced(begin, read_size) > end) {
CAF_LOG_ERROR("range_check failed");
throw std::out_of_range("binary_deserializer::read_range()");
}
}
template <class T>
void apply_int(binary_deserializer& bd, T& x) {
T tmp;
bd.apply_raw(sizeof(T), &tmp);
x = detail::from_network_order(tmp);
}
template <class T>
void apply_float(binary_deserializer& bd, T& x) {
typename detail::ieee_754_trait<T>::packed_type tmp;
apply_int(bd, tmp);
x = detail::unpack754(tmp);
}
} // namespace <anonmyous>
binary_deserializer::binary_deserializer(actor_system& sys,
const void* buf, size_t buf_size)
: binary_deserializer(sys, buf, advanced(buf, buf_size)) {
// nop
}
binary_deserializer::binary_deserializer(execution_unit* ctx,
const void* buf,
size_t buf_size)
: binary_deserializer(ctx, buf, advanced(buf, buf_size)) {
// nop
}
binary_deserializer::binary_deserializer(actor_system& sys,
const void* first, const void* last)
: binary_deserializer(sys.dummy_execution_unit(), first, last) {
// nop
}
binary_deserializer::binary_deserializer(execution_unit* ctx,
const void* first, const void* last)
: deserializer(ctx),
pos_(first),
end_(last) {
// nop
}
bool binary_deserializer::at_end() const {
return pos_ == end_;
}
/// with same semantics as `strncmp(this->pos_, buf, num_bytes)`.
bool binary_deserializer::buf_equals(const void* buf, size_t num_bytes) {
auto bytes_left = static_cast<size_t>(std::distance(as_char_pointer(pos_),
as_char_pointer(end_)));
if (bytes_left < num_bytes)
return false;
return strncmp(as_char_pointer(pos_), as_char_pointer(buf), num_bytes) == 0;
}
binary_deserializer& binary_deserializer::advance(ptrdiff_t num_bytes) {
pos_ = advanced(pos_, num_bytes);
return *this;
}
void binary_deserializer::begin_object(uint16_t& nr, std::string& name) {
apply_int(*this, nr);
if (nr == 0)
apply(name);
}
void binary_deserializer::end_object() {
// nop
}
void binary_deserializer::begin_sequence(size_t& result) {
CAF_LOG_TRACE("");
static_assert(sizeof(size_t) >= sizeof(uint32_t),
"sizeof(size_t) < sizeof(uint32_t)");
uint32_t tmp;
apply_int(*this, tmp);
result = static_cast<size_t>(tmp);
}
void binary_deserializer::end_sequence() {
// nop
}
void binary_deserializer::apply_raw(size_t num_bytes, void* storage) {
range_check(pos_, end_, num_bytes);
memcpy(storage, pos_, num_bytes);
pos_ = advanced(pos_, num_bytes);
}
void binary_deserializer::apply_builtin(builtin type, void* val) {
CAF_ASSERT(val != nullptr);
switch (type) {
case i8_v:
case u8_v:
apply_raw(sizeof(uint8_t), val);
break;
case i16_v:
case u16_v:
apply_int(*this, *reinterpret_cast<uint16_t*>(val));
break;
case i32_v:
case u32_v:
apply_int(*this, *reinterpret_cast<uint32_t*>(val));
break;
case i64_v:
case u64_v:
apply_int(*this, *reinterpret_cast<uint64_t*>(val));
break;
case float_v:
apply_float(*this, *reinterpret_cast<float*>(val));
break;
case double_v:
apply_float(*this, *reinterpret_cast<double*>(val));
break;
case ldouble_v: {
// the IEEE-754 conversion does not work for long double
// => fall back to string serialization (event though it sucks)
std::string tmp;
apply(tmp);
std::istringstream iss{std::move(tmp)};
iss >> *reinterpret_cast<long double*>(val);
break;
}
case string8_v: {
auto& str = *reinterpret_cast<std::string*>(val);
uint32_t str_size;
apply_int(*this, str_size);
range_check(pos_, end_, str_size);
str.clear();
str.reserve(str_size);
auto last = advanced(pos_, str_size);
std::copy(reinterpret_cast<const char*>(pos_),
reinterpret_cast<const char*>(last),
std::back_inserter(str));
pos_ = last;
break;
}
case string16_v: {
auto& str = *reinterpret_cast<std::string*>(val);
uint32_t str_size;
apply_int(*this, str_size);
range_check(pos_, end_, str_size * sizeof(uint16_t));
str.clear();
str.reserve(str_size);
auto last = advanced(pos_, str_size);
std::copy(reinterpret_cast<const uint16_t*>(pos_),
reinterpret_cast<const uint16_t*>(last),
std::back_inserter(str));
pos_ = last;
break;
}
case string32_v: {
auto& str = *reinterpret_cast<std::string*>(val);
uint32_t str_size;
apply_int(*this, str_size);
range_check(pos_, end_, str_size * sizeof(uint32_t));
str.clear();
str.reserve(str_size);
auto last = advanced(pos_, str_size);
std::copy(reinterpret_cast<const uint32_t*>(pos_),
reinterpret_cast<const uint32_t*>(last),
std::back_inserter(str));
pos_ = last;
break;
}
}
}
} // namespace caf
......@@ -254,7 +254,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
return;
}
std::vector<char> buf;
binary_serializer bs{self->context(), std::back_inserter(buf)};
binary_serializer bs{std::back_inserter(buf), self->context()};
self->save_state(bs, 0);
auto sender = node.sender;
// request(...)
......@@ -300,7 +300,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
self->bhvr_stack().pop_back();
self->is_migrated_from(false);
}
binary_deserializer bd{self->context(), buf.data(), buf.size()};
binary_deserializer bd{buf.data(), buf.size(), self->context()};
self->load_state(bd, 0);
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
......
......@@ -46,6 +46,7 @@
#include <type_traits>
#include "caf/message.hpp"
#include "caf/streambuf.hpp"
#include "caf/serializer.hpp"
#include "caf/ref_counted.hpp"
#include "caf/deserializer.hpp"
......@@ -160,14 +161,14 @@ struct fixture {
template <class T, class... Ts>
vector<char> serialize(T& x, Ts&... xs) {
vector<char> buf;
binary_serializer bs{&context, std::back_inserter(buf)};
binary_serializer bs{&context, buf};
apply(bs, x, xs...);
return buf;
}
template <class T, class... Ts>
void deserialize(const vector<char>& buf, T& x, Ts&... xs) {
binary_deserializer bd{&context, buf.data(), buf.size()};
binary_deserializer bd{&context, buf};
apply(bd, x, xs...);
}
......@@ -381,6 +382,7 @@ CAF_TEST(multiple_messages) {
CAF_CHECK(is_message(m2).equal(i32, te, str, rs));
}
CAF_TEST(type_erased_value) {
auto buf = serialize(str);
type_erased_value_ptr ptr{new type_erased_value_impl<std::string>};
......@@ -411,4 +413,28 @@ CAF_TEST(type_erased_tuple) {
CAF_CHECK_EQUAL(to_string(tview), deep_to_string(std::make_tuple(str, i32)));
}
CAF_TEST(streambuf_serialization) {
auto data = std::string{"The quick brown fox jumps over the lazy dog"};
std::vector<char> buf;
// First, we check the standard use case in CAF where stream serializers own
// their stream buffers.
stream_serializer<vectorbuf> bs{vectorbuf{buf}};
bs << data;
stream_deserializer<charbuf> bd{charbuf{buf}};
std::string target;
bd >> target;
CAF_CHECK(data == target);
// Second, we test another use case where the serializers only keep
// references of the stream buffers.
buf.clear();
target.clear();
vectorbuf vb{buf};
stream_serializer<vectorbuf&> vs{vb};
vs << data;
charbuf cb{buf};
stream_deserializer<charbuf&> vd{cb};
vd >> target;
CAF_CHECK(data == target);
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -17,115 +17,94 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/binary_serializer.hpp"
#include "caf/config.hpp"
#include <limits>
#define CAF_SUITE streambuf
#include "caf/test/unit_test.hpp"
#include "caf/detail/ieee_754.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/streambuf.hpp"
namespace caf {
using namespace caf;
namespace {
template <class T>
void apply_int(binary_serializer& bs, T x) {
auto y = detail::to_network_order(x);
bs.apply_raw(sizeof(T), &y);
}
} // namespace <anonmyous>
void binary_serializer::begin_object(uint16_t& nr, std::string& name) {
apply(nr);
if (nr == 0)
apply(name);
}
void binary_serializer::end_object() {
// nop
CAF_TEST(signed_arraybuf) {
auto data = std::string{"The quick brown fox jumps over the lazy dog"};
arraybuf<char> ab{data};
// Let's read some.
CAF_CHECK_EQUAL(static_cast<size_t>(ab.in_avail()), data.size());
CAF_CHECK_EQUAL(ab.sgetc(), 'T');
std::string buf;
buf.resize(3);
size_t got = ab.sgetn(&buf[0], 3);
CAF_CHECK_EQUAL(got, 3u);
CAF_CHECK_EQUAL(buf, "The");
CAF_CHECK_EQUAL(ab.sgetc(), ' ');
// Exhaust the stream.
buf.resize(data.size());
got = ab.sgetn(&buf[0] + 3, data.size() - 3);
CAF_CHECK_EQUAL(got, data.size() - 3);
CAF_CHECK_EQUAL(data, buf);
CAF_CHECK_EQUAL(ab.in_avail(), 0);
// No more.
auto c = ab.sgetc();
CAF_CHECK_EQUAL(c, charbuf::traits_type::eof());
// Reset the stream and write into it.
ab.setbuf(&data[0], data.size());
CAF_CHECK_EQUAL(static_cast<size_t>(ab.in_avail()), data.size());
auto put = ab.sputn("One", 3);
CAF_CHECK_EQUAL(put, 3);
CAF_CHECK(data.compare(0, 3, "One") == 0);
}
void binary_serializer::begin_sequence(size_t& list_size) {
auto s = static_cast<uint32_t>(list_size);
apply(s);
CAF_TEST(unsigned_arraybuf) {
std::vector<uint8_t> data = {1, 2, 3, 4};
arraybuf<uint8_t> ab{data};
decltype(data) buf;
std::copy(std::istreambuf_iterator<uint8_t>{&ab},
std::istreambuf_iterator<uint8_t>{},
std::back_inserter(buf));
CAF_CHECK_EQUAL(data, buf);
}
void binary_serializer::end_sequence() {
// nop
CAF_TEST(containerbuf) {
std::string data{
"Habe nun, ach! Philosophie,\n"
"Juristerei und Medizin,\n"
"Und leider auch Theologie\n"
"Durchaus studiert, mit heißem Bemühn.\n"
"Da steh ich nun, ich armer Tor!\n"
"Und bin so klug als wie zuvor"
};
// Write some data.
std::vector<char> buf;
vectorbuf vb{buf};
auto put = vb.sputn(data.data(), data.size());
CAF_CHECK_EQUAL(static_cast<size_t>(put), data.size());
put = vb.sputn(";", 1);
CAF_CHECK_EQUAL(put, 1);
std::string target;
std::copy(buf.begin(), buf.end(), std::back_inserter(target));
CAF_CHECK_EQUAL(data + ';', target);
// Check "overflow" on a new stream.
buf.clear();
vectorbuf vb2{buf};
auto chr = vb2.sputc('x');
CAF_CHECK_EQUAL(chr, 'x');
// Let's read some data into a stream.
buf.clear();
containerbuf<std::string> scb{data};
std::copy(std::istreambuf_iterator<char>{&scb},
std::istreambuf_iterator<char>{},
std::back_inserter(buf));
CAF_CHECK_EQUAL(buf.size(), data.size());
CAF_CHECK(std::equal(buf.begin(), buf.end(), data.begin() /*, data.end() */));
// We're done, nothing to see here, please move along.
CAF_CHECK_EQUAL(scb.sgetc(), containerbuf<std::string>::traits_type::eof());
// Let's read again, but now in one big block.
buf.clear();
containerbuf<std::string> sib2{data};
buf.resize(data.size());
size_t got = sib2.sgetn(&buf[0], buf.size());
CAF_CHECK_EQUAL(got, data.size());
CAF_CHECK_EQUAL(buf.size(), data.size());
CAF_CHECK(std::equal(buf.begin(), buf.end(), data.begin() /*, data.end() */));
}
void binary_serializer::apply_raw(size_t num_bytes, void* data) {
out_(reinterpret_cast<char*>(data), num_bytes);
}
void binary_serializer::apply_builtin(builtin type, void* val) {
CAF_ASSERT(val != nullptr);
switch (type) {
case i8_v:
case u8_v:
apply_raw(sizeof(uint8_t), val);
break;
case i16_v:
case u16_v:
apply_int(*this, *reinterpret_cast<uint16_t*>(val));
break;
case i32_v:
case u32_v:
apply_int(*this, *reinterpret_cast<uint32_t*>(val));
break;
case i64_v:
case u64_v:
apply_int(*this, *reinterpret_cast<uint64_t*>(val));
break;
case float_v:
apply_int(*this, detail::pack754(*reinterpret_cast<float*>(val)));
break;
case double_v:
apply_int(*this, detail::pack754(*reinterpret_cast<double*>(val)));
break;
case ldouble_v: {
// 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)
<< *reinterpret_cast<long double*>(val);
auto tmp = oss.str();
auto s = static_cast<uint32_t>(tmp.size());
apply(s);
apply_raw(tmp.size(), const_cast<char*>(tmp.c_str()));
break;
}
case string8_v: {
auto str = reinterpret_cast<std::string*>(val);
auto s = static_cast<uint32_t>(str->size());
apply(s);
apply_raw(str->size(), const_cast<char*>(str->c_str()));
break;
}
case string16_v: {
auto& str = *reinterpret_cast<std::u16string*>(val);
auto s = static_cast<uint32_t>(str.size());
apply(s);
for (auto c : str) {
// the standard does not guarantee that char16_t is exactly 16 bits...
auto tmp = static_cast<uint16_t>(c);
apply_raw(sizeof(tmp), &tmp);
}
break;
}
case string32_v: {
auto& str = *reinterpret_cast<std::u32string*>(val);
auto s = static_cast<uint32_t>(str.size());
apply(s);
for (auto c : str) {
// the standard does not guarantee that char32_t is exactly 32 bits...
auto tmp = static_cast<uint32_t>(c);
apply_raw(sizeof(tmp), &tmp);
}
break;
}
}
}
} // namespace caf
......@@ -19,6 +19,7 @@
#include "caf/io/basp/instance.hpp"
#include "caf/streambuf.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/binary_deserializer.hpp"
......@@ -65,7 +66,7 @@ connection_state instance::handle(execution_unit* ctx,
return err();
}
} else {
binary_deserializer bd{ctx, dm.buf.data(), dm.buf.size()};
binary_deserializer bd{ctx, dm.buf};
bd >> hdr;
CAF_LOG_DEBUG(CAF_ARG(hdr));
if (! valid(hdr)) {
......@@ -81,7 +82,7 @@ connection_state instance::handle(execution_unit* ctx,
&& hdr.dest_node != this_node_) {
auto path = lookup(hdr.dest_node);
if (path) {
binary_serializer bs{ctx, std::back_inserter(path->wr_buf)};
binary_serializer bs{ctx, path->wr_buf};
bs << hdr;
if (payload)
bs.apply_raw(payload->size(), payload->data());
......@@ -119,7 +120,7 @@ connection_state instance::handle(execution_unit* ctx,
actor_id aid = invalid_actor_id;
std::set<std::string> sigs;
if (payload_valid()) {
binary_deserializer bd{ctx, payload->data(), payload->size()};
binary_deserializer bd{ctx, *payload};
bd >> aid >> sigs;
}
// close self connection after handshake is done
......@@ -176,7 +177,7 @@ connection_state instance::handle(execution_unit* ctx,
&& tbl_.lookup_direct(hdr.source_node) == invalid_connection_handle
&& tbl_.add_indirect(last_hop, hdr.source_node))
callee_.learned_new_node_indirectly(hdr.source_node);
binary_deserializer bd{ctx, payload->data(), payload->size()};
binary_deserializer bd{ctx, *payload};
std::vector<strong_actor_ptr> forwarding_stack;
message msg;
bd >> forwarding_stack >> msg;
......@@ -334,7 +335,7 @@ void instance::write(execution_unit* ctx,
<< CAF_ARG(source_actor) << CAF_ARG(dest_actor));
if (!pw) {
uint32_t zero = 0;
binary_serializer bs{ctx, std::back_inserter(buf)};
binary_serializer bs{ctx, buf};
bs << source_node
<< dest_node
<< source_actor
......@@ -344,16 +345,16 @@ void instance::write(execution_unit* ctx,
<< operation_data;
} else {
// reserve space in the buffer to write the payload later on
auto wr_pos = static_cast<ptrdiff_t>(buf.size());
auto wr_pos = buf.size();
char placeholder[basp::header_size];
buf.insert(buf.end(), std::begin(placeholder), std::end(placeholder));
auto pl_pos = buf.size();
{ // lifetime scope of first serializer (write payload)
binary_serializer bs{ctx, std::back_inserter(buf)};
binary_serializer bs{ctx, buf};
(*pw)(bs);
}
// write broker message to the reserved space
binary_serializer bs2{ctx, buf.begin() + wr_pos};
stream_serializer<charbuf> bs2{ctx, buf.data() + wr_pos, pl_pos - wr_pos};
auto plen = static_cast<uint32_t>(buf.size() - pl_pos);
bs2 << source_node
<< dest_node
......
......@@ -140,7 +140,7 @@ public:
uint32_t serialized_size(const message& msg) {
buffer buf;
binary_serializer bs{mpx_, back_inserter(buf)};
binary_serializer bs{mpx_, buf};
bs << msg;
return static_cast<uint32_t>(buf.size());
}
......@@ -233,7 +233,7 @@ public:
template <class... Ts>
void to_payload(buffer& buf, const Ts&... xs) {
binary_serializer bs{mpx_, std::back_inserter(buf)};
binary_serializer bs{mpx_, buf};
to_payload(bs, xs...);
}
......@@ -254,7 +254,7 @@ public:
std::pair<basp::header, buffer> from_buf(const buffer& buf) {
basp::header hdr;
binary_deserializer bd{mpx_, buf.data(), buf.size()};
binary_deserializer bd{mpx_, buf};
bd >> hdr;
buffer payload;
if (hdr.payload_len > 0) {
......@@ -327,7 +327,7 @@ public:
std::tie(hdr, buf) = read_from_out_buf(hdl);
CAF_MESSAGE("dispatch output buffer for connection " << hdl.id());
CAF_REQUIRE(hdr.operation == basp::message_type::dispatch_message);
binary_deserializer source{mpx_, buf.data(), buf.size()};
binary_deserializer source{mpx_, buf};
std::vector<strong_actor_ptr> stages;
message msg;
source >> stages >> msg;
......@@ -372,7 +372,7 @@ public:
CAF_MESSAGE("output buffer has " << ob.size() << " bytes");
basp::header hdr;
{ // lifetime scope of source
binary_deserializer source{this_->mpx(), ob.data(), ob.size()};
binary_deserializer source{this_->mpx(), ob};
source >> hdr;
}
buffer payload;
......
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