Commit 625ff651 authored by Dominik Charousset's avatar Dominik Charousset

Remove unused stream multiplexers

parent 113bbae9
......@@ -55,7 +55,6 @@ set (LIBCAF_CORE_SRCS
src/group.cpp
src/group_manager.cpp
src/group_module.cpp
src/incoming_stream_multiplexer.cpp
src/invoke_result_visitor.cpp
src/local_actor.cpp
src/logger.cpp
......@@ -70,7 +69,6 @@ set (LIBCAF_CORE_SRCS
src/message_view.cpp
src/monitorable_actor.cpp
src/node_id.cpp
src/outgoing_stream_multiplexer.cpp
src/parse_ini.cpp
src/pretty_type_name.cpp
src/private_thread.cpp
......@@ -96,7 +94,6 @@ set (LIBCAF_CORE_SRCS
src/stream_handler.cpp
src/stream_id.cpp
src/stream_msg_visitor.cpp
src/stream_multiplexer.cpp
src/stream_priority.cpp
src/stream_sink.cpp
src/stream_source.cpp
......
......@@ -129,7 +129,7 @@ public:
friend class abstract_actor;
/// The number of actors implictly spawned by the actor system on startup.
static constexpr size_t num_internal_actors = 3;
static constexpr size_t num_internal_actors = 2;
/// Returns the ID of an internal actor by its name.
/// @pre x in {'SpawnServ', 'ConfigServ', 'StreamServ'}
......@@ -148,12 +148,6 @@ public:
return internal_actors_[internal_actor_id(atom("ConfigServ"))];
}
/// Returns the internal actor for managing streams that
/// cross network boundaries.
inline const strong_actor_ptr& stream_serv() const {
return internal_actors_[internal_actor_id(atom("StreamServ"))];
}
actor_system() = delete;
actor_system(const actor_system&) = delete;
actor_system& operator=(const actor_system&) = delete;
......@@ -545,10 +539,6 @@ private:
internal_actors_[internal_actor_id(atom("ConfigServ"))] = std::move(x);
}
/// Sets the internal actor for managing streams that
/// cross network boundaries. Called in middleman::start.
void stream_serv(strong_actor_ptr x);
std::atomic<size_t> ids_;
uniform_type_info_map types_;
node_id node_;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* 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_DETAIL_INCOMING_STREAM_MULTIPLEXER_HPP
#define CAF_DETAIL_INCOMING_STREAM_MULTIPLEXER_HPP
#include <deque>
#include <cstdint>
#include <algorithm>
#include <unordered_map>
#include "caf/actor.hpp"
#include "caf/error.hpp"
#include "caf/local_actor.hpp"
#include "caf/stream_msg.hpp"
#include "caf/detail/stream_multiplexer.hpp"
namespace caf {
namespace detail {
// Forwards messages from local actors to a remote stream_serv.
class incoming_stream_multiplexer : public stream_multiplexer {
public:
/// Allow `variant` to recognize this type as a visitor.
using result_type = void;
incoming_stream_multiplexer(local_actor* self, backend& service);
void operator()(stream_msg& x);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::open&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::ack_open&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::batch&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::ack_batch&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::close&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::abort&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::downstream_failed&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::upstream_failed&);
private:
// Forwards the current stream_msg upstream.
// @pre `current_stream_msg != nullptr`
void forward_to_upstream();
// Forwards the current stream_msg downstream.
// @pre `current_stream_msg != nullptr`
void forward_to_downstream();
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_INCOMING_STREAM_MULTIPLEXER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* 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_DETAIL_STREAM_SERV_DOWNSTREAM_HPP
#define CAF_DETAIL_STREAM_SERV_DOWNSTREAM_HPP
#include <deque>
#include <cstdint>
#include <algorithm>
#include <unordered_map>
#include "caf/actor.hpp"
#include "caf/error.hpp"
#include "caf/stream_msg.hpp"
#include "caf/local_actor.hpp"
#include "caf/detail/stream_multiplexer.hpp"
namespace caf {
namespace detail {
// Forwards messages from local actors to a remote stream_serv.
class outgoing_stream_multiplexer : public stream_multiplexer {
public:
/// Allow `variant` to recognize this type as a visitor.
using result_type = void;
outgoing_stream_multiplexer(local_actor* self, backend& service);
void operator()(stream_msg& x);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::open&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::ack_open&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::batch&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::ack_batch&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::close&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::abort&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::downstream_failed&);
/// @pre `self_->current_sender() != nullptr && current_stream_state_valid()`
void operator()(stream_msg::upstream_failed&);
private:
// Forwards the current stream_msg upstream.
// @pre `current_stream_msg != nullptr`
void forward_to_upstream();
// Forwards the current stream_msg downstream.
// @pre `current_stream_msg != nullptr`
void forward_to_downstream();
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_STREAM_SERV_DOWNSTREAM_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* 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_DETAIL_STREAM_MULTIPLEXER_HPP
#define CAF_DETAIL_STREAM_MULTIPLEXER_HPP
#include <deque>
#include <vector>
#include <cstdint>
#include <algorithm>
#include <unordered_map>
#include "caf/node_id.hpp"
#include "caf/optional.hpp"
#include "caf/stream_msg.hpp"
#include "caf/message_id.hpp"
#include "caf/local_actor.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/actor_control_block.hpp"
namespace caf {
namespace detail {
// Forwards messages from local actors to a remote stream_serv.
class stream_multiplexer {
public:
/// Represents a path to a remote spawn server and stores required meta data.
struct remote_path {
remote_path(remote_path&&) = default;
remote_path& operator=(remote_path&&) = default;
inline remote_path(strong_actor_ptr ptr)
: hdl(ptr),
credit(1),
in_flight(1) {
// We start at credit 1 and in_flight 1. This means sending the first
// message to a remote spawn server does not require previous handshaking
// (other than establishing a connection).
}
/// Handle to a remote stream server.
strong_actor_ptr hdl;
/// Buffer for outgoing messages (sent to the BASP broker).
std::deque<mailbox_element_ptr> buf;
/// Available credit for sending messages.
int32_t credit;
/// Capacity that we have granted the remote stream server.
int32_t in_flight;
};
/// Maps node IDs to remote paths.
using remote_paths = std::unordered_map<node_id, remote_path>;
/// The backend of a stream server downstream establishes connections to
/// remote stream servers via node ID.
class backend {
public:
backend(actor basp_ref);
virtual ~backend();
/// Returns a remote actor representing the stream serv of node `nid`.
/// Returns an invalid handle if a) `nid` is invalid or identitifies this
/// node, or b) the backend could not establish a connection.
virtual strong_actor_ptr remote_stream_serv(const node_id& nid) = 0;
/// Returns a reference to the basp_ broker.
inline actor& basp() {
return basp_;
}
/// Returns all known remote stream servers and available credit.
inline remote_paths& remotes() {
return remotes_;
}
inline const remote_paths& remotes() const {
return remotes_;
}
/// Queries whether `x` is a known remote node.
inline bool has_remote_path(const node_id& x) const {
return remotes().count(x) > 0;
}
/// Adds `ptr` as remote stream serv on `x`. This is a no-op if `x` already
/// has a known path.
inline void add_remote_path(const node_id& x, strong_actor_ptr ptr) {
remotes().emplace(x, std::move(ptr));
}
/// Called whenever `nid` grants us `x` more credit.
// @pre `current_stream_state_ != streams_.end()`
void add_credit(const node_id& nid, int32_t x);
// Drains as much from the buffer by sending messages to the remote
// spawn_serv as possible, i.e., as many messages as credit is available.
void drain_buf(remote_path& path);
protected:
/// A reference to the basp_ broker.
actor basp_;
/// Known remote stream servers and available credit.
remote_paths remotes_;
};
/// Stores previous and next stage for a stream as well as the corresponding
/// remote path.
struct stream_state {
strong_actor_ptr prev_stage;
strong_actor_ptr next_stage;
remote_path* rpath;
};
// Maps stream ID to stream states.
using stream_states = std::unordered_map<stream_id, stream_state>;
/// Creates a new stream multiplexer for `self`, using `service` to connect to
/// remote spawn servers, and `basp` to send messages to remotes.
/// @pre `self != nullptr && basp != nullptr`
stream_multiplexer(local_actor* self, backend& service);
/// Adds stream state for `current_stream_msg_->sid`.
stream_states::iterator add_stream(strong_actor_ptr prev,
strong_actor_ptr next,
remote_path* current_path);
/// Removes `current_stream_state_`.
void remove_current_stream();
/// Queries whether stream `x` is managed by this multiplexer.
inline bool has_stream(const stream_id& x) const {
return streams_.count(x) > 0;
}
/// Queries the number of open streams.
inline size_t num_streams() const {
return streams_.size();
}
protected:
inline bool current_stream_state_valid() const {
return current_stream_state_ != streams_.end();
}
// Dispatches `x` on the subtype `T`.
template <class T>
static void dispatch(T& derived, stream_msg& x) {
// Reject anonymous messages.
auto prev = derived.self_->current_sender();
if (prev != nullptr) {
// Set state for the message handlers.
derived.current_stream_msg_ = &x;
derived.current_stream_state_ = derived.streams_.find(x.sid);
// Make sure that handshakes are not received twice and drop
// non-handshake messages if no state for the stream is found.
if (holds_alternative<stream_msg::open>(x.content)) {
if (derived.current_stream_state_ == derived.streams_.end()) {
derived(get<stream_msg::open>(x.content));
} else {
CAF_LOG_ERROR("Received multiple handshakes for stream.");
derived.fail(sec::upstream_already_exists);
}
} else {
if (derived.current_stream_state_ != derived.streams_.end()) {
visit(derived, x.content);
} else {
CAF_LOG_ERROR("Unable to access required stream and/or path state.");
derived.fail(sec::invalid_stream_state);
}
}
}
}
// Returns a reference to the remote stream server instance for `nid`
// if a remote stream_serv is known or connecting is successful.
optional<remote_path&> get_remote_or_try_connect(const node_id& nid);
// Returns a reference to the stream state for `sid`.
optional<stream_state&> state_for(const stream_id& sid);
/// Assings new capacity (credit) to remote stream servers.
/// @pre `current_remote_path_ != remotes().end()`
void manage_credit();
// Aborts the current stream with error `reason`.
// @pre `current_stream_msg != nullptr`
void fail(error reason, strong_actor_ptr predecessor,
strong_actor_ptr successor = nullptr);
// Aborts the current stream with error `reason`, assuming `state_for` returns
// valid predecessor and successor.
// @pre `current_stream_msg != nullptr`
void fail(error reason);
// Sends message `x` to the local actor `dest`.
void send_local(strong_actor_ptr& dest, stream_msg&& x,
std::vector<strong_actor_ptr> stages = {},
message_id mid = message_id::make());
// Creates a new message for the BASP broker.
inline mailbox_element_ptr
make_basp_message(remote_path& path, message&& x,
std::vector<strong_actor_ptr> stages = {},
message_id mid = message_id::make()) {
return make_mailbox_element(self_->ctrl(), message_id::make(), {},
forward_atom::value,
strong_actor_ptr{self_->ctrl()},
std::move(stages), path.hdl, mid, std::move(x));
}
// Sends message `x` to the remote stream server `dest`.
inline void send_remote(remote_path& path, stream_msg&& x,
std::vector<strong_actor_ptr> stages = {},
message_id mid = message_id::make()) {
path.buf.emplace_back(make_basp_message(path, make_message(std::move(x)),
std::move(stages), mid));
service_.drain_buf(path);
}
// Sends the control message `x` to the remote stream server `dest`. A control
// message signals capaticity and therefore does not use credit on its own and
// is sent immediately.
inline void send_remote_ctrl(remote_path& path, message&& x) {
basp()->enqueue(make_basp_message(path, std::move(x)), self_->context());
}
/// Returns a reference to the basp_ broker.
inline actor& basp() const {
return service_.basp();
}
/// Returns all known remote stream servers and available credit.
inline remote_paths& remotes() {
return service_.remotes();
}
/// Returns all known remote stream servers and available credit.
inline const remote_paths& remotes() const {
return service_.remotes();
}
// Stores which stream is currently processed.
stream_msg* current_stream_msg_ = nullptr;
/// Stores which stream state belongs to `current_stream_msg_`.
stream_states::iterator current_stream_state_;
// The parent actor.
local_actor* self_;
// The remoting backend.
backend& service_;
private:
// Open streams.
stream_states streams_;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_STREAM_MULTIPLEXER_HPP
......@@ -32,7 +32,7 @@ class forwarding_actor_proxy : public actor_proxy {
public:
using forwarding_stack = std::vector<strong_actor_ptr>;
forwarding_actor_proxy(actor_config& cfg, actor dest, actor stream_serv1);
forwarding_actor_proxy(actor_config& cfg, actor dest);
~forwarding_actor_proxy() override;
......@@ -50,7 +50,6 @@ private:
mutable detail::shared_spinlock mtx_;
actor broker_;
actor stream_serv_;
};
} // namespace caf
......
......@@ -438,9 +438,4 @@ actor_system::dyn_spawn_impl(const std::string& name, message& args,
return std::move(res.first);
}
void actor_system::stream_serv(strong_actor_ptr x) {
internal_actors_[internal_actor_id(atom("StreamServ"))] = std::move(x);
registry_.put(atom("StreamServ"), stream_serv());
}
} // namespace caf
......@@ -29,11 +29,9 @@
namespace caf {
forwarding_actor_proxy::forwarding_actor_proxy(actor_config& cfg, actor dest,
actor stream_serv)
forwarding_actor_proxy::forwarding_actor_proxy(actor_config& cfg, actor dest)
: actor_proxy(cfg),
broker_(std::move(dest)),
stream_serv_(std::move(stream_serv)) {
broker_(std::move(dest)) {
// nop
}
......@@ -85,11 +83,10 @@ bool forwarding_actor_proxy::remove_backlink(abstract_actor* x) {
void forwarding_actor_proxy::kill_proxy(execution_unit* ctx, error rsn) {
CAF_ASSERT(ctx != nullptr);
actor tmp[2];
actor tmp;
{ // lifetime scope of guard
std::unique_lock<detail::shared_spinlock> guard(mtx_);
broker_.swap(tmp[0]); // manually break cycle
stream_serv_.swap(tmp[1]); // manually break cycle
broker_.swap(tmp); // manually break cycle
}
cleanup(std::move(rsn), ctx);
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* 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/detail/incoming_stream_multiplexer.hpp"
#include "caf/send.hpp"
#include "caf/variant.hpp"
#include "caf/to_string.hpp"
#include "caf/local_actor.hpp"
namespace caf {
namespace detail {
incoming_stream_multiplexer::incoming_stream_multiplexer(local_actor* self,
backend& service)
: stream_multiplexer(self, service) {
// nop
}
void incoming_stream_multiplexer::operator()(stream_msg& x) {
CAF_LOG_TRACE(CAF_ARG(x));
CAF_ASSERT(self_->current_mailbox_element() != nullptr);
dispatch(*this, x);
}
void incoming_stream_multiplexer::operator()(stream_msg::open& x) {
CAF_LOG_TRACE(CAF_ARG(x));
CAF_ASSERT(current_stream_msg_ != nullptr);
auto prev = std::move(x.prev_stage);
// Make sure we have a previous stage.
if (!prev) {
CAF_LOG_WARNING("received stream_msg::open without previous stage");
return fail(sec::invalid_upstream, nullptr);
}
// Make sure we have a next stage.
auto cme = self_->current_mailbox_element();
if (!cme || cme->stages.empty()) {
CAF_LOG_WARNING("received stream_msg::open without next stage");
return fail(sec::invalid_downstream, std::move(prev));
}
auto successor = cme->stages.back();
cme->stages.pop_back();
// Our prev always is the remote stream server proxy.
auto current_remote_path = remotes().emplace(prev->node(), prev).first;
current_stream_state_ = add_stream(std::move(prev), successor,
&current_remote_path->second);
// Rewrite handshake and forward it to the next stage.
x.prev_stage = self_->ctrl();
auto ptr = make_mailbox_element(
cme->sender, cme->mid, std::move(cme->stages),
make<stream_msg::open>(current_stream_msg_->sid, std::move(x.msg),
self_->ctrl(), x.original_stage, x.priority,
x.redeployable));
successor->enqueue(std::move(ptr), self_->context());
// Send out demand upstream.
manage_credit();
}
void incoming_stream_multiplexer::operator()(stream_msg::ack_open&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
forward_to_upstream();
}
void incoming_stream_multiplexer::operator()(stream_msg::batch&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
forward_to_downstream();
}
void incoming_stream_multiplexer::operator()(stream_msg::ack_batch&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
forward_to_upstream();
}
void incoming_stream_multiplexer::operator()(stream_msg::close&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
forward_to_downstream();
remove_current_stream();
}
void incoming_stream_multiplexer::operator()(stream_msg::abort& x) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
if (current_stream_state_->second.prev_stage == self_->current_sender())
fail(x.reason, nullptr, current_stream_state_->second.next_stage);
else
fail(x.reason, current_stream_state_->second.prev_stage);
remove_current_stream();
}
void incoming_stream_multiplexer::operator()(stream_msg::downstream_failed&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
// TODO: implement me
}
void incoming_stream_multiplexer::operator()(stream_msg::upstream_failed&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
// TODO: implement me
}
void incoming_stream_multiplexer::forward_to_upstream() {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
send_remote(*current_stream_state_->second.rpath,
std::move(*current_stream_msg_));
}
void incoming_stream_multiplexer::forward_to_downstream() {
CAF_ASSERT(current_stream_msg_ != nullptr);
// When forwarding downstream, we also have to manage upstream credit.
manage_credit();
send_local(current_stream_state_->second.next_stage,
std::move(*current_stream_msg_));
}
} // namespace detail
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* 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/detail/outgoing_stream_multiplexer.hpp"
#include "caf/send.hpp"
#include "caf/variant.hpp"
#include "caf/to_string.hpp"
#include "caf/local_actor.hpp"
namespace caf {
namespace detail {
outgoing_stream_multiplexer::outgoing_stream_multiplexer(local_actor* self,
backend& service)
: stream_multiplexer(self, service) {
// nop
}
void outgoing_stream_multiplexer::operator()(stream_msg& x) {
CAF_LOG_TRACE(CAF_ARG(x));
CAF_ASSERT(self_->current_mailbox_element() != nullptr);
dispatch(*this, x);
}
void outgoing_stream_multiplexer::operator()(stream_msg::open& x) {
CAF_LOG_TRACE(CAF_ARG(x));
CAF_ASSERT(current_stream_msg_ != nullptr);
auto predecessor = std::move(x.prev_stage);
// Make sure we a previous stage.
if (!predecessor) {
CAF_LOG_WARNING("received stream_msg::open without previous stage");
return fail(sec::invalid_upstream, nullptr);
}
// Make sure we don't receive a handshake for an already open stream.
if (has_stream(current_stream_msg_->sid)) {
CAF_LOG_WARNING("received stream_msg::open twice");
return fail(sec::upstream_already_exists, std::move(predecessor));
}
// Make sure we have a next stage.
auto cme = self_->current_mailbox_element();
if (cme->stages.empty()) {
CAF_LOG_WARNING("received stream_msg::open without next stage");
return fail(sec::invalid_downstream, std::move(predecessor));
}
auto successor = cme->stages.back();
// Get a connection to the responsible stream server.
auto path = get_remote_or_try_connect(successor->node());
if (!path) {
CAF_LOG_WARNING("cannot connect to remote stream server");
return fail(sec::cannot_connect_to_node, std::move(predecessor));
}
// Update state and send handshake to remote stream_serv (via
// middleman/basp_broker).
add_stream(std::move(predecessor), path->hdl, &(*path));
// Send handshake to remote stream_serv (via middleman/basp_broker). We need
// to send this message as `current_sender`. We have do bypass the queue
// since `send_remote` always sends the message from `self_`.
auto ptr = make_mailbox_element(
cme->sender, message_id::make(), {}, forward_atom::value, cme->sender,
std::move(cme->stages), path->hdl, cme->mid,
make_message(make<stream_msg::open>(
current_stream_msg_->sid, std::move(x.msg), self_->ctrl(),
x.original_stage, x.priority, x.redeployable)));
basp()->enqueue(std::move(ptr), self_->context());
}
void outgoing_stream_multiplexer::operator()(stream_msg::ack_open&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
forward_to_upstream();
}
void outgoing_stream_multiplexer::operator()(stream_msg::batch&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
forward_to_downstream();
}
void outgoing_stream_multiplexer::operator()(stream_msg::ack_batch&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
forward_to_upstream();
}
void outgoing_stream_multiplexer::operator()(stream_msg::close&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
forward_to_downstream();
remove_current_stream();
}
void outgoing_stream_multiplexer::operator()(stream_msg::abort& x) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
if (current_stream_state_->second.prev_stage == self_->current_sender())
fail(x.reason, nullptr, current_stream_state_->second.next_stage);
else
fail(x.reason, current_stream_state_->second.prev_stage);
remove_current_stream();
}
void outgoing_stream_multiplexer::operator()(stream_msg::downstream_failed&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
// TODO: implement me
}
void outgoing_stream_multiplexer::operator()(stream_msg::upstream_failed&) {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
// TODO: implement me
}
void outgoing_stream_multiplexer::forward_to_upstream() {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
manage_credit();
send_local(current_stream_state_->second.prev_stage,
std::move(*current_stream_msg_));
}
void outgoing_stream_multiplexer::forward_to_downstream() {
CAF_ASSERT(current_stream_msg_ != nullptr);
CAF_ASSERT(current_stream_state_valid());
send_remote(*current_stream_state_->second.rpath,
std::move(*current_stream_msg_));
}
} // namespace detail
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* 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/detail/stream_multiplexer.hpp"
#include "caf/send.hpp"
#include "caf/variant.hpp"
#include "caf/to_string.hpp"
#include "caf/local_actor.hpp"
#include "caf/stream_aborter.hpp"
namespace caf {
namespace detail {
stream_multiplexer::backend::backend(actor basp_ref) : basp_(basp_ref) {
// nop
}
stream_multiplexer::backend::~backend() {
// nop
}
void stream_multiplexer::backend::add_credit(const node_id& nid, int32_t x) {
auto i = remotes().find(nid);
if (i != remotes().end()) {
auto& path = i->second;
path.credit += x;
drain_buf(path);
}
}
void stream_multiplexer::backend::drain_buf(remote_path& path) {
CAF_LOG_TRACE(CAF_ARG(path));
auto n = std::min(path.credit, static_cast<int32_t>(path.buf.size()));
if (n > 0) {
auto b = path.buf.begin();
auto e = b + n;
for (auto i = b; i != e; ++i)
basp()->enqueue(std::move(*i), nullptr);
path.buf.erase(b, e);
path.credit -= static_cast<int32_t>(n);
}
}
stream_multiplexer::stream_multiplexer(local_actor* self, backend& service)
: self_(self),
service_(service) {
CAF_ASSERT(self_ != nullptr);
}
stream_multiplexer::stream_states::iterator
stream_multiplexer::add_stream(strong_actor_ptr prev, strong_actor_ptr next,
remote_path* path) {
CAF_LOG_TRACE(CAF_ARG(prev) << CAF_ARG(next) << CAF_ARG(path));
auto sid = current_stream_msg_->sid;
auto x = streams_.emplace(sid, stream_state{prev, next, path});
if (x.second) {
stream_aborter::add(prev, self_->address(), sid);
stream_aborter::add(next, self_->address(), sid);
}
return x.first;
}
void stream_multiplexer::remove_current_stream() {
auto& sid = current_stream_state_->first;
auto& st = current_stream_state_->second;
stream_aborter::del(st.prev_stage, self_->address(), sid);
stream_aborter::del(st.next_stage, self_->address(), sid);
streams_.erase(current_stream_state_);
}
optional<stream_multiplexer::remote_path&>
stream_multiplexer::get_remote_or_try_connect(const node_id& nid) {
auto i = remotes().find(nid);
if (i != remotes().end())
return i->second;
auto res = service_.remote_stream_serv(nid);
if (res)
return remotes().emplace(nid, std::move(res)).first->second;
return none;
}
optional<stream_multiplexer::stream_state&>
stream_multiplexer::state_for(const stream_id& sid) {
auto i = streams_.find(sid);
if (i != streams_.end())
return i->second;
return none;
}
void stream_multiplexer::manage_credit() {
auto& path = *current_stream_state_->second.rpath;
// todo: actual, adaptive credit management
if (--path.in_flight == 0) {
int32_t new_remote_credit = 5;
path.in_flight += new_remote_credit;
send_remote_ctrl(
path, make_message(sys_atom::value, ok_atom::value, new_remote_credit));
}
}
void stream_multiplexer::fail(error reason, strong_actor_ptr predecessor,
strong_actor_ptr successor) {
CAF_ASSERT(current_stream_msg_ != nullptr);
if (predecessor)
unsafe_send_as(self_, predecessor,
make<stream_msg::abort>(current_stream_msg_->sid, reason));
if (successor)
unsafe_send_as(self_, successor,
make<stream_msg::abort>(current_stream_msg_->sid, reason));
auto rp = self_->make_response_promise();
if (!rp.async())
rp.deliver(std::move(reason));
}
void stream_multiplexer::fail(error reason) {
CAF_ASSERT(current_stream_msg_ != nullptr);
auto i = streams_.find(current_stream_msg_->sid);
if (i != streams_.end()) {
fail(std::move(reason), std::move(i->second.prev_stage),
std::move(i->second.next_stage));
streams_.erase(i);
} else {
fail(std::move(reason), nullptr, nullptr);
}
}
void stream_multiplexer::send_local(strong_actor_ptr& dest, stream_msg&& x,
std::vector<strong_actor_ptr> stages,
message_id mid) {
CAF_ASSERT(dest != nullptr);
dest->enqueue(make_mailbox_element(self_->ctrl(), mid, std::move(stages),
std::move(x)),
self_->context());
}
} // namespace detail
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* 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 <set>
#include <map>
#include <string>
#include <numeric>
#include <fstream>
#include <iostream>
#include <iterator>
#include <unordered_set>
#define CAF_SUITE manual_stream_management
#include "caf/test/dsl.hpp"
#include "caf/upstream_path.hpp"
#include "caf/stream_source.hpp"
#include "caf/filtering_downstream.hpp"
#include "caf/policy/pull5.hpp"
#include "caf/policy/push5.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/basp_broker.hpp"
#include "caf/io/network/test_multiplexer.hpp"
using std::cout;
using std::endl;
using std::string;
using namespace caf;
// -- Type aliases -------------------------------------------------------------
using peer_atom = atom_constant<atom("peer")>;
using key_type = string;
using value_type = int;
using filter_type = std::vector<key_type>;
using element_type = std::pair<key_type, value_type>;
using stream_type = stream<element_type>;
// -- Convenience functions ----------------------------------------------------
/// Returns `true` if `x` is selected by `f`, `false` otherwise.
bool selected(const filter_type& f, const element_type& x) {
using std::get;
for (auto& key : f)
if (key == get<0>(x))
return true;
return false;
}
// -- Class definitions --------------------------------------------------------
struct core_state;
/// A stream governor dispatches incoming data from all publishers to local
/// subscribers as well as peers. Its primary job is to avoid routing loops by
/// not forwarding data from a peer back to itself.
class stream_governor : public stream_handler {
public:
// -- Nested types -----------------------------------------------------------
struct peer_data {
filter_type filter;
policy::push5<element_type> out;
stream_id incoming_sid;
peer_data(filter_type y, local_actor* self, const stream_id& sid)
: filter(std::move(y)),
out(self, sid) {
// nop
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, peer_data& x) {
return f(x.filter, x.out, x.incoming_sid);
}
};
using peer_data_ptr = std::unique_ptr<peer_data>;
using peer_map = std::unordered_map<strong_actor_ptr, peer_data_ptr>;
using local_downstream = filtering_downstream<element_type, key_type>;
// -- Constructors and destructors -------------------------------------------
stream_governor(core_state* state);
// -- Accessors --------------------------------------------------------------
inline const peer_map& peers() const {
return peers_;
}
inline bool has_peer(const strong_actor_ptr& hdl) const {
return peers_.count(hdl) > 0;
}
inline local_downstream& local_subscribers() {
return local_subscribers_;
}
// -- Mutators ---------------------------------------------------------------
template <class... Ts>
void new_stream(const strong_actor_ptr& hdl, const stream_id& sid,
std::tuple<Ts...> xs) {
CAF_ASSERT(hdl != nullptr);
stream_type token{sid};
auto ys = std::tuple_cat(std::make_tuple(token), std::move(xs));
new_stream(hdl, token, make_message_from_tuple(std::move(ys)));
}
peer_data* add_peer(strong_actor_ptr ptr, filter_type filter);
// -- Overridden member functions of `stream_handler` ------------------------
error add_downstream(strong_actor_ptr& hdl) override;
void downstream_demand(downstream_path* path, long demand);
error confirm_downstream(const strong_actor_ptr& rebind_from,
strong_actor_ptr& hdl, long initial_demand,
bool redeployable) override;
error downstream_ack(strong_actor_ptr& hdl, int64_t, long demand) override;
error push() override;
expected<long> add_upstream(strong_actor_ptr& hdl, const stream_id& sid,
stream_priority prio) override;
error upstream_batch(strong_actor_ptr& hdl, int64_t, long,
message& xs) override;
error close_upstream(strong_actor_ptr& hdl) override;
void abort(strong_actor_ptr& cause, const error& reason) override;
bool done() const override;
message make_output_token(const stream_id&) const override;
long downstream_credit() const;
long downstream_buffer_size() const;
void assign_credit();
private:
void new_stream(const strong_actor_ptr& hdl, const stream_type& token,
message msg);
core_state* state_;
policy::pull5 in_;
local_downstream local_subscribers_;
peer_map peers_;
};
struct core_state {
/// Requested topics on this core.
filter_type filter;
/// Multiplexes local streams as well as stream for peers.
intrusive_ptr<stream_governor> governor;
/// List of all known publishers. Whenever we change the `filter` on a core,
/// we need to send the updated filter to all publishers.
std::vector<strong_actor_ptr> peers;
/// Stream ID used by the governor.
stream_id sid;
/// Set of pending handshake requests.
std::set<strong_actor_ptr> pending_peers;
/// Pointer to the owning actor.
event_based_actor* self;
/// Name of this actor type.
static const char* name;
void init(event_based_actor* s, filter_type initial_filter) {
self = s;
filter = std::move(initial_filter);
sid =
stream_id{self->ctrl(),
self->new_request_id(message_priority::normal).integer_value()};
governor = make_counted<stream_governor>(this);
self->streams().emplace(sid, governor);
}
strong_actor_ptr prev_peer_from_handshake() {
auto& xs = self->current_mailbox_element()->content();
strong_actor_ptr res;
if (xs.match_elements<stream_msg>()) {
auto& x = xs.get_as<stream_msg>(0);
if (holds_alternative<stream_msg::open>(x.content)) {
res = get<stream_msg::open>(x.content).prev_stage;
}
}
return res;
}
};
const char* core_state::name = "core";
// -- Implementation of `stream_governor` --------------------------------------
stream_governor::stream_governor(core_state* state)
: state_(state),
in_(state->self),
local_subscribers_(state->self, state->sid) {
// nop
}
stream_governor::peer_data* stream_governor::add_peer(strong_actor_ptr hdl,
filter_type filter) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(filter));
auto ptr = new peer_data{std::move(filter), state_->self, state_->sid};
ptr->out.add_path(hdl);
auto res = peers_.emplace(std::move(hdl), peer_data_ptr{ptr});
return res.second ? ptr : nullptr;
}
error stream_governor::add_downstream(strong_actor_ptr&) {
CAF_LOG_ERROR("add_downstream on governor called");
return sec::invalid_stream_state;
}
void stream_governor::downstream_demand(downstream_path* path, long demand) {
path->open_credit += demand;
push();
assign_credit();
}
error stream_governor::confirm_downstream(const strong_actor_ptr& rebind_from,
strong_actor_ptr& hdl,
long initial_demand,
bool redeployable) {
CAF_LOG_TRACE(CAF_ARG(rebind_from) << CAF_ARG(hdl)
<< CAF_ARG(initial_demand) << CAF_ARG(redeployable));
CAF_IGNORE_UNUSED(redeployable);
auto path = local_subscribers_.find(rebind_from);
if (path) {
if (!local_subscribers_.confirm_path(rebind_from, hdl, initial_demand)) {
CAF_LOG_ERROR("Cannot rebind to registered downstream.");
return sec::invalid_stream_state;
}
downstream_demand(path, initial_demand);
return none;
}
auto i = peers_.find(rebind_from);
if (i != peers_.end()) {
auto uptr = std::move(i->second);
peers_.erase(i);
auto res = peers_.emplace(hdl, std::move(uptr));
if (!res.second) {
CAF_LOG_ERROR("Cannot rebind to registered downstream.");
return sec::invalid_stream_state;
}
CAF_LOG_DEBUG("Confirmed path to another core"
<< CAF_ARG(rebind_from) << CAF_ARG(hdl));
res.first->second->out.confirm_path(rebind_from, hdl, initial_demand);
auto pp = res.first->second->out.find(hdl);
if (!pp) {
CAF_LOG_ERROR("Unable to find peer after confirming it");
return sec::invalid_downstream;
}
downstream_demand(pp, initial_demand);
return none;
}
CAF_LOG_ERROR("Cannot confirm path to unknown downstream.");
return sec::invalid_downstream;
}
error stream_governor::downstream_ack(strong_actor_ptr& hdl, int64_t,
long demand) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(demand));
auto path = local_subscribers_.find(hdl);
if (path) {
downstream_demand(path, demand);
return none;
}
auto i = peers_.find(hdl);
if (i != peers_.end()) {
auto pp = i->second->out.find(hdl);
if (!pp)
return sec::invalid_stream_state;
CAF_LOG_DEBUG("grant" << demand << "new credit to" << hdl);
downstream_demand(pp, demand);
return none;
}
return sec::invalid_downstream;
}
error stream_governor::push() {
CAF_LOG_TRACE("");
if (local_subscribers_.buf_size() > 0)
local_subscribers_.emit_batches();
for (auto& kvp : peers_) {
auto& out = kvp.second->out;
if (out.buf_size() > 0)
out.emit_batches();
}
return none;
}
expected<long> stream_governor::add_upstream(strong_actor_ptr& hdl,
const stream_id& sid,
stream_priority prio) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(sid) << CAF_ARG(prio));
if (hdl)
return in_.add_path(hdl, sid, prio, downstream_credit());
return sec::invalid_argument;
}
error stream_governor::upstream_batch(strong_actor_ptr& hdl, int64_t xs_id,
long xs_size, message& xs) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(xs_size) << CAF_ARG(xs));
using std::get;
// Sanity checking.
auto path = in_.find(hdl);
if (!path)
return sec::invalid_upstream;
if (xs_size > path->assigned_credit)
return sec::invalid_stream_state;
if (!xs.match_elements<std::vector<element_type>>())
return sec::unexpected_message;
// Unwrap `xs`.
auto& vec = xs.get_mutable_as<std::vector<element_type>>(0);
// Decrease credit assigned to `hdl` and get currently available downstream
// credit on all paths.
CAF_LOG_DEBUG(CAF_ARG(path->assigned_credit));
path->last_batch_id = xs_id;
path->assigned_credit -= xs_size;
// Forward data to all other peers.
for (auto& kvp : peers_)
if (kvp.first != hdl) {
auto& out = kvp.second->out;
for (const auto& x : vec)
if (selected(kvp.second->filter, x))
out.push(x);
if (out.buf_size() > 0) {
out.emit_batches();
}
}
// Move elements from `xs` to the buffer for local subscribers.
for (auto& x : vec)
local_subscribers_.push(std::move(x));
local_subscribers_.emit_batches();
// Grant new credit to upstream if possible.
assign_credit();
return none;
}
error stream_governor::close_upstream(strong_actor_ptr& hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
if (in_.remove_path(hdl))
return none;
return sec::invalid_upstream;
}
void stream_governor::abort(strong_actor_ptr& hdl, const error& reason) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(reason));
if (hdl != nullptr) {
if (local_subscribers_.remove_path(hdl))
return;
auto i = peers_.find(hdl);
if (i != peers_.end()) {
auto& pd = *i->second;
state_->self->streams().erase(pd.incoming_sid);
peers_.erase(i);
} else {
in_.abort(hdl, reason);
}
} else {
local_subscribers_.abort(hdl, reason);
in_.abort(hdl, reason);
for (auto& kvp : peers_)
kvp.second->out.abort(hdl, reason);
peers_.clear();
}
}
bool stream_governor::done() const {
return false;
}
message stream_governor::make_output_token(const stream_id& x) const {
return make_message(stream<element_type>{x});
}
long stream_governor::downstream_credit() const {
auto min_peer_credit = [&] {
return std::accumulate(peers_.begin(), peers_.end(),
std::numeric_limits<long>::max(),
[](long x, const peer_map::value_type& y) {
return std::min(x, y.second->out.min_credit());
});
};
constexpr long min_buffer_size = 5l;
if (local_subscribers_.num_paths() == 0)
return (peers_.empty() ? 0l : min_peer_credit()) + min_buffer_size;
return (peers_.empty()
? local_subscribers_.min_credit()
: std::min(local_subscribers_.min_credit(), min_peer_credit()))
+ min_buffer_size;
}
long stream_governor::downstream_buffer_size() const {
auto result = local_subscribers_.buf_size();
for (auto& kvp : peers_)
result += std::max(result, kvp.second->out.buf_size());
return result;
}
void stream_governor::assign_credit() {
CAF_LOG_TRACE("");
auto current_size = downstream_buffer_size();
auto desired_size = downstream_credit();
CAF_LOG_DEBUG(CAF_ARG(current_size) << CAF_ARG(desired_size));
if (current_size < desired_size)
in_.assign_credit(desired_size - current_size);
}
void stream_governor::new_stream(const strong_actor_ptr& hdl,
const stream_type& token, message msg) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(token) << CAF_ARG(msg));
CAF_ASSERT(hdl != nullptr);
auto self = state_->self;
hdl->enqueue(make_mailbox_element(self->ctrl(), message_id::make(), {},
make<stream_msg::open>(
token.id(), std::move(msg), self->ctrl(),
hdl, stream_priority::normal, false)),
self->context());
self->streams().emplace(token.id(), this);
}
// -- Implementation of core actor ---------------------------------------------
namespace {
behavior core(stateful_actor<core_state>* self, filter_type initial_filter) {
self->state.init(self, std::move(initial_filter));
return {
// -- Peering requests from local actors, i.e., "step 0". ------------------
[=](peer_atom, strong_actor_ptr remote_core) -> result<void> {
auto& st = self->state;
// Sanity checking.
if (remote_core == nullptr)
return sec::invalid_argument;
// Create necessary state and send message to remote core if not already
// peering with B.
if (!st.governor->has_peer(remote_core))
self->send(actor{self} * actor_cast<actor>(remote_core),
peer_atom::value, st.filter);
return unit;
},
// -- 3-way handshake for establishing peering streams between A and B. ----
// -- A (this node) performs steps #1 and #3. B performs #2 and #4. --------
// Step #1: A demands B shall establish a stream back to A. A has
// subscribers to the topics `ts`.
[=](peer_atom, filter_type& peer_ts) -> stream_type {
auto& st = self->state;
// Reject anonymous peering requests.
auto p = self->current_sender();
if (p == nullptr) {
CAF_LOG_DEBUG("Drop anonymous peering request.");
return invalid_stream;
}
// Ignore unexpected handshakes as well as handshakes that collide
// with an already pending handshake.
if (st.pending_peers.count(p) > 0) {
CAF_LOG_DEBUG("Drop repeated peering request.");
return invalid_stream;
}
auto peer_ptr = st.governor->add_peer(p, std::move(peer_ts));
if (peer_ptr == nullptr) {
CAF_LOG_DEBUG("Drop peering request of already known peer.");
return invalid_stream;
}
st.pending_peers.emplace(std::move(p));
auto& next = self->current_mailbox_element()->stages.back();
CAF_ASSERT(next != nullptr);
auto token = std::make_tuple(st.filter);
self->fwd_stream_handshake<element_type>(st.sid, token);
return {st.sid, st.governor};
},
// step #2: B establishes a stream to A, sending its own local subscriptions
[=](const stream_type& in, filter_type& filter) {
auto& st = self->state;
// Reject anonymous peering requests and unrequested handshakes.
auto p = st.prev_peer_from_handshake();
if (p == nullptr) {
CAF_LOG_DEBUG("Drop anonymous peering request.");
return;
}
// Ignore duplicates.
if (st.governor->has_peer(p)) {
CAF_LOG_DEBUG("Drop repeated handshake phase #2.");
return;
}
// Add state to actor.
auto peer_ptr = st.governor->add_peer(p, std::move(filter));
peer_ptr->incoming_sid = in.id();
self->streams().emplace(in.id(), st.governor);
// Start streaming in opposite direction.
st.governor->new_stream(p, st.sid, std::make_tuple(ok_atom::value));
},
// step #3: A establishes a stream to B
// (now B has a stream to A and vice versa)
[=](const stream_type& in, ok_atom) {
CAF_LOG_TRACE(CAF_ARG(in));
auto& st = self->state;
// Reject anonymous peering requests and unrequested handshakes.
auto p = st.prev_peer_from_handshake();
if (p == nullptr) {
CAF_LOG_DEBUG("Ignored anonymous peering request.");
return;
}
// Reject step #3 handshake if this actor didn't receive a step #1
// handshake previously.
auto i = st.pending_peers.find(p);
if (i == st.pending_peers.end()) {
CAF_LOG_WARNING("Received a step #3 handshake, but no #1 previously.");
return;
}
st.pending_peers.erase(i);
auto res = self->streams().emplace(in.id(), st.governor);
if (!res.second) {
CAF_LOG_WARNING("Stream already existed.");
}
},
// -- Communication to local actors: incoming streams and subscriptions. ---
[=](join_atom, filter_type& filter) -> expected<stream_type> {
auto& st = self->state;
auto& cs = self->current_sender();
if (cs == nullptr)
return sec::cannot_add_downstream;
auto& stages = self->current_mailbox_element()->stages;
if (stages.empty()) {
CAF_LOG_ERROR("Cannot join a data stream without downstream.");
auto rp = self->make_response_promise();
rp.deliver(sec::no_downstream_stages_defined);
return stream_type{stream_id{nullptr, 0}, nullptr};
}
auto next = stages.back();
CAF_ASSERT(next != nullptr);
std::tuple<> token;
self->fwd_stream_handshake<element_type>(st.sid, token);
st.governor->local_subscribers().add_path(cs);
st.governor->local_subscribers().set_filter(cs, std::move(filter));
return stream_type{st.sid, st.governor};
},
[=](const stream_type& in) {
auto& st = self->state;
auto& cs = self->current_sender();
if (cs == nullptr) {
return;
}
self->streams().emplace(in.id(), st.governor);
}
};
}
void driver(event_based_actor* self, const actor& sink) {
using buf_type = std::vector<element_type>;
self->new_stream(
// Destination.
sink,
// Initialize send buffer with 10 elements.
[](buf_type& xs) {
xs = buf_type{{"a", 0}, {"b", 0}, {"a", 1}, {"a", 2}, {"b", 1},
{"b", 2}, {"a", 3}, {"b", 3}, {"a", 4}, {"a", 5}};
},
// Get next element.
[](buf_type& xs, downstream<element_type>& out, size_t num) {
auto n = std::min(num, xs.size());
for (size_t i = 0u; i < n; ++i)
out.push(xs[i]);
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
},
// Did we reach the end?.
[](const buf_type& xs) {
return xs.empty();
},
// Handle result of the stream.
[](expected<void>) {
// nop
},
policy::arg<policy::push5<element_type>>::value
);
}
struct consumer_state {
std::vector<element_type> xs;
};
void consumer(stateful_actor<consumer_state>* self, filter_type ts,
const actor& src) {
self->send(self * src, join_atom::value, std::move(ts));
self->become(
[=](const stream_type& in) {
self->add_sink(
// Input stream.
in,
// Initialize state.
[](unit_t&) {
// nop
},
// Process single element.
[=](unit_t&, element_type x) {
self->state.xs.emplace_back(std::move(x));
},
// Cleanup.
[](unit_t&) {
// nop
},
policy::arg<policy::pull5>::value
);
},
[=](get_atom) {
return self->state.xs;
}
);
}
struct config : actor_system_config {
public:
config() {
add_message_type<element_type>("element");
logger_file_name = "streamlog";
}
};
using fixture = test_coordinator_fixture<config>;
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(manual_stream_management, fixture)
CAF_TEST(two_peers) {
// Spawn core actors.
auto core1 = sys.spawn(core, filter_type{"a", "b", "c"});
auto core2 = sys.spawn(core, filter_type{"a", "b", "c"});
sched.run();
// Connect a consumer (leaf) to core2.
auto leaf = sys.spawn(consumer, filter_type{"b"}, core2);
sched.run_once();
expect((atom_value, filter_type),
from(leaf).to(core2).with(join_atom::value, filter_type{"b"}));
expect((stream_msg::open), from(_).to(leaf).with(_, core2, _, _, false));
expect((stream_msg::ack_open), from(leaf).to(core2).with(_, 5, _, false));
// Initiate handshake between core1 and core2.
self->send(core1, peer_atom::value, actor_cast<strong_actor_ptr>(core2));
expect((peer_atom, strong_actor_ptr), from(self).to(core1).with(_, core2));
// Step #1: core1 ---> ('peer', filter_type) ---> core2
expect((peer_atom, filter_type),
from(core1).to(core2).with(_, filter_type{"a", "b", "c"}));
// Step #2: core1 <--- (stream_msg::open) <--- core2
expect((stream_msg::open),
from(_).to(core1).with(
std::make_tuple(_, filter_type{"a", "b", "c"}), core2, _, _,
false));
// Step #3: core1 ---> (stream_msg::open) ---> core2
// core1 ---> (stream_msg::ack_open) ---> core2
expect((stream_msg::open), from(_).to(core2).with(_, core1, _, _, false));
expect((stream_msg::ack_open), from(core1).to(core2).with(_, 5, _, false));
expect((stream_msg::ack_open), from(core2).to(core1).with(_, 5, _, false));
// There must be no communication pending at this point.
CAF_REQUIRE(!sched.has_job());
// Spin up driver on core1.
auto d1 = sys.spawn(driver, core1);
CAF_MESSAGE("d1: " << to_string(d1));
sched.run_once();
expect((stream_msg::open), from(_).to(core1).with(_, d1, _, _, false));
expect((stream_msg::ack_open), from(core1).to(d1).with(_, 5, _, false));
// Data flows from driver to core1 to core2 and finally to leaf.
using buf = std::vector<element_type>;
expect((stream_msg::batch),
from(d1).to(core1)
.with(5, buf{{"a", 0}, {"b", 0}, {"a", 1}, {"a", 2}, {"b", 1}}, 0));
expect((stream_msg::batch),
from(core1).to(core2)
.with(5, buf{{"a", 0}, {"b", 0}, {"a", 1}, {"a", 2}, {"b", 1}}, 0));
expect((stream_msg::batch),
from(core2).to(leaf)
.with(2, buf{{"b", 0}, {"b", 1}}, 0));
expect((stream_msg::ack_batch), from(core2).to(core1).with(5, 0));
expect((stream_msg::ack_batch), from(core1).to(d1).with(5, 0));
// Check log of the consumer.
self->send(leaf, get_atom::value);
sched.prioritize(leaf);
sched.run_once();
self->receive(
[](const buf& xs) {
buf expected{{"b", 0}, {"b", 1}};
CAF_REQUIRE_EQUAL(xs, expected);
}
);
// Shutdown.
CAF_MESSAGE("Shutdown core actors.");
anon_send_exit(core1, exit_reason::user_shutdown);
anon_send_exit(core2, exit_reason::user_shutdown);
sched.run();
}
CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* 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/config.hpp"
#define CAF_SUITE stream_multiplexing
#include "caf/test/dsl.hpp"
#include <string>
#include "caf/all.hpp"
#include "caf/raw_event_based_actor.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/detail/incoming_stream_multiplexer.hpp"
#include "caf/detail/outgoing_stream_multiplexer.hpp"
using namespace caf;
namespace {
struct dummy_basp_state {
static const char* name;
};
const char* dummy_basp_state::name = "dummy_basp";
behavior dummy_basp(stateful_actor<dummy_basp_state>*) {
return {
[=](forward_atom, strong_actor_ptr& src,
std::vector<strong_actor_ptr>& fwd_stack, strong_actor_ptr& dest,
message_id mid, message& msg) {
CAF_REQUIRE(dest != nullptr);
dest->enqueue(make_mailbox_element(std::move(src), mid,
std::move(fwd_stack), std::move(msg)),
nullptr);
}
};
}
struct streamer_state {
static const char* name;
};
const char* streamer_state::name = "streamer";
void streamer_impl(stateful_actor<streamer_state>* self, const actor& dest) {
using buf = std::deque<int>;
self->new_stream(
// destination of the stream
dest,
// initialize state
[](buf& xs) {
xs = buf{1, 2, 3, 4, 5, 6, 7, 8, 9};
},
// get next element
[](buf& xs, downstream<int>& out, size_t num) {
auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i)
out.push(xs[i]);
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
},
// check whether we reached the end
[](const buf& xs) {
return xs.empty();
},
// handle result of the stream
[](expected<int>) {
// nop
}
);
}
struct sum_up_state {
static const char* name;
};
const char* sum_up_state::name = "sum_up";
behavior sum_up_impl(stateful_actor<sum_up_state>* self) {
return {
[=](stream<int>& in) {
return self->add_sink(
// input stream
in,
// initialize state
[](int& x) {
x = 0;
},
// processing step
[](int& x, int y) {
x += y;
},
// cleanup and produce result message
[](int& x) -> int {
return x;
}
);
}
};
}
struct fixture;
class stream_serv_class : public raw_event_based_actor,
public detail::stream_multiplexer::backend {
public:
stream_serv_class(actor_config& cfg, actor basp, fixture& parent)
: raw_event_based_actor(cfg),
detail::stream_multiplexer::backend(std::move(basp)),
fixture_(parent),
incoming_(this, *this),
outgoing_(this, *this) {
// nop
}
behavior make_behavior() override {
return {
[=](stream_msg& x) -> delegated<message> {
// Dispatching depends on the direction of the message.
if (outgoing_.has_stream(x.sid)) {
outgoing_(x);
} else {
incoming_(x);
}
return {};
},
[=](sys_atom, stream_msg& x) -> delegated<message> {
// Stream message received from a proxy, always results in a new
// stream from a local actor to a remote node.
CAF_REQUIRE(holds_alternative<stream_msg::open>(x.content));
outgoing_(x);
return {};
},
[=](sys_atom, ok_atom, int32_t credit) {
CAF_ASSERT(current_mailbox_element() != nullptr);
auto cme = current_mailbox_element();
if (cme->sender != nullptr) {
auto& nid = cme->sender->node();
add_credit(nid, credit);
} else {
CAF_LOG_ERROR("Received credit from an anonmyous stream server.");
}
},
[=](exit_msg& x) {
quit(x.reason);
}
};
}
const char* name() const override {
return "stream_serv";
}
void on_exit() override {
CAF_CHECK_EQUAL(incoming_.num_streams(), 0u);
CAF_CHECK_EQUAL(outgoing_.num_streams(), 0u);
CAF_CHECK(streams().empty());
remotes().clear();
}
strong_actor_ptr remote_stream_serv(const node_id& nid) override;
private:
fixture& fixture_;
detail::incoming_stream_multiplexer incoming_;
detail::outgoing_stream_multiplexer outgoing_;
};
// Simulates a regular forwarding_actor_proxy by pushing a handle to the
// original to the forwarding stack and redirecting each message to the
// stream_serv.
class pseudo_proxy : public raw_event_based_actor {
public:
pseudo_proxy(actor_config& cfg, actor stream_serv, actor original)
: raw_event_based_actor(cfg),
stream_serv_(std::move(stream_serv)),
original_(std::move(original)){
// nop
}
void enqueue(mailbox_element_ptr x, execution_unit* context) override {
x->stages.emplace_back(actor_cast<strong_actor_ptr>(original_));
auto msg = x->move_content_to_message();
auto prefix = make_message(sys_atom::value);
stream_serv_->enqueue(make_mailbox_element(std::move(x->sender), x->mid,
std::move(x->stages),
prefix + msg),
context);
}
const char* name() const override {
return "pseudo_proxy";
}
private:
actor stream_serv_;
actor original_;
};
struct fixture : test_coordinator_fixture<> {
actor basp;
actor streamer;
actor sum_up;
actor sum_up_proxy;
actor stream_serv1;
actor stream_serv2;
fixture() {
basp = sys.spawn(dummy_basp);
sum_up = sys.spawn(sum_up_impl);
stream_serv1 = sys.spawn<stream_serv_class>(basp, *this);
stream_serv2 = sys.spawn<stream_serv_class>(basp, *this);
sum_up_proxy = sys.spawn<pseudo_proxy>(stream_serv1, sum_up);
CAF_MESSAGE("basp: " << to_string(basp));
CAF_MESSAGE("sum_up: " << to_string(sum_up));
CAF_MESSAGE("stream_serv1: " << to_string(stream_serv1));
CAF_MESSAGE("stream_serv2: " << to_string(stream_serv2));
CAF_MESSAGE("sum_up_proxy: " << to_string(sum_up_proxy));
sched.run();
}
~fixture() {
kill_em_all();
}
void kill_em_all() {
for (auto x : {basp, streamer, sum_up, stream_serv1, stream_serv2})
anon_send_exit(x, exit_reason::kill);
sched.run();
}
};
strong_actor_ptr stream_serv_class::remote_stream_serv(const node_id&) {
auto res = strong_actor_ptr{ctrl()} == fixture_.stream_serv1
? fixture_.stream_serv2
: fixture_.stream_serv1;
return actor_cast<strong_actor_ptr>(res);
// actor_cast<strong_actor_ptr>(fixture_.sum_up);
}
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(outgoing_stream_multiplexer_tests, fixture)
#define expect_network_traffic(source, destination) \
expect((atom_value, strong_actor_ptr, std::vector<strong_actor_ptr>, \
strong_actor_ptr, message_id, message), \
from(source).to(basp).with(forward_atom::value, source, _, \
destination, message_id::make(), _));
// Our first setup simply checks whether outgoing_stream_multiplexer intercepts
// stream handshakes. For this to happen, the forwarding actor proxy
// sum_up_proxy needs to re-write the initial `stream_msg::open`. It pushes
// "itself" (the actor it represents) onto the forwarding stack and redirects
// the handshake to the stream_serv (self). outgoing_stream_multiplexer then
// sends an ACK to the previous stage and a new OPEN to the remote stream_serv
// (which is missing in this simple setup).
CAF_TEST(stream_interception) {
streamer = sys.spawn(streamer_impl, sum_up_proxy);
CAF_MESSAGE("streamer: " << to_string(streamer));
sched.run_once();
// streamer --('sys' stream_msg::open)--> stream_serv1
expect((atom_value, stream_msg),
from(streamer).to(stream_serv1).with(sys_atom::value, _));
// streamer [via stream_serv1 / BASP] --(stream_msg::open)--> stream_serv2
expect_network_traffic(streamer, stream_serv2);
expect((stream_msg::open),
from(_).to(stream_serv2)
.with(_, stream_serv1, sum_up_proxy, _, _, false));
// stream_serv2 [via BASP] --('sys', 'ok', 5)--> stream_serv1
expect_network_traffic(stream_serv2, stream_serv1);
expect((atom_value, atom_value, int32_t),
from(stream_serv2).to(stream_serv1)
.with(sys_atom::value, ok_atom::value, 5));
// stream_serv2 --(stream_msg::open)--> sum_up
expect((stream_msg::open),
from(_).to(sum_up).with(_, stream_serv2, sum_up_proxy, _, _, false));
// sum_up --(stream_msg::ack_open)--> stream_serv2
expect((stream_msg::ack_open),
from(sum_up).to(stream_serv2).with(sum_up_proxy, 5, _, false));
// stream_serv2 [via BASP] --(stream_msg::ack_open)--> stream_serv1
expect_network_traffic(stream_serv2, stream_serv1);
expect((stream_msg::ack_open),
from(stream_serv2).to(stream_serv1).with(_, 5, _, false));
// stream_serv1 --('sys', 'ok', 5)--> stream_serv2
expect_network_traffic(stream_serv1, stream_serv2);
expect((atom_value, atom_value, int32_t),
from(stream_serv1).to(stream_serv2)
.with(sys_atom::value, ok_atom::value, 5));
// stream_serv1 --(stream_msg::ack_open)--> streamer
expect((stream_msg::ack_open),
from(stream_serv1).to(streamer).with(_, 5, _, false));
// streamer --(stream_msg::batch)--> stream_serv1
expect((stream_msg::batch), from(streamer).to(stream_serv1).with(5, _, 0));
// stream_serv1 [via BASP] --(stream_msg::batch)--> stream_serv2
expect_network_traffic(stream_serv1, stream_serv2);
expect((stream_msg::batch),
from(stream_serv1).to(stream_serv2).with(5, _, 0));
// stream_serv1 --(stream_msg::batch)--> sum_up
expect((stream_msg::batch),
from(stream_serv2).to(sum_up).with(5, _, 0));
// sum_up --(stream_msg::ack_batch)--> stream_serv1
expect((stream_msg::ack_batch),
from(sum_up).to(stream_serv2).with(5, 0));
// stream_serv2 [via BASP] --(stream_msg::ack_batch)--> stream_serv1
expect_network_traffic(stream_serv2, stream_serv1);
expect((stream_msg::ack_batch),
from(stream_serv2).to(stream_serv1).with(5, 0));
// stream_serv1 --(stream_msg::ack_batch)--> streamer
expect((stream_msg::ack_batch),
from(stream_serv1).to(streamer).with(5, 0));
// streamer --(stream_msg::batch)--> stream_serv1
expect((stream_msg::batch), from(streamer).to(stream_serv1).with(4, _, 1));
// stream_serv1 [via BASP] --(stream_msg::batch)--> stream_serv2
expect_network_traffic(stream_serv1, stream_serv2);
expect((stream_msg::batch),
from(stream_serv1).to(stream_serv2).with(4, _, 1));
// stream_serv1 --(stream_msg::batch)--> sum_up
expect((stream_msg::batch),
from(stream_serv2).to(sum_up).with(4, _, 1));
// sum_up --(stream_msg::ack_batch)--> stream_serv1
expect((stream_msg::ack_batch),
from(sum_up).to(stream_serv2).with(4, 1));
// stream_serv2 [via BASP] --(stream_msg::ack_batch)--> stream_serv1
expect_network_traffic(stream_serv2, stream_serv1);
expect((stream_msg::ack_batch),
from(stream_serv2).to(stream_serv1).with(4, 1));
// stream_serv1 --(stream_msg::ack_batch)--> streamer
expect((stream_msg::ack_batch),
from(stream_serv1).to(streamer).with(4, 1));
// streamer --(stream_msg::close)--> stream_serv1
expect((stream_msg::close), from(streamer).to(stream_serv1).with());
// stream_serv1 [via BASP] --(stream_msg::close)--> stream_serv2
expect_network_traffic(stream_serv1, stream_serv2);
expect((stream_msg::close), from(stream_serv1).to(stream_serv2).with());
// stream_serv1 --(stream_msg::close)--> sum_up
expect((stream_msg::close), from(stream_serv2).to(sum_up).with());
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -88,8 +88,7 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
auto mm = &system().middleman();
actor_config cfg;
auto res = make_actor<forwarding_actor_proxy, strong_actor_ptr>(
aid, nid, &(self->home_system()), cfg, self,
actor_cast<actor>(self->home_system().stream_serv()));
aid, nid, &(self->home_system()), cfg, self);
strong_actor_ptr selfptr{self->ctrl()};
res->get()->attach_functor([=](const error& rsn) {
mm->backend().post([=] {
......
......@@ -56,8 +56,6 @@
#include "caf/detail/safe_equal.hpp"
#include "caf/detail/get_root_uuid.hpp"
#include "caf/detail/get_mac_addresses.hpp"
#include "caf/detail/incoming_stream_multiplexer.hpp"
#include "caf/detail/outgoing_stream_multiplexer.hpp"
#ifdef CAF_USE_ASIO
#include "caf/io/network/asio_multiplexer.hpp"
......@@ -273,106 +271,10 @@ void middleman::start() {
}};
backend().thread_id(thread_.get_id());
}
// Default implementation of the stream server.
class stream_serv : public raw_event_based_actor,
public detail::stream_multiplexer::backend {
public:
stream_serv(actor_config& cfg, actor basp_ref)
: raw_event_based_actor(cfg),
detail::stream_multiplexer::backend(std::move(basp_ref)),
incoming_(this, *this),
outgoing_(this, *this) {
// nop
}
const char* name() const override {
return "stream_serv";
}
behavior make_behavior() override {
return {
[=](stream_msg& x) -> delegated<message> {
CAF_LOG_TRACE(CAF_ARG(x));
// Dispatching depends on the direction of the message.
if (outgoing_.has_stream(x.sid)) {
outgoing_(x);
} else {
incoming_(x);
}
return {};
},
[=](sys_atom, stream_msg& x) -> delegated<message> {
CAF_LOG_TRACE(CAF_ARG(x));
// Stream message received from a proxy
outgoing_(x);
return {};
},
[=](sys_atom, ok_atom, int32_t credit) {
CAF_LOG_TRACE(CAF_ARG(credit));
CAF_ASSERT(current_mailbox_element() != nullptr);
auto cme = current_mailbox_element();
if (cme->sender != nullptr) {
auto& nid = cme->sender->node();
add_credit(nid, credit);
} else {
CAF_LOG_ERROR("Received credit from an anonmyous stream server.");
}
},
[=](exit_msg& x) {
CAF_LOG_TRACE(CAF_ARG(x));
if (x.reason)
quit(x.reason);
},
// Connects both incoming_ and outgoing_ to nid.
[=](connect_atom, const node_id& nid) {
CAF_LOG_TRACE(CAF_ARG(nid));
send(basp_, forward_atom::value, nid, atom("ConfigServ"),
make_message(get_atom::value, atom("StreamServ")));
},
// Assumes `ptr` is a remote spawn server.
[=](strong_actor_ptr& ptr) {
CAF_LOG_TRACE(CAF_ARG(ptr));
if (ptr) {
add_remote_path(ptr->node(), ptr);
}
}
};
}
strong_actor_ptr remote_stream_serv(const node_id& nid) override {
CAF_LOG_TRACE(CAF_ARG(nid));
strong_actor_ptr result;
// Ask remote config server for a handle to the remote spawn server.
scoped_actor self{system()};
self->send(basp_, forward_atom::value, nid, atom("ConfigServ"),
make_message(get_atom::value, atom("StreamServ")));
// Time out after 5 minutes.
self->receive(
[&](strong_actor_ptr& addr) {
result = std::move(addr);
},
after(std::chrono::minutes(5)) >> [] {
CAF_LOG_INFO("Accessing a remote spawn server timed out.");
}
);
return result;
}
void on_exit() override {
// Make sure to not keep references to remotes after shutdown.
remotes().clear();
}
private:
detail::incoming_stream_multiplexer incoming_;
detail::outgoing_stream_multiplexer outgoing_;
};
// Spawn utility actors.
auto basp = named_broker<basp_broker>(atom("BASP"));
manager_ = make_middleman_actor(system(), basp);
auto hdl = actor_cast<actor>(basp);
auto ssi = system().spawn<stream_serv, lazy_init + hidden>(std::move(hdl));
system().stream_serv(actor_cast<strong_actor_ptr>(std::move(ssi)));
}
void middleman::stop() {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* 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 <string>
#include <numeric>
#include <fstream>
#include <iostream>
#include <iterator>
#define CAF_SUITE io_remote_streams
#include "caf/test/io_dsl.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/basp_broker.hpp"
#include "caf/io/network/test_multiplexer.hpp"
using std::cout;
using std::endl;
using std::string;
using namespace caf;
using namespace caf::io;
namespace {
class remoting_config : public actor_system_config {
public:
remoting_config() {
load<middleman, network::test_multiplexer>();
add_message_type<stream<int>>("stream<int>");
add_message_type<std::vector<int>>("vector<int>");
middleman_detach_utility_actors = false;
}
};
using sub_fixture = test_node_fixture_t<remoting_config>;
} // namespace <anonymous>
struct dsl_path_info {
sub_fixture& host;
actor receiver;
dsl_path_info(sub_fixture& x, actor y)
: host(x), receiver(std::move(y)) {
// nop
}
dsl_path_info(sub_fixture& x, strong_actor_ptr y)
: host(x), receiver(actor_cast<actor>(std::move(y))) {
// nop
}
};
#define expect_on_path(types, fields, ...) \
CAF_MESSAGE(">>> " << #types << " on path " << #__VA_ARGS__); \
{ \
std::vector<dsl_path_info> path{{__VA_ARGS__}}; \
for (auto x : path) { \
network_traffic(); \
expect_on(x.host, types, from(_).to(x.receiver).fields); \
} \
} \
CAF_MESSAGE("<<< path done")
CAF_TEST_FIXTURE_SCOPE(netstreams, point_to_point_fixture_t<remoting_config>)
CAF_TEST(stream_crossing_the_wire) {
// TODO: stream servers are currently disabled, because they break many
// possible setups by hiding remote actor handles. They must be
// re-implemented in a transparent fashion.
CAF_CHECK(true);
}
/*
CAF_TEST(stream_crossing_the_wire) {
CAF_MESSAGE("earth stream serv: " << to_string(earth.stream_serv));
CAF_MESSAGE("mars stream serv: " << to_string(mars.stream_serv));
// Connect the buffers of mars and earth to setup a pseudo-network.
mars.peer = &earth;
earth.peer = &mars;
// Setup: earth (streamer_without_result) streams to mars (drop_all).
CAF_MESSAGE("spawn drop_all sink on mars");
auto sink = mars.sys.spawn(drop_all);
// This test uses two fixtures for keeping state.
earth.conn = connection_handle::from_int(1);
mars.conn = connection_handle::from_int(2);
mars.acc = accept_handle::from_int(3);
// Run any initialization code.
exec_all();
// Prepare publish and remote_actor calls.
CAF_MESSAGE("prepare connections on earth and mars");
prepare_connection(mars, earth, "mars", 8080u);
// Publish sink on mars.
CAF_MESSAGE("publish sink on mars");
mars.publish(sink, 8080);
// Get a proxy on earth.
CAF_MESSAGE("connect from earth to mars");
auto proxy = earth.remote_actor("mars", 8080);
CAF_MESSAGE("got proxy: " << to_string(proxy) << ", spawn streamer on earth");
CAF_MESSAGE("establish remote stream paths");
// Establish remote paths between the two stream servers. This step is
// necessary to prevent a deadlock in `stream_serv::remote_stream_serv` due
// to blocking communication with the BASP broker.
anon_send(actor_cast<actor>(earth.stream_serv), connect_atom::value,
mars.stream_serv->node());
anon_send(actor_cast<actor>(mars.stream_serv), connect_atom::value,
earth.stream_serv->node());
exec_all();
CAF_MESSAGE("start streaming");
// Start streaming.
auto source = earth.sys.spawn(streamer_without_result, proxy);
earth.sched.run_once();
// source ----('sys', stream_msg::open)----> earth.stream_serv
expect_on(earth, (atom_value, stream_msg),
from(source).to(earth.stream_serv).with(sys_atom::value, _));
// --------------(stream_msg::open)-------------->
// earth.stream_serv -> mars.stream_serv -> sink
expect_on_path(
(stream_msg::open), with(_, _, _, _, _, false),
{mars, mars.stream_serv}, {mars, sink});
// mars.stream_serv --('sys', 'ok', 5)--> earth.stream_serv
network_traffic();
expect_on(earth, (atom_value, atom_value, int32_t),
from(_).to(earth.stream_serv)
.with(sys_atom::value, ok_atom::value, 5));
// -----------------(stream_msg::ack_open)------------------>
// sink -> mars.stream_serv -> earth.stream_serv -> source
expect_on_path(
(stream_msg::ack_open), with(_, 5, _, false),
{mars, mars.stream_serv}, {earth, earth.stream_serv}, {earth, source});
// earth.stream_serv --('sys', 'ok', 5)--> mars.stream_serv
network_traffic();
expect_on(mars, (atom_value, atom_value, int32_t),
from(_).to(mars.stream_serv)
.with(sys_atom::value, ok_atom::value, 5));
// -------------------(stream_msg::batch)------------------->
// source -> earth.stream_serv -> mars.stream_serv -> sink
expect_on_path(
(stream_msg::batch), with(5, std::vector<int>{1, 2, 3, 4, 5}, 0),
{earth, earth.stream_serv},
{mars, mars.stream_serv}, {mars, sink});
// -----------------(stream_msg::ack_batch)------------------>
// sink -> mars.stream_serv -> earth.stream_serv -> source
expect_on_path(
(stream_msg::ack_batch), with(5, 0),
{mars, mars.stream_serv}, {earth, earth.stream_serv}, {earth, source});
// -------------------(stream_msg::batch)------------------->
// source -> earth.stream_serv -> mars.stream_serv -> sink
expect_on_path(
(stream_msg::batch), with(4, std::vector<int>{6, 7, 8, 9}, 1),
{earth, earth.stream_serv},
{mars, mars.stream_serv}, {mars, sink});
// -----------------(stream_msg::ack_batch)------------------>
// sink -> mars.stream_serv -> earth.stream_serv -> source
expect_on_path(
(stream_msg::ack_batch), with(4, 1),
{mars, mars.stream_serv}, {earth, earth.stream_serv}, {earth, source});
// -------------------(stream_msg::close)------------------->
// source -> earth.stream_serv -> mars.stream_serv -> sink
expect_on_path(
(stream_msg::close), with(),
{earth, earth.stream_serv},
{mars, mars.stream_serv}, {mars, sink});
// sink ----(result: <empty>)---> source
network_traffic();
expect_on(earth, (void), from(proxy).to(source).with());
// sink --(void)--> source
anon_send_exit(sink, exit_reason::user_shutdown);
mars.sched.run();
anon_send_exit(source, exit_reason::user_shutdown);
earth.sched.run();
}
*/
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -36,13 +36,11 @@ public:
caf::io::connection_handle conn;
caf::io::accept_handle acc;
test_node_fixture* peer = nullptr;
caf::strong_actor_ptr stream_serv;
test_node_fixture()
: mm(this->sys.middleman()),
mpx(dynamic_cast<caf::io::network::test_multiplexer&>(mm.backend())),
basp(get_basp_broker()),
stream_serv(this->sys.stream_serv()) {
basp(get_basp_broker()) {
// nop
}
......
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