Commit 70289e0d authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Revise stream messages

Replace stream_id with stream_slot and separate stream messages into
three types: stream_handshake_msg, downstream_msg, and upstream_msg.

This patch breaks streaming in an ongoing effort to simplify the design
and to integrate adaptive credit management.
parent a0653c14
...@@ -98,9 +98,7 @@ set (LIBCAF_CORE_SRCS ...@@ -98,9 +98,7 @@ set (LIBCAF_CORE_SRCS
src/stream_gatherer.cpp src/stream_gatherer.cpp
src/stream_gatherer.cpp src/stream_gatherer.cpp
src/stream_gatherer_impl.cpp src/stream_gatherer_impl.cpp
src/stream_id.cpp
src/stream_manager.cpp src/stream_manager.cpp
src/stream_msg_visitor.cpp
src/stream_priority.cpp src/stream_priority.cpp
src/stream_scatterer.cpp src/stream_scatterer.cpp
src/stream_scatterer_impl.cpp src/stream_scatterer_impl.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_DOWNSTREAM_MSG_HPP
#define CAF_DOWNSTREAM_MSG_HPP
#include <utility>
#include <vector>
#include <cstdint>
#include "caf/actor_addr.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/atom.hpp"
#include "caf/message.hpp"
#include "caf/stream_priority.hpp"
#include "caf/stream_slot.hpp"
#include "caf/variant.hpp"
#include "caf/tag/boxing_type.hpp"
#include "caf/detail/type_list.hpp"
namespace caf {
/// Stream messages that travel downstream, i.e., batches and close messages.
struct downstream_msg : tag::boxing_type {
// -- nested types -----------------------------------------------------------
/// Transmits stream data.
struct batch {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
/// Size of the type-erased vector<T> (used credit).
int32_t xs_size;
/// A type-erased vector<T> containing the elements of the batch.
message xs;
/// ID of this batch (ascending numbering).
int64_t id;
};
/// Orderly shuts down a stream after receiving an ACK for the last batch.
struct close {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
};
/// Propagates a fatal error from sources to sinks.
struct forced_close {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = downstream_msg;
/// Reason for shutting down the stream.
error reason;
};
// -- member types -----------------------------------------------------------
/// Lists all possible options for the payload.
using alternatives = detail::type_list<batch, close, forced_close>;
/// Stores one of `alternatives`.
using content_type = variant<batch, close, forced_close>;
// -- constructors, destructors, and assignment operators --------------------
template <class T>
downstream_msg(stream_slot id, actor_addr addr, T&& x)
: slot(id),
sender(std::move(addr)),
content(std::forward<T>(x)) {
// nop
}
downstream_msg() = default;
downstream_msg(downstream_msg&&) = default;
downstream_msg(const downstream_msg&) = default;
downstream_msg& operator=(downstream_msg&&) = default;
downstream_msg& operator=(const downstream_msg&) = default;
// -- member variables -------------------------------------------------------
/// ID of the affected stream.
stream_slot slot;
/// Address of the sender. Identifies the up- or downstream actor sending
/// this message. Note that abort messages can get send after `sender`
/// already terminated. Hence, `current_sender()` can be `nullptr`, because
/// no strong pointers can be formed any more and receiver would receive an
/// anonymous message.
actor_addr sender;
/// Palyoad of the message.
content_type content;
};
/// Allows the testing DSL to unbox `stream_msg` automagically.
template <class T>
const T& get(const downstream_msg& x) {
return get<T>(x.content);
}
/// Allows the testing DSL to check whether `stream_msg` holds a `T`.
template <class T>
bool is(const downstream_msg& x) {
return holds_alternative<T>(x.content);
}
/// @relates downstream_msg
template <class T, class... Ts>
detail::enable_if_tt<detail::tl_contains<downstream_msg::alternatives, T>,
downstream_msg>
make(stream_slot slot, actor_addr addr, Ts&&... xs) {
return {slot, std::move(addr), T{std::forward<Ts>(xs)...}};
}
/// @relates downstream_msg::batch
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f,
downstream_msg::batch& x) {
return f(meta::type_name("batch"), meta::omittable(), x.xs_size, x.xs, x.id);
}
/// @relates downstream_msg::close
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, downstream_msg::close&) {
return f(meta::type_name("close"));
}
/// @relates downstream_msg::forced_close
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f,
downstream_msg::forced_close& x) {
return f(meta::type_name("forced_close"), x.reason);
}
/// @relates downstream_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, downstream_msg& x) {
return f(meta::type_name("stream_msg"), x.slot, x.sender, x.content);
}
} // namespace caf
#endif // CAF_DOWNSTREAM_MSG_HPP
...@@ -92,44 +92,44 @@ public: ...@@ -92,44 +92,44 @@ public:
detail::apply_args(f, indices, substreams_); detail::apply_args(f, indices, substreams_);
} }
path_ptr add_path(const stream_id& sid, strong_actor_ptr origin, path_ptr add_path(stream_slot slot, strong_actor_ptr origin,
strong_actor_ptr sink_ptr, strong_actor_ptr sink_ptr,
mailbox_element::forwarding_stack stages, mailbox_element::forwarding_stack stages,
message_id handshake_mid, message handshake_data, message_id handshake_mid, message handshake_data,
stream_priority prio, bool redeployable) override { stream_priority prio, bool redeployable) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(origin) << CAF_ARG(sink_ptr) CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(origin) << CAF_ARG(sink_ptr)
<< CAF_ARG(stages) << CAF_ARG(handshake_mid) << CAF_ARG(stages) << CAF_ARG(handshake_mid)
<< CAF_ARG(handshake_data) << CAF_ARG(prio) << CAF_ARG(handshake_data) << CAF_ARG(prio)
<< CAF_ARG(redeployable)); << CAF_ARG(redeployable));
auto ptr = substream_by_handshake_type(handshake_data); auto ptr = substream_by_handshake_type(handshake_data);
if (!ptr) if (!ptr)
return nullptr; return nullptr;
return ptr->add_path(sid, std::move(origin), std::move(sink_ptr), return ptr->add_path(slot, std::move(origin), std::move(sink_ptr),
std::move(stages), handshake_mid, std::move(stages), handshake_mid,
std::move(handshake_data), prio, redeployable); std::move(handshake_data), prio, redeployable);
} }
path_ptr confirm_path(const stream_id& sid, const actor_addr& from, path_ptr confirm_path(stream_slot slot, const actor_addr& from,
strong_actor_ptr to, long initial_demand, strong_actor_ptr to, long initial_demand,
long desired_batch_size, bool redeployable) override { long desired_batch_size, bool redeployable) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(from) << CAF_ARG(to) CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(from) << CAF_ARG(to)
<< CAF_ARG(initial_demand) << CAF_ARG(redeployable)); << CAF_ARG(initial_demand) << CAF_ARG(redeployable));
return first_hit([&](pointer ptr) -> outbound_path* { return first_hit([&](pointer ptr) -> outbound_path* {
// Note: we cannot blindly try `confirm_path` on each scatterer, because // Note: we cannot blindly try `confirm_path` on each scatterer, because
// this will trigger forced_close messages. // this will trigger forced_close messages.
if (ptr->find(sid, from) == nullptr) if (ptr->find(slot, from) == nullptr)
return nullptr; return nullptr;
return ptr->confirm_path(sid, from, to, initial_demand, return ptr->confirm_path(slot, from, to, initial_demand,
desired_batch_size, redeployable); desired_batch_size, redeployable);
}); });
} }
bool remove_path(const stream_id& sid, const actor_addr& addr, error reason, bool remove_path(stream_slot slot, const actor_addr& addr, error reason,
bool silent) override { bool silent) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(addr) << CAF_ARG(reason) CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(addr) << CAF_ARG(reason)
<< CAF_ARG(silent)); << CAF_ARG(silent));
return std::any_of(begin(), end(), [&](pointer x) { return std::any_of(begin(), end(), [&](pointer x) {
return x->remove_path(sid, addr, reason, silent); return x->remove_path(slot, addr, reason, silent);
}); });
} }
...@@ -177,8 +177,8 @@ public: ...@@ -177,8 +177,8 @@ public:
ptr->emit_batches(); ptr->emit_batches();
} }
path_ptr find(const stream_id& sid, const actor_addr& x) override{ path_ptr find(stream_slot slot, const actor_addr& x) override{
return first_hit([&](const_pointer ptr) { return ptr->find(sid, x); }); return first_hit([&](const_pointer ptr) { return ptr->find(slot, x); });
} }
path_ptr path_at(size_t idx) override { path_ptr path_at(size_t idx) override {
......
...@@ -104,7 +104,6 @@ class response_promise; ...@@ -104,7 +104,6 @@ class response_promise;
class event_based_actor; class event_based_actor;
class type_erased_tuple; class type_erased_tuple;
class type_erased_value; class type_erased_value;
class stream_msg_visitor;
class actor_control_block; class actor_control_block;
class actor_system_config; class actor_system_config;
class uniform_type_info_map; class uniform_type_info_map;
...@@ -115,9 +114,10 @@ class forwarding_actor_proxy; ...@@ -115,9 +114,10 @@ class forwarding_actor_proxy;
struct unit_t; struct unit_t;
struct exit_msg; struct exit_msg;
struct down_msg; struct down_msg;
struct stream_msg;
struct timeout_msg; struct timeout_msg;
struct upstream_msg;
struct group_down_msg; struct group_down_msg;
struct downstream_msg;
struct invalid_actor_t; struct invalid_actor_t;
struct invalid_actor_addr_t; struct invalid_actor_addr_t;
struct illegal_message_element; struct illegal_message_element;
......
...@@ -22,11 +22,11 @@ ...@@ -22,11 +22,11 @@
#include <cstddef> #include <cstddef>
#include <cstdint> #include <cstdint>
#include "caf/stream_id.hpp" #include "caf/actor_control_block.hpp"
#include "caf/stream_msg.hpp"
#include "caf/stream_aborter.hpp" #include "caf/stream_aborter.hpp"
#include "caf/stream_priority.hpp" #include "caf/stream_priority.hpp"
#include "caf/actor_control_block.hpp" #include "caf/stream_slot.hpp"
#include "caf/upstream_msg.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
...@@ -39,16 +39,16 @@ public: ...@@ -39,16 +39,16 @@ public:
static constexpr const auto aborter_type = stream_aborter::source_aborter; static constexpr const auto aborter_type = stream_aborter::source_aborter;
/// Message type for propagating graceful shutdowns. /// Message type for propagating graceful shutdowns.
using regular_shutdown = stream_msg::drop; using regular_shutdown = upstream_msg::drop;
/// Message type for propagating errors. /// Message type for propagating errors.
using irregular_shutdown = stream_msg::forced_drop; using irregular_shutdown = upstream_msg::forced_drop;
/// Pointer to the parent actor. /// Pointer to the parent actor.
local_actor* self; local_actor* self;
/// Stream ID used on this source. /// Slot ID for the source.
stream_id sid; stream_slot slot;
/// Handle to the source. /// Handle to the source.
strong_actor_ptr hdl; strong_actor_ptr hdl;
...@@ -77,7 +77,7 @@ public: ...@@ -77,7 +77,7 @@ public:
error shutdown_reason; error shutdown_reason;
/// Constructs a path for given handle and stream ID. /// Constructs a path for given handle and stream ID.
inbound_path(local_actor* selfptr, const stream_id& id, strong_actor_ptr ptr); inbound_path(local_actor* selfptr, stream_slot id, strong_actor_ptr ptr);
~inbound_path(); ~inbound_path();
...@@ -91,14 +91,15 @@ public: ...@@ -91,14 +91,15 @@ public:
void emit_ack_batch(long new_demand); void emit_ack_batch(long new_demand);
static void emit_irregular_shutdown(local_actor* self, const stream_id& sid, static void emit_irregular_shutdown(local_actor* self, stream_slot sid,
const strong_actor_ptr& hdl, const strong_actor_ptr& hdl,
error reason); error reason);
}; };
/// @relates inbound_path
template <class Inspector> template <class Inspector>
typename Inspector::return_type inspect(Inspector& f, inbound_path& x) { typename Inspector::return_type inspect(Inspector& f, inbound_path& x) {
return f(meta::type_name("inbound_path"), x.hdl, x.sid, x.prio, return f(meta::type_name("inbound_path"), x.hdl, x.slot, x.prio,
x.last_acked_batch_id, x.last_batch_id, x.assigned_credit); x.last_acked_batch_id, x.last_batch_id, x.assigned_credit);
} }
......
...@@ -30,12 +30,12 @@ public: ...@@ -30,12 +30,12 @@ public:
~invalid_stream_gatherer() override; ~invalid_stream_gatherer() override;
path_ptr add_path(const stream_id& sid, strong_actor_ptr x, path_ptr add_path(stream_slot slot, strong_actor_ptr x,
strong_actor_ptr original_stage, stream_priority prio, strong_actor_ptr original_stage, stream_priority prio,
long available_credit, bool redeployable, long available_credit, bool redeployable,
response_promise result_cb) override; response_promise result_cb) override;
bool remove_path(const stream_id& sid, const actor_addr& x, error reason, bool remove_path(stream_slot slot, const actor_addr& x, error reason,
bool silent) override; bool silent) override;
void close(message reason) override; void close(message reason) override;
...@@ -52,7 +52,7 @@ public: ...@@ -52,7 +52,7 @@ public:
path_ptr path_at(size_t index) override; path_ptr path_at(size_t index) override;
path_ptr find(const stream_id& sid, const actor_addr& x) override; path_ptr find(stream_slot slot, const actor_addr& x) override;
void assign_credit(long downstream_capacity) override; void assign_credit(long downstream_capacity) override;
......
...@@ -30,17 +30,17 @@ public: ...@@ -30,17 +30,17 @@ public:
~invalid_stream_scatterer() override; ~invalid_stream_scatterer() override;
path_ptr add_path(const stream_id& sid, strong_actor_ptr origin, path_ptr add_path(stream_slot slot, strong_actor_ptr origin,
strong_actor_ptr sink_ptr, strong_actor_ptr sink_ptr,
mailbox_element::forwarding_stack stages, mailbox_element::forwarding_stack stages,
message_id handshake_mid, message handshake_data, message_id handshake_mid, message handshake_data,
stream_priority prio, bool redeployable) override; stream_priority prio, bool redeployable) override;
path_ptr confirm_path(const stream_id& sid, const actor_addr& from, path_ptr confirm_path(stream_slot slot, const actor_addr& from,
strong_actor_ptr to, long initial_demand, strong_actor_ptr to, long initial_demand,
long desired_batch_size, bool redeployable) override; long desired_batch_size, bool redeployable) override;
bool remove_path(const stream_id& sid, const actor_addr& x, bool remove_path(stream_slot slot, const actor_addr& x,
error reason, bool silent) override; error reason, bool silent) override;
bool paths_clean() const override; bool paths_clean() const override;
...@@ -61,7 +61,7 @@ public: ...@@ -61,7 +61,7 @@ public:
void emit_batches() override; void emit_batches() override;
path_type* find(const stream_id& sid, const actor_addr& x) override; path_type* find(stream_slot slot, const actor_addr& x) override;
long credit() const override; long credit() const override;
......
...@@ -24,11 +24,11 @@ ...@@ -24,11 +24,11 @@
#include <cstdint> #include <cstdint>
#include <cstddef> #include <cstddef>
#include "caf/actor_control_block.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/stream_id.hpp"
#include "caf/stream_msg.hpp"
#include "caf/stream_aborter.hpp" #include "caf/stream_aborter.hpp"
#include "caf/actor_control_block.hpp" #include "caf/stream_slot.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
...@@ -41,10 +41,10 @@ public: ...@@ -41,10 +41,10 @@ public:
static constexpr const auto aborter_type = stream_aborter::sink_aborter; static constexpr const auto aborter_type = stream_aborter::sink_aborter;
/// Message type for propagating graceful shutdowns. /// Message type for propagating graceful shutdowns.
using regular_shutdown = stream_msg::close; using regular_shutdown = downstream_msg::close;
/// Message type for propagating errors. /// Message type for propagating errors.
using irregular_shutdown = stream_msg::forced_close; using irregular_shutdown = downstream_msg::forced_close;
/// Stores information about the initiator of the steam. /// Stores information about the initiator of the steam.
struct client_data { struct client_data {
...@@ -52,12 +52,12 @@ public: ...@@ -52,12 +52,12 @@ public:
message_id mid; message_id mid;
}; };
/// Slot ID for the sink.
stream_slot slot;
/// Pointer to the parent actor. /// Pointer to the parent actor.
local_actor* self; local_actor* self;
/// Stream ID used by the sink.
stream_id sid;
/// Handle to the sink. /// Handle to the sink.
strong_actor_ptr hdl; strong_actor_ptr hdl;
...@@ -80,7 +80,7 @@ public: ...@@ -80,7 +80,7 @@ public:
int64_t next_ack_id; int64_t next_ack_id;
/// Caches batches until receiving an ACK. /// Caches batches until receiving an ACK.
std::deque<std::pair<int64_t, stream_msg::batch>> unacknowledged_batches; std::deque<std::pair<int64_t, downstream_msg::batch>> unacknowledged_batches;
/// Caches the initiator of the stream (client) with the original request ID /// Caches the initiator of the stream (client) with the original request ID
/// until the stream handshake is either confirmed or aborted. Once /// until the stream handshake is either confirmed or aborted. Once
...@@ -92,7 +92,7 @@ public: ...@@ -92,7 +92,7 @@ public:
error shutdown_reason; error shutdown_reason;
/// Constructs a path for given handle and stream ID. /// Constructs a path for given handle and stream ID.
outbound_path(local_actor* selfptr, const stream_id& id, outbound_path(local_actor* selfptr, stream_slot id,
strong_actor_ptr ptr); strong_actor_ptr ptr);
~outbound_path(); ~outbound_path();
...@@ -109,14 +109,15 @@ public: ...@@ -109,14 +109,15 @@ public:
/// `xs_size` and increments `next_batch_id` by 1. /// `xs_size` and increments `next_batch_id` by 1.
void emit_batch(long xs_size, message xs); void emit_batch(long xs_size, message xs);
static void emit_irregular_shutdown(local_actor* self, const stream_id& sid, static void emit_irregular_shutdown(local_actor* self, stream_slot slot,
const strong_actor_ptr& hdl, const strong_actor_ptr& hdl,
error reason); error reason);
}; };
/// @relates outbound_path
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, outbound_path& x) { typename Inspector::result_type inspect(Inspector& f, outbound_path& x) {
return f(meta::type_name("outbound_path"), x.hdl, x.sid, x.next_batch_id, return f(meta::type_name("outbound_path"), x.hdl, x.slot, x.next_batch_id,
x.open_credit, x.redeployable, x.unacknowledged_batches); x.open_credit, x.redeployable, x.unacknowledged_batches);
} }
......
...@@ -307,11 +307,7 @@ public: ...@@ -307,11 +307,7 @@ public:
// -- stream management ------------------------------------------------------ // -- stream management ------------------------------------------------------
/// Returns a new stream ID. /*
stream_id make_stream_id() {
return {ctrl(), new_request_id(message_priority::normal).integer_value()};
}
/// Creates a new stream source and starts streaming to `dest`. /// Creates a new stream source and starts streaming to `dest`.
/// @param dest Actor handle to the stream destination. /// @param dest Actor handle to the stream destination.
/// @param xs User-defined handshake payload. /// @param xs User-defined handshake payload.
...@@ -597,6 +593,7 @@ public: ...@@ -597,6 +593,7 @@ public:
/// manually trigger batches in a source after receiving more data to send. /// manually trigger batches in a source after receiving more data to send.
void trigger_downstreams(); void trigger_downstreams();
*/
/// @cond PRIVATE /// @cond PRIVATE
// -- timeout management ----------------------------------------------------- // -- timeout management -----------------------------------------------------
...@@ -646,8 +643,7 @@ public: ...@@ -646,8 +643,7 @@ public:
inline bool has_behavior() const { inline bool has_behavior() const {
return !bhvr_stack_.empty() return !bhvr_stack_.empty()
|| !awaited_responses_.empty() || !awaited_responses_.empty()
|| !multiplexed_responses_.empty() || !multiplexed_responses_.empty();
|| !streams_.empty();
} }
inline behavior& current_behavior() { inline behavior& current_behavior() {
...@@ -674,6 +670,7 @@ public: ...@@ -674,6 +670,7 @@ public:
return make_message_from_tuple(std::move(ys)); return make_message_from_tuple(std::move(ys));
} }
/*
/// Tries to add a new sink to the stream manager `mgr`. /// Tries to add a new sink to the stream manager `mgr`.
/// @param mgr Pointer to the responsible stream manager. /// @param mgr Pointer to the responsible stream manager.
/// @param sid The ID used for communicating to the sink. /// @param sid The ID used for communicating to the sink.
...@@ -775,7 +772,6 @@ public: ...@@ -775,7 +772,6 @@ public:
return true; return true;
} }
template <class T, class... Ts> template <class T, class... Ts>
void fwd_stream_handshake(const stream_id& sid, std::tuple<Ts...>& xs, void fwd_stream_handshake(const stream_id& sid, std::tuple<Ts...>& xs,
bool ignore_mid = false) { bool ignore_mid = false) {
...@@ -797,6 +793,7 @@ public: ...@@ -797,6 +793,7 @@ public:
if (!ignore_mid) if (!ignore_mid)
mptr->mid.mark_as_answered(); mptr->mid.mark_as_answered();
} }
*/
/// @endcond /// @endcond
...@@ -864,10 +861,6 @@ protected: ...@@ -864,10 +861,6 @@ protected:
/// Pointer to a private thread object associated with a detached actor. /// Pointer to a private thread object associated with a detached actor.
detail::private_thread* private_thread_; detail::private_thread* private_thread_;
// TODO: this type is quite heavy in terms of memory, maybe use vector?
/// Holds state for all streams running through this actor.
streams_map streams_;
# ifndef CAF_NO_EXCEPTIONS # ifndef CAF_NO_EXCEPTIONS
/// Customization point for setting a default exception callback. /// Customization point for setting a default exception callback.
exception_handler exception_handler_; exception_handler exception_handler_;
......
...@@ -20,7 +20,7 @@ ...@@ -20,7 +20,7 @@
#define CAF_STREAM_HPP #define CAF_STREAM_HPP
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/stream_id.hpp" #include "caf/stream_slot.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
...@@ -40,24 +40,23 @@ public: ...@@ -40,24 +40,23 @@ public:
// -- constructors and destructors ------------------------------------------- // -- constructors and destructors -------------------------------------------
stream() = default;
stream(stream&&) = default; stream(stream&&) = default;
stream(const stream&) = default; stream(const stream&) = default;
stream& operator=(stream&&) = default; stream& operator=(stream&&) = default;
stream& operator=(const stream&) = default; stream& operator=(const stream&) = default;
stream(none_t) : stream() { stream(none_t = none) : slot_(0) {
// nop // nop
} }
stream(stream_id sid) : id_(std::move(sid)) { stream(invalid_stream_t) : stream(none) {
// nop // nop
} }
/// Convenience constructor for returning the result of `self->new_stream` /// Convenience constructor for returning the result of `self->new_stream`
/// and similar functions. /// and similar functions.
stream(stream_id sid, stream_manager_ptr sptr) stream(stream_slot id, stream_manager_ptr sptr = nullptr)
: id_(std::move(sid)), : slot_(id),
ptr_(std::move(sptr)) { ptr_(std::move(sptr)) {
// nop // nop
} }
...@@ -65,20 +64,16 @@ public: ...@@ -65,20 +64,16 @@ public:
/// Convenience constructor for returning the result of `self->new_stream` /// Convenience constructor for returning the result of `self->new_stream`
/// and similar functions. /// and similar functions.
stream(stream other, stream_manager_ptr sptr) stream(stream other, stream_manager_ptr sptr)
: id_(std::move(other.id_)), : slot_(std::move(other.slot_)),
ptr_(std::move(sptr)) { ptr_(std::move(sptr)) {
// nop // nop
} }
stream(invalid_stream_t) {
// nop
}
// -- accessors -------------------------------------------------------------- // -- accessors --------------------------------------------------------------
/// Returns the unique identifier for this stream. /// Returns the actor-specific stream slot ID.
inline const stream_id& id() const { inline stream_slot slot() const {
return id_; return slot_;
} }
/// Returns the handler assigned to this stream on this actor. /// Returns the handler assigned to this stream on this actor.
...@@ -90,22 +85,16 @@ public: ...@@ -90,22 +85,16 @@ public:
template <class Inspector> template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, stream& x) { friend typename Inspector::result_type inspect(Inspector& f, stream& x) {
return f(meta::type_name("stream"), x.id_); return f(meta::type_name("stream"), x.slot_);
} }
private: private:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
stream_id id_; stream_slot slot_;
stream_manager_ptr ptr_; stream_manager_ptr ptr_;
}; };
/// @relates stream
template <class T>
inline bool operator==(const stream<T>& x, const stream<T>& y) {
return x.id() == y.id();
}
/// @relates stream /// @relates stream
constexpr invalid_stream_t invalid_stream = invalid_stream_t{}; constexpr invalid_stream_t invalid_stream = invalid_stream_t{};
......
...@@ -19,10 +19,10 @@ ...@@ -19,10 +19,10 @@
#ifndef CAF_STREAM_ABORTER_HPP #ifndef CAF_STREAM_ABORTER_HPP
#define CAF_STREAM_ABORTER_HPP #define CAF_STREAM_ABORTER_HPP
#include "caf/fwd.hpp"
#include "caf/stream_id.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/attachable.hpp" #include "caf/attachable.hpp"
#include "caf/fwd.hpp"
#include "caf/stream_slot.hpp"
namespace caf { namespace caf {
...@@ -35,41 +35,38 @@ public: ...@@ -35,41 +35,38 @@ public:
struct token { struct token {
const actor_addr& observer; const actor_addr& observer;
const stream_id& sid; stream_slot slot;
mode m; mode m;
static constexpr size_t token_type = attachable::token::stream_aborter; static constexpr size_t token_type = attachable::token::stream_aborter;
}; };
stream_aborter(actor_addr&& observed, actor_addr&& observer,
stream_slot slot, mode m);
~stream_aborter() override; ~stream_aborter() override;
void actor_exited(const error& rsn, execution_unit* host) override; void actor_exited(const error& rsn, execution_unit* host) override;
bool matches(const attachable::token& what) override; bool matches(const attachable::token& what) override;
inline static attachable_ptr make(actor_addr observed, actor_addr observer,
const stream_id& sid, mode m) {
return attachable_ptr{
new stream_aborter(std::move(observed), std::move(observer), sid, m)};
}
/// Adds a stream aborter to `observed`. /// Adds a stream aborter to `observed`.
static void add(strong_actor_ptr observed, actor_addr observer, static void add(strong_actor_ptr observed, actor_addr observer,
const stream_id& sid, mode m); stream_slot slot, mode m);
/// Removes a stream aborter from `observed`. /// Removes a stream aborter from `observed`.
static void del(strong_actor_ptr observed, const actor_addr& observer, static void del(strong_actor_ptr observed, const actor_addr& observer,
const stream_id& sid, mode m); stream_slot slot, mode m);
private: private:
stream_aborter(actor_addr&& observed, actor_addr&& observer,
const stream_id& type, mode m);
actor_addr observed_; actor_addr observed_;
actor_addr observer_; actor_addr observer_;
stream_id sid_; stream_slot slot_;
mode mode_; mode mode_;
}; };
attachable_ptr make_stream_aborter(actor_addr observed, actor_addr observer,
stream_slot slot, stream_aborter::mode m);
} // namespace caf } // namespace caf
......
...@@ -28,7 +28,6 @@ ...@@ -28,7 +28,6 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/stream_msg.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/inbound_path.hpp" #include "caf/inbound_path.hpp"
...@@ -92,7 +91,7 @@ public: ...@@ -92,7 +91,7 @@ public:
} }
// -- static utility functions for path vectors ------------------------------ // -- static utility functions for path vectors ------------------------------
/// Sorts `xs` in descending order by available credit. /// Sorts `xs` in descending order by available credit.
template <class PathContainer> template <class PathContainer>
static void sort_by_credit(PathContainer& xs) { static void sort_by_credit(PathContainer& xs) {
...@@ -112,10 +111,10 @@ public: ...@@ -112,10 +111,10 @@ public:
/// Finds the path for `ptr` and returns a pointer to it. /// Finds the path for `ptr` and returns a pointer to it.
template <class PathContainer, class Handle> template <class PathContainer, class Handle>
static path_ptr find(PathContainer& xs, const stream_id& sid, static path_ptr find(PathContainer& xs, stream_slot slot,
const Handle& x) { const Handle& x) {
auto predicate = [&](const typename PathContainer::value_type& y) { auto predicate = [&](const typename PathContainer::value_type& y) {
return y->hdl == x && y->sid == sid; return y->hdl == x && y->slot == slot;
}; };
auto e = xs.end(); auto e = xs.end();
auto i = std::find_if(xs.begin(), e, predicate); auto i = std::find_if(xs.begin(), e, predicate);
...@@ -127,9 +126,9 @@ public: ...@@ -127,9 +126,9 @@ public:
/// Finds the path for `ptr` and returns an iterator to it. /// Finds the path for `ptr` and returns an iterator to it.
template <class PathContainer, class Handle> template <class PathContainer, class Handle>
static typename PathContainer::iterator static typename PathContainer::iterator
iter_find(PathContainer& xs, const stream_id& sid, const Handle& x) { iter_find(PathContainer& xs, stream_slot slot, const Handle& x) {
auto predicate = [&](const typename PathContainer::value_type& y) { auto predicate = [&](const typename PathContainer::value_type& y) {
return y->hdl == x && y->sid == sid; return y->hdl == x && y->slot == slot;
}; };
return std::find_if(xs.begin(), xs.end(), predicate); return std::find_if(xs.begin(), xs.end(), predicate);
} }
...@@ -158,7 +157,7 @@ public: ...@@ -158,7 +157,7 @@ public:
return false; return false;
} }
auto& p = *(*i); auto& p = *(*i);
stream_aborter::del(p.hdl, self_->address(), p.sid, aborter_type); stream_aborter::del(p.hdl, self_->address(), p.slot, aborter_type);
if (silent) if (silent)
p.hdl = nullptr; p.hdl = nullptr;
if (reason != none) if (reason != none)
...@@ -171,11 +170,11 @@ public: ...@@ -171,11 +170,11 @@ public:
// -- implementation of common methods --------------------------------------- // -- implementation of common methods ---------------------------------------
bool remove_path(const stream_id& sid, const actor_addr& x, bool remove_path(stream_slot slot, const actor_addr& x,
error reason, bool silent) override { error reason, bool silent) override {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(x) CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(x)
<< CAF_ARG(reason) << CAF_ARG(silent)); << CAF_ARG(reason) << CAF_ARG(silent));
return remove_path(iter_find(paths_, sid, x), std::move(reason), silent); return remove_path(iter_find(paths_, slot, x), std::move(reason), silent);
} }
void abort(error reason) override { void abort(error reason) override {
...@@ -215,23 +214,23 @@ public: ...@@ -215,23 +214,23 @@ public:
using super::find; using super::find;
path_ptr find(const stream_id& sid, const actor_addr& x) override { path_ptr find(stream_slot slot, const actor_addr& x) override {
return find(paths_, sid, x); return find(paths_, slot, x);
} }
protected: protected:
/// Adds a path to the edge without emitting messages. /// Adds a path to the edge without emitting messages.
path_ptr add_path_impl(const stream_id& sid, strong_actor_ptr x) { path_ptr add_path_impl(stream_slot slot, strong_actor_ptr x) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(sid)); CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(slot));
stream_aborter::add(x, self_->address(), sid, aborter_type); stream_aborter::add(x, self_->address(), slot, aborter_type);
paths_.emplace_back(new path_type(self_, sid, std::move(x))); paths_.emplace_back(new path_type(self_, slot, std::move(x)));
return paths_.back().get(); return paths_.back().get();
} }
template <class F> template <class F>
void close_impl(F f) { void close_impl(F f) {
for (auto& x : paths_) { for (auto& x : paths_) {
stream_aborter::del(x->hdl, self_->address(), x->sid, aborter_type); stream_aborter::del(x->hdl, self_->address(), x->slot, aborter_type);
f(*x); f(*x);
} }
paths_.clear(); paths_.clear();
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include <utility> #include <utility>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/stream_slot.hpp"
namespace caf { namespace caf {
...@@ -47,7 +48,7 @@ public: ...@@ -47,7 +48,7 @@ public:
// -- pure virtual memeber functions ----------------------------------------- // -- pure virtual memeber functions -----------------------------------------
/// Adds a path to the edge and emits `ack_open` to the source. /// Adds a path to the edge and emits `ack_open` to the source.
/// @param sid Stream ID used by the source. /// @param slot Stream ID used by the source.
/// @param x Handle to the source. /// @param x Handle to the source.
/// @param original_stage Actor that received the stream handshake initially. /// @param original_stage Actor that received the stream handshake initially.
/// @param prio Priority of data on this path. /// @param prio Priority of data on this path.
...@@ -59,13 +60,13 @@ public: ...@@ -59,13 +60,13 @@ public:
/// it until the gatherer acknowledges the handshake. The /// it until the gatherer acknowledges the handshake. The
/// callback is invalid if the stream has a next stage. /// callback is invalid if the stream has a next stage.
/// @returns The added path on success, `nullptr` otherwise. /// @returns The added path on success, `nullptr` otherwise.
virtual path_ptr add_path(const stream_id& sid, strong_actor_ptr x, virtual path_ptr add_path(stream_slot slot, strong_actor_ptr x,
strong_actor_ptr original_stage, strong_actor_ptr original_stage,
stream_priority prio, long available_credit, stream_priority prio, long available_credit,
bool redeployable, response_promise result_cb) = 0; bool redeployable, response_promise result_cb) = 0;
/// Removes a path from the gatherer. /// Removes a path from the gatherer.
virtual bool remove_path(const stream_id& sid, const actor_addr& x, virtual bool remove_path(stream_slot slot, const actor_addr& x,
error reason, bool silent) = 0; error reason, bool silent) = 0;
/// Removes all paths gracefully. /// Removes all paths gracefully.
...@@ -87,11 +88,11 @@ public: ...@@ -87,11 +88,11 @@ public:
virtual void continuous(bool value) = 0; virtual void continuous(bool value) = 0;
/// Returns the stored state for `x` if `x` is a known path and associated to /// Returns the stored state for `x` if `x` is a known path and associated to
/// `sid`, otherwise `nullptr`. /// `slot`, otherwise `nullptr`.
virtual path_ptr find(const stream_id& sid, const actor_addr& x) = 0; virtual path_ptr find(stream_slot slot, const actor_addr& x) = 0;
/// Returns the stored state for `x` if `x` is a known path and associated to /// Returns the stored state for `x` if `x` is a known path and associated to
/// `sid`, otherwise `nullptr`. /// `slot`, otherwise `nullptr`.
virtual path_ptr path_at(size_t index) = 0; virtual path_ptr path_at(size_t index) = 0;
/// Assigns new credit to all sources. /// Assigns new credit to all sources.
...@@ -103,11 +104,11 @@ public: ...@@ -103,11 +104,11 @@ public:
// -- convenience functions -------------------------------------------------- // -- convenience functions --------------------------------------------------
/// Removes a path from the gatherer. /// Removes a path from the gatherer.
bool remove_path(const stream_id& sid, const strong_actor_ptr& x, bool remove_path(stream_slot slot, const strong_actor_ptr& x,
error reason, bool silent); error reason, bool silent);
/// Convenience function for calling `find(x, actor_cast<actor_addr>(x))`. /// Convenience function for calling `find(x, actor_cast<actor_addr>(x))`.
path_ptr find(const stream_id& sid, const strong_actor_ptr& x); path_ptr find(stream_slot slot, const strong_actor_ptr& x);
}; };
} // namespace caf } // namespace caf
......
...@@ -41,12 +41,12 @@ public: ...@@ -41,12 +41,12 @@ public:
~stream_gatherer_impl(); ~stream_gatherer_impl();
path_ptr add_path(const stream_id& sid, strong_actor_ptr x, path_ptr add_path(stream_slot slot, strong_actor_ptr x,
strong_actor_ptr original_stage, stream_priority prio, strong_actor_ptr original_stage, stream_priority prio,
long available_credit, bool redeployable, long available_credit, bool redeployable,
response_promise result_cb) override; response_promise result_cb) override;
bool remove_path(const stream_id& sid, const actor_addr& x, error reason, bool remove_path(stream_slot slot, const actor_addr& x, error reason,
bool silent) override; bool silent) override;
void close(message result) override; void close(message result) override;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* 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_STREAM_ID_HPP
#define CAF_STREAM_ID_HPP
#include <cstdint>
#include "caf/actor_addr.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/detail/comparable.hpp"
namespace caf {
class stream_id : detail::comparable<stream_id> {
public:
stream_id(stream_id&&) = default;
stream_id(const stream_id&) = default;
stream_id& operator=(stream_id&&) = default;
stream_id& operator=(const stream_id&) = default;
stream_id();
stream_id(none_t);
stream_id(actor_addr origin_actor, uint64_t origin_nr);
stream_id(actor_control_block* origin_actor, uint64_t origin_nr);
stream_id(const strong_actor_ptr& origin_actor, uint64_t origin_nr);
int64_t compare(const stream_id& other) const;
actor_addr origin;
uint64_t nr;
inline bool valid() const {
return origin != nullptr;
}
};
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_id& x) {
return f(meta::type_name("stream_id"), x.origin, x.nr);
}
} // namespace caf
namespace std {
template <>
struct hash<caf::stream_id> {
size_t operator()(const caf::stream_id& x) const {
auto tmp = reinterpret_cast<ptrdiff_t>(x.origin.get())
^ static_cast<ptrdiff_t>(x.nr);
return static_cast<size_t>(tmp);
}
};
} // namespace std
#endif // CAF_STREAM_ID_HPP
...@@ -24,8 +24,9 @@ ...@@ -24,8 +24,9 @@
#include <cstddef> #include <cstddef>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/ref_counted.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/ref_counted.hpp"
#include "caf/stream_slot.hpp"
namespace caf { namespace caf {
...@@ -35,8 +36,10 @@ class stream_manager : public ref_counted { ...@@ -35,8 +36,10 @@ class stream_manager : public ref_counted {
public: public:
~stream_manager() override; ~stream_manager() override;
/// Handles `stream_msg::open` messages. /// Handles `stream_msg::open` messages by creating a new slot for incoming
/// @returns Initial credit to the source. /// traffic.
/// @param slot Slot ID used by the sender, i.e., the slot ID for upstream
/// messages back to the sender.
/// @param hdl Handle to the sender. /// @param hdl Handle to the sender.
/// @param original_stage Handle to the initial receiver of the handshake. /// @param original_stage Handle to the initial receiver of the handshake.
/// @param priority Affects credit assignment and maximum bandwidth. /// @param priority Affects credit assignment and maximum bandwidth.
...@@ -46,21 +49,22 @@ public: ...@@ -46,21 +49,22 @@ public:
/// Ignored when returning `nullptr`, because the previous /// Ignored when returning `nullptr`, because the previous
/// stage is responsible for it until this manager /// stage is responsible for it until this manager
/// acknowledges the handshake. /// acknowledges the handshake.
/// @returns An error if the stream manager rejects the handshake.
/// @pre `hdl != nullptr` /// @pre `hdl != nullptr`
virtual error open(const stream_id& sid, strong_actor_ptr hdl, virtual error open(stream_slot slot, strong_actor_ptr hdl,
strong_actor_ptr original_stage, stream_priority priority, strong_actor_ptr original_stage, stream_priority priority,
bool redeployable, response_promise result_cb); bool redeployable, response_promise result_cb);
/// Handles `stream_msg::ack_open` messages, i.e., finalizes the stream /// Handles `stream_msg::ack_open` messages, i.e., finalizes the stream
/// handshake. /// handshake.
/// @param sid ID of the outgoing stream. /// @param slot Slot ID used by the sender.
/// @param rebind_from Receiver of the original `open` message. /// @param rebind_from Receiver of the original `open` message.
/// @param rebind_to Sender of this confirmation. /// @param rebind_to Sender of this confirmation.
/// @param initial_demand Credit received with this `ack_open`. /// @param initial_demand Credit received with this `ack_open`.
/// @param redeployable Denotes whether the runtime can redeploy /// @param redeployable Denotes whether the runtime can redeploy
/// `rebind_to` on failure. /// `rebind_to` on failure.
/// @pre `hdl != nullptr` /// @pre `hdl != nullptr`
virtual error ack_open(const stream_id& sid, const actor_addr& rebind_from, virtual error ack_open(stream_slot slot, const actor_addr& rebind_from,
strong_actor_ptr rebind_to, long initial_demand, strong_actor_ptr rebind_to, long initial_demand,
long desired_batch_size, bool redeployable); long desired_batch_size, bool redeployable);
...@@ -71,7 +75,7 @@ public: ...@@ -71,7 +75,7 @@ public:
/// @param xs_id ID of this batch (must be ACKed). /// @param xs_id ID of this batch (must be ACKed).
/// @pre `hdl != nullptr` /// @pre `hdl != nullptr`
/// @pre `xs_size > 0` /// @pre `xs_size > 0`
virtual error batch(const stream_id& sid, const actor_addr& hdl, long xs_size, virtual error batch(stream_slot slot, const actor_addr& hdl, long xs_size,
message& xs, int64_t xs_id); message& xs, int64_t xs_id);
/// Handles `stream_msg::ack_batch` messages. /// Handles `stream_msg::ack_batch` messages.
...@@ -79,19 +83,19 @@ public: ...@@ -79,19 +83,19 @@ public:
/// @param new_demand New credit for sending data. /// @param new_demand New credit for sending data.
/// @param cumulative_batch_id Id of last handled batch. /// @param cumulative_batch_id Id of last handled batch.
/// @pre `hdl != nullptr` /// @pre `hdl != nullptr`
virtual error ack_batch(const stream_id& sid, const actor_addr& hdl, virtual error ack_batch(stream_slot slot, const actor_addr& hdl,
long new_demand, long desired_batch_size, long new_demand, long desired_batch_size,
int64_t cumulative_batch_id); int64_t cumulative_batch_id);
/// Handles `stream_msg::close` messages. /// Handles `stream_msg::close` messages.
/// @param hdl Handle to the sender. /// @param hdl Handle to the sender.
/// @pre `hdl != nullptr` /// @pre `hdl != nullptr`
virtual error close(const stream_id& sid, const actor_addr& hdl); virtual error close(stream_slot slot, const actor_addr& hdl);
/// Handles `stream_msg::drop` messages. /// Handles `stream_msg::drop` messages.
/// @param hdl Handle to the sender. /// @param hdl Handle to the sender.
/// @pre `hdl != nullptr` /// @pre `hdl != nullptr`
virtual error drop(const stream_id& sid, const actor_addr& hdl); virtual error drop(stream_slot slot, const actor_addr& hdl);
/// Handles `stream_msg::drop` messages. The default implementation calls /// Handles `stream_msg::drop` messages. The default implementation calls
/// `abort(reason)` and returns `sec::unhandled_stream_error`. /// `abort(reason)` and returns `sec::unhandled_stream_error`.
...@@ -99,7 +103,7 @@ public: ...@@ -99,7 +103,7 @@ public:
/// @param reason Reported error from the source. /// @param reason Reported error from the source.
/// @pre `hdl != nullptr` /// @pre `hdl != nullptr`
/// @pre `err != none` /// @pre `err != none`
virtual error forced_close(const stream_id& sid, const actor_addr& hdl, virtual error forced_close(stream_slot slot, const actor_addr& hdl,
error reason); error reason);
/// Handles `stream_msg::drop` messages. The default implementation calls /// Handles `stream_msg::drop` messages. The default implementation calls
...@@ -108,11 +112,11 @@ public: ...@@ -108,11 +112,11 @@ public:
/// @param reason Reported error from the sink. /// @param reason Reported error from the sink.
/// @pre `hdl != nullptr` /// @pre `hdl != nullptr`
/// @pre `err != none` /// @pre `err != none`
virtual error forced_drop(const stream_id& sid, const actor_addr& hdl, virtual error forced_drop(stream_slot slot, const actor_addr& hdl,
error reason); error reason);
/// Adds a new sink to the stream. /// Adds a new sink to the stream.
virtual bool add_sink(const stream_id& sid, strong_actor_ptr origin, virtual bool add_sink(stream_slot slot, strong_actor_ptr origin,
strong_actor_ptr sink_ptr, strong_actor_ptr sink_ptr,
mailbox_element::forwarding_stack stages, mailbox_element::forwarding_stack stages,
message_id handshake_mid, message handshake_data, message_id handshake_mid, message handshake_data,
...@@ -120,7 +124,7 @@ public: ...@@ -120,7 +124,7 @@ public:
/// Adds the source `hdl` to a stream. Convenience function for calling /// Adds the source `hdl` to a stream. Convenience function for calling
/// `in().add_path(sid, hdl).second`. /// `in().add_path(sid, hdl).second`.
virtual bool add_source(const stream_id& sid, strong_actor_ptr source_ptr, virtual bool add_source(stream_slot slot, strong_actor_ptr source_ptr,
strong_actor_ptr original_stage, stream_priority prio, strong_actor_ptr original_stage, stream_priority prio,
bool redeployable, response_promise result_cb); bool redeployable, response_promise result_cb);
...@@ -190,7 +194,7 @@ protected: ...@@ -190,7 +194,7 @@ protected:
local_actor* self_; local_actor* self_;
/// Keeps track of pending handshakes. /// Keeps track of pending handshakes.
}; };
/// A reference counting pointer to a `stream_manager`. /// A reference counting pointer to a `stream_manager`.
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* 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_STREAM_MSG_VISITOR_HPP
#define CAF_STREAM_MSG_VISITOR_HPP
#include <utility>
#include <unordered_map>
#include "caf/fwd.hpp"
#include "caf/stream_msg.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/stream_manager.hpp"
#include "caf/scheduled_actor.hpp"
namespace caf {
class stream_msg_visitor {
public:
using map_type = std::unordered_map<stream_id, intrusive_ptr<stream_manager>>;
using iterator = typename map_type::iterator;
using result_type = bool;
stream_msg_visitor(scheduled_actor* self, const stream_msg& msg,
behavior* bhvr);
result_type operator()(stream_msg::open& x);
result_type operator()(stream_msg::ack_open& x);
result_type operator()(stream_msg::batch& x);
result_type operator()(stream_msg::ack_batch& x);
result_type operator()(stream_msg::close& x);
result_type operator()(stream_msg::drop& x);
result_type operator()(stream_msg::forced_close& x);
result_type operator()(stream_msg::forced_drop& x);
private:
// Invokes `f` on the stream manager. On error calls `abort` on the manager
// and removes it from the streams map.
template <class F>
bool invoke(F f) {
auto e = self_->streams().end();
auto i = self_->streams().find(sid_);
if (i == e)
return false;
auto err = f(i->second);
if (err != none) {
i->second->abort(std::move(err));
self_->streams().erase(i);
} else if (i->second->done()) {
CAF_LOG_DEBUG("manager reported done, remove from streams");
i->second->close();
self_->streams().erase(i);
}
return true;
}
scheduled_actor* self_;
const stream_id& sid_;
const actor_addr& sender_;
behavior* bhvr_;
};
} // namespace caf
#endif // CAF_STREAM_MSG_VISITOR_HPP
...@@ -21,7 +21,7 @@ ...@@ -21,7 +21,7 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/stream_id.hpp" #include "caf/stream_slot.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
...@@ -32,24 +32,23 @@ namespace caf { ...@@ -32,24 +32,23 @@ namespace caf {
template <class T> template <class T>
class stream_result { class stream_result {
public: public:
stream_result() = default;
stream_result(stream_result&&) = default; stream_result(stream_result&&) = default;
stream_result(const stream_result&) = default; stream_result(const stream_result&) = default;
stream_result& operator=(stream_result&&) = default; stream_result& operator=(stream_result&&) = default;
stream_result& operator=(const stream_result&) = default; stream_result& operator=(const stream_result&) = default;
stream_result(none_t) : stream_result() { stream_result(none_t = none) : slot_(0) {
// nop // nop
} }
stream_result(stream_id sid) : id_(std::move(sid)) { stream_result(stream_slot id) : slot_(id) {
// nop // nop
} }
/// Convenience constructor for returning the result of `self->new_stream_result` /// Convenience constructor for returning the result of `self->new_stream_result`
/// and similar functions. /// and similar functions.
stream_result(stream_id sid, stream_manager_ptr sptr) stream_result(stream_slot id, stream_manager_ptr sptr)
: id_(std::move(sid)), : slot_(id),
ptr_(std::move(sptr)) { ptr_(std::move(sptr)) {
// nop // nop
} }
...@@ -57,14 +56,14 @@ public: ...@@ -57,14 +56,14 @@ public:
/// Convenience constructor for returning the result of `self->new_stream_result` /// Convenience constructor for returning the result of `self->new_stream_result`
/// and similar functions. /// and similar functions.
stream_result(stream_result other, stream_manager_ptr sptr) stream_result(stream_result other, stream_manager_ptr sptr)
: id_(std::move(other.id_)), : slot_(other.slot_),
ptr_(std::move(sptr)) { ptr_(std::move(sptr)) {
// nop // nop
} }
/// Returns the unique identifier for this stream_result. /// Returns the unique identifier for this stream_result.
inline const stream_id& id() const { inline stream_slot slot() const {
return id_; return slot_;
} }
/// Returns the handler assigned to this stream_result on this actor. /// Returns the handler assigned to this stream_result on this actor.
...@@ -73,12 +72,13 @@ public: ...@@ -73,12 +72,13 @@ public:
} }
template <class Inspector> template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, stream_result& x) { friend typename Inspector::result_type inspect(Inspector& f,
return f(meta::type_name("stream_result"), x.id_); stream_result& x) {
return f(meta::type_name("stream_result"), x.slot_);
} }
private: private:
stream_id id_; stream_slot slot_;
stream_manager_ptr ptr_; stream_manager_ptr ptr_;
}; };
......
...@@ -21,10 +21,11 @@ ...@@ -21,10 +21,11 @@
#include <cstddef> #include <cstddef>
#include "caf/fwd.hpp"
#include "caf/duration.hpp" #include "caf/duration.hpp"
#include "caf/optional.hpp" #include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/optional.hpp"
#include "caf/stream_slot.hpp"
namespace caf { namespace caf {
...@@ -47,19 +48,19 @@ public: ...@@ -47,19 +48,19 @@ public:
/// Adds a path to the edge. /// Adds a path to the edge.
/// @returns The added path on success, `nullptr` otherwise. /// @returns The added path on success, `nullptr` otherwise.
virtual path_ptr add_path(const stream_id& sid, strong_actor_ptr origin, virtual path_ptr add_path(stream_slot slot, strong_actor_ptr origin,
strong_actor_ptr sink_ptr, strong_actor_ptr sink_ptr,
mailbox_element::forwarding_stack stages, mailbox_element::forwarding_stack stages,
message_id handshake_mid, message handshake_data, message_id handshake_mid, message handshake_data,
stream_priority prio, bool redeployable) = 0; stream_priority prio, bool redeployable) = 0;
/// Adds a path to a sink and initiates the handshake. /// Adds a path to a sink and initiates the handshake.
virtual path_ptr confirm_path(const stream_id& sid, const actor_addr& from, virtual path_ptr confirm_path(stream_slot slot, const actor_addr& from,
strong_actor_ptr to, long initial_demand, strong_actor_ptr to, long initial_demand,
long desired_batch_size, bool redeployable) = 0; long desired_batch_size, bool redeployable) = 0;
/// Removes a path from the scatterer. /// Removes a path from the scatterer.
virtual bool remove_path(const stream_id& sid, const actor_addr& x, virtual bool remove_path(stream_slot slot, const actor_addr& x,
error reason, bool silent) = 0; error reason, bool silent) = 0;
/// Returns `true` if there is no data pending and no unacknowledged batch on /// Returns `true` if there is no data pending and no unacknowledged batch on
...@@ -89,7 +90,7 @@ public: ...@@ -89,7 +90,7 @@ public:
/// Returns the stored state for `x` if `x` is a known path and associated to /// Returns the stored state for `x` if `x` is a known path and associated to
/// `sid`, otherwise `nullptr`. /// `sid`, otherwise `nullptr`.
virtual path_ptr find(const stream_id& sid, const actor_addr& x) = 0; virtual path_ptr find(stream_slot slot, const actor_addr& x) = 0;
/// Returns the stored state for `x` if `x` is a known path and associated to /// Returns the stored state for `x` if `x` is a known path and associated to
/// `sid`, otherwise `nullptr`. /// `sid`, otherwise `nullptr`.
...@@ -122,11 +123,11 @@ public: ...@@ -122,11 +123,11 @@ public:
// -- convenience functions -------------------------------------------------- // -- convenience functions --------------------------------------------------
/// Removes a path from the scatterer. /// Removes a path from the scatterer.
bool remove_path(const stream_id& sid, const strong_actor_ptr& x, bool remove_path(stream_slot slot, const strong_actor_ptr& x,
error reason, bool silent); error reason, bool silent);
/// Convenience function for calling `find(x, actor_cast<actor_addr>(x))`. /// Convenience function for calling `find(x, actor_cast<actor_addr>(x))`.
path_ptr find(const stream_id& sid, const strong_actor_ptr& x); path_ptr find(stream_slot slot, const strong_actor_ptr& x);
}; };
} // namespace caf } // namespace caf
......
...@@ -97,7 +97,7 @@ public: ...@@ -97,7 +97,7 @@ public:
static long min_desired_batch_size(const PathContainer& xs) { static long min_desired_batch_size(const PathContainer& xs) {
return !xs.empty() ? fold_desired_batch_size( return !xs.empty() ? fold_desired_batch_size(
xs, std::numeric_limits<long>::max(), xs, std::numeric_limits<long>::max(),
[](long x, long y) { return std::min(x, y); }) [](long x, long y) { return std::min(x, y); })
: 0l; : 0l;
} }
...@@ -131,13 +131,13 @@ public: ...@@ -131,13 +131,13 @@ public:
void close() override; void close() override;
path_ptr add_path(const stream_id& sid, strong_actor_ptr origin, path_ptr add_path(stream_slot slot, strong_actor_ptr origin,
strong_actor_ptr sink_ptr, strong_actor_ptr sink_ptr,
mailbox_element::forwarding_stack stages, mailbox_element::forwarding_stack stages,
message_id handshake_mid, message handshake_data, message_id handshake_mid, message handshake_data,
stream_priority prio, bool redeployable) override; stream_priority prio, bool redeployable) override;
path_ptr confirm_path(const stream_id& sid, const actor_addr& from, path_ptr confirm_path(stream_slot slot, const actor_addr& from,
strong_actor_ptr to, long initial_demand, strong_actor_ptr to, long initial_demand,
long desired_batch_size, bool redeployable) override; long desired_batch_size, bool redeployable) override;
......
...@@ -16,42 +16,16 @@ ...@@ -16,42 +16,16 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/stream_id.hpp" #ifndef CAF_STREAM_SLOT_HPP
#define CAF_STREAM_SLOT_HPP
#include <cstddef> #include <cstdint>
namespace caf { namespace caf {
stream_id::stream_id() : origin(nullptr), nr(0) { /// Actor-specific identifier for stream traffic.
// nop using stream_slot = uint64_t;
}
stream_id::stream_id(none_t) : stream_id() {
// nop
}
stream_id::stream_id(actor_addr origin_actor, uint64_t origin_nr)
: origin(std::move(origin_actor)),
nr(origin_nr) {
// nop
}
stream_id::stream_id(actor_control_block* origin_actor, uint64_t origin_nr)
: stream_id(origin_actor->address(), origin_nr) {
// nop
}
stream_id::stream_id(const strong_actor_ptr& origin_actor, uint64_t origin_nr)
: stream_id(origin_actor->address(), origin_nr) {
// nop
}
int64_t stream_id::compare(const stream_id& other) const {
auto r0 = static_cast<ptrdiff_t>(origin.get() - other.origin.get());
if (r0 != 0)
return static_cast<int64_t>(r0);
return static_cast<int64_t>(nr) - static_cast<int64_t>(other.nr);
}
} // namespace caf } // namespace caf
#endif // CAF_STREAM_SLOT_HPP
...@@ -100,7 +100,7 @@ protected: ...@@ -100,7 +100,7 @@ protected:
} else if (out_.buffered() > 0) { } else if (out_.buffered() > 0) {
push(); push();
} else { } else {
auto sid = path->sid; auto sid = path->slot;
auto hdl = path->hdl; auto hdl = path->hdl;
out().remove_path(sid, hdl, none, false); out().remove_path(sid, hdl, none, false);
} }
......
...@@ -100,7 +100,7 @@ protected: ...@@ -100,7 +100,7 @@ protected:
push(); push();
else if (in_.closed()) { else if (in_.closed()) {
// don't pass path->hdl: path can become invalid // don't pass path->hdl: path can become invalid
auto sid = path->sid; auto sid = path->slot;
out_.remove_path(sid, hdl, none, false); out_.remove_path(sid, hdl, none, false);
} }
auto current_size = out_.buffered(); auto current_size = out_.buffered();
......
...@@ -23,16 +23,14 @@ ...@@ -23,16 +23,14 @@
#include <cstdint> #include <cstdint>
#include <type_traits> #include <type_traits>
#include "caf/fwd.hpp"
#include "caf/group.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/deep_to_string.hpp" #include "caf/deep_to_string.hpp"
#include "caf/fwd.hpp"
#include "caf/group.hpp"
#include "caf/stream_slot.hpp"
#include "caf/meta/type_name.hpp" #include "caf/meta/type_name.hpp"
#include "caf/detail/tbind.hpp"
#include "caf/detail/type_list.hpp"
namespace caf { namespace caf {
/// Sent to all links when an actor is terminated. /// Sent to all links when an actor is terminated.
...@@ -92,6 +90,35 @@ typename Inspector::result_type inspect(Inspector& f, timeout_msg& x) { ...@@ -92,6 +90,35 @@ typename Inspector::result_type inspect(Inspector& f, timeout_msg& x) {
return f(meta::type_name("timeout_msg"), x.timeout_id); return f(meta::type_name("timeout_msg"), x.timeout_id);
} }
/// Initiates a stream handhsake.
struct stream_handshake_msg {
/// Reserved slot on the source.
stream_slot slot;
/// Contains a type-erased stream<T> object as first argument followed by
/// any number of user-defined additional handshake data.
message msg;
/// Identifies the previous stage in the pipeline.
strong_actor_ptr prev_stage;
/// Identifies the original receiver of this message.
strong_actor_ptr original_stage;
/// Configures the priority for stream elements.
stream_priority priority;
/// Tells the downstream whether rebindings can occur on this path.
bool redeployable;
};
/// @relates stream_handshake_msg
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_handshake_msg& x) {
return f(meta::type_name("stream_handshake_msg"), x.slot, x.msg, x.prev_stage,
x.original_stage, x.priority, x.redeployable);
}
} // namespace caf } // namespace caf
#endif // CAF_SYSTEM_MESSAGES_HPP #endif // CAF_SYSTEM_MESSAGES_HPP
...@@ -16,6 +16,9 @@ ...@@ -16,6 +16,9 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#ifndef CAF_TAG_BOXING_TYPE_HPP
#define CAF_TAG_BOXING_TYPE_HPP
namespace caf { namespace caf {
namespace tag { namespace tag {
...@@ -24,3 +27,5 @@ struct boxing_type {}; ...@@ -24,3 +27,5 @@ struct boxing_type {};
} // namespace tag } // namespace tag
} // namespace caf } // namespace caf
#endif // CAF_TAG_BOXING_TYPE_HPP
...@@ -45,8 +45,8 @@ using sorted_builtin_types = ...@@ -45,8 +45,8 @@ using sorted_builtin_types =
atom_value, // @atom atom_value, // @atom
std::vector<char>, // @charbuf std::vector<char>, // @charbuf
down_msg, // @down down_msg, // @down
downstream_msg, // @downstream_msg
duration, // @duration duration, // @duration
timestamp, // @timestamp
error, // @error error, // @error
exit_msg, // @exit exit_msg, // @exit
group, // @group group, // @group
...@@ -60,12 +60,12 @@ using sorted_builtin_types = ...@@ -60,12 +60,12 @@ using sorted_builtin_types =
message_id, // @message_id message_id, // @message_id
node_id, // @node node_id, // @node
std::string, // @str std::string, // @str
stream_msg, // @stream_msg
std::map<std::string, std::string>, // @strmap std::map<std::string, std::string>, // @strmap
strong_actor_ptr, // @strong_actor_ptr strong_actor_ptr, // @strong_actor_ptr
std::set<std::string>, // @strset std::set<std::string>, // @strset
std::vector<std::string>, // @strvec std::vector<std::string>, // @strvec
timeout_msg, // @timeout timeout_msg, // @timeout
timestamp, // @timestamp
uint16_t, // @u16 uint16_t, // @u16
std::u16string, // @u16_str std::u16string, // @u16_str
uint32_t, // @u32 uint32_t, // @u32
...@@ -73,6 +73,7 @@ using sorted_builtin_types = ...@@ -73,6 +73,7 @@ using sorted_builtin_types =
uint64_t, // @u64 uint64_t, // @u64
uint8_t, // @u8 uint8_t, // @u8
unit_t, // @unit unit_t, // @unit
upstream_msg, // @upstream_msg
weak_actor_ptr, // @weak_actor_ptr weak_actor_ptr, // @weak_actor_ptr
bool, // bool bool, // bool
double, // double double, // double
......
...@@ -23,7 +23,6 @@ ...@@ -23,7 +23,6 @@
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/locks.hpp" #include "caf/locks.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/stream_msg.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
namespace caf { namespace caf {
......
...@@ -25,10 +25,10 @@ ...@@ -25,10 +25,10 @@
namespace caf { namespace caf {
inbound_path::inbound_path(local_actor* selfptr, const stream_id& id, inbound_path::inbound_path(local_actor* selfptr, stream_slot id,
strong_actor_ptr ptr) strong_actor_ptr ptr)
: self(selfptr), : self(selfptr),
sid(std::move(id)), slot(id),
hdl(std::move(ptr)), hdl(std::move(ptr)),
prio(stream_priority::normal), prio(stream_priority::normal),
last_acked_batch_id(0), last_acked_batch_id(0),
...@@ -42,11 +42,12 @@ inbound_path::inbound_path(local_actor* selfptr, const stream_id& id, ...@@ -42,11 +42,12 @@ inbound_path::inbound_path(local_actor* selfptr, const stream_id& id,
inbound_path::~inbound_path() { inbound_path::~inbound_path() {
if (hdl) { if (hdl) {
if (shutdown_reason == none) if (shutdown_reason == none)
unsafe_send_as(self, hdl, make<stream_msg::drop>(sid, self->address())); unsafe_send_as(self, hdl,
make<upstream_msg::drop>(slot, self->address()));
else else
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<stream_msg::forced_drop>(sid, self->address(), make<upstream_msg::forced_drop>(slot, self->address(),
shutdown_reason)); shutdown_reason));
} }
} }
...@@ -63,8 +64,8 @@ void inbound_path::emit_ack_open(actor_addr rebind_from, ...@@ -63,8 +64,8 @@ void inbound_path::emit_ack_open(actor_addr rebind_from,
redeployable = is_redeployable; redeployable = is_redeployable;
auto batch_size = static_cast<int32_t>(desired_batch_size); auto batch_size = static_cast<int32_t>(desired_batch_size);
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<stream_msg::ack_open>( make<upstream_msg::ack_open>(
sid, self->address(), std::move(rebind_from), self->ctrl(), slot, self->address(), std::move(rebind_from), self->ctrl(),
static_cast<int32_t>(initial_demand), batch_size, static_cast<int32_t>(initial_demand), batch_size,
is_redeployable)); is_redeployable));
} }
...@@ -75,18 +76,19 @@ void inbound_path::emit_ack_batch(long new_demand) { ...@@ -75,18 +76,19 @@ void inbound_path::emit_ack_batch(long new_demand) {
assigned_credit += new_demand; assigned_credit += new_demand;
auto batch_size = static_cast<int32_t>(desired_batch_size); auto batch_size = static_cast<int32_t>(desired_batch_size);
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<stream_msg::ack_batch>(sid, self->address(), make<upstream_msg::ack_batch>(slot, self->address(),
static_cast<int32_t>(new_demand), static_cast<int32_t>(new_demand),
batch_size, last_batch_id)); batch_size, last_batch_id));
} }
void inbound_path::emit_irregular_shutdown(local_actor* self, void inbound_path::emit_irregular_shutdown(local_actor* self,
const stream_id& sid, stream_slot slot,
const strong_actor_ptr& hdl, const strong_actor_ptr& hdl,
error reason) { error reason) {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl) << CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(hdl) << CAF_ARG(reason));
unsafe_send_as(self, hdl, make<stream_msg::forced_drop>(sid, self->address(), unsafe_send_as(
std::move(reason))); self, hdl,
make<upstream_msg::forced_drop>(slot, self->address(), std::move(reason)));
} }
} // namespace caf } // namespace caf
...@@ -28,14 +28,14 @@ invalid_stream_gatherer::~invalid_stream_gatherer() { ...@@ -28,14 +28,14 @@ invalid_stream_gatherer::~invalid_stream_gatherer() {
} }
stream_gatherer::path_ptr stream_gatherer::path_ptr
invalid_stream_gatherer::add_path(const stream_id&, strong_actor_ptr, invalid_stream_gatherer::add_path(stream_slot, strong_actor_ptr,
strong_actor_ptr, stream_priority, long, bool, strong_actor_ptr, stream_priority, long, bool,
response_promise) { response_promise) {
CAF_LOG_ERROR("invalid_stream_gatherer::add_path called"); CAF_LOG_ERROR("invalid_stream_gatherer::add_path called");
return nullptr; return nullptr;
} }
bool invalid_stream_gatherer::remove_path(const stream_id&, const actor_addr&, bool invalid_stream_gatherer::remove_path(stream_slot, const actor_addr&,
error, bool) { error, bool) {
CAF_LOG_ERROR("invalid_stream_gatherer::remove_path called"); CAF_LOG_ERROR("invalid_stream_gatherer::remove_path called");
return false; return false;
...@@ -69,7 +69,7 @@ stream_gatherer::path_type* invalid_stream_gatherer::path_at(size_t) { ...@@ -69,7 +69,7 @@ stream_gatherer::path_type* invalid_stream_gatherer::path_at(size_t) {
return nullptr; return nullptr;
} }
stream_gatherer::path_type* invalid_stream_gatherer::find(const stream_id&, stream_gatherer::path_type* invalid_stream_gatherer::find(stream_slot,
const actor_addr&) { const actor_addr&) {
return nullptr; return nullptr;
} }
......
...@@ -31,7 +31,7 @@ invalid_stream_scatterer::~invalid_stream_scatterer() { ...@@ -31,7 +31,7 @@ invalid_stream_scatterer::~invalid_stream_scatterer() {
} }
stream_scatterer::path_ptr stream_scatterer::path_ptr
invalid_stream_scatterer::add_path(const stream_id&, strong_actor_ptr, invalid_stream_scatterer::add_path(stream_slot, strong_actor_ptr,
strong_actor_ptr, strong_actor_ptr,
mailbox_element::forwarding_stack, mailbox_element::forwarding_stack,
message_id, message, stream_priority, bool) { message_id, message, stream_priority, bool) {
...@@ -40,13 +40,13 @@ invalid_stream_scatterer::add_path(const stream_id&, strong_actor_ptr, ...@@ -40,13 +40,13 @@ invalid_stream_scatterer::add_path(const stream_id&, strong_actor_ptr,
} }
stream_scatterer::path_ptr stream_scatterer::path_ptr
invalid_stream_scatterer::confirm_path(const stream_id&, const actor_addr&, invalid_stream_scatterer::confirm_path(stream_slot, const actor_addr&,
strong_actor_ptr, long, long, bool) { strong_actor_ptr, long, long, bool) {
CAF_LOG_ERROR("invalid_stream_scatterer::confirm_path called"); CAF_LOG_ERROR("invalid_stream_scatterer::confirm_path called");
return nullptr; return nullptr;
} }
bool invalid_stream_scatterer::remove_path(const stream_id&, const actor_addr&, bool invalid_stream_scatterer::remove_path(stream_slot, const actor_addr&,
error, bool) { error, bool) {
CAF_LOG_ERROR("invalid_stream_scatterer::remove_path called"); CAF_LOG_ERROR("invalid_stream_scatterer::remove_path called");
return false; return false;
...@@ -88,7 +88,7 @@ void invalid_stream_scatterer::emit_batches() { ...@@ -88,7 +88,7 @@ void invalid_stream_scatterer::emit_batches() {
// nop // nop
} }
stream_scatterer::path_type* invalid_stream_scatterer::find(const stream_id&, stream_scatterer::path_type* invalid_stream_scatterer::find(stream_slot,
const actor_addr&) { const actor_addr&) {
return nullptr; return nullptr;
} }
......
...@@ -25,10 +25,10 @@ ...@@ -25,10 +25,10 @@
namespace caf { namespace caf {
outbound_path::outbound_path(local_actor* selfptr, const stream_id& id, outbound_path::outbound_path(local_actor* selfptr, stream_slot id,
strong_actor_ptr ptr) strong_actor_ptr ptr)
: self(selfptr), : slot(id),
sid(id), self(selfptr),
hdl(std::move(ptr)), hdl(std::move(ptr)),
next_batch_id(0), next_batch_id(0),
open_credit(0), open_credit(0),
...@@ -42,11 +42,12 @@ outbound_path::~outbound_path() { ...@@ -42,11 +42,12 @@ outbound_path::~outbound_path() {
CAF_LOG_TRACE(CAF_ARG(shutdown_reason)); CAF_LOG_TRACE(CAF_ARG(shutdown_reason));
if (hdl) { if (hdl) {
if (shutdown_reason == none) if (shutdown_reason == none)
unsafe_send_as(self, hdl, make<stream_msg::close>(sid, self->address())); unsafe_send_as(self, hdl,
make<downstream_msg::close>(slot, self->address()));
else else
unsafe_send_as( unsafe_send_as(self, hdl,
self, hdl, make<downstream_msg::forced_close>(slot, self->address(),
make<stream_msg::forced_close>(sid, self->address(), shutdown_reason)); shutdown_reason));
} }
if (shutdown_reason != none) if (shutdown_reason != none)
unsafe_response(self, std::move(cd.hdl), no_stages, cd.mid, unsafe_response(self, std::move(cd.hdl), no_stages, cd.mid,
...@@ -69,9 +70,9 @@ void outbound_path::emit_open(strong_actor_ptr origin, ...@@ -69,9 +70,9 @@ void outbound_path::emit_open(strong_actor_ptr origin,
redeployable = is_redeployable; redeployable = is_redeployable;
hdl->enqueue( hdl->enqueue(
make_mailbox_element(std::move(origin), handshake_mid, std::move(stages), make_mailbox_element(std::move(origin), handshake_mid, std::move(stages),
make_message(make<stream_msg::open>( make_message(stream_handshake_msg{
sid, self->address(), std::move(handshake_data), slot, std::move(handshake_data), self->ctrl(), hdl,
self->ctrl(), hdl, prio, is_redeployable))), prio, is_redeployable})),
self->context()); self->context());
} }
...@@ -79,19 +80,22 @@ void outbound_path::emit_batch(long xs_size, message xs) { ...@@ -79,19 +80,22 @@ void outbound_path::emit_batch(long xs_size, message xs) {
CAF_LOG_TRACE(CAF_ARG(xs_size) << CAF_ARG(xs)); CAF_LOG_TRACE(CAF_ARG(xs_size) << CAF_ARG(xs));
open_credit -= xs_size; open_credit -= xs_size;
auto bid = next_batch_id++; auto bid = next_batch_id++;
stream_msg::batch batch{static_cast<int32_t>(xs_size), std::move(xs), bid}; downstream_msg::batch batch{static_cast<int32_t>(xs_size), std::move(xs),
bid};
if (redeployable) if (redeployable)
unacknowledged_batches.emplace_back(bid, batch); unacknowledged_batches.emplace_back(bid, batch);
unsafe_send_as(self, hdl, stream_msg{sid, self->address(), std::move(batch)}); unsafe_send_as(self, hdl,
downstream_msg{slot, self->address(), std::move(batch)});
} }
void outbound_path::emit_irregular_shutdown(local_actor* self, void outbound_path::emit_irregular_shutdown(local_actor* self,
const stream_id& sid, stream_slot slot,
const strong_actor_ptr& hdl, const strong_actor_ptr& hdl,
error reason) { error reason) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
unsafe_send_as(self, hdl, make<stream_msg::forced_close>(sid, self->address(), unsafe_send_as(self, hdl,
std::move(reason))); make<downstream_msg::forced_close>(slot, self->address(),
std::move(reason)));
} }
} // namespace caf } // namespace caf
...@@ -21,7 +21,6 @@ ...@@ -21,7 +21,6 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/to_string.hpp" #include "caf/to_string.hpp"
#include "caf/actor_ostream.hpp" #include "caf/actor_ostream.hpp"
#include "caf/stream_msg_visitor.hpp"
#include "caf/detail/private_thread.hpp" #include "caf/detail/private_thread.hpp"
#include "caf/detail/sync_request_bouncer.hpp" #include "caf/detail/sync_request_bouncer.hpp"
...@@ -192,6 +191,7 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) { ...@@ -192,6 +191,7 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
// Clear all state. // Clear all state.
awaited_responses_.clear(); awaited_responses_.clear();
multiplexed_responses_.clear(); multiplexed_responses_.clear();
/*
if (fail_state != none) if (fail_state != none)
for (auto& kvp : streams_) for (auto& kvp : streams_)
kvp.second->abort(fail_state); kvp.second->abort(fail_state);
...@@ -199,6 +199,7 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) { ...@@ -199,6 +199,7 @@ bool scheduled_actor::cleanup(error&& fail_state, execution_unit* host) {
for (auto& kvp : streams_) for (auto& kvp : streams_)
kvp.second->close(); kvp.second->close();
streams_.clear(); streams_.clear();
*/
// Dispatch to parent's `cleanup` function. // Dispatch to parent's `cleanup` function.
return local_actor::cleanup(std::move(fail_state), host); return local_actor::cleanup(std::move(fail_state), host);
} }
...@@ -276,10 +277,12 @@ void scheduled_actor::quit(error x) { ...@@ -276,10 +277,12 @@ void scheduled_actor::quit(error x) {
// -- stream management -------------------------------------------------------- // -- stream management --------------------------------------------------------
/*
void scheduled_actor::trigger_downstreams() { void scheduled_actor::trigger_downstreams() {
for (auto& s : streams_) for (auto& s : streams_)
s.second->push(); s.second->push();
} }
*/
// -- timeout management ------------------------------------------------------- // -- timeout management -------------------------------------------------------
...@@ -382,11 +385,13 @@ scheduled_actor::categorize(mailbox_element& x) { ...@@ -382,11 +385,13 @@ scheduled_actor::categorize(mailbox_element& x) {
call_handler(error_handler_, this, err); call_handler(error_handler_, this, err);
return message_category::internal; return message_category::internal;
} }
/*
case make_type_token<stream_msg>(): { case make_type_token<stream_msg>(): {
auto& bs = bhvr_stack(); auto& bs = bhvr_stack();
handle_stream_msg(x, bs.empty() ? nullptr : &bs.back()); handle_stream_msg(x, bs.empty() ? nullptr : &bs.back());
return message_category::internal; return message_category::internal;
} }
*/
default: default:
return message_category::ordinary; return message_category::ordinary;
} }
...@@ -402,17 +407,22 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) { ...@@ -402,17 +407,22 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
auto ordinary_invoke = [](ptr_t, behavior& f, mailbox_element& in) -> bool { auto ordinary_invoke = [](ptr_t, behavior& f, mailbox_element& in) -> bool {
return f(in.content()) != none; return f(in.content()) != none;
}; };
auto stream_invoke = [](ptr_t p, behavior& f, mailbox_element& in) -> bool { auto stream_invoke = [](ptr_t, behavior&, mailbox_element&) -> bool {
/*
// The only legal stream message in a response is `stream_open`. // The only legal stream message in a response is `stream_open`.
auto& var = in.content().get_as<stream_msg>(0).content; auto& var = in.content().get_as<stream_msg>(0).content;
if (holds_alternative<stream_msg::open>(var)) if (holds_alternative<stream_msg::open>(var))
return p->handle_stream_msg(in, &f); return p->handle_stream_msg(in, &f);
*/
return false; return false;
}; };
auto select_invoke_fun = [&]() -> fun_t { auto select_invoke_fun = [&]() -> fun_t {
return ordinary_invoke;
/*
if (x.content().type_token() != make_type_token<stream_msg>()) if (x.content().type_token() != make_type_token<stream_msg>())
return ordinary_invoke; return ordinary_invoke;
return stream_invoke; return stream_invoke;
*/
}; };
// Short-circuit awaited responses. // Short-circuit awaited responses.
if (!awaited_responses_.empty()) { if (!awaited_responses_.empty()) {
...@@ -622,6 +632,7 @@ bool scheduled_actor::finalize() { ...@@ -622,6 +632,7 @@ bool scheduled_actor::finalize() {
return true; return true;
} }
/*
bool scheduled_actor::handle_stream_msg(mailbox_element& x, bool scheduled_actor::handle_stream_msg(mailbox_element& x,
behavior* active_behavior) { behavior* active_behavior) {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
...@@ -684,5 +695,6 @@ bool scheduled_actor::add_source(const stream_manager_ptr& mgr, ...@@ -684,5 +695,6 @@ bool scheduled_actor::add_source(const stream_manager_ptr& mgr,
std::move(opn.original_stage), opn.priority, std::move(opn.original_stage), opn.priority,
opn.redeployable, std::move(result_cb)); opn.redeployable, std::move(result_cb));
} }
*/
} // namespace caf } // namespace caf
...@@ -19,11 +19,12 @@ ...@@ -19,11 +19,12 @@
#include "caf/stream_aborter.hpp" #include "caf/stream_aborter.hpp"
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/actor_cast.hpp"
#include "caf/stream_msg.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/upstream_msg.hpp"
namespace caf { namespace caf {
...@@ -37,15 +38,15 @@ void stream_aborter::actor_exited(const error& rsn, execution_unit* host) { ...@@ -37,15 +38,15 @@ void stream_aborter::actor_exited(const error& rsn, execution_unit* host) {
if (observer != nullptr) if (observer != nullptr)
{ {
if (mode_ == source_aborter) if (mode_ == source_aborter)
observer->enqueue( observer->enqueue(nullptr, make_message_id(),
nullptr, make_message_id(), make_message(caf::make<downstream_msg::forced_close>(
make_message(caf::make<stream_msg::forced_close>(sid_, observed_, rsn)), slot_, observed_, rsn)),
host); host);
else else
observer->enqueue( observer->enqueue(nullptr, make_message_id(),
nullptr, make_message_id(), make_message(caf::make<upstream_msg::forced_drop>(
make_message(caf::make<stream_msg::forced_drop>(sid_, observed_, rsn)), slot_, observed_, rsn)),
host); host);
} }
} }
...@@ -53,30 +54,37 @@ bool stream_aborter::matches(const attachable::token& what) { ...@@ -53,30 +54,37 @@ bool stream_aborter::matches(const attachable::token& what) {
if (what.subtype != attachable::token::stream_aborter) if (what.subtype != attachable::token::stream_aborter)
return false; return false;
auto& ot = *reinterpret_cast<const token*>(what.ptr); auto& ot = *reinterpret_cast<const token*>(what.ptr);
return ot.observer == observer_ && ot.sid == sid_; return ot.observer == observer_ && ot.slot == slot_;
} }
stream_aborter::stream_aborter(actor_addr&& observed, actor_addr&& observer, stream_aborter::stream_aborter(actor_addr&& observed, actor_addr&& observer,
const stream_id& sid, mode m) stream_slot slot, mode m)
: observed_(std::move(observed)), : observed_(std::move(observed)),
observer_(std::move(observer)), observer_(std::move(observer)),
sid_(sid), slot_(slot),
mode_(m) { mode_(m) {
// nop // nop
} }
void stream_aborter::add(strong_actor_ptr observed, actor_addr observer, void stream_aborter::add(strong_actor_ptr observed, actor_addr observer,
const stream_id& sid, mode m) { stream_slot slot, mode m) {
CAF_LOG_TRACE(CAF_ARG(observed) << CAF_ARG(observer) << CAF_ARG(sid)); CAF_LOG_TRACE(CAF_ARG(observed) << CAF_ARG(observer) << CAF_ARG(slot));
observed->get()->attach(make(observed->address(), std::move(observer), auto ptr = make_stream_aborter(observed->address(), std::move(observer),
sid, m)); slot, m);
observed->get()->attach(std::move(ptr));
} }
void stream_aborter::del(strong_actor_ptr observed, const actor_addr& observer, void stream_aborter::del(strong_actor_ptr observed, const actor_addr& observer,
const stream_id& sid, mode m) { stream_slot slot, mode m) {
CAF_LOG_TRACE(CAF_ARG(observed) << CAF_ARG(observer) << CAF_ARG(sid)); CAF_LOG_TRACE(CAF_ARG(observed) << CAF_ARG(observer) << CAF_ARG(slot));
token tk{observer, sid, m}; token tk{observer, slot, m};
observed->get()->detach(tk); observed->get()->detach(tk);
} }
attachable_ptr make_stream_aborter(actor_addr observed, actor_addr observer,
stream_slot slot, stream_aborter::mode m) {
return attachable_ptr{
new stream_aborter(std::move(observed), std::move(observer), slot, m)};
}
} // namespace caf } // namespace caf
...@@ -29,15 +29,15 @@ stream_gatherer::~stream_gatherer() { ...@@ -29,15 +29,15 @@ stream_gatherer::~stream_gatherer() {
// nop // nop
} }
bool stream_gatherer::remove_path(const stream_id& sid, bool stream_gatherer::remove_path(stream_slot slot, const strong_actor_ptr& x,
const strong_actor_ptr& x, error reason, error reason, bool silent) {
bool silent) { return remove_path(slot, actor_cast<actor_addr>(x), std::move(reason),
return remove_path(sid, actor_cast<actor_addr>(x), std::move(reason), silent); silent);
} }
stream_gatherer::path_type* stream_gatherer::find(const stream_id& sid, stream_gatherer::path_type* stream_gatherer::find(stream_slot slot,
const strong_actor_ptr& x) { const strong_actor_ptr& x) {
return find(sid, actor_cast<actor_addr>(x)); return find(slot, actor_cast<actor_addr>(x));
} }
} // namespace caf } // namespace caf
...@@ -30,20 +30,20 @@ stream_gatherer_impl::~stream_gatherer_impl() { ...@@ -30,20 +30,20 @@ stream_gatherer_impl::~stream_gatherer_impl() {
} }
stream_gatherer::path_ptr stream_gatherer::path_ptr
stream_gatherer_impl::add_path(const stream_id& sid, strong_actor_ptr hdl, stream_gatherer_impl::add_path(stream_slot slot, strong_actor_ptr hdl,
strong_actor_ptr original_stage, strong_actor_ptr original_stage,
stream_priority prio, long available_credit, stream_priority prio, long available_credit,
bool redeployable, response_promise result_cb) { bool redeployable, response_promise result_cb) {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl) << CAF_ARG(original_stage) CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(hdl) << CAF_ARG(original_stage)
<< CAF_ARG(prio) << CAF_ARG(available_credit)); << CAF_ARG(prio) << CAF_ARG(available_credit));
CAF_ASSERT(hdl != nullptr); CAF_ASSERT(hdl != nullptr);
if (find(sid, hdl) != nullptr) { if (find(slot, hdl) != nullptr) {
inbound_path::emit_irregular_shutdown(self_, sid, hdl, inbound_path::emit_irregular_shutdown(self_, slot, hdl,
sec::cannot_add_upstream); sec::cannot_add_upstream);
return nullptr; return nullptr;
} }
auto ptr = add_path_impl(sid, std::move(hdl)); auto ptr = add_path_impl(slot, std::move(hdl));
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
assignment_vec_.emplace_back(ptr, 0l); assignment_vec_.emplace_back(ptr, 0l);
if (result_cb.pending()) if (result_cb.pending())
...@@ -54,19 +54,19 @@ stream_gatherer_impl::add_path(const stream_id& sid, strong_actor_ptr hdl, ...@@ -54,19 +54,19 @@ stream_gatherer_impl::add_path(const stream_id& sid, strong_actor_ptr hdl,
return ptr; return ptr;
} }
bool stream_gatherer_impl::remove_path(const stream_id& sid, bool stream_gatherer_impl::remove_path(stream_slot slot,
const actor_addr& x, error reason, const actor_addr& x, error reason,
bool silent) { bool silent) {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(x) CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(x)
<< CAF_ARG(reason) << CAF_ARG(silent)); << CAF_ARG(reason) << CAF_ARG(silent));
auto pred = [&](const assignment_pair& y) { auto pred = [&](const assignment_pair& y) {
return y.first->sid == sid && y.first->hdl == x; return y.first->slot == slot && y.first->hdl == x;
}; };
auto e = assignment_vec_.end(); auto e = assignment_vec_.end();
auto i = std::find_if(assignment_vec_.begin(), e, pred); auto i = std::find_if(assignment_vec_.begin(), e, pred);
if (i != e) { if (i != e) {
assignment_vec_.erase(i); assignment_vec_.erase(i);
return super::remove_path(sid, x, std::move(reason), silent); return super::remove_path(slot, x, std::move(reason), silent);
} }
return false; return false;
} }
...@@ -75,7 +75,7 @@ void stream_gatherer_impl::close(message result) { ...@@ -75,7 +75,7 @@ void stream_gatherer_impl::close(message result) {
CAF_LOG_TRACE(CAF_ARG(result) << CAF_ARG2("remaining paths", paths_.size()) CAF_LOG_TRACE(CAF_ARG(result) << CAF_ARG2("remaining paths", paths_.size())
<< CAF_ARG2("listener", listeners_.size())); << CAF_ARG2("listener", listeners_.size()));
for (auto& path : paths_) for (auto& path : paths_)
stream_aborter::del(path->hdl, self_->address(), path->sid, aborter_type); stream_aborter::del(path->hdl, self_->address(), path->slot, aborter_type);
paths_.clear(); paths_.clear();
assignment_vec_.clear(); assignment_vec_.clear();
for (auto& listener : listeners_) for (auto& listener : listeners_)
...@@ -85,7 +85,7 @@ void stream_gatherer_impl::close(message result) { ...@@ -85,7 +85,7 @@ void stream_gatherer_impl::close(message result) {
void stream_gatherer_impl::abort(error reason) { void stream_gatherer_impl::abort(error reason) {
for (auto& path : paths_) { for (auto& path : paths_) {
stream_aborter::del(path->hdl, self_->address(), path->sid, aborter_type); stream_aborter::del(path->hdl, self_->address(), path->slot, aborter_type);
path->shutdown_reason = reason; path->shutdown_reason = reason;
} }
paths_.clear(); paths_.clear();
......
...@@ -38,50 +38,55 @@ stream_manager::~stream_manager() { ...@@ -38,50 +38,55 @@ stream_manager::~stream_manager() {
// nop // nop
} }
error stream_manager::open(const stream_id& sid, strong_actor_ptr hdl, error stream_manager::open(stream_slot slot, strong_actor_ptr hdl,
strong_actor_ptr original_stage, strong_actor_ptr original_stage,
stream_priority prio, bool redeployable, stream_priority prio, bool redeployable,
response_promise result_cb) { response_promise result_cb) {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl) << CAF_ARG(original_stage) CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(hdl) << CAF_ARG(original_stage)
<< CAF_ARG(prio) << CAF_ARG(redeployable)); << CAF_ARG(prio) << CAF_ARG(redeployable));
/*
if (hdl == nullptr) if (hdl == nullptr)
return sec::invalid_argument; return sec::invalid_argument;
if (in().add_path(sid, hdl, std::move(original_stage), prio, if (in().add_path(slot, hdl, std::move(original_stage), prio,
out().credit(), redeployable, std::move(result_cb)) out().credit(), redeployable, std::move(result_cb))
!= nullptr) != nullptr)
return none; return none;
*/
return sec::cannot_add_upstream; return sec::cannot_add_upstream;
} }
error stream_manager::ack_open(const stream_id& sid, error stream_manager::ack_open(stream_slot slot,
const actor_addr& rebind_from, const actor_addr& rebind_from,
strong_actor_ptr rebind_to, long initial_demand, strong_actor_ptr rebind_to, long initial_demand,
long desired_batch_size, bool redeployable) { long desired_batch_size, bool redeployable) {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(rebind_from) << CAF_ARG(rebind_to) CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(rebind_from) << CAF_ARG(rebind_to)
<< CAF_ARG(initial_demand) << CAF_ARG(redeployable)); << CAF_ARG(initial_demand) << CAF_ARG(redeployable));
/*
if (rebind_from == nullptr) if (rebind_from == nullptr)
return sec::invalid_argument; return sec::invalid_argument;
if (rebind_to == nullptr) { if (rebind_to == nullptr) {
auto from_ptr = actor_cast<strong_actor_ptr>(rebind_from); auto from_ptr = actor_cast<strong_actor_ptr>(rebind_from);
if (from_ptr) if (from_ptr)
out().remove_path(sid, from_ptr, sec::invalid_downstream, false); out().remove_path(slot, from_ptr, sec::invalid_downstream, false);
return sec::invalid_downstream; return sec::invalid_downstream;
} }
auto ptr = out().confirm_path(sid, rebind_from, std::move(rebind_to), auto ptr = out().confirm_path(slot, rebind_from, std::move(rebind_to),
initial_demand, desired_batch_size, initial_demand, desired_batch_size,
redeployable); redeployable);
if (ptr == nullptr) if (ptr == nullptr)
return sec::invalid_downstream; return sec::invalid_downstream;
downstream_demand(ptr, initial_demand); downstream_demand(ptr, initial_demand);
*/
return none; return none;
} }
error stream_manager::batch(const stream_id& sid, const actor_addr& hdl, error stream_manager::batch(stream_slot slot, const actor_addr& hdl,
long xs_size, message& xs, int64_t xs_id) { long xs_size, message& xs, int64_t xs_id) {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl) << CAF_ARG(xs_size) CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(hdl) << CAF_ARG(xs_size)
<< CAF_ARG(xs) << CAF_ARG(xs_id)); << CAF_ARG(xs) << CAF_ARG(xs_id));
CAF_ASSERT(hdl != nullptr); CAF_ASSERT(hdl != nullptr);
auto ptr = in().find(sid, hdl); /*
auto ptr = in().find(slot, hdl);
if (ptr == nullptr) { if (ptr == nullptr) {
CAF_LOG_WARNING("received batch for unknown stream"); CAF_LOG_WARNING("received batch for unknown stream");
return sec::invalid_downstream; return sec::invalid_downstream;
...@@ -101,95 +106,119 @@ error stream_manager::batch(const stream_id& sid, const actor_addr& hdl, ...@@ -101,95 +106,119 @@ error stream_manager::batch(const stream_id& sid, const actor_addr& hdl,
in().assign_credit(desired_size - current_size); in().assign_credit(desired_size - current_size);
} }
return err; return err;
*/
return none;
} }
error stream_manager::ack_batch(const stream_id& sid, const actor_addr& hdl, error stream_manager::ack_batch(stream_slot slot, const actor_addr& hdl,
long demand, long batch_size, int64_t) { long demand, long batch_size, int64_t) {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl) << CAF_ARG(demand)); CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(hdl) << CAF_ARG(demand));
auto ptr = out().find(sid, hdl); /*
auto ptr = out().find(slot, hdl);
if (ptr == nullptr) if (ptr == nullptr)
return sec::invalid_downstream; return sec::invalid_downstream;
ptr->open_credit += demand; ptr->open_credit += demand;
if (ptr->desired_batch_size != batch_size) if (ptr->desired_batch_size != batch_size)
ptr->desired_batch_size = batch_size; ptr->desired_batch_size = batch_size;
downstream_demand(ptr, demand); downstream_demand(ptr, demand);
*/
return none; return none;
} }
error stream_manager::close(const stream_id& sid, const actor_addr& hdl) { error stream_manager::close(stream_slot slot, const actor_addr& hdl) {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(hdl)); CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(hdl));
if (in().remove_path(sid, hdl, none, true) && in().closed()) /*
if (in().remove_path(slot, hdl, none, true) && in().closed())
input_closed(none); input_closed(none);
*/
return none; return none;
} }
error stream_manager::drop(const stream_id& sid, const actor_addr& hdl) { error stream_manager::drop(stream_slot slot, const actor_addr& hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl)); CAF_LOG_TRACE(CAF_ARG(hdl));
if (out().remove_path(sid, hdl, none, true) && out().closed()) /*
if (out().remove_path(slot, hdl, none, true) && out().closed())
output_closed(none); output_closed(none);
*/
return none; return none;
} }
error stream_manager::forced_close(const stream_id& sid, const actor_addr& hdl, error stream_manager::forced_close(stream_slot slot, const actor_addr& hdl,
error reason) { error reason) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(reason));
CAF_IGNORE_UNUSED(hdl); CAF_IGNORE_UNUSED(hdl);
in().remove_path(sid, hdl, reason, true); /*
in().remove_path(slot, hdl, reason, true);
abort(std::move(reason)); abort(std::move(reason));
return sec::unhandled_stream_error; return sec::unhandled_stream_error;
*/
return none;
} }
error stream_manager::forced_drop(const stream_id& sid, const actor_addr& hdl, error stream_manager::forced_drop(stream_slot slot, const actor_addr& hdl,
error reason) { error reason) {
/*
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(reason));
CAF_IGNORE_UNUSED(hdl); CAF_IGNORE_UNUSED(hdl);
out().remove_path(sid, hdl, reason, true); out().remove_path(slot, hdl, reason, true);
abort(std::move(reason)); abort(std::move(reason));
return sec::unhandled_stream_error; return sec::unhandled_stream_error;
*/
return none;
} }
void stream_manager::abort(caf::error reason) { void stream_manager::abort(caf::error reason) {
/*
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
in().abort(reason); in().abort(reason);
input_closed(reason); input_closed(reason);
out().abort(reason); out().abort(reason);
output_closed(std::move(reason)); output_closed(std::move(reason));
*/
} }
void stream_manager::close() { void stream_manager::close() {
/*
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
in().close(make_final_result()); in().close(make_final_result());
input_closed(none); input_closed(none);
out().close(); out().close();
output_closed(none); output_closed(none);
*/
} }
bool stream_manager::add_sink(const stream_id& sid, strong_actor_ptr origin, bool stream_manager::add_sink(stream_slot slot, strong_actor_ptr origin,
strong_actor_ptr sink_ptr, strong_actor_ptr sink_ptr,
mailbox_element::forwarding_stack stages, mailbox_element::forwarding_stack stages,
message_id handshake_mid, message handshake_data, message_id handshake_mid, message handshake_data,
stream_priority prio, bool redeployable) { stream_priority prio, bool redeployable) {
return out().add_path(sid, std::move(origin), std::move(sink_ptr), /*
return out().add_path(slot, std::move(origin), std::move(sink_ptr),
std::move(stages), handshake_mid, std::move(stages), handshake_mid,
std::move(handshake_data), prio, redeployable) std::move(handshake_data), prio, redeployable)
!= nullptr; != nullptr;
*/
} }
bool stream_manager::add_source(const stream_id& sid, bool stream_manager::add_source(stream_slot slot,
strong_actor_ptr source_ptr, strong_actor_ptr source_ptr,
strong_actor_ptr original_stage, strong_actor_ptr original_stage,
stream_priority prio, bool redeployable, stream_priority prio, bool redeployable,
response_promise result_cb) { response_promise result_cb) {
/*
// TODO: out().credit() gives the same amount of credit to any number of new // TODO: out().credit() gives the same amount of credit to any number of new
// sources -> feedback needed // sources -> feedback needed
return in().add_path(sid, std::move(source_ptr), std::move(original_stage), return in().add_path(slot, std::move(source_ptr), std::move(original_stage),
prio, out().credit(), redeployable, std::move(result_cb)) prio, out().credit(), redeployable, std::move(result_cb))
!= nullptr; != nullptr;
*/
} }
void stream_manager::push() { void stream_manager::push() {
/*
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
out().emit_batches(); out().emit_batches();
*/
} }
bool stream_manager::generate_messages() { bool stream_manager::generate_messages() {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* 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/stream_msg_visitor.hpp"
#include "caf/send.hpp"
#include "caf/logger.hpp"
#include "caf/inbound_path.hpp"
#include "caf/outbound_path.hpp"
#include "caf/scheduled_actor.hpp"
namespace caf {
stream_msg_visitor::stream_msg_visitor(scheduled_actor* self,
const stream_msg& msg, behavior* bhvr)
: self_(self),
sid_(msg.sid),
sender_(msg.sender),
bhvr_(bhvr) {
CAF_ASSERT(sender_ != nullptr);
}
auto stream_msg_visitor::operator()(stream_msg::open& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
CAF_ASSERT(x.prev_stage != nullptr);
CAF_ASSERT(x.original_stage != nullptr);
auto& predecessor = x.prev_stage;
// Convenience function for aborting the stream on error.
auto fail = [&](error err) -> result_type {
inbound_path::emit_irregular_shutdown(self_, sid_, predecessor,
std::move(err));
return false;
};
// Sanity checks.
if (!predecessor) {
CAF_LOG_WARNING("received stream_msg::open with empty prev_stage");
return fail(sec::invalid_upstream);
}
if (bhvr_ == nullptr) {
CAF_LOG_WARNING("received stream_msg::open with empty behavior");
return fail(sec::stream_init_failed);
}
if (self_->streams().count(sid_) != 0) {
CAF_LOG_WARNING("received duplicate stream_msg::open");
return fail(sec::stream_init_failed);
}
// Invoke behavior of parent to perform handshake.
(*bhvr_)(x.msg);
if (self_->streams().count(sid_) == 0) {
CAF_LOG_WARNING("actor did not provide a stream "
"handler after receiving handshake:"
<< CAF_ARG(x.msg));
return fail(sec::stream_init_failed);
}
return true;
}
auto stream_msg_visitor::operator()(stream_msg::close&) -> result_type {
CAF_LOG_TRACE("");
return invoke([&](stream_manager_ptr& mgr) {
return mgr->close(sid_, sender_);
});
}
auto stream_msg_visitor::operator()(stream_msg::drop&) -> result_type {
CAF_LOG_TRACE("");
return invoke([&](stream_manager_ptr& mgr) {
return mgr->drop(sid_, sender_);
});
}
auto stream_msg_visitor::operator()(stream_msg::forced_close& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
return invoke([&](stream_manager_ptr& mgr) {
return mgr->forced_close(sid_, sender_, std::move(x.reason));
});
}
auto stream_msg_visitor::operator()(stream_msg::forced_drop& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
return invoke([&](stream_manager_ptr& mgr) {
return mgr->forced_drop(sid_, sender_, std::move(x.reason));
});
}
auto stream_msg_visitor::operator()(stream_msg::ack_open& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
return invoke([&](stream_manager_ptr& mgr) {
return mgr->ack_open(sid_, x.rebind_from, std::move(x.rebind_to),
x.initial_demand, x.desired_batch_size,
x.redeployable);
});
}
auto stream_msg_visitor::operator()(stream_msg::batch& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
return invoke([&](stream_manager_ptr& mgr) {
return mgr->batch(sid_, sender_, static_cast<long>(x.xs_size), x.xs, x.id);
});
}
auto stream_msg_visitor::operator()(stream_msg::ack_batch& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
return invoke([&](stream_manager_ptr& mgr) {
return mgr->ack_batch(sid_, sender_, x.new_capacity, x.desired_batch_size,
x.acknowledged_id);
});
}
} // namespace caf
...@@ -29,15 +29,16 @@ stream_scatterer::~stream_scatterer() { ...@@ -29,15 +29,16 @@ stream_scatterer::~stream_scatterer() {
// nop // nop
} }
bool stream_scatterer::remove_path(const stream_id& sid, bool stream_scatterer::remove_path(stream_slot slot,
const strong_actor_ptr& x, error reason, const strong_actor_ptr& x, error reason,
bool silent) { bool silent) {
return remove_path(sid, actor_cast<actor_addr>(x), std::move(reason), silent); return remove_path(slot, actor_cast<actor_addr>(x), std::move(reason),
silent);
} }
stream_scatterer::path_type* stream_scatterer::find(const stream_id& sid, stream_scatterer::path_type* stream_scatterer::find(stream_slot slot,
const strong_actor_ptr& x) { const strong_actor_ptr& x) {
return find(sid, actor_cast<actor_addr>(x)); return find(slot, actor_cast<actor_addr>(x));
} }
} // namespace caf } // namespace caf
...@@ -34,17 +34,17 @@ stream_scatterer_impl::~stream_scatterer_impl() { ...@@ -34,17 +34,17 @@ stream_scatterer_impl::~stream_scatterer_impl() {
} }
stream_scatterer::path_ptr stream_scatterer::path_ptr
stream_scatterer_impl::add_path(const stream_id& sid, strong_actor_ptr origin, stream_scatterer_impl::add_path(stream_slot slot, strong_actor_ptr origin,
strong_actor_ptr sink_ptr, strong_actor_ptr sink_ptr,
mailbox_element::forwarding_stack stages, mailbox_element::forwarding_stack stages,
message_id handshake_mid, message_id handshake_mid,
message handshake_data, stream_priority prio, message handshake_data, stream_priority prio,
bool redeployable) { bool redeployable) {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(origin) << CAF_ARG(sink_ptr) CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(origin) << CAF_ARG(sink_ptr)
<< CAF_ARG(stages) << CAF_ARG(handshake_mid) << CAF_ARG(stages) << CAF_ARG(handshake_mid)
<< CAF_ARG(handshake_data) << CAF_ARG(prio) << CAF_ARG(handshake_data) << CAF_ARG(prio)
<< CAF_ARG(redeployable)); << CAF_ARG(redeployable));
auto ptr = add_path_impl(sid, std::move(sink_ptr)); auto ptr = add_path_impl(slot, std::move(sink_ptr));
if (ptr != nullptr) if (ptr != nullptr)
ptr->emit_open(std::move(origin), std::move(stages), handshake_mid, ptr->emit_open(std::move(origin), std::move(stages), handshake_mid,
std::move(handshake_data), prio, redeployable); std::move(handshake_data), prio, redeployable);
...@@ -52,15 +52,15 @@ stream_scatterer_impl::add_path(const stream_id& sid, strong_actor_ptr origin, ...@@ -52,15 +52,15 @@ stream_scatterer_impl::add_path(const stream_id& sid, strong_actor_ptr origin,
} }
stream_scatterer::path_ptr stream_scatterer_impl::confirm_path( stream_scatterer::path_ptr stream_scatterer_impl::confirm_path(
const stream_id& sid, const actor_addr& from, strong_actor_ptr to, stream_slot slot, const actor_addr& from, strong_actor_ptr to,
long initial_demand, long desired_batch_size, bool redeployable) { long initial_demand, long desired_batch_size, bool redeployable) {
CAF_LOG_TRACE(CAF_ARG(sid) << CAF_ARG(from) << CAF_ARG(to) CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(from) << CAF_ARG(to)
<< CAF_ARG(initial_demand) << CAF_ARG(desired_batch_size) << CAF_ARG(initial_demand) << CAF_ARG(desired_batch_size)
<< CAF_ARG(redeployable)); << CAF_ARG(redeployable));
auto ptr = find(sid, from); auto ptr = find(slot, from);
if (ptr == nullptr) { if (ptr == nullptr) {
CAF_LOG_WARNING("cannot confirm unknown path"); CAF_LOG_WARNING("cannot confirm unknown path");
outbound_path::emit_irregular_shutdown(self_, sid, std::move(to), outbound_path::emit_irregular_shutdown(self_, slot, std::move(to),
sec::invalid_downstream); sec::invalid_downstream);
return nullptr; return nullptr;
} }
...@@ -83,13 +83,13 @@ bool stream_scatterer_impl::paths_clean() const { ...@@ -83,13 +83,13 @@ bool stream_scatterer_impl::paths_clean() const {
void stream_scatterer_impl::close() { void stream_scatterer_impl::close() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
for (auto& path : paths_) for (auto& path : paths_)
stream_aborter::del(path->hdl, self_->address(), path->sid, aborter_type); stream_aborter::del(path->hdl, self_->address(), path->slot, aborter_type);
paths_.clear(); paths_.clear();
} }
void stream_scatterer_impl::abort(error reason) { void stream_scatterer_impl::abort(error reason) {
for (auto& path : paths_) { for (auto& path : paths_) {
stream_aborter::del(path->hdl, self_->address(), path->sid, aborter_type); stream_aborter::del(path->hdl, self_->address(), path->slot, aborter_type);
path->shutdown_reason = reason; path->shutdown_reason = reason;
} }
paths_.clear(); paths_.clear();
......
...@@ -28,23 +28,23 @@ ...@@ -28,23 +28,23 @@
#include <algorithm> #include <algorithm>
#include <type_traits> #include <type_traits>
#include "caf/locks.hpp" #include "caf/abstract_group.hpp"
#include "caf/string_algorithms.hpp" #include "caf/actor_cast.hpp"
#include "caf/actor_factory.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/duration.hpp"
#include "caf/group.hpp" #include "caf/group.hpp"
#include "caf/locks.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/type_nr.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/duration.hpp"
#include "caf/timestamp.hpp"
#include "caf/actor_cast.hpp"
#include "caf/stream_msg.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_factory.hpp"
#include "caf/abstract_group.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/message_builder.hpp" #include "caf/message_builder.hpp"
#include "caf/actor_system_config.hpp" #include "caf/proxy_registry.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/timestamp.hpp"
#include "caf/type_nr.hpp"
#include "caf/upstream_msg.hpp"
#include "caf/detail/safe_equal.hpp" #include "caf/detail/safe_equal.hpp"
#include "caf/detail/scope_guard.hpp" #include "caf/detail/scope_guard.hpp"
...@@ -60,8 +60,8 @@ const char* numbered_type_names[] = { ...@@ -60,8 +60,8 @@ const char* numbered_type_names[] = {
"@atom", "@atom",
"@charbuf", "@charbuf",
"@down", "@down",
"@downstream_msg",
"@duration", "@duration",
"@timestamp",
"@error", "@error",
"@exit", "@exit",
"@group", "@group",
...@@ -75,12 +75,12 @@ const char* numbered_type_names[] = { ...@@ -75,12 +75,12 @@ const char* numbered_type_names[] = {
"@message_id", "@message_id",
"@node", "@node",
"@str", "@str",
"@stream_msg",
"@strmap", "@strmap",
"@strong_actor_ptr", "@strong_actor_ptr",
"@strset", "@strset",
"@strvec", "@strvec",
"@timeout", "@timeout",
"@timestamp",
"@u16", "@u16",
"@u16str", "@u16str",
"@u32", "@u32",
...@@ -88,6 +88,7 @@ const char* numbered_type_names[] = { ...@@ -88,6 +88,7 @@ const char* numbered_type_names[] = {
"@u64", "@u64",
"@u8", "@u8",
"@unit", "@unit",
"@upstream_msg",
"@weak_actor_ptr", "@weak_actor_ptr",
"bool", "bool",
"double", "double",
......
This diff is collapsed.
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* 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. *
******************************************************************************/
// this test is essentially a subset of streaming.cpp and tests whether the
// 3-stage pipeline compiles and runs when using a type-safe version of each
// stage
#include <deque>
#include <string>
#include <vector>
#define CAF_SUITE typed_stream_stages
#include "caf/test/dsl.hpp"
using std::string;
using namespace caf;
namespace {
using file_reader_actor = typed_actor<replies_to<string>
::with<stream<int>, string>>;
using filter_actor = typed_actor<replies_to<stream<int>, string>
::with<stream<int>, string>>;
using sum_up_actor = typed_actor<replies_to<stream<int>, string>::with<int>>;
file_reader_actor::behavior_type file_reader(file_reader_actor::pointer self) {
using buf = std::deque<int>;
return {
[=](std::string& fname) {
CAF_CHECK_EQUAL(fname, "test.txt");
return self->make_source(
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
// 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();
}
);
}
};
}
filter_actor::behavior_type filter(filter_actor::pointer self) {
return {
[=](stream<int>& in, std::string& fname) {
CAF_CHECK_EQUAL(fname, "test.txt");
return self->make_stage(
// input stream
in,
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
// initialize state
[=](unit_t&) {
// nop
},
// processing step
[=](unit_t&, downstream<int>& out, int x) {
if ((x & 0x01) != 0)
out.push(x);
},
// cleanup
[=](unit_t&) {
// nop
}
);
}
};
}
sum_up_actor::behavior_type sum_up(sum_up_actor::pointer self) {
return {
[=](stream<int>& in, std::string& fname) {
CAF_CHECK_EQUAL(fname, "test.txt");
return self->make_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;
}
);
}
};
}
using fixture = test_coordinator_fixture<>;
using pipeline_actor = typed_actor<replies_to<string>::with<int>>;
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(typed_streaming_tests, fixture)
CAF_TEST(depth3_pipeline) {
scoped_actor self{sys};
auto source = sys.spawn(file_reader);
auto stage = sys.spawn(filter);
auto sink = sys.spawn(sum_up);
auto pipeline = (sink * stage) * source;
CAF_MESSAGE("source: " << to_string(source));
CAF_MESSAGE("stage: " << to_string(stage));
CAF_MESSAGE("sink: " << to_string(sink));
CAF_MESSAGE("pipeline: " << to_string(pipeline));
static_assert(std::is_same<decltype(pipeline), pipeline_actor>::value,
"pipeline composition returned wrong type");
sched.run();
sched.after_next_enqueue([&] { sched.run(); });
self->request(pipeline, infinite, "test.txt").receive(
[&](int x) {
CAF_CHECK_EQUAL(x, 25);
},
[&](error& err) {
CAF_FAIL("error: " << sys.render(err));
}
);
}
CAF_TEST_FIXTURE_SCOPE_END()
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