Commit 863f2977 authored by Dominik Charousset's avatar Dominik Charousset

Fix bug with new node_id serialization format

parent 46019288
...@@ -56,9 +56,11 @@ class proper_actor_base : public Policies::resume_policy::template ...@@ -56,9 +56,11 @@ class proper_actor_base : public Policies::resume_policy::template
message msg, message msg,
execution_unit* eu) override { execution_unit* eu) override {
CAF_PUSH_AID(dptr()->id()); CAF_PUSH_AID(dptr()->id());
/*
CAF_LOG_DEBUG(CAF_TARG(sender, to_string) CAF_LOG_DEBUG(CAF_TARG(sender, to_string)
<< ", " << CAF_MARG(mid, integer_value) << ", " << ", " << CAF_MARG(mid, integer_value) << ", "
<< CAF_TARG(msg, to_string)); << CAF_TARG(msg, to_string));
*/
scheduling_policy().enqueue(dptr(), sender, mid, msg, eu); scheduling_policy().enqueue(dptr(), sender, mid, msg, eu);
} }
......
...@@ -46,8 +46,8 @@ void actor_namespace::write(serializer* sink, const actor_addr& addr) { ...@@ -46,8 +46,8 @@ void actor_namespace::write(serializer* sink, const actor_addr& addr) {
node_id::host_id_type zero; node_id::host_id_type zero;
std::fill(zero.begin(), zero.end(), 0); std::fill(zero.begin(), zero.end(), 0);
sink->write_value(static_cast<actor_id>(0)); // actor id sink->write_value(static_cast<actor_id>(0)); // actor id
sink->write_value(static_cast<uint32_t>(0)); // process id
sink->write_raw(node_id::host_id_size, zero.data()); // host id sink->write_raw(node_id::host_id_size, zero.data()); // host id
sink->write_value(static_cast<uint32_t>(0)); // process id
} else { } else {
// register locally running actors to be able to deserialize them later // register locally running actors to be able to deserialize them later
if (!addr.is_remote()) { if (!addr.is_remote()) {
...@@ -56,8 +56,8 @@ void actor_namespace::write(serializer* sink, const actor_addr& addr) { ...@@ -56,8 +56,8 @@ void actor_namespace::write(serializer* sink, const actor_addr& addr) {
} }
auto pinf = addr.node(); auto pinf = addr.node();
sink->write_value(addr.id()); // actor id sink->write_value(addr.id()); // actor id
sink->write_value(pinf.process_id()); // process id
sink->write_raw(node_id::host_id_size, pinf.host_id().data()); // host id sink->write_raw(node_id::host_id_size, pinf.host_id().data()); // host id
sink->write_value(pinf.process_id()); // process id
} }
} }
...@@ -65,8 +65,8 @@ actor_addr actor_namespace::read(deserializer* source) { ...@@ -65,8 +65,8 @@ actor_addr actor_namespace::read(deserializer* source) {
CAF_REQUIRE(source != nullptr); CAF_REQUIRE(source != nullptr);
node_id::host_id_type hid; node_id::host_id_type hid;
auto aid = source->read<uint32_t>(); // actor id auto aid = source->read<uint32_t>(); // actor id
auto pid = source->read<uint32_t>(); // process id
source->read_raw(node_id::host_id_size, hid.data()); // host id source->read_raw(node_id::host_id_size, hid.data()); // host id
auto pid = source->read<uint32_t>(); // process id
node_id this_node = detail::singletons::get_node_id(); node_id this_node = detail::singletons::get_node_id();
if (aid == 0 && pid == 0) { if (aid == 0 && pid == 0) {
// 0:0 identifies an invalid actor // 0:0 identifies an invalid actor
......
...@@ -139,21 +139,27 @@ class string_serializer : public serializer, public dummy_backend { ...@@ -139,21 +139,27 @@ class string_serializer : public serializer, public dummy_backend {
out(mout), out(mout),
m_namespace(*this), m_namespace(*this),
m_after_value(false), m_after_value(false),
m_obj_just_opened(false) {} m_obj_just_opened(false) {
// nop
}
void begin_object(const uniform_type_info* uti) { void begin_object(const uniform_type_info* uti) {
clear(); clear();
string tname = uti->name(); string tname = uti->name();
m_open_objects.push(tname); m_open_objects.push(tname);
// do not print type names for strings and atoms // do not print type names for strings and atoms
if (!isbuiltin(tname)) if (!isbuiltin(tname) && tname != "@message") {
// suppress output of "@message ( ... )" because it's redundant
// since each message immediately calls begin_object(...)
// for the typed tuple it's containing
out << tname; out << tname;
else if (tname.compare(0, 3, "@<>") == 0) { m_obj_just_opened = true;
} else if (tname.compare(0, 3, "@<>") == 0) {
std::vector<string> subtypes; std::vector<string> subtypes;
split(subtypes, tname, is_any_of("+"), token_compress_on); split(subtypes, tname, is_any_of("+"), token_compress_on);
}
m_obj_just_opened = true; m_obj_just_opened = true;
} }
}
void end_object() { void end_object() {
if (m_obj_just_opened) { if (m_obj_just_opened) {
...@@ -162,7 +168,8 @@ class string_serializer : public serializer, public dummy_backend { ...@@ -162,7 +168,8 @@ class string_serializer : public serializer, public dummy_backend {
} }
m_after_value = true; m_after_value = true;
if (!m_open_objects.empty()) { if (!m_open_objects.empty()) {
if (!isbuiltin(m_open_objects.top())) { auto& open_obj = m_open_objects.top();
if (!isbuiltin(open_obj) && open_obj != "@message") {
out << (m_after_value ? " )" : ")"); out << (m_after_value ? " )" : ")");
} }
m_open_objects.pop(); m_open_objects.pop();
...@@ -318,6 +325,12 @@ class string_deserializer : public deserializer, public dummy_backend { ...@@ -318,6 +325,12 @@ class string_deserializer : public deserializer, public dummy_backend {
void read_value(primitive_variant& storage) override { void read_value(primitive_variant& storage) override {
integrity_check(); integrity_check();
skip_space_and_comma(); skip_space_and_comma();
if (m_open_objects.top() == "@node") {
// example node_id: c5c978f5bc5c7e4975e407bb03c751c9374480d9:63768
// deserialization will call read_raw() followed by read_value()
// and ':' must be ignored
consume(':');
}
string::iterator substr_end; string::iterator substr_end;
auto find_if_cond = [](char c)->bool { auto find_if_cond = [](char c)->bool {
switch (c) { switch (c) {
...@@ -402,8 +415,9 @@ class string_deserializer : public deserializer, public dummy_backend { ...@@ -402,8 +415,9 @@ class string_deserializer : public deserializer, public dummy_backend {
void read_raw(size_t buf_size, void* vbuf) override { void read_raw(size_t buf_size, void* vbuf) override {
if (buf_size == node_id::host_id_size && !m_open_objects.empty() if (buf_size == node_id::host_id_size && !m_open_objects.empty()
&& m_open_objects.top() == "@node") { && (m_open_objects.top() == "@actor"
// node ids are formatted as process_id@host_id || m_open_objects.top() == "@actor_addr")) {
// actor addresses are formatted as actor_id@host_id:process_id
// this read_raw reads the host_id, i.e., we need // this read_raw reads the host_id, i.e., we need
// to skip the '@' character // to skip the '@' character
consume('@'); consume('@');
......
...@@ -339,14 +339,14 @@ void deserialize_impl(message& atref, deserializer* source) { ...@@ -339,14 +339,14 @@ void deserialize_impl(message& atref, deserializer* source) {
} }
void serialize_impl(const node_id& nid, serializer* sink) { void serialize_impl(const node_id& nid, serializer* sink) {
sink->write_value(nid.process_id());
sink->write_raw(nid.host_id().size(), nid.host_id().data()); sink->write_raw(nid.host_id().size(), nid.host_id().data());
sink->write_value(nid.process_id());
} }
void deserialize_impl(node_id& nid, deserializer* source) { void deserialize_impl(node_id& nid, deserializer* source) {
node_id::host_id_type hid; node_id::host_id_type hid;
auto pid = source->read<uint32_t>();
source->read_raw(node_id::host_id_size, hid.data()); source->read_raw(node_id::host_id_size, hid.data());
auto pid = source->read<uint32_t>();
auto is_zero = [](uint8_t value) { return value == 0; }; auto is_zero = [](uint8_t value) { return value == 0; };
if (pid == 0 && std::all_of(hid.begin(), hid.end(), is_zero)) { if (pid == 0 && std::all_of(hid.begin(), hid.end(), is_zero)) {
// invalid process information // invalid process information
......
...@@ -53,14 +53,12 @@ namespace { ...@@ -53,14 +53,12 @@ namespace {
struct struct_a { struct struct_a {
int x; int x;
int y; int y;
}; };
struct struct_b { struct struct_b {
struct_a a; struct_a a;
int z; int z;
list<int> ints; list<int> ints;
}; };
using strmap = map<string, u16string>; using strmap = map<string, u16string>;
...@@ -68,12 +66,10 @@ using strmap = map<string, u16string>; ...@@ -68,12 +66,10 @@ using strmap = map<string, u16string>;
struct struct_c { struct struct_c {
strmap strings; strmap strings;
set<int> ints; set<int> ints;
}; };
struct raw_struct { struct raw_struct {
string str; string str;
}; };
bool operator==(const raw_struct& lhs, const raw_struct& rhs) { bool operator==(const raw_struct& lhs, const raw_struct& rhs) {
...@@ -118,7 +114,6 @@ enum class test_enum { ...@@ -118,7 +114,6 @@ enum class test_enum {
a, a,
b, b,
c c
}; };
} // namespace <anonymous> } // namespace <anonymous>
...@@ -145,7 +140,10 @@ int main() { ...@@ -145,7 +140,10 @@ int main() {
auto nid_str = to_string(nid); auto nid_str = to_string(nid);
CAF_PRINT("nid_str = " << nid_str); CAF_PRINT("nid_str = " << nid_str);
auto nid2 = from_string<node_id>(nid_str); auto nid2 = from_string<node_id>(nid_str);
CAF_CHECK(nid2 && nid == *nid2); CAF_CHECK(nid2);
if (nid2) {
CAF_CHECK_EQUAL(to_string(nid), to_string(*nid2));
}
/* /*
auto oarr = new detail::object_array; auto oarr = new detail::object_array;
......
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