Commit b876e4cc authored by Dominik Charousset's avatar Dominik Charousset

updated binary_serializer to use util::buffer as data sink

parent b5f0a0e6
...@@ -32,7 +32,9 @@ ...@@ -32,7 +32,9 @@
#define CPPA_BINARY_SERIALIZER_HPP #define CPPA_BINARY_SERIALIZER_HPP
#include <utility> #include <utility>
#include "cppa/serializer.hpp" #include "cppa/serializer.hpp"
#include "cppa/util/buffer.hpp"
namespace cppa { namespace cppa {
...@@ -44,20 +46,16 @@ namespace detail { class binary_writer; } ...@@ -44,20 +46,16 @@ namespace detail { class binary_writer; }
*/ */
class binary_serializer : public serializer { class binary_serializer : public serializer {
friend class detail::binary_writer;
char* m_begin;
char* m_end;
char* m_wr_pos;
// make sure that it's safe to write num_bytes bytes to m_wr_pos // make sure that it's safe to write num_bytes bytes to m_wr_pos
void acquire(size_t num_bytes); void acquire(size_t num_bytes);
public: public:
binary_serializer(); /**
* @brief Creates a binary serializer writing to @p write_buffer.
~binary_serializer(); * @warning @p write_buffer must be guaranteed to outlive @p this
*/
binary_serializer(util::buffer* write_buffer);
void begin_object(const std::string& tname); void begin_object(const std::string& tname);
...@@ -74,23 +72,15 @@ class binary_serializer : public serializer { ...@@ -74,23 +72,15 @@ class binary_serializer : public serializer {
void write_raw(size_t num_bytes, const void* data); void write_raw(size_t num_bytes, const void* data);
/** /**
* @brief Returns the number of written bytes. * @note does <b>not</b> affect the underlying write buffer
*/
size_t size() const;
/**
* @brief Returns a pointer to the internal buffer.
*/ */
const char* data() const; void reset();
size_t sendable_size() const;
const char* sendable_data(); private:
/** size_t m_obj_count;
* @brief Resets the internal buffer. size_t m_begin_pos;
*/ util::buffer* m_sink;
void reset();
}; };
......
...@@ -38,43 +38,34 @@ ...@@ -38,43 +38,34 @@
using std::enable_if; using std::enable_if;
namespace cppa {
namespace { namespace {
constexpr size_t chunk_size = 512; constexpr size_t chunk_size = 512;
constexpr size_t ui32_size = sizeof(std::uint32_t); constexpr size_t ui32_size = sizeof(std::uint32_t);
} // namespace <anonymous>
namespace cppa {
namespace detail {
class binary_writer { class binary_writer {
binary_serializer* m_serializer;
public: public:
binary_writer(binary_serializer* s) : m_serializer(s) { } binary_writer(util::buffer* sink) : m_sink(sink) { }
template<typename T> template<typename T>
inline static void write_int(binary_serializer* bs, const T& value) { static inline void write_int(util::buffer* sink, const T& value) {
memcpy(bs->m_wr_pos, &value, sizeof(T)); sink->write(sizeof(T), &value, util::grow_if_needed);
bs->m_wr_pos += sizeof(T);
} }
inline static void write_string(binary_serializer* bs, static inline void write_string(util::buffer* sink,
const std::string& str) { const std::string& str) {
write_int(bs, static_cast<std::uint32_t>(str.size())); write_int(sink, static_cast<std::uint32_t>(str.size()));
memcpy(bs->m_wr_pos, str.c_str(), str.size()); sink->write(str.size(), str.c_str(), util::grow_if_needed);
bs->m_wr_pos += str.size();
} }
template<typename T> template<typename T>
void operator()(const T& value, void operator()(const T& value,
typename enable_if<std::is_integral<T>::value>::type* = 0) { typename enable_if<std::is_integral<T>::value>::type* = 0) {
m_serializer->acquire(sizeof(T)); write_int(m_sink, value);
write_int(m_serializer, value);
} }
template<typename T> template<typename T>
...@@ -85,96 +76,71 @@ class binary_writer { ...@@ -85,96 +76,71 @@ class binary_writer {
} }
void operator()(const std::string& str) { void operator()(const std::string& str) {
m_serializer->acquire(sizeof(std::uint32_t) + str.size()); write_string(m_sink, str);
write_string(m_serializer, str);
} }
void operator()(const std::u16string& str) { void operator()(const std::u16string& str) {
m_serializer->acquire(sizeof(std::uint32_t) + str.size()); write_int(m_sink, static_cast<std::uint32_t>(str.size()));
write_int(m_serializer, static_cast<std::uint32_t>(str.size()));
for (char16_t c : str) { for (char16_t c : str) {
// force writer to use exactly 16 bit // force writer to use exactly 16 bit
write_int(m_serializer, static_cast<std::uint16_t>(c)); write_int(m_sink, static_cast<std::uint16_t>(c));
} }
} }
void operator()(const std::u32string& str) { void operator()(const std::u32string& str) {
m_serializer->acquire(sizeof(std::uint32_t) + str.size()); write_int(m_sink, static_cast<std::uint32_t>(str.size()));
write_int(m_serializer, static_cast<std::uint32_t>(str.size()));
for (char32_t c : str) { for (char32_t c : str) {
// force writer to use exactly 32 bit // force writer to use exactly 32 bit
write_int(m_serializer, static_cast<std::uint32_t>(c)); write_int(m_sink, static_cast<std::uint32_t>(c));
} }
} }
}; private:
} // namespace detail util::buffer* m_sink;
binary_serializer::binary_serializer() : m_begin(0), m_end(0), m_wr_pos(0) { };
}
binary_serializer::~binary_serializer() { } // namespace <anonymous>
delete[] m_begin;
}
void binary_serializer::acquire(size_t num_bytes) { binary_serializer::binary_serializer(util::buffer* buf)
if (!m_begin) { : m_obj_count(0), m_begin_pos(0), m_sink(buf) {
num_bytes += ui32_size;
size_t new_size = chunk_size;
while (new_size <= num_bytes) {
new_size += chunk_size;
}
m_begin = new char[new_size];
m_end = m_begin + new_size;
m_wr_pos = m_begin + ui32_size;
}
else {
char* next_wr_pos = m_wr_pos + num_bytes;
if (next_wr_pos > m_end) {
size_t new_size = static_cast<size_t>(m_end - m_begin)
+ chunk_size;
while ((m_begin + new_size) < next_wr_pos) {
new_size += chunk_size;
}
char* begin = new char[new_size];
auto used_bytes = static_cast<size_t>(m_wr_pos - m_begin);
if (used_bytes > 0) {
memcpy(m_begin, begin, used_bytes);
}
delete[] m_begin;
m_begin = begin;
m_end = m_begin + new_size;
m_wr_pos = m_begin + used_bytes;
}
}
} }
void binary_serializer::begin_object(const std::string& tname) { void binary_serializer::begin_object(const std::string& tname) {
acquire(sizeof(std::uint32_t) + tname.size()); if (++m_obj_count == 1) {
detail::binary_writer::write_string(this, tname); // store a dummy size in the buffer that is
// eventually updated on matching end_object()
m_begin_pos = m_sink->size();
std::uint32_t dummy_size = 0;
m_sink->write(sizeof(std::uint32_t), &dummy_size, util::grow_if_needed);
}
binary_writer::write_string(m_sink, tname);
} }
void binary_serializer::end_object() { void binary_serializer::end_object() {
if (--m_obj_count == 0) {
// update the size in the buffer
auto data = m_sink->data();
auto s = static_cast<std::uint32_t>(m_sink->size()
- (m_begin_pos + ui32_size));
auto wr_pos = data + m_begin_pos;
memcpy(wr_pos, &s, sizeof(std::uint32_t));
}
} }
void binary_serializer::begin_sequence(size_t list_size) { void binary_serializer::begin_sequence(size_t list_size) {
acquire(sizeof(std::uint32_t)); binary_writer::write_int(m_sink, static_cast<std::uint32_t>(list_size));
detail::binary_writer::write_int(this,
static_cast<std::uint32_t>(list_size));
} }
void binary_serializer::end_sequence() { void binary_serializer::end_sequence() { }
}
void binary_serializer::write_value(const primitive_variant& value) { void binary_serializer::write_value(const primitive_variant& value) {
value.apply(detail::binary_writer(this)); value.apply(binary_writer(m_sink));
} }
void binary_serializer::write_raw(size_t num_bytes, const void* data) { void binary_serializer::write_raw(size_t num_bytes, const void* data) {
acquire(num_bytes); m_sink->write(num_bytes, data, util::grow_if_needed);
memcpy(m_wr_pos, data, num_bytes);
m_wr_pos += num_bytes;
} }
void binary_serializer::write_tuple(size_t size, void binary_serializer::write_tuple(size_t size,
...@@ -185,26 +151,8 @@ void binary_serializer::write_tuple(size_t size, ...@@ -185,26 +151,8 @@ void binary_serializer::write_tuple(size_t size,
} }
} }
size_t binary_serializer::sendable_size() const {
return static_cast<size_t>(m_wr_pos - m_begin);
}
size_t binary_serializer::size() const {
return sendable_size() - ui32_size;
}
const char* binary_serializer::data() const {
return m_begin + ui32_size;
}
const char* binary_serializer::sendable_data() {
auto s = static_cast<std::uint32_t>(size());
memcpy(m_begin, &s, ui32_size);
return m_begin;
}
void binary_serializer::reset() { void binary_serializer::reset() {
m_wr_pos = m_begin + ui32_size; m_obj_count = 0;
} }
} // namespace cppa } // namespace cppa
...@@ -105,8 +105,9 @@ mm_message::~mm_message() { ...@@ -105,8 +105,9 @@ mm_message::~mm_message() {
void mailman_loop(intrusive::single_reader_queue<mm_message>& q) { void mailman_loop(intrusive::single_reader_queue<mm_message>& q) {
bool done = false; bool done = false;
util::buffer wr_buf;
// serializes outgoing messages // serializes outgoing messages
binary_serializer bs; binary_serializer bs(&wr_buf);
// connected tcp peers // connected tcp peers
std::map<process_information, util::io_stream_ptr_pair> peers; std::map<process_information, util::io_stream_ptr_pair> peers;
std::unique_ptr<mm_message> msg; std::unique_ptr<mm_message> msg;
...@@ -120,12 +121,13 @@ void mailman_loop(intrusive::single_reader_queue<mm_message>& q) { ...@@ -120,12 +121,13 @@ void mailman_loop(intrusive::single_reader_queue<mm_message>& q) {
auto i = peers.find(*target_peer); auto i = peers.find(*target_peer);
if (i != peers.end()) { if (i != peers.end()) {
bool disconnect_peer = false; bool disconnect_peer = false;
bs.reset();
wr_buf.reset();
try { try {
bs << out_msg; bs << out_msg;
DEBUG("--> " << to_string(out_msg)); DEBUG("--> " << to_string(out_msg));
DEBUG("outgoing message size: " << bs.size()); DEBUG("outgoing message size: " << bs.size());
i->second.second->write(bs.sendable_data(), i->second.second->write(wr_buf.data(), wr_buf.size());
bs.sendable_size());
} }
// something went wrong; close connection to this peer // something went wrong; close connection to this peer
catch (std::exception& e) { catch (std::exception& e) {
...@@ -138,7 +140,6 @@ void mailman_loop(intrusive::single_reader_queue<mm_message>& q) { ...@@ -138,7 +140,6 @@ void mailman_loop(intrusive::single_reader_queue<mm_message>& q) {
//post_office_close_socket(peer_fd); //post_office_close_socket(peer_fd);
peers.erase(i); peers.erase(i);
} }
bs.reset();
} }
else { else {
DEBUG("message to an unknown peer: " << to_string(out_msg)); DEBUG("message to an unknown peer: " << to_string(out_msg));
......
...@@ -60,6 +60,8 @@ using namespace cppa::util; ...@@ -60,6 +60,8 @@ using namespace cppa::util;
using cppa::detail::type_to_ptype; using cppa::detail::type_to_ptype;
using cppa::detail::ptype_to_type; using cppa::detail::ptype_to_type;
namespace { const size_t ui32size = sizeof(std::uint32_t); }
struct struct_a { struct struct_a {
int x; int x;
int y; int y;
...@@ -127,19 +129,24 @@ int main() { ...@@ -127,19 +129,24 @@ int main() {
} }
{ {
any_tuple ttup = make_cow_tuple(1, 2, actor_ptr(self)); any_tuple ttup = make_cow_tuple(1, 2, actor_ptr(self));
binary_serializer bs; util::buffer wr_buf;
binary_serializer bs(&wr_buf);
bs << ttup; bs << ttup;
binary_deserializer bd(bs.data(), bs.size()); std::uint32_t serialized_size;
memcpy(&serialized_size, wr_buf.data(), ui32size);
CPPA_CHECK_EQUAL(serialized_size, wr_buf.size() - ui32size);
binary_deserializer bd(wr_buf.data() + ui32size, wr_buf.size());
any_tuple ttup2; any_tuple ttup2;
uniform_typeid<any_tuple>()->deserialize(&ttup2, &bd); uniform_typeid<any_tuple>()->deserialize(&ttup2, &bd);
CPPA_CHECK(ttup == ttup2); CPPA_CHECK(ttup == ttup2);
} }
{ {
// serialize b1 to buf // serialize b1 to buf
binary_serializer bs; util::buffer wr_buf;
binary_serializer bs(&wr_buf);
bs << atuple1; bs << atuple1;
// deserialize b2 from buf // deserialize b2 from buf
binary_deserializer bd(bs.data(), bs.size()); binary_deserializer bd(wr_buf.data() + ui32size, wr_buf.size());
any_tuple atuple2; any_tuple atuple2;
uniform_typeid<any_tuple>()->deserialize(&atuple2, &bd); uniform_typeid<any_tuple>()->deserialize(&atuple2, &bd);
try { try {
...@@ -163,9 +170,10 @@ int main() { ...@@ -163,9 +170,10 @@ int main() {
CPPA_ERROR("msg1str != to_string(msg1)"); CPPA_ERROR("msg1str != to_string(msg1)");
cerr << "to_string(msg1) = " << msg1_tostring << endl; cerr << "to_string(msg1) = " << msg1_tostring << endl;
} }
binary_serializer bs; util::buffer wr_buf;
binary_serializer bs(&wr_buf);
bs << msg1; bs << msg1;
binary_deserializer bd(bs.data(), bs.size()); binary_deserializer bd(wr_buf.data() + ui32size, wr_buf.size());
object obj1; object obj1;
bd >> obj1; bd >> obj1;
object obj2 = from_string(to_string(msg1)); object obj2 = from_string(to_string(msg1));
...@@ -224,10 +232,11 @@ int main() { ...@@ -224,10 +232,11 @@ int main() {
// verify // verify
CPPA_CHECK_EQUAL((to_string(b1)), b1str); { CPPA_CHECK_EQUAL((to_string(b1)), b1str); {
// serialize b1 to buf // serialize b1 to buf
binary_serializer bs; util::buffer wr_buf;
binary_serializer bs(&wr_buf);
bs << b1; bs << b1;
// deserialize b2 from buf // deserialize b2 from buf
binary_deserializer bd(bs.data(), bs.size()); binary_deserializer bd(wr_buf.data() + ui32size, wr_buf.size());
object res; object res;
bd >> res; bd >> res;
CPPA_CHECK_EQUAL(res.type()->name(), "struct_b"); CPPA_CHECK_EQUAL(res.type()->name(), "struct_b");
...@@ -250,10 +259,11 @@ int main() { ...@@ -250,10 +259,11 @@ int main() {
struct_c c1{strmap{{"abc", u"cba" }, { "x", u"y" }}, std::set<int>{9, 4, 5}}; struct_c c1{strmap{{"abc", u"cba" }, { "x", u"y" }}, std::set<int>{9, 4, 5}};
struct_c c2; { struct_c c2; {
// serialize c1 to buf // serialize c1 to buf
binary_serializer bs; util::buffer wr_buf;
binary_serializer bs(&wr_buf);
bs << c1; bs << c1;
// serialize c2 from buf // serialize c2 from buf
binary_deserializer bd(bs.data(), bs.size()); binary_deserializer bd(wr_buf.data() + ui32size, wr_buf.size());
object res; object res;
bd >> res; bd >> res;
CPPA_CHECK_EQUAL(res.type()->name(), "struct_c"); CPPA_CHECK_EQUAL(res.type()->name(), "struct_c");
......
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