Commit 6374186c authored by Joseph Noir's avatar Joseph Noir

Move ep context to separate file, start ordering

parent eb05530e
......@@ -52,6 +52,7 @@ struct header {
node_id dest_node;
actor_id source_actor;
actor_id dest_actor;
uint16_t sequence_number;
inline header(message_type m_operation, uint8_t m_flags,
uint32_t m_payload_len, uint64_t m_operation_data,
......@@ -64,7 +65,25 @@ struct header {
source_node(std::move(m_source_node)),
dest_node(std::move(m_dest_node)),
source_actor(m_source_actor),
dest_actor(m_dest_actor) {
dest_actor(m_dest_actor),
sequence_number(0) {
// nop
}
inline header(message_type m_operation, uint8_t m_flags,
uint32_t m_payload_len, uint64_t m_operation_data,
node_id m_source_node, node_id m_dest_node,
actor_id m_source_actor, actor_id m_dest_actor,
uint16_t m_sequence_number)
: operation(m_operation),
flags(m_flags),
payload_len(m_payload_len),
operation_data(m_operation_data),
source_node(std::move(m_source_node)),
dest_node(std::move(m_dest_node)),
source_actor(m_source_actor),
dest_actor(m_dest_actor),
sequence_number(m_sequence_number) {
// nop
}
......@@ -88,7 +107,7 @@ typename Inspector::result_type inspect(Inspector& f, header& hdr) {
meta::omittable(), pad,
meta::omittable(), pad,
hdr.flags, hdr.payload_len, hdr.operation_data, hdr.source_node,
hdr.dest_node, hdr.source_actor, hdr.dest_actor);
hdr.dest_node, hdr.source_actor, hdr.dest_actor, hdr.sequence_number);
}
/// @relates header
......@@ -118,7 +137,8 @@ bool valid(const header& hdr);
constexpr size_t header_size = node_id::serialized_size * 2
+ sizeof(actor_id) * 2
+ sizeof(uint32_t) * 2
+ sizeof(uint64_t);
+ sizeof(uint64_t)
+ sizeof(uint16_t);
/// @}
......
......@@ -26,6 +26,7 @@
#include "caf/io/hook.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/endpoint_context.hpp"
#include "caf/io/basp/header.hpp"
#include "caf/io/basp/buffer_type.hpp"
......@@ -122,10 +123,10 @@ public:
bool is_payload);
/// Handles a received datagram
bool handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr);
bool handle(execution_unit* ctx, new_datagram_msg& dm, endpoint_context& ep);
/// Handles a new UDP endpoint msg
bool handle(execution_unit* ctx, new_endpoint_msg& em, header& hdr);
bool handle(execution_unit* ctx, new_endpoint_msg& em, endpoint_context& ep);
/// Sends heartbeat messages to all valid nodes those are directly connected.
void handle_heartbeat(execution_unit* ctx);
......
......@@ -35,9 +35,11 @@
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/io/basp/all.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/visitors.hpp"
#include "caf/io/typed_broker.hpp"
#include "caf/io/endpoint_context.hpp"
namespace caf {
namespace io {
......@@ -96,24 +98,6 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// nop
}
// stores meta information for active endpoints
struct endpoint_context {
// denotes what message we expect from the remote node next
basp::connection_state cstate;
// our currently processed BASP header
basp::header hdr;
// the handle for I/O operations
variant<connection_handle, dgram_scribe_handle> hdl;
// network-agnostic node identifier
node_id id;
// port
uint16_t remote_port;
// pending operations to be performed after handshake completed
optional<response_promise> callback;
// TODO: ordering
// TODO: reliability
};
void set_context(connection_handle hdl);
void set_context(dgram_scribe_handle hdl);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_IO_ENDPOINT_CONTEXT_HPP
#define CAF_IO_ENDPOINT_CONTEXT_HPP
#include <unordered_map>
#include "caf/variant.hpp"
#include "caf/response_promise.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/io/dgram_scribe_handle.hpp"
#include "caf/io/basp/header.hpp"
#include "caf/io/basp/connection_state.hpp"
namespace caf {
namespace io {
// stores meta information for active endpoints
struct endpoint_context {
// denotes what message we expect from the remote node next
basp::connection_state cstate;
// our currently processed BASP header
basp::header hdr;
// the handle for I/O operations
variant<connection_handle, dgram_scribe_handle> hdl;
// network-agnostic node identifier
node_id id;
// port
uint16_t remote_port;
// pending operations to be performed after handshake completed
optional<response_promise> callback;
uint16_t seq_incoming;
uint16_t seq_outgoing;
std::unordered_map<uint16_t, std::pair<basp::header,std::vector<char>>> pending;
};
} // namespace io
} // namespace caf
#endif // CAF_IO_ENDPOINT_CONTEXT_HPP
......@@ -462,7 +462,8 @@ void basp_broker_state::set_context(connection_handle hdl) {
basp::header{basp::message_type::server_handshake,
0, 0, 0, none, none,
invalid_actor_id, invalid_actor_id},
hdl, none, 0, none
hdl, none, 0, none,
0, 0
}
).first;
}
......@@ -482,7 +483,8 @@ void basp_broker_state::set_context(dgram_scribe_handle hdl) {
basp::header{basp::message_type::server_handshake,
0, 0, 0, none, none,
invalid_actor_id, invalid_actor_id},
hdl, none, 0, none
hdl, none, 0, none,
0, 0
}
).first;
}
......@@ -580,9 +582,7 @@ behavior basp_broker::make_behavior() {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
state.set_context(msg.handle);
auto& ctx = *state.this_context;
// TODO: ordering
// TODO: reliability?
auto is_open = state.instance.handle(context(), msg, ctx.hdr);
auto is_open = state.instance.handle(context(), msg, ctx);
if (!is_open) {
if (ctx.callback) {
CAF_LOG_WARNING("failed to handshake with remote node"
......@@ -679,7 +679,7 @@ behavior basp_broker::make_behavior() {
CAF_LOG_TRACE(CAF_ARG(msg.handle));
state.set_context(msg.handle);
auto& ctx = *state.this_context;
auto is_open = state.instance.handle(context(), msg, ctx.hdr);
auto is_open = state.instance.handle(context(), msg, ctx);
if (!is_open) {
if (ctx.callback) {
CAF_LOG_WARNING("failed to handshake with remote node"
......
......@@ -44,7 +44,8 @@ std::string to_string(const header &hdr) {
<< to_string(hdr.source_node) << ", "
<< to_string(hdr.dest_node) << ", "
<< hdr.source_actor << ", "
<< hdr.dest_actor
<< hdr.dest_actor << ", "
<< hdr.sequence_number
<< "}";
return oss.str();
}
......@@ -57,7 +58,8 @@ bool operator==(const header& lhs, const header& rhs) {
&& lhs.source_node == rhs.source_node
&& lhs.dest_node == rhs.dest_node
&& lhs.source_actor == rhs.source_actor
&& lhs.dest_actor == rhs.dest_actor;
&& lhs.dest_actor == rhs.dest_actor
&& lhs.sequence_number == rhs.sequence_number;
}
namespace {
......
......@@ -118,7 +118,8 @@ connection_state instance::handle(execution_unit* ctx,
return await_header;
}
bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
bool instance::handle(execution_unit* ctx, new_datagram_msg& dm,
endpoint_context& ep) {
using itr_t = std::vector<char>::iterator;
// function object providing cleanup code on errors
auto err = [&]() -> bool {
......@@ -137,28 +138,32 @@ bool instance::handle(execution_unit* ctx, new_datagram_msg& dm, header& hdr) {
dm.buf.resize(basp::header_size);
// extract header
binary_deserializer bd{ctx, dm.buf};
auto e = bd(hdr);
if (e || !valid(hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr));
auto e = bd(ep.hdr);
if (e || !valid(ep.hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(ep.hdr));
std::cerr << "Received invalid header!" << std::endl;
return err();
}
CAF_LOG_DEBUG(CAF_ARG(hdr));
CAF_LOG_DEBUG(CAF_ARG(ep.hdr));
std::vector<char>* payload = nullptr;
if (hdr.payload_len > 0) {
if (ep.hdr.payload_len > 0) {
payload = &pl_buf;
if (payload->size() != hdr.payload_len) {
if (payload->size() != ep.hdr.payload_len) {
CAF_LOG_WARNING("received invalid payload");
return err();
}
}
if (!handle_msg(ctx, dm.handle, hdr, payload, false, none))
// TODO: Ordering
// TODO: Reliability
if (!handle_msg(ctx, dm.handle, ep.hdr, payload, false, none))
return err();
return true;
};
bool instance::handle(execution_unit* ctx, new_endpoint_msg& em, header& hdr) {
bool instance::handle(execution_unit* ctx, new_endpoint_msg& em,
endpoint_context& ep) {
using itr_t = std::vector<char>::iterator;
// function object providing cleanup code on errors
auto err = [&]() -> bool {
......@@ -177,22 +182,22 @@ bool instance::handle(execution_unit* ctx, new_endpoint_msg& em, header& hdr) {
em.buf.resize(basp::header_size);
// extract header
binary_deserializer bd{ctx, em.buf};
auto e = bd(hdr);
if (e || !valid(hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(hdr));
auto e = bd(ep.hdr);
if (e || !valid(ep.hdr)) {
CAF_LOG_WARNING("received invalid header:" << CAF_ARG(ep.hdr));
std::cerr << "Received invalid header!" << std::endl;
return err();
}
CAF_LOG_DEBUG(CAF_ARG(hdr));
CAF_LOG_DEBUG(CAF_ARG(ep.hdr));
std::vector<char>* payload = nullptr;
if (hdr.payload_len > 0) {
if (ep.hdr.payload_len > 0) {
payload = &pl_buf;
if (payload->size() != hdr.payload_len) {
if (payload->size() != ep.hdr.payload_len) {
CAF_LOG_WARNING("received invalid payload");
return err();
}
}
if (!handle_msg(ctx, em.handle, hdr, payload, false, em.port))
if (!handle_msg(ctx, em.handle, ep.hdr, payload, false, em.port))
return err();
return true;
}
......
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