Commit 3e470a42 authored by Dominik Charousset's avatar Dominik Charousset

Modularize BASP implementation

parent 269acb49
......@@ -103,7 +103,7 @@
/// @author Dominik Charousset <dominik.charousset (at) haw-hamburg.de>
///
/// @mainpage libcaf
/// @mainpage CAF
///
/// @section Intro Introduction
///
......
......@@ -25,9 +25,8 @@
namespace caf {
/// Implementation class for spawning composable states directly as actors.
template <class State>
class composable_behavior_based_actor : public stateful_actor
<State, typename State::actor_base> {
template <class State, class Base = typename State::actor_base>
class composable_behavior_based_actor : public stateful_actor<State, Base> {
public:
static_assert(! std::is_abstract<State>::value,
"State is abstract, please make sure to override all "
......
......@@ -7,7 +7,6 @@ file(GLOB_RECURSE LIBCAF_IO_HDRS "caf/*.hpp")
# list cpp files excluding platform-dependent files
set (LIBCAF_IO_SRCS
src/basp.cpp
src/basp_broker.cpp
src/abstract_broker.cpp
src/broker.cpp
......@@ -22,7 +21,12 @@ set (LIBCAF_IO_SRCS
src/stream_manager.cpp
src/test_multiplexer.cpp
src/acceptor_manager.cpp
src/multiplexer.cpp)
src/multiplexer.cpp
# BASP files
src/header.cpp
src/message_type.cpp
src/routing_table.cpp
src/instance.cpp)
add_custom_target(libcaf_io)
......
......@@ -20,7 +20,7 @@
#ifndef CAF_IO_ALL_HPP
#define CAF_IO_ALL_HPP
#include "caf/io/basp.hpp"
#include "caf/io/basp/all.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/basp_broker.hpp"
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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_BASP_ALL_HPP
#define CAF_IO_BASP_ALL_HPP
#include "caf/io/basp/header.hpp"
#include "caf/io/basp/version.hpp"
#include "caf/io/basp/instance.hpp"
#include "caf/io/basp/error_code.hpp"
#include "caf/io/basp/buffer_type.hpp"
#include "caf/io/basp/message_type.hpp"
#include "caf/io/basp/routing_table.hpp"
#include "caf/io/basp/connection_state.hpp"
/// @defgroup BASP Binary Actor Sytem Protocol
///
/// # Protocol Overview
///
/// The "Binary Actor Sytem Protocol" (BASP) is **not** a network protocol.
/// It is a specification for the "Remote Method Invocation" (RMI) interface
/// used by distributed instances of CAF. The purpose of BASP is unify the
/// structure of RMI calls in order to simplify processing and implementation.
/// Hence, BASP is independent of any underlying network technology,
/// and assumes a reliable communication channel.
///
///
/// The RMI interface of CAF enables network-transparent monitoring and linking
/// as well as global message dispatching to actors running on different nodes.
///
/// ![](basp_overview.png)
///
/// The figure above illustrates the phyiscal as well as the logical view
/// of a distributed CAF application. Note that the actors used for the
/// BASP communication ("BASP Brokers") are not part of the logical system
/// view and are in fact not visible to other actors. A BASP Broker creates
/// proxy actors that represent actors running on different nodes. It is worth
/// mentioning that two instances of CAF running on the same physical machine
/// are considered two different nodes in BASP.
///
/// BASP has two objectives:
///
/// - **Forward messages sent to a proxy to the actor it represents**
///
/// Whenever a proxy instance receives a message, it forwards this message to
/// its parent (a BASP Broker). This message is then serialized and forwarded
/// over the network. If no direct connection between the node sending the
/// message and the node receiving it exists, intermediate BASP Brokers will
/// forward it until it the message reaches its destination.
///
/// - **Synchronize the state of an actor with all of its proxies**.
///
/// Whenever a node learns the address of a remotely running actor, it
/// creates Ma local proxy instance representing this actor and sends an
/// `announce_proxy_instance` to the node hosting the actor. Whenever an actor
/// terminates, the hosting node sends `kill_proxy_instance` messages to all
/// nodes that have a proxy for this actor. This enables network-transparent
/// actor monitoring. There are two possible ways addresses can be learned:
///
/// + A client connects to a remotely running (published) actor via
/// `remote_actor`. In this case, the `server_handshake` will contain the
/// address of the published actor.
///
/// + Receiving `dispatch_message`. Whenever an actor message arrives,
/// it usually contains the address of the sender. Further, the message
/// itself can contain addresses to other actors that the BASP Broker
/// will get aware of while deserializing the message object
/// from the payload.
///
/// # Node IDs
///
/// The ID of a node consists of a 120 bit hash and the process ID. Note that
/// we use "node" as a synonym for "CAF instance". The hash is generated from
/// "low-level" characteristics of a machine such as the UUID of the root
/// file system and available MAC addresses. The only purpose of the node ID
/// is to generate a network-wide unique identifier. By adding the process ID,
/// CAF disambiguates multiple instances running on the same phyisical machine.
///
/// # Header Format
///
/// ![](basp_header.png)
///
/// - **Operation ID**: 4 bytes.
///
/// This field indicates what BASP function this datagram represents.
/// The value is an `uint32_t` representation of `message_type`.
///
/// - **Payload Length**: 4 bytes.
///
/// The length of the data following this header as `uint32_t`,
/// measured in bytes.
///
/// - **Operation Data**: 8 bytes.
///
/// This field contains.
///
/// - **Source Node ID**: 18 bytes.
///
/// The address of the source node. See [Node IDs](# Node IDs).
///
/// - **Destination Node ID**: 18 bytes.
///
/// The address of the destination node. See [Node IDs](# Node IDs).
/// Upon receiving this datagram, a BASP Broker compares this node ID
/// to its own ID. On a mismatch, it selects the next hop and forwards
/// this datagram unchanged.
///
/// - **Source Actor ID**: 4 bytes.
///
/// This field contains the ID of the sending actor or 0 for anonymously
/// sent messages. The *full address* of an actor is the combination of
/// the node ID and the actor ID. Thus, every actor can be unambigiously
/// identified by combining these two IDs.
///
/// - **Destination Actor ID**: 4 bytes.
///
/// This field contains the ID of the receiving actor or 0 for BASP
/// functions that do not require
///
/// # Example
///
/// The following diagram models a distributed application
/// with three nodes. The pseudo code for the application can be found
/// in the three grey boxes, while the resulting BASP messaging
/// is shown in UML sequence diagram notation. More details about
/// individual BASP message types can be found in the documentation
/// of {@link message_type} below.
///
/// ![](basp_sequence.png)
#endif // CAF_IO_BASP_ALL_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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_BASP_BUFFER_TYPE_HPP
#define CAF_IO_BASP_BUFFER_TYPE_HPP
#include <vector>
namespace caf {
namespace io {
namespace basp {
/// @addtogroup BASP
/// Storage type for raw bytes.
using buffer_type = std::vector<char>;
/// @}
} // namespace basp
} // namespace io
} // namespace caf
#endif // CAF_IO_BASP_BUFFER_TYPE_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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_BASP_CONNECTION_STATE_HPP
#define CAF_IO_BASP_CONNECTION_STATE_HPP
namespace caf {
namespace io {
namespace basp {
/// @addtogroup BASP
/// Denotes the state of a connection between to BASP nodes.
enum connection_state {
/// Indicates that a connection is established and this node is
/// waiting for the next BASP header.
await_header,
/// Indicates that this node has received a header with non-zero payload
/// and is waiting for the data.
await_payload,
/// Indicates that this connection no longer exists.
close_connection
};
/// @relates connection_state
constexpr const char* to_string(connection_state x) {
return x == await_header ? "await_header"
: (x == await_payload ? "await_payload"
: "close_connection");
}
/// @}
} // namespace basp
} // namespace io
} // namespace caf
#endif // CAF_IO_BASP_CONNECTION_STATE_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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__HPP
#define CAF_IO__HPP
#include <cstdint>
namespace caf {
namespace io {
namespace basp {
/// @addtogroup BASP
/// Describes an error during forwarding of BASP messages.
enum class error_code : uint64_t {
/// Indicates that a forwarding node had no route
/// to the destination.
no_route_to_destination = 0x01,
/// Indicates that a forwarding node detected
/// a loop in the forwarding path.
loop_detected = 0x02
};
/// @relates error_code
constexpr const char* to_string(error_code x) {
return x == error_code::no_route_to_destination ? "no_route_to_destination"
: "loop_detected";
}
/// @}
} // namespace basp
} // namespace io
} // namespace caf
#endif // CAF_IO__HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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_BASP_HEADER_HPP
#define CAF_IO_BASP_HEADER_HPP
#include <string>
#include <cstdint>
#include "caf/node_id.hpp"
#include "caf/io/basp/message_type.hpp"
namespace caf {
namespace io {
namespace basp {
/// @addtogroup BASP
/// The header of a Binary Actor System Protocol (BASP) message.
/// A BASP header consists of a routing part, i.e., source and
/// destination, as well as an operation and operation data. Several
/// message types consist of only a header.
struct header {
message_type operation;
uint32_t payload_len;
uint64_t operation_data;
node_id source_node;
node_id dest_node;
actor_id source_actor;
actor_id dest_actor;
};
/// @relates header
template <class T>
void serialize(T& in_or_out, header& hdr, const unsigned int) {
in_or_out & hdr.source_node;
in_or_out & hdr.dest_node;
in_or_out & hdr.source_actor;
in_or_out & hdr.dest_actor;
in_or_out & hdr.payload_len;
in_or_out & hdr.operation;
in_or_out & hdr.operation_data;
}
/// @relates header
std::string to_string(const header& hdr);
/// @relates header
bool operator==(const header& lhs, const header& rhs);
/// @relates header
inline bool operator!=(const header& lhs, const header& rhs) {
return !(lhs == rhs);
}
/// Checks whether given header contains a handshake.
inline bool is_handshake(const header& hdr) {
return hdr.operation == message_type::server_handshake
|| hdr.operation == message_type::client_handshake;
}
/// Checks wheter given header contains a heartbeat.
inline bool is_heartbeat(const header& hdr) {
return hdr.operation == message_type::heartbeat;
}
/// Checks whether given BASP header is valid.
/// @relates header
bool valid(const header& hdr);
/// Size of a BASP header in serialized form
constexpr size_t header_size = node_id::serialized_size * 2
+ sizeof(actor_id) * 2
+ sizeof(uint32_t) * 2
+ sizeof(uint64_t);
/// @}
} // namespace basp
} // namespace io
} // namespace caf
#endif // CAF_IO_BASP_HEADER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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_BASP_MESSAGE_TYPE_HPP
#define CAF_IO_BASP_MESSAGE_TYPE_HPP
#include <string>
#include <cstdint>
namespace caf {
namespace io {
namespace basp {
/// @addtogroup BASP
/// Describes the first header field of a BASP message and determines the
/// interpretation of the other header fields.
enum class message_type : uint32_t {
/// Send from server, i.e., the node with a published actor, to client,
/// i.e., node that initiates a new connection using remote_actor().
///
/// ![](server_handshake.png)
server_handshake = 0x00,
/// Send from client to server after it has successfully received the
/// server_handshake to establish the connection.
///
/// ![](client_handshake.png)
client_handshake = 0x01,
/// Transmits a message from `source_node:source_actor` to
/// `dest_node:dest_actor`.
///
/// ![](dispatch_message.png)
dispatch_message = 0x02,
/// Informs the receiving node that the sending node has created a proxy
/// instance for one of its actors. Causes the receiving node to attach
/// a functor to the actor that triggers a kill_proxy_instance
/// message on termination.
///
/// ![](announce_proxy_instance.png)
announce_proxy_instance = 0x03,
/// Informs the receiving node that it has a proxy for an actor
/// that has been terminated.
///
/// ![](kill_proxy_instance.png)
kill_proxy_instance = 0x04,
/// Send to remote node which has direct connection.
///
/// ![](heartbeat.png)
heartbeat = 0x05,
};
/// @relates message_type
std::string to_string(message_type);
/// @}
} // namespace basp
} // namespace io
} // namespace caf
#endif // CAF_IO_BASP_MESSAGE_TYPE_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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_BASP_ROUTING_TABLE_HPP
#define CAF_IO_BASP_ROUTING_TABLE_HPP
#include <unordered_map>
#include <unordered_set>
#include "caf/node_id.hpp"
#include "caf/callback.hpp"
#include "caf/io/abstract_broker.hpp"
#include "caf/io/basp/buffer_type.hpp"
namespace caf {
namespace io {
namespace basp {
/// @addtogroup BASP
/// Stores routing information for a single broker participating as
/// BASP peer and provides both direct and indirect paths.
class routing_table {
public:
explicit routing_table(abstract_broker* parent);
virtual ~routing_table();
/// Describes a routing path to a node.
struct route {
buffer_type& wr_buf;
const node_id& next_hop;
connection_handle hdl;
};
/// Describes a function object for erase operations that
/// is called for each indirectly lost connection.
using erase_callback = callback<const node_id&>;
/// Returns a route to `target` or `none` on error.
maybe<route> lookup(const node_id& target);
/// Returns the ID of the peer connected via `hdl` or
/// `invalid_node_id` if `hdl` is unknown.
node_id lookup_direct(const connection_handle& hdl) const;
/// Returns the handle offering a direct connection to `nid` or
/// `invalid_connection_handle` if no direct connection to `nid` exists.
connection_handle lookup_direct(const node_id& nid) const;
/// Returns the next hop that would be chosen for `nid`
/// or `invalid_node_id` if there's no indirect route to `nid`.
node_id lookup_indirect(const node_id& nid) const;
/// Flush output buffer for `r`.
void flush(const route& r);
/// Adds a new direct route to the table.
/// @pre `hdl != invalid_connection_handle && nid != invalid_node_id`
void add_direct(const connection_handle& hdl, const node_id& dest);
/// Adds a new indirect route to the table.
bool add_indirect(const node_id& hop, const node_id& dest);
/// Blacklist the route to `dest` via `hop`.
void blacklist(const node_id& hop, const node_id& dest);
/// Removes a direct connection and calls `cb` for any node
/// that became unreachable as a result of this operation,
/// including the node that is assigned as direct path for `hdl`.
void erase_direct(const connection_handle& hdl, erase_callback& cb);
/// Removes any entry for indirect connection to `dest` and returns
/// `true` if `dest` had an indirect route, otherwise `false`.
bool erase_indirect(const node_id& dest);
/// Queries whether `dest` is reachable.
bool reachable(const node_id& dest);
/// Removes all direct and indirect routes to `dest` and calls
/// `cb` for any node that became unreachable as a result of this
/// operation, including `dest`.
/// @returns the number of removed routes (direct and indirect)
size_t erase(const node_id& dest, erase_callback& cb);
public:
template <class Map, class Fallback>
typename Map::mapped_type
get_opt(const Map& m, const typename Map::key_type& k, Fallback&& x) const {
auto i = m.find(k);
if (i != m.end())
return i->second;
return std::forward<Fallback>(x);
}
using node_id_set = std::unordered_set<node_id>;
using indirect_entries = std::unordered_map<node_id, // dest
node_id_set>; // hop
abstract_broker* parent_;
std::unordered_map<connection_handle, node_id> direct_by_hdl_;
std::unordered_map<node_id, connection_handle> direct_by_nid_;
indirect_entries indirect_;
indirect_entries blacklist_;
};
/// @}
} // namespace basp
} // namespace io
} // namespace caf
#endif // CAF_IO_BASP_ROUTING_TABLE_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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_BASP_VERSION_HPP
#define CAF_IO_BASP_VERSION_HPP
namespace caf {
namespace io {
namespace basp {
/// @addtogroup BASP
/// The current BASP version. Different BASP versions will not
/// be able to exchange messages.
constexpr uint64_t version = 1;
/// @}
} // namespace basp
} // namespace io
} // namespace caf
#endif // CAF_IO_BASP_VERSION_HPP
......@@ -34,8 +34,9 @@
#include "caf/binary_deserializer.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/io/basp.hpp"
#include "caf/io/basp/all.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/typed_broker.hpp"
namespace caf {
namespace io {
......@@ -134,6 +135,8 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
}
};
/// A broker implementation for the Binary Actor System Protocol (BASP).
class basp_broker : public stateful_actor<basp_broker_state, broker> {
public:
......
......@@ -32,7 +32,7 @@
#include "caf/actor_registry.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/io/basp.hpp"
#include "caf/io/basp/all.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/network/interfaces.hpp"
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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. *
******************************************************************************/
#include "caf/io/basp/header.hpp"
#include <sstream>
namespace caf {
namespace io {
namespace basp {
std::string to_string(const header &hdr) {
std::ostringstream oss;
oss << "{"
<< to_string(hdr.operation) << ", "
<< hdr.payload_len << ", "
<< hdr.operation_data << ", "
<< to_string(hdr.source_node) << ", "
<< to_string(hdr.dest_node) << ", "
<< hdr.source_actor << ", "
<< hdr.dest_actor
<< "}";
return oss.str();
}
bool operator==(const header& lhs, const header& rhs) {
return lhs.operation == rhs.operation
&& lhs.payload_len == rhs.payload_len
&& lhs.operation_data == rhs.operation_data
&& lhs.source_node == rhs.source_node
&& lhs.dest_node == rhs.dest_node
&& lhs.source_actor == rhs.source_actor
&& lhs.dest_actor == rhs.dest_actor;
}
namespace {
bool valid(const node_id& val) {
return val != invalid_node_id;
}
template <class T>
bool zero(T val) {
return val == 0;
}
bool server_handshake_valid(const header& hdr) {
return valid(hdr.source_node)
&& ! valid(hdr.dest_node)
&& zero(hdr.dest_actor)
&& ! zero(hdr.operation_data);
}
bool client_handshake_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);
}
bool dispatch_message_valid(const header& hdr) {
return valid(hdr.dest_node)
&& ! zero(hdr.dest_actor)
&& ! zero(hdr.payload_len);
}
bool announce_proxy_instance_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);
}
bool kill_proxy_instance_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);
}
bool heartbeat_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>
bool valid(const header& hdr) {
switch (hdr.operation) {
default:
return false; // invalid operation field
case message_type::server_handshake:
return server_handshake_valid(hdr);
case message_type::client_handshake:
return client_handshake_valid(hdr);
case message_type::dispatch_message:
return dispatch_message_valid(hdr);
case message_type::announce_proxy_instance:
return announce_proxy_instance_valid(hdr);
case message_type::kill_proxy_instance:
return kill_proxy_instance_valid(hdr);
case message_type::heartbeat:
return heartbeat_valid(hdr);
}
}
} // namespace basp
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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. *
******************************************************************************/
#include "caf/io/basp/message_type.hpp"
namespace caf {
namespace io {
namespace basp {
std::string to_string(message_type x) {
switch (x) {
case message_type::server_handshake:
return "server_handshake";
case message_type::client_handshake:
return "client_handshake";
case message_type::dispatch_message:
return "dispatch_message";
case message_type::announce_proxy_instance:
return "announce_proxy_instance";
case message_type::kill_proxy_instance:
return "kill_proxy_instance";
case message_type::heartbeat:
return "heartbeat";
default:
return "???";
}
}
} // namespace basp
} // namespace io
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* 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. *
******************************************************************************/
#include "caf/io/basp/routing_table.hpp"
#include "caf/io/middleman.hpp"
namespace caf {
namespace io {
namespace basp {
routing_table::routing_table(abstract_broker* parent) : parent_(parent) {
// nop
}
routing_table::~routing_table() {
// nop
}
maybe<routing_table::route> routing_table::lookup(const node_id& target) {
auto hdl = lookup_direct(target);
if (hdl != invalid_connection_handle)
return route{parent_->wr_buf(hdl), target, hdl};
// pick first available indirect route
auto i = indirect_.find(target);
if (i != indirect_.end()) {
auto& hops = i->second;
while (! hops.empty()) {
auto& hop = *hops.begin();
hdl = lookup_direct(hop);
if (hdl != invalid_connection_handle)
return route{parent_->wr_buf(hdl), hop, hdl};
else
hops.erase(hops.begin());
}
}
return none;
}
void routing_table::flush(const route& r) {
parent_->flush(r.hdl);
}
node_id routing_table::lookup_direct(const connection_handle& hdl) const {
return get_opt(direct_by_hdl_, hdl, invalid_node_id);
}
connection_handle routing_table::lookup_direct(const node_id& nid) const {
return get_opt(direct_by_nid_, nid, invalid_connection_handle);
}
node_id routing_table::lookup_indirect(const node_id& nid) const {
auto i = indirect_.find(nid);
if (i == indirect_.end())
return invalid_node_id;
if (i->second.empty())
return invalid_node_id;
return *i->second.begin();
}
void routing_table::blacklist(const node_id& hop, const node_id& dest) {
blacklist_[dest].emplace(hop);
auto i = indirect_.find(dest);
if (i == indirect_.end())
return;
i->second.erase(hop);
if (i->second.empty())
indirect_.erase(i);
}
void routing_table::erase_direct(const connection_handle& hdl,
erase_callback& cb) {
auto i = direct_by_hdl_.find(hdl);
if (i == direct_by_hdl_.end())
return;
cb(i->second);
parent_->parent().notify<hook::connection_lost>(i->second);
direct_by_nid_.erase(i->second);
direct_by_hdl_.erase(i);
}
bool routing_table::erase_indirect(const node_id& dest) {
auto i = indirect_.find(dest);
if (i == indirect_.end())
return false;
if (parent_->parent().has_hook())
for (auto& nid : i->second)
parent_->parent().notify<hook::route_lost>(nid, dest);
indirect_.erase(i);
return true;
}
void routing_table::add_direct(const connection_handle& hdl,
const node_id& nid) {
CAF_ASSERT(direct_by_hdl_.count(hdl) == 0);
CAF_ASSERT(direct_by_nid_.count(nid) == 0);
direct_by_hdl_.emplace(hdl, nid);
direct_by_nid_.emplace(nid, hdl);
parent_->parent().notify<hook::new_connection_established>(nid);
}
bool routing_table::add_indirect(const node_id& hop, const node_id& dest) {
auto i = blacklist_.find(dest);
if (i == blacklist_.end() || i->second.count(hop) == 0) {
auto& hops = indirect_[dest];
auto added_first = hops.empty();
hops.emplace(hop);
parent_->parent().notify<hook::new_route_added>(hop, dest);
return added_first;
}
return false; // blacklisted
}
bool routing_table::reachable(const node_id& dest) {
return direct_by_nid_.count(dest) > 0 || indirect_.count(dest) > 0;
}
size_t routing_table::erase(const node_id& dest, erase_callback& cb) {
cb(dest);
size_t res = 0;
auto i = indirect_.find(dest);
if (i != indirect_.end()) {
res = i->second.size();
for (auto& nid : i->second) {
cb(nid);
parent_->parent().notify<hook::route_lost>(nid, dest);
}
indirect_.erase(i);
}
auto hdl = lookup_direct(dest);
if (hdl != invalid_connection_handle) {
direct_by_hdl_.erase(hdl);
direct_by_nid_.erase(dest);
parent_->parent().notify<hook::connection_lost>(dest);
++res;
}
return res;
}
} // namespace basp
} // namespace io
} // namespace caf
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