Commit 91946b53 authored by Dominik Charousset's avatar Dominik Charousset

Re-organize BASP header + documentation update

The BASP header has been re-organized for better readability and in advance
of future extensions. This is a breaking change!
parent a8c30ef2
......@@ -527,7 +527,7 @@ configure_file("${CMAKE_CURRENT_SOURCE_DIR}/manual/variables.tex.in"
# check for doxygen and add custom "doc" target to Makefile
find_package(Doxygen)
if(DOXYGEN_FOUND)
configure_file("${CMAKE_CURRENT_SOURCE_DIR}/Doxyfile.in"
configure_file("${CMAKE_CURRENT_SOURCE_DIR}/doc/Doxyfile.in"
"${CMAKE_CURRENT_SOURCE_DIR}/Doxyfile"
@ONLY)
add_custom_target(doc "${DOXYGEN_EXECUTABLE}"
......
......@@ -549,7 +549,7 @@ WARN_LOGFILE =
# directories like "/usr/src/myproject". Separate the files or directories
# with spaces.
INPUT = @CMAKE_HOME_DIRECTORY@/libcaf_core/caf @CMAKE_HOME_DIRECTORY@/libcaf_core/caf/mixin @CMAKE_HOME_DIRECTORY@/libcaf_core/caf/policy @CMAKE_HOME_DIRECTORY@/libcaf_io/caf/io
INPUT = @CMAKE_HOME_DIRECTORY@/libcaf_core/caf @CMAKE_HOME_DIRECTORY@/libcaf_core/caf/mixin @CMAKE_HOME_DIRECTORY@/libcaf_core/caf/policy @CMAKE_HOME_DIRECTORY@/libcaf_io/caf/io @CMAKE_HOME_DIRECTORY@/libcaf_io/caf/io/network
# This tag can be used to specify the character encoding of the source files
# that doxygen parses. Internally doxygen uses the UTF-8 encoding, which is
......@@ -626,7 +626,7 @@ EXAMPLE_RECURSIVE = YES
# directories that contain image that are included in the documentation (see
# the \image command).
IMAGE_PATH =
IMAGE_PATH = @CMAKE_HOME_DIRECTORY@/doc/
# The INPUT_FILTER tag can be used to specify a program that doxygen should
# invoke to filter for each input file. Doxygen will invoke the filter program
......
......@@ -148,8 +148,7 @@
/// The {@link math_actor.cpp Math Actor Example} shows the usage
/// of {@link receive_loop} and {@link caf::arg_match arg_match}.
/// The {@link dining_philosophers.cpp Dining Philosophers Example}
/// introduces event-based actors and includes a lot of `libcaf
/// features.
/// introduces event-based actors covers various features of CAF.
///
/// @namespace caf
/// Root namespace of libcaf.
......@@ -164,35 +163,38 @@
/// Contains policies encapsulating characteristics or algorithms.
///
/// @namespace caf::io
/// Contains all network-related classes and functions.
/// Contains all IO-related classes and functions.
///
/// @namespace caf::io::network
/// Contains classes and functions used for network abstraction.
///
/// @namespace caf::io::basp
/// Contains all classes and functions for the Binary Actor Sytem Protocol.
///
/// @defgroup MessageHandling Message handling.
/// @defgroup MessageHandling Message Handling
///
/// This is the beating heart of `libcaf`. Actor programming is
/// all about message handling.
/// This is the beating heart of CAF, since actor programming is
/// a message oriented programming paradigm.
///
/// A message in `libcaf` is a n-tuple of values (with size >= 1)
/// You can use almost every type in a messages - as long as it is announced,
/// i.e., known by the type system of `libcaf`.
/// A message in CAF is a n-tuple of values (with size >= 1).
/// You can use almost every type in a messages as long as it is announced,
/// i.e., known by the type system of CAF.
///
/// @defgroup BlockingAPI Blocking API.
/// @defgroup BlockingAPI Blocking API
///
/// Blocking functions to receive messages.
///
/// The blocking API of libcaf is intended to be used for migrating
/// previously threaded applications. When writing new code, you should use
/// ibcafs nonblocking become/unbecome API.
/// The blocking API of CAF is intended to be used for migrating
/// previously threaded applications. When writing new code, you should
/// consider the nonblocking API based on `become` and `unbecome` first.
///
/// @section Send Send messages
/// @section Send Sending Messages
///
/// The function `send` can be used to send a message to an actor.
/// The first argument is the receiver of the message followed by any number
/// of values:
///
/// @code
/// ~~
/// // spawn some actors
/// auto a1 = spawn(...);
/// auto a2 = spawn(...);
......@@ -206,14 +208,14 @@
/// send(a1, msg);
/// send(a2, msg);
/// send(a3, msg);
/// @endcode
/// ~~
///
/// @section Receive Receive messages
///
/// The function `receive` takes a `behavior` as argument. The behavior
/// is a list of { pattern >> callback } rules.
///
/// @code
/// ~~
/// receive
/// (
/// on(atom("hello"), arg_match) >> [](const std::string& msg)
......@@ -226,7 +228,7 @@
/// return make_message(atom("result"), i0 + i1 + i2);
/// }
/// );
/// @endcode
/// ~~
///
/// Please read the manual for further details about pattern matching.
///
......@@ -239,7 +241,7 @@
/// what operation the sender of a message wanted by receiving just two integers.
///
/// Example actor:
/// @code
/// ~~
/// void math_actor() {
/// receive_loop (
/// on(atom("plus"), arg_match) >> [](int a, int b) {
......@@ -250,9 +252,9 @@
/// }
/// );
/// }
/// @endcode
/// ~~
///
/// @section ReceiveLoops Receive loops
/// @section ReceiveLoops Receive Loops
///
/// Previous examples using `receive` create behaviors on-the-fly.
/// This is inefficient in a loop since the argument passed to receive
......@@ -270,7 +272,7 @@
/// `receive_while` creates a functor evaluating a lambda expression.
/// The loop continues until the given lambda returns `false`. A simple example:
///
/// @code
/// ~~
/// // receive two integers
/// vector<int> received_values;
/// receive_while([&]() { return received_values.size() < 2; }) (
......@@ -279,23 +281,23 @@
/// }
/// );
/// // ...
/// @endcode
/// ~~
///
/// `receive_for` is a simple ranged-based loop:
///
/// @code
/// ~~
/// std::vector<int> vec {1, 2, 3, 4};
/// auto i = vec.begin();
/// receive_for(i, vec.end()) (
/// on(atom("get")) >> [&]() -> message { return {atom("result"), *i}; }
/// );
/// @endcode
/// ~~
///
/// `do_receive` returns a functor providing the function `until` that
/// takes a lambda expression. The loop continues until the given lambda
/// returns true. Example:
///
/// @code
/// ~~
/// // receive ints until zero was received
/// vector<int> received_values;
/// do_receive (
......@@ -305,29 +307,29 @@
/// )
/// .until([&]() { return received_values.back() == 0 });
/// // ...
/// @endcode
/// ~~
///
/// @section FutureSend Send delayed messages
/// @section FutureSend Sending Delayed Messages
///
/// The function `delayed_send` provides a simple way to delay a message.
/// This is particularly useful for recurring events, e.g., periodical polling.
/// Usage example:
///
/// @code
/// delayed_send(self, std::chrono::seconds(1), atom("poll"));
/// ~~
/// delayed_send(self, std::chrono::seconds(1), poll_atom::value);
/// receive_loop (
/// // ...
/// on(atom("poll")) >> [] {
/// [](poll_atom) {
/// // ... poll something ...
/// // and do it again after 1sec
/// delayed_send(self, std::chrono::seconds(1), atom("poll"));
/// delayed_send(self, std::chrono::seconds(1), poll_atom::value);
/// }
/// );
/// @endcode
/// ~~
///
/// See also the {@link dancing_kirby.cpp dancing kirby example}.
///
/// @defgroup ImplicitConversion Implicit type conversions.
/// @defgroup ImplicitConversion Implicit Type Conversions
///
/// The message passing of `libcaf` prohibits pointers in messages because
/// it enforces network transparent messaging.
......@@ -337,7 +339,7 @@
/// It also converts unicode literals to the corresponding STL container.
///
/// A few examples:
/// @code
/// ~~
/// // sends an std::string containing "hello actor!" to itself
/// send(self, "hello actor!");
///
......@@ -355,9 +357,9 @@
/// // equal to: on(std::string("hello actor!"))
/// on("hello actor!") >> [] { }
/// );
/// @endcode
/// ~~
///
/// @defgroup ActorCreation Actor creation.
/// @defgroup ActorCreation Creating Actors
// examples
......
......@@ -22,7 +22,7 @@
namespace caf {
/// @ingroup ActorCreation
/// @addtogroup ActorCreation
/// @{
/// Stores options passed to the `spawn` function family.
......
......@@ -12,6 +12,7 @@ set (LIBCAF_IO_SRCS
src/abstract_broker.cpp
src/broker.cpp
src/default_multiplexer.cpp
src/doorman.cpp
src/max_msg_size.cpp
src/middleman.cpp
src/hook.cpp
......@@ -22,6 +23,7 @@ set (LIBCAF_IO_SRCS
src/remote_group.cpp
src/manager.cpp
src/set_middleman.cpp
src/scribe.cpp
src/stream_manager.cpp
src/test_multiplexer.cpp
src/unpublish.cpp
......
......@@ -23,10 +23,12 @@
#include <vector>
#include <unordered_map>
#include "caf/local_actor.hpp"
#include "caf/detail/intrusive_partitioned_list.hpp"
#include "caf/io/fwd.hpp"
#include "caf/local_actor.hpp"
#include "caf/io/scribe.hpp"
#include "caf/io/doorman.hpp"
#include "caf/io/accept_handle.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/system_messages.hpp"
......@@ -40,141 +42,60 @@ namespace io {
class middleman;
/// @defgroup Broker Actor-based Network Abstraction
///
/// Brokers provide an actor-based abstraction for low-level network IO.
/// The central component in the network abstraction of CAF is the
/// `middleman`. It connects any number of brokers to a `multiplexer`,
/// which implements a low-level IO event loop.
///
/// ![Relation between middleman, multiplexer, and broker](broker.png)
///
/// Brokers do *not* operate on sockets or other platform-dependent
/// communication primitives. Instead, brokers use a `connection_handle`
/// to identify a reliable, end-to-end byte stream (e.g. a TCP connection)
/// and `accept_handle` to identify a communication endpoint others can
/// connect to via its port.
///
/// Each `connection_handle` is associated with a `scribe` that provides
/// access to an output buffer as well as a `flush` operation to request
/// sending its content via the network. Instead of actively receiving data,
/// brokers configure a scribe to asynchronously receive data, e.g.,
/// `self->configure_read(hdl, receive_policy::exactly(1024))` would
/// configure the scribe associated to `hdl` to receive *exactly* 1024 bytes
/// and generate a `new_data_msg` message for the broker once the
/// data is available. The buffer in this message will be re-used by the
/// scribe to minimize memory usage and heap allocations.
///
/// Each `accept_handle` is associated with a `doorman` that will create
/// a `new_connection_msg` whenever a new connection was established.
///
/// All `scribe` and `doorman` instances are managed by the `multiplexer`
///
/// A broker mediates between actor systems and other components in the network.
/// @ingroup Broker
class abstract_broker : public local_actor {
public:
using buffer_type = std::vector<char>;
class continuation;
virtual ~abstract_broker();
// even brokers need friends
friend class scribe;
friend class doorman;
friend class continuation;
void enqueue(const actor_addr&, message_id,
message, execution_unit*) override;
void enqueue(mailbox_element_ptr, execution_unit*) override;
void launch(execution_unit* eu, bool lazy, bool hide);
/// Called after this broker has finished execution.
void cleanup(uint32_t reason);
/// Manages a low-level IO device for the `broker`.
class servant {
public:
friend class abstract_broker;
void set_broker(abstract_broker* ptr);
virtual ~servant();
protected:
virtual void remove_from_broker() = 0;
virtual message disconnect_message() = 0;
inline abstract_broker* parent() {
return broker_;
}
servant(abstract_broker* ptr);
void disconnect(bool invoke_disconnect_message);
bool disconnected_;
abstract_broker* broker_;
};
/// Manages a stream.
class scribe : public network::stream_manager, public servant {
public:
scribe(abstract_broker* parent, connection_handle hdl);
~scribe();
/// Implicitly starts the read loop on first call.
virtual void configure_read(receive_policy::config config) = 0;
/// Grants access to the output buffer.
virtual buffer_type& wr_buf() = 0;
/// Flushes the output buffer, i.e., sends the content of
/// the buffer via the network.
virtual void flush() = 0;
inline connection_handle hdl() const {
return hdl_;
}
void io_failure(network::operation op) override;
void consume(const void* data, size_t num_bytes) override;
protected:
virtual buffer_type& rd_buf() = 0;
inline new_data_msg& read_msg() {
return read_msg_.get_as_mutable<new_data_msg>(0);
}
inline const new_data_msg& read_msg() const {
return read_msg_.get_as<new_data_msg>(0);
}
void remove_from_broker() override;
message disconnect_message() override;
connection_handle hdl_;
message read_msg_;
};
using scribe_ptr = intrusive_ptr<scribe>;
/// Manages incoming connections.
class doorman : public network::acceptor_manager, public servant {
public:
doorman(abstract_broker* parent, accept_handle hdl, uint16_t local_port);
~doorman();
inline accept_handle hdl() const {
return hdl_;
}
void io_failure(network::operation op) override;
// needs to be launched explicitly
virtual void launch() = 0;
uint16_t port() const {
return port_;
}
protected:
void remove_from_broker() override;
message disconnect_message() override;
inline new_connection_msg& accept_msg() {
return accept_msg_.get_as_mutable<new_connection_msg>(0);
}
inline const new_connection_msg& accept_msg() const {
return accept_msg_.get_as<new_connection_msg>(0);
}
accept_handle hdl_;
message accept_msg_;
uint16_t port_;
};
using doorman_ptr = intrusive_ptr<doorman>;
// a broker needs friends
friend class scribe;
friend class doorman;
friend class continuation;
virtual ~abstract_broker();
/// Starts running this broker in the `middleman`.
void launch(execution_unit* eu, bool lazy, bool hide);
/// Modifies the receive policy for given connection.
/// @param hdl Identifies the affected connection.
......@@ -182,7 +103,7 @@ public:
void configure_read(connection_handle hdl, receive_policy::config config);
/// Returns the write buffer for given connection.
buffer_type& wr_buf(connection_handle hdl);
std::vector<char>& wr_buf(connection_handle hdl);
/// Writes `data` into the buffer for given connection.
void write(connection_handle hdl, size_t data_size, const void* data);
......@@ -195,37 +116,47 @@ public:
return scribes_.size();
}
/// Returns all handles of all `scribe` instances attached to this broker.
std::vector<connection_handle> connections() const;
/// @cond PRIVATE
/// Returns the middleman instance this broker belongs to.
inline middleman& parent() {
return mm_;
}
inline void add_scribe(const scribe_ptr& ptr) {
scribes_.emplace(ptr->hdl(), ptr);
}
/// Adds a `scribe` instance to this broker.
void add_scribe(const intrusive_ptr<scribe>& ptr);
/// Tries to connect to `host` on given `port` and creates
/// a new scribe describing the connection afterwards.
/// @returns The handle of the new `scribe` on success.
/// @throws network_error Thrown if given `host` was unreachable or invalid.
connection_handle add_tcp_scribe(const std::string& host, uint16_t port);
/// Assigns a detached `scribe` instance identified by `hdl`
/// from the `multiplexer` to this broker.
void assign_tcp_scribe(connection_handle hdl);
/// Creates and assigns a new `scribe` from given native socked `fd`.
connection_handle add_tcp_scribe(network::native_socket fd);
inline void add_doorman(const doorman_ptr& ptr) {
doormen_.emplace(ptr->hdl(), ptr);
if (is_initialized()) {
ptr->launch();
}
}
/// Adds a `doorman` instance to this broker.
void add_doorman(const intrusive_ptr<doorman>& ptr);
/// Tries to open a local port and creates a `doorman` managing
/// it on success. If `port == 0`, then the broker will ask
/// the operating system to pick a random port.
/// @returns The handle of the new `doorman` and the assigned port.
/// @throws network_error Thrown if `port` was unavailable.
std::pair<accept_handle, uint16_t> add_tcp_doorman(uint16_t port = 0,
const char* in = nullptr,
bool reuse_addr = false);
/// Assigns a detached `doorman` instance identified by `hdl`
/// from the `multiplexer` to this broker.
void assign_tcp_doorman(accept_handle hdl);
/// Creates and assigns a new `doorman` from given native socked `fd`.
accept_handle add_tcp_doorman(network::native_socket fd);
/// Returns the local port associated to `hdl` or `0` if `hdl` is invalid.
......@@ -234,10 +165,11 @@ public:
/// Returns the handle associated to given local `port` or `none`.
optional<accept_handle> hdl_by_port(uint16_t port);
/// Invokes `msg` on this broker.
void invoke_message(mailbox_element_ptr& msg);
void invoke_message(const actor_addr& sender,
message_id mid, message& msg);
/// Creates a mailbox element from given data and invoke it.
void invoke_message(const actor_addr& sender, message_id mid, message& msg);
/// Closes all connections and acceptors.
void close_all();
......@@ -260,9 +192,12 @@ protected:
abstract_broker(middleman& parent_ref);
using doorman_map = std::unordered_map<accept_handle, doorman_ptr>;
using doorman_map = std::unordered_map<accept_handle, intrusive_ptr<doorman>>;
using scribe_map = std::unordered_map<connection_handle,
intrusive_ptr<scribe>>;
using scribe_map = std::unordered_map<connection_handle, scribe_ptr>;
/// @cond PRIVATE
// meta programming utility
inline doorman_map& get_map(accept_handle) {
......@@ -275,13 +210,14 @@ protected:
}
// meta programming utility (not implemented)
static doorman_ptr ptr_of(accept_handle);
static intrusive_ptr<doorman> ptr_of(accept_handle);
// meta programming utility (not implemented)
static scribe_ptr ptr_of(connection_handle);
static intrusive_ptr<scribe> ptr_of(connection_handle);
/// @endcond
/// Returns the `multiplexer` running this broker.
network::multiplexer& backend();
/// Returns a `scribe` or `doorman` identified by `hdl`.
......@@ -309,6 +245,7 @@ protected:
return result;
}
/// Tries to invoke a message from the cache.
bool invoke_message_from_cache();
private:
......
......@@ -44,6 +44,45 @@ namespace caf {
namespace io {
namespace basp {
/// @defgroup BASP Binary Actor Sytem Protocol
///
/// This protocol is used to orchestrate CAF peers in the network.
/// BASP is independent of any underlying network technology, since it
/// operates only on the granularity of `node_id` addresses. Its
/// implementation only assumes end-to-end byte stream connections.
/// BASP forms an overlay consisting of all participating nodes
/// (or peers) and enables global communiation between all actors
/// on all nodes.
///
/// ![](basp_header.png)
///
/// BASP has two primary objectives:
///
/// - Establish routing paths for all participating nodes, i.e.,
/// allow BASP nodes to exchange messages even if no direct
/// connection exists.
///
/// - Propagate actor terminations. Whenever a node learns the
/// address of a remotely running actor, it creates a 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.
///
/// 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)
/// @addtogroup BASP
/// @{
/// The current BASP version. Different BASP versions will not
/// be able to exchange messages.
constexpr uint64_t version = 2;
......@@ -51,52 +90,110 @@ constexpr uint64_t version = 2;
/// Storage type for raw bytes.
using buffer_type = std::vector<char>;
/// 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,
/// Informs the receiving node that a message it sent
/// did not reach its destination.
///
/// ![](dispatch_error.png)
dispatch_error = 0x05
};
/// @relates message_type
std::string to_string(message_type);
/// 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;
uint32_t payload_len;
uint32_t operation;
uint64_t operation_data;
};
/// @relates header
std::string to_string(const header& hdr);
/// @relates header
inline bool operator==(const header& lhs, const header& rhs) {
return 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.payload_len == rhs.payload_len
&& lhs.operation == rhs.operation
&& lhs.operation_data == rhs.operation_data;
}
void read_hdr(deserializer& source, header& hdr);
/// @relates header
void write_hdr(serializer& sink, 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 BASP header is valid.
/// @relates header
void read_hdr(deserializer&, header& hdr);
bool valid(const header& hdr);
/// @relates header
void write_hdr(serializer&, const header& hdr);
/// Deserialize a BASP message header from `source`.
void read_hdr(serializer& sink, header& hdr);
/// Serialize a BASP message header to `sink`.
void write_hdr(deserializer& source, const header& hdr);
/// Size of a BASP header in serialized form
constexpr size_t header_size =
node_id::host_id_size * 2 + sizeof(uint32_t) * 2 +
sizeof(actor_id) * 2 + sizeof(uint32_t) * 2 + sizeof(uint64_t);
/// Describes an error during forwarding of BASP messages.
enum class error : 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
};
/// Denotes the state of a connection between to BASP nodes.
enum connection_state {
/// Indicates that this node just started and is waiting for server handshake.
await_server_handshake,
/// Indicates that this node just accepted a connection, sent its
/// handshake to the client, and is now waiting for the client's response.
await_client_handshake,
/// Indicates that a connection is established and this node is
/// waiting for the next BASP header.
await_header,
......@@ -301,20 +398,18 @@ public:
/// Writes a header (build from the arguments)
/// followed by its payload to `storage`.
void write(buffer_type& storage,
message_type operation,
uint32_t* payload_len,
uint64_t operation_data,
const node_id& source_node,
const node_id& dest_node,
actor_id source_actor,
actor_id dest_actor,
uint32_t* payload_len,
uint32_t operation,
uint64_t operation_data,
payload_writer* writer = nullptr,
bool suppress_auto_size_prefixing = false);
payload_writer* writer = nullptr);
/// Writes a header followed by its payload to `storage`.
void write(buffer_type& storage, header& hdr,
payload_writer* writer = nullptr,
bool suppress_auto_size_prefixing = false);
payload_writer* writer = nullptr);
/// Writes the server handshake containing the information of the
/// actor published at `port` to `buf`. If `port == none` or
......@@ -326,7 +421,7 @@ public:
void write_dispatch_error(buffer_type& buf,
const node_id& source_node,
const node_id& dest_node,
uint64_t error_code,
error error_code,
const header& original_hdr,
const buffer_type& payload);
......@@ -347,212 +442,13 @@ private:
callee& callee_;
};
/// Deserialize a BASP message header from `source`.
void read_hdr(serializer& sink, header& hdr);
/// Serialize a BASP message header to `sink`.
void write_hdr(deserializer& source, const header& hdr);
/// Size of a BASP header in serialized form
constexpr size_t header_size =
node_id::host_id_size * 2 + sizeof(uint32_t) * 2 +
sizeof(actor_id) * 2 + sizeof(uint32_t) * 2 + sizeof(uint64_t);
inline bool valid(const node_id& val) {
return val != invalid_node_id;
}
inline bool invalid(const node_id& val) {
return ! valid(val);
}
template <class T>
inline bool zero(T val) {
return val == 0;
}
template <class T>
inline bool nonzero(T aid) {
return ! zero(aid);
}
/// Send from server, i.e., the node with a published actor, to client,
/// i.e., node that initiates a new connection using remote_actor().
///
/// Field | Assignment
/// ---------------|----------------------------------------------------------
/// source_node | ID of server
/// dest_node | invalid
/// source_actor | 0
/// dest_actor | 0
/// payload_len | size of actor id + interface definition
/// operation_data | BASP version of the server
constexpr uint32_t server_handshake = 0x00;
inline bool server_handshake_valid(const header& hdr) {
return valid(hdr.source_node)
&& invalid(hdr.dest_node)
&& zero(hdr.source_actor)
&& zero(hdr.dest_actor)
&& nonzero(hdr.payload_len)
&& nonzero(hdr.operation_data);
}
/// Send from client to server after it has successfully received the
/// server_handshake to establish the connection.
///
/// Field | Assignment
/// ---------------|----------------------------------------------------------
/// source_node | ID of client
/// dest_node | ID of server
/// source_actor | 0
/// dest_actor | 0
/// payload_len | 0
/// operation_data | 0
constexpr uint32_t client_handshake = 0x01;
inline 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);
}
/// Transmits a message from source_node:source_actor to
/// dest_node:dest_actor.
///
/// Field | Assignment
/// ---------------|----------------------------------------------------------
/// source_node | ID of sending node (invalid in case of anon_send)
/// dest_node | ID of receiving node
/// source_actor | ID of sending actor (invalid in case of anon_send)
/// dest_actor | ID of receiving actor, must not be invalid
/// payload_len | > 0
/// operation_data | message ID (0 for asynchronous messages)
///
/// Payload:
/// - the first 4 bytes are the size of the serialized message
/// - serialized message
/// - any number of node IDs: each forwarding node adds its
/// node ID to the payload, allowing the receiving node to backtrace
/// the full path of this message (also enables loop detection)
constexpr uint32_t dispatch_message = 0x02;
inline bool dispatch_message_valid(const header& hdr) {
return valid(hdr.dest_node)
&& nonzero(hdr.dest_actor)
&& nonzero(hdr.payload_len);
}
/// 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.
///
/// Field | Assignment
/// ---------------|----------------------------------------------------------
/// source_node | ID of sending node
/// dest_node | ID of receiving node
/// source_actor | 0
/// dest_actor | ID of monitored actor
/// payload_len | 0
/// operation_data | 0
constexpr uint32_t announce_proxy_instance = 0x03;
inline 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)
&& nonzero(hdr.dest_actor)
&& zero(hdr.payload_len)
&& zero(hdr.operation_data);
}
/// Informs the receiving node that it has a proxy for an actor
/// that has been terminated.
///
/// Field | Assignment
/// ---------------|----------------------------------------------------------
/// source_node | ID of sending node
/// dest_node | ID of receiving node
/// source_actor | ID of monitored actor
/// dest_actor | 0
/// payload_len | 0
/// operation_data | exit reason (uint32)
constexpr uint32_t kill_proxy_instance = 0x04;
inline bool kill_proxy_instance_valid(const header& hdr) {
return valid(hdr.source_node)
&& valid(hdr.dest_node)
&& hdr.source_node != hdr.dest_node
&& nonzero(hdr.source_actor)
&& zero(hdr.dest_actor)
&& zero(hdr.payload_len)
&& nonzero(hdr.operation_data);
}
namespace error {
/// Indicates that a forwarding node had no route
/// to the destination.
static constexpr uint64_t no_route_to_destination = 0x01;
/// Indicates that a forwarding node detected
/// a loop in the forwarding path.
static constexpr uint64_t loop_detected = 0x02;
} // namespace error
/// Informs the receiving node that a message it sent
/// did not reach its destination.
///
/// Field | Assignment
/// ---------------|----------------------------------------------------------
/// source_node | ID of sending node
/// dest_node | ID of receiving node
/// source_actor | 0
/// dest_actor | 0
/// payload_len | > 0 (size of message object and forwarding nodes)
/// operation_data | error code (see `basp::error`)
constexpr uint32_t dispatch_error = 0x05;
inline 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)
&& nonzero(hdr.payload_len)
&& nonzero(hdr.operation_data);
}
/// Checks whether given header contains a handshake.
inline bool is_handshake(const header& hdr) {
return hdr.operation == server_handshake || hdr.operation == client_handshake;
return hdr.operation == message_type::server_handshake
|| hdr.operation == message_type::client_handshake;
}
/// Checks whether given header is valid.
inline bool valid(const header& hdr) {
switch (hdr.operation) {
default:
return false; // invalid operation field
case server_handshake:
return server_handshake_valid(hdr);
case client_handshake:
return client_handshake_valid(hdr);
case dispatch_message:
return dispatch_message_valid(hdr);
case announce_proxy_instance:
return announce_proxy_instance_valid(hdr);
case kill_proxy_instance:
return kill_proxy_instance_valid(hdr);
}
}
/// @}
} // namespace basp
} // namespace io
......
......@@ -32,8 +32,9 @@
namespace caf {
namespace io {
/// A broker mediates between actor systems and other components in the network.
/// @extends local_actor
/// Describes a dynamically typed broker.
/// @extends abstract_broker
/// @ingroup Broker
class broker : public abstract_event_based_actor<behavior, false,
abstract_broker> {
public:
......@@ -50,7 +51,7 @@ public:
CAF_ASSERT(sptr->hdl() == hdl);
return spawn_functor(nullptr,
[sptr](broker* forked) {
sptr->set_broker(forked);
sptr->set_parent(forked);
forked->scribes_.emplace(sptr->hdl(), sptr);
},
fun, hdl, std::forward<Ts>(xs)...);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_DOORMAN_HPP
#define CAF_IO_DOORMAN_HPP
#include <cstddef>
#include "caf/message.hpp"
#include "caf/io/accept_handle.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/network/acceptor_manager.hpp"
namespace caf {
namespace io {
/// Manages incoming connections.
/// @ingroup Broker
class doorman : public network::acceptor_manager {
public:
doorman(abstract_broker* parent, accept_handle hdl, uint16_t local_port);
~doorman();
inline accept_handle hdl() const {
return hdl_;
}
void io_failure(network::operation op) override;
// needs to be launched explicitly
virtual void launch() = 0;
uint16_t port() const {
return port_;
}
protected:
void detach_from_parent() override;
message detach_message() override;
inline new_connection_msg& accept_msg() {
return accept_msg_.get_as_mutable<new_connection_msg>(0);
}
inline const new_connection_msg& accept_msg() const {
return accept_msg_.get_as<new_connection_msg>(0);
}
accept_handle hdl_;
message accept_msg_;
uint16_t port_;
};
} // namespace io
} // namespace caf
#endif // CAF_IO_DOORMAN_HPP
......@@ -187,7 +187,7 @@ public:
"Cannot fork: new broker misses required handlers");
return spawn_functor_impl<no_spawn_options, impl>(
nullptr, [&sptr](abstract_broker* forked) {
sptr->set_broker(forked);
sptr->set_parent(forked);
forked->add_scribe(sptr);
},
std::move(fun), hdl, std::forward<Ts>(xs)...);
......
......@@ -23,11 +23,11 @@
namespace caf {
namespace io {
class basp_broker;
class abstract_broker;
class broker;
class middleman;
class basp_broker;
class receive_policy;
class abstract_broker;
namespace network {
......
......@@ -30,6 +30,8 @@ namespace network {
/// callbacks for incoming connections as well as for error handling.
class acceptor_manager : public manager {
public:
acceptor_manager(abstract_broker* ptr);
~acceptor_manager();
/// Called by the underlying IO device to indicate that
......
......@@ -115,7 +115,7 @@ template <class Socket>
connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
Socket&& sock) {
CAF_LOG_TRACE("");
class impl : public abstract_broker::scribe {
class impl : public scribe {
public:
impl(abstract_broker* ptr, Socket&& s)
: scribe(ptr, network::conn_hdl_from_socket(s)),
......@@ -130,16 +130,16 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
launch();
}
}
abstract_broker::buffer_type& wr_buf() override {
std::vector<char>& wr_buf() override {
return stream_.wr_buf();
}
abstract_broker::buffer_type& rd_buf() override {
std::vector<char>& rd_buf() override {
return stream_.rd_buf();
}
void stop_reading() override {
CAF_LOG_TRACE("");
stream_.stop_reading();
disconnect(false);
detach(false);
}
void flush() override {
CAF_LOG_TRACE("");
......@@ -156,7 +156,7 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
bool launched_;
stream<Socket> stream_;
};
abstract_broker::scribe_ptr ptr = make_counted<impl>(self, std::move(sock));
auto ptr = make_counted<impl>(self, std::move(sock));
self->add_scribe(ptr);
return ptr->hdl();
}
......@@ -210,7 +210,7 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self,
default_socket_acceptor&& sock) {
CAF_LOG_TRACE("sock.fd = " << sock.native_handle());
CAF_ASSERT(sock.native_handle() != network::invalid_native_socket);
class impl : public abstract_broker::doorman {
class impl : public doorman {
public:
impl(abstract_broker* ptr, default_socket_acceptor&& s,
network::asio_multiplexer& am)
......@@ -230,7 +230,7 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self,
void stop_reading() override {
CAF_LOG_TRACE("");
acceptor_.stop();
disconnect(false);
detach(false);
}
void launch() override {
CAF_LOG_TRACE("");
......@@ -240,8 +240,7 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self,
private:
network::acceptor<default_socket_acceptor> acceptor_;
};
abstract_broker::doorman_ptr ptr
= make_counted<impl>(self, std::move(sock), *this);
auto ptr = make_counted<impl>(self, std::move(sock), *this);
self->add_doorman(ptr);
return ptr->hdl();
}
......
......@@ -20,9 +20,11 @@
#ifndef CAF_IO_NETWORK_MANAGER_HPP
#define CAF_IO_NETWORK_MANAGER_HPP
#include "caf/message.hpp"
#include "caf/ref_counted.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/io/fwd.hpp"
#include "caf/io/network/operation.hpp"
namespace caf {
......@@ -33,14 +35,44 @@ namespace network {
/// for various IO operations.
class manager : public ref_counted {
public:
manager(abstract_broker* parent_ptr);
~manager();
/// Sets the parent for this manager.
/// @pre `parent() == nullptr`
void set_parent(abstract_broker* ptr);
/// Returns the parent broker of this manager.
inline abstract_broker* parent() {
return parent_;
}
/// Returns `true` if this manager has a parent, `false` otherwise.
inline bool detached() const {
return parent_ == nullptr;
}
/// Detach this manager from its parent and invoke `detach_message()``
/// if `invoke_detach_message == true`.
void detach(bool invoke_detach_message);
/// Causes the manager to stop read operations on its IO device.
/// Unwritten bytes are still send before the socket will be closed.
virtual void stop_reading() = 0;
/// Called by the underlying IO device to report failures.
virtual void io_failure(operation op) = 0;
protected:
/// Creates a message signalizing a disconnect to the parent.
virtual message detach_message() = 0;
/// Detaches this manager from its parent.
virtual void detach_from_parent() = 0;
private:
abstract_broker* parent_;
};
} // namespace network
......
......@@ -32,6 +32,8 @@ namespace network {
/// for incoming data as well as for error handling.
class stream_manager : public manager {
public:
stream_manager(abstract_broker* ptr);
~stream_manager();
/// Called by the underlying IO device whenever it received data.
......
......@@ -77,7 +77,7 @@ public:
bool& stopped_reading(connection_handle hdl);
abstract_broker::scribe_ptr& impl_ptr(connection_handle hdl);
intrusive_ptr<scribe>& impl_ptr(connection_handle hdl);
uint16_t& port(accept_handle hdl);
......@@ -85,7 +85,7 @@ public:
/// `false` otherwise.
bool& stopped_reading(accept_handle hdl);
abstract_broker::doorman_ptr& impl_ptr(accept_handle hdl);
intrusive_ptr<doorman>& impl_ptr(accept_handle hdl);
void add_pending_connect(accept_handle src, connection_handle hdl);
......@@ -128,13 +128,13 @@ private:
buffer_type wr_buf;
receive_policy::config recv_conf;
bool stopped_reading = false;
abstract_broker::scribe_ptr ptr;
intrusive_ptr<scribe> ptr;
};
struct doorman_data {
uint16_t port;
bool stopped_reading = false;
abstract_broker::doorman_ptr ptr;
intrusive_ptr<doorman> ptr;
};
std::mutex mx_;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_SCRIBE_HPP
#define CAF_IO_SCRIBE_HPP
#include <vector>
#include "caf/message.hpp"
#include "caf/io/receive_policy.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/network/stream_manager.hpp"
namespace caf {
namespace io {
/// Manages a stream.
/// @ingroup Broker
class scribe : public network::stream_manager {
public:
scribe(abstract_broker* parent, connection_handle hdl);
~scribe();
/// Implicitly starts the read loop on first call.
virtual void configure_read(receive_policy::config config) = 0;
/// Grants access to the output buffer.
virtual std::vector<char>& wr_buf() = 0;
/// Flushes the output buffer, i.e., sends the content of
/// the buffer via the network.
virtual void flush() = 0;
inline connection_handle hdl() const {
return hdl_;
}
void io_failure(network::operation op) override;
void consume(const void* data, size_t num_bytes) override;
protected:
virtual std::vector<char>& rd_buf() = 0;
inline new_data_msg& read_msg() {
return read_msg_.get_as_mutable<new_data_msg>(0);
}
inline const new_data_msg& read_msg() const {
return read_msg_.get_as<new_data_msg>(0);
}
void detach_from_parent() override;
message detach_message() override;
private:
connection_handle hdl_;
message read_msg_;
};
} // namespace io
} // namespace caf
#endif // CAF_IO_SCRIBE_HPP
......@@ -97,114 +97,6 @@ void abstract_broker::cleanup(uint32_t reason) {
deref(); // release implicit reference count from middleman
}
void abstract_broker::servant::set_broker(abstract_broker* new_broker) {
if (! disconnected_) {
broker_ = new_broker;
}
}
abstract_broker::servant::~servant() {
CAF_LOG_TRACE("");
}
abstract_broker::servant::servant(abstract_broker* ptr) : disconnected_(false), broker_(ptr) {
// nop
}
void abstract_broker::servant::disconnect(bool invoke_disconnect_message) {
CAF_LOG_TRACE("");
if (! disconnected_) {
CAF_LOG_DEBUG("disconnect servant from broker");
disconnected_ = true;
remove_from_broker();
if (invoke_disconnect_message) {
auto msg = disconnect_message();
broker_->invoke_message(broker_->address(),invalid_message_id, msg);
}
}
}
abstract_broker::scribe::scribe(abstract_broker* ptr, connection_handle conn_hdl)
: servant(ptr),
hdl_(conn_hdl) {
std::vector<char> tmp;
read_msg_ = make_message(new_data_msg{hdl_, std::move(tmp)});
}
void abstract_broker::scribe::remove_from_broker() {
CAF_LOG_TRACE("hdl = " << hdl().id());
broker_->scribes_.erase(hdl());
}
abstract_broker::scribe::~scribe() {
CAF_LOG_TRACE("");
}
message abstract_broker::scribe::disconnect_message() {
return make_message(connection_closed_msg{hdl()});
}
void abstract_broker::scribe::consume(const void*, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(num_bytes));
if (disconnected_) {
// we are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an IO event causes
// the broker to call close_all() while the pollset contained
// further activities for the broker
return;
}
auto& buf = rd_buf();
CAF_ASSERT(buf.size() >= num_bytes);
// make sure size is correct, swap into message, and then call client
buf.resize(num_bytes);
read_msg().buf.swap(buf);
broker_->invoke_message(invalid_actor_addr, invalid_message_id, read_msg_);
// swap buffer back to stream and implicitly flush wr_buf()
if (broker_->exit_reason() == exit_reason::not_exited) {
read_msg().buf.swap(buf);
flush();
}
}
void abstract_broker::scribe::io_failure(network::operation op) {
CAF_LOG_TRACE("id = " << hdl().id()
<< ", " << CAF_TARG(op, static_cast<int>));
// keep compiler happy when compiling w/o logging
static_cast<void>(op);
disconnect(true);
}
abstract_broker::doorman::doorman(abstract_broker* ptr,
accept_handle acc_hdl,
uint16_t p)
: servant(ptr),
hdl_(acc_hdl),
port_(p) {
auto hdl2 = connection_handle::from_int(-1);
accept_msg_ = make_message(new_connection_msg{hdl_, hdl2});
}
abstract_broker::doorman::~doorman() {
CAF_LOG_TRACE("");
}
void abstract_broker::doorman::remove_from_broker() {
CAF_LOG_TRACE("hdl = " << hdl().id());
broker_->doormen_.erase(hdl());
}
message abstract_broker::doorman::disconnect_message() {
return make_message(acceptor_closed_msg{hdl()});
}
void abstract_broker::doorman::io_failure(network::operation op) {
CAF_LOG_TRACE("id = " << hdl().id() << ", "
<< CAF_TARG(op, static_cast<int>));
// keep compiler happy when compiling w/o logging
static_cast<void>(op);
disconnect(true);
}
abstract_broker::~abstract_broker() {
CAF_LOG_TRACE("");
}
......@@ -216,7 +108,7 @@ void abstract_broker::configure_read(connection_handle hdl,
by_id(hdl).configure_read(cfg);
}
abstract_broker::buffer_type& abstract_broker::wr_buf(connection_handle hdl) {
std::vector<char>& abstract_broker::wr_buf(connection_handle hdl) {
return by_id(hdl).wr_buf();
}
......@@ -240,6 +132,9 @@ std::vector<connection_handle> abstract_broker::connections() const {
return result;
}
void abstract_broker::add_scribe(const intrusive_ptr<scribe>& ptr) {
scribes_.emplace(ptr->hdl(), ptr);
}
connection_handle abstract_broker::add_tcp_scribe(const std::string& hostname,
uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hostname) << ", " << CAF_ARG(port));
......@@ -257,6 +152,12 @@ abstract_broker::add_tcp_scribe(network::native_socket fd) {
return backend().add_tcp_scribe(this, fd);
}
void abstract_broker::add_doorman(const intrusive_ptr<doorman>& ptr) {
doormen_.emplace(ptr->hdl(), ptr);
if (is_initialized())
ptr->launch();
}
std::pair<accept_handle, uint16_t>
abstract_broker::add_tcp_doorman(uint16_t port, const char* in,
bool reuse_addr) {
......@@ -351,13 +252,13 @@ void abstract_broker::invoke_message(mailbox_element_ptr& ptr) {
}
void abstract_broker::invoke_message(const actor_addr& sender,
message_id mid, message& msg) {
message_id mid,
message& msg) {
auto ptr = mailbox_element::make(sender, mid, message{});
ptr->msg.swap(msg);
invoke_message(ptr);
if (ptr) {
if (ptr)
ptr->msg.swap(msg);
}
}
void abstract_broker::close_all() {
......
......@@ -23,6 +23,10 @@ namespace caf {
namespace io {
namespace network {
acceptor_manager::acceptor_manager(abstract_broker* ptr) : manager(ptr) {
// nop
}
acceptor_manager::~acceptor_manager() {
// nop
}
......
......@@ -37,33 +37,155 @@ namespace basp {
* free functions *
******************************************************************************/
std::string to_string(message_type x) {
switch (x) {
case basp::message_type::server_handshake:
return "server_handshake";
case basp::message_type::client_handshake:
return "client_handshake";
case basp::message_type::dispatch_message:
return "dispatch_message";
case basp::message_type::announce_proxy_instance:
return "announce_proxy_instance";
case basp::message_type::kill_proxy_instance:
return "kill_proxy_instance";
case basp::message_type::dispatch_error:
return "dispatch_error";
default:
return "???";
}
}
std::string to_string(const header &hdr) {
std::ostringstream oss;
oss << "{" << to_string(hdr.source_node) << ", " << to_string(hdr.dest_node)
<< ", " << hdr.source_actor << ", " << hdr.dest_actor << ", "
<< hdr.payload_len << ", " << hdr.operation << ", " << hdr.operation_data
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();
}
void read_hdr(deserializer& source, header& hdr) {
source.read(reinterpret_cast<uint32_t&>(hdr.operation))
.read(hdr.payload_len)
.read(hdr.operation_data);
hdr.source_node.deserialize(source);
hdr.dest_node.deserialize(source);
source.read(hdr.source_actor)
.read(hdr.dest_actor)
.read(hdr.payload_len)
.read(hdr.operation)
.read(hdr.operation_data);
.read(hdr.dest_actor);
}
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.dest_node.serialize(sink);
sink.write(hdr.source_actor)
.write(hdr.dest_actor)
.write(hdr.payload_len)
.write(hdr.operation)
.write(hdr.operation_data);
.write(hdr.dest_actor);
}
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.source_actor)
&& zero(hdr.dest_actor)
&& ! zero(hdr.payload_len)
&& ! 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 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>
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::dispatch_error:
return dispatch_error_valid(hdr);
}
}
/******************************************************************************
......@@ -208,8 +330,37 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
CAF_LOG_WARNING("received invalid header");
return err();
}
// needs forwarding?
if (! is_handshake(hdr) && hdr.dest_node != this_node_) {
auto path = lookup(hdr.dest_node);
if (path) {
auto& wr_buf = path->wr_buf;
binary_serializer bs{std::back_inserter(wr_buf), &get_namespace()};
write_hdr(bs, hdr);
wr_buf.insert(wr_buf.end(), payload->begin(), payload->end());
tbl_.flush(*path);
} else {
CAF_LOG_INFO("cannot forward message, no route to destination");
if (hdr.source_node != this_node_) {
auto reverse_path = lookup(hdr.source_node);
if (! reverse_path) {
CAF_LOG_WARNING("cannot send error message: no route to source");
} else {
write_dispatch_error(reverse_path->wr_buf,
this_node_,
hdr.source_node,
error::no_route_to_destination,
hdr,
*payload);
}
} else {
CAF_LOG_WARNING("lost packet with probably spoofed source");
}
}
return await_header;
}
switch (hdr.operation) {
case server_handshake: {
case message_type::server_handshake: {
binary_deserializer bd{payload->data(), payload->size(),
&get_namespace()};
actor_id aid;
......@@ -238,15 +389,15 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
return err();
}
write(path->wr_buf,
message_type::client_handshake, nullptr, 0,
this_node_, hdr.source_node,
invalid_actor_id, invalid_actor_id,
nullptr, client_handshake, 0);
invalid_actor_id, invalid_actor_id);
// tell client to create proxy etc.
callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(*path);
break;
}
case client_handshake:
case message_type::client_handshake:
if (tbl_.lookup_direct(hdr.source_node) != invalid_connection_handle) {
CAF_LOG_INFO("received second client handshake for "
<< to_string(hdr.source_node) << ", close connection");
......@@ -255,77 +406,18 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
// add direct route to this node
tbl_.add_direct(dm.handle, hdr.source_node);
break;
case dispatch_message: {
if (hdr.dest_node == this_node_) {
// message is addressed to us
binary_deserializer bd{payload->data(), payload->size(),
&get_namespace()};
// skip size of serialized message
bd.advance(sizeof(uint32_t));
message msg;
msg.deserialize(bd);
callee_.deliver(hdr.source_node, hdr.source_actor,
hdr.dest_node, hdr.dest_actor, msg,
message_id::from_integer_value(hdr.operation_data));
} else {
// message needs forwarding
auto path = lookup(hdr.dest_node);
if (path) {
// detect loop (skip message, then iterate forwarding nodes)
binary_deserializer bd{payload->data(), payload->size(),
&get_namespace()};
auto msg_size = bd.read<uint32_t>();
bd.advance(msg_size);
while (! bd.at_end()) {
bool loop_detected = false;
if (! bd.buf_equals(this_node_.host_id().data(),
node_id::host_id_size))
loop_detected = true;
else
loop_detected = bd.advance(node_id::host_id_size)
.read<uint32_t>();
if (loop_detected) {
CAF_LOG_WARNING("cannot send error message: no route to source");
auto reverse_path = lookup(hdr.source_node);
if (! reverse_path) {
CAF_LOG_WARNING("cannot send error message: no route to source");
} else {
write_dispatch_error(reverse_path->wr_buf,
this_node_,
hdr.source_node,
error::no_route_to_destination,
hdr,
*payload);
}
return await_header;
}
}
// append this node to the forwaring path
hdr.payload_len += node_id::serialized_size;
auto& wr_buf = path->wr_buf;
binary_serializer bs{std::back_inserter(wr_buf), &get_namespace()};
write_hdr(bs, hdr);
wr_buf.insert(wr_buf.end(), payload->begin(), payload->end());
this_node_.serialize(bs);
tbl_.flush(*path);
} else {
CAF_LOG_INFO("cannot forward message, no route to destination");
auto reverse_path = lookup(hdr.source_node);
if (! reverse_path) {
CAF_LOG_WARNING("cannot send error message: no route to source");
} else {
write_dispatch_error(reverse_path->wr_buf,
this_node_,
hdr.source_node,
error::no_route_to_destination,
hdr,
*payload);
}
}
}
case message_type::dispatch_message: {
// message is addressed to us
binary_deserializer bd{payload->data(), payload->size(),
&get_namespace()};
message msg;
msg.deserialize(bd);
callee_.deliver(hdr.source_node, hdr.source_actor,
hdr.dest_node, hdr.dest_actor, msg,
message_id::from_integer_value(hdr.operation_data));
break;
}
case dispatch_error:
case message_type::dispatch_error:
// needs forwarding?
if (hdr.dest_node != this_node_) {
auto path = lookup(hdr.dest_node);
......@@ -343,7 +435,7 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
}
return await_header;
}
switch (hdr.operation_data) {
switch (static_cast<error>(hdr.operation_data)) {
case error::loop_detected:
case error::no_route_to_destination:
// TODO: do some better error handling for detected loops
......@@ -372,10 +464,10 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
break;
}
break;
case announce_proxy_instance:
case message_type::announce_proxy_instance:
callee_.proxy_announced(hdr.source_node, hdr.dest_actor);
break;
case kill_proxy_instance:
case message_type::kill_proxy_instance:
callee_.kill_proxy(hdr.source_node, hdr.source_actor,
static_cast<uint32_t>(hdr.operation_data));
break;
......@@ -478,29 +570,32 @@ bool instance::dispatch(const actor_addr& sender, const actor_addr& receiver,
auto writer = make_callback([&](serializer& sink) {
msg.serialize(sink);
});
header hdr{sender ? sender->node() : this_node(), receiver->node(),
sender ? sender->id() : invalid_actor_id, receiver->id(),
0, dispatch_message, mid.integer_value()};
header hdr{message_type::dispatch_message, 0, mid.integer_value(),
sender ? sender->node() : this_node(), receiver->node(),
sender ? sender->id() : invalid_actor_id, receiver->id()};
write(path->wr_buf, hdr, &writer);
flush(*path);
return true;
}
void instance::write(buffer_type& buf, const node_id& source_node,
const node_id& dest_node, actor_id source_actor,
actor_id dest_actor, uint32_t* payload_len,
uint32_t operation, uint64_t operation_data,
payload_writer* pw,
bool suppress_auto_size_prefixing) {
void instance::write(buffer_type& buf,
message_type operation,
uint32_t* payload_len,
uint64_t operation_data,
const node_id& source_node,
const node_id& dest_node,
actor_id source_actor,
actor_id dest_actor,
payload_writer* pw) {
if (! pw) {
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);
dest_node.serialize(bs);
bs.write(source_actor)
.write(dest_actor)
.write(uint32_t{0})
.write(operation)
.write(operation_data);
.write(dest_actor);
} else {
// reserve space in the buffer to write the payload later on
auto wr_pos = static_cast<ptrdiff_t>(buf.size());
......@@ -509,37 +604,26 @@ void instance::write(buffer_type& buf, const node_id& source_node,
auto pl_pos = buf.size();
{ // lifetime scope of first serializer (write payload)
binary_serializer bs1{std::back_inserter(buf), &get_namespace()};
if (operation == dispatch_message && ! suppress_auto_size_prefixing) {
// prefix buffer with its size for dispatch_message
bs1.write(uint32_t{0});
(*pw)(bs1);
bs1.reset(buf.begin() + static_cast<ptrdiff_t>(pl_pos));
bs1.write(static_cast<uint32_t>(buf.size() - pl_pos
- sizeof(uint32_t)));
} else {
(*pw)(bs1);
}
(*pw)(bs1);
}
// write broker message to the reserved space
binary_serializer bs2{buf.begin() + wr_pos, &get_namespace()};
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);
dest_node.serialize(bs2);
bs2.write(source_actor)
.write(dest_actor)
.write(plen)
.write(operation)
.write(operation_data);
.write(dest_actor);
if (payload_len)
*payload_len = plen;
}
}
void instance::write(buffer_type& buf, header& hdr, payload_writer* pw,
bool suppress_auto_size_prefixing) {
write(buf, hdr.source_node, hdr.dest_node, hdr.source_actor, hdr.dest_actor,
&hdr.payload_len, hdr.operation, hdr.operation_data, pw,
suppress_auto_size_prefixing);
void instance::write(buffer_type& buf, header& hdr, payload_writer* pw) {
write(buf, hdr.operation, &hdr.payload_len, hdr.operation_data,
hdr.source_node, hdr.dest_node, hdr.source_actor, hdr.dest_actor, pw);
}
void instance::write_server_handshake(buffer_type& out_buf,
......@@ -559,24 +643,26 @@ void instance::write_server_handshake(buffer_type& out_buf,
for (auto& sig : pa->second)
sink << sig;
});
header hdr{this_node_, invalid_node_id,
invalid_actor_id, invalid_actor_id,
0, server_handshake, version};
header hdr{message_type::server_handshake, 0, version,
this_node_, invalid_node_id,
invalid_actor_id, invalid_actor_id};
write(out_buf, hdr, &writer);
}
void instance::write_dispatch_error(buffer_type& buf,
const node_id& source_node,
const node_id& dest_node,
uint64_t error_code,
error error_code,
const header& original_hdr,
const buffer_type& payload) {
auto writer = make_callback([&](serializer& sink) {
write_hdr(sink, original_hdr);
sink.write_raw(payload.size(), payload.data());
});
header hdr{source_node, dest_node, invalid_actor_id, invalid_actor_id,
0, kill_proxy_instance, error_code};
header hdr{message_type::kill_proxy_instance, 0,
static_cast<uint64_t>(error_code),
source_node, dest_node,
invalid_actor_id, invalid_actor_id};
write(buf, hdr, &writer);
}
......@@ -584,8 +670,9 @@ void instance::write_kill_proxy_instance(buffer_type& buf,
const node_id& dest_node,
actor_id aid,
uint32_t rsn) {
header hdr{this_node_, dest_node, aid, invalid_actor_id,
0, kill_proxy_instance, rsn};
header hdr{message_type::kill_proxy_instance, 0, rsn,
this_node_, dest_node,
aid, invalid_actor_id};
write(buf, hdr);
}
......
......@@ -83,9 +83,9 @@ actor_proxy_ptr basp_broker_state::make_proxy(const node_id& nid,
});
// tell remote side we are monitoring this actor now
instance.write(self->wr_buf(this_context->hdl),
basp::message_type::announce_proxy_instance, nullptr, 0,
this_node(), nid,
invalid_actor_id, aid,
nullptr, basp::announce_proxy_instance, 0);
invalid_actor_id, aid);
instance.tbl().flush(*path);
self->parent().notify<hook::new_remote_actor>(res->address());
return res;
......
......@@ -733,7 +733,7 @@ void default_multiplexer::dispatch_runnable(runnable_ptr ptr) {
connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
default_socket&& sock) {
CAF_LOG_TRACE("");
class impl : public abstract_broker::scribe {
class impl : public scribe {
public:
impl(abstract_broker* ptr, default_socket&& s)
: scribe(ptr, network::conn_hdl_from_socket(s)),
......@@ -746,16 +746,16 @@ connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
stream_.configure_read(config);
if (! launched_) launch();
}
abstract_broker::buffer_type& wr_buf() override {
std::vector<char>& wr_buf() override {
return stream_.wr_buf();
}
abstract_broker::buffer_type& rd_buf() override {
std::vector<char>& rd_buf() override {
return stream_.rd_buf();
}
void stop_reading() override {
CAF_LOG_TRACE("");
stream_.stop_reading();
disconnect(false);
detach(false);
}
void flush() override {
CAF_LOG_TRACE("");
......@@ -771,7 +771,7 @@ connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
bool launched_;
stream<default_socket> stream_;
};
abstract_broker::scribe_ptr ptr = make_counted<impl>(self, std::move(sock));
auto ptr = make_counted<impl>(self, std::move(sock));
self->add_scribe(ptr);
return ptr->hdl();
}
......@@ -781,7 +781,7 @@ default_multiplexer::add_tcp_doorman(abstract_broker* self,
default_socket_acceptor&& sock) {
CAF_LOG_TRACE("sock.fd = " << sock.fd());
CAF_ASSERT(sock.fd() != network::invalid_native_socket);
class impl : public abstract_broker::doorman {
class impl : public doorman {
public:
impl(abstract_broker* ptr, default_socket_acceptor&& s)
: doorman(ptr, network::accept_hdl_from_socket(s), port_of_fd(s.fd())),
......@@ -799,7 +799,7 @@ default_multiplexer::add_tcp_doorman(abstract_broker* self,
void stop_reading() override {
CAF_LOG_TRACE("");
acceptor_.stop_reading();
disconnect(false);
detach(false);
}
void launch() override {
CAF_LOG_TRACE("");
......@@ -808,7 +808,7 @@ default_multiplexer::add_tcp_doorman(abstract_broker* self,
private:
network::acceptor<default_socket_acceptor> acceptor_;
};
abstract_broker::doorman_ptr ptr = make_counted<impl>(self, std::move(sock));
auto ptr = make_counted<impl>(self, std::move(sock));
self->add_doorman(ptr);
return ptr->hdl();
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/doorman.hpp"
#include "caf/detail/logging.hpp"
#include "caf/io/abstract_broker.hpp"
namespace caf {
namespace io {
doorman::doorman(abstract_broker* ptr, accept_handle acc_hdl, uint16_t p)
: network::acceptor_manager(ptr),
hdl_(acc_hdl),
accept_msg_(make_message(new_connection_msg{hdl_, connection_handle{}})),
port_(p) {
// nop
}
doorman::~doorman() {
// nop
}
void doorman::detach_from_parent() {
CAF_LOG_TRACE("hdl = " << hdl().id());
parent()->doormen_.erase(hdl());
}
message doorman::detach_message() {
return make_message(acceptor_closed_msg{hdl()});
}
void doorman::io_failure(network::operation op) {
CAF_LOG_TRACE("id = " << hdl().id() << ", "
<< CAF_TARG(op, static_cast<int>));
// keep compiler happy when compiling w/o logging
static_cast<void>(op);
detach(true);
}
} // namespace io
} // namespace caf
......@@ -19,14 +19,40 @@
#include "caf/io/network/manager.hpp"
#include "caf/detail/logging.hpp"
#include "caf/io/abstract_broker.hpp"
namespace caf {
namespace io {
namespace network {
manager::manager(abstract_broker* ptr) : parent_(ptr) {
// nop
}
manager::~manager() {
// nop
}
void manager::set_parent(abstract_broker* ptr) {
if (! detached())
parent_ = ptr;
}
void manager::detach(bool invoke_disconnect_message) {
CAF_LOG_TRACE("");
if (! detached()) {
CAF_LOG_DEBUG("disconnect servant from broker");
detach_from_parent();
if (invoke_disconnect_message) {
auto msg = detach_message();
parent_->invoke_message(parent_->address(),invalid_message_id, msg);
}
parent_ = nullptr;
}
}
} // namespace network
} // 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/scribe.hpp"
#include "caf/detail/logging.hpp"
#include "caf/io/abstract_broker.hpp"
namespace caf {
namespace io {
scribe::scribe(abstract_broker* ptr, connection_handle conn_hdl)
: network::stream_manager(ptr),
hdl_(conn_hdl) {
std::vector<char> tmp;
read_msg_ = make_message(new_data_msg{hdl_, std::move(tmp)});
}
void scribe::detach_from_parent() {
CAF_LOG_TRACE("hdl = " << hdl().id());
parent()->scribes_.erase(hdl());
}
scribe::~scribe() {
CAF_LOG_TRACE("");
}
message scribe::detach_message() {
return make_message(connection_closed_msg{hdl()});
}
void scribe::consume(const void*, size_t num_bytes) {
CAF_LOG_TRACE(CAF_ARG(num_bytes));
if (detached()) {
// we are already disconnected from the broker while the multiplexer
// did not yet remove the socket, this can happen if an IO event causes
// the broker to call close_all() while the pollset contained
// further activities for the broker
return;
}
auto& buf = rd_buf();
CAF_ASSERT(buf.size() >= num_bytes);
// make sure size is correct, swap into message, and then call client
buf.resize(num_bytes);
read_msg().buf.swap(buf);
parent()->invoke_message(invalid_actor_addr, invalid_message_id, read_msg_);
// swap buffer back to stream and implicitly flush wr_buf()
read_msg().buf.swap(buf);
flush();
}
void scribe::io_failure(network::operation op) {
CAF_LOG_TRACE("id = " << hdl().id()
<< ", " << CAF_TARG(op, static_cast<int>));
// keep compiler happy when compiling w/o logging
static_cast<void>(op);
detach(true);
}
} // namespace io
} // namespace caf
......@@ -23,6 +23,10 @@ namespace caf {
namespace io {
namespace network {
stream_manager::stream_manager(abstract_broker* ptr) : manager(ptr) {
// nop
}
stream_manager::~stream_manager() {
// nop
}
......
......@@ -40,32 +40,32 @@ connection_handle test_multiplexer::new_tcp_scribe(const std::string& host,
void test_multiplexer::assign_tcp_scribe(abstract_broker* ptr,
connection_handle hdl) {
class impl : public abstract_broker::scribe {
class impl : public scribe {
public:
impl(abstract_broker* self, connection_handle ch, test_multiplexer* mpx)
: abstract_broker::scribe(self, ch),
: scribe(self, ch),
mpx_(mpx) {
// nop
}
void configure_read(receive_policy::config config) {
void configure_read(receive_policy::config config) override {
mpx_->read_config(hdl()) = config;
}
abstract_broker::buffer_type& wr_buf() {
std::vector<char>& wr_buf() override {
return mpx_->output_buffer(hdl());
}
abstract_broker::buffer_type& rd_buf() {
std::vector<char>& rd_buf() override {
return mpx_->input_buffer(hdl());
}
void stop_reading() {
void stop_reading() override {
mpx_->stopped_reading(hdl()) = true;
disconnect(false);
detach(false);
}
void flush() {
void flush() override {
// nop
}
......@@ -106,7 +106,7 @@ test_multiplexer::new_tcp_doorman(uint16_t port, const char*, bool) {
void test_multiplexer::assign_tcp_doorman(abstract_broker* ptr,
accept_handle hdl) {
class impl : public abstract_broker::doorman {
class impl : public doorman {
public:
impl(abstract_broker* self, accept_handle ah, test_multiplexer* mpx)
: doorman(self, ah, mpx->port(ah)),
......@@ -127,7 +127,7 @@ void test_multiplexer::assign_tcp_doorman(abstract_broker* ptr,
void stop_reading() {
mpx_->stopped_reading(hdl()) = true;
disconnect(false);
detach(false);
}
void launch() {
......@@ -204,7 +204,7 @@ bool& test_multiplexer::stopped_reading(connection_handle hdl) {
return scribe_data_[hdl].stopped_reading;
}
abstract_broker::scribe_ptr& test_multiplexer::impl_ptr(connection_handle hdl) {
intrusive_ptr<scribe>& test_multiplexer::impl_ptr(connection_handle hdl) {
return scribe_data_[hdl].ptr;
}
......@@ -216,7 +216,7 @@ bool& test_multiplexer::stopped_reading(accept_handle hdl) {
return doorman_data_[hdl].stopped_reading;
}
abstract_broker::doorman_ptr& test_multiplexer::impl_ptr(accept_handle hdl) {
intrusive_ptr<doorman>& test_multiplexer::impl_ptr(accept_handle hdl) {
return doorman_data_[hdl].ptr;
}
......
......@@ -63,25 +63,6 @@ string hexstr(const buffer& buf) {
return oss.str();
}
string get_operation(uint64_t operation) {
switch (operation) {
case basp::server_handshake:
return "server_handshake";
case basp::client_handshake:
return "client_handshake";
case basp::dispatch_message:
return "dispatch_message";
case basp::announce_proxy_instance:
return "announce_proxy_instance";
case basp::kill_proxy_instance:
return "kill_proxy_instance";
case basp::dispatch_error:
return "dispatch_error";
default:
return "???";
}
}
class fixture {
public:
fixture() {
......@@ -199,7 +180,7 @@ public:
using payload_writer = basp::instance::payload_writer;
void to_buf(buffer& buf, basp::header& hdr, payload_writer* writer) {
instance().write(buf, hdr, writer, true);
instance().write(buf, hdr, writer);
}
template <class T, class... Ts>
......@@ -243,15 +224,21 @@ public:
// technically, the server handshake arrives
// before we send the client handshake
mock(hdl,
{remote_node(i), this_node(),
invalid_actor_id, invalid_actor_id,
0, basp::client_handshake, 0})
{basp::message_type::client_handshake, 0, 0,
remote_node(i), this_node(),
invalid_actor_id, invalid_actor_id})
.expect(hdl,
{this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id,
0, basp::server_handshake, basp::version},
{basp::message_type::server_handshake, 0, basp::version,
this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id},
published_actor_id,
published_actor_ifs);
// test whether basp instance correctly updates the
// routing table upon receiving client handshakes
auto path = tbl().lookup(remote_node(i));
CAF_REQUIRE(path != none);
CAF_CHECK(path->hdl == remote_hdl(i));
CAF_CHECK(path->next_hop == remote_node(i));
}
std::pair<basp::header, buffer> read_from_out_buf(connection_handle hdl) {
......@@ -264,26 +251,15 @@ public:
return result;
}
void dispatch_out_buf(connection_handle hdl, vector<node_id> hops = {}) {
void dispatch_out_buf(connection_handle hdl) {
basp::header hdr;
buffer buf;
std::tie(hdr, buf) = read_from_out_buf(hdl);
CAF_MESSAGE("dispatch output buffer for connection " << hdl.id());
CAF_REQUIRE(hdr.operation == basp::dispatch_message);
CAF_REQUIRE(hdr.operation == basp::message_type::dispatch_message);
message msg;
auto source = make_deserializer(buf);
auto msg_size = source.read<uint32_t>();
auto expected_payload_len = static_cast<uint32_t>(
msg_size + sizeof(uint32_t) + (node_id::serialized_size * hops.size()));
CAF_CHECK(expected_payload_len == hdr.payload_len);
msg.deserialize(source);
vector<node_id> forwarder_nodes;
for (size_t i = 0; i < hops.size(); ++i) {
node_id nid;
nid.deserialize(source);
forwarder_nodes.emplace_back(std::move(nid));
}
CAF_CHECK(hops == forwarder_nodes);
auto registry = detail::singletons::get_actor_registry();
auto src = registry->get(hdr.source_actor);
auto dest = registry->get(hdr.dest_actor);
......@@ -309,7 +285,7 @@ public:
template <class... Ts>
mock_t& expect(connection_handle hdl, basp::header hdr, const Ts&... xs) {
CAF_MESSAGE("expect " << num++ << ". sent message to be a "
<< get_operation(hdr.operation));
<< to_string(hdr.operation));
buffer buf;
this_->to_buf(buf, hdr, nullptr, xs...);
buffer& ob = this_->mpx()->output_buffer(hdl);
......@@ -340,7 +316,7 @@ public:
template <class... Ts>
mock_t mock(connection_handle hdl, basp::header hdr, const Ts&... xs) {
CAF_MESSAGE("virtually send " << get_operation(hdr.operation));
CAF_MESSAGE("virtually send " << to_string(hdr.operation));
buffer buf;
to_buf(buf, hdr, nullptr, xs...);
mpx()->virtual_send(hdl, buf);
......@@ -378,10 +354,11 @@ CAF_TEST(empty_server_handshake) {
auto is_zero = [](char c) {
return c == 0;
};
basp::header expected{this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id,
basp::header expected{basp::message_type::server_handshake,
sizeof(actor_id) + sizeof(uint32_t),
basp::server_handshake, basp::version};
basp::version,
this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id};
CAF_CHECK(basp::valid(hdr));
CAF_CHECK(basp::is_handshake(hdr));
CAF_CHECK(hdr == expected);
......@@ -396,9 +373,9 @@ CAF_TEST(non_empty_server_handshake) {
{"caf::replies_to<@u16>::with<@u16>"});
instance().write_server_handshake(buf, 4242);
buffer expected_buf;
basp::header expected{this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id,
0, basp::server_handshake, basp::version};
basp::header expected{basp::message_type::server_handshake, 0, basp::version,
this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id};
to_buf(expected_buf, expected, nullptr,
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"});
CAF_CHECK(hexstr(buf) == hexstr(expected_buf));
......@@ -406,24 +383,15 @@ CAF_TEST(non_empty_server_handshake) {
CAF_TEST(client_handshake_and_dispatch) {
connect_node(0, none, invalid_actor_id, {});
// test whether basp instance correctly updates the
// routing table upon receiving client handshakes
auto path = tbl().lookup(remote_node(0));
CAF_CHECK(path
&& path->hdl == remote_hdl(0)
&& path->next_hop == remote_node(0));
// send a message via `dispatch` from node 0
auto msg = make_message(1, 2, 3);
auto msg_size = serialized_size(msg);
mock(remote_hdl(0),
{remote_node(0), this_node(), pseudo_remote(0)->id(), self()->id(),
0, basp::dispatch_message, 0},
msg_size,
msg)
{basp::message_type::dispatch_message, 0, 0,
remote_node(0), this_node(), pseudo_remote(0)->id(), self()->id()},
make_message(1, 2, 3))
.expect(remote_hdl(0),
{this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id(),
0, basp::announce_proxy_instance, 0});
{basp::message_type::announce_proxy_instance, 0, 0,
this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id()});
// must've created a proxy for our remote actor
CAF_REQUIRE(get_namespace().count_proxies(remote_node(0)) == 1);
// must've send remote node a message that this proxy is monitored now
......@@ -453,21 +421,17 @@ CAF_TEST(message_forwarding) {
connect_node(0, none, invalid_actor_id, {});
connect_node(1, none, invalid_actor_id, {});
auto msg = make_message(1, 2, 3);
auto msg_size = serialized_size(msg);
// send a message from node 0 to node 1, forwarded by this node
mock(remote_hdl(0),
{remote_node(0), remote_node(1),
invalid_actor_id, pseudo_remote(1)->id(),
0, basp::dispatch_message, 0},
msg_size,
{basp::message_type::dispatch_message, 0, 0,
remote_node(0), remote_node(1),
invalid_actor_id, pseudo_remote(1)->id()},
msg)
.expect(remote_hdl(1),
{remote_node(0), remote_node(1),
invalid_actor_id, pseudo_remote(1)->id(),
0, basp::dispatch_message, 0},
msg_size,
msg,
this_node());
{basp::message_type::dispatch_message, 0, 0,
remote_node(0), remote_node(1),
invalid_actor_id, pseudo_remote(1)->id()},
msg);
}
CAF_TEST(publish_and_connect) {
......@@ -490,19 +454,19 @@ CAF_TEST(remote_actor_and_send) {
// build a fake server handshake containing the id of our first pseudo actor
CAF_TEST_VERBOSE("server handshake => client handshake + proxy announcement");
mock(remote_hdl(0),
{remote_node(0), invalid_node_id,
invalid_actor_id, invalid_actor_id,
0, basp::server_handshake, basp::version},
{basp::message_type::server_handshake, 0, basp::version,
remote_node(0), invalid_node_id,
invalid_actor_id, invalid_actor_id},
pseudo_remote(0)->id(),
uint32_t{0})
.expect(remote_hdl(0),
{this_node(), remote_node(0),
invalid_actor_id, invalid_actor_id,
0, basp::client_handshake, 0})
{basp::message_type::client_handshake, 0, 0,
this_node(), remote_node(0),
invalid_actor_id, invalid_actor_id})
.expect(remote_hdl(0),
{this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id(), 0,
basp::announce_proxy_instance, 0});
{basp::message_type::announce_proxy_instance, 0, 0,
this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id()});
// basp broker should've send the proxy
f.await(
[&](ok_atom, actor_addr res) {
......@@ -526,19 +490,16 @@ CAF_TEST(remote_actor_and_send) {
mpx()->exec_runnable(); // process forwarded message in basp_broker
mock()
.expect(remote_hdl(0),
{this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id(),
0, basp::dispatch_message, 0},
serialized_size(make_message(42)),
{basp::message_type::dispatch_message, 0, 0,
this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id()},
make_message(42));
auto msg = make_message("hi there!");
auto msg_size = serialized_size(msg);
CAF_MESSAGE("send message via BASP (from proxy)");
mock(remote_hdl(0),
{remote_node(0), this_node(),
pseudo_remote(0)->id(), self()->id(),
0, basp::dispatch_message, 0},
msg_size,
{basp::message_type::dispatch_message, 0, 0,
remote_node(0), this_node(),
pseudo_remote(0)->id(), self()->id()},
make_message("hi there!"));
self()->receive(
[&](const string& str) {
......@@ -563,21 +524,19 @@ CAF_TEST(actor_serialize_and_deserialize) {
auto prx = get_namespace().get_or_put(remote_node(0), pseudo_remote(0)->id());
mock()
.expect(remote_hdl(0),
{this_node(), prx->node(),
invalid_actor_id, prx->id(),
0, basp::announce_proxy_instance, 0});
{basp::message_type::announce_proxy_instance, 0, 0,
this_node(), prx->node(),
invalid_actor_id, prx->id()});
CAF_CHECK(prx->node() == remote_node(0));
CAF_CHECK(prx->id() == pseudo_remote(0)->id());
auto testee = spawn(testee_impl);
registry()->put(testee->id(), testee->address());
CAF_MESSAGE("send message via BASP (from proxy)");
auto msg = make_message(prx->address());
auto msg_size = serialized_size(msg);
mock(remote_hdl(0),
{prx->node(), this_node(),
prx->id(), testee->id(),
0, basp::dispatch_message, 0},
msg_size,
{basp::message_type::dispatch_message, 0, 0,
prx->node(), this_node(),
prx->id(), testee->id()},
msg);
// testee must've responded (process forwarded message in BASP broker)
CAF_MESSAGE("exec runnable, i.e., handle response from testee");
......@@ -585,10 +544,9 @@ CAF_TEST(actor_serialize_and_deserialize) {
// output buffer must contain the reflected message
mock()
.expect(remote_hdl(0),
{this_node(), prx->node(),
testee->id(), prx->id(),
0, basp::dispatch_message, 0},
msg_size,
{basp::message_type::dispatch_message, 0, 0,
this_node(), prx->node(),
testee->id(), prx->id()},
msg);
}
......
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