Commit e1f69c79 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/streaming-tweaks'

parents 23c12a3e 39c79fa4
......@@ -43,11 +43,7 @@ behavior.
\begin{center}
\begin{tabular}{ll}
\textbf{Types} & ~ \\
\hline
\lstinline^rtti_pair^ & \lstinline^std::pair<uint16_t, const std::type_info*>^ \\
\hline
~ & ~ \\ \textbf{Observers} & ~ \\
\textbf{Observers} & ~ \\
\hline
\lstinline^bool empty()^ & Returns whether this message is empty. \\
\hline
......
......@@ -91,6 +91,7 @@ set(LIBCAF_CORE_SRCS
src/response_promise.cpp
src/resumable.cpp
src/ripemd_160.cpp
src/rtti_pair.cpp
src/runtime_settings_map.cpp
src/scheduled_actor.cpp
src/scoped_actor.cpp
......
......@@ -53,14 +53,6 @@
namespace caf {
using rtti_pair = std::pair<uint16_t, const std::type_info*>;
using rtti_pair_vec = std::vector<rtti_pair>;
using rtti_pair_vec_triple = std::tuple<rtti_pair_vec,
rtti_pair_vec,
rtti_pair_vec>;
template <class T>
struct mpi_field_access {
std::string operator()(const uniform_type_info_map& types) {
......
......@@ -21,11 +21,12 @@
#include <tuple>
#include <stdexcept>
#include "caf/type_nr.hpp"
#include "caf/serializer.hpp"
#include "caf/deserializer.hpp"
#include "caf/deep_to_string.hpp"
#include "caf/deserializer.hpp"
#include "caf/make_type_erased_value.hpp"
#include "caf/rtti_pair.hpp"
#include "caf/serializer.hpp"
#include "caf/type_nr.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/safe_equal.hpp"
......@@ -65,20 +66,6 @@ struct void_ptr_access {
}
};
template <class T, uint16_t N = type_nr<T>::value>
struct tuple_vals_type_helper {
static typename message_data::rtti_pair get() noexcept {
return {N, nullptr};
}
};
template <class T>
struct tuple_vals_type_helper<T, 0> {
static typename message_data::rtti_pair get() noexcept {
return {0, &typeid(T)};
}
};
template <class Base, class... Ts>
class tuple_vals_impl : public Base {
public:
......@@ -90,8 +77,6 @@ public:
using super = message_data;
using rtti_pair = typename message_data::rtti_pair;
using data_type = std::tuple<Ts...>;
// -- friend functions -------------------------------------------------------
......@@ -107,7 +92,7 @@ public:
template <class... Us>
tuple_vals_impl(Us&&... xs)
: data_(std::forward<Us>(xs)...),
types_{{tuple_vals_type_helper<Ts>::get()...}} {
types_{{make_rtti_pair<Ts>()...}} {
// nop
}
......
......@@ -21,8 +21,11 @@
#include <cstddef>
#include <cstdint>
#include "caf/actor_clock.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/downstream_msg.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/rtti_pair.hpp"
#include "caf/stream_aborter.hpp"
#include "caf/stream_manager.hpp"
#include "caf/stream_priority.hpp"
......@@ -30,8 +33,6 @@
#include "caf/timestamp.hpp"
#include "caf/upstream_msg.hpp"
#include "caf/meta/type_name.hpp"
namespace caf {
/// State for a path to an upstream actor (source).
......@@ -70,9 +71,6 @@ public:
/// Amount of credit we assign sources after receiving `open`.
static constexpr int initial_credit = 50;
/// Keep track of measurements for the last X batches.
static constexpr size_t stats_sampling_size = 16;
/// Stores statistics for measuring complexity of incoming batches.
struct stats_t {
/// Wraps a time measurement for a single processed batch.
......@@ -91,27 +89,38 @@ public:
int32_t items_per_batch;
};
stats_t();
/// Total number of elements in all processed batches.
int64_t num_elements;
/// Stores `stats_sampling_size` measurements in a ring.
std::vector<measurement> measurements;
/// Elapsed time for processing all elements of all batches.
timespan processing_time;
/// Current position in `measurements`
size_t ring_iter;
stats_t();
/// Returns the maximum number of items this actor could handle for given
/// cycle length with a minimum of 1.
calculation_result calculate(timespan cycle, timespan desired_complexity);
/// Stores a new measurement in the ring buffer.
/// Adds a measurement to this statistic.
void store(measurement x);
/// Resets this statistic.
void reset();
};
/// Summarizes how many elements we processed during the last cycle and how
/// much time we spent processing those elements.
stats_t stats;
/// Stores the time point of the last credit decision for this source.
actor_clock::time_point last_credit_decision;
/// Stores the time point of the last credit decision for this source.
actor_clock::time_point next_credit_decision;
/// Constructs a path for given handle and stream ID.
inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
strong_actor_ptr ptr);
strong_actor_ptr ptr, rtti_pair input_type);
~inbound_path();
......@@ -138,7 +147,8 @@ public:
/// @param cycle Time between credit rounds.
/// @param desired_batch_complexity Desired processing time per batch.
void emit_ack_batch(local_actor* self, int32_t queued_items,
int32_t max_downstream_capacity, timespan cycle,
int32_t max_downstream_capacity,
actor_clock::time_point now, timespan cycle,
timespan desired_batch_complexity);
/// Returns whether the path received no input since last emitting
......@@ -155,6 +165,9 @@ public:
static void emit_irregular_shutdown(local_actor* self, stream_slots slots,
const strong_actor_ptr& hdl,
error reason);
private:
actor_clock& clock();
};
/// @relates inbound_path
......
......@@ -613,3 +613,20 @@ inline caf::actor_id caf_set_aid_dummy() { return 0; }
"TERMINATE ; ID =" << thisptr->id() \
<< "; REASON =" << deep_to_string(rsn).c_str() \
<< "; NODE =" << thisptr->node())
// -- macros for logging streaming-related events ------------------------------
/// The log component for logging streaming-related events that are crucial for
/// understanding handshaking, credit decisions, etc.
#define CAF_LOG_STREAM_COMPONENT "caf_stream"
#if CAF_LOG_LEVEL >= CAF_LOG_LEVEL_DEBUG
#define CAF_STREAM_LOG_DEBUG(output) \
CAF_LOG_IMPL(CAF_LOG_STREAM_COMPONENT, CAF_LOG_LEVEL_DEBUG, output)
#define CAF_STREAM_LOG_DEBUG_IF(condition, output) \
if (condition) \
CAF_LOG_IMPL(CAF_LOG_STREAM_COMPONENT, CAF_LOG_LEVEL_DEBUG, output)
#else
#define CAF_STREAM_LOG_DEBUG(unused) CAF_VOID_STMT
#define CAF_STREAM_LOG_DEBUG_IF(unused1, unused2) CAF_VOID_STMT
#endif
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <cstdint>
#include <type_traits>
#include <typeinfo>
#include <utility>
#include "caf/type_nr.hpp"
namespace caf {
/// Bundles the type number with its C++ `type_info` object. The type number is
/// non-zero for builtin types and the pointer to the `type_info` object is
/// non-null for custom types.
using rtti_pair = std::pair<uint16_t, const std::type_info*>;
/// @relates rtti_pair
template <class T>
typename std::enable_if<type_nr<T>::value == 0, rtti_pair>::type
make_rtti_pair() {
return {0, &typeid(T)};
}
/// @relates rtti_pair
template <class T>
typename std::enable_if<type_nr<T>::value != 0, rtti_pair>::type
make_rtti_pair() {
auto n = type_nr<T>::value;
return {n, nullptr};
}
/// @relates rtti_pair
std::string to_string(rtti_pair x);
} // namespace caf
......@@ -734,7 +734,8 @@ public:
/// Creates a new path for incoming stream traffic from `sender`.
virtual inbound_path* make_inbound_path(stream_manager_ptr mgr,
stream_slots slots,
strong_actor_ptr sender);
strong_actor_ptr sender,
rtti_pair rtti);
/// Silently closes incoming stream traffic on `slot`.
virtual void erase_inbound_path_later(stream_slot slot);
......
......@@ -18,9 +18,10 @@
#pragma once
#include <vector>
#include <cstdint>
#include <cstddef>
#include <cstdint>
#include <string>
#include <vector>
#include "caf/actor.hpp"
#include "caf/actor_cast.hpp"
......@@ -246,7 +247,7 @@ public:
/// @private
template <class In>
stream_slot add_unchecked_inbound_path(const stream<In>&) {
return add_unchecked_inbound_path_impl();
return add_unchecked_inbound_path_impl(make_rtti_pair<In>());
}
/// Adds a new outbound path to `rp.next()`.
......@@ -265,7 +266,7 @@ public:
/// Adds the current sender as an inbound path.
/// @pre Current message is an `open_stream_msg`.
stream_slot add_unchecked_inbound_path_impl();
stream_slot add_unchecked_inbound_path_impl(rtti_pair rtti);
protected:
// -- modifiers for self -----------------------------------------------------
......
......@@ -20,12 +20,14 @@
#include <algorithm>
#include <tuple>
#include <typeinfo>
#include "caf/detail/type_traits.hpp"
#include "caf/inbound_path.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/rtti_pair.hpp"
#include "caf/stream_manager.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/type_nr.hpp"
namespace caf {
......@@ -62,10 +64,13 @@ public:
/// Creates a new input path to the current sender.
inbound_stream_slot<input_type> add_inbound_path(const stream<input_type>&) {
return {add_unchecked_inbound_path_impl()};
auto rtti = make_rtti_pair<input_type>();
return {this->add_unchecked_inbound_path_impl(rtti)};
}
private:
// -- member variables -------------------------------------------------------
downstream_manager dummy_out_;
};
......
......@@ -23,24 +23,20 @@
#include <cstdint>
#include <typeinfo>
#include "caf/detail/apply_args.hpp"
#include "caf/detail/pseudo_tuple.hpp"
#include "caf/detail/try_match.hpp"
#include "caf/fwd.hpp"
#include "caf/type_nr.hpp"
#include "caf/optional.hpp"
#include "caf/rtti_pair.hpp"
#include "caf/type_erased_value.hpp"
#include "caf/detail/try_match.hpp"
#include "caf/detail/apply_args.hpp"
#include "caf/detail/pseudo_tuple.hpp"
#include "caf/type_nr.hpp"
namespace caf {
/// Represents a tuple of type-erased values.
class type_erased_tuple {
public:
// -- member types -----------------------------------------------------------
using rtti_pair = std::pair<uint16_t, const std::type_info*>;
// -- constructors, destructors, and assignment operators --------------------
type_erased_tuple() = default;
......
......@@ -22,8 +22,9 @@
#include <typeinfo>
#include <functional>
#include "caf/fwd.hpp"
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/rtti_pair.hpp"
#include "caf/type_nr.hpp"
namespace caf {
......@@ -31,10 +32,6 @@ namespace caf {
/// Represents a single type-erased value.
class type_erased_value {
public:
// -- member types -----------------------------------------------------------
using rtti_pair = std::pair<uint16_t, const std::type_info*>;
// -- constructors, destructors, and assignment operators --------------------
virtual ~type_erased_value();
......
......@@ -77,7 +77,7 @@ uint32_t concatenated_tuple::type_token() const noexcept {
return type_token_;
}
message_data::rtti_pair concatenated_tuple::type(size_t pos) const noexcept {
rtti_pair concatenated_tuple::type(size_t pos) const noexcept {
CAF_ASSERT(pos < size());
auto selected = select(pos);
return selected.first->type(selected.second);
......
......@@ -71,7 +71,7 @@ uint32_t decorated_tuple::type_token() const noexcept {
return type_token_;
}
message_data::rtti_pair decorated_tuple::type(size_t pos) const noexcept {
rtti_pair decorated_tuple::type(size_t pos) const noexcept {
CAF_ASSERT(pos < size());
return decorated_->type(mapping_[pos]);
}
......
......@@ -25,9 +25,8 @@
namespace caf {
inbound_path::stats_t::stats_t() : ring_iter(0) {
measurement x{0, timespan{0}};
measurements.resize(stats_sampling_size, x);
inbound_path::stats_t::stats_t() : num_elements(0), processing_time(0) {
// nop
}
auto inbound_path::stats_t::calculate(timespan c, timespan d)
......@@ -37,12 +36,7 @@ auto inbound_path::stats_t::calculate(timespan c, timespan d)
// instead of C.
// We compute our values in 64-bit for more precision before truncating to a
// 32-bit integer type at the end.
int64_t total_ns = 0;
int64_t total_items = 0;
for (auto& x : measurements) {
total_ns += x.calculation_time.count();
total_items += x.batch_size;
}
int64_t total_ns = processing_time.count();
if (total_ns == 0)
return {1, 1};
/// Helper for truncating a 64-bit integer to a 32-bit integer with a minimum
......@@ -57,17 +51,22 @@ auto inbound_path::stats_t::calculate(timespan c, timespan d)
};
// Instead of C * (N / t) we calculate (C * N) / t to avoid double conversion
// and rounding errors.
return {clamp((c.count() * total_items) / total_ns),
clamp((d.count() * total_items) / total_ns)};
return {clamp((c.count() * num_elements) / total_ns),
clamp((d.count() * num_elements) / total_ns)};
}
void inbound_path::stats_t::store(measurement x) {
measurements[ring_iter] = x;
ring_iter = (ring_iter + 1) % stats_sampling_size;
num_elements += x.batch_size;
processing_time += x.calculation_time;
}
void inbound_path::stats_t::reset() {
num_elements = 0;
processing_time = timespan{0};
}
inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
strong_actor_ptr ptr)
strong_actor_ptr ptr, rtti_pair in_type)
: mgr(std::move(mgr_ptr)),
hdl(std::move(ptr)),
slots(id),
......@@ -77,6 +76,10 @@ inbound_path::inbound_path(stream_manager_ptr mgr_ptr, stream_slots id,
last_acked_batch_id(0),
last_batch_id(0) {
mgr->register_input_path(this);
CAF_STREAM_LOG_DEBUG(mgr->self()->name()
<< "opens input stream with element type"
<< *mgr->self()->system().types().portable_name(in_type)
<< "at slot" << id.receiver << "from" << hdl);
}
inbound_path::~inbound_path() {
......@@ -85,14 +88,29 @@ inbound_path::~inbound_path() {
void inbound_path::handle(downstream_msg::batch& x) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x));
auto& clk = clock();
auto batch_size = x.xs_size;
assigned_credit -= batch_size;
last_batch_id = x.id;
auto& clock = mgr->self()->clock();
auto t0 = clock.now();
auto t0 = clk.now();
CAF_STREAM_LOG_DEBUG(mgr->self()->name() << "handles batch of size"
<< batch_size << "on slot" << slots.receiver << "with"
<< assigned_credit << "assigned credit");
if (assigned_credit <= batch_size) {
assigned_credit = 0;
// Do not log a message when "running out of credit" for the first batch
// that can easily consume the initial credit in one shot.
CAF_STREAM_LOG_DEBUG_IF(next_credit_decision.time_since_epoch().count() > 0,
mgr->self()->name() << "ran out of credit at slot"
<< slots.receiver << "with approx."
<< (next_credit_decision - t0)
<< "until next cycle");
} else {
assigned_credit -= batch_size;
CAF_ASSERT(assigned_credit >= 0);
}
mgr->handle(this, x);
auto t1 = clock.now();
auto dt = clock.difference(atom("batch"), batch_size, t0, t1);
auto t1 = clk.now();
auto dt = clk.difference(atom("batch"), batch_size, t0, t1);
stats.store({batch_size, dt});
mgr->push();
}
......@@ -101,6 +119,7 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(rebind_from));
// Update state.
assigned_credit = mgr->acquire_credit(this, initial_credit);
CAF_ASSERT(assigned_credit >= 0);
// Make sure we receive errors from this point on.
stream_aborter::add(hdl, self->address(), slots.receiver,
stream_aborter::source_aborter);
......@@ -109,18 +128,24 @@ void inbound_path::emit_ack_open(local_actor* self, actor_addr rebind_from) {
make<upstream_msg::ack_open>(
slots.invert(), self->address(), std::move(rebind_from),
self->ctrl(), assigned_credit, desired_batch_size));
last_credit_decision = clock().now();
}
void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
int32_t max_downstream_capacity,
timespan cycle, timespan complexity) {
actor_clock::time_point now, timespan cycle,
timespan complexity) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(queued_items)
<< CAF_ARG(max_downstream_capacity) << CAF_ARG(cycle)
<< CAF_ARG(complexity));
CAF_IGNORE_UNUSED(queued_items);
auto x = stats.calculate(cycle, complexity);
// Update timestamps.
last_credit_decision = now;
next_credit_decision = now + cycle;
// Hand out enough credit to fill our queue for 2 cycles but never exceed
// the downstream capacity.
auto x = stats.calculate(cycle, complexity);
auto stats_guard = detail::make_scope_guard([&] { stats.reset(); });
auto max_capacity = std::min(x.max_throughput * 2, max_downstream_capacity);
CAF_ASSERT(max_capacity > 0);
// Protect against overflow on `assigned_credit`.
......@@ -128,15 +153,23 @@ void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
// Compute the amount of credit we grant in this round.
auto credit = std::min(std::max(max_capacity - assigned_credit, 0),
max_new_credit);
CAF_ASSERT(credit >= 0);
// The manager can restrict or adjust the amount of credit.
credit = std::min(mgr->acquire_credit(this, credit), max_new_credit);
CAF_STREAM_LOG_DEBUG(mgr->self()->name() << "grants" << credit
<< "new credit at slot" << slots.receiver
<< "after receiving" << stats.num_elements
<< "elements that took" << stats.processing_time
<< CAF_ARG2("max_throughput", x.max_throughput)
<< CAF_ARG(max_downstream_capacity)
<< CAF_ARG(assigned_credit));
if (credit == 0 && up_to_date())
return;
CAF_LOG_DEBUG(CAF_ARG(assigned_credit) << CAF_ARG(max_capacity)
<< CAF_ARG(queued_items) << CAF_ARG(credit)
<< CAF_ARG(desired_batch_size));
if (credit > 0)
assigned_credit += credit;
assigned_credit += credit;
CAF_ASSERT(assigned_credit >= 0);
desired_batch_size = static_cast<int32_t>(x.items_per_batch);
unsafe_send_as(self, hdl,
make<upstream_msg::ack_batch>(slots.invert(), self->address(),
......@@ -177,4 +210,9 @@ void inbound_path::emit_irregular_shutdown(local_actor* self,
make<upstream_msg::forced_drop>(slots.invert(), self->address(),
std::move(reason)));
}
actor_clock& inbound_path::clock() {
return mgr->self()->clock();
}
} // namespace caf
......@@ -77,7 +77,7 @@ uint32_t merged_tuple::type_token() const noexcept {
return type_token_;
}
merged_tuple::rtti_pair merged_tuple::type(size_t pos) const noexcept {
rtti_pair merged_tuple::type(size_t pos) const noexcept {
CAF_ASSERT(pos < mapping_.size());
auto& p = mapping_[pos];
return data_[p.first]->type(p.second);
......
......@@ -77,7 +77,7 @@ uint32_t message::type_token() const noexcept {
return vals_ != nullptr ? vals_->type_token() : 0xFFFFFFFF;
}
message::rtti_pair message::type(size_t pos) const noexcept {
rtti_pair message::type(size_t pos) const noexcept {
CAF_ASSERT(vals_ != nullptr);
return vals_->type(pos);
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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/rtti_pair.hpp"
namespace caf {
std::string to_string(rtti_pair x) {
std::string result = "(";
result += std::to_string(x.first);
result += ", ";
result += x.second != nullptr ? x.second->name() : "<null>";
result += ")";
return result;
}
} // namespace caf
......@@ -879,14 +879,15 @@ scheduled_actor::urgent_queue& scheduled_actor::get_urgent_queue() {
inbound_path* scheduled_actor::make_inbound_path(stream_manager_ptr mgr,
stream_slots slots,
strong_actor_ptr sender) {
strong_actor_ptr sender,
rtti_pair rtti) {
static constexpr size_t queue_index = downstream_queue_index;
using policy_type = policy::downstream_messages::nested;
auto& qs = get<queue_index>(mailbox_.queue().queues()).queues();
auto res = qs.emplace(slots.receiver, policy_type{nullptr});
if (!res.second)
return nullptr;
auto path = new inbound_path(std::move(mgr), slots, std::move(sender));
auto path = new inbound_path(std::move(mgr), slots, std::move(sender), rtti);
res.first->second.policy().handler.reset(path);
return path;
}
......@@ -1120,20 +1121,6 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
CAF_LOG_DEBUG("no match in behavior, fall back to default handler");
return fallback();
case match_case::result::match: {
/*
if (f.ptr == nullptr) {
CAF_LOG_WARNING("actor did not return a stream manager after "
"handling open_stream_msg");
fail(sec::stream_init_failed, "behavior did not create a manager");
return im_dropped;
}
stream_slots path_id{osm.slot, f.in_slot};
auto path = make_inbound_path(f.ptr, path_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.
build_pipeline(f.in_slot, f.out_slot, std::move(f.ptr));
*/
return im_success;
}
default:
......@@ -1168,7 +1155,7 @@ scheduled_actor::advance_streams(actor_clock::time_point now) {
auto inptr = kvp.second.policy().handler.get();
auto bs = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(this, bs, inptr->mgr->out().max_capacity(),
cycle, bc);
now, cycle, bc);
}
}
return stream_ticks_.next_timeout(now, {max_batch_delay_ticks_,
......
......@@ -26,12 +26,13 @@
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/inbound_path.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/logger.hpp"
#include "caf/message.hpp"
#include "caf/outbound_path.hpp"
#include "caf/response_promise.hpp"
#include "caf/scheduled_actor.hpp"
#include "caf/sec.hpp"
#include "caf/type_nr.hpp"
namespace caf {
......@@ -149,6 +150,7 @@ void stream_manager::advance() {
CAF_LOG_TRACE("");
// Try to emit more credit.
if (!inbound_paths_.empty()) {
auto now = self_->clock().now();
auto& cfg = self_->system().config();
auto bc = cfg.stream_desired_batch_complexity;
auto interval = cfg.stream_credit_round_interval;
......@@ -159,7 +161,8 @@ void stream_manager::advance() {
// Ignore inbound paths of other managers.
if (inptr->mgr.get() == this) {
auto bs = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(self_, bs, out().max_capacity(), interval, bc);
inptr->emit_ack_batch(self_, bs, out().max_capacity(), now, interval,
bc);
}
}
}
......@@ -286,7 +289,7 @@ stream_manager::add_unchecked_outbound_path_impl(message handshake) {
return add_unchecked_outbound_path_impl(rp, std::move(handshake));
}
stream_slot stream_manager::add_unchecked_inbound_path_impl() {
stream_slot stream_manager::add_unchecked_inbound_path_impl(rtti_pair rtti) {
CAF_LOG_TRACE("");
auto x = self_->current_mailbox_element();
if (x == nullptr || !x->content().match_elements<open_stream_msg>()) {
......@@ -309,7 +312,8 @@ stream_slot stream_manager::add_unchecked_inbound_path_impl() {
}
auto slot = assign_next_slot();
stream_slots path_id{osm.slot, slot};
auto ptr = self_->make_inbound_path(this, path_id, std::move(osm.prev_stage));
auto ptr = self_->make_inbound_path(this, path_id, std::move(osm.prev_stage),
rtti);
CAF_ASSERT(ptr != nullptr);
ptr->emit_ack_open(self_, actor_cast<actor_addr>(osm.original_stage));
return slot;
......
......@@ -55,9 +55,9 @@ void push(std::deque<T>& xs, downstream<T>& out, size_t num) {
}
VARARGS_TESTEE(int_file_reader, size_t buf_size) {
using buf = std::deque<int>;
using buf = std::deque<int32_t>;
return {
[=](string& fname) -> output_stream<int> {
[=](string& fname) -> output_stream<int32_t> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
return self->make_source(
// initialize state
......@@ -66,7 +66,7 @@ VARARGS_TESTEE(int_file_reader, size_t buf_size) {
std::iota(xs.begin(), xs.end(), 1);
},
// get next element
[](buf& xs, downstream<int>& out, size_t num) {
[](buf& xs, downstream<int32_t>& out, size_t num) {
push(xs, out, num);
},
// check whether we reached the end
......@@ -107,7 +107,7 @@ TESTEE_STATE(sum_up) {
TESTEE(sum_up) {
using intptr = int*;
return {
[=](stream<int>& in) {
[=](stream<int32_t>& in) {
return self->make_sink(
// input stream
in,
......@@ -116,7 +116,7 @@ TESTEE(sum_up) {
x = &self->state.x;
},
// processing step
[](intptr& x, int y) {
[](intptr& x, int32_t y) {
*x += y;
},
// cleanup and produce result message
......@@ -241,7 +241,7 @@ TESTEE(stream_multiplexer) {
stg->out().assign<string_downstream_manager>(result);
return result;
},
[=](const stream<int>& in) {
[=](const stream<int32_t>& in) {
CAF_MESSAGE("received handshake for integers");
return self->state.stage->add_unchecked_inbound_path(in);
},
......@@ -252,6 +252,12 @@ TESTEE(stream_multiplexer) {
};
}
struct config : actor_system_config {
config() {
add_message_type<std::deque<std::string>>("deque<string>");
}
};
using fixture = test_coordinator_fixture<>;
} // namespace <anonymous>
......
......@@ -39,12 +39,6 @@ namespace {
struct fixture {
inbound_path::stats_t x;
size_t sampling_size = inbound_path::stats_sampling_size;
fixture() {
CAF_CHECK_EQUAL(x.measurements.size(), sampling_size);
CAF_CHECK_EQUAL(sampling_size % 2, 0u);
}
void calculate(int32_t total_items, int32_t total_time) {
int32_t c = 1000;
......@@ -86,26 +80,11 @@ CAF_TEST(one_store) {
}
CAF_TEST(multiple_stores) {
CAF_MESSAGE("store a measurement: (50, 500ns), (60, 400ns), (40, 600ns)");
CAF_MESSAGE("store measurements: (50, 500ns), (60, 400ns), (40, 600ns)");
store(50, 500);
store(40, 600);
store(60, 400);
calculate(150, 1500);
}
CAF_TEST(overriding_stores) {
CAF_MESSAGE("fill measurements with (100, 1000ns)");
for (size_t i = 0; i < sampling_size; ++i)
store(100, 1000);
calculate(100, 1000);
CAF_MESSAGE("override first half of the measurements with (10, 1000ns)");
for (size_t i = 0; i < sampling_size / 2; ++i)
store(10, 1000);
calculate(55, 1000);
CAF_MESSAGE("override second half of the measurements with (10, 1000ns)");
for (size_t i = 0; i < sampling_size / 2; ++i)
store(10, 1000);
calculate(10, 1000);
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -16,16 +16,18 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/config.hpp"
#include "caf/message.hpp"
#define CAF_SUITE match
#include "caf/test/unit_test.hpp"
#include <functional>
#include "caf/make_type_erased_tuple_view.hpp"
#include "caf/message_builder.hpp"
#include "caf/message_handler.hpp"
#include "caf/make_type_erased_tuple_view.hpp"
#include "caf/rtti_pair.hpp"
using namespace caf;
using namespace std;
......@@ -35,17 +37,6 @@ using ho_atom = atom_constant<atom("ho")>;
namespace {
using rtti_pair = std::pair<uint16_t, const std::type_info*>;
std::string to_string(const rtti_pair& x) {
std::string result = "(";
result += std::to_string(x.first);
result += ", ";
result += x.second != nullptr ? x.second->name() : "<null>";
result += ")";
return result;
}
struct fixture {
using array_type = std::array<bool, 4>;
......
......@@ -233,11 +233,11 @@ public:
using downstream_manager = broadcast_downstream_manager<int>;
struct driver final : public stream_source_driver<downstream_manager> {
public:
driver(int sentinel) : x_(0), sentinel_(sentinel) {
driver(int32_t sentinel) : x_(0), sentinel_(sentinel) {
// nop
}
void pull(downstream<int>& out, size_t hint) override {
void pull(downstream<int32_t>& out, size_t hint) override {
auto y = std::min(sentinel_, x_ + static_cast<int>(hint));
while (x_ < y)
out.push(x_++);
......@@ -247,8 +247,8 @@ public:
return x_ == sentinel_;
}
private:
int x_;
int sentinel_;
int32_t x_;
int32_t sentinel_;
};
auto mgr = detail::make_stream_source<driver>(this, num_messages);
auto res = mgr->add_outbound_path(ref.ctrl());
......@@ -260,9 +260,9 @@ public:
using downstream_manager = broadcast_downstream_manager<int>;
struct driver final : public stream_stage_driver<int, downstream_manager> {
public:
using super = stream_stage_driver<int, downstream_manager>;
using super = stream_stage_driver<int32_t, downstream_manager>;
driver(downstream_manager& out, vector<int>* log)
driver(downstream_manager& out, vector<int32_t>* log)
: super(out),
log_(log) {
// nop
......@@ -342,7 +342,8 @@ public:
for (auto& kvp : qs) {
auto inptr = kvp.second.policy().handler.get();
auto bs = static_cast<int32_t>(kvp.second.total_task_size());
inptr->emit_ack_batch(this, bs, 30, cycle, desired_batch_complexity);
inptr->emit_ack_batch(this, bs, 30, now(), cycle,
desired_batch_complexity);
}
}
};
......@@ -350,13 +351,15 @@ public:
}
inbound_path* make_inbound_path(stream_manager_ptr mgr, stream_slots slots,
strong_actor_ptr sender) override {
strong_actor_ptr sender,
rtti_pair rtti) override {
using policy_type = policy::downstream_messages::nested;
auto res = get<dmsg_id::value>(mbox.queues())
.queues().emplace(slots.receiver, policy_type{nullptr});
if (!res.second)
return nullptr;
auto path = new inbound_path(std::move(mgr), slots, std::move(sender));
auto path = new inbound_path(std::move(mgr), slots, std::move(sender),
rtti);
res.first->second.policy().handler.reset(path);
return path;
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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 rtti_pair
#include "caf/rtti_pair.hpp"
#include "caf/test/dsl.hpp"
using namespace caf;
namespace {
struct foo {
int x;
int y;
};
} // namespace <anonymous>
CAF_TEST(make_rtti_pair) {
auto n = type_nr<int32_t>::value;
CAF_REQUIRE_NOT_EQUAL(n, 0u);
auto x1 = make_rtti_pair<int32_t>();
CAF_CHECK_EQUAL(x1.first, n);
CAF_CHECK_EQUAL(x1.second, nullptr);
auto x2 = make_rtti_pair<foo>();
CAF_CHECK_EQUAL(x2.first, 0);
CAF_CHECK_EQUAL(x2.second, &typeid(foo));
}
CAF_TEST(to_string) {
auto n = type_nr<int32_t>::value;
CAF_CHECK_EQUAL(to_string(make_rtti_pair<int32_t>()),
"(" + std::to_string(n) + ", <null>)");
}
......@@ -172,7 +172,13 @@ TESTEE(log_consumer) {
};
}
using fixture = test_coordinator_fixture<>;
struct config : actor_system_config {
config() {
add_message_type<value_type>("value_type");
}
};
using fixture = test_coordinator_fixture<config>;
} // namespace <anonymous>
......
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