Commit 803e718d authored by Dominik Charousset's avatar Dominik Charousset

Restore BASP to version 1

Since the future development of BASP is unclear at this time, it makes no sense
to introduce breaking changes to the protocol that are likely to get broken
again soon.
parent 99556335
...@@ -85,7 +85,7 @@ namespace basp { ...@@ -85,7 +85,7 @@ namespace basp {
/// The current BASP version. Different BASP versions will not /// The current BASP version. Different BASP versions will not
/// be able to exchange messages. /// be able to exchange messages.
constexpr uint64_t version = 2; constexpr uint64_t version = 1;
/// Storage type for raw bytes. /// Storage type for raw bytes.
using buffer_type = std::vector<char>; using buffer_type = std::vector<char>;
...@@ -123,13 +123,7 @@ enum class message_type : uint32_t { ...@@ -123,13 +123,7 @@ enum class message_type : uint32_t {
/// that has been terminated. /// that has been terminated.
/// ///
/// ![](kill_proxy_instance.png) /// ![](kill_proxy_instance.png)
kill_proxy_instance = 0x04, kill_proxy_instance = 0x04
/// Informs the receiving node that a message it sent
/// did not reach its destination.
///
/// ![](dispatch_error.png)
dispatch_error = 0x05
}; };
/// @relates message_type /// @relates message_type
......
...@@ -49,8 +49,6 @@ std::string to_string(message_type x) { ...@@ -49,8 +49,6 @@ std::string to_string(message_type x) {
return "announce_proxy_instance"; return "announce_proxy_instance";
case basp::message_type::kill_proxy_instance: case basp::message_type::kill_proxy_instance:
return "kill_proxy_instance"; return "kill_proxy_instance";
case basp::message_type::dispatch_error:
return "dispatch_error";
default: default:
return "???"; return "???";
} }
...@@ -72,24 +70,24 @@ std::string to_string(const header &hdr) { ...@@ -72,24 +70,24 @@ std::string to_string(const header &hdr) {
void read_hdr(deserializer& source, header& hdr) { void read_hdr(deserializer& source, header& hdr) {
uint32_t operation; uint32_t operation;
source.read(operation);
hdr.operation = static_cast<basp::message_type>(operation);
source.read(hdr.payload_len)
.read(hdr.operation_data);
hdr.source_node.deserialize(source); hdr.source_node.deserialize(source);
hdr.dest_node.deserialize(source); hdr.dest_node.deserialize(source);
source.read(hdr.source_actor) source.read(hdr.source_actor)
.read(hdr.dest_actor); .read(hdr.dest_actor)
.read(hdr.payload_len)
.read(operation)
.read(hdr.operation_data);
hdr.operation = static_cast<basp::message_type>(operation);
} }
void write_hdr(serializer& sink, const header& hdr) { void write_hdr(serializer& sink, const header& hdr) {
sink.write(static_cast<uint32_t>(hdr.operation))
.write(hdr.payload_len)
.write(hdr.operation_data);
hdr.source_node.serialize(sink); hdr.source_node.serialize(sink);
hdr.dest_node.serialize(sink); hdr.dest_node.serialize(sink);
sink.write(hdr.source_actor) sink.write(hdr.source_actor)
.write(hdr.dest_actor); .write(hdr.dest_actor)
.write(hdr.payload_len)
.write(static_cast<uint32_t>(hdr.operation))
.write(hdr.operation_data);
} }
bool operator==(const header& lhs, const header& rhs) { bool operator==(const header& lhs, const header& rhs) {
...@@ -116,10 +114,10 @@ bool zero(T val) { ...@@ -116,10 +114,10 @@ bool zero(T val) {
bool server_handshake_valid(const header& hdr) { bool server_handshake_valid(const header& hdr) {
return valid(hdr.source_node) return valid(hdr.source_node)
&& ! valid(hdr.dest_node) && ! valid(hdr.dest_node)
&& zero(hdr.source_actor)
&& zero(hdr.dest_actor) && zero(hdr.dest_actor)
&& ! zero(hdr.payload_len) && ! zero(hdr.operation_data)
&& ! zero(hdr.operation_data); && ( (! zero(hdr.source_actor) && ! zero(hdr.payload_len))
|| (zero(hdr.source_actor) && zero(hdr.payload_len)));
} }
bool client_handshake_valid(const header& hdr) { bool client_handshake_valid(const header& hdr) {
...@@ -159,16 +157,6 @@ bool kill_proxy_instance_valid(const header& hdr) { ...@@ -159,16 +157,6 @@ bool kill_proxy_instance_valid(const header& hdr) {
&& ! zero(hdr.operation_data); && ! zero(hdr.operation_data);
} }
bool dispatch_error_valid(const header& hdr) {
return valid(hdr.source_node)
&& valid(hdr.dest_node)
&& hdr.source_node != hdr.dest_node
&& zero(hdr.source_actor)
&& zero(hdr.dest_actor)
&& ! zero(hdr.payload_len)
&& ! zero(hdr.operation_data);
}
} // namespace <anonymous> } // namespace <anonymous>
bool valid(const header& hdr) { bool valid(const header& hdr) {
...@@ -185,8 +173,6 @@ bool valid(const header& hdr) { ...@@ -185,8 +173,6 @@ bool valid(const header& hdr) {
return announce_proxy_instance_valid(hdr); return announce_proxy_instance_valid(hdr);
case message_type::kill_proxy_instance: case message_type::kill_proxy_instance:
return kill_proxy_instance_valid(hdr); return kill_proxy_instance_valid(hdr);
case message_type::dispatch_error:
return dispatch_error_valid(hdr);
} }
} }
...@@ -329,7 +315,8 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr, ...@@ -329,7 +315,8 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
} }
CAF_LOG_DEBUG("hdr = " << to_string(hdr)); CAF_LOG_DEBUG("hdr = " << to_string(hdr));
if (! valid(hdr)) { if (! valid(hdr)) {
CAF_LOG_WARNING("received invalid header"); CAF_LOG_WARNING("received invalid header of type "
<< to_string(hdr.operation));
return err(); return err();
} }
// needs forwarding? // needs forwarding?
...@@ -368,13 +355,13 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr, ...@@ -368,13 +355,13 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
// handle message to ourselves // handle message to ourselves
switch (hdr.operation) { switch (hdr.operation) {
case message_type::server_handshake: { case message_type::server_handshake: {
if (! payload_valid()) actor_id aid = invalid_actor_id;
return err(); std::set<std::string> sigs;
if (payload_valid()) {
binary_deserializer bd{payload->data(), payload->size(), binary_deserializer bd{payload->data(), payload->size(),
&get_namespace()}; &get_namespace()};
actor_id aid;
std::set<std::string> sigs;
bd >> aid >> sigs; bd >> aid >> sigs;
}
// close self connection after handshake is done // close self connection after handshake is done
if (hdr.source_node == this_node_) { if (hdr.source_node == this_node_) {
CAF_LOG_INFO("close connection to self immediately"); CAF_LOG_INFO("close connection to self immediately");
...@@ -427,15 +414,6 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr, ...@@ -427,15 +414,6 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
message_id::from_integer_value(hdr.operation_data)); message_id::from_integer_value(hdr.operation_data));
break; break;
} }
case message_type::dispatch_error:
switch (static_cast<error>(hdr.operation_data)) {
case error::loop_detected:
case error::no_route_to_destination:
// TODO: do some actual error handling
CAF_LOG_ERROR("a message got lost");
break;
}
break;
case message_type::announce_proxy_instance: case message_type::announce_proxy_instance:
callee_.proxy_announced(hdr.source_node, hdr.dest_actor); callee_.proxy_announced(hdr.source_node, hdr.dest_actor);
break; break;
...@@ -526,9 +504,8 @@ size_t instance::remove_published_actor(const actor_addr& whom, uint16_t port, ...@@ -526,9 +504,8 @@ size_t instance::remove_published_actor(const actor_addr& whom, uint16_t port,
++i; ++i;
} }
} }
return result;
} }
return 0; return result;
} }
bool instance::dispatch(const actor_addr& sender, const actor_addr& receiver, bool instance::dispatch(const actor_addr& sender, const actor_addr& receiver,
...@@ -561,13 +538,13 @@ void instance::write(buffer_type& buf, ...@@ -561,13 +538,13 @@ void instance::write(buffer_type& buf,
payload_writer* pw) { payload_writer* pw) {
if (! pw) { if (! pw) {
binary_serializer bs{std::back_inserter(buf), &get_namespace()}; binary_serializer bs{std::back_inserter(buf), &get_namespace()};
bs.write(static_cast<uint32_t>(operation))
.write(uint32_t{0})
.write(operation_data);
source_node.serialize(bs); source_node.serialize(bs);
dest_node.serialize(bs); dest_node.serialize(bs);
bs.write(source_actor) bs.write(source_actor)
.write(dest_actor); .write(dest_actor)
.write(uint32_t{0})
.write(static_cast<uint32_t>(operation))
.write(operation_data);
} else { } else {
// reserve space in the buffer to write the payload later on // reserve space in the buffer to write the payload later on
auto wr_pos = static_cast<ptrdiff_t>(buf.size()); auto wr_pos = static_cast<ptrdiff_t>(buf.size());
...@@ -581,13 +558,13 @@ void instance::write(buffer_type& buf, ...@@ -581,13 +558,13 @@ void instance::write(buffer_type& buf,
// write broker message to the reserved space // write broker message to the reserved space
binary_serializer bs2{buf.begin() + wr_pos, &get_namespace()}; binary_serializer bs2{buf.begin() + wr_pos, &get_namespace()};
auto plen = static_cast<uint32_t>(buf.size() - pl_pos); auto plen = static_cast<uint32_t>(buf.size() - pl_pos);
bs2.write(static_cast<uint32_t>(operation))
.write(plen)
.write(operation_data);
source_node.serialize(bs2); source_node.serialize(bs2);
dest_node.serialize(bs2); dest_node.serialize(bs2);
bs2.write(source_actor) bs2.write(source_actor)
.write(dest_actor); .write(dest_actor)
.write(plen)
.write(static_cast<uint32_t>(operation))
.write(operation_data);
if (payload_len) if (payload_len)
*payload_len = plen; *payload_len = plen;
} }
...@@ -608,7 +585,6 @@ void instance::write_server_handshake(buffer_type& out_buf, ...@@ -608,7 +585,6 @@ void instance::write_server_handshake(buffer_type& out_buf,
} }
auto writer = make_callback([&](serializer& sink) { auto writer = make_callback([&](serializer& sink) {
if (! pa) { if (! pa) {
sink << invalid_actor_id << static_cast<uint32_t>(0);
return; return;
} }
sink << pa->first.id() << static_cast<uint32_t>(pa->second.size()); sink << pa->first.id() << static_cast<uint32_t>(pa->second.size());
...@@ -617,7 +593,7 @@ void instance::write_server_handshake(buffer_type& out_buf, ...@@ -617,7 +593,7 @@ void instance::write_server_handshake(buffer_type& out_buf,
}); });
header hdr{message_type::server_handshake, 0, version, header hdr{message_type::server_handshake, 0, version,
this_node_, invalid_node_id, this_node_, invalid_node_id,
invalid_actor_id, invalid_actor_id}; pa ? pa->first.id() : invalid_actor_id, invalid_actor_id};
write(out_buf, hdr, &writer); write(out_buf, hdr, &writer);
} }
......
...@@ -223,16 +223,22 @@ public: ...@@ -223,16 +223,22 @@ public:
mpx_->accept_connection(src); mpx_->accept_connection(src);
// technically, the server handshake arrives // technically, the server handshake arrives
// before we send the client handshake // before we send the client handshake
mock(hdl, auto m = mock(hdl,
{basp::message_type::client_handshake, 0, 0, {basp::message_type::client_handshake, 0, 0,
remote_node(i), this_node(), remote_node(i), this_node(),
invalid_actor_id, invalid_actor_id}) invalid_actor_id, invalid_actor_id});
.expect(hdl, if (published_actor_id != invalid_actor_id)
m.expect(hdl,
{basp::message_type::server_handshake, 0, basp::version, {basp::message_type::server_handshake, 0, basp::version,
this_node(), invalid_node_id, this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id}, published_actor_id, invalid_actor_id},
published_actor_id, published_actor_id,
published_actor_ifs); published_actor_ifs);
else
m.expect(hdl,
{basp::message_type::server_handshake, 0, basp::version,
this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id});
// test whether basp instance correctly updates the // test whether basp instance correctly updates the
// routing table upon receiving client handshakes // routing table upon receiving client handshakes
auto path = tbl().lookup(remote_node(i)); auto path = tbl().lookup(remote_node(i));
...@@ -351,18 +357,15 @@ CAF_TEST(empty_server_handshake) { ...@@ -351,18 +357,15 @@ CAF_TEST(empty_server_handshake) {
basp::header hdr; basp::header hdr;
buffer payload; buffer payload;
std::tie(hdr, payload) = from_buf(buf); std::tie(hdr, payload) = from_buf(buf);
auto is_zero = [](char c) {
return c == 0;
};
basp::header expected{basp::message_type::server_handshake, basp::header expected{basp::message_type::server_handshake,
sizeof(actor_id) + sizeof(uint32_t), 0,
basp::version, basp::version,
this_node(), invalid_node_id, this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id}; invalid_actor_id, invalid_actor_id};
CAF_CHECK(basp::valid(hdr)); CAF_CHECK(basp::valid(hdr));
CAF_CHECK(basp::is_handshake(hdr)); CAF_CHECK(basp::is_handshake(hdr));
CAF_CHECK(hdr == expected); CAF_CHECK(hdr == expected);
CAF_CHECK(all_of(payload.begin(), payload.end(), is_zero)); CAF_CHECK(payload.empty());
} }
CAF_TEST(non_empty_server_handshake) { CAF_TEST(non_empty_server_handshake) {
...@@ -375,7 +378,7 @@ CAF_TEST(non_empty_server_handshake) { ...@@ -375,7 +378,7 @@ CAF_TEST(non_empty_server_handshake) {
buffer expected_buf; buffer expected_buf;
basp::header expected{basp::message_type::server_handshake, 0, basp::version, basp::header expected{basp::message_type::server_handshake, 0, basp::version,
this_node(), invalid_node_id, this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id}; self()->id(), invalid_actor_id};
to_buf(expected_buf, expected, nullptr, to_buf(expected_buf, expected, nullptr,
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"}); self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"});
CAF_CHECK(hexstr(buf) == hexstr(expected_buf)); CAF_CHECK(hexstr(buf) == hexstr(expected_buf));
...@@ -456,7 +459,7 @@ CAF_TEST(remote_actor_and_send) { ...@@ -456,7 +459,7 @@ CAF_TEST(remote_actor_and_send) {
mock(remote_hdl(0), mock(remote_hdl(0),
{basp::message_type::server_handshake, 0, basp::version, {basp::message_type::server_handshake, 0, basp::version,
remote_node(0), invalid_node_id, remote_node(0), invalid_node_id,
invalid_actor_id, invalid_actor_id}, pseudo_remote(0)->id(), invalid_actor_id},
pseudo_remote(0)->id(), pseudo_remote(0)->id(),
uint32_t{0}) uint32_t{0})
.expect(remote_hdl(0), .expect(remote_hdl(0),
......
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