Commit 9008efc4 authored by Dominik Charousset's avatar Dominik Charousset Committed by Marian Triebe

Add early experimental streaming support

parent 55241fd8
...@@ -95,8 +95,24 @@ set (LIBCAF_CORE_SRCS ...@@ -95,8 +95,24 @@ set (LIBCAF_CORE_SRCS
src/type_erased_tuple.cpp src/type_erased_tuple.cpp
src/uniform_type_info_map.cpp src/uniform_type_info_map.cpp
src/unprofiled.cpp src/unprofiled.cpp
src/greedy.cpp
src/broadcast.cpp
src/anycast.cpp
src/work_stealing.cpp src/work_stealing.cpp
src/work_sharing.cpp) src/work_sharing.cpp
src/abstract_downstream.cpp
src/abstract_upstream.cpp
src/downstream_path.cpp
src/downstream_policy.cpp
src/stream.cpp
src/stream_handler.cpp
src/stream_msg_visitor.cpp
src/stream_priority.cpp
src/stream_sink.cpp
src/stream_source.cpp
src/stream_stage.cpp
src/upstream_path.cpp
src/upstream_policy.cpp)
add_custom_target(libcaf_core) add_custom_target(libcaf_core)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_ABSTRACT_DOWNSTREAM_HPP
#define CAF_ABSTRACT_DOWNSTREAM_HPP
#include <set>
#include <vector>
#include <memory>
#include <numeric>
#include <unordered_map>
#include "caf/fwd.hpp"
#include "caf/atom.hpp"
#include "caf/optional.hpp"
#include "caf/stream_id.hpp"
namespace caf {
class abstract_downstream {
public:
using topics = std::vector<atom_value>;
/// A set of downstream-defined topic subscriptions.
using topics_set = std::set<topics>;
using path = downstream_path;
using path_cref = const path&;
/// A unique pointer to a downstream path.
using path_uptr = std::unique_ptr<path>;
/// A raw pointer to a downstream path.
using path_ptr = path*;
/// Stores all available paths.
using path_list = std::vector<path_uptr>;
/// List of views to paths.
using path_ptr_list = std::vector<path_ptr>;
/// Stores all available paths sorted by topics.
using path_map = std::unordered_map<topics, path_ptr_list>;
using policy_ptr = std::unique_ptr<downstream_policy>;
abstract_downstream(local_actor* selfptr, const stream_id& sid,
policy_ptr ptr);
virtual ~abstract_downstream();
/// Returns the total available credit for all sinks in O(n).
size_t total_credit() const;
/// Returns the maximum credit of all sinks in O(n).
size_t max_credit() const;
/// Returns the minimal credit of all sinks in O(n).
size_t min_credit() const;
/// Broadcasts the first `min_credit()` elements of the buffer on all paths
/// for all topics.
virtual void broadcast() = 0;
/// Sends `total_credit()` elements of the buffer to any available path
/// on any topic.
virtual void anycast() = 0;
/// Returns the size of the output buffer.
virtual size_t buf_size() const = 0;
bool add_path(strong_actor_ptr ptr, std::vector<atom_value> filter,
bool redeployable);
bool remove_path(strong_actor_ptr& ptr);
/// Removes all paths.
void close();
/// Sends an abort message to all paths and closes the stream.
void abort(strong_actor_ptr& cause, const error& reason);
optional<path&> find(const strong_actor_ptr& ptr) const;
inline local_actor* self() const {
return self_;
}
inline downstream_policy& policy() const {
return *policy_;
}
inline const stream_id& sid() const {
return sid_;
}
inline bool closed() const {
return paths_.empty();
}
protected:
void recalculate_active_filters();
void send_batch(downstream_path& dest, size_t chunk_size, message chunk);
/// Sorts `paths_` in descending order by available credit.
void sort_by_credit();
template <class F>
size_t fold_paths(size_t init, F f) const {
auto b = paths_.begin();
auto e = paths_.end();
auto g = [&](size_t x, const path_uptr& y) {
return f(x, *y);
};
return b != e ? std::accumulate(b, e, init, g) : 0;
}
local_actor* self_;
stream_id sid_;
path_list paths_;
std::unique_ptr<downstream_policy> policy_;
topics_set active_filters_;
};
} // namespace caf
#endif // CAF_ABSTRACT_DOWNSTREAM_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_ABSTRACT_UPSTREAM_HPP
#define CAF_ABSTRACT_UPSTREAM_HPP
#include <vector>
#include <memory>
#include <unordered_map>
#include "caf/fwd.hpp"
#include "caf/upstream_path.hpp"
#include "caf/upstream_policy.hpp"
namespace caf {
class abstract_upstream {
public:
using path = upstream_path;
using path_cref = const path&;
using path_uptr = std::unique_ptr<path>;
using path_ptr = path*;
using path_list = std::vector<path_uptr>;
using path_ptr_list = std::vector<path_ptr>;
/// Stores available paths sorted by priority.
using path_map = std::unordered_map<stream_priority, path_ptr_list>;
using policy_ptr = std::unique_ptr<upstream_policy>;
/// @pre `self != nullptr`
/// @pre `policy != nullptr`
abstract_upstream(local_actor* self, policy_ptr policy);
virtual ~abstract_upstream();
void abort(strong_actor_ptr& cause, const error& reason);
error pull(strong_actor_ptr& hdl, size_t n);
error pull(size_t n);
expected<size_t> add_path(strong_actor_ptr hdl, const stream_id& sid,
stream_priority prio);
bool remove_path(const strong_actor_ptr& hdl);
inline local_actor* self() const {
return self_;
}
/// Queries whether all upstream paths were closed.
inline bool closed() const {
return paths_.empty();
}
protected:
local_actor* self_;
path_list paths_;
policy_ptr policy_;
};
} // namespace caf
#endif // CAF_ABSTRACT_UPSTREAM_HPP
...@@ -34,6 +34,7 @@ ...@@ -34,6 +34,7 @@
#include "caf/actor_factory.hpp" #include "caf/actor_factory.hpp"
#include "caf/is_typed_actor.hpp" #include "caf/is_typed_actor.hpp"
#include "caf/type_erased_value.hpp" #include "caf/type_erased_value.hpp"
#include "caf/named_actor_config.hpp"
#include "caf/detail/safe_equal.hpp" #include "caf/detail/safe_equal.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
...@@ -49,6 +50,9 @@ public: ...@@ -49,6 +50,9 @@ public:
using hook_factory_vector = std::vector<hook_factory>; using hook_factory_vector = std::vector<hook_factory>;
template <class K, class V>
using hash_map = std::unordered_map<K, V>;
using module_factory = std::function<actor_system::module* (actor_system&)>; using module_factory = std::function<actor_system::module* (actor_system&)>;
using module_factory_vector = std::vector<module_factory>; using module_factory_vector = std::vector<module_factory>;
...@@ -77,6 +81,8 @@ public: ...@@ -77,6 +81,8 @@ public:
// -- nested classes --------------------------------------------------------- // -- nested classes ---------------------------------------------------------
using named_actor_config_map = hash_map<std::string, named_actor_config>;
class opt_group { class opt_group {
public: public:
opt_group(option_vector& xs, const char* category); opt_group(option_vector& xs, const char* category);
...@@ -294,6 +300,9 @@ public: ...@@ -294,6 +300,9 @@ public:
// -- utility for caf-run ---------------------------------------------------- // -- utility for caf-run ----------------------------------------------------
// Config parameter for individual actor types.
named_actor_config_map named_actor_configs;
int (*slave_mode_fun)(actor_system&, const actor_system_config&); int (*slave_mode_fun)(actor_system&, const actor_system_config&);
protected: protected:
......
...@@ -36,6 +36,7 @@ ...@@ -36,6 +36,7 @@
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/others.hpp" #include "caf/others.hpp"
#include "caf/result.hpp" #include "caf/result.hpp"
#include "caf/stream.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/behavior.hpp" #include "caf/behavior.hpp"
......
...@@ -64,6 +64,11 @@ public: ...@@ -64,6 +64,11 @@ public:
elements_.emplace_back(std::move(what)); elements_.emplace_back(std::move(what));
} }
template <class... Ts>
inline void emplace_back(Ts&&... xs) {
elements_.emplace_back(std::forward<Ts>(xs)...);
}
inline void cleanup() { inline void cleanup() {
erased_elements_.clear(); erased_elements_.clear();
} }
......
...@@ -31,7 +31,8 @@ namespace detail { ...@@ -31,7 +31,8 @@ namespace detail {
template <class F, template <class F,
int Args = tl_size<typename get_callable_trait<F>::arg_types>::value> int Args = tl_size<typename get_callable_trait<F>::arg_types>::value>
struct functor_attachable : attachable { struct functor_attachable : attachable {
static_assert(Args == 1, "Only 0 or 1 arguments for F are supported"); static_assert(Args == 1 || Args == 2,
"Only 0, 1 or 2 arguments for F are supported");
F functor_; F functor_;
functor_attachable(F arg) : functor_(std::move(arg)) { functor_attachable(F arg) : functor_(std::move(arg)) {
// nop // nop
...@@ -42,6 +43,18 @@ struct functor_attachable : attachable { ...@@ -42,6 +43,18 @@ struct functor_attachable : attachable {
static constexpr size_t token_type = attachable::token::anonymous; static constexpr size_t token_type = attachable::token::anonymous;
}; };
template <class F>
struct functor_attachable<F, 2> : attachable {
F functor_;
functor_attachable(F arg) : functor_(std::move(arg)) {
// nop
}
void actor_exited(const error& x, execution_unit* y) override {
functor_(x, y);
}
};
template <class F> template <class F>
struct functor_attachable<F, 0> : attachable { struct functor_attachable<F, 0> : attachable {
F functor_; F functor_;
......
...@@ -70,6 +70,9 @@ struct tl_head<type_list<T0, Ts...>> { ...@@ -70,6 +70,9 @@ struct tl_head<type_list<T0, Ts...>> {
using type = T0; using type = T0;
}; };
template <class List>
using tl_head_t = typename tl_head<List>::type;
// type_list tail(type_list) // type_list tail(type_list)
/// Gets the tail of `List`. /// Gets the tail of `List`.
...@@ -377,7 +380,7 @@ struct tl_reverse { ...@@ -377,7 +380,7 @@ struct tl_reverse {
using type = typename tl_reverse_impl<List>::type; using type = typename tl_reverse_impl<List>::type;
}; };
// bool find(list, type) // type find(list, predicate)
/// Finds the first element of type `What` beginning at index `Pos`. /// Finds the first element of type `What` beginning at index `Pos`.
template <template <class> class Pred, class... Ts> template <template <class> class Pred, class... Ts>
...@@ -407,6 +410,9 @@ struct tl_find<type_list<Ts...>, Pred> { ...@@ -407,6 +410,9 @@ struct tl_find<type_list<Ts...>, Pred> {
using type = typename tl_find_impl<Pred, Ts...>::type; using type = typename tl_find_impl<Pred, Ts...>::type;
}; };
template <class List, template <class> class Pred>
using tl_find_t = typename tl_find<List, Pred>::type;
// bool forall(predicate) // bool forall(predicate)
/// Tests whether a predicate holds for all elements of a list. /// Tests whether a predicate holds for all elements of a list.
...@@ -592,6 +598,9 @@ struct tl_map<type_list<Ts...>, Funs...> { ...@@ -592,6 +598,9 @@ struct tl_map<type_list<Ts...>, Funs...> {
using type = type_list<typename tl_apply_all<Ts, Funs...>::type...>; using type = type_list<typename tl_apply_all<Ts, Funs...>::type...>;
}; };
template <class List, template <class> class... Funs>
using tl_map_t = typename tl_map<List, Funs...>::type;
/// Creates a new list by applying a `Fun` to each element which /// Creates a new list by applying a `Fun` to each element which
/// returns `TraitResult` for `Trait`. /// returns `TraitResult` for `Trait`.
template <class List, template <class> class Trait, bool TRes, template <class List, template <class> class Trait, bool TRes,
...@@ -762,6 +771,9 @@ struct tl_filter_type<type_list<T...>, Type> { ...@@ -762,6 +771,9 @@ struct tl_filter_type<type_list<T...>, Type> {
>::type; >::type;
}; };
template <class List, class T>
using tl_filter_type_t = typename tl_filter_type<List, T>::type;
/// Creates a new list containing all elements which /// Creates a new list containing all elements which
/// are not equal to `Type`. /// are not equal to `Type`.
template <class List, class Type> template <class List, class Type>
......
...@@ -33,9 +33,6 @@ namespace caf { // forward declarations ...@@ -33,9 +33,6 @@ namespace caf { // forward declarations
template <class... Ts> template <class... Ts>
class typed_actor; class typed_actor;
template <class R>
class typed_continue_helper;
} // namespace caf } // namespace caf
namespace caf { namespace caf {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_DOWNSTREAM_HPP
#define CAF_DOWNSTREAM_HPP
#include <deque>
#include <vector>
#include "caf/stream_msg.hpp"
#include "caf/downstream_path.hpp"
#include "caf/abstract_downstream.hpp"
namespace caf {
template <class T>
class downstream final : public abstract_downstream {
public:
/// Topic filters defined by a downstream actor.
using topics = abstract_downstream::topics;
using topics_set = abstract_downstream::topics_set;
/// A chunk of data for sending batches downstream.
using chunk = std::vector<T>;
/// Chunks split into multiple categories via filters.
using categorized_chunks = std::unordered_map<topics, chunk>;
downstream(local_actor* ptr, const stream_id& sid,
typename abstract_downstream::policy_ptr policy)
: abstract_downstream(ptr, sid, std::move(policy)) {
// nop
}
template <class... Ts>
void push(Ts&&... xs) {
buf_.emplace_back(std::forward<Ts>(xs)...);
}
std::deque<T>& buf() {
return buf_;
}
const std::deque<T>& buf() const {
return buf_;
}
void broadcast() override {
auto chunk = get_chunk(min_credit());
auto csize = chunk.size();
if (csize == 0)
return;
auto wrapped_chunk = make_message(std::move(chunk));
for (auto& x : paths_) {
x->open_credit -= csize;
send_batch(*x, csize, wrapped_chunk);
}
}
void anycast() override {
sort_by_credit();
for (auto& x : paths_) {
auto chunk = get_chunk(x->open_credit);
auto csize = chunk.size();
if (csize == 0)
return;
x->open_credit -= csize;
auto wrapped_chunk = make_message(std::move(chunk));
send_batch(*x, csize, std::move(wrapped_chunk));
}
}
size_t buf_size() const override {
return buf_.size();
}
private:
/// Returns the minimum of `desired` and `buf_.size()`.
size_t clamp_chunk_size(size_t desired) const {
return std::min(desired, buf_.size());
}
using scratch_space_value = std::tuple<T, bool, atom_value>;
using scratch_space = std::vector<scratch_space_value>;
/// Iterator over a `scratch_space` that behaves like a move
/// iterator over `vector<T>`.
struct scratch_space_move_iter {
typename scratch_space::iterator i;
scratch_space_move_iter& operator++() {
++i;
return *this;
}
T&& operator*() {
return std::move(std::get<0>(*i));
}
bool operator!=(const scratch_space_move_iter& x) {
return i != x.i;
}
};
/// @pre `n <= buf_.size()`
chunk get_chunk(size_t n) {
chunk xs;
if (n > 0) {
xs.reserve(n);
if (n < buf_.size()) {
auto first = buf_.begin();
auto last = first + static_cast<ptrdiff_t>(n);
std::move(first, last, std::back_inserter(xs));
buf_.erase(first, last);
} else {
std::move(buf_.begin(), buf_.end(), std::back_inserter(xs));
buf_.clear();
}
}
return xs;
}
/// @pre `n <= buf_.size()`
chunk get_chunk(size_t n, const topics& filter) {
if (filter.empty())
return get_chunk(n);
chunk xs;
if (n > 0) {
xs.reserve(n);
auto in_filter = [&](atom_value x) {
auto e = filter.end();
return std::find(filter.begin(), e, x) != e;
};
// temporarily stores a T with the result of in_filter()
scratch_space ys;
move_from_buf_until_nth_match(ys, n, in_filter);
auto is_marked = [](const scratch_space_value& x) {
return std::get<1>(x);
};
// helper for iterating over ys as if using a move iterator
// over vector<T>
// parition range into result messages and message to move back into buf_
scratch_space_move_iter first{ys.begin()};
scratch_space_move_iter last{ys.end()};
scratch_space_move_iter pp{std::stable_partition(first.i, last.i,
is_marked)};
xs.insert(xs.end(), first, pp);
buf_.insert(buf_.begin(), pp, last);
}
return std::move(xs);
}
categorized_chunks get_chunks(size_t n, const topics_set& filters) {
categorized_chunks res;
if (filters.empty()) {
res.emplace(topics{}, get_chunk(n));
} else if (filters.size() == 1) {
res.emplace(*filters.begin(), get_chunk(n, *filters.begin()));
} else {
// reserve sufficient space for chunks
for (auto& filter : filters) {
chunk tmp;
tmp.reserve(n);
res.emplace(filter, std::move(tmp));
}
// get up to n elements from buffer
auto in_filter = [](const topics_set& filter, atom_value x) {
auto e = filter.end();
return filter.empty() || std::find(filter.begin(), e, x) != e;
};
auto in_any_filter = [&](atom_value x) {
for (auto& filter : filters)
if (in_filter(filter, x))
return true;
return false;
};
scratch_space ys;
move_from_buf_until_nth_match(ys, n, in_any_filter);
// parition range into result messages and message to move back into buf_
auto is_marked = [](const scratch_space_value& x) {
return std::get<1>(x);
};
auto first = ys.begin();
auto last = ys.end();
auto pp = std::stable_partition(first, last, is_marked);
// copy matched values into according chunks
for (auto& r : res)
for (auto& y : ys)
if (in_filter(r.first, std::get<2>(y)))
r.second.emplace_back(std::get<0>(y));
// move unused items back into place
buf_.insert(buf_.begin(), scratch_space_move_iter{pp},
scratch_space_move_iter{last});
}
return res;
}
// moves elements from `buf_` to `tmp` until n elements matching the
// predicate were moved or the buffer was fully consumed.
// @pre `n > 0`
template <class F>
void move_from_buf_until_nth_match(scratch_space& ys, size_t n, F pred) {
CAF_ASSERT(n > 0);
size_t included = 0; // nr. of included elements
for (auto i = buf_.begin(); i != buf_.end(); ++i) {
auto topic = this->policy().categorize(*i);
auto match = pred(topic);
ys.emplace_back(std::move(*i), match, topic);
if (match && ++included == n) {
buf_.erase(buf_.begin(), i + 1);
return;
}
}
// consumed whole buffer
buf_.clear();
}
std::deque<T> buf_;
};
} // namespace caf
#endif // CAF_DOWNSTREAM_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_DOWNSTREAM_PATH_HPP
#define CAF_DOWNSTREAM_PATH_HPP
#include <deque>
#include <vector>
#include <cstdint>
#include <cstddef>
#include "caf/fwd.hpp"
#include "caf/stream_msg.hpp"
namespace caf {
/// Denotes a downstream actor in a stream topology. All downstream actors use
/// the stream ID registered with the hosting downstream object.
class downstream_path {
public:
using topics = std::vector<atom_value>;
/// Handle to the downstream actor.
strong_actor_ptr ptr;
/// Next expected batch ID.
int64_t next_batch_id;
/// Currently available credit for this path.
size_t open_credit;
/// Subscribed topics on this path (empty for all).
topics filter;
/// Stores whether the downstream actor is failsafe, i.e., allows the runtime
/// to redeploy it on failure. If this field is set to `false` then
/// `unacknowledged_batches` is unused.
bool redeployable;
/// Caches batches until receiving an ACK.
std::deque<std::pair<int64_t, stream_msg::batch>> unacknowledged_batches;
downstream_path(strong_actor_ptr p, topics ts, bool redeploy);
};
} // namespace caf
#endif // CAF_DOWNSTREAM_PATH_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_DOWNSTREAM_POLICY_HPP
#define CAF_DOWNSTREAM_POLICY_HPP
#include <cstddef>
#include "caf/fwd.hpp"
namespace caf {
/// Implements dispatching to downstream paths.
class downstream_policy {
public:
virtual ~downstream_policy();
/// Returns the topic name for `x`. The default implementation returns an
/// empty atom value, indicating a stream without topics.
virtual atom_value categorize(type_erased_value& x) const;
/// Queries the optimal amount of data for the next `push` operation to `x`.
virtual size_t desired_buffer_size(const abstract_downstream& x) = 0;
/// Pushes data to the downstream paths of `x`.
virtual void push(abstract_downstream& x) = 0;
// TODO: callbacks für new and closed downstream pahts
/// Reclaim credit of closed downstream `y`.
//virtual void reclaim(abstract_downstream& x, downstream_path& y) = 0;
};
} // namespace caf
#endif // CAF_DOWNSTREAM_POLICY_HPP
...@@ -28,13 +28,15 @@ namespace caf { ...@@ -28,13 +28,15 @@ namespace caf {
// -- 1 param templates -------------------------------------------------------- // -- 1 param templates --------------------------------------------------------
template <class> class param; template <class> class param;
template <class> class stream;
template <class> class optional; template <class> class optional;
template <class> class expected; template <class> class expected;
template <class> class upstream;
template <class> class downstream;
template <class> class intrusive_ptr; template <class> class intrusive_ptr;
template <class> class behavior_type_of; template <class> class behavior_type_of;
template <class> class trivial_match_case; template <class> class trivial_match_case;
template <class> class weak_intrusive_ptr; template <class> class weak_intrusive_ptr;
template <class> class typed_continue_helper;
template <class> struct timeout_definition; template <class> struct timeout_definition;
...@@ -54,41 +56,52 @@ template <class...> class typed_event_based_actor; ...@@ -54,41 +56,52 @@ template <class...> class typed_event_based_actor;
// -- classes ------------------------------------------------------------------ // -- classes ------------------------------------------------------------------
class actor; class actor;
class group;
class error; class error;
class group;
class message; class message;
class node_id; class node_id;
class duration;
class behavior; class behavior;
class duration;
class resumable; class resumable;
class actor_addr; class actor_addr;
class actor_pool; class actor_pool;
class message_id; class message_id;
class serializer; class serializer;
class actor_proxy; class actor_proxy;
class ref_counted;
class local_actor; class local_actor;
class ref_counted;
class stream_sink;
class actor_config; class actor_config;
class actor_system; class actor_system;
class deserializer; class deserializer;
class group_module; class group_module;
class message_view; class message_view;
class scoped_actor; class scoped_actor;
class stream_stage;
class stream_source;
class upstream_path;
class abstract_actor; class abstract_actor;
class abstract_group; class abstract_group;
class actor_registry; class actor_registry;
class blocking_actor; class blocking_actor;
class execution_unit; class execution_unit;
class proxy_registry; class proxy_registry;
class stream_handler;
class upstream_policy;
class actor_companion; class actor_companion;
class continue_helper; class continue_helper;
class downstream_path;
class mailbox_element; class mailbox_element;
class message_handler; class message_handler;
class scheduled_actor; class scheduled_actor;
class response_promise; class response_promise;
class abstract_upstream;
class downstream_policy;
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 abstract_downstream;
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;
...@@ -99,6 +112,8 @@ class forwarding_actor_proxy; ...@@ -99,6 +112,8 @@ class forwarding_actor_proxy;
struct unit_t; struct unit_t;
struct exit_msg; struct exit_msg;
struct down_msg; struct down_msg;
struct stream_id;
struct stream_msg;
struct timeout_msg; struct timeout_msg;
struct group_down_msg; struct group_down_msg;
struct invalid_actor_t; struct invalid_actor_t;
...@@ -108,6 +123,7 @@ struct prohibit_top_level_spawn_marker; ...@@ -108,6 +123,7 @@ struct prohibit_top_level_spawn_marker;
// -- enums -------------------------------------------------------------------- // -- enums --------------------------------------------------------------------
enum class stream_priority;
enum class atom_value : uint64_t; enum class atom_value : uint64_t;
// -- aliases ------------------------------------------------------------------ // -- aliases ------------------------------------------------------------------
......
...@@ -20,7 +20,9 @@ ...@@ -20,7 +20,9 @@
#ifndef CAF_INTRUSIVE_PTR_HPP #ifndef CAF_INTRUSIVE_PTR_HPP
#define CAF_INTRUSIVE_PTR_HPP #define CAF_INTRUSIVE_PTR_HPP
#include <string>
#include <cstddef> #include <cstddef>
#include <cinttypes>
#include <algorithm> #include <algorithm>
#include <stdexcept> #include <stdexcept>
#include <type_traits> #include <type_traits>
...@@ -230,6 +232,16 @@ bool operator<(const intrusive_ptr<T>& x, const intrusive_ptr<T>& y) { ...@@ -230,6 +232,16 @@ bool operator<(const intrusive_ptr<T>& x, const intrusive_ptr<T>& y) {
return x.get() < y.get(); return x.get() < y.get();
} }
template <class T>
std::string to_string(const intrusive_ptr<T>& x) {
auto v = reinterpret_cast<uintptr_t>(x.get());
// we convert to hex representation, i.e.,
// one byte takes two characters + null terminator + "0x" prefix
char buf[sizeof(v) * 2 + 3];
sprintf(buf, "%" PRIxPTR, v);
return buf;
}
} // namespace caf } // namespace caf
#endif // CAF_INTRUSIVE_PTR_HPP #endif // CAF_INTRUSIVE_PTR_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_MIXIN_HAS_DOWNSTREAMS_HPP
#define CAF_MIXIN_HAS_DOWNSTREAMS_HPP
#include <cstddef>
#include "caf/sec.hpp"
#include "caf/actor_control_block.hpp"
namespace caf {
namespace mixin {
/// Mixin for streams with any number of downstreams.
template <class Base, class Subtype>
class has_downstreams : public Base {
public:
using topics = typename Base::topics;
error add_downstream(strong_actor_ptr& ptr, const topics& filter,
size_t init, bool is_redeployable) final {
if (!ptr)
return sec::invalid_argument;
if (out().add_path(ptr, filter, is_redeployable)) {
dptr()->downstream_demand(ptr, init);
return none;
}
return sec::downstream_already_exists;
}
error trigger_send(strong_actor_ptr&) {
if (out().buf_size() > 0)
out().policy().push(out());
return none;
}
private:
Subtype* dptr() {
return static_cast<Subtype*>(this);
}
abstract_downstream& out() {
return dptr()->out();
}
};
} // namespace mixin
} // namespace caf
#endif // CAF_MIXIN_HAS_DOWNSTREAMS_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_MIXIN_HAS_UPSTREAMS_HPP
#define CAF_MIXIN_HAS_UPSTREAMS_HPP
#include "caf/sec.hpp"
#include "caf/expected.hpp"
#include "caf/stream_id.hpp"
#include "caf/stream_priority.hpp"
#include "caf/actor_control_block.hpp"
namespace caf {
namespace mixin {
/// Mixin for streams with has number of upstream actors.
template <class Base, class Subtype>
class has_upstreams : public Base {
public:
expected<size_t> add_upstream(strong_actor_ptr& ptr, const stream_id& sid,
stream_priority prio) final {
if (!ptr)
return sec::invalid_argument;
return in().add_path(ptr, sid, prio);
}
error close_upstream(strong_actor_ptr& ptr) final {
if (in().remove_path(ptr)) {
if (in().closed())
dptr()->last_upstream_closed();
return none;
}
return sec::invalid_upstream;
}
error pull(size_t amount) final {
return in().pull(amount);
}
private:
Subtype* dptr() {
return static_cast<Subtype*>(this);
}
abstract_upstream& in() {
return dptr()->in();
}
};
} // namespace mixin
} // namespace caf
#endif // CAF_MIXIN_HAS_UPSTREAMS_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_POLICY_ANYCAST_HPP
#define CAF_POLICY_ANYCAST_HPP
#include "caf/downstream_policy.hpp"
namespace caf {
namespace policy {
class anycast final : public downstream_policy {
public:
void push(abstract_downstream& out) override;
size_t desired_buffer_size(const abstract_downstream& out) override;
};
} // namespace policy
} // namespace caf
#endif // CAF_POLICY_ANYCAST_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_POLICY_BROADCAST_HPP
#define CAF_POLICY_BROADCAST_HPP
#include "caf/downstream_policy.hpp"
namespace caf {
namespace policy {
class broadcast final : public downstream_policy {
public:
void push(abstract_downstream& out) override;
size_t desired_buffer_size(const abstract_downstream& out) override;
};
} // namespace policy
} // namespace caf
#endif // CAF_POLICY_BROADCAST_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_POLICY_GREEDY_HPP
#define CAF_POLICY_GREEDY_HPP
#include "caf/upstream_policy.hpp"
namespace caf {
namespace policy {
/// Sends ACKs as early and often as possible.
class greedy final : public upstream_policy {
public:
size_t initial_credit(abstract_upstream& x, upstream_path& y) override;
void reclaim(abstract_upstream& x, upstream_path& y) override;
};
} // namespace policy
} // namespace caf
#endif // CAF_POLICY_GREEDY_HPP
...@@ -28,7 +28,6 @@ ...@@ -28,7 +28,6 @@
#include "caf/typed_behavior.hpp" #include "caf/typed_behavior.hpp"
#include "caf/continue_helper.hpp" #include "caf/continue_helper.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/typed_continue_helper.hpp"
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
#include "caf/detail/typed_actor_util.hpp" #include "caf/detail/typed_actor_util.hpp"
......
...@@ -136,6 +136,12 @@ public: ...@@ -136,6 +136,12 @@ public:
error err; error err;
}; };
template <class T>
struct is_result : std::false_type {};
template <class... Ts>
struct is_result<result<Ts...>> : std::true_type {};
} // namespace caf } // namespace caf
#endif // CAF_RESULT_HPP #endif // CAF_RESULT_HPP
...@@ -35,6 +35,13 @@ ...@@ -35,6 +35,13 @@
#include "caf/actor_marker.hpp" #include "caf/actor_marker.hpp"
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/scheduled_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/stream_sink_impl.hpp"
#include "caf/stream_stage_impl.hpp"
#include "caf/stream_source_impl.hpp"
#include "caf/policy/greedy.hpp"
#include "caf/policy/anycast.hpp"
#include "caf/policy/broadcast.hpp"
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
...@@ -285,6 +292,111 @@ public: ...@@ -285,6 +292,111 @@ public:
} }
# endif // CAF_NO_EXCEPTIONS # endif // CAF_NO_EXCEPTIONS
// -- stream management ------------------------------------------------------
/// Adds a stream source to this actor.
template <class Init, class Getter, class ClosedPredicate>
stream<typename stream_source_trait_t<Getter>::output>
add_source(Init init, Getter getter, ClosedPredicate pred) {
CAF_ASSERT(current_mailbox_element() != nullptr);
using type = typename stream_source_trait_t<Getter>::output;
using state_type = typename stream_source_trait_t<Getter>::state;
static_assert(std::is_same<
void (state_type&),
typename detail::get_callable_trait<Init>::fun_sig
>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
bool (const state_type&),
typename detail::get_callable_trait<ClosedPredicate>::fun_sig
>::value,
"Expected signature `bool (const State&)` for "
"closed_predicate function");
if (current_mailbox_element()->stages.empty()) {
CAF_LOG_ERROR("cannot create a stream data source without downstream");
auto rp = make_response_promise();
rp.deliver(sec::no_downstream_stages_defined);
return stream_id{nullptr, 0};
}
stream_id sid{ctrl(),
new_request_id(message_priority::normal).integer_value()};
fwd_stream_handshake<type>(sid);
using impl = stream_source_impl<Getter, ClosedPredicate>;
std::unique_ptr<downstream_policy> p{new policy::anycast};
auto ptr = make_counted<impl>(this, sid, std::move(p), std::move(getter),
std::move(pred));
init(ptr->state());
streams_.emplace(std::move(sid), std::move(ptr));
return sid;
}
/// Adds a stream stage to this actor.
template <class In, class Init, class Fun, class Cleanup>
stream<typename stream_stage_trait_t<Fun>::output>
add_stage(stream<In>& in, Init init, Fun fun, Cleanup cleanup) {
CAF_ASSERT(current_mailbox_element() != nullptr);
using output_type = typename stream_stage_trait_t<Fun>::output;
using state_type = typename stream_stage_trait_t<Fun>::state;
static_assert(std::is_same<
void (state_type&),
typename detail::get_callable_trait<Init>::fun_sig
>::value,
"Expected signature `void (State&)` for init function");
if (current_mailbox_element()->stages.empty()) {
CAF_LOG_ERROR("cannot create a stream data source without downstream");
return stream_id{nullptr, 0};
}
auto sid = in.id();
fwd_stream_handshake<output_type>(in.id());
using impl = stream_stage_impl<Fun, Cleanup>;
std::unique_ptr<downstream_policy> dptr{new policy::anycast};
std::unique_ptr<upstream_policy> uptr{new policy::greedy};
auto ptr = make_counted<impl>(this, sid, std::move(uptr), std::move(dptr),
std::move(fun), std::move(cleanup));
init(ptr->state());
streams_.emplace(sid, std::move(ptr));
return std::move(sid);
}
/// Adds a stream sink to this actor.
template <class In, class Init, class Fun, class Finalize>
result<typename stream_sink_trait_t<Fun, Finalize>::output>
add_sink(stream<In>& in, Init init, Fun fun, Finalize finalize) {
CAF_ASSERT(current_mailbox_element() != nullptr);
delegated<typename stream_sink_trait_t<Fun, Finalize>::output> dummy_res;
//using output_type = typename stream_sink_trait_t<Fun, Finalize>::output;
using state_type = typename stream_sink_trait_t<Fun, Finalize>::state;
static_assert(std::is_same<
void (state_type&),
typename detail::get_callable_trait<Init>::fun_sig
>::value,
"Expected signature `void (State&)` for init function");
static_assert(std::is_same<
void (state_type&, In),
typename detail::get_callable_trait<Fun>::fun_sig
>::value,
"Expected signature `void (State&, Input)` "
"for consume function");
auto mptr = current_mailbox_element();
if (!mptr) {
CAF_LOG_ERROR("add_sink called outside of a message handler");
return dummy_res;
}
using impl = stream_sink_impl<Fun, Finalize>;
std::unique_ptr<upstream_policy> p{new policy::greedy};
auto ptr = make_counted<impl>(this, std::move(p), std::move(mptr->sender),
std::move(mptr->stages), mptr->mid,
std::move(fun), std::move(finalize));
init(ptr->state());
streams_.emplace(in.id(), std::move(ptr));
return dummy_res;
}
inline std::unordered_map<stream_id, intrusive_ptr<stream_handler>>&
streams() {
return streams_;
}
/// @cond PRIVATE /// @cond PRIVATE
// -- timeout management ----------------------------------------------------- // -- timeout management -----------------------------------------------------
...@@ -353,6 +465,10 @@ public: ...@@ -353,6 +465,10 @@ public:
/// @returns `true` if cleanup code was called, `false` otherwise. /// @returns `true` if cleanup code was called, `false` otherwise.
bool finalize(); bool finalize();
inline detail::behavior_stack& bhvr_stack() {
return bhvr_stack_;
}
/// @endcond /// @endcond
protected: protected:
...@@ -388,6 +504,24 @@ protected: ...@@ -388,6 +504,24 @@ protected:
swap(g, f); swap(g, f);
} }
template <class T>
void fwd_stream_handshake(const stream_id& sid) {
auto mptr = current_mailbox_element();
auto& stages = mptr->stages;
CAF_ASSERT(!stages.empty());
CAF_ASSERT(stages.back() != nullptr);
auto next = std::move(stages.back());
stages.pop_back();
stream<T> token{sid};
next->enqueue(make_mailbox_element(
mptr->sender, mptr->mid, std::move(stages),
make<stream_msg::open>(sid, make_message(token), ctrl(),
stream_priority::normal,
std::vector<atom_value>{}, false)),
context());
mptr->mid.mark_as_answered();
}
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Stores user-defined callbacks for message handling. /// Stores user-defined callbacks for message handling.
...@@ -417,6 +551,10 @@ protected: ...@@ -417,6 +551,10 @@ 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.
std::unordered_map<stream_id, intrusive_ptr<stream_handler>> 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_;
......
...@@ -90,6 +90,22 @@ enum class sec : uint8_t { ...@@ -90,6 +90,22 @@ enum class sec : uint8_t {
runtime_error, runtime_error,
/// Linking to a remote actor failed because actor no longer exists. /// Linking to a remote actor failed because actor no longer exists.
remote_linking_failed, remote_linking_failed,
/// Adding an upstream to a stream failed.
cannot_add_upstream,
/// Adding an upstream to a stream failed because it already exists.
upstream_already_exists,
/// Unable to process upstream messages because upstream is invalid.
invalid_upstream,
/// Adding a downstream to a stream failed.
cannot_add_downstream,
/// Adding a downstream to a stream failed because it already exists.
downstream_already_exists,
/// Unable to process downstream messages because downstream is invalid.
invalid_downstream,
/// Cannot start streaming without next stage.
no_downstream_stages_defined,
/// Actor failed to initialize state after receiving a stream handshake.
stream_init_failed,
/// A function view was called without assigning an actor first. /// A function view was called without assigning an actor first.
bad_function_call bad_function_call
}; };
......
...@@ -26,6 +26,7 @@ ...@@ -26,6 +26,7 @@
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/typed_actor.hpp" #include "caf/typed_actor.hpp"
#include "caf/local_actor.hpp"
#include "caf/response_type.hpp" #include "caf/response_type.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/is_message_sink.hpp" #include "caf/is_message_sink.hpp"
...@@ -67,6 +68,33 @@ void send_as(const Source& src, const Dest& dest, Ts&&... xs) { ...@@ -67,6 +68,33 @@ void send_as(const Source& src, const Dest& dest, Ts&&... xs) {
nullptr, std::forward<Ts>(xs)...); nullptr, std::forward<Ts>(xs)...);
} }
template <class Source, class Dest, class... Ts>
void unsafe_send_as(Source* src, const Dest& dest, Ts&&... xs) {
actor_cast<abstract_actor*>(dest)->eq_impl(message_id::make(), src->ctrl(),
src->context(),
std::forward<Ts>(xs)...);
}
template <class... Ts>
void unsafe_response(local_actor* self, strong_actor_ptr src,
std::vector<strong_actor_ptr> stages, message_id mid,
Ts&&... xs) {
strong_actor_ptr next;
if (stages.empty()) {
next = src;
src = self->ctrl();
if (mid.is_request())
mid = mid.response_id();
} else {
next = std::move(stages.back());
stages.pop_back();
}
if (next)
next->enqueue(make_mailbox_element(std::move(src), mid, std::move(stages),
std::forward<Ts>(xs)...),
self->context());
}
/// Anonymously sends `dest` a message. /// Anonymously sends `dest` a message.
template <message_priority P = message_priority::normal, template <message_priority P = message_priority::normal,
class Dest = actor, class... Ts> class Dest = actor, class... Ts>
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_HPP
#define CAF_STREAM_HPP
#include "caf/stream_id.hpp"
#include "caf/meta/type_name.hpp"
namespace caf {
/// Identifies an unbound sequence of messages.
template <class T>
class stream {
public:
stream() = default;
stream(stream&&) = default;
stream(const stream&) = default;
stream& operator=(stream&&) = default;
stream& operator=(const stream&) = default;
stream(stream_id sid) : id_(std::move(sid)) {
// nop
}
inline const stream_id& id() const {
return id_;
}
template <class Inspector>
friend typename Inspector::result_type inspect(Inspector& f, stream& x) {
return f(meta::type_name("stream"), x.id_);
}
private:
stream_id id_;
};
} // namespace caf
#endif // CAF_STREAM_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_HANDLER_HPP
#define CAF_STREAM_HANDLER_HPP
#include <vector>
#include <cstdint>
#include <cstddef>
#include "caf/fwd.hpp"
#include "caf/ref_counted.hpp"
namespace caf {
/// Manages a single stream with any number of down- and upstream actors.
class stream_handler : public ref_counted {
public:
~stream_handler();
// -- member types -----------------------------------------------------------
using topics = std::vector<atom_value>;
// -- handler for downstream events ------------------------------------------
/// Add a new downstream actor to the stream.
/// @param hdl Handle to the new downstream actor.
/// @param filter Subscribed topics (empty for all).
/// @param initial_demand Credit received with `ack_open`.
/// @param redeployable Denotes whether the runtime can redeploy
/// the downstream actor on failure.
/// @pre `hdl != nullptr`
virtual error add_downstream(strong_actor_ptr& hdl, const topics& filter,
size_t initial_demand, bool redeployable);
/// Handles ACK message from a downstream actor.
/// @pre `hdl != nullptr`
virtual error downstream_ack(strong_actor_ptr& hdl, int64_t batch_id);
/// Handles new demand from a downstream actor.
/// @pre `hdl != nullptr`
/// @pre `new_demand > 0`
virtual error downstream_demand(strong_actor_ptr& hdl, size_t new_demand);
/// Push new data to downstream by sending batches. The amount of pushed data
/// is limited by the available credit.
virtual error push();
// -- handler for upstream events --------------------------------------------
/// Add a new upstream actor to the stream and return an initial credit.
virtual expected<size_t> add_upstream(strong_actor_ptr& hdl,
const stream_id& sid,
stream_priority prio);
/// Handles data from an upstream actor.
virtual error upstream_batch(strong_actor_ptr& hdl, size_t, message& xs);
/// Closes an upstream.
virtual error close_upstream(strong_actor_ptr& hdl);
/// Pull up to `n` new data items from upstream by sending a demand message.
virtual error pull(size_t n);
// -- handler for stream-wide events -----------------------------------------
/// Shutdown the stream due to a fatal error.
virtual void abort(strong_actor_ptr& cause, const error& reason) = 0;
// -- accessors --------------------------------------------------------------
virtual bool done() const = 0;
/// Queries whether this handler is a sink, i.e.,
/// does not accept downstream actors.
virtual bool is_sink() const;
/// Queries whether this handler is a hdl, i.e.,
/// does not accepts upstream actors.
virtual bool is_source() const;
/// Returns a type-erased `stream<T>` as handshake token for downstream
/// actors. Returns an empty message for sinks.
virtual message make_output_token(const stream_id&) const;
};
} // namespace caf
#endif // CAF_STREAM_HANDLER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_ID_HPP
#define CAF_STREAM_ID_HPP
#include <cstdint>
#include "caf/meta/type_name.hpp"
#include "caf/actor_control_block.hpp"
namespace caf {
struct stream_id {
strong_actor_ptr origin;
uint64_t nr;
};
inline bool operator==(const stream_id& x, const stream_id& y) {
return x.origin == y.origin && x.nr == y.nr;
}
inline bool operator<(const stream_id& x, const stream_id& y) {
return x.origin == y.origin ? x.nr < y.nr : x.origin < y.origin;
}
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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_MSG_HPP
#define CAF_STREAM_MSG_HPP
#include <vector>
#include <cstdint>
#include "caf/atom.hpp"
#include "caf/message.hpp"
#include "caf/variant.hpp"
#include "caf/stream_id.hpp"
#include "caf/stream_priority.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/tag/boxing_type.hpp"
#include "caf/detail/type_list.hpp"
namespace caf {
/// Stream communication messages for handshaking, ACKing, data transmission,
/// etc.
struct stream_msg : tag::boxing_type {
/// Initiates a stream handshake.
struct open {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = stream_msg;
/// A type-erased stream<T> object for picking the correct message
/// handler of the receiving actor.
message token;
/// A pointer to the previous stage in the pipeline.
strong_actor_ptr prev_stage;
/// Configures the priority for stream elements.
stream_priority priority;
/// Available topics for this stream. An empty vector indicates that the
/// upstream does provide only a single channel for this stream.
std::vector<atom_value> topics;
/// Tells the downstream whether rebindings can occur on this path.
bool redeployable;
};
/// Acknowledges a previous `open` message and finalizes a stream handshake.
/// Also signalizes initial demand.
struct ack_open {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = stream_msg;
/// Grants credit to the source.
int32_t initial_demand;
/// Subscribes to a subset of the stream if non-empty. Otherwise, the
/// upstream sends all data of the stream.
std::vector<atom_value> filter;
/// Tells the upstream whether rebindings can occur on this path.
bool redeployable;
};
/// Transmits stream data.
struct batch {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = stream_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;
};
/// Cumulatively acknowledges received batches and signalizes new demand from
/// a sink to its source.
struct ack_batch {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = stream_msg;
/// Newly available credit.
int32_t new_capacity;
/// Cumulative ack ID.
int64_t acknowledged_id;
};
/// Closes a stream after receiving an ACK for the last batch.
struct close {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = stream_msg;
};
/// Propagates fatal errors.
struct abort {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = stream_msg;
/// Reason for shutting down the stream.
error reason;
};
/// Send by the runtime if a downstream path failed. The receiving actor
/// awaits a `resume` message afterwards if the downstream path was
/// redeployable. Otherwise, this results in a fatal error.
struct downstream_failed {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = stream_msg;
/// Exit reason of the failing downstream path.
error reason;
};
/// Send by the runtime if an upstream path failed. The receiving actor
/// awaits a `resume` message afterwards if the upstream path was
/// redeployable. Otherwise, this results in a fatal error.
struct upstream_failed {
/// Allows the testing DSL to unbox this type automagically.
using outer_type = stream_msg;
/// Exit reason of the failing upstream path.
error reason;
};
/// Lists all possible options for the payload.
using content_alternatives = detail::type_list<open, ack_open, batch,
ack_batch, close, abort,
downstream_failed,
upstream_failed>;
/// Stores one of `content_alternatives`.
using content_type = variant<open, ack_open, batch, ack_batch, close,
abort, downstream_failed, upstream_failed>;
/// ID of the affected stream.
stream_id sid;
/// Palyoad of the message.
content_type content;
template <class T>
stream_msg(const stream_id& id, T&& x)
: sid(id),
content(std::forward<T>(x)) {
// nop
}
stream_msg() = default;
stream_msg(stream_msg&&) = default;
stream_msg(const stream_msg&) = default;
stream_msg& operator=(stream_msg&&) = default;
stream_msg& operator=(const stream_msg&) = default;
};
/// Allows the testing DSL to unbox `stream_msg` automagically.
template <class T>
const T& get(const stream_msg& x) {
return get<T>(x.content);
}
template <class T, class... Ts>
typename std::enable_if<
detail::tl_contains<
stream_msg::content_alternatives,
T
>::value,
stream_msg
>::type
make(const stream_id& sid, Ts&&... xs) {
return {sid, T{std::forward<Ts>(xs)...}};
}
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_msg::open& x) {
return f(meta::type_name("open"), x.token, x.prev_stage, x.priority,
x.topics, x.redeployable);
}
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_msg::ack_open& x) {
return f(meta::type_name("ok"), x.initial_demand, x.filter, x.redeployable);
}
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_msg::batch& x) {
return f(meta::type_name("batch"), meta::omittable(), x.xs_size, x.xs, x.id);
}
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f,
stream_msg::ack_batch& x) {
return f(meta::type_name("demand"), x.new_capacity, x.acknowledged_id);
}
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_msg::close&) {
return f(meta::type_name("close"));
}
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_msg::abort& x) {
return f(meta::type_name("abort"), x.reason);
}
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f,
stream_msg::downstream_failed& x) {
return f(meta::type_name("downstream_failed"), x.reason);
}
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f,
stream_msg::upstream_failed& x) {
return f(meta::type_name("upstream_failed"), x.reason);
}
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_msg& x) {
return f(meta::type_name("stream_msg"), x.sid, x.content);
}
} // namespace caf
#endif // CAF_STREAM_MSG_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_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_handler.hpp"
namespace caf {
class stream_msg_visitor {
public:
using map_type = std::unordered_map<stream_id, intrusive_ptr<stream_handler>>;
using iterator = typename map_type::iterator;
using result_type = std::pair<error, iterator>;
stream_msg_visitor(scheduled_actor* self, stream_id& sid,
iterator pos, iterator end);
result_type operator()(stream_msg::open& x);
result_type operator()(stream_msg::ack_open&);
result_type operator()(stream_msg::batch&);
result_type operator()(stream_msg::ack_batch&);
result_type operator()(stream_msg::close&);
result_type operator()(stream_msg::abort&);
result_type operator()(stream_msg::downstream_failed&);
result_type operator()(stream_msg::upstream_failed&);
private:
scheduled_actor* self_;
stream_id& sid_;
iterator i_;
iterator e_;
};
} // namespace caf
#endif // CAF_STREAM_MSG_VISITOR_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_PRIORITY_HPP
#define CAF_STREAM_PRIORITY_HPP
#include <string>
namespace caf {
/// Categorizes individual streams.
enum class stream_priority {
/// Denotes best-effort traffic.
low,
/// Denotes traffic with moderate timing requirements.
normal,
/// Denotes soft-realtime traffic.
high
};
std::string to_string(stream_priority x);
} // namespace caf
#endif // CAF_STREAM_PRIORITY_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SINK_HPP
#define CAF_STREAM_SINK_HPP
#include <vector>
#include "caf/extend.hpp"
#include "caf/message_id.hpp"
#include "caf/stream_handler.hpp"
#include "caf/upstream_policy.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/mixin/has_upstreams.hpp"
namespace caf {
/// Represents a terminal stream stage.
class stream_sink : public extend<stream_handler, stream_sink>::
with<mixin::has_upstreams> {
public:
stream_sink(abstract_upstream* in_ptr, strong_actor_ptr&& orig_sender,
std::vector<strong_actor_ptr>&& trailing_stages, message_id mid);
bool done() const;
error upstream_batch(strong_actor_ptr& src, size_t xs_size,
message& xs) final;
void abort(strong_actor_ptr& cause, const error& reason) final;
void last_upstream_closed();
inline abstract_upstream& in() {
return *in_ptr_;
}
protected:
/// Consumes a batch.
virtual error consume(message& xs) = 0;
/// Computes the final result after consuming all batches of the stream.
virtual message finalize() = 0;
private:
abstract_upstream* in_ptr_;
strong_actor_ptr original_sender_;
std::vector<strong_actor_ptr> next_stages_;
message_id original_msg_id_;
};
} // namespace caf
#endif // CAF_STREAM_SINK_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SINK_IMPL_HPP
#define CAF_STREAM_SINK_IMPL_HPP
#include "caf/upstream.hpp"
#include "caf/stream_sink.hpp"
#include "caf/stream_sink_trait.hpp"
namespace caf {
template <class Fun, class Finalize>
class stream_sink_impl : public stream_sink {
public:
using trait = stream_sink_trait_t<Fun, Finalize>;
using state_type = typename trait::state;
using input_type = typename trait::input;
using output_type = typename trait::output;
stream_sink_impl(local_actor* self, std::unique_ptr<upstream_policy> policy,
strong_actor_ptr&& orig_sender,
std::vector<strong_actor_ptr>&& svec, message_id mid,
Fun fun, Finalize fin)
: stream_sink(&in_, std::move(orig_sender), std::move(svec), mid),
fun_(std::move(fun)),
fin_(std::move(fin)),
in_(self, std::move(policy)) {
// nop
}
error consume(message& msg) final {
using vec_type = std::vector<output_type>;
if (msg.match_elements<vec_type>()) {
auto& xs = msg.get_as<vec_type>(0);
for (auto& x : xs)
fun_(state_, x);
return none;
}
return sec::unexpected_message;
}
message finalize() final {
return make_message(fin_(state_));
}
state_type& state() {
return state_;
}
private:
state_type state_;
Fun fun_;
Finalize fin_;
upstream<output_type> in_;
};
} // namespace caf
#endif // CAF_STREAM_SINK_IMPL_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SINK_TRAIT_HPP
#define CAF_STREAM_SINK_TRAIT_HPP
#include "caf/detail/type_traits.hpp"
namespace caf {
template <class Fun, class Fin>
struct stream_sink_trait;
template <class State, class In, class Out>
struct stream_sink_trait<void (State&, In), Out (State&)> {
using state = State;
using input = In;
using output = Out;
};
template <class Fun, class Fin>
using stream_sink_trait_t =
stream_sink_trait<typename detail::get_callable_trait<Fun>::fun_sig,
typename detail::get_callable_trait<Fin>::fun_sig>;
} // namespace caf
#endif // CAF_STREAM_SINK_TRAIT_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SOURCE_HPP
#define CAF_STREAM_SOURCE_HPP
#include <memory>
#include "caf/extend.hpp"
#include "caf/stream_handler.hpp"
#include "caf/downstream_policy.hpp"
#include "caf/mixin/has_downstreams.hpp"
namespace caf {
class stream_source : public extend<stream_handler, stream_source>::
with<mixin::has_downstreams> {
public:
stream_source(abstract_downstream* out_ptr);
~stream_source();
bool done() const final;
error downstream_demand(strong_actor_ptr& hdl, size_t value) final;
void abort(strong_actor_ptr& cause, const error& reason) final;
inline abstract_downstream& out() {
return *out_ptr_;
}
protected:
/// Queries the current amount of elements in the output buffer.
virtual size_t buf_size() const = 0;
/// Generate new elements for the output buffer. The size hint `n` indicates
/// how much elements can be shipped immediately.
virtual void generate(size_t n) = 0;
/// Queries whether the source is exhausted.
virtual bool at_end() const = 0;
private:
abstract_downstream* out_ptr_;
};
} // namespace caf
#endif // CAF_STREAM_SOURCE_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SOURCE_IMPL_HPP
#define CAF_STREAM_SOURCE_IMPL_HPP
#include "caf/downstream.hpp"
#include "caf/stream_source.hpp"
#include "caf/stream_source_trait.hpp"
namespace caf {
template <class Fun, class Predicate>
class stream_source_impl : public stream_source {
public:
using trait = stream_source_trait_t<Fun>;
using state_type = typename trait::state;
using output_type = typename trait::output;
stream_source_impl(local_actor* self,
const stream_id& sid,
std::unique_ptr<downstream_policy> policy, Fun fun,
Predicate pred)
: stream_source(&out_),
fun_(std::move(fun)),
pred_(std::move(pred)),
out_(self, sid, std::move(policy)) {
// nop
}
void generate(size_t num) final {
fun_(state_, out_, num);
}
size_t buf_size() const final {
return out_.buf().size();
}
bool at_end() const final {
return pred_(state_);
}
state_type& state() {
return state_;
}
private:
state_type state_;
Fun fun_;
Predicate pred_;
downstream<output_type> out_;
};
} // namespace caf
#endif // CAF_STREAM_SOURCE_IMPL_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SOURCE_TRAIT_HPP
#define CAF_STREAM_SOURCE_TRAIT_HPP
#include "caf/detail/type_traits.hpp"
namespace caf {
template <class F>
struct stream_source_trait;
template <class State, class T>
struct stream_source_trait<void (State&, downstream<T>&, size_t)> {
using output = T;
using state = State;
};
template <class F>
using stream_source_trait_t =
stream_source_trait<typename detail::get_callable_trait<F>::fun_sig>;
} // namespace caf
#endif // CAF_STREAM_SOURCE_TRAIT_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_STAGE_HPP
#define CAF_STREAM_STAGE_HPP
#include "caf/fwd.hpp"
#include "caf/extend.hpp"
#include "caf/stream_handler.hpp"
#include "caf/mixin/has_upstreams.hpp"
#include "caf/mixin/has_downstreams.hpp"
namespace caf {
/// Models a stream stage with up- and downstreams.
class stream_stage : public extend<stream_handler, stream_stage>::
with<mixin::has_upstreams, mixin::has_downstreams> {
public:
// -- constructors, destructors, and assignment operators --------------------
stream_stage(abstract_upstream* in_ptr, abstract_downstream* out_ptr);
// -- overrides --------------------------------------------------------------
bool done() const final;
void abort(strong_actor_ptr&, const error&) final;
error downstream_demand(strong_actor_ptr&, size_t) final;
error upstream_batch(strong_actor_ptr&, size_t, message&) final;
void last_upstream_closed();
inline abstract_upstream& in() {
return *in_ptr_;
}
inline abstract_downstream& out() {
return *out_ptr_;
}
protected:
virtual error process_batch(message& msg) = 0;
private:
abstract_upstream* in_ptr_;
abstract_downstream* out_ptr_;
};
} // namespace caf
#endif // CAF_STREAM_STAGE_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_STAGE_IMPL_HPP
#define CAF_STREAM_STAGE_IMPL_HPP
#include "caf/upstream.hpp"
#include "caf/downstream.hpp"
#include "caf/stream_stage.hpp"
#include "caf/stream_stage_trait.hpp"
namespace caf {
template <class Fun, class Cleanup>
class stream_stage_impl : public stream_stage {
public:
using trait = stream_stage_trait_t<Fun>;
using state_type = typename trait::state;
using input_type = typename trait::input;
using output_type = typename trait::output;
stream_stage_impl(local_actor* self,
const stream_id& sid,
typename abstract_upstream::policy_ptr in_policy,
typename abstract_downstream::policy_ptr out_policy,
Fun fun, Cleanup cleanup)
: stream_stage(&in_, &out_),
fun_(std::move(fun)),
cleanup_(std::move(cleanup)),
in_(self, std::move(in_policy)),
out_(self, sid, std::move(out_policy)) {
// nop
}
error process_batch(message& msg) final {
using vec_type = std::vector<output_type>;
if (msg.match_elements<vec_type>()) {
auto& xs = msg.get_as<vec_type>(0);
for (auto& x : xs)
fun_(state_, out_, x);
return none;
}
return sec::unexpected_message;
}
message make_output_token(const stream_id& x) const final {
return make_message(stream<output_type>{x});
}
state_type& state() {
return state_;
}
private:
state_type state_;
Fun fun_;
Cleanup cleanup_;
upstream<output_type> in_;
downstream<input_type> out_;
};
} // namespace caf
#endif // CAF_STREAM_STAGE_IMPL_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_STAGE_TRAIT_HPP
#define CAF_STREAM_STAGE_TRAIT_HPP
#include "caf/fwd.hpp"
#include "caf/detail/type_traits.hpp"
namespace caf {
template <class F>
struct stream_stage_trait;
template <class State, class In, class Out>
struct stream_stage_trait<void (State&, downstream<Out>&, In)> {
using state = State;
using input = In;
using output = Out;
};
template <class F>
using stream_stage_trait_t =
stream_stage_trait<typename detail::get_callable_trait<F>::fun_sig>;
} // namespace caf
#endif // CAF_STREAM_STAGE_TRAIT_HPP
...@@ -61,6 +61,7 @@ using sorted_builtin_types = ...@@ -61,6 +61,7 @@ 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
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_UPSTREAM_HPP
#define CAF_UPSTREAM_HPP
#include "caf/abstract_upstream.hpp"
namespace caf {
template <class T>
class upstream : public abstract_upstream {
public:
upstream(local_actor* self, typename abstract_upstream::policy_ptr ptr)
: abstract_upstream(self, std::move(ptr)) {
// nop
}
};
} // namespace caf
#endif // CAF_UPSTREAM_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_UPSTREAM_PATH_HPP
#define CAF_UPSTREAM_PATH_HPP
#include <cstddef>
#include <cstdint>
#include "caf/stream_id.hpp"
#include "caf/stream_priority.hpp"
#include "caf/actor_control_block.hpp"
namespace caf {
/// Denotes an upstream actor in a stream topology. Each upstream actor can
/// refer to the stream using a different stream ID.
class upstream_path {
public:
/// Handle to the upstream actor.
strong_actor_ptr hdl;
/// Stream ID used on this upstream path.
stream_id sid;
/// Priority of this input channel.
stream_priority prio;
/// ID of the last received batch we have acknowledged.
int64_t last_acked_batch_id;
/// ID of the last received batch.
int64_t last_batch_id;
/// Amount of credit we have signaled upstream.
size_t assigned_credit;
upstream_path(strong_actor_ptr hdl, const stream_id& id, stream_priority p);
};
} // namespace caf
#endif // CAF_UPSTREAM_PATH_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_UPSTREAM_POLICY_HPP
#define CAF_UPSTREAM_POLICY_HPP
#include <cstdint>
#include "caf/fwd.hpp"
namespace caf {
class upstream_policy {
public:
virtual ~upstream_policy();
/// Queries the initial credit for the new path `y` in `x`.
virtual size_t initial_credit(abstract_upstream& x, upstream_path& y) = 0;
/// Reclaim credit of closed upstream `y`.
virtual void reclaim(abstract_upstream& x, upstream_path& y) = 0;
};
} // namespace caf
#endif // CAF_UPSTREAM_POLICY_HPP
...@@ -155,6 +155,15 @@ public: ...@@ -155,6 +155,15 @@ public:
return type_ == Pos; return type_ == Pos;
} }
template <class T>
bool is() const {
using namespace detail;
int_token<tl_index_where<type_list<Ts...>,
tbind<is_same_ish, T>::template type>::value>
token;
return is(token);
}
template <int Pos> template <int Pos>
const typename detail::tl_at<types, Pos>::type& const typename detail::tl_at<types, Pos>::type&
get(std::integral_constant<int, Pos> token) const { get(std::integral_constant<int, Pos> token) const {
...@@ -318,6 +327,11 @@ apply_visitor(Visitor& visitor, variant<Ts...>& data) { ...@@ -318,6 +327,11 @@ apply_visitor(Visitor& visitor, variant<Ts...>& data) {
return data.apply(visitor); return data.apply(visitor);
} }
template <class T, class... Ts>
bool holds_alternative(const variant<Ts...>& data) {
return data.template is<T>();
}
/// @relates variant /// @relates variant
template <class T> template <class T>
struct variant_compare_helper { struct variant_compare_helper {
......
...@@ -316,6 +316,7 @@ public: ...@@ -316,6 +316,7 @@ public:
******************************************************************************/ ******************************************************************************/
actor abstract_coordinator::printer() const { actor abstract_coordinator::printer() const {
CAF_ASSERT(printer_ != nullptr);
return actor_cast<actor>(printer_); return actor_cast<actor>(printer_);
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/abstract_downstream.hpp"
#include "caf/send.hpp"
#include "caf/downstream_path.hpp"
#include "caf/downstream_policy.hpp"
namespace caf {
abstract_downstream::abstract_downstream(local_actor* selfptr,
const stream_id& sid,
std::unique_ptr<downstream_policy> ptr)
: self_(selfptr), sid_(sid), policy_(std::move(ptr)) {
// nop
}
abstract_downstream::~abstract_downstream() {
// nop
}
size_t abstract_downstream::total_credit() const {
auto f = [](size_t x, path_cref y) {
// printf("f(%d, %d)\n", (int) x, (int) y.open_credit);
return x + y.open_credit;
};
return fold_paths(0, f);
}
size_t abstract_downstream::max_credit() const {
auto f = [](size_t x, path_cref y) { return std::max(x, y.open_credit); };
return fold_paths(0, f);
}
size_t abstract_downstream::min_credit() const {
auto f = [](size_t x, path_cref y) { return std::min(x, y.open_credit); };
return fold_paths(std::numeric_limits<size_t>::max(), f);
}
bool abstract_downstream::add_path(strong_actor_ptr ptr,
std::vector<atom_value> filter,
bool redeployable) {
auto predicate = [&](const path_uptr& x) { return x->ptr == ptr; };
if (std::none_of(paths_.begin(), paths_.end(), predicate)) {
paths_.emplace_back(
new path(std::move(ptr), std::move(filter), redeployable));
recalculate_active_filters();
return true;
}
return false;
}
bool abstract_downstream::remove_path(strong_actor_ptr& ptr) {
auto predicate = [&](const path_uptr& x) { return x->ptr == ptr; };
auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, predicate);
if (i != e) {
CAF_ASSERT((*i)->ptr != nullptr);
if (i != paths_.end() - 1)
std::swap(*i, paths_.back());
auto x = std::move(paths_.back());
paths_.pop_back();
unsafe_send_as(self_, x->ptr, make<stream_msg::close>(sid_));
//policy_->reclaim(this, x);
return true;
}
return false;
}
void abstract_downstream::close() {
for (auto& x : paths_)
unsafe_send_as(self_, x->ptr, make<stream_msg::close>(sid_));
paths_.clear();
}
void abstract_downstream::abort(strong_actor_ptr& cause, const error& reason) {
for (auto& x : paths_)
if (x->ptr != cause)
unsafe_send_as(self_, x->ptr,
make<stream_msg::abort>(this->sid_, reason));
}
auto abstract_downstream::find(const strong_actor_ptr& ptr) const
-> optional<path&> {
auto predicate = [&](const path_uptr& y) {
return y->ptr == ptr;
};
auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, predicate);
if (i != e)
return *(*i);
return none;
}
void abstract_downstream::recalculate_active_filters() {
active_filters_.clear();
for (auto& x : paths_)
active_filters_.emplace(x->filter);
}
void abstract_downstream::send_batch(downstream_path& dest, size_t chunk_size,
message chunk) {
auto scs = static_cast<int32_t>(chunk_size);
auto batch_id = dest.next_batch_id++;
stream_msg::batch batch{scs, std::move(chunk), batch_id};
if (dest.redeployable)
dest.unacknowledged_batches.emplace_back(batch_id, batch);
unsafe_send_as(self_, dest.ptr, stream_msg{sid_, std::move(batch)});
}
void abstract_downstream::sort_by_credit() {
auto cmp = [](const path_uptr& x, const path_uptr& y) {
return x->open_credit > y->open_credit;
};
std::sort(paths_.begin(), paths_.end(), cmp);
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/abstract_upstream.hpp"
#include "caf/send.hpp"
#include "caf/stream_msg.hpp"
namespace caf {
abstract_upstream::abstract_upstream(local_actor* selfptr,
abstract_upstream::policy_ptr p)
: self_(selfptr),
policy_(std::move(p)) {
// nop
}
abstract_upstream::~abstract_upstream() {
// nop
}
void abstract_upstream::abort(strong_actor_ptr& cause, const error& reason) {
for (auto& x : paths_)
if (x->hdl != cause)
unsafe_send_as(self_, x->hdl, make<stream_msg::abort>(x->sid, reason));
}
error abstract_upstream::pull(strong_actor_ptr& hdl, size_t n) {
CAF_ASSERT(hdl != nullptr);
auto has_hdl = [&](const path_uptr& x) {
CAF_ASSERT(x != nullptr);
return x->hdl == hdl;
};
auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, has_hdl);
if (i != e) {
auto& x = *(*i);
CAF_ASSERT(x.hdl == hdl);
x.assigned_credit += n;
unsafe_send_as(self_, x.hdl,
make<stream_msg::ack_batch>(x.sid, static_cast<int32_t>(n),
x.last_batch_id++));
return none;
}
return sec::invalid_upstream;
}
error abstract_upstream::pull(size_t n) {
// TODO: upstream policy
if (!paths_.empty()) {
pull(paths_.front()->hdl, n);
return none;
}
return sec::invalid_upstream;
}
expected<size_t> abstract_upstream::add_path(strong_actor_ptr hdl,
const stream_id& sid,
stream_priority prio) {
CAF_ASSERT(hdl != nullptr);
auto has_hdl = [&](const path_uptr& x) {
CAF_ASSERT(x != nullptr);
return x->hdl == hdl;
};
auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, has_hdl);
if (i == e) {
paths_.emplace_back(new path(std::move(hdl), sid, prio));
return policy_->initial_credit(*this, *paths_.back());
}
return sec::upstream_already_exists;
}
bool abstract_upstream::remove_path(const strong_actor_ptr& hdl) {
auto has_hdl = [&](const path_uptr& x) {
CAF_ASSERT(x != nullptr);
return x->hdl == hdl;
};
auto e = paths_.end();
auto i = std::find_if(paths_.begin(), e, has_hdl);
if (i != e) {
if (i != e - 1)
std::swap(*i, paths_.back());
paths_.pop_back();
return true;
}
return false;
}
} // namespace caf
...@@ -33,13 +33,20 @@ namespace caf { ...@@ -33,13 +33,20 @@ namespace caf {
namespace { namespace {
using option_vector = actor_system_config::option_vector; using option_vector = actor_system_config::option_vector;
const char actor_conf_prefix[] = "actor:";
constexpr size_t actor_conf_prefix_size = 6;
class actor_system_config_reader { class actor_system_config_reader {
public: public:
using sink = std::function<void (size_t, config_value&, using sink = std::function<void (size_t, config_value&,
optional<std::ostream&>)>; optional<std::ostream&>)>;
actor_system_config_reader(option_vector& xs, option_vector& ys) { using named_actor_sink = std::function<void (size_t, const std::string&,
config_value&)>;
actor_system_config_reader(option_vector& xs, option_vector& ys,
named_actor_sink na_sink)
: named_actor_sink_(std::move(na_sink)){
add_opts(xs); add_opts(xs);
add_opts(ys); add_opts(ys);
} }
...@@ -49,17 +56,28 @@ public: ...@@ -49,17 +56,28 @@ public:
sinks_.emplace(x->full_name(), x->to_sink()); sinks_.emplace(x->full_name(), x->to_sink());
} }
void operator()(size_t ln, const std::string& name, config_value& cv) { void operator()(size_t ln, const std::string& name, config_value& cv,
optional<std::ostream&> out) {
auto i = sinks_.find(name); auto i = sinks_.find(name);
if (i != sinks_.end()) if (i != sinks_.end()) {
(i->second)(ln, cv, none); (i->second)(ln, cv, none);
else return;
std::cerr << "error in line " << ln }
<< R"(: unrecognized parameter name ")" << name << R"(")"; // check whether this is an individual actor config
if (name.compare(0, actor_conf_prefix_size, actor_conf_prefix) == 0) {
auto substr = name.substr(actor_conf_prefix_size);
named_actor_sink_(ln, substr, cv);
return;
}
if (out)
*out << "error in line " << ln
<< ": unrecognized parameter name \"" << name << "\""
<< std::endl;
} }
private: private:
std::map<std::string, sink> sinks_; std::map<std::string, sink> sinks_;
named_actor_sink named_actor_sink_;
}; };
} // namespace <anonymous> } // namespace <anonymous>
...@@ -228,12 +246,36 @@ actor_system_config& actor_system_config::parse(message& args, ...@@ -228,12 +246,36 @@ actor_system_config& actor_system_config::parse(message& args,
std::istream& ini) { std::istream& ini) {
// (2) content of the INI file overrides hard-coded defaults // (2) content of the INI file overrides hard-coded defaults
if (ini.good()) { if (ini.good()) {
actor_system_config_reader consumer{options_, custom_options_}; using conf_sink = std::function<void (size_t, config_value&,
auto f = [&](size_t ln, std::string str, optional<std::ostream&>)>;
config_value& x, optional<std::ostream&>) { using conf_sinks = std::unordered_map<std::string, conf_sink>;
consumer(ln, std::move(str), x); using conf_mapping = std::pair<option_vector, conf_sinks>;
hash_map<std::string, conf_mapping> ovs;
auto nac_sink = [&](size_t ln, const std::string& nm, config_value& cv) {
std::string actor_name{nm.begin(), std::find(nm.begin(), nm.end(), '.')};
auto ac = named_actor_configs.find(actor_name);
if (ac == named_actor_configs.end())
ac = named_actor_configs.emplace(actor_name,
named_actor_config{}).first;
auto& ov = ovs[actor_name];
if (ov.first.empty()) {
opt_group(ov.first, ac->first.c_str())
.add(ac->second.strategy, "strategy", "")
.add(ac->second.low_watermark, "low-watermark", "")
.add(ac->second.max_pending, "max-pending", "");
for (auto& opt : ov.first)
ov.second.emplace(opt->full_name(), opt->to_sink());
}
auto i = ov.second.find(nm);
if (i != ov.second.end())
i->second(ln, cv, none);
else
std::cerr << "error in line " << ln
<< ": unrecognized parameter name \"" << nm << "\""
<< std::endl;
}; };
detail::parse_ini(ini, f, std::cerr); actor_system_config_reader consumer{options_, custom_options_, nac_sink};
detail::parse_ini(ini, consumer, std::cerr);
} }
// (3) CLI options override the content of the INI file // (3) CLI options override the content of the INI file
std::string dummy; // caf#config-file either ignored or already open std::string dummy; // caf#config-file either ignored or already open
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/policy/anycast.hpp"
#include "caf/abstract_downstream.hpp"
namespace caf {
namespace policy {
void anycast::push(abstract_downstream& out) {
out.anycast();
}
size_t anycast::desired_buffer_size(const abstract_downstream& out) {
return out.total_credit();
}
} // namespace policy
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/policy/broadcast.hpp"
#include "caf/abstract_downstream.hpp"
namespace caf {
namespace policy {
void broadcast::push(abstract_downstream& out) {
out.broadcast();
}
size_t broadcast::desired_buffer_size(const abstract_downstream& out) {
return out.min_credit();
}
} // namespace policy
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/downstream_path.hpp"
namespace caf {
downstream_path::downstream_path(strong_actor_ptr p, downstream_path::topics ts,
bool redeploy)
: ptr(std::move(p)),
next_batch_id(0),
open_credit(0),
filter(std::move(ts)),
redeployable(redeploy) {
// nop
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/downstream_policy.hpp"
#include "caf/atom.hpp"
namespace caf {
downstream_policy::~downstream_policy() {
// nop
}
atom_value downstream_policy::categorize(type_erased_value&) const {
return atom("");
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/policy/greedy.hpp"
namespace caf {
namespace policy {
size_t greedy::initial_credit(abstract_upstream&, upstream_path&) {
return 5;
}
void greedy::reclaim(abstract_upstream&, upstream_path&) {
// nop
}
} // namespace policy
} // namespace caf
...@@ -22,6 +22,7 @@ ...@@ -22,6 +22,7 @@
#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"
...@@ -370,6 +371,24 @@ scheduled_actor::categorize(mailbox_element& x) { ...@@ -370,6 +371,24 @@ 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>(): {
auto sm = content.move_if_unshared<stream_msg>(0);
auto e = streams_.end();
stream_msg_visitor f{this, sm.sid, streams_.find(sm.sid), e};
auto res = apply_visitor(f, sm.content);
auto i = res.second;
if (res.first) {
if (i != e) {
i->second->abort(current_sender(), std::move(res.first));
streams_.erase(i);
}
} else if (i != e) {
if (i->second->done()) {
streams_.erase(i);
}
}
return message_category::internal;
}
default: default:
return message_category::ordinary; return message_category::ordinary;
} }
......
...@@ -56,6 +56,14 @@ const char* sec_strings[] = { ...@@ -56,6 +56,14 @@ const char* sec_strings[] = {
"no_proxy_registry", "no_proxy_registry",
"runtime_error", "runtime_error",
"remote_linking_failed", "remote_linking_failed",
"cannot_add_upstream",
"upstream_already_exists",
"invalid_upstream",
"cannot_add_downstream",
"downstream_already_exists",
"invalid_downstream",
"no_downstream_stages_defined",
"stream_init_failed",
"bad_function_call" "bad_function_call"
}; };
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/stream.hpp"
namespace caf {
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/stream_handler.hpp"
#include "caf/sec.hpp"
#include "caf/error.hpp"
#include "caf/logger.hpp"
#include "caf/message.hpp"
#include "caf/expected.hpp"
namespace caf {
stream_handler::~stream_handler() {
// nop
}
error stream_handler::add_downstream(strong_actor_ptr&,
const stream_handler::topics&, size_t,
bool) {
CAF_LOG_ERROR("Cannot add downstream to a stream marked as no-downstreams");
return sec::cannot_add_downstream;
}
error stream_handler::downstream_ack(strong_actor_ptr&, int64_t) {
CAF_LOG_ERROR("Received downstream messages in "
"a stream marked as no-downstreams");
return sec::invalid_downstream;
}
error stream_handler::downstream_demand(strong_actor_ptr&, size_t) {
CAF_LOG_ERROR("Received downstream messages in "
"a stream marked as no-downstreams");
return sec::invalid_downstream;
}
error stream_handler::push() {
CAF_LOG_ERROR("Cannot push to a stream marked as no-downstreams");
return sec::invalid_downstream;
}
expected<size_t> stream_handler::add_upstream(strong_actor_ptr&,
const stream_id&,
stream_priority) {
CAF_LOG_ERROR("Cannot add upstream to a stream marked as no-upstreams");
return sec::cannot_add_upstream;
}
error stream_handler::upstream_batch(strong_actor_ptr&, size_t, message&) {
CAF_LOG_ERROR("Received upstream messages in "
"a stream marked as no-upstreams");
return sec::invalid_upstream;
}
error stream_handler::close_upstream(strong_actor_ptr&) {
CAF_LOG_ERROR("Received upstream messages in "
"a stream marked as no-upstreams");
return sec::invalid_upstream;
}
error stream_handler::pull(size_t) {
CAF_LOG_ERROR("Cannot pull from a stream marked as no-upstreams");
return sec::invalid_upstream;
}
bool stream_handler::is_sink() const {
return false;
}
bool stream_handler::is_source() const {
return false;
}
message stream_handler::make_output_token(const stream_id&) const {
return make_message();
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/stream_msg_visitor.hpp"
#include "caf/send.hpp"
#include "caf/logger.hpp"
#include "caf/scheduled_actor.hpp"
namespace caf {
stream_msg_visitor::stream_msg_visitor(scheduled_actor* self, stream_id& sid,
iterator i, iterator last)
: self_(self),
sid_(sid),
i_(i),
e_(last) {
// nop
}
auto stream_msg_visitor::operator()(stream_msg::open& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
CAF_ASSERT(self_->current_mailbox_element() != nullptr);
if (i_ != e_)
return {sec::downstream_already_exists, e_};
auto& predecessor = x.prev_stage;
auto fail = [&](error reason) -> result_type {
auto rp = self_->make_response_promise();
rp.deliver(reason);
unsafe_send_as(self_, predecessor, make<stream_msg::abort>(sid_, reason));
return {std::move(reason), e_};
};
if (!predecessor) {
CAF_LOG_WARNING("received stream_msg::open with empty prev_stage");
return fail(sec::invalid_upstream);
}
auto& bs = self_->bhvr_stack();
if (bs.empty()) {
CAF_LOG_WARNING("cannot open stream in actor without behavior");
return fail(sec::stream_init_failed);
}
auto bhvr = self_->bhvr_stack().back();
auto res = bhvr(x.token);
if (!res) {
CAF_LOG_WARNING("stream handshake failed: actor did not respond to token:"
<< CAF_ARG(x.token));
return fail(sec::stream_init_failed);
}
i_ = self_->streams().find(sid_);
if (i_ == e_) {
CAF_LOG_WARNING("stream handshake failed: actor did not provide a stream "
"handler after receiving token:"
<< CAF_ARG(x.token));
return fail(sec::stream_init_failed);
} else {
auto& handler = i_->second;
// store upstream actor
auto initial_credit = handler->add_upstream(x.prev_stage, sid_, x.priority);
if (initial_credit) {
// check whether we are a stage in a longer pipeline and send more
// stream_open messages if required
auto ic = static_cast<int32_t>(*initial_credit);
std::vector<atom_value> filter;
unsafe_send_as(self_, predecessor,
make_message(make<stream_msg::ack_open>(
std::move(sid_), ic, std::move(filter), false)));
return {none, i_};
} else {
printf("error: %s\n", self_->system().render(initial_credit.error()).c_str());
self_->streams().erase(i_);
return fail(std::move(initial_credit.error()));
}
}
}
auto stream_msg_visitor::operator()(stream_msg::close&) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
if (i_ != e_) {
i_->second->close_upstream(self_->current_sender());
return {none, i_->second->done() ? i_ : e_};
}
return {sec::invalid_upstream, e_};
}
auto stream_msg_visitor::operator()(stream_msg::abort& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
if (i_ == e_) {
CAF_LOG_DEBUG("received stream_msg::abort for unknown stream");
return {sec::unexpected_message, e_};
}
return {std::move(x.reason), i_};
}
auto stream_msg_visitor::operator()(stream_msg::ack_open& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
if (i_ != e_)
return {i_->second->add_downstream(self_->current_sender(), x.filter,
static_cast<size_t>(x.initial_demand),
false),
i_};
CAF_LOG_DEBUG("received stream_msg::ok for unknown stream");
return {sec::unexpected_message, e_};
}
auto stream_msg_visitor::operator()(stream_msg::batch& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
if (i_ != e_)
return {i_->second->upstream_batch(self_->current_sender(),
static_cast<size_t>(x.xs_size), x.xs),
i_};
CAF_LOG_DEBUG("received stream_msg::batch for unknown stream");
return {sec::unexpected_message, e_};
}
auto stream_msg_visitor::operator()(stream_msg::ack_batch& x) -> result_type {
CAF_LOG_TRACE(CAF_ARG(x));
if (i_ != e_)
return {i_->second->downstream_demand(self_->current_sender(),
static_cast<size_t>(x.new_capacity)),
i_};
CAF_LOG_DEBUG("received stream_msg::batch for unknown stream");
return {sec::unexpected_message, e_};
}
auto stream_msg_visitor::operator()(stream_msg::downstream_failed&) -> result_type {
// TODO
return {none, e_};
}
auto stream_msg_visitor::operator()(stream_msg::upstream_failed&) -> result_type {
// TODO
return {none, e_};
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/stream_priority.hpp"
namespace caf {
std::string to_string(stream_priority x) {
switch (x) {
default: return "invalid";
case stream_priority::low: return "low";
case stream_priority::normal: return "normal";
case stream_priority::high: return "high";
}
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/stream_sink.hpp"
#include "caf/send.hpp"
#include "caf/abstract_upstream.hpp"
namespace caf {
stream_sink::stream_sink(abstract_upstream* in_ptr,
strong_actor_ptr&& orig_sender,
std::vector<strong_actor_ptr>&& trailing_stages,
message_id mid)
: in_ptr_(in_ptr),
original_sender_(std::move(orig_sender)),
next_stages_(std::move(trailing_stages)),
original_msg_id_(mid) {
// nop
}
bool stream_sink::done() const {
// we are done streaming if no upstream paths remain
return in_ptr_->closed();
}
error stream_sink::upstream_batch(strong_actor_ptr& src, size_t xs_size,
message& xs) {
auto err = in_ptr_->pull(src, xs_size);
if (!err)
consume(xs);
return err;
}
void stream_sink::abort(strong_actor_ptr& cause, const error& reason) {
unsafe_send_as(in_ptr_->self(), original_sender_, reason);
in_ptr_->abort(cause, reason);
}
void stream_sink::last_upstream_closed() {
unsafe_response(in().self(), std::move(original_sender_),
std::move(next_stages_), original_msg_id_, finalize());
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/stream_source.hpp"
#include "caf/sec.hpp"
#include "caf/error.hpp"
#include "caf/logger.hpp"
#include "caf/downstream_path.hpp"
#include "caf/abstract_downstream.hpp"
namespace caf {
stream_source::stream_source(abstract_downstream* out_ptr) : out_ptr_(out_ptr) {
// nop
}
stream_source::~stream_source() {
// nop
}
bool stream_source::done() const {
// we are done streaming if no downstream paths remain
return out_ptr_->closed();
}
error stream_source::downstream_demand(strong_actor_ptr& hdl, size_t value) {
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(value));
auto path = out_ptr_->find(hdl);
if (path) {
path->open_credit += value;
if (!at_end()) {
// produce new elements
auto current_size = buf_size();
auto size_hint = out_ptr_->policy().desired_buffer_size(*out_ptr_);
if (current_size < size_hint)
generate(size_hint - current_size);
return trigger_send(hdl);
} else if(buf_size() > 0) {
// transmit cached elements before closing paths
return trigger_send(hdl);
} else if (out_ptr_->remove_path(hdl)) {
return none;
}
}
return sec::invalid_downstream;
}
void stream_source::abort(strong_actor_ptr& cause, const error& reason) {
out_ptr_->abort(cause, reason);
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/stream_stage.hpp"
#include "caf/sec.hpp"
#include "caf/logger.hpp"
#include "caf/downstream_path.hpp"
#include "caf/abstract_upstream.hpp"
#include "caf/downstream_policy.hpp"
#include "caf/abstract_downstream.hpp"
namespace caf {
stream_stage::stream_stage(abstract_upstream* in_ptr,
abstract_downstream* out_ptr)
: in_ptr_(in_ptr),
out_ptr_(out_ptr) {
// nop
}
bool stream_stage::done() const {
return in_ptr_->closed() && out_ptr_->closed();
}
error stream_stage::upstream_batch(strong_actor_ptr& hdl, size_t xs_size,
message& xs) {
auto err = in_ptr_->pull(hdl, xs_size);
if (!err) {
process_batch(xs);
strong_actor_ptr dummy;
trigger_send(dummy);
}
return err;
}
error stream_stage::downstream_demand(strong_actor_ptr& hdl, size_t value) {
auto path = out_ptr_->find(hdl);
if (path) {
path->open_credit += value;
if(out_ptr_->buf_size() > 0) {
return trigger_send(hdl);
} else if (in_ptr_->closed()) {
if (!out_ptr_->remove_path(hdl)) {
return sec::invalid_downstream;
}
}
return none;
}
return sec::invalid_downstream;
}
void stream_stage::abort(strong_actor_ptr& cause, const error& reason) {
in_ptr_->abort(cause, reason);
out_ptr_->abort(cause, reason);
}
void stream_stage::last_upstream_closed() {
if (out().buf_size() == 0)
out().close();
}
} // namespace caf
...@@ -34,10 +34,12 @@ ...@@ -34,10 +34,12 @@
#include "caf/group.hpp" #include "caf/group.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/duration.hpp"
#include "caf/timestamp.hpp" #include "caf/timestamp.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/stream_msg.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/actor_factory.hpp" #include "caf/actor_factory.hpp"
#include "caf/abstract_group.hpp" #include "caf/abstract_group.hpp"
...@@ -45,7 +47,6 @@ ...@@ -45,7 +47,6 @@
#include "caf/message_builder.hpp" #include "caf/message_builder.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/type_nr.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"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
...@@ -75,6 +76,7 @@ const char* numbered_type_names[] = { ...@@ -75,6 +76,7 @@ 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",
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/upstream_path.hpp"
namespace caf {
upstream_path::upstream_path(strong_actor_ptr ptr, const stream_id& id,
stream_priority p)
: hdl(std::move(ptr)),
sid(id),
prio(p),
last_acked_batch_id(0),
last_batch_id(0),
assigned_credit(0) {
// nop
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/upstream_policy.hpp"
namespace caf {
upstream_policy::~upstream_policy() {
// nop
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include <string>
#include <numeric>
#include <fstream>
#include <iostream>
#include <iterator>
#define CAF_SUITE streaming
#include "caf/test/dsl.hpp"
using std::cout;
using std::endl;
using std::string;
using namespace caf;
namespace {
behavior file_reader(event_based_actor* self) {
using buf = std::deque<int>;
return {
[=](const std::string&) -> stream<int> {
return self->add_source(
// 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();
}
);
}
};
}
behavior filter(event_based_actor* self) {
return {
[=](stream<int>& in) -> stream<int> {
return self->add_stage(
// input stream
in,
// initialize state
[=](unit_t&) {
// nop
},
// processing step
[=](unit_t&, downstream<int>& out, int x) {
if (x & 0x01)
out.push(x);
},
// cleanup
[=](unit_t&) {
// nop
}
);
}
};
}
behavior broken_filter(event_based_actor*) {
return {
[=](stream<int>& x) -> stream<int> {
return x;
}
};
}
behavior sum_up(event_based_actor* self) {
return {
[=](stream<int>& in) {
return self->add_sink(
// input stream
in,
// initialize state
[](int& x) {
x = 0;
},
// processing step
[](int& x, int y) {
x += y;
},
// cleanup and produce result message
[](int& x) -> int {
return x;
}
);
}
};
}
using fixture = test_coordinator_fixture<>;
} // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(streaming_tests, fixture)
CAF_TEST(no_downstream) {
CAF_MESSAGE("opening streams must fail if no downstream stage exists");
auto source = sys.spawn(file_reader);
self->send(source, "test.txt");
sched.run();
CAF_CHECK_EQUAL(fetch_result(), sec::no_downstream_stages_defined);
CAF_CHECK(deref(source).streams().empty());
}
CAF_TEST(broken_pipeline) {
CAF_MESSAGE("streams must abort if a stage fails to initialize its state");
auto source = sys.spawn(file_reader);
auto stage = sys.spawn(broken_filter);
auto pipeline = stage * source;
sched.run();
// self --("test.txt")--> source
self->send(pipeline, "test.txt");
expect((std::string), from(self).to(source).with("test.txt"));
// source --(stream_msg::open)--> stage
expect((stream_msg::open), from(self).to(stage).with(_, source, _, _, false));
CAF_CHECK(!deref(source).streams().empty());
CAF_CHECK(deref(stage).streams().empty());
// stage --(stream_msg::abort)--> source
expect((stream_msg::abort),
from(stage).to(source).with(sec::stream_init_failed));
CAF_CHECK(deref(source).streams().empty());
CAF_CHECK(deref(stage).streams().empty());
CAF_CHECK_EQUAL(fetch_result(), sec::stream_init_failed);
}
CAF_TEST(incomplete_pipeline) {
CAF_MESSAGE("streams must abort if not reaching a sink");
auto source = sys.spawn(file_reader);
auto stage = sys.spawn(filter);
auto pipeline = stage * source;
sched.run();
// self --("test.txt")--> source
self->send(pipeline, "test.txt");
expect((std::string), from(self).to(source).with("test.txt"));
// source --(stream_msg::open)--> stage
CAF_REQUIRE(sched.prioritize(stage));
expect((stream_msg::open), from(self).to(stage).with(_, source, _, _, false));
CAF_CHECK(!deref(source).streams().empty());
CAF_CHECK(deref(stage).streams().empty());
// stage --(stream_msg::abort)--> source
expect((stream_msg::abort),
from(stage).to(source).with(sec::stream_init_failed));
CAF_CHECK(deref(source).streams().empty());
CAF_CHECK(deref(stage).streams().empty());
CAF_CHECK_EQUAL(fetch_result(), sec::stream_init_failed);
}
CAF_TEST(depth2_pipeline) {
auto source = sys.spawn(file_reader);
auto sink = sys.spawn(sum_up);
auto pipeline = sink * source;
// run initialization code
sched.run();
// self -------("test.txt")-------> source
self->send(pipeline, "test.txt");
expect((std::string), from(self).to(source).with("test.txt"));
CAF_CHECK(!deref(source).streams().empty());
CAF_CHECK(deref(sink).streams().empty());
// source ----(stream_msgreturn ::open)----> sink
expect((stream_msg::open), from(self).to(sink).with(_, source, _, _, false));
// source <----(stream_msg::ack_open)------ sink
expect((stream_msg::ack_open), from(sink).to(source).with(5, _, false));
// source ----(stream_msg::batch)---> sink
expect((stream_msg::batch),
from(source).to(sink).with(5, std::vector<int>{1, 2, 3, 4, 5}, 0));
// source <--(stream_msg::ack_batch)---- sink
expect((stream_msg::ack_batch), from(sink).to(source).with(5, 0));
// source ----(stream_msg::batch)---> sink
expect((stream_msg::batch),
from(source).to(sink).with(4, std::vector<int>{6, 7, 8, 9}, 1));
// source <--(stream_msg::ack_batch)---- sink
expect((stream_msg::ack_batch), from(sink).to(source).with(4, 1));
// source ----(stream_msg::close)---> sink
expect((stream_msg::close), from(source).to(sink).with());
CAF_CHECK(deref(source).streams().empty());
CAF_CHECK(deref(sink).streams().empty());
}
CAF_TEST(depth3_pipeline) {
CAF_MESSAGE("check fully initialized pipeline");
auto source = sys.spawn(file_reader);
auto stage = sys.spawn(filter);
auto sink = sys.spawn(sum_up);
auto pipeline = self * sink * stage * source;
// run initialization code
sched.run();
// self --("test.txt")--> source
CAF_CHECK(self->mailbox().empty());
self->send(pipeline, "test.txt");
expect((std::string), from(self).to(source).with("test.txt"));
CAF_CHECK(self->mailbox().empty());
CAF_CHECK(!deref(source).streams().empty());
CAF_CHECK(deref(stage).streams().empty());
CAF_CHECK(deref(sink).streams().empty());
// source --(stream_msg::open)--> stage
expect((stream_msg::open), from(self).to(stage).with(_, source, _, _, false));
CAF_CHECK(!deref(source).streams().empty());
CAF_CHECK(!deref(stage).streams().empty());
CAF_CHECK(deref(sink).streams().empty());
// stage --(stream_msg::open)--> sink
expect((stream_msg::open), from(self).to(sink).with(_, stage, _, _, false));
CAF_CHECK(!deref(source).streams().empty());
CAF_CHECK(!deref(stage).streams().empty());
CAF_CHECK(!deref(sink).streams().empty());
// sink --(stream_msg::ack_open)--> stage
expect((stream_msg::ack_open), from(sink).to(stage).with(5, _, false));
// stage --(stream_msg::ack_open)--> source
expect((stream_msg::ack_open), from(stage).to(source).with(5, _, false));
// source --(stream_msg::batch)--> stage
expect((stream_msg::batch),
from(source).to(stage).with(5, std::vector<int>{1, 2, 3, 4, 5}, 0));
// stage --(stream_msg::batch)--> sink
expect((stream_msg::batch),
from(stage).to(sink).with(3, std::vector<int>{1, 3, 5}, 0));
// stage --(stream_msg::batch)--> source
expect((stream_msg::ack_batch), from(stage).to(source).with(5, 0));
// sink --(stream_msg::batch)--> stage
expect((stream_msg::ack_batch), from(sink).to(stage).with(3, 0));
// source --(stream_msg::batch)--> stage
expect((stream_msg::batch),
from(source).to(stage).with(4, std::vector<int>{6, 7, 8, 9}, 1));
// stage --(stream_msg::batch)--> sink
expect((stream_msg::batch),
from(stage).to(sink).with(2, std::vector<int>{7, 9}, 1));
// stage --(stream_msg::batch)--> source
expect((stream_msg::ack_batch), from(stage).to(source).with(4, 1));
// sink --(stream_msg::batch)--> stage
expect((stream_msg::ack_batch), from(sink).to(stage).with(2, 1));
// source ----(stream_msg::close)---> stage
expect((stream_msg::close), from(source).to(stage).with());
// stage ----(stream_msg::close)---> sink
expect((stream_msg::close), from(stage).to(sink).with());
// sink ----(result: 25)---> self
expect((int), from(sink).to(self).with(25));
//auto res = fetch_result<int>();
//CAF_CHECK_EQUAL(res, 25);
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -185,7 +185,10 @@ public: ...@@ -185,7 +185,10 @@ public:
std::tuple<const Ts&...> peek() { std::tuple<const Ts&...> peek() {
CAF_REQUIRE(dest_ != nullptr); CAF_REQUIRE(dest_ != nullptr);
auto ptr = dest_->mailbox().peek(); auto ptr = dest_->mailbox().peek();
CAF_REQUIRE(ptr->content().match_elements<Ts...>()); if (!ptr->content().match_elements<Ts...>()) {
CAF_FAIL("Message does not match expected pattern: " << to_string(ptr->content()));
}
//CAF_REQUIRE(ptr->content().match_elements<Ts...>());
return ptr->content().get_as_tuple<Ts...>(); return ptr->content().get_as_tuple<Ts...>();
} }
......
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