Commit 3e2ac362 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Update in-/output paths and remove gatherers

parent c9586ff1
...@@ -54,7 +54,6 @@ set (LIBCAF_CORE_SRCS ...@@ -54,7 +54,6 @@ set (LIBCAF_CORE_SRCS
src/group_manager.cpp src/group_manager.cpp
src/group_module.cpp src/group_module.cpp
src/inbound_path.cpp src/inbound_path.cpp
src/invalid_stream_gatherer.cpp
src/invalid_stream_scatterer.cpp src/invalid_stream_scatterer.cpp
src/invoke_result_visitor.cpp src/invoke_result_visitor.cpp
src/local_actor.cpp src/local_actor.cpp
...@@ -75,8 +74,6 @@ set (LIBCAF_CORE_SRCS ...@@ -75,8 +74,6 @@ set (LIBCAF_CORE_SRCS
src/pretty_type_name.cpp src/pretty_type_name.cpp
src/private_thread.cpp src/private_thread.cpp
src/proxy_registry.cpp src/proxy_registry.cpp
src/pull5_gatherer.cpp
src/random_gatherer.cpp
src/raw_event_based_actor.cpp src/raw_event_based_actor.cpp
src/ref_counted.cpp src/ref_counted.cpp
src/replies_to.cpp src/replies_to.cpp
...@@ -95,9 +92,6 @@ set (LIBCAF_CORE_SRCS ...@@ -95,9 +92,6 @@ set (LIBCAF_CORE_SRCS
src/splitter.cpp src/splitter.cpp
src/stream.cpp src/stream.cpp
src/stream_aborter.cpp src/stream_aborter.cpp
src/stream_gatherer.cpp
src/stream_gatherer.cpp
src/stream_gatherer_impl.cpp
src/stream_manager.cpp src/stream_manager.cpp
src/stream_priority.cpp src/stream_priority.cpp
src/stream_scatterer.cpp src/stream_scatterer.cpp
......
...@@ -79,7 +79,6 @@ ...@@ -79,7 +79,6 @@
#include "caf/message_handler.hpp" #include "caf/message_handler.hpp"
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/fused_scatterer.hpp" #include "caf/fused_scatterer.hpp"
#include "caf/random_gatherer.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/abstract_channel.hpp" #include "caf/abstract_channel.hpp"
#include "caf/may_have_timeout.hpp" #include "caf/may_have_timeout.hpp"
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_PULL5_GATHERER_HPP
#define CAF_PULL5_GATHERER_HPP
#include "caf/random_gatherer.hpp"
namespace caf {
namespace detail {
/// Always pulls exactly 5 elements from sources. Used in unit tests only.
class pull5_gatherer : public random_gatherer {
public:
using super = random_gatherer;
pull5_gatherer(local_actor* selfptr, stream_scatterer&);
void assign_credit(long available) override;
long initial_credit(long, inbound_path*) override;
};
} // namespace detail
} // namespace caf
#endif // CAF_PULL5_GATHERER_HPP
...@@ -24,7 +24,6 @@ ...@@ -24,7 +24,6 @@
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/random_gatherer.hpp"
#include "caf/scheduled_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/broadcast_scatterer.hpp" #include "caf/broadcast_scatterer.hpp"
...@@ -62,9 +61,6 @@ namespace detail { ...@@ -62,9 +61,6 @@ namespace detail {
/// /// Groups multiple topics into a single selection filter. /// /// Groups multiple topics into a single selection filter.
/// using filter_type = ...; /// using filter_type = ...;
/// ///
/// /// Policy for gathering data from peers.
/// using gatherer_type = ...;
///
/// /// Policy for scattering data to peers. /// /// Policy for scattering data to peers.
/// using scatterer_type = ...; /// using scatterer_type = ...;
/// ///
...@@ -105,8 +101,6 @@ public: ...@@ -105,8 +101,6 @@ public:
using super = stream_manager; using super = stream_manager;
using gatherer_type = typename Policy::gatherer_type;
using scatterer_type = typename Policy::scatterer_type; using scatterer_type = typename Policy::scatterer_type;
// --- constructors and destructors ------------------------------------------ // --- constructors and destructors ------------------------------------------
...@@ -186,10 +180,6 @@ public: ...@@ -186,10 +180,6 @@ public:
return false; return false;
} }
gatherer_type& in() override {
return in_;
}
scatterer_type& out() override { scatterer_type& out() override {
return out_; return out_;
} }
...@@ -235,7 +225,6 @@ public: ...@@ -235,7 +225,6 @@ public:
private: private:
scheduled_actor* self_; scheduled_actor* self_;
gatherer_type in_;
scatterer_type out_; scatterer_type out_;
Policy policy_; Policy policy_;
}; };
......
...@@ -83,8 +83,8 @@ struct downstream_msg : tag::boxing_type { ...@@ -83,8 +83,8 @@ struct downstream_msg : tag::boxing_type {
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
template <class T> template <class T>
downstream_msg(stream_slot id, actor_addr addr, T&& x) downstream_msg(stream_slots s, actor_addr addr, T&& x)
: slot(id), : slots(s),
sender(std::move(addr)), sender(std::move(addr)),
content(std::forward<T>(x)) { content(std::forward<T>(x)) {
// nop // nop
...@@ -99,7 +99,7 @@ struct downstream_msg : tag::boxing_type { ...@@ -99,7 +99,7 @@ struct downstream_msg : tag::boxing_type {
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// ID of the affected stream. /// ID of the affected stream.
stream_slot slot; stream_slots slots;
/// Address of the sender. Identifies the up- or downstream actor sending /// Address of the sender. Identifies the up- or downstream actor sending
/// this message. Note that abort messages can get send after `sender` /// this message. Note that abort messages can get send after `sender`
...@@ -128,8 +128,8 @@ bool is(const downstream_msg& x) { ...@@ -128,8 +128,8 @@ bool is(const downstream_msg& x) {
template <class T, class... Ts> template <class T, class... Ts>
detail::enable_if_tt<detail::tl_contains<downstream_msg::alternatives, T>, detail::enable_if_tt<detail::tl_contains<downstream_msg::alternatives, T>,
downstream_msg> downstream_msg>
make(stream_slot slot, actor_addr addr, Ts&&... xs) { make(stream_slots slots, actor_addr addr, Ts&&... xs) {
return {slot, std::move(addr), T{std::forward<Ts>(xs)...}}; return {slots, std::move(addr), T{std::forward<Ts>(xs)...}};
} }
/// @relates downstream_msg::batch /// @relates downstream_msg::batch
...@@ -155,7 +155,7 @@ typename Inspector::result_type inspect(Inspector& f, ...@@ -155,7 +155,7 @@ typename Inspector::result_type inspect(Inspector& f,
/// @relates downstream_msg /// @relates downstream_msg
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, downstream_msg& x) { typename Inspector::result_type inspect(Inspector& f, downstream_msg& x) {
return f(meta::type_name("stream_msg"), x.slot, x.sender, x.content); return f(meta::type_name("stream_msg"), x.slots, x.sender, x.content);
} }
} // namespace caf } // namespace caf
......
...@@ -93,8 +93,6 @@ class blocking_actor; ...@@ -93,8 +93,6 @@ class blocking_actor;
class execution_unit; class execution_unit;
class proxy_registry; class proxy_registry;
class stream_manager; class stream_manager;
class random_gatherer;
class stream_gatherer;
class actor_companion; class actor_companion;
class mailbox_element; class mailbox_element;
class message_handler; class message_handler;
......
...@@ -44,12 +44,12 @@ public: ...@@ -44,12 +44,12 @@ public:
/// Message type for propagating errors. /// Message type for propagating errors.
using irregular_shutdown = upstream_msg::forced_drop; using irregular_shutdown = upstream_msg::forced_drop;
/// Slot IDs for sender (hdl) and receiver (self).
stream_slots slots;
/// Pointer to the parent actor. /// Pointer to the parent actor.
local_actor* self; local_actor* self;
/// Slot ID for the source.
stream_slot slot;
/// Handle to the source. /// Handle to the source.
strong_actor_ptr hdl; strong_actor_ptr hdl;
...@@ -77,7 +77,7 @@ public: ...@@ -77,7 +77,7 @@ public:
error shutdown_reason; error shutdown_reason;
/// Constructs a path for given handle and stream ID. /// Constructs a path for given handle and stream ID.
inbound_path(local_actor* selfptr, stream_slot id, strong_actor_ptr ptr); inbound_path(local_actor* selfptr, stream_slots id, strong_actor_ptr ptr);
~inbound_path(); ~inbound_path();
...@@ -91,7 +91,7 @@ public: ...@@ -91,7 +91,7 @@ public:
void emit_ack_batch(long new_demand); void emit_ack_batch(long new_demand);
static void emit_irregular_shutdown(local_actor* self, stream_slot sid, static void emit_irregular_shutdown(local_actor* self, stream_slots sid,
const strong_actor_ptr& hdl, const strong_actor_ptr& hdl,
error reason); error reason);
}; };
...@@ -99,7 +99,7 @@ public: ...@@ -99,7 +99,7 @@ public:
/// @relates inbound_path /// @relates inbound_path
template <class Inspector> template <class Inspector>
typename Inspector::return_type inspect(Inspector& f, inbound_path& x) { typename Inspector::return_type inspect(Inspector& f, inbound_path& x) {
return f(meta::type_name("inbound_path"), x.hdl, x.slot, x.prio, return f(meta::type_name("inbound_path"), x.hdl, x.slots, x.prio,
x.last_acked_batch_id, x.last_batch_id, x.assigned_credit); x.last_acked_batch_id, x.last_batch_id, x.assigned_credit);
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_INVALID_STREAM_GATHERER_HPP
#define CAF_INVALID_STREAM_GATHERER_HPP
#include "caf/stream_gatherer.hpp"
namespace caf {
/// Type-erased policy for receiving data from sources.
class invalid_stream_gatherer : public stream_gatherer {
public:
invalid_stream_gatherer() = default;
~invalid_stream_gatherer() override;
path_ptr add_path(stream_slot slot, strong_actor_ptr x,
strong_actor_ptr original_stage, stream_priority prio,
long available_credit, bool redeployable,
response_promise result_cb) override;
bool remove_path(stream_slot slot, const actor_addr& x, error reason,
bool silent) override;
void close(message reason) override;
void abort(error reason) override;
long num_paths() const override;
bool closed() const override;
bool continuous() const override;
void continuous(bool value) override;
path_ptr path_at(size_t index) override;
path_ptr find(stream_slot slot, const actor_addr& x) override;
void assign_credit(long downstream_capacity) override;
long initial_credit(long downstream_capacity, path_type* x) override;
};
} // namespace caf
#endif // CAF_INVALID_STREAM_GATHERER_HPP
...@@ -134,6 +134,10 @@ public: ...@@ -134,6 +134,10 @@ public:
return *this; return *this;
} }
const type_erased_tuple& content() const override {
return *this;
}
message move_content_to_message() override { message move_content_to_message() override {
message_factory f; message_factory f;
auto& xs = this->data(); auto& xs = this->data();
...@@ -170,6 +174,10 @@ public: ...@@ -170,6 +174,10 @@ public:
return *this; return *this;
} }
const type_erased_tuple& content() const override {
return *this;
}
message move_content_to_message() override { message move_content_to_message() override {
message_factory f; message_factory f;
auto& xs = this->data(); auto& xs = this->data();
......
...@@ -303,7 +303,12 @@ public: ...@@ -303,7 +303,12 @@ public:
/// @cond PRIVATE /// @cond PRIVATE
/// @pre `!empty()` /// @pre `!empty()`
type_erased_tuple& content() { inline type_erased_tuple& content() {
CAF_ASSERT(vals_);
return *vals_;
}
inline const type_erased_tuple& content() const {
CAF_ASSERT(vals_); CAF_ASSERT(vals_);
return *vals_; return *vals_;
} }
......
...@@ -31,6 +31,8 @@ public: ...@@ -31,6 +31,8 @@ public:
virtual type_erased_tuple& content() = 0; virtual type_erased_tuple& content() = 0;
virtual const type_erased_tuple& content() const = 0;
virtual message move_content_to_message() = 0; virtual message move_content_to_message() = 0;
virtual message copy_content_to_message() const = 0; virtual message copy_content_to_message() const = 0;
......
...@@ -52,8 +52,8 @@ public: ...@@ -52,8 +52,8 @@ public:
message_id mid; message_id mid;
}; };
/// Slot ID for the sink. /// Slot IDs for sender (self) and receiver (hdl).
stream_slot slot; stream_slots slots;
/// Pointer to the parent actor. /// Pointer to the parent actor.
local_actor* self; local_actor* self;
...@@ -92,7 +92,7 @@ public: ...@@ -92,7 +92,7 @@ public:
error shutdown_reason; error shutdown_reason;
/// Constructs a path for given handle and stream ID. /// Constructs a path for given handle and stream ID.
outbound_path(local_actor* selfptr, stream_slot id, outbound_path(local_actor* selfptr, stream_slots id,
strong_actor_ptr ptr); strong_actor_ptr ptr);
~outbound_path(); ~outbound_path();
...@@ -109,7 +109,7 @@ public: ...@@ -109,7 +109,7 @@ public:
/// `xs_size` and increments `next_batch_id` by 1. /// `xs_size` and increments `next_batch_id` by 1.
void emit_batch(long xs_size, message xs); void emit_batch(long xs_size, message xs);
static void emit_irregular_shutdown(local_actor* self, stream_slot slot, static void emit_irregular_shutdown(local_actor* self, stream_slots slots,
const strong_actor_ptr& hdl, const strong_actor_ptr& hdl,
error reason); error reason);
}; };
...@@ -117,7 +117,7 @@ public: ...@@ -117,7 +117,7 @@ public:
/// @relates outbound_path /// @relates outbound_path
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, outbound_path& x) { typename Inspector::result_type inspect(Inspector& f, outbound_path& x) {
return f(meta::type_name("outbound_path"), x.hdl, x.slot, x.next_batch_id, return f(meta::type_name("outbound_path"), x.hdl, x.slots, x.next_batch_id,
x.open_credit, x.redeployable, x.unacknowledged_batches); x.open_credit, x.redeployable, x.unacknowledged_batches);
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_RANDOM_GATHERER_HPP
#define CAF_RANDOM_GATHERER_HPP
#include "caf/fwd.hpp"
#include "caf/stream_gatherer_impl.hpp"
namespace caf {
/// Pulls data from sources in arbitrary order.
class random_gatherer : public stream_gatherer_impl {
public:
using super = stream_gatherer_impl;
random_gatherer(local_actor* selfptr, stream_scatterer&);
~random_gatherer() override;
void assign_credit(long downstream_capacity) override;
long initial_credit(long downstream_capacity, path_type* x) override;
};
} // namespace caf
#endif // CAF_RANDOM_GATHERER_HPP
...@@ -37,7 +37,6 @@ ...@@ -37,7 +37,6 @@
#include "caf/stream_result.hpp" #include "caf/stream_result.hpp"
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/scheduled_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/random_gatherer.hpp"
#include "caf/stream_sink_impl.hpp" #include "caf/stream_sink_impl.hpp"
#include "caf/stream_stage_impl.hpp" #include "caf/stream_stage_impl.hpp"
#include "caf/stream_source_impl.hpp" #include "caf/stream_source_impl.hpp"
......
...@@ -35,13 +35,13 @@ public: ...@@ -35,13 +35,13 @@ public:
struct token { struct token {
const actor_addr& observer; const actor_addr& observer;
stream_slot slot; stream_slots slots;
mode m; mode m;
static constexpr size_t token_type = attachable::token::stream_aborter; static constexpr size_t token_type = attachable::token::stream_aborter;
}; };
stream_aborter(actor_addr&& observed, actor_addr&& observer, stream_aborter(actor_addr&& observed, actor_addr&& observer,
stream_slot slot, mode m); stream_slots slots, mode m);
~stream_aborter() override; ~stream_aborter() override;
...@@ -51,21 +51,21 @@ public: ...@@ -51,21 +51,21 @@ public:
/// Adds a stream aborter to `observed`. /// Adds a stream aborter to `observed`.
static void add(strong_actor_ptr observed, actor_addr observer, static void add(strong_actor_ptr observed, actor_addr observer,
stream_slot slot, mode m); stream_slots slots, mode m);
/// Removes a stream aborter from `observed`. /// Removes a stream aborter from `observed`.
static void del(strong_actor_ptr observed, const actor_addr& observer, static void del(strong_actor_ptr observed, const actor_addr& observer,
stream_slot slot, mode m); stream_slots slots, mode m);
private: private:
actor_addr observed_; actor_addr observed_;
actor_addr observer_; actor_addr observer_;
stream_slot slot_; stream_slots slots_;
mode mode_; mode mode_;
}; };
attachable_ptr make_stream_aborter(actor_addr observed, actor_addr observer, attachable_ptr make_stream_aborter(actor_addr observed, actor_addr observer,
stream_slot slot, stream_aborter::mode m); stream_slots slot, stream_aborter::mode m);
} // namespace caf } // namespace caf
......
...@@ -111,26 +111,23 @@ public: ...@@ -111,26 +111,23 @@ public:
/// Finds the path for `ptr` and returns a pointer to it. /// Finds the path for `ptr` and returns a pointer to it.
template <class PathContainer, class Handle> template <class PathContainer, class Handle>
static path_ptr find(PathContainer& xs, stream_slot slot, static path_ptr find(PathContainer&, stream_slot, const Handle&) {
const Handle& x) { // TODO: implement me
auto predicate = [&](const typename PathContainer::value_type& y) {
return y->hdl == x && y->slot == slot;
};
auto e = xs.end();
auto i = std::find_if(xs.begin(), e, predicate);
if (i != e)
return &(*(*i)); // Ugly, but works for both raw and smart pointers.
return nullptr; return nullptr;
} }
/// Finds the path for `ptr` and returns an iterator to it. /// Finds the path for `ptr` and returns an iterator to it.
template <class PathContainer, class Handle> template <class PathContainer, class Handle>
static typename PathContainer::iterator static typename PathContainer::iterator
iter_find(PathContainer& xs, stream_slot slot, const Handle& x) { iter_find(PathContainer& xs, stream_slot, const Handle&) {
// TODO: implement me
return xs.end();
/*
auto predicate = [&](const typename PathContainer::value_type& y) { auto predicate = [&](const typename PathContainer::value_type& y) {
return y->hdl == x && y->slot == slot; return y->hdl == x && y->slot == slot;
}; };
return std::find_if(xs.begin(), xs.end(), predicate); return std::find_if(xs.begin(), xs.end(), predicate);
*/
} }
// -- accessors -------------------------------------------------------------- // -- accessors --------------------------------------------------------------
...@@ -149,7 +146,10 @@ public: ...@@ -149,7 +146,10 @@ public:
using super::remove_path; using super::remove_path;
bool remove_path(path_uptr_iter i, error reason, bool silent) { bool remove_path(path_uptr_iter, error, bool) {
// TODO: implement me
return true;
/*
CAF_LOG_TRACE(CAF_ARG(reason) << CAF_ARG(silent)); CAF_LOG_TRACE(CAF_ARG(reason) << CAF_ARG(silent));
auto e = paths_.end(); auto e = paths_.end();
if (i == e) { if (i == e) {
...@@ -166,6 +166,7 @@ public: ...@@ -166,6 +166,7 @@ public:
std::swap(*i, paths_.back()); std::swap(*i, paths_.back());
paths_.pop_back(); paths_.pop_back();
return true; return true;
*/
} }
// -- implementation of common methods --------------------------------------- // -- implementation of common methods ---------------------------------------
...@@ -220,11 +221,15 @@ public: ...@@ -220,11 +221,15 @@ public:
protected: protected:
/// Adds a path to the edge without emitting messages. /// Adds a path to the edge without emitting messages.
path_ptr add_path_impl(stream_slot slot, strong_actor_ptr x) { path_ptr add_path_impl(stream_slot, strong_actor_ptr) {
// TODO: implement me
return nullptr;
/*
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(slot)); CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(slot));
stream_aborter::add(x, self_->address(), slot, aborter_type); stream_aborter::add(x, self_->address(), slot, aborter_type);
paths_.emplace_back(new path_type(self_, slot, std::move(x))); paths_.emplace_back(new path_type(self_, slot, std::move(x)));
return paths_.back().get(); return paths_.back().get();
*/
} }
template <class F> template <class F>
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_GATHERER_HPP
#define CAF_STREAM_GATHERER_HPP
#include <vector>
#include <cstdint>
#include <utility>
#include "caf/fwd.hpp"
#include "caf/stream_slot.hpp"
namespace caf {
/// Type-erased policy for receiving data from sources.
class stream_gatherer {
public:
// -- member types -----------------------------------------------------------
/// Type of a single path to a data source.
using path_type = inbound_path;
/// Pointer to a single path to a data source.
using path_ptr = path_type*;
// -- constructors, destructors, and assignment operators --------------------
stream_gatherer() = default;
virtual ~stream_gatherer();
// -- pure virtual memeber functions -----------------------------------------
/// Adds a path to the edge and emits `ack_open` to the source.
/// @param slot Stream ID used by the source.
/// @param x Handle to the source.
/// @param original_stage Actor that received the stream handshake initially.
/// @param prio Priority of data on this path.
/// @param available_credit Maximum credit for granting to the source.
/// @param redeployable Stores whether the source can re-appear after aborts.
/// @param result_cb Callback for the listener of the final stream result.
/// The gatherer must ignore the promise when returning
/// `nullptr`, because the previous stage is responsible for
/// it until the gatherer acknowledges the handshake. The
/// callback is invalid if the stream has a next stage.
/// @returns The added path on success, `nullptr` otherwise.
virtual path_ptr add_path(stream_slot slot, strong_actor_ptr x,
strong_actor_ptr original_stage,
stream_priority prio, long available_credit,
bool redeployable, response_promise result_cb) = 0;
/// Removes a path from the gatherer.
virtual bool remove_path(stream_slot slot, const actor_addr& x,
error reason, bool silent) = 0;
/// Removes all paths gracefully.
virtual void close(message result) = 0;
/// Removes all paths with an error message.
virtual void abort(error reason) = 0;
/// Returns the number of paths managed on this edge.
virtual long num_paths() const = 0;
/// Returns `true` if no downstream exists, `false` otherwise.
virtual bool closed() const = 0;
/// Returns whether this edge remains open after the last path is removed.
virtual bool continuous() const = 0;
/// Sets whether this edge remains open after the last path is removed.
virtual void continuous(bool value) = 0;
/// Returns the stored state for `x` if `x` is a known path and associated to
/// `slot`, otherwise `nullptr`.
virtual path_ptr find(stream_slot slot, const actor_addr& x) = 0;
/// Returns the stored state for `x` if `x` is a known path and associated to
/// `slot`, otherwise `nullptr`.
virtual path_ptr path_at(size_t index) = 0;
/// Assigns new credit to all sources.
virtual void assign_credit(long downstream_capacity) = 0;
/// Calculates initial credit for `x` after adding it to the gatherer.
virtual long initial_credit(long downstream_capacity, path_ptr x) = 0;
// -- convenience functions --------------------------------------------------
/// Removes a path from the gatherer.
bool remove_path(stream_slot slot, const strong_actor_ptr& x,
error reason, bool silent);
/// Convenience function for calling `find(x, actor_cast<actor_addr>(x))`.
path_ptr find(stream_slot slot, const strong_actor_ptr& x);
};
} // namespace caf
#endif // CAF_STREAM_GATHERER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_GATHERER_IMPL_HPP
#define CAF_STREAM_GATHERER_IMPL_HPP
#include <vector>
#include <cstdint>
#include <utility>
#include "caf/fwd.hpp"
#include "caf/stream_gatherer.hpp"
#include "caf/stream_edge_impl.hpp"
#include "caf/response_promise.hpp"
namespace caf {
/// Type-erased policy for receiving data from sources.
class stream_gatherer_impl : public stream_edge_impl<stream_gatherer> {
public:
using super = stream_edge_impl<stream_gatherer>;
using assignment_pair = std::pair<path_type*, long>;
stream_gatherer_impl(local_actor* selfptr);
~stream_gatherer_impl();
path_ptr add_path(stream_slot slot, strong_actor_ptr x,
strong_actor_ptr original_stage, stream_priority prio,
long available_credit, bool redeployable,
response_promise result_cb) override;
bool remove_path(stream_slot slot, const actor_addr& x, error reason,
bool silent) override;
void close(message result) override;
void abort(error reason) override;
protected:
void emit_credits();
std::vector<assignment_pair> assignment_vec_;
/// Listeners for the final result.
std::vector<response_promise> listeners_;
};
} // namespace caf
#endif // CAF_STREAM_GATHERER_IMPL_HPP
...@@ -144,9 +144,6 @@ public: ...@@ -144,9 +144,6 @@ public:
// -- implementation hooks for all children classes -------------------------- // -- implementation hooks for all children classes --------------------------
/// Returns the stream edge for incoming data.
virtual stream_gatherer& in() = 0;
/// Returns the stream edge for outgoing data. /// Returns the stream edge for outgoing data.
virtual stream_scatterer& out() = 0; virtual stream_scatterer& out() = 0;
......
...@@ -20,7 +20,6 @@ ...@@ -20,7 +20,6 @@
#define CAF_STREAM_SLOT_HPP #define CAF_STREAM_SLOT_HPP
#include <cstdint> #include <cstdint>
#include <tuple>
#include "caf/detail/comparable.hpp" #include "caf/detail/comparable.hpp"
...@@ -38,6 +37,10 @@ struct stream_slots : detail::comparable<stream_slots>{ ...@@ -38,6 +37,10 @@ struct stream_slots : detail::comparable<stream_slots>{
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
constexpr stream_slots() : sender(0), receiver(0) {
// nop
}
constexpr stream_slots(stream_slot sender_slot, stream_slot receiver_slot) constexpr stream_slots(stream_slot sender_slot, stream_slot receiver_slot)
: sender(sender_slot), : sender(sender_slot),
receiver(receiver_slot) { receiver(receiver_slot) {
...@@ -62,7 +65,7 @@ struct stream_slots : detail::comparable<stream_slots>{ ...@@ -62,7 +65,7 @@ struct stream_slots : detail::comparable<stream_slots>{
/// @relates stream_slots /// @relates stream_slots
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_slots& x) { typename Inspector::result_type inspect(Inspector& f, stream_slots& x) {
return f(std::tie(x.sender, x.receiver)); return f(x.sender, x.receiver);
} }
} // namespace caf } // namespace caf
......
...@@ -24,7 +24,6 @@ ...@@ -24,7 +24,6 @@
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
#include "caf/stream_manager.hpp" #include "caf/stream_manager.hpp"
#include "caf/stream_source_trait.hpp" #include "caf/stream_source_trait.hpp"
#include "caf/invalid_stream_gatherer.hpp"
namespace caf { namespace caf {
...@@ -48,9 +47,6 @@ public: ...@@ -48,9 +47,6 @@ public:
return at_end() && out_.paths_clean(); return at_end() && out_.paths_clean();
} }
invalid_stream_gatherer& in() override {
return in_;
}
DownstreamPolicy& out() override { DownstreamPolicy& out() override {
return out_; return out_;
...@@ -88,22 +84,8 @@ protected: ...@@ -88,22 +84,8 @@ protected:
return pred_(state_); return pred_(state_);
} }
void downstream_demand(outbound_path* path, long) override { void downstream_demand(outbound_path*, long) override {
CAF_LOG_TRACE(CAF_ARG(path)); // TODO: implment me
if (!at_end()) {
auto dbs = out_.desired_batch_size();
generate_messages();
while (out_.buffered() >= dbs && out_.credit() >= dbs) {
push();
generate_messages();
}
} else if (out_.buffered() > 0) {
push();
} else {
auto sid = path->slot;
auto hdl = path->hdl;
out().remove_path(sid, hdl, none, false);
}
} }
private: private:
...@@ -111,7 +93,6 @@ private: ...@@ -111,7 +93,6 @@ private:
Fun fun_; Fun fun_;
Predicate pred_; Predicate pred_;
DownstreamPolicy out_; DownstreamPolicy out_;
invalid_stream_gatherer in_;
}; };
} // namespace caf } // namespace caf
......
...@@ -93,20 +93,8 @@ protected: ...@@ -93,20 +93,8 @@ protected:
return make_message(stream<output_type>{x}); return make_message(stream<output_type>{x});
} }
void downstream_demand(outbound_path* path, long) override { void downstream_demand(outbound_path*, long) override {
CAF_LOG_TRACE(CAF_ARG(path)); // TODO: implement me
auto hdl = path->hdl;
if(out_.buffered() > 0)
push();
else if (in_.closed()) {
// don't pass path->hdl: path can become invalid
auto sid = path->slot;
out_.remove_path(sid, hdl, none, false);
}
auto current_size = out_.buffered();
auto desired_size = out_.credit();
if (current_size < desired_size)
in_.assign_credit(desired_size - current_size);
} }
private: private:
......
...@@ -90,8 +90,8 @@ typename Inspector::result_type inspect(Inspector& f, timeout_msg& x) { ...@@ -90,8 +90,8 @@ typename Inspector::result_type inspect(Inspector& f, timeout_msg& x) {
return f(meta::type_name("timeout_msg"), x.timeout_id); return f(meta::type_name("timeout_msg"), x.timeout_id);
} }
/// Initiates a stream handhsake. /// Demands the receiver to open a new stream from the sender to the receiver.
struct stream_handshake_msg { struct open_stream_msg {
/// Reserved slot on the source. /// Reserved slot on the source.
stream_slot slot; stream_slot slot;
...@@ -114,7 +114,7 @@ struct stream_handshake_msg { ...@@ -114,7 +114,7 @@ struct stream_handshake_msg {
/// @relates stream_handshake_msg /// @relates stream_handshake_msg
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, stream_handshake_msg& x) { typename Inspector::result_type inspect(Inspector& f, open_stream_msg& x) {
return f(meta::type_name("stream_handshake_msg"), x.slot, x.msg, x.prev_stage, return f(meta::type_name("stream_handshake_msg"), x.slot, x.msg, x.prev_stage,
x.original_stage, x.priority, x.redeployable); x.original_stage, x.priority, x.redeployable);
} }
......
...@@ -109,8 +109,8 @@ struct upstream_msg : tag::boxing_type { ...@@ -109,8 +109,8 @@ struct upstream_msg : tag::boxing_type {
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
template <class T> template <class T>
upstream_msg(stream_slot id, actor_addr addr, T&& x) upstream_msg(stream_slots id, actor_addr addr, T&& x)
: slot(id), : slots(id),
sender(std::move(addr)), sender(std::move(addr)),
content(std::forward<T>(x)) { content(std::forward<T>(x)) {
// nop // nop
...@@ -124,8 +124,8 @@ struct upstream_msg : tag::boxing_type { ...@@ -124,8 +124,8 @@ struct upstream_msg : tag::boxing_type {
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// ID of the affected stream. /// Stream slots of sender and receiver.
stream_slot slot; stream_slots slots;
/// Address of the sender. Identifies the up- or downstream actor sending /// Address of the sender. Identifies the up- or downstream actor sending
/// this message. Note that abort messages can get send after `sender` /// this message. Note that abort messages can get send after `sender`
...@@ -154,8 +154,8 @@ bool is(const upstream_msg& x) { ...@@ -154,8 +154,8 @@ bool is(const upstream_msg& x) {
template <class T, class... Ts> template <class T, class... Ts>
detail::enable_if_tt<detail::tl_contains<upstream_msg::alternatives, T>, detail::enable_if_tt<detail::tl_contains<upstream_msg::alternatives, T>,
upstream_msg> upstream_msg>
make(stream_slot slot, actor_addr addr, Ts&&... xs) { make(stream_slots slots, actor_addr addr, Ts&&... xs) {
return {slot, std::move(addr), T{std::forward<Ts>(xs)...}}; return {slots, std::move(addr), T{std::forward<Ts>(xs)...}};
} }
/// @relates upstream_msg::ack_open /// @relates upstream_msg::ack_open
...@@ -190,7 +190,7 @@ typename Inspector::result_type inspect(Inspector& f, ...@@ -190,7 +190,7 @@ typename Inspector::result_type inspect(Inspector& f,
/// @relates upstream_msg /// @relates upstream_msg
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, upstream_msg& x) { typename Inspector::result_type inspect(Inspector& f, upstream_msg& x) {
return f(meta::type_name("stream_msg"), x.slot, x.sender, x.content); return f(meta::type_name("stream_msg"), x.slots, x.sender, x.content);
} }
} // namespace caf } // namespace caf
......
...@@ -25,10 +25,10 @@ ...@@ -25,10 +25,10 @@
namespace caf { namespace caf {
inbound_path::inbound_path(local_actor* selfptr, stream_slot id, inbound_path::inbound_path(local_actor* selfptr, stream_slots id,
strong_actor_ptr ptr) strong_actor_ptr ptr)
: self(selfptr), : slots(id),
slot(id), self(selfptr),
hdl(std::move(ptr)), hdl(std::move(ptr)),
prio(stream_priority::normal), prio(stream_priority::normal),
last_acked_batch_id(0), last_acked_batch_id(0),
...@@ -43,11 +43,11 @@ inbound_path::~inbound_path() { ...@@ -43,11 +43,11 @@ inbound_path::~inbound_path() {
if (hdl) { if (hdl) {
if (shutdown_reason == none) if (shutdown_reason == none)
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<upstream_msg::drop>(slot, self->address())); make<upstream_msg::drop>(slots.invert(), self->address()));
else else
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<upstream_msg::forced_drop>(slot, self->address(), make<upstream_msg::forced_drop>(
shutdown_reason)); slots.invert(), self->address(), shutdown_reason));
} }
} }
...@@ -65,7 +65,7 @@ void inbound_path::emit_ack_open(actor_addr rebind_from, ...@@ -65,7 +65,7 @@ void inbound_path::emit_ack_open(actor_addr rebind_from,
auto batch_size = static_cast<int32_t>(desired_batch_size); auto batch_size = static_cast<int32_t>(desired_batch_size);
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<upstream_msg::ack_open>( make<upstream_msg::ack_open>(
slot, self->address(), std::move(rebind_from), self->ctrl(), slots, self->address(), std::move(rebind_from), self->ctrl(),
static_cast<int32_t>(initial_demand), batch_size, static_cast<int32_t>(initial_demand), batch_size,
is_redeployable)); is_redeployable));
} }
...@@ -76,19 +76,19 @@ void inbound_path::emit_ack_batch(long new_demand) { ...@@ -76,19 +76,19 @@ void inbound_path::emit_ack_batch(long new_demand) {
assigned_credit += new_demand; assigned_credit += new_demand;
auto batch_size = static_cast<int32_t>(desired_batch_size); auto batch_size = static_cast<int32_t>(desired_batch_size);
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<upstream_msg::ack_batch>(slot, self->address(), make<upstream_msg::ack_batch>(slots, self->address(),
static_cast<int32_t>(new_demand), static_cast<int32_t>(new_demand),
batch_size, last_batch_id)); batch_size, last_batch_id));
} }
void inbound_path::emit_irregular_shutdown(local_actor* self, void inbound_path::emit_irregular_shutdown(local_actor* self,
stream_slot slot, stream_slots slots,
const strong_actor_ptr& hdl, const strong_actor_ptr& hdl,
error reason) { error reason) {
CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(hdl) << CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(hdl) << CAF_ARG(reason));
unsafe_send_as( unsafe_send_as(
self, hdl, self, hdl,
make<upstream_msg::forced_drop>(slot, self->address(), std::move(reason))); make<upstream_msg::forced_drop>(slots, self->address(), std::move(reason)));
} }
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/invalid_stream_gatherer.hpp"
#include "caf/logger.hpp"
#include "caf/response_promise.hpp"
namespace caf {
invalid_stream_gatherer::~invalid_stream_gatherer() {
// nop
}
stream_gatherer::path_ptr
invalid_stream_gatherer::add_path(stream_slot, strong_actor_ptr,
strong_actor_ptr, stream_priority, long, bool,
response_promise) {
CAF_LOG_ERROR("invalid_stream_gatherer::add_path called");
return nullptr;
}
bool invalid_stream_gatherer::remove_path(stream_slot, const actor_addr&,
error, bool) {
CAF_LOG_ERROR("invalid_stream_gatherer::remove_path called");
return false;
}
void invalid_stream_gatherer::close(message) {
// nop
}
void invalid_stream_gatherer::abort(error) {
// nop
}
long invalid_stream_gatherer::num_paths() const {
return 0;
}
bool invalid_stream_gatherer::closed() const {
return true;
}
bool invalid_stream_gatherer::continuous() const {
return false;
}
void invalid_stream_gatherer::continuous(bool) {
// nop
}
stream_gatherer::path_type* invalid_stream_gatherer::path_at(size_t) {
return nullptr;
}
stream_gatherer::path_type* invalid_stream_gatherer::find(stream_slot,
const actor_addr&) {
return nullptr;
}
void invalid_stream_gatherer::assign_credit(long) {
// nop
}
long invalid_stream_gatherer::initial_credit(long, path_type*) {
return 0;
}
} // namespace caf
...@@ -50,7 +50,11 @@ public: ...@@ -50,7 +50,11 @@ public:
} }
type_erased_tuple& content() override { type_erased_tuple& content() override {
return *msg_.vals().raw_ptr(); return msg_.content();
}
const type_erased_tuple& content() const override {
return msg_.content();
} }
message move_content_to_message() override { message move_content_to_message() override {
......
...@@ -25,9 +25,9 @@ ...@@ -25,9 +25,9 @@
namespace caf { namespace caf {
outbound_path::outbound_path(local_actor* selfptr, stream_slot id, outbound_path::outbound_path(local_actor* selfptr, stream_slots id,
strong_actor_ptr ptr) strong_actor_ptr ptr)
: slot(id), : slots(id),
self(selfptr), self(selfptr),
hdl(std::move(ptr)), hdl(std::move(ptr)),
next_batch_id(0), next_batch_id(0),
...@@ -43,10 +43,10 @@ outbound_path::~outbound_path() { ...@@ -43,10 +43,10 @@ outbound_path::~outbound_path() {
if (hdl) { if (hdl) {
if (shutdown_reason == none) if (shutdown_reason == none)
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<downstream_msg::close>(slot, self->address())); make<downstream_msg::close>(slots, self->address()));
else else
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<downstream_msg::forced_close>(slot, self->address(), make<downstream_msg::forced_close>(slots, self->address(),
shutdown_reason)); shutdown_reason));
} }
if (shutdown_reason != none) if (shutdown_reason != none)
...@@ -69,10 +69,10 @@ void outbound_path::emit_open(strong_actor_ptr origin, ...@@ -69,10 +69,10 @@ void outbound_path::emit_open(strong_actor_ptr origin,
cd = client_data{origin, handshake_mid}; cd = client_data{origin, handshake_mid};
redeployable = is_redeployable; redeployable = is_redeployable;
hdl->enqueue( hdl->enqueue(
make_mailbox_element(std::move(origin), handshake_mid, std::move(stages), make_mailbox_element(
make_message(stream_handshake_msg{ std::move(origin), handshake_mid, std::move(stages),
slot, std::move(handshake_data), self->ctrl(), hdl, make_message(open_stream_msg{slots.sender, std::move(handshake_data),
prio, is_redeployable})), self->ctrl(), hdl, prio, is_redeployable})),
self->context()); self->context());
} }
...@@ -85,16 +85,16 @@ void outbound_path::emit_batch(long xs_size, message xs) { ...@@ -85,16 +85,16 @@ void outbound_path::emit_batch(long xs_size, message xs) {
if (redeployable) if (redeployable)
unacknowledged_batches.emplace_back(bid, batch); unacknowledged_batches.emplace_back(bid, batch);
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
downstream_msg{slot, self->address(), std::move(batch)}); downstream_msg{slots, self->address(), std::move(batch)});
} }
void outbound_path::emit_irregular_shutdown(local_actor* self, void outbound_path::emit_irregular_shutdown(local_actor* self,
stream_slot slot, stream_slots slots,
const strong_actor_ptr& hdl, const strong_actor_ptr& hdl,
error reason) { error reason) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
unsafe_send_as(self, hdl, unsafe_send_as(self, hdl,
make<downstream_msg::forced_close>(slot, self->address(), make<downstream_msg::forced_close>(slots, self->address(),
std::move(reason))); 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/detail/pull5_gatherer.hpp"
namespace caf {
namespace detail {
pull5_gatherer::pull5_gatherer(local_actor* selfptr, stream_scatterer& out)
: super(selfptr, out) {
// nop
}
void pull5_gatherer::assign_credit(long available) {
CAF_LOG_TRACE(CAF_ARG(available));
for (auto& kvp : assignment_vec_) {
auto x = std::min(available, 5l - kvp.first->assigned_credit);
available -= x;
kvp.second = x;
}
emit_credits();
}
long pull5_gatherer::initial_credit(long, inbound_path*) {
return 5;
}
} // namespace detail
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/random_gatherer.hpp"
namespace caf {
random_gatherer::random_gatherer(local_actor* selfptr, stream_scatterer&)
: super(selfptr) {
// nop
}
random_gatherer::~random_gatherer() {
// nop
}
void random_gatherer::assign_credit(long available) {
CAF_LOG_TRACE(CAF_ARG(available));
if (available <= 0 || assignment_vec_.empty())
return;
auto max_credit = available / static_cast<long>(assignment_vec_.size());
if (max_credit == 0)
return;
for (auto& kvp : assignment_vec_) {
auto x = std::min(available, max_credit - kvp.first->assigned_credit);
available -= x;
kvp.second = x;
}
emit_credits();
}
long random_gatherer::initial_credit(long available, path_type*) {
return available;
}
/*
void random_gatherer::fill_assignment_vec(long downstream_credit) {
CAF_LOG_TRACE(CAF_ARG(downstream_credit));
// Zero-out assignment vector if no credit is available at downstream paths.
if (downstream_credit <= 0) {
for (auto& x : assignment_vec_)
x.second = 0;
return;
}
// Assign credit to upstream paths until no more credit is available. We must
// make sure to write to each element in the vector.
auto available = downstream_credit;
for (auto& p : assignment_vec_) {
auto& x = p.first->assigned_credit; // current value
auto y = std::min(max_credit(), x + available);
auto delta = y - x;
if (delta >= min_credit_assignment()) {
p.second = delta;
available -= delta;
} else {
p.second = 0;
}
}
}
*/
} // namespace caf
...@@ -40,12 +40,12 @@ void stream_aborter::actor_exited(const error& rsn, execution_unit* host) { ...@@ -40,12 +40,12 @@ void stream_aborter::actor_exited(const error& rsn, execution_unit* host) {
if (mode_ == source_aborter) if (mode_ == source_aborter)
observer->enqueue(nullptr, make_message_id(), observer->enqueue(nullptr, make_message_id(),
make_message(caf::make<downstream_msg::forced_close>( make_message(caf::make<downstream_msg::forced_close>(
slot_, observed_, rsn)), slots_, observed_, rsn)),
host); host);
else else
observer->enqueue(nullptr, make_message_id(), observer->enqueue(nullptr, make_message_id(),
make_message(caf::make<upstream_msg::forced_drop>( make_message(caf::make<upstream_msg::forced_drop>(
slot_, observed_, rsn)), slots_, observed_, rsn)),
host); host);
} }
} }
...@@ -54,37 +54,37 @@ bool stream_aborter::matches(const attachable::token& what) { ...@@ -54,37 +54,37 @@ bool stream_aborter::matches(const attachable::token& what) {
if (what.subtype != attachable::token::stream_aborter) if (what.subtype != attachable::token::stream_aborter)
return false; return false;
auto& ot = *reinterpret_cast<const token*>(what.ptr); auto& ot = *reinterpret_cast<const token*>(what.ptr);
return ot.observer == observer_ && ot.slot == slot_; return ot.observer == observer_ && ot.slots == slots_;
} }
stream_aborter::stream_aborter(actor_addr&& observed, actor_addr&& observer, stream_aborter::stream_aborter(actor_addr&& observed, actor_addr&& observer,
stream_slot slot, mode m) stream_slots slots, mode m)
: observed_(std::move(observed)), : observed_(std::move(observed)),
observer_(std::move(observer)), observer_(std::move(observer)),
slot_(slot), slots_(slots),
mode_(m) { mode_(m) {
// nop // nop
} }
void stream_aborter::add(strong_actor_ptr observed, actor_addr observer, void stream_aborter::add(strong_actor_ptr observed, actor_addr observer,
stream_slot slot, mode m) { stream_slots slots, mode m) {
CAF_LOG_TRACE(CAF_ARG(observed) << CAF_ARG(observer) << CAF_ARG(slot)); CAF_LOG_TRACE(CAF_ARG(observed) << CAF_ARG(observer) << CAF_ARG(slots));
auto ptr = make_stream_aborter(observed->address(), std::move(observer), auto ptr = make_stream_aborter(observed->address(), std::move(observer),
slot, m); slots, m);
observed->get()->attach(std::move(ptr)); observed->get()->attach(std::move(ptr));
} }
void stream_aborter::del(strong_actor_ptr observed, const actor_addr& observer, void stream_aborter::del(strong_actor_ptr observed, const actor_addr& observer,
stream_slot slot, mode m) { stream_slots slots, mode m) {
CAF_LOG_TRACE(CAF_ARG(observed) << CAF_ARG(observer) << CAF_ARG(slot)); CAF_LOG_TRACE(CAF_ARG(observed) << CAF_ARG(observer) << CAF_ARG(slots));
token tk{observer, slot, m}; token tk{observer, slots, m};
observed->get()->detach(tk); observed->get()->detach(tk);
} }
attachable_ptr make_stream_aborter(actor_addr observed, actor_addr observer, attachable_ptr make_stream_aborter(actor_addr observed, actor_addr observer,
stream_slot slot, stream_aborter::mode m) { stream_slots slots, stream_aborter::mode m) {
return attachable_ptr{ return attachable_ptr{
new stream_aborter(std::move(observed), std::move(observer), slot, m)}; new stream_aborter(std::move(observed), std::move(observer), slots, m)};
} }
} // namespace caf } // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_gatherer.hpp"
#include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp"
#include "caf/inbound_path.hpp"
#include "caf/actor_control_block.hpp"
namespace caf {
stream_gatherer::~stream_gatherer() {
// nop
}
bool stream_gatherer::remove_path(stream_slot slot, const strong_actor_ptr& x,
error reason, bool silent) {
return remove_path(slot, actor_cast<actor_addr>(x), std::move(reason),
silent);
}
stream_gatherer::path_type* stream_gatherer::find(stream_slot slot,
const strong_actor_ptr& x) {
return find(slot, actor_cast<actor_addr>(x));
}
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_gatherer_impl.hpp"
namespace caf {
stream_gatherer_impl::stream_gatherer_impl(local_actor* selfptr)
: super(selfptr) {
// nop
}
stream_gatherer_impl::~stream_gatherer_impl() {
// nop
}
stream_gatherer::path_ptr
stream_gatherer_impl::add_path(stream_slot slot, strong_actor_ptr hdl,
strong_actor_ptr original_stage,
stream_priority prio, long available_credit,
bool redeployable, response_promise result_cb) {
CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(hdl) << CAF_ARG(original_stage)
<< CAF_ARG(prio) << CAF_ARG(available_credit));
CAF_ASSERT(hdl != nullptr);
if (find(slot, hdl) != nullptr) {
inbound_path::emit_irregular_shutdown(self_, slot, hdl,
sec::cannot_add_upstream);
return nullptr;
}
auto ptr = add_path_impl(slot, std::move(hdl));
CAF_ASSERT(ptr != nullptr);
assignment_vec_.emplace_back(ptr, 0l);
if (result_cb.pending())
listeners_.emplace_back(std::move(result_cb));
ptr->prio = prio;
ptr->emit_ack_open(actor_cast<actor_addr>(original_stage),
initial_credit(available_credit, ptr), redeployable);
return ptr;
}
bool stream_gatherer_impl::remove_path(stream_slot slot,
const actor_addr& x, error reason,
bool silent) {
CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(x)
<< CAF_ARG(reason) << CAF_ARG(silent));
auto pred = [&](const assignment_pair& y) {
return y.first->slot == slot && y.first->hdl == x;
};
auto e = assignment_vec_.end();
auto i = std::find_if(assignment_vec_.begin(), e, pred);
if (i != e) {
assignment_vec_.erase(i);
return super::remove_path(slot, x, std::move(reason), silent);
}
return false;
}
void stream_gatherer_impl::close(message result) {
CAF_LOG_TRACE(CAF_ARG(result) << CAF_ARG2("remaining paths", paths_.size())
<< CAF_ARG2("listener", listeners_.size()));
for (auto& path : paths_)
stream_aborter::del(path->hdl, self_->address(), path->slot, aborter_type);
paths_.clear();
assignment_vec_.clear();
for (auto& listener : listeners_)
listener.deliver(result);
listeners_.clear();
}
void stream_gatherer_impl::abort(error reason) {
for (auto& path : paths_) {
stream_aborter::del(path->hdl, self_->address(), path->slot, aborter_type);
path->shutdown_reason = reason;
}
paths_.clear();
assignment_vec_.clear();
for (auto& listener : listeners_)
listener.deliver(reason);
listeners_.clear();
}
void stream_gatherer_impl::emit_credits() {
for (auto& kvp : assignment_vec_)
if (kvp.second > 0)
kvp.first->emit_ack_batch(kvp.second);
}
} // namespace caf
...@@ -29,7 +29,6 @@ ...@@ -29,7 +29,6 @@
#include "caf/outbound_path.hpp" #include "caf/outbound_path.hpp"
#include "caf/response_promise.hpp" #include "caf/response_promise.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/stream_gatherer.hpp"
#include "caf/stream_scatterer.hpp" #include "caf/stream_scatterer.hpp"
namespace caf { namespace caf {
......
...@@ -51,9 +51,11 @@ stream_scatterer_impl::add_path(stream_slot slot, strong_actor_ptr origin, ...@@ -51,9 +51,11 @@ stream_scatterer_impl::add_path(stream_slot slot, strong_actor_ptr origin,
return ptr; return ptr;
} }
stream_scatterer::path_ptr stream_scatterer_impl::confirm_path( stream_scatterer::path_ptr
stream_slot slot, const actor_addr& from, strong_actor_ptr to, stream_scatterer_impl::confirm_path(stream_slot, const actor_addr&,
long initial_demand, long desired_batch_size, bool redeployable) { strong_actor_ptr, long, long, bool) {
return nullptr; // TODO: implement me
/*
CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(from) << CAF_ARG(to) CAF_LOG_TRACE(CAF_ARG(slot) << CAF_ARG(from) << CAF_ARG(to)
<< CAF_ARG(initial_demand) << CAF_ARG(desired_batch_size) << CAF_ARG(initial_demand) << CAF_ARG(desired_batch_size)
<< CAF_ARG(redeployable)); << CAF_ARG(redeployable));
...@@ -71,6 +73,7 @@ stream_scatterer::path_ptr stream_scatterer_impl::confirm_path( ...@@ -71,6 +73,7 @@ stream_scatterer::path_ptr stream_scatterer_impl::confirm_path(
if (ptr->desired_batch_size != desired_batch_size) if (ptr->desired_batch_size != desired_batch_size)
ptr->desired_batch_size = desired_batch_size; ptr->desired_batch_size = desired_batch_size;
return ptr; return ptr;
*/
} }
bool stream_scatterer_impl::paths_clean() const { bool stream_scatterer_impl::paths_clean() const {
...@@ -82,17 +85,24 @@ bool stream_scatterer_impl::paths_clean() const { ...@@ -82,17 +85,24 @@ bool stream_scatterer_impl::paths_clean() const {
void stream_scatterer_impl::close() { void stream_scatterer_impl::close() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// TODO: implement me
/*
for (auto& path : paths_) for (auto& path : paths_)
stream_aborter::del(path->hdl, self_->address(), path->slot, aborter_type); stream_aborter::del(path->hdl, self_->address(), path->slot, aborter_type);
paths_.clear(); paths_.clear();
*/
} }
void stream_scatterer_impl::abort(error reason) { void stream_scatterer_impl::abort(error reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
// TODO: implement me
/*
for (auto& path : paths_) { for (auto& path : paths_) {
stream_aborter::del(path->hdl, self_->address(), path->slot, aborter_type); stream_aborter::del(path->hdl, self_->address(), path->slot, aborter_type);
path->shutdown_reason = reason; path->shutdown_reason = reason;
} }
paths_.clear(); paths_.clear();
*/
} }
long stream_scatterer_impl::total_credit() const { long stream_scatterer_impl::total_credit() const {
......
...@@ -132,26 +132,26 @@ CAF_TEST(high_priority) { ...@@ -132,26 +132,26 @@ CAF_TEST(high_priority) {
CAF_TEST(upstream_msg_static) { CAF_TEST(upstream_msg_static) {
auto m1 = make_mailbox_element(nullptr, make_message_id(), no_stages, auto m1 = make_mailbox_element(nullptr, make_message_id(), no_stages,
make<upstream_msg::drop>(0, nullptr)); make<upstream_msg::drop>({0, 0}, nullptr));
CAF_CHECK(m1->mid.category() == message_id::upstream_message_category); CAF_CHECK(m1->mid.category() == message_id::upstream_message_category);
} }
CAF_TEST(upstream_msg_dynamic) { CAF_TEST(upstream_msg_dynamic) {
auto m1 = auto m1 = make_mailbox_element(
make_mailbox_element(nullptr, make_message_id(), no_stages, nullptr, make_message_id(), no_stages,
make_message(make<upstream_msg::drop>(0, nullptr))); make_message(make<upstream_msg::drop>({0, 0}, nullptr)));
CAF_CHECK(m1->mid.category() == message_id::upstream_message_category); CAF_CHECK(m1->mid.category() == message_id::upstream_message_category);
} }
CAF_TEST(downstream_msg_static) { CAF_TEST(downstream_msg_static) {
auto m1 = make_mailbox_element(nullptr, make_message_id(), no_stages, auto m1 = make_mailbox_element(nullptr, make_message_id(), no_stages,
make<downstream_msg::close>(0, nullptr)); make<downstream_msg::close>({0, 0}, nullptr));
CAF_CHECK(m1->mid.category() == message_id::downstream_message_category); CAF_CHECK(m1->mid.category() == message_id::downstream_message_category);
} }
CAF_TEST(downstream_msg_dynamic) { CAF_TEST(downstream_msg_dynamic) {
auto m1 = auto m1 = make_mailbox_element(
make_mailbox_element(nullptr, make_message_id(), no_stages, nullptr, make_message_id(), no_stages,
make_message(make<downstream_msg::close>(0, nullptr))); make_message(make<downstream_msg::close>({0, 0}, nullptr)));
CAF_CHECK(m1->mid.category() == message_id::downstream_message_category); CAF_CHECK(m1->mid.category() == message_id::downstream_message_category);
} }
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