Commit a18c0cc6 authored by Dominik Charousset's avatar Dominik Charousset

Remove paths_ from stream_scatterer

Move path-related state to `stream_scatterer_impl` to allow non-owning
scatterer types (such as `fused_scatterer`).
parent 50aea7f9
......@@ -96,6 +96,7 @@ set (LIBCAF_CORE_SRCS
src/stream_manager.cpp
src/stream_priority.cpp
src/stream_scatterer.cpp
src/stream_scatterer_impl.cpp
src/stringification_inspector.cpp
src/sync_request_bouncer.cpp
src/term.cpp
......
......@@ -108,11 +108,11 @@ public:
}
protected:
void about_to_erase(typename super::map_type::iterator i, bool silent,
void about_to_erase(outbound_path* ptr, bool silent,
error* reason) override {
CAF_LOG_DEBUG("remove cache:" << CAF_ARG2("slot", i->second->slots.sender));
state_map_.erase(i->second->slots.sender);
super::about_to_erase(i, silent, reason);
CAF_LOG_DEBUG("remove cache:" << CAF_ARG2("slot", ptr->slots.sender));
state_map_.erase(ptr->slots.sender);
super::about_to_erase(ptr, silent, reason);
}
private:
......
......@@ -24,21 +24,19 @@
#include <cstddef>
#include <iterator>
#include "caf/actor_control_block.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
#include "caf/stream_scatterer.hpp"
#include "caf/stream_scatterer_impl.hpp"
namespace caf {
/// Mixin for streams with any number of downstreams. `Subtype` must provide a
/// member function `buf()` returning a queue with `std::deque`-like interface.
template <class T>
class buffered_scatterer : public stream_scatterer {
class buffered_scatterer : public stream_scatterer_impl {
public:
// -- member types -----------------------------------------------------------
using super = stream_scatterer;
using super = stream_scatterer_impl;
using value_type = T;
......
......@@ -78,7 +78,6 @@ class node_id;
class behavior;
class duration;
class resumable;
class stream_id;
class actor_addr;
class actor_pool;
class message_id;
......@@ -122,6 +121,7 @@ struct unit_t;
struct exit_msg;
struct down_msg;
struct timeout_msg;
struct stream_slots;
struct upstream_msg;
struct group_down_msg;
struct downstream_msg;
......
......@@ -30,6 +30,14 @@ public:
~invalid_stream_scatterer() override;
size_t num_paths() const noexcept override;
path_ptr add_path(stream_slots slots, strong_actor_ptr target) override;
unique_path_ptr take_path(stream_slot slots) noexcept override;
path_ptr path(stream_slot slots) noexcept override;
void emit_batches() override;
void force_emit_batches() override;
......@@ -39,6 +47,14 @@ public:
size_t buffered() const noexcept override;
message make_handshake_token(stream_slot slot) const override;
protected:
void for_each_path_impl(path_visitor& f) override;
bool check_paths_impl(path_algorithm algo,
path_predicate& pred) const noexcept override;
void clear_paths() override;
};
} // namespace caf
......
......@@ -19,17 +19,9 @@
#ifndef CAF_STREAM_SCATTERER_HPP
#define CAF_STREAM_SCATTERER_HPP
#include <cstddef>
#include <memory>
#include "caf/actor_control_block.hpp"
#include "caf/duration.hpp"
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/optional.hpp"
#include "caf/stream_slot.hpp"
#include "caf/detail/unordered_flat_map.hpp"
namespace caf {
......@@ -45,10 +37,26 @@ public:
using path_ptr = path_type*;
/// Unique pointer to an outbound path.
using path_unique_ptr = std::unique_ptr<path_type>;
/// Maps slots to paths.
using map_type = detail::unordered_flat_map<stream_slot, path_unique_ptr>;
using unique_path_ptr = std::unique_ptr<path_type>;
/// Function object for iterating over all paths.
struct path_visitor {
virtual ~path_visitor();
virtual void operator()(outbound_path& x) = 0;
};
/// Predicate object for paths.
struct path_predicate {
virtual ~path_predicate();
virtual bool operator()(const outbound_path& x) const noexcept = 0;
};
/// Selects a check algorithms.
enum path_algorithm {
all_of,
any_of,
none_of
};
// -- constructors, destructors, and assignment operators --------------------
......@@ -56,60 +64,91 @@ public:
virtual ~stream_scatterer();
// -- properties -------------------------------------------------------------
local_actor* self() const {
return self_;
}
// -- meta information -------------------------------------------------------
/// Returns `true` if thie scatterer belongs to a sink, i.e., terminates the
/// stream and never has outbound paths.
virtual bool terminal() const noexcept;
// -- path management --------------------------------------------------------
/// Adds a path to `target` to the scatterer.
/// @returns The added path on success, `nullptr` otherwise.
virtual path_ptr add_path(stream_slots slots, strong_actor_ptr target);
/// Applies `f` to each path.
template <class F>
void for_each_path(F f) {
struct impl : path_visitor {
F fun;
impl(F x) : fun(std::move(x)) {
// nop
}
void operator()(outbound_path& x) override {
fun(x);
}
};
impl g{std::move(f)};
for_each_path_impl(g);
}
/// Checks whether `predicate` holds true for all paths.
template <class Predicate>
bool all_paths(Predicate predicate) const noexcept {
return check_paths(path_algorithm::all_of, std::move(predicate));
}
/// Checks whether `predicate` holds true for any path.
template <class Predicate>
bool any_path(Predicate predicate) const noexcept {
return check_paths(path_algorithm::any_of, std::move(predicate));
}
/// Checks whether `predicate` holds true for no path.
template <class Predicate>
bool no_path(Predicate predicate) const noexcept {
return check_paths(path_algorithm::none_of, std::move(predicate));
}
/// Returns the current number of paths.
size_t num_paths() const noexcept;
virtual size_t num_paths() const noexcept = 0;
/// Adds a path to `target` to the scatterer.
/// @returns The added path on success, `nullptr` otherwise.
virtual path_ptr add_path(stream_slots slots, strong_actor_ptr target) = 0;
/// Removes a path from the scatterer and returns it.
path_unique_ptr take_path(stream_slot slots) noexcept;
virtual unique_path_ptr take_path(stream_slot slots) noexcept = 0;
/// Returns the path associated to `slots` or `nullptr`.
path_ptr path(stream_slot slots) noexcept;
virtual path_ptr path(stream_slot slots) noexcept = 0;
/// Returns `true` if there is no data pending and no unacknowledged batch on
/// any path.
bool clean() const noexcept;
/// Removes all paths gracefully.
void close();
virtual void close();
/// Removes all paths with an error message.
void abort(error reason);
virtual void abort(error reason);
/// Returns `true` if no downstream exists and `!continuous()`,
/// `false` otherwise.
/// Returns `num_paths() == 0`.
inline bool empty() const noexcept {
return paths_.empty();
return num_paths() == 0;
}
/// Returns the minimum amount of credit on all output paths.
size_t min_credit() const noexcept;
size_t min_credit() const;
/// Returns the maximum amount of credit on all output paths.
size_t max_credit() const noexcept;
size_t max_credit() const;
/// Returns the total amount of credit on all output paths, i.e., the sum of
/// all individual credits.
size_t total_credit() const noexcept;
// -- state access -----------------------------------------------------------
local_actor* self() const {
return self_;
}
// -- meta information -------------------------------------------------------
/// Returns `true` if thie scatterer belongs to a sink, i.e., terminates the
/// stream and never has outbound paths.
virtual bool terminal() const noexcept;
// -- pure virtual memeber functions -----------------------------------------
size_t total_credit() const;
/// Sends batches to sinks.
virtual void emit_batches() = 0;
......@@ -128,17 +167,46 @@ public:
/// this scatterer.
virtual message make_handshake_token(stream_slot slot) const = 0;
/// Silently removes all paths.
virtual void clear_paths() = 0;
protected:
// -- customization points ---------------------------------------------------
/// Applies `f` to each path.
virtual void for_each_path_impl(path_visitor& f) = 0;
/// Dispatches the predicate to `std::all_of`, `std::any_of`, or
/// `std::none_of`.
virtual bool check_paths_impl(path_algorithm algo,
path_predicate& pred) const noexcept = 0;
/// Emits a regular (`reason == nullptr`) or irregular (`reason != nullptr`)
/// shutdown if `silent == false`.
/// @warning moves `*reason` if `reason == nullptr`
virtual void about_to_erase(map_type::iterator i, bool silent, error* reason);
virtual void about_to_erase(path_ptr ptr, bool silent, error* reason);
// -- helper functions -------------------------------------------------------
/// Delegates to `check_paths_impl`.
template <class Predicate>
bool check_paths(path_algorithm algorithm,
Predicate predicate) const noexcept {
struct impl : path_predicate {
Predicate fun;
impl(Predicate x) : fun(std::move(x)) {
// nop
}
bool operator()(const outbound_path& x) const noexcept override {
return fun(x);
}
};
impl g{std::move(predicate)};
return check_paths_impl(algorithm, g);
}
// -- member variables -------------------------------------------------------
map_type paths_;
local_actor* self_;
};
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_STREAM_SCATTERER_IMPL_HPP
#define CAF_STREAM_SCATTERER_IMPL_HPP
#include <cstddef>
#include <memory>
#include "caf/stream_scatterer.hpp"
#include "caf/detail/unordered_flat_map.hpp"
namespace caf {
/// Type-erased policy for dispatching data to sinks.
class stream_scatterer_impl : public stream_scatterer {
public:
// -- member types -----------------------------------------------------------
/// Base type.
using super = stream_scatterer;
/// Maps slots to paths.
using map_type = detail::unordered_flat_map<stream_slot, unique_path_ptr>;
// -- constructors, destructors, and assignment operators --------------------
explicit stream_scatterer_impl(local_actor* self);
virtual ~stream_scatterer_impl();
// -- path management --------------------------------------------------------
size_t num_paths() const noexcept override;
path_ptr add_path(stream_slots slots, strong_actor_ptr target) override;
unique_path_ptr take_path(stream_slot slots) noexcept override;
path_ptr path(stream_slot slots) noexcept override;
void clear_paths() override;
protected:
void for_each_path_impl(path_visitor& f) override;
bool check_paths_impl(path_algorithm algo,
path_predicate& pred) const noexcept override;
// -- member variables -------------------------------------------------------
map_type paths_;
};
} // namespace caf
#endif // CAF_STREAM_SCATTERER_IMPL_HPP
......@@ -19,6 +19,7 @@
#include "caf/invalid_stream_scatterer.hpp"
#include "caf/logger.hpp"
#include "caf/outbound_path.hpp"
#include "caf/stream.hpp"
namespace caf {
......@@ -32,6 +33,24 @@ invalid_stream_scatterer::~invalid_stream_scatterer() {
// nop
}
size_t invalid_stream_scatterer::num_paths() const noexcept {
return 0;
}
auto invalid_stream_scatterer::add_path(stream_slots, strong_actor_ptr)
-> path_ptr {
return nullptr;
}
auto invalid_stream_scatterer::take_path(stream_slot) noexcept
-> unique_path_ptr {
return unique_path_ptr{nullptr};
}
auto invalid_stream_scatterer::path(stream_slot) noexcept -> path_ptr {
return nullptr;
}
void invalid_stream_scatterer::emit_batches() {
// nop
}
......@@ -52,4 +71,26 @@ message invalid_stream_scatterer::make_handshake_token(stream_slot slot) const {
return make_message(stream<message>{slot});
}
void invalid_stream_scatterer::for_each_path_impl(path_visitor&) {
// nop
}
bool invalid_stream_scatterer::check_paths_impl(path_algorithm algo,
path_predicate&)
const noexcept {
// Return the result for empty ranges as specified by the C++ standard.
switch (algo) {
default: // all_of
return true;
case path_algorithm::any_of:
return false;
case path_algorithm::none_of:
return true;
}
}
void invalid_stream_scatterer::clear_paths() {
// nop
}
} // namespace caf
......@@ -28,140 +28,81 @@
namespace caf {
namespace {
using pointer = stream_scatterer::path_ptr;
using unique_pointer = stream_scatterer::path_unique_ptr;
} // namespace <anonymous>
stream_scatterer::stream_scatterer(local_actor* self) : self_(self) {
stream_scatterer::stream_scatterer::path_visitor::~path_visitor() {
// nop
}
stream_scatterer::~stream_scatterer() {
stream_scatterer::stream_scatterer::path_predicate::~path_predicate() {
// nop
}
pointer stream_scatterer::add_path(stream_slots slots,
strong_actor_ptr target) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(target));
auto res = paths_.emplace(slots.sender, nullptr);
if (res.second) {
auto ptr = new outbound_path(slots, std::move(target));
res.first->second.reset(ptr);
return ptr;
}
return nullptr;
}
size_t stream_scatterer::num_paths() const noexcept {
return paths_.size();
}
unique_pointer stream_scatterer::take_path(stream_slot slot) noexcept {
unique_pointer result;
auto i = paths_.find(slot);
if (i != paths_.end()) {
result.swap(i->second);
paths_.erase(i);
}
return result;
stream_scatterer::stream_scatterer(local_actor* self) : self_(self) {
// nop
}
pointer stream_scatterer::path(stream_slot slot) noexcept {
auto i = paths_.find(slot);
return i != paths_.end() ? i->second.get() : nullptr;
stream_scatterer::~stream_scatterer() {
// nop
}
bool stream_scatterer::clean() const noexcept {
auto pred = [](const map_type::value_type& kvp) {
auto& p = *kvp.second;
return p.next_batch_id == p.next_ack_id;
auto pred = [](const outbound_path& x) {
return x.next_batch_id == x.next_ack_id;
};
return buffered() == 0 && std::all_of(paths_.begin(), paths_.end(), pred);
return buffered() == 0 && all_paths(pred);
}
void stream_scatterer::close() {
CAF_LOG_TRACE(CAF_ARG(paths_));
if (paths_.empty())
return;
for (auto i = paths_.begin(); i != paths_.end(); ++i)
about_to_erase(i, false, nullptr);
paths_.clear();
CAF_LOG_TRACE("");
for_each_path([&](outbound_path& x) { about_to_erase(&x, false, nullptr); });
clear_paths();
}
void stream_scatterer::abort(error reason) {
CAF_LOG_TRACE(CAF_ARG(reason) << CAF_ARG(paths_));
if (paths_.empty())
return;
auto i = paths_.begin();
auto s = paths_.end() - 1;
for (; i != s; ++i) {
CAF_LOG_TRACE(CAF_ARG(reason));
for_each_path([&](outbound_path& x) {
auto tmp = reason;
about_to_erase(i, false, &tmp);
}
about_to_erase(i, false, &reason);
paths_.clear();
}
namespace {
template <class BinOp>
struct accumulate_helper {
using value_type = stream_scatterer::map_type::value_type;
BinOp op;
size_t operator()(size_t x, const value_type& y) const noexcept {
return op(x, y.second->open_credit);
}
size_t operator()(const value_type& x, size_t y) const noexcept {
return op(x.second->open_credit, y);
}
};
template <class BinOp>
struct accumulate_helper<BinOp> make_accumulate_helper(BinOp f) {
return {f};
about_to_erase(&x, false, &tmp);
});
clear_paths();
}
template <class F, class T, class Container>
T fold(F fun, T init, const Container& xs) {
return std::accumulate(xs.begin(), xs.end(), std::move(init),
make_accumulate_helper(std::move(fun)));
}
using bin_op_fptr = const size_t& (*)(const size_t&, const size_t&);
} // namespace <anonymous>
size_t stream_scatterer::min_credit() const noexcept {
size_t stream_scatterer::min_credit() const {
if (empty())
return 0u;
return fold(static_cast<bin_op_fptr>(&std::min<size_t>),
std::numeric_limits<size_t>::max(), paths_);
auto result = std::numeric_limits<size_t>::max();
const_cast<stream_scatterer*>(this)->for_each_path([&](outbound_path& x) {
result = std::min(result, static_cast<size_t>(x.open_credit));
});
return result;
}
size_t stream_scatterer::max_credit() const noexcept {
return fold(static_cast<bin_op_fptr>(&std::max<size_t>),
size_t{0u}, paths_);
size_t stream_scatterer::max_credit() const {
size_t result = 0;
const_cast<stream_scatterer*>(this)->for_each_path([&](outbound_path& x) {
result = std::max(result, static_cast<size_t>(x.open_credit));
});
return result;
}
size_t stream_scatterer::total_credit() const noexcept {
return fold(std::plus<size_t>{}, size_t{0u}, paths_);
size_t stream_scatterer::total_credit() const {
size_t result = 0;
const_cast<stream_scatterer*>(this)->for_each_path([&](outbound_path& x) {
result += static_cast<size_t>(x.open_credit);
});
return result;
}
bool stream_scatterer::terminal() const noexcept {
return false;
}
void stream_scatterer::about_to_erase(map_type::iterator i, bool silent,
void stream_scatterer::about_to_erase(outbound_path* ptr, bool silent,
error* reason) {
if (!silent) {
if (reason == nullptr)
i->second->emit_regular_shutdown(self_);
ptr->emit_regular_shutdown(self_);
else
i->second->emit_irregular_shutdown(self_, std::move(*reason));
ptr->emit_irregular_shutdown(self_, std::move(*reason));
}
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/stream_scatterer_impl.hpp"
#include <functional>
#include "caf/logger.hpp"
#include "caf/outbound_path.hpp"
namespace caf {
stream_scatterer_impl::stream_scatterer_impl(local_actor* self) : super(self) {
// nop
}
stream_scatterer_impl::~stream_scatterer_impl() {
// nop
}
outbound_path* stream_scatterer_impl::add_path(stream_slots slots,
strong_actor_ptr target) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(target));
auto res = paths_.emplace(slots.sender, nullptr);
if (res.second) {
auto ptr = new outbound_path(slots, std::move(target));
res.first->second.reset(ptr);
return ptr;
}
return nullptr;
}
size_t stream_scatterer_impl::num_paths() const noexcept {
return paths_.size();
}
auto stream_scatterer_impl::take_path(stream_slot slot) noexcept
-> unique_path_ptr {
unique_path_ptr result;
auto i = paths_.find(slot);
if (i != paths_.end()) {
about_to_erase(i->second.get(), true, nullptr);
result.swap(i->second);
paths_.erase(i);
}
return result;
}
auto stream_scatterer_impl::path(stream_slot slot) noexcept -> path_ptr {
auto i = paths_.find(slot);
return i != paths_.end() ? i->second.get() : nullptr;
}
void stream_scatterer_impl::clear_paths() {
paths_.clear();
}
void stream_scatterer_impl::for_each_path_impl(path_visitor& f) {
for (auto& kvp : paths_)
f(*kvp.second);
}
bool stream_scatterer_impl::check_paths_impl(path_algorithm algo,
path_predicate& pred) const noexcept {
auto f = [&](const map_type::value_type& x) {
return pred(*x.second);
};
switch (algo) {
default: // all_of
return std::all_of(paths_.begin(), paths_.end(), f);
case path_algorithm::any_of:
return std::any_of(paths_.begin(), paths_.end(), f);
case path_algorithm::none_of:
return std::none_of(paths_.begin(), paths_.end(), f);
}
}
} // namespace caf
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment