Commit 9c4f5804 authored by Dominik Charousset's avatar Dominik Charousset

Merge pull request #1023

parents bf152e63 fc84df5a
......@@ -124,6 +124,17 @@ enum class sec : uint8_t {
socket_operation_failed = 45,
/// A resource is temporarily unavailable or would block.
unavailable_or_would_block,
/// Connection refused because of incompatible CAF versions.
incompatible_versions,
/// Connection refused because of incompatible application IDs.
incompatible_application_ids,
/// The middleman received a malformed BASP message from another node.
malformed_basp_message,
/// The middleman closed a connection because it failed to serialize or
/// deserialize a payload.
serializing_basp_payload_failed,
/// The middleman closed a connection to itself or an already connected node.
redundant_connection,
/// Resolving a path on a remote node failed.
remote_lookup_failed,
};
......
......@@ -18,6 +18,7 @@
#pragma once
#include "caf/sec.hpp"
namespace caf {
namespace io {
......@@ -25,7 +26,8 @@ namespace basp {
/// @addtogroup BASP
/// Denotes the state of a connection between to BASP nodes.
/// Denotes the state of a connection between two BASP nodes. Overlaps with
/// `sec` (these states get converted to an error by the BASP instance).
enum connection_state {
/// Indicates that a connection is established and this node is
/// waiting for the next BASP header.
......@@ -34,14 +36,67 @@ enum connection_state {
/// and is waiting for the data.
await_payload,
/// Indicates that this connection no longer exists.
close_connection
close_connection,
/// See `sec::incompatible_versions`.
incompatible_versions,
/// See `sec::incompatible_application_ids`.
incompatible_application_ids,
/// See `sec::malformed_basp_message`.
malformed_basp_message,
/// See `sec::serializing_basp_payload_failed`.
serializing_basp_payload_failed,
/// See `sec::redundant_connection`.
redundant_connection,
/// See `sec::no_route_to_receiving_node`.
no_route_to_receiving_node
};
/// Returns whether the connection state requries a shutdown of the socket
/// connection.
/// @relates connection_state
inline bool requires_shutdown(connection_state x) {
// Any enum value other than await_header (0) and await_payload (1) signal the
// BASP broker to shutdown the connection.
return static_cast<int>(x) > 1;
}
/// Converts the connection state to a system error code if it holds one of the
/// overlapping values. Otherwise returns `sec::none`.
/// @relates connection_state
inline sec to_sec(connection_state x) {
switch (x) {
default:
return sec::none;
case incompatible_versions:
return sec::incompatible_versions;
case incompatible_application_ids:
return sec::incompatible_application_ids;
case malformed_basp_message:
return sec::malformed_basp_message;
case serializing_basp_payload_failed:
return sec::serializing_basp_payload_failed;
case redundant_connection:
return sec::redundant_connection;
case no_route_to_receiving_node:
return sec::no_route_to_receiving_node;
}
}
/// @relates connection_state
inline std::string to_string(connection_state x) {
return x == await_header ? "await_header"
: (x == await_payload ? "await_payload"
: "close_connection");
// TODO: auto-generate this to_string implementation
static constexpr const char* strings[] = {
"await_header",
"await_payload",
"close_connection",
"incompatible_versions",
"incompatible_application_ids",
"malformed_basp_message",
"serializing_basp_payload_failed",
"redundant_connection",
"no_route_to_receiving_node",
};
return strings[static_cast<size_t>(x)];
}
/// @}
......
......@@ -225,8 +225,8 @@ public:
return system().config();
}
bool handle(execution_unit* ctx, connection_handle hdl, header& hdr,
std::vector<char>* payload);
connection_state handle(execution_unit* ctx, connection_handle hdl,
header& hdr, std::vector<char>* payload);
private:
void forward(execution_unit* ctx, const node_id& dest_node, const header& hdr,
......
......@@ -111,7 +111,7 @@ public:
void set_context(connection_handle hdl);
/// Cleans up any state for `hdl`.
void connection_cleanup(connection_handle hdl);
void connection_cleanup(connection_handle hdl, sec code);
/// Sends a basp::down_message message to a remote node.
void send_basp_down_message(const node_id& nid, actor_id aid, error err);
......
......@@ -54,10 +54,10 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
header& hdr, bool is_payload) {
CAF_LOG_TRACE(CAF_ARG(dm) << CAF_ARG(is_payload));
// function object providing cleanup code on errors
auto err = [&]() -> connection_state {
auto err = [&](connection_state code) {
if (auto nid = tbl_.erase_direct(dm.handle))
callee_.purge_state(nid);
return close_connection;
return code;
};
std::vector<char>* payload = nullptr;
if (is_payload) {
......@@ -65,14 +65,14 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
if (payload->size() != hdr.payload_len) {
CAF_LOG_WARNING("received invalid payload, expected"
<< hdr.payload_len << "bytes, got" << payload->size());
return err();
return err(malformed_basp_message);
}
} else {
binary_deserializer bd{ctx, dm.buf};
auto e = bd(hdr);
if (e || !valid(hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr));
return err();
return err(malformed_basp_message);
}
if (hdr.payload_len > 0) {
CAF_LOG_DEBUG("await payload before processing further");
......@@ -80,9 +80,7 @@ connection_state instance::handle(execution_unit* ctx, new_data_msg& dm,
}
}
CAF_LOG_DEBUG(CAF_ARG(hdr));
if (!handle(ctx, dm.handle, hdr, payload))
return err();
return await_header;
return handle(ctx, dm.handle, hdr, payload);
}
void instance::handle_heartbeat(execution_unit* ctx) {
......@@ -286,18 +284,18 @@ void instance::write_heartbeat(execution_unit* ctx, buffer_type& buf) {
write(ctx, buf, hdr);
}
bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
std::vector<char>* payload) {
connection_state instance::handle(execution_unit* ctx, connection_handle hdl,
header& hdr, std::vector<char>* payload) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(hdr));
// Check payload validity.
if (payload == nullptr) {
if (hdr.payload_len != 0) {
CAF_LOG_WARNING("invalid payload");
return false;
CAF_LOG_WARNING("missing payload");
return malformed_basp_message;
}
} else if (hdr.payload_len != payload->size()) {
CAF_LOG_WARNING("invalid payload");
return false;
CAF_LOG_WARNING("actual payload size differs from advertised size");
return malformed_basp_message;
}
// Dispatch by message type.
switch (hdr.operation) {
......@@ -311,7 +309,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (auto err = bd(source_node, app_ids, aid, sigs)) {
CAF_LOG_WARNING("unable to deserialize payload of server handshake:"
<< ctx->system().render(err));
return false;
return serializing_basp_payload_failed;
}
// Check the application ID.
auto whitelist = get_or(config(), "middleman.app-identifiers",
......@@ -321,20 +319,20 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (i == app_ids.end()) {
CAF_LOG_WARNING("refuse to connect to server due to app ID mismatch:"
<< CAF_ARG(app_ids) << CAF_ARG(whitelist));
return false;
return incompatible_application_ids;
}
// Close connection to ourselves immediately after sending client HS.
if (source_node == this_node_) {
CAF_LOG_DEBUG("close connection to self immediately");
callee_.finalize_handshake(source_node, aid, sigs);
return false;
return redundant_connection;
}
// Close this connection if we already have a direct connection.
if (tbl_.lookup_direct(source_node)) {
CAF_LOG_DEBUG(
"close redundant direct connection:" << CAF_ARG(source_node));
callee_.finalize_handshake(source_node, aid, sigs);
return false;
return redundant_connection;
}
// Add direct route to this node and remove any indirect entry.
CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(source_node));
......@@ -344,7 +342,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
auto path = tbl_.lookup(source_node);
if (!path) {
CAF_LOG_ERROR("no route to host after server handshake");
return false;
return no_route_to_receiving_node;
}
write_client_handshake(ctx, callee_.get_buffer(path->hdl));
callee_.learned_new_node_directly(source_node, was_indirect);
......@@ -359,7 +357,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (auto err = bd(source_node)) {
CAF_LOG_WARNING("unable to deserialize payload of client handshake:"
<< ctx->system().render(err));
return false;
return serializing_basp_payload_failed;
}
// Drop repeated handshakes.
if (tbl_.lookup_direct(source_node)) {
......@@ -383,11 +381,11 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
CAF_LOG_WARNING(
"unable to deserialize source and destination for routed message:"
<< ctx->system().render(err));
return false;
return serializing_basp_payload_failed;
}
if (dest_node != this_node_) {
forward(ctx, dest_node, hdr, *payload);
return true;
return await_header;
}
auto last_hop = tbl_.lookup_direct(hdl);
if (source_node != none && source_node != this_node_
......@@ -441,7 +439,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (auto err = bd(source_node, dest_node)) {
CAF_LOG_WARNING("unable to deserialize payload of monitor message:"
<< ctx->system().render(err));
return false;
return serializing_basp_payload_failed;
}
if (dest_node == this_node_)
callee_.proxy_announced(source_node, hdr.dest_actor);
......@@ -458,7 +456,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (auto err = bd(source_node, dest_node, fail_state)) {
CAF_LOG_WARNING("unable to deserialize payload of down message:"
<< ctx->system().render(err));
return false;
return serializing_basp_payload_failed;
}
if (dest_node == this_node_) {
// Delay this message to make sure we don't skip in-flight messages.
......@@ -481,10 +479,10 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
}
default: {
CAF_LOG_ERROR("invalid operation");
return false;
return malformed_basp_message;
}
}
return true;
return await_header;
}
void instance::forward(execution_unit* ctx, const node_id& dest_node,
......
......@@ -128,8 +128,8 @@ behavior basp_broker::make_behavior() {
auto& ctx = *this_context;
auto next = instance.handle(context(), msg, ctx.hdr,
ctx.cstate == basp::await_payload);
if (next == basp::close_connection) {
connection_cleanup(msg.handle);
if (requires_shutdown(next)) {
connection_cleanup(msg.handle, to_sec(next));
close(msg.handle);
return;
}
......@@ -226,7 +226,9 @@ behavior basp_broker::make_behavior() {
delete_atom::value, msg.handle));
},
// received from the message handler above for connection_closed_msg
[=](delete_atom, connection_handle hdl) { connection_cleanup(hdl); },
[=](delete_atom, connection_handle hdl) {
connection_cleanup(hdl, sec::none);
},
// received from underlying broker implementation
[=](const acceptor_closed_msg& msg) {
CAF_LOG_TRACE("");
......@@ -564,8 +566,8 @@ void basp_broker::set_context(connection_handle hdl) {
t_last_hop = &i->second.id;
}
void basp_broker::connection_cleanup(connection_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
void basp_broker::connection_cleanup(connection_handle hdl, sec code) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(code));
// Remove handle from the routing table and clean up any node-specific state
// we might still have.
if (auto nid = instance.tbl().erase_direct(hdl))
......@@ -577,8 +579,9 @@ void basp_broker::connection_cleanup(connection_handle hdl) {
auto& ref = i->second;
CAF_ASSERT(i->first == ref.hdl);
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
CAF_LOG_DEBUG("connection closed during handshake:" << CAF_ARG(code));
auto x = code != sec::none ? code : sec::disconnect_during_handshake;
ref.callback->deliver(x);
}
ctx.erase(i);
}
......
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