Commit 39c79fa4 authored by Dominik Charousset's avatar Dominik Charousset

Add stream logging to inbound_path

parent 537de9e6
......@@ -25,6 +25,7 @@
#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"
......@@ -119,7 +120,7 @@ public:
/// 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();
......
......@@ -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_;
};
......
......@@ -66,7 +66,7 @@ void inbound_path::stats_t::reset() {
}
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),
......@@ -76,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() {
......@@ -88,7 +92,7 @@ void inbound_path::handle(downstream_msg::batch& x) {
auto batch_size = x.xs_size;
last_batch_id = x.id;
auto t0 = clk.now();
CAF_STREAM_LOG_DEBUG("handle batch of size"
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) {
......@@ -96,10 +100,10 @@ void inbound_path::handle(downstream_msg::batch& x) {
// 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,
"source at slot" << slots.receiver
<< "ran out of credit with approx."
<< (next_credit_decision - t0)
<< "until next cycle");
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);
......@@ -138,16 +142,10 @@ void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
// Update timestamps.
last_credit_decision = now;
next_credit_decision = now + cycle;
// Short-circuit if we didn't receive anything during the last cycle.
if (stats.num_elements == 0)
return;
auto x = stats.calculate(cycle, complexity);
std::cout << "stats = " << stats.num_elements << "/"
<< deep_to_string(stats.processing_time) << " => "
<< x.max_throughput << "/" << x.items_per_batch << std::endl;
stats.reset();
// 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`.
......@@ -158,15 +156,20 @@ void inbound_path::emit_ack_batch(local_actor* self, int32_t queued_items,
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;
CAF_ASSERT(assigned_credit >= 0);
}
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(),
......
......@@ -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:
......
......@@ -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 {
......@@ -288,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>()) {
......@@ -311,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>
......
......@@ -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
......@@ -351,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;
}
......
......@@ -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