Commit 8ff35a7e authored by Dominik Charousset's avatar Dominik Charousset

Add more logging output

parent a319a0c4
...@@ -71,9 +71,11 @@ public: ...@@ -71,9 +71,11 @@ public:
} }
bool generate_messages() override { bool generate_messages() override {
CAF_LOG_TRACE("");
if (at_end_) if (at_end_)
return false; return false;
auto hint = out_.capacity(); auto hint = out_.capacity();
CAF_LOG_DEBUG(CAF_ARG(hint));
if (hint == 0) if (hint == 0)
return false; return false;
downstream<output_type> ds{out_.buf()}; downstream<output_type> ds{out_.buf()};
......
...@@ -354,6 +354,7 @@ scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) { ...@@ -354,6 +354,7 @@ scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) {
mailbox_element_ptr ptr; mailbox_element_ptr ptr;
// Timeout for calling `advance_streams`. // Timeout for calling `advance_streams`.
while (handled_msgs < max_throughput) { while (handled_msgs < max_throughput) {
CAF_LOG_DEBUG("start new DRR round");
// TODO: maybe replace '3' with configurable / adaptive value? // TODO: maybe replace '3' with configurable / adaptive value?
// Dispatch on the different message categories in our mailbox. // Dispatch on the different message categories in our mailbox.
if (!mailbox_.new_round(3, f).consumed_items) { if (!mailbox_.new_round(3, f).consumed_items) {
...@@ -368,6 +369,7 @@ scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) { ...@@ -368,6 +369,7 @@ scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) {
if (now >= tout) if (now >= tout)
tout = advance_streams(now); tout = advance_streams(now);
} }
CAF_LOG_DEBUG("max throughput reached");
reset_timeouts_if_needed(); reset_timeouts_if_needed();
if (mailbox().try_block()) if (mailbox().try_block())
return resumable::awaiting_message; return resumable::awaiting_message;
...@@ -426,6 +428,7 @@ sec scheduled_actor::build_pipeline(stream_manager_ptr mgr) { ...@@ -426,6 +428,7 @@ sec scheduled_actor::build_pipeline(stream_manager_ptr mgr) {
// -- timeout management ------------------------------------------------------- // -- timeout management -------------------------------------------------------
uint64_t scheduled_actor::set_receive_timeout(actor_clock::time_point x) { uint64_t scheduled_actor::set_receive_timeout(actor_clock::time_point x) {
CAF_LOG_TRACE(x);
setf(has_timeout_flag); setf(has_timeout_flag);
return set_timeout(receive_atom::value, x); return set_timeout(receive_atom::value, x);
} }
...@@ -461,6 +464,7 @@ bool scheduled_actor::is_active_receive_timeout(uint64_t tid) const { ...@@ -461,6 +464,7 @@ bool scheduled_actor::is_active_receive_timeout(uint64_t tid) const {
} }
uint64_t scheduled_actor::set_stream_timeout(actor_clock::time_point x) { uint64_t scheduled_actor::set_stream_timeout(actor_clock::time_point x) {
CAF_LOG_TRACE(x);
if (x == actor_clock::time_point::max()) if (x == actor_clock::time_point::max())
return 0; return 0;
return set_timeout(stream_atom::value, x); return set_timeout(stream_atom::value, x);
...@@ -484,6 +488,7 @@ void scheduled_actor::add_multiplexed_response_handler(message_id response_id, ...@@ -484,6 +488,7 @@ void scheduled_actor::add_multiplexed_response_handler(message_id response_id,
scheduled_actor::message_category scheduled_actor::message_category
scheduled_actor::categorize(mailbox_element& x) { scheduled_actor::categorize(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x));
auto& content = x.content(); auto& content = x.content();
switch (content.type_token()) { switch (content.type_token()) {
case make_type_token<atom_value, atom_value, std::string>(): case make_type_token<atom_value, atom_value, std::string>():
...@@ -511,11 +516,12 @@ scheduled_actor::categorize(mailbox_element& x) { ...@@ -511,11 +516,12 @@ scheduled_actor::categorize(mailbox_element& x) {
auto& tm = content.get_as<timeout_msg>(0); auto& tm = content.get_as<timeout_msg>(0);
auto tid = tm.timeout_id; auto tid = tm.timeout_id;
if (tm.type == receive_atom::value) { if (tm.type == receive_atom::value) {
CAF_LOG_DEBUG("handle timeout message"); CAF_LOG_DEBUG("handle ordinary timeout message");
if (is_active_receive_timeout(tid) && !bhvr_stack_.empty()) if (is_active_receive_timeout(tid) && !bhvr_stack_.empty())
bhvr_stack_.back().handle_timeout(); bhvr_stack_.back().handle_timeout();
} else { } else {
CAF_ASSERT(tm.type == atom("stream")); CAF_ASSERT(tm.type == atom("stream"));
CAF_LOG_DEBUG("handle stream timeout message");
set_stream_timeout(advance_streams(clock().now())); set_stream_timeout(advance_streams(clock().now()));
} }
return message_category::internal; return message_category::internal;
......
...@@ -58,6 +58,7 @@ void stream_manager::handle(inbound_path*, downstream_msg::forced_close& x) { ...@@ -58,6 +58,7 @@ void stream_manager::handle(inbound_path*, downstream_msg::forced_close& x) {
} }
void stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) { void stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x));
auto path = out().add_path(slots.invert(), x.rebind_to); auto path = out().add_path(slots.invert(), x.rebind_to);
path->open_credit = x.initial_demand; path->open_credit = x.initial_demand;
path->desired_batch_size = x.desired_batch_size; path->desired_batch_size = x.desired_batch_size;
...@@ -67,6 +68,7 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) { ...@@ -67,6 +68,7 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) {
} }
void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) { void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x));
auto path = out().path(slots.invert()); auto path = out().path(slots.invert());
if (path != nullptr) { if (path != nullptr) {
path->open_credit += x.new_capacity; path->open_credit += x.new_capacity;
......
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