Commit de53c367 authored by Dominik Charousset's avatar Dominik Charousset

Re-implement rudimentary source->sink streaming

parent b3bd97ad
......@@ -416,6 +416,8 @@ public:
bool cleanup(error&& fail_state, execution_unit* host) override;
sec build_pipeline(stream_manager_ptr);
/// @endcond
private:
......
......@@ -88,9 +88,9 @@ struct dmi<optional<Y> (Xs...)> : dmi<Y (Xs...)> {};
template <class Y, class... Xs>
struct dmi<expected<Y> (Xs...)> : dmi<Y (Xs...)> {};
// case #5: function returning an annotated_stream<>
// case #5: function returning an output_stream<>
template <class Y, class... Ys, class... Xs>
struct dmi<annotated_stream<Y, Ys...> (Xs...)> : dmi<Y (Xs...)> {
struct dmi<output_stream<Y, Ys...> (Xs...)> : dmi<Y (Xs...)> {
using type =
typed_mpi<type_list<typename param_decay<Xs>::type...>,
output_tuple<stream<Y>, strip_and_convert_t<Ys>...>>;
......
......@@ -56,6 +56,11 @@ public:
delegate(x);
}
void operator()(stream_slot, stream_manager_ptr& ptr) override {
// TODO: error handling
self_->build_pipeline(std::move(ptr));
}
private:
void deliver(response_promise& rp, error& x) {
CAF_LOG_DEBUG("report error back to requesting actor");
......
......@@ -30,6 +30,7 @@
#include "caf/expected.hpp"
#include "caf/optional.hpp"
#include "caf/make_message.hpp"
#include "caf/stream_result.hpp"
#include "caf/detail/int_list.hpp"
#include "caf/detail/apply_args.hpp"
......@@ -37,6 +38,8 @@
namespace caf {
namespace detail {
/// Inspects the result of message handlers and triggers type-depended actions
/// such as generating result messages.
class invoke_result_visitor {
public:
virtual ~invoke_result_visitor();
......@@ -45,26 +48,36 @@ public:
// nop
}
// severs as catch-all case for all values producing
// no result such as response_promise
// -- virtual handlers -------------------------------------------------------
/// Called whenever no result messages gets produced, e.g., when returning a
/// `response_promise`.
virtual void operator()() = 0;
// called if the message handler explicitly returned an error
/// Called if the message handler returned an error.
virtual void operator()(error&) = 0;
// called if the message handler returned any "ordinary" value
/// Called if the message handler returned any "ordinary" value.
virtual void operator()(message&) = 0;
// called if the message handler returns explictly none or unit
/// Called if the message handler returned a "stream<T>",
/// `output_stream<T, ...>`, or `stream_result<T>`.
virtual void operator()(stream_slot, stream_manager_ptr&) = 0;
/// Called if the message handler returns "nothing", for example a
/// default-constructed `optional<T>`.
virtual void operator()(const none_t&) = 0;
// map unit to an empty message
// -- on-the-fly type conversions --------------------------------------------
/// Called if the message handler returns `void` or `unit_t`.
inline void operator()(const unit_t&) {
message empty_msg;
(*this)(empty_msg);
}
// unwrap optionals
/// Unwraps an `optional<T>` by recursively calling the visitor with either
/// `none_t` or `T`.
template <class T>
void operator()(optional<T>& x) {
if (x)
......@@ -73,7 +86,8 @@ public:
(*this)(none);
}
// unwrap expecteds
/// Unwraps an `expected<T>` by recursively calling the visitor with either
/// `error` or `T`.
template <class T>
void operator()(expected<T>& x) {
if (x)
......@@ -82,71 +96,88 @@ public:
(*this)(x.error());
}
// convert values to messages
/// Wraps arbitrary values into a `message` and calls the visitor recursively.
template <class... Ts>
void operator()(Ts&... xs) {
auto tmp = make_message(std::move(xs)...);
(*this)(tmp);
}
// unwrap tuples
/// Wraps the tuple into a `message` and calls the visitor recursively with
/// its contents.
template <class... Ts>
void operator()(std::tuple<Ts...>& xs) {
apply_args(*this, get_indices(xs), xs);
}
// disambiguations
/// Disambiguates the variadic `operator<Ts...>()`.
inline void operator()(none_t& x) {
(*this)(const_cast<const none_t&>(x));
}
/// Disambiguates the variadic `operator<Ts...>()`.
inline void operator()(unit_t& x) {
(*this)(const_cast<const unit_t&>(x));
}
// special purpose handler that don't procude results
// -- special-purpose handlers that don't procude results --------------------
/// Calls `(*this)()`.
inline void operator()(response_promise&) {
(*this)();
}
/// Calls `(*this)()`.
template <class... Ts>
void operator()(typed_response_promise<Ts...>&) {
(*this)();
}
/// Calls `(*this)()`.
template <class... Ts>
void operator()(delegated<Ts...>&) {
(*this)();
}
/// Calls `(*this)(x.ptr)`.
template <class T>
void operator()(stream<T>&) {
(*this)();
void operator()(stream<T>& x) {
(*this)(x.slot(), x.ptr());
}
/// Calls `(*this)(x.ptr)`.
template <class T, class... Us>
void operator()(output_stream<T, Us...>& x) {
(*this)(x.slot(), x.ptr());
}
// visit API - returns true if T was visited, false if T was skipped
/// Calls `(*this)(x.ptr)`.
template <class T>
void operator()(stream_result<T>& x) {
(*this)(x.slot(), x.ptr());
}
// -- visit API: return true if T was visited, false if T was skipped --------
/// Delegates `x` to the appropriate handler and returns `true`.
template <class T>
bool visit(T& x) {
(*this)(x);
return true;
}
/// Returns `false`.
inline bool visit(skip_t&) {
return false;
}
/// Returns `false`.
inline bool visit(const skip_t&) {
return false;
}
template <class T>
bool visit(stream<T>&) {
return true;
}
/// Returns `false` if `x != none`, otherwise calls the void handler and
/// returns `true`..
inline bool visit(optional<skip_t>& x) {
if (x)
return false;
......@@ -154,6 +185,7 @@ public:
return true;
}
/// Dispatches on the runtime-type of `x`.
template <class... Ts>
bool visit(result<Ts...>& x) {
switch (x.flag) {
......
......@@ -64,6 +64,7 @@ public:
if (x.xs.match_elements<vec_type>()) {
auto& xs = x.xs.get_mutable_as<vec_type>(0);
driver_.process(std::move(xs));
return;
}
CAF_LOG_ERROR("received unexpected batch type (dropped)");
}
......
......@@ -43,7 +43,7 @@ public:
using stream_type = stream<output_type>;
using annotated_stream_type = typename super::annotated_stream_type;
using output_stream_type = typename super::output_stream_type;
using tuple_type = std::tuple<Ts...>;
......
......@@ -83,6 +83,10 @@ public:
return hint != out_.capacity();
}
message make_handshake() const override {
return make_message_from_tuple(driver_.make_handshake());
}
private:
bool at_end_;
Driver driver_;
......
......@@ -47,7 +47,7 @@ public:
using stream_type = stream<output_type>;
using annotated_stream_type = typename super::annotated_stream_type;
using output_stream_type = typename super::output_stream_type;
using tuple_type = std::tuple<Ts...>;
......
......@@ -29,16 +29,19 @@
namespace caf {
namespace detail {
/// Drains a mailbox and sends an error message to each unhandled request.
struct sync_request_bouncer {
error rsn;
explicit sync_request_bouncer(error r);
void operator()(const strong_actor_ptr& sender, const message_id& mid) const;
void operator()(const mailbox_element& e) const;
template <class Key, class Queue>
/// Unwrap WDRR queues. Nesting WDRR queues results in a Key/Queue prefix for
/// each layer of nesting.
template <class Key, class Queue, class... Ts>
intrusive::task_result operator()(const Key&, const Queue&,
const mailbox_element& x) const {
(*this)(x);
const Ts&... xs) const {
(*this)(xs...);
return intrusive::task_result::resume;
}
};
......
......@@ -62,6 +62,11 @@ public:
/// Configures the time interval per tick.
void interval(duration_type x);
/// Returns the time interval per tick.
inline duration_type interval() const {
return interval_;
}
/// Advances time and calls `consumer` for each emitted tick.
template <class F>
void update(time_point now, F& consumer) {
......
......@@ -57,7 +57,7 @@ template <class...> class typed_event_based_actor;
// -- variadic templates with 1 fixed argument ---------------------------------
template <class, class...> class annotated_stream;
template <class, class...> class output_stream;
// -- classes ------------------------------------------------------------------
......
......@@ -28,6 +28,7 @@
#include "caf/stream_manager.hpp"
#include "caf/stream_priority.hpp"
#include "caf/stream_slot.hpp"
#include "caf/timestamp.hpp"
#include "caf/upstream_msg.hpp"
#include "caf/meta/type_name.hpp"
......@@ -81,7 +82,7 @@ public:
struct stats_t {
struct measurement {
long batch_size;
std::chrono::nanoseconds calculation_time;
timespan calculation_time;
};
struct calculation_result {
......@@ -99,8 +100,7 @@ public:
/// Returns the maximum number of items this actor could handle for given
/// cycle length with a minimum of 1.
calculation_result calculate(std::chrono::nanoseconds cycle,
std::chrono::nanoseconds desired_complexity);
calculation_result calculate(timespan cycle, timespan desired_complexity);
/// Stores a new measurement in the ring buffer.
void store(measurement x);
......@@ -129,9 +129,8 @@ public:
/// @param self Points to the parent actor.
/// @param queued_items Counts the accumulated size of all batches that are
/// currently waiting in the mailbox.
void emit_ack_batch(local_actor* self, long queued_items,
std::chrono::nanoseconds cycle,
std::chrono::nanoseconds desired_batch_complexity);
void emit_ack_batch(local_actor* self, long queued_items, timespan cycle,
timespan desired_batch_complexity);
/// Sends a `stream_msg::close` on this path.
void emit_regular_shutdown(local_actor* self);
......
......@@ -97,6 +97,8 @@ public:
bool remove_backlink(abstract_actor* x) override;
error fail_state() const;
/// @endcond
protected:
......
......@@ -22,6 +22,7 @@
#include <string>
#include "caf/illegal_message_element.hpp"
#include "caf/stream.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_pair.hpp"
......@@ -36,9 +37,6 @@ std::string replies_to_type_name(size_t input_size,
const std::string* output_opt1);
/// @endcond
template <class...>
struct output_stream {};
template <class...>
struct output_tuple {};
......@@ -70,8 +68,9 @@ struct replies_to {
using with = typed_mpi<detail::type_list<Is...>, output_tuple<Os...>>;
/// @private
template <class... Os>
using with_stream = typed_mpi<detail::type_list<Is...>, output_stream<Os...>>;
template <class O, class... Os>
using with_stream = typed_mpi<detail::type_list<Is...>,
output_stream<O, Os...>>;
};
template <class... Is>
......
......@@ -56,6 +56,7 @@
#include "caf/policy/urgent_messages.hpp"
#include "caf/detail/behavior_stack.hpp"
#include "caf/detail/tick_emitter.hpp"
#include "caf/detail/stream_sink_driver_impl.hpp"
#include "caf/detail/stream_sink_impl.hpp"
#include "caf/detail/stream_source_driver_impl.hpp"
......@@ -104,10 +105,6 @@ public:
/// Categorizes incoming messages.
enum class message_category {
/// Denotes an expired and thus obsolete timeout.
expired_timeout,
/// Triggers the currently active timeout.
timeout,
/// Triggers the current behavior.
ordinary,
/// Triggers handlers for system messages such as `exit_msg` or `down_msg`.
......@@ -141,7 +138,8 @@ public:
using upstream_queue = intrusive::drr_queue<policy::upstream_messages>;
/// Stores downstream messages.
using downstream_queue = intrusive::drr_queue<policy::downstream_messages>;
using downstream_queue =
intrusive::wdrr_dynamic_multiplexed_queue<policy::downstream_messages>;
/// Configures the FIFO inbox with four nested queues:
///
......@@ -166,6 +164,10 @@ public:
static constexpr size_t default_queue_index = 0;
static constexpr size_t upstream_queue_index = 1;
static constexpr size_t downstream_queue_index = 2;
static constexpr size_t urgent_queue_index = 3;
};
......@@ -214,7 +216,9 @@ public:
mailbox_element&);
/// Consumes downstream messages.
intrusive::task_result operator()(size_t, downstream_queue&,
intrusive::task_result
operator()(size_t, downstream_queue&, stream_slot slot,
policy::downstream_messages::nested_queue_type&,
mailbox_element&);
// Dispatches asynchronous messages with high and normal priority to the
......@@ -403,20 +407,17 @@ public:
template <class Driver,
class Scatterer = broadcast_scatterer<typename Driver::output_type>,
class... Ts>
typename Driver::annotated_stream_type make_source(Ts&&... xs) {
auto slot = next_slot();
typename Driver::output_stream_type make_source(Ts&&... xs) {
auto ptr = detail::make_stream_source<Driver, Scatterer>(
this, std::forward<Ts>(xs)...);
ptr->generate_messages();
pending_stream_managers_.emplace(slot, ptr);
return {slot, std::move(ptr)};
return {0, std::move(ptr)};
}
template <class... Ts, class Init, class Pull, class Done,
class Scatterer =
broadcast_scatterer<typename stream_source_trait_t<Pull>::output>,
class Trait = stream_source_trait_t<Pull>>
annotated_stream<typename Trait::output, detail::decay_t<Ts>...>
output_stream<typename Trait::output, detail::decay_t<Ts>...>
make_source(std::tuple<Ts...> xs, Init init, Pull pull, Done done,
policy::arg<Scatterer> = {}) {
using tuple_type = std::tuple<detail::decay_t<Ts>...>;
......@@ -432,7 +433,10 @@ public:
auto slot = next_slot();
stream_slots id{in.slot(), slot};
auto ptr = detail::make_stream_sink<Driver>(this, std::forward<Ts>(xs)...);
stream_managers_.emplace(id, ptr);
if (!add_stream_manager(id, ptr)) {
CAF_LOG_WARNING("unable to add a stream manager for a sink");
return {0, nullptr};
}
return {slot, std::move(ptr)};
}
......@@ -460,7 +464,7 @@ public:
`stream_manager`. template <class Handle, class... Ts, class Init, class
Getter, class ClosedPredicate, class ResHandler, class Scatterer =
broadcast_scatterer< typename stream_source_trait_t<Getter>::output>>
annotated_stream<typename stream_source_trait_t<Getter>::output, Ts...>
output_stream<typename stream_source_trait_t<Getter>::output, Ts...>
make_source(const Handle& dest, std::tuple<Ts...> xs, Init init,
Getter getter, ClosedPredicate pred, ResHandler res_handler,
policy::arg<Scatterer> scatterer_type = {}) {
......@@ -531,7 +535,7 @@ public:
/// @returns A stream object with a pointer to the generated
`stream_manager`. template <class Init, class... Ts, class Getter, class
ClosedPredicate, class Scatterer = broadcast_scatterer< typename
stream_source_trait_t<Getter>::output>> annotated_stream<typename
stream_source_trait_t<Getter>::output>> output_stream<typename
stream_source_trait_t<Getter>::output, Ts...> make_source(std::tuple<Ts...>
xs, Init init, Getter getter, ClosedPredicate pred, policy::arg<Scatterer>
scatterer_type = {}) { CAF_IGNORE_UNUSED(scatterer_type); using type =
......@@ -598,7 +602,7 @@ public:
`stream_manager`. template <class In, class... Ts, class Init, class Fun,
class Cleanup, class Gatherer = random_gatherer, class Scatterer =
broadcast_scatterer<typename stream_stage_trait_t<Fun>::output>>
annotated_stream<typename stream_stage_trait_t<Fun>::output, Ts...>
output_stream<typename stream_stage_trait_t<Fun>::output, Ts...>
make_stage(const stream<In>& in, std::tuple<Ts...> xs, Init init, Fun fun,
Cleanup cleanup, policy::arg<Gatherer, Scatterer> policies = {})
{ CAF_IGNORE_UNUSED(policies); CAF_ASSERT(current_mailbox_element() !=
......@@ -722,16 +726,29 @@ public:
*/
/// @cond PRIVATE
/// Builds the pipeline after receiving an `open_stream_msg`. Sends a stream
/// handshake to the next actor if `mgr->out().terminal() == false`,
/// otherwise
/// @private
sec build_pipeline(stream_manager_ptr mgr);
// -- timeout management -----------------------------------------------------
/// Requests a new timeout and returns its ID.
uint32_t request_timeout(const duration& d);
uint64_t set_receive_timeout(actor_clock::time_point x);
/// Requests a new timeout for the current behavior and returns its ID.
uint64_t set_receive_timeout();
/// Resets the timeout if `timeout_id` is the active timeout.
void reset_timeout(uint32_t timeout_id);
void reset_receive_timeout(uint64_t timeout_id);
/// Returns whether `timeout_id` is currently active.
bool is_active_timeout(uint32_t tid) const;
bool is_active_receive_timeout(uint64_t tid) const;
/// Requests a new timeout and returns its ID.
uint64_t set_stream_timeout(actor_clock::time_point x);
// -- message processing -----------------------------------------------------
......@@ -942,11 +959,15 @@ public:
void erase_inbound_paths_later(const stream_manager* mgr,
error reason) override;
// -- handling of upstream message -------------------------------------------
void handle(stream_slots slots, actor_addr& sender,
upstream_msg::ack_open& x);
protected:
/// @cond PRIVATE
/// Returns a currently unused slot.
stream_slot next_slot();
// -- utility functions for invoking default handler -------------------------
/// Utility function that swaps `f` into a temporary before calling it
/// and restoring `f` only if it has not been replaced by the user.
......@@ -978,7 +999,29 @@ protected:
swap(g, f);
}
//bool handle_stream_msg(mailbox_element& x, behavior* active_behavior);
// -- timeout management -----------------------------------------------------
/// Requests a new timeout and returns its ID.
uint64_t set_timeout(atom_value type, actor_clock::time_point x);
// -- stream processing ------------------------------------------------------
/// Returns a currently unused slot.
stream_slot next_slot();
/// Adds a new stream manager to the actor and starts cycle management if
/// needed.
bool add_stream_manager(stream_slots id, stream_manager_ptr ptr);
/// Removes a new stream manager.
bool remove_stream_manager(stream_slots id);
/// @pre `x.content().match_elements<open_stream_msg>()`
invoke_message_result handle_open_stream_msg(mailbox_element& x);
/// Advances credit and batch timeouts and returns the timestamp when to call
/// this function again.
actor_clock::time_point advance_streams(actor_clock::time_point now);
// -- member variables -------------------------------------------------------
......@@ -989,7 +1032,7 @@ protected:
detail::behavior_stack bhvr_stack_;
/// Identifies the timeout messages we are currently waiting for.
uint32_t timeout_id_;
uint64_t timeout_id_;
/// Stores callbacks for awaited responses.
std::forward_list<pending_response> awaited_responses_;
......@@ -1016,6 +1059,15 @@ protected:
/// yet received an ACK.
std::map<stream_slot, stream_manager_ptr> pending_stream_managers_;
/// Controls batch and credit timeouts.
detail::tick_emitter stream_ticks_;
/// Number of ticks per batch delay.
long max_batch_delay_ticks_;
/// Number of ticks of each credit round.
long credit_round_ticks_;
/// Pointer to a private thread object associated with a detached actor.
detail::private_thread* private_thread_;
......
......@@ -77,7 +77,12 @@ public:
}
/// Returns the handler assigned to this stream on this actor.
inline const stream_manager_ptr& ptr() const {
inline stream_manager_ptr& ptr() noexcept {
return ptr_;
}
/// Returns the handler assigned to this stream on this actor.
inline const stream_manager_ptr& ptr() const noexcept {
return ptr_;
}
......@@ -101,7 +106,7 @@ constexpr invalid_stream_t invalid_stream = invalid_stream_t{};
/// Identifies an unbound sequence of messages annotated with the additional
/// handshake arguments emitted to the next stage.
template <class T, class... Ts>
class annotated_stream final : public stream<T> {
class output_stream final : public stream<T> {
public:
/// Dennotes the supertype.
using super = stream<T>;
......
......@@ -36,7 +36,8 @@ namespace caf {
/// @relates stream_msg
class stream_manager : public ref_counted {
public:
stream_manager(local_actor* selfptr);
stream_manager(local_actor* selfptr,
stream_priority prio = stream_priority::normal);
~stream_manager() override;
......@@ -96,12 +97,11 @@ public:
virtual void send_handshake(strong_actor_ptr dest, stream_slot slot,
strong_actor_ptr stream_origin,
mailbox_element::forwarding_stack fwd_stack,
message_id handshake_mid, stream_priority prio);
message_id handshake_mid);
/// Sends a handshake to `dest`.
/// @pre `dest != nullptr`
void send_handshake(strong_actor_ptr dest, stream_slot slot,
stream_priority prio);
void send_handshake(strong_actor_ptr dest, stream_slot slot);
// -- implementation hooks for sources ---------------------------------------
......@@ -119,6 +119,9 @@ public:
/// safely.
virtual bool done() const = 0;
/// Advances time.
virtual void cycle_timeout(size_t cycle_nr);
// -- input path management --------------------------------------------------
/// Informs the manager that a new input path opens.
......@@ -127,6 +130,12 @@ public:
/// Informs the manager that an input path closes.
virtual void deregister_input_path(inbound_path* x) noexcept;
// -- mutators ---------------------------------------------------------------
/// Adds a response promise to a sink for delivering the final result.
/// @pre `out().terminal() == true`
void add_promise(response_promise x);
// -- inline functions -------------------------------------------------------
inline local_actor* self() {
......@@ -170,6 +179,12 @@ protected:
/// Keeps track of pending handshakes.
long pending_handshakes_;
/// Configures the importance of outgoing traffic.
stream_priority priority_;
/// Stores response promises for dellivering the final result.
std::vector<response_promise> promises_;
};
/// A reference counting pointer to a `stream_manager`.
......
......@@ -62,12 +62,17 @@ public:
}
/// Returns the unique identifier for this stream_result.
inline stream_slot slot() const {
inline stream_slot slot() const noexcept {
return slot_;
}
/// Returns the handler assigned to this stream_result on this actor.
inline const stream_manager_ptr& ptr() const {
/// Returns the handler assigned to this stream on this actor.
inline stream_manager_ptr& ptr() noexcept {
return ptr_;
}
/// Returns the handler assigned to this stream on this actor.
inline const stream_manager_ptr& ptr() const noexcept {
return ptr_;
}
......
......@@ -36,7 +36,7 @@ public:
using stream_type = stream<output_type>;
using annotated_stream_type = annotated_stream<output_type, HandshakeData...>;
using output_stream_type = output_stream<output_type, HandshakeData...>;
using handshake_tuple_type = std::tuple<stream_type, HandshakeData...>;
......
......@@ -39,7 +39,7 @@ public:
using stream_type = stream<output_type>;
using annotated_stream_type = annotated_stream<output_type, HandshakeData...>;
using output_stream_type = output_stream<output_type, HandshakeData...>;
using handshake_tuple_type = std::tuple<stream_type, HandshakeData...>;
......
......@@ -80,14 +80,16 @@ typename Inspector::result_type inspect(Inspector& f, group_down_msg& x) {
/// Signalizes a timeout event.
/// @note This message is handled implicitly by the runtime system.
struct timeout_msg {
/// Type of the timeout (either `receive_atom` or `cycle_atom`).
atom_value type;
/// Actor-specific timeout ID.
uint32_t timeout_id;
uint64_t timeout_id;
};
/// @relates timeout_msg
template <class Inspector>
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.type, x.timeout_id);
}
/// Demands the receiver to open a new stream from the sender to the receiver.
......@@ -112,7 +114,7 @@ struct open_stream_msg {
/// @relates stream_handshake_msg
template <class Inspector>
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("open_stream_msg"), x.slot, x.msg, x.prev_stage,
x.original_stage, x.priority);
}
......
......@@ -73,6 +73,10 @@ public:
void operator()(const none_t&) override {
(*this)();
}
void operator()(stream_slot, stream_manager_ptr&) override {
(*this)();
}
};
} // namespace <anonymous>
......
......@@ -259,14 +259,19 @@ bool blocking_actor::await_data(timeout_type timeout) {
}
mailbox_element_ptr blocking_actor::dequeue() {
mailbox().fetch_more();
auto& qs = mailbox_.queue().queues();
auto& q1 = get<mailbox_policy::default_queue_index>(qs);
auto ptr = q1.take_front();
if (ptr == nullptr) {
auto& q2 = get<mailbox_policy::urgent_queue_index>(qs);
ptr = q2.take_front();
auto& q1 = get<mailbox_policy::urgent_queue_index>(qs);
if (!q1.empty()) {
q1.inc_deficit(1);
return q1.take_front();
}
return ptr;
auto& q2 = get<mailbox_policy::default_queue_index>(qs);
if (!q2.empty()) {
q2.inc_deficit(1);
return q2.take_front();
}
return nullptr;
}
void blocking_actor::varargs_tup_receive(receive_cond& rcc, message_id mid,
......@@ -285,6 +290,11 @@ void blocking_actor::varargs_tup_receive(receive_cond& rcc, message_id mid,
}
}
sec blocking_actor::build_pipeline(stream_manager_ptr) {
CAF_LOG_ERROR("blocking_actor::connect_pipeline called");
return sec::bad_function_call;
}
size_t blocking_actor::attach_functor(const actor& x) {
return attach_functor(actor_cast<strong_actor_ptr>(x));
}
......
......@@ -26,12 +26,11 @@
namespace caf {
inbound_path::stats_t::stats_t() : ring_iter(0) {
measurement x{0, std::chrono::nanoseconds(0)};
measurement x{0, timespan{0}};
measurements.resize(stats_sampling_size, x);
}
auto inbound_path::stats_t::calculate(std::chrono::nanoseconds c,
std::chrono::nanoseconds d)
auto inbound_path::stats_t::calculate(timespan c, timespan d)
-> calculation_result {
// Max throughput = C * (N / t), where C = cycle length, N = measured items,
// and t = measured time. Desired batch size is the same formula with D
......@@ -94,8 +93,7 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
}
void inbound_path::emit_ack_batch(local_actor* self, long queued_items,
std::chrono::nanoseconds cycle,
std::chrono::nanoseconds complexity){
timespan cycle, timespan complexity) {
auto x = stats.calculate(cycle, complexity);
auto credit = std::max(x.max_throughput - (assigned_credit - queued_items),
0l);
......
......@@ -187,6 +187,11 @@ bool monitorable_actor::remove_backlink(abstract_actor* x) {
return detach_impl(tk, true) > 0;
}
error monitorable_actor::fail_state() const {
std::unique_lock<std::mutex> guard{mtx_};
return fail_state_;
}
size_t monitorable_actor::detach_impl(const attachable::token& what,
bool stop_on_hit, bool dry_run) {
CAF_LOG_TRACE(CAF_ARG(stop_on_hit) << CAF_ARG(dry_run));
......
......@@ -67,7 +67,7 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
auto& tm = content.get_as<timeout_msg>(0);
auto tid = tm.timeout_id;
CAF_ASSERT(!x.mid.valid());
if (is_active_timeout(tid)) {
if (is_active_receive_timeout(tid)) {
CAF_LOG_DEBUG("handle timeout message");
if (bhvr_stack_.empty())
return im_dropped;
......
......@@ -19,6 +19,7 @@
#include "caf/scheduled_actor.hpp"
#include "caf/actor_ostream.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/config.hpp"
#include "caf/inbound_path.hpp"
#include "caf/to_string.hpp"
......@@ -109,7 +110,14 @@ scheduled_actor::scheduled_actor(actor_config& cfg)
, exception_handler_(default_exception_handler)
# endif // CAF_NO_EXCEPTIONS
{
// nop
auto& sys_cfg = home_system().config();
auto interval = sys_cfg.streaming_tick_duration_us;
CAF_ASSERT(interval != 0);
stream_ticks_.interval(std::chrono::microseconds(interval));
max_batch_delay_ticks_ = sys_cfg.streaming_max_batch_delay_us / interval;
credit_round_ticks_ = sys_cfg.streaming_credit_round_interval_us / interval;
CAF_LOG_DEBUG(CAF_ARG(max_batch_delay_ticks_)
<< CAF_ARG(credit_round_ticks_));
}
scheduled_actor::~scheduled_actor() {
......@@ -229,22 +237,111 @@ void scheduled_actor::intrusive_ptr_release_impl() {
intrusive_ptr_release(ctrl());
}
intrusive::task_result
scheduled_actor::mailbox_visitor:: operator()(size_t, upstream_queue&,
mailbox_element&) {
// TODO: implement me
return intrusive::task_result::stop;
intrusive::task_result scheduled_actor::mailbox_visitor::
operator()(size_t, upstream_queue&, mailbox_element& x) {
CAF_ASSERT(x.content().type_token() == make_type_token<upstream_msg>());
struct visitor {
scheduled_actor* selfptr;
upstream_msg& um;
void operator()(upstream_msg::ack_open& y) {
selfptr->handle(um.slots, um.sender, y);
}
void operator()(upstream_msg::ack_batch& x) {
// Dispatch to the responsible manager.
auto slots = um.slots;
auto& managers = selfptr->stream_managers_;
auto i = managers.find(slots);
if (i == managers.end()) {
CAF_LOG_INFO("no manager available for ack_batch:" << CAF_ARG(slots));
return;
}
i->second->handle(um.slots, x);
if (i->second->done()) {
CAF_LOG_INFO("done sending:" << CAF_ARG(slots));
i->second->close();
managers.erase(i);
}
}
void operator()(upstream_msg::drop&) {
CAF_LOG_ERROR("implement me");
}
void operator()(upstream_msg::forced_drop&) {
CAF_LOG_ERROR("implement me");
}
};
auto& um = x.content().get_mutable_as<upstream_msg>(0);
visitor f{self, um};
visit(f, um.content);
return intrusive::task_result::resume;
}
intrusive::task_result
scheduled_actor::mailbox_visitor:: operator()(size_t, downstream_queue&,
mailbox_element&) {
// TODO: implement me
intrusive::task_result scheduled_actor::mailbox_visitor::
operator()(size_t, downstream_queue& qs, stream_slot,
policy::downstream_messages::nested_queue_type& q,
mailbox_element& x) {
CAF_ASSERT(x.content().type_token() == make_type_token<downstream_msg>());
struct visitor {
scheduled_actor* thisptr;
downstream_queue& qs_ref;
policy::downstream_messages::nested_queue_type& q_ref;
inbound_path* inptr;
downstream_msg& dm;
intrusive::task_result operator()(downstream_msg::batch& y) {
inptr->handle(y);
if (inptr->mgr->done()) {
CAF_LOG_DEBUG("path is done receiving and closes its manager");
inptr->mgr->close();
}
return intrusive::task_result::resume;
}
intrusive::task_result operator()(downstream_msg::close& y) {
CAF_LOG_TRACE(CAF_ARG(y));
auto& managers = thisptr->stream_managers_;
auto slots = dm.slots;
auto i = managers.find(slots);
if (i == managers.end()) {
CAF_LOG_DEBUG("no manager found for dispatching close message:"
<< CAF_ARG(slots));
return intrusive::task_result::resume;
}
i->second->handle(inptr, y);
q_ref.policy().handler.reset();
qs_ref.erase_later(slots.receiver);
if (!i->second->done()) {
CAF_LOG_DEBUG("inbound path closed:" << CAF_ARG(slots));
managers.erase(i);
} else {
CAF_LOG_DEBUG("manager closed its final inbound path:"
<< CAF_ARG(slots));
// Close the manager and remove it on all registered slots.
auto mgr = i->second;
mgr->close();
auto j = managers.begin();
while (j != managers.end()) {
if (j->second == mgr)
j = managers.erase(j);
else
++j;
}
}
return intrusive::task_result::resume;
}
intrusive::task_result operator()(downstream_msg::forced_close&) {
CAF_LOG_ERROR("implement me");
return intrusive::task_result::stop;
}
};
auto inptr = q.policy().handler.get();
if (inptr == nullptr)
return intrusive::task_result::stop;
auto& dm = x.content().get_mutable_as<downstream_msg>(0);
visitor f{self, qs, q, inptr, dm};
return visit(f, dm.content);
}
intrusive::task_result
scheduled_actor::mailbox_visitor::operator()(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG(handled_msgs));
switch (self->reactivate(x)) {
case activation_result::terminated:
result = resume_result::done;
......@@ -263,26 +360,40 @@ scheduled_actor::mailbox_visitor::operator()(mailbox_element& x) {
resumable::resume_result
scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) {
CAF_PUSH_AID(id());
CAF_LOG_TRACE(CAF_ARG(max_throughput));
if (!activate(ctx))
return resume_result::done;
size_t handled_msgs = 0;
auto reset_timeout_if_needed = [&] {
if (handled_msgs > 0 && !bhvr_stack_.empty())
request_timeout(bhvr_stack_.back().timeout());
actor_clock::time_point tout{actor_clock::duration_type{0}};
auto reset_timeouts_if_needed = [&] {
// Set a new receive timeout if we called our behavior at least once.
if (handled_msgs > 0)
set_receive_timeout();
// Set a new stream timeout if we called `advance_streams` at least once.
if (!stream_managers_.empty() && tout.time_since_epoch().count() != 0) {
set_stream_timeout(tout);
}
};
auto result = resume_result::awaiting_message;
mailbox_visitor f{this, result, handled_msgs, max_throughput};
mailbox_element_ptr ptr;
// Timeout for calling `advance_streams`.
while (handled_msgs < max_throughput) {
// TODO: maybe replace '3' with configurable / adaptive value?
// Dispatch on the different message categories in our mailbox.
if (!mailbox_.new_round(3, f).consumed_items) {
reset_timeout_if_needed();
reset_timeouts_if_needed();
if (mailbox().try_block())
return resumable::awaiting_message;
}
// Immediately stop if the visitor reports an error.
if (result != awaiting_message)
return result;
auto now = clock().now();
if (now >= tout)
tout = advance_streams(now);
}
reset_timeout_if_needed();
reset_timeouts_if_needed();
if (mailbox().try_block())
return resumable::awaiting_message;
// time's up
......@@ -312,37 +423,74 @@ void scheduled_actor::trigger_downstreams() {
}
*/
sec scheduled_actor::build_pipeline(stream_manager_ptr mgr) {
auto next = take_current_next_stage();
if (next == nullptr) {
auto rp = make_response_promise();
if (mgr->out().terminal()) {
// Response will be delivered by the sink after stream completion.
mgr->add_promise(std::move(rp));
return sec::none;
}
// Manager requires a next stage.
CAF_LOG_INFO("broken stream pipeline");
rp.deliver(sec::no_downstream_stages_defined);
return sec::no_downstream_stages_defined;
} else if (mgr->out().terminal()) {
// Cannot connect a sink to a next stage.
return sec::cannot_add_downstream;
}
auto slot = next_slot();
mgr->send_handshake(std::move(next), slot, current_sender(),
take_current_forwarding_stack(), current_message_id());
mgr->generate_messages();
pending_stream_managers_.emplace(slot, std::move(mgr));
return sec::none;
}
// -- timeout management -------------------------------------------------------
uint32_t scheduled_actor::request_timeout(const duration& d) {
uint64_t scheduled_actor::set_receive_timeout(actor_clock::time_point x) {
setf(has_timeout_flag);
return set_timeout(receive_atom::value, x);
}
uint64_t scheduled_actor::set_receive_timeout() {
CAF_LOG_TRACE("");
if (bhvr_stack_.empty())
return 0;
auto d = bhvr_stack_.back().timeout();
if (!d.valid()) {
unsetf(has_timeout_flag);
return 0;
}
setf(has_timeout_flag);
auto result = ++timeout_id_;
auto msg = make_message(timeout_msg{result});
CAF_LOG_TRACE("send new timeout_msg, " << CAF_ARG(timeout_id_));
if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s
enqueue(ctrl(), invalid_message_id, std::move(msg), context());
} else {
auto id = ++timeout_id_;
auto type = receive_atom::value;
eq_impl(make_message_id(), nullptr, context(), timeout_msg{type, id});
return id;
}
auto t = clock().now();
t += d;
clock().set_receive_timeout(t, this, result);
}
return result;
return set_receive_timeout(t);
}
void scheduled_actor::reset_timeout(uint32_t timeout_id) {
if (is_active_timeout(timeout_id))
void scheduled_actor::reset_receive_timeout(uint64_t timeout_id) {
if (is_active_receive_timeout(timeout_id))
unsetf(has_timeout_flag);
}
bool scheduled_actor::is_active_timeout(uint32_t tid) const {
bool scheduled_actor::is_active_receive_timeout(uint64_t tid) const {
return getf(has_timeout_flag) && timeout_id_ == tid;
}
uint64_t scheduled_actor::set_stream_timeout(actor_clock::time_point x) {
if (x == actor_clock::time_point::max())
return 0;
return set_timeout(stream_atom::value, x);
}
// -- message processing -------------------------------------------------------
void scheduled_actor::add_awaited_response_handler(message_id response_id,
......@@ -384,11 +532,18 @@ scheduled_actor::categorize(mailbox_element& x) {
}
return message_category::ordinary;
case make_type_token<timeout_msg>(): {
CAF_ASSERT(!x.mid.valid());
auto& tm = content.get_as<timeout_msg>(0);
auto tid = tm.timeout_id;
CAF_ASSERT(!x.mid.valid());
return is_active_timeout(tid) ? message_category::timeout
: message_category::expired_timeout;
if (tm.type == receive_atom::value) {
CAF_LOG_DEBUG("handle timeout message");
if (is_active_receive_timeout(tid) && !bhvr_stack_.empty())
bhvr_stack_.back().handle_timeout();
} else {
CAF_ASSERT(tm.type == atom("stream"));
set_stream_timeout(advance_streams(clock().now()));
}
return message_category::internal;
}
case make_type_token<exit_msg>(): {
auto em = content.move_if_unshared<exit_msg>(0);
......@@ -413,13 +568,10 @@ scheduled_actor::categorize(mailbox_element& x) {
call_handler(error_handler_, this, err);
return message_category::internal;
}
/*
case make_type_token<stream_msg>(): {
auto& bs = bhvr_stack();
handle_stream_msg(x, bs.empty() ? nullptr : &bs.back());
case make_type_token<open_stream_msg>(): {
handle_open_stream_msg(x);
return message_category::internal;
}
*/
default:
return message_category::ordinary;
}
......@@ -488,19 +640,9 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
}
// Dispatch on the content of x.
switch (categorize(x)) {
case message_category::expired_timeout:
CAF_LOG_DEBUG("dropped expired timeout message");
return im_dropped;
case message_category::internal:
CAF_LOG_DEBUG("handled system message");
return im_success;
case message_category::timeout: {
CAF_LOG_DEBUG("handle timeout message");
if (bhvr_stack_.empty())
return im_dropped;
bhvr_stack_.back().handle_timeout();
return im_success;
}
case message_category::ordinary: {
detail::default_invoke_result_visitor<scheduled_actor> visitor{this};
bool skipped = false;
......@@ -599,8 +741,8 @@ auto scheduled_actor::activate(execution_unit* ctx, mailbox_element& x)
if (!activate(ctx))
return activation_result::terminated;
auto res = reactivate(x);
if (res == activation_result::success && !bhvr_stack_.empty())
request_timeout(bhvr_stack_.back().timeout());
if (res == activation_result::success)
set_receive_timeout();
return res;
}
......@@ -646,8 +788,8 @@ void scheduled_actor::do_become(behavior bhvr, bool discard_old) {
if (discard_old && !bhvr_stack_.empty())
bhvr_stack_.pop_back();
// request_timeout simply resets the timeout when it's invalid
request_timeout(bhvr.timeout());
bhvr_stack_.push_back(std::move(bhvr));
set_receive_timeout();
}
bool scheduled_actor::finalize() {
......@@ -680,23 +822,22 @@ void scheduled_actor::push_to_cache(mailbox_element_ptr ptr) {
inbound_path* scheduled_actor::make_inbound_path(stream_manager_ptr mgr,
stream_slots slots,
strong_actor_ptr sender) {
/*
auto res = get<2>(mailbox_.queues()).queues().emplace(slots.receiver, nullptr);
auto& qs = mailbox_.queue().queues();
auto res = get<2>(qs).queues().emplace(slots.receiver, nullptr);
if (!res.second)
return nullptr;
auto path = new inbound_path(std::move(mgr), slots, std::move(sender));
res.first->second.policy().handler.reset(path);
return path;
*/
}
void scheduled_actor::erase_inbound_path_later(stream_slot slot) {
/*
get<2>(mailbox_.queues()).erase_later(slot);
*/
constexpr size_t id = mailbox_policy::downstream_queue_index;
get<id>(mailbox_.queue().queues()).erase_later(slot);
}
void scheduled_actor::erase_inbound_path_later(stream_slot slot, error reason) {
void scheduled_actor::erase_inbound_path_later(stream_slot, error) {
CAF_LOG_ERROR("implement me");
/*
using fn = void (*)(local_actor*, inbound_path&, error&);
fn regular = [](local_actor* self, inbound_path& in, error&) {
......@@ -716,16 +857,18 @@ void scheduled_actor::erase_inbound_path_later(stream_slot slot, error reason) {
*/
}
void scheduled_actor::erase_inbound_paths_later(const stream_manager* mgr) {
/*
for (auto& kvp : get<2>(mailbox_.queues()).queues())
if (kvp.second.policy().handler->mgr == mgr)
void scheduled_actor::erase_inbound_paths_later(const stream_manager* ptr) {
constexpr size_t id = mailbox_policy::downstream_queue_index;
for (auto& kvp : get<id>(mailbox_.queue().queues()).queues()) {
auto& path = kvp.second.policy().handler;
if (path != nullptr && path->mgr == ptr)
erase_inbound_path_later(kvp.first);
*/
}
}
void scheduled_actor::erase_inbound_paths_later(const stream_manager* mgr,
error reason) {
void scheduled_actor::erase_inbound_paths_later(const stream_manager*,
error) {
CAF_LOG_ERROR("implement me");
/*
using fn = void (*)(local_actor*, inbound_path&, error&);
fn regular = [](local_actor* self, inbound_path& in, error&) {
......@@ -745,6 +888,35 @@ void scheduled_actor::erase_inbound_paths_later(const stream_manager* mgr,
*/
}
void scheduled_actor::handle(stream_slots slots, actor_addr& sender,
upstream_msg::ack_open& x) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(sender) << CAF_ARG(x));
CAF_ASSERT(sender == x.rebind_to);
// Get the manager for that stream, move it from `pending_managers_` to
// `managers_`, and handle `x`.
auto i = pending_stream_managers_.find(slots.receiver);
if (i == pending_stream_managers_.end()) {
CAF_LOG_WARNING("found no corresponding manager for received ack_open");
return;
}
auto ptr = std::move(i->second);
pending_stream_managers_.erase(i);
if (!add_stream_manager(slots, ptr)) {
CAF_LOG_WARNING("unable to add stream manager after receiving ack_open");
return;
}
ptr->handle(slots, x);
}
uint64_t scheduled_actor::set_timeout(atom_value type,
actor_clock::time_point x) {
CAF_LOG_TRACE(CAF_ARG(type) << CAF_ARG(x));
auto id = ++timeout_id_;
CAF_LOG_DEBUG("set timeout:" << CAF_ARG(type) << CAF_ARG(x));
clock().set_ordinary_timeout(x, this, type, id);
return id;
}
stream_slot scheduled_actor::next_slot() {
stream_slot result = 0;
auto nslot = [](stream_slot x) -> stream_slot {
......@@ -757,4 +929,114 @@ stream_slot scheduled_actor::next_slot() {
return result;
}
bool scheduled_actor::add_stream_manager(stream_slots id,
stream_manager_ptr ptr) {
CAF_LOG_TRACE(CAF_ARG(id));
if (stream_managers_.empty())
stream_ticks_.start(clock().now());
auto result = stream_managers_.emplace(id, std::move(ptr)).second;
return result;
}
bool scheduled_actor::remove_stream_manager(stream_slots id) {
if (stream_managers_.erase(id) == 0)
return false;
if (stream_managers_.empty())
stream_ticks_.stop();
return true;
}
invoke_message_result
scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x));
// Fetches a stream manger from a behavior.
struct visitor : detail::invoke_result_visitor {
stream_manager_ptr ptr;
stream_slots id;
void operator()() override {
// nop
}
void operator()(error&) override {
// nop
}
void operator()(message&) override {
// nop
}
void operator()(stream_slot slot, stream_manager_ptr& x) override {
id.receiver = slot;
ptr = std::move(x);
}
void operator()(const none_t&) override {
// nop
}
};
auto& bs = bhvr_stack();
if (bs.empty()) {
// TODO: send forced_drop
return im_dropped;
}
// Extract the handshake part of the message.
CAF_ASSERT(x.content().match_elements<open_stream_msg>());
auto& osm = x.content().get_mutable_as<open_stream_msg>(0);
visitor f;
f.id.sender = osm.slot;
auto res = (bs.back())(f, osm.msg);
switch (res) {
case match_case::result::no_match:
// TODO: send forced_drop
CAF_LOG_DEBUG("unmatched open_stream_msg content:" << osm.msg);
return im_dropped;
case match_case::result::match: {
if (f.ptr == nullptr) {
CAF_LOG_WARNING("actor did not return a stream manager after "
"receiving open_stream_msg");
// TODO: send forced_drop
return im_dropped;
}
auto path = make_inbound_path(f.ptr, f.id, std::move(osm.prev_stage));
CAF_ASSERT(path != nullptr);
path->emit_ack_open(this, actor_cast<actor_addr>(osm.original_stage));
// Propagate handshake down the pipeline.
// TODO: error handling
build_pipeline(std::move(f.ptr));
return im_success;
}
default:
return im_skipped; // nop
}
}
actor_clock::time_point
scheduled_actor::advance_streams(actor_clock::time_point now) {
CAF_LOG_TRACE("");
if (!stream_ticks_.started())
return actor_clock::time_point::max();
/// Advance time for driving forced batches and credit.
auto bitmask = stream_ticks_.timeouts(now, {max_batch_delay_ticks_,
credit_round_ticks_});
// Force batches on all output paths.
if ((bitmask & 0x01) != 0) {
for (auto& kvp : stream_managers_)
kvp.second->out().force_emit_batches();
}
// Fill up credit on each input path.
if ((bitmask & 0x02) != 0) {
auto cycle = stream_ticks_.interval();
cycle *= credit_round_ticks_;
auto bc_us = home_system().config().streaming_desired_batch_commplexity_us;
auto bc = std::chrono::microseconds(bc_us);
auto& qs = get<2>(mailbox_.queue().queues()).queues();
for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get();
inptr->emit_ack_batch(this, kvp.second.total_task_size(), cycle, bc);
}
}
return stream_ticks_.next_timeout(now, {max_batch_delay_ticks_,
credit_round_ticks_});
}
} // namespace caf
......@@ -34,9 +34,10 @@
namespace caf {
stream_manager::stream_manager(local_actor* selfptr)
stream_manager::stream_manager(local_actor* selfptr, stream_priority prio)
: self_(selfptr),
pending_handshakes_(0) {
pending_handshakes_(0),
priority_(prio) {
// nop
}
......@@ -86,6 +87,11 @@ void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) {
void stream_manager::close() {
out().close();
self_->erase_inbound_paths_later(this);
if (!promises_.empty()) {
auto msg = make_final_result();
for (auto& x : promises_)
x.deliver(msg);
}
}
void stream_manager::abort(error reason) {
......@@ -107,27 +113,30 @@ bool stream_manager::congested() const {
void stream_manager::send_handshake(strong_actor_ptr dest, stream_slot slot,
strong_actor_ptr client,
mailbox_element::forwarding_stack fwd_stack,
message_id mid, stream_priority prio) {
message_id mid) {
CAF_ASSERT(dest != nullptr);
++pending_handshakes_;
dest->enqueue(
make_mailbox_element(
std::move(client), mid, std::move(fwd_stack),
open_stream_msg{slot, make_handshake(), self_->ctrl(), dest, prio}),
open_stream_msg{slot, make_handshake(), self_->ctrl(), dest, priority_}),
self_->context());
}
void stream_manager::send_handshake(strong_actor_ptr dest, stream_slot slot,
stream_priority prio) {
void stream_manager::send_handshake(strong_actor_ptr dest, stream_slot slot) {
mailbox_element::forwarding_stack fwd_stack;
send_handshake(std::move(dest), slot, nullptr, std::move(fwd_stack),
make_message_id(), prio);
make_message_id());
}
bool stream_manager::generate_messages() {
return false;
}
void stream_manager::cycle_timeout(size_t) {
// TODO: make pure virtual
}
void stream_manager::register_input_path(inbound_path* ptr) {
inbound_paths_.emplace_back(ptr);
}
......@@ -142,6 +151,12 @@ void stream_manager::deregister_input_path(inbound_path* ptr) noexcept {
inbound_paths_.pop_back();
}
void stream_manager::add_promise(response_promise x) {
CAF_LOG_TRACE(CAF_ARG(x));
CAF_ASSERT(out().terminal());
promises_.emplace_back(std::move(x));
}
message stream_manager::make_final_result() {
return none;
}
......@@ -156,7 +171,7 @@ void stream_manager::output_closed(error) {
}
message stream_manager::make_handshake() const {
CAF_LOG_ERROR("stream_manager::make_output_token called");
CAF_LOG_ERROR("stream_manager::make_handshake called");
return none;
}
......
......@@ -28,9 +28,6 @@
using namespace std;
using namespace caf;
// TODO: use std::literals::chrono_literals when switching to C++14
using ns = std::chrono::nanoseconds;
template <class... Ts>
void print(const char* format, Ts... xs) {
char buf[200];
......@@ -60,13 +57,14 @@ struct fixture {
c, n, t, m);
print("- items/batch B = max(D * N / t, 1) = max(%ld * %ld / %ld, 1) = %ld",
d, n, t, b);
auto cr = x.calculate(ns(1000), ns(100));
auto cr = x.calculate(timespan(1000), timespan(100));
CAF_CHECK_EQUAL(cr.items_per_batch, b);
CAF_CHECK_EQUAL(cr.max_throughput, m);
}
void store(long batch_size, long calculation_time_ns) {
inbound_path::stats_t::measurement m{batch_size, ns{calculation_time_ns}};
inbound_path::stats_t::measurement m{batch_size,
timespan{calculation_time_ns}};
x.store(m);
}
};
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#define CAF_SUITE local_streaming
#include "caf/test/dsl.hpp"
#include <memory>
#include <numeric>
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/stateful_actor.hpp"
using std::string;
using std::vector;
using namespace caf;
namespace {
struct file_reader_state {
static const char* name;
};
const char* file_reader_state::name = "file_reader";
behavior file_reader(stateful_actor<file_reader_state>* self, size_t buf_size) {
using buf = std::deque<int>;
return {
[=](string& fname) -> output_stream<int, string> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return self->make_source(
// forward file name in handshake to next stage
std::forward_as_tuple(std::move(fname)),
// initialize state
[&](buf& xs) {
xs.resize(buf_size);
std::iota(xs.begin(), xs.end(), 0);
},
// get next element
[=](buf& xs, downstream<int>& out, size_t num) {
CAF_MESSAGE("push " << num << " more messages downstream");
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();
}
);
}
};
}
struct sum_up_state {
static const char* name;
};
const char* sum_up_state::name = "sum_up";
behavior sum_up(stateful_actor<sum_up_state>* self) {
return {
[=](stream<int>& in, string& fname) {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->make_sink(
// input stream
in,
// initialize state
[](int& x) {
x = 0;
},
// processing step
[](int& x, int y) {
x += y;
},
// cleanup and produce result message
[](int& x) -> int {
return x;
}
);
}
};
}
struct fixture : test_coordinator_fixture<> {
};
error fail_state(const actor& x) {
auto ptr = actor_cast<abstract_actor*>(x);
return dynamic_cast<monitorable_actor&>(*ptr).fail_state();
}
} // namespace <anonymous>
// -- unit tests ---------------------------------------------------------------
CAF_TEST_FIXTURE_SCOPE(local_streaming_tests, fixture)
CAF_TEST(depth_2_pipeline) {
std::chrono::microseconds cycle{cfg.streaming_credit_round_interval_us};
sched.clock().current_time += cycle;
auto src = sys.spawn(file_reader, 10);
auto snk = sys.spawn(sum_up);
auto pipeline = snk * src;
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(snk));
// self --("numbers.txt")-----------> source sink
self->send(pipeline, "numbers.txt");
expect((string), from(self).to(src).with("numbers.txt"));
// self source --(open_stream_msg)-------> sink
expect((open_stream_msg), from(self).to(snk));
// self source <--------------(ack_open)-- sink
expect((upstream_msg::ack_open), from(snk).to(src));
// self source --(batch)-----------------> sink
expect((downstream_msg::batch), from(src).to(snk));
// self source <-------------(ack_batch)-- sink
sched.clock().current_time += cycle;
sched.dispatch();
expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(src).to(src));
expect((upstream_msg::ack_batch), from(snk).to(src));
// self source --(close)-----------------> sink
expect((downstream_msg::close), from(src).to(snk));
// self <-----------------------------------------------------(result)-- sink
expect((int), from(snk).to(self).with(45));
CAF_CHECK_EQUAL(fail_state(snk), exit_reason::normal);
CAF_CHECK_EQUAL(fail_state(src), exit_reason::normal);
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -261,7 +261,7 @@ public:
int sentinel_;
};
auto ptr = detail::make_stream_source<driver>(this, num_messages);
ptr->send_handshake(ref.ctrl(), slot, stream_priority::normal);
ptr->send_handshake(ref.ctrl(), slot);
ptr->generate_messages();
pending_managers_.emplace(slot, std::move(ptr));
}
......@@ -288,7 +288,7 @@ public:
vector<int>* log_;
};
forwarder = detail::make_stream_stage<driver>(this, &data);
forwarder->send_handshake(ref.ctrl(), slot, stream_priority::normal);
forwarder->send_handshake(ref.ctrl(), slot);
pending_managers_.emplace(slot, forwarder);
}
......@@ -483,8 +483,6 @@ struct msg_visitor {
auto& dm = x.content().get_mutable_as<downstream_msg>(0);
auto f = detail::make_overload(
[&](downstream_msg::batch& y) {
TRACE(self->name(), batch, CAF_ARG2("size", y.xs_size),
CAF_ARG2("remaining_credit", inptr->assigned_credit - y.xs_size));
inptr->handle(y);
if (inptr->mgr->done()) {
CAF_MESSAGE(self->name()
......@@ -599,7 +597,7 @@ struct fixture {
template <class... Ts>
void next_cycle(Ts&... xs) {
entity* es[] = {&xs...};
CAF_MESSAGE("advance clock by " << tc.credit_interval.count() << "ns");
//CAF_MESSAGE("advance clock by " << tc.credit_interval.count() << "ns");
tc.global_time += tc.credit_interval;
for (auto e : es)
e->advance_time();
......@@ -650,7 +648,7 @@ CAF_TEST(depth_2_pipeline_single_round) {
}
CAF_TEST(depth_2_pipeline_multiple_rounds) {
constexpr size_t num_messages = 200000;
constexpr size_t num_messages = 2000;
alice.start_streaming(bob, num_messages);
loop_until([&] { return done_streaming(); }, alice, bob);
CAF_CHECK_EQUAL(bob.data, make_iota(0, num_messages));
......@@ -667,7 +665,7 @@ CAF_TEST(depth_3_pipeline_single_round) {
}
CAF_TEST(depth_3_pipeline_multiple_rounds) {
constexpr size_t num_messages = 200000;
constexpr size_t num_messages = 2000;
bob.forward_to(carl);
alice.start_streaming(bob, num_messages);
CAF_MESSAGE("loop over alice and bob until bob is congested");
......
......@@ -322,8 +322,13 @@ struct fixture {
actor_system system;
scoped_actor self;
fixture() : system(cfg.add_message_type<get_state_msg>("get_state_msg")),
self(system) {
static actor_system_config& init(actor_system_config& cfg) {
cfg.add_message_type<get_state_msg>("get_state_msg");
cfg.parse(test::engine::argc(), test::engine::argv());
return cfg;
}
fixture() : system(init(cfg)), self(system) {
// nop
}
......@@ -376,10 +381,11 @@ CAF_TEST_FIXTURE_SCOPE(typed_spawn_tests, fixture)
******************************************************************************/
CAF_TEST(typed_spawns) {
// run test series with typed_server(1|2)
CAF_MESSAGE("run test series with typed_server1");
test_typed_spawn(system.spawn(typed_server1));
self->await_all_other_actors_done();
CAF_MESSAGE("finished test series with `typed_server1`");
CAF_MESSAGE("run test series with typed_server2");
test_typed_spawn(system.spawn(typed_server2));
self->await_all_other_actors_done();
CAF_MESSAGE("finished test series with `typed_server2`");
......
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