Commit ac2f6f23 authored by Dominik Charousset's avatar Dominik Charousset

convert invoke_message_result to an enum class

parent 8607ad2f
......@@ -7,6 +7,7 @@ file(GLOB_RECURSE LIBCAF_CORE_HDRS "caf/*.hpp")
enum_to_string("caf/exit_reason.hpp" "exit_reason_strings.cpp")
enum_to_string("caf/intrusive/inbox_result.hpp" "inbox_result_strings.cpp")
enum_to_string("caf/intrusive/task_result.hpp" "task_result_strings.cpp")
enum_to_string("caf/invoke_message_result.hpp" "invoke_msg_result_strings.cpp")
enum_to_string("caf/message_priority.hpp" "message_priority_strings.cpp")
enum_to_string("caf/pec.hpp" "pec_strings.cpp")
enum_to_string("caf/sec.hpp" "sec_strings.cpp")
......@@ -16,6 +17,7 @@ enum_to_string("caf/stream_priority.hpp" "stream_priority_strings.cpp")
set(LIBCAF_CORE_SRCS
"${CMAKE_CURRENT_BINARY_DIR}/exit_reason_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/inbox_result_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/invoke_msg_result_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/message_priority_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/pec_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/sec_strings.cpp"
......
......@@ -177,6 +177,7 @@ enum class atom_value : uint64_t;
enum class byte : uint8_t;
enum class sec : uint8_t;
enum class stream_priority;
enum class invoke_message_result;
// -- aliases ------------------------------------------------------------------
......
......@@ -22,17 +22,21 @@
namespace caf {
enum invoke_message_result {
im_success,
im_skipped,
im_dropped
/// Stores the result of a message invocation.
enum class invoke_message_result {
/// Indicates that the actor consumed the message.
consumed,
/// Indicates that the actor left the message in the mailbox.
skipped,
/// Indicates that the actor discarded the message based on meta data. For
/// example, timeout messages for already received requests usually get
/// dropped without calling any user-defined code.
dropped,
};
inline std::string to_string(invoke_message_result x) {
return x == im_success ? "im_success"
: (x == im_skipped ? "im_skipped" : "im_dropped" );
}
/// @relates invoke_message_result
std::string to_string(invoke_message_result);
} // namespace caf
......@@ -23,7 +23,7 @@
namespace caf {
raw_event_based_actor::raw_event_based_actor(actor_config& cfg)
: event_based_actor(cfg) {
: event_based_actor(cfg) {
// nop
}
......@@ -36,30 +36,30 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
auto& pr = awaited_responses_.front();
// skip all messages until we receive the currently awaited response
if (x.mid != pr.first)
return im_skipped;
return invoke_message_result::skipped;
if (!pr.second(x.content())) {
// try again with error if first attempt failed
auto msg = make_message(make_error(sec::unexpected_response,
x.move_content_to_message()));
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
pr.second(msg);
}
awaited_responses_.pop_front();
return im_success;
return invoke_message_result::consumed;
}
// handle multiplexed responses
if (x.mid.is_response()) {
auto mrh = multiplexed_responses_.find(x.mid);
// neither awaited nor multiplexed, probably an expired timeout
if (mrh == multiplexed_responses_.end())
return im_dropped;
return invoke_message_result::dropped;
if (!mrh->second(x.content())) {
// try again with error if first attempt failed
auto msg = make_message(make_error(sec::unexpected_response,
x.move_content_to_message()));
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
mrh->second(msg);
}
multiplexed_responses_.erase(mrh);
return im_success;
return invoke_message_result::consumed;
}
auto& content = x.content();
// handle timeout messages
......@@ -70,12 +70,12 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
if (is_active_receive_timeout(tid)) {
CAF_LOG_DEBUG("handle timeout message");
if (bhvr_stack_.empty())
return im_dropped;
return invoke_message_result::dropped;
bhvr_stack_.back().handle_timeout();
return im_success;
return invoke_message_result::consumed;
}
CAF_LOG_DEBUG("dropped expired timeout message");
return im_dropped;
return invoke_message_result::dropped;
}
// handle everything else as ordinary message
detail::default_invoke_result_visitor<event_based_actor> visitor{this};
......@@ -103,7 +103,8 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
};
if (bhvr_stack_.empty()) {
call_default_handler();
return !skipped ? im_success : im_skipped;
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
}
auto& bhvr = bhvr_stack_.back();
switch (bhvr(visitor, x.content())) {
......@@ -115,7 +116,8 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
case match_case::no_match:
call_default_handler();
}
return !skipped ? im_success : im_skipped;
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
// should be unreachable
CAF_CRITICAL("invalid message type");
}
......
......@@ -47,8 +47,7 @@ result<message> reflect_and_quit(scheduled_actor* ptr, message_view& x) {
result<message> print_and_drop(scheduled_actor* ptr, message_view& x) {
CAF_LOG_WARNING("unexpected message" << CAF_ARG(x.content()));
aout(ptr) << "*** unexpected message [id: " << ptr->id()
<< ", name: " << ptr->name() << "]: "
<< x.content().stringify()
<< ", name: " << ptr->name() << "]: " << x.content().stringify()
<< std::endl;
return sec::unexpected_message;
}
......@@ -90,7 +89,7 @@ void scheduled_actor::default_exit_handler(scheduled_actor* ptr, exit_msg& x) {
default_error_handler(ptr, x.reason);
}
# ifndef CAF_NO_EXCEPTIONS
#ifndef CAF_NO_EXCEPTIONS
error scheduled_actor::default_exception_handler(pointer ptr,
std::exception_ptr& x) {
CAF_ASSERT(x != nullptr);
......@@ -98,8 +97,8 @@ error scheduled_actor::default_exception_handler(pointer ptr,
std::rethrow_exception(x);
} catch (const std::exception& e) {
aout(ptr) << "*** unhandled exception: [id: " << ptr->id()
<< ", name: " << ptr->name() << ", exception typeid: "
<< typeid(e).name() << "]: " << e.what()
<< ", name: " << ptr->name()
<< ", exception typeid: " << typeid(e).name() << "]: " << e.what()
<< std::endl;
} catch (...) {
aout(ptr) << "*** unhandled exception: [id: " << ptr->id()
......@@ -108,23 +107,24 @@ error scheduled_actor::default_exception_handler(pointer ptr,
}
return sec::runtime_error;
}
# endif // CAF_NO_EXCEPTIONS
#endif // CAF_NO_EXCEPTIONS
// -- constructors and destructors ---------------------------------------------
scheduled_actor::scheduled_actor(actor_config& cfg)
: super(cfg),
mailbox_(unit, unit, unit, unit, unit),
timeout_id_(0),
default_handler_(print_and_drop),
error_handler_(default_error_handler),
down_handler_(default_down_handler),
exit_handler_(default_exit_handler),
private_thread_(nullptr)
# ifndef CAF_NO_EXCEPTIONS
, exception_handler_(default_exception_handler)
# endif // CAF_NO_EXCEPTIONS
{
: super(cfg),
mailbox_(unit, unit, unit, unit, unit),
timeout_id_(0),
default_handler_(print_and_drop),
error_handler_(default_error_handler),
down_handler_(default_down_handler),
exit_handler_(default_exit_handler),
private_thread_(nullptr)
#ifndef CAF_NO_EXCEPTIONS
,
exception_handler_(default_exception_handler)
#endif // CAF_NO_EXCEPTIONS
{
auto& sys_cfg = home_system().config();
auto interval = sys_cfg.stream_tick_duration();
CAF_ASSERT(interval.count() > 0);
......@@ -138,7 +138,7 @@ scheduled_actor::scheduled_actor(actor_config& cfg)
credit_round_ticks_ = div(sys_cfg.stream_credit_round_interval, interval);
CAF_ASSERT(credit_round_ticks_ > 0);
CAF_LOG_DEBUG(CAF_ARG(interval) << CAF_ARG(max_batch_delay_ticks_)
<< CAF_ARG(credit_round_ticks_));
<< CAF_ARG(credit_round_ticks_));
}
scheduled_actor::~scheduled_actor() {
......@@ -279,8 +279,9 @@ struct upstream_msg_visitor {
} // namespace
intrusive::task_result scheduled_actor::mailbox_visitor::
operator()(size_t, upstream_queue&, mailbox_element& x) {
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>());
self->current_mailbox_element(&x);
CAF_LOG_RECEIVE_EVENT((&x));
......@@ -310,9 +311,9 @@ struct downstream_msg_visitor {
inptr->handle(x);
// The sender slot can be 0. This is the case for forced_close or
// forced_drop messages from stream aborters.
CAF_ASSERT(inptr->slots == dm.slots
|| (dm.slots.sender == 0
&& dm.slots.receiver == inptr->slots.receiver));
CAF_ASSERT(
inptr->slots == dm.slots
|| (dm.slots.sender == 0 && dm.slots.receiver == inptr->slots.receiver));
// TODO: replace with `if constexpr` when switching to C++17
if (std::is_same<T, downstream_msg::close>::value
|| std::is_same<T, downstream_msg::forced_close>::value) {
......@@ -324,8 +325,7 @@ struct downstream_msg_visitor {
mgr->stop();
}
return intrusive::task_result::stop;
}
else if (mgr->done()) {
} else if (mgr->done()) {
CAF_LOG_DEBUG("path is done receiving and closes its manager");
selfptr->erase_stream_manager(mgr);
mgr->stop();
......@@ -337,10 +337,9 @@ struct downstream_msg_visitor {
} // namespace
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) {
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_LOG_TRACE(CAF_ARG(x) << CAF_ARG(handled_msgs));
self->current_mailbox_element(&x);
CAF_LOG_RECEIVE_EVENT((&x));
......@@ -359,9 +358,8 @@ scheduled_actor::mailbox_visitor::operator()(mailbox_element& x) {
case activation_result::terminated:
return intrusive::task_result::stop;
case activation_result::success:
return ++handled_msgs < max_throughput
? intrusive::task_result::resume
: intrusive::task_result::stop_all;
return ++handled_msgs < max_throughput ? intrusive::task_result::resume
: intrusive::task_result::stop_all;
case activation_result::skipped:
return intrusive::task_result::skip;
default:
......@@ -369,8 +367,8 @@ scheduled_actor::mailbox_visitor::operator()(mailbox_element& x) {
}
}
resumable::resume_result
scheduled_actor::resume(execution_unit* ctx, size_t max_throughput) {
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))
......@@ -515,9 +513,7 @@ uint64_t scheduled_actor::set_stream_timeout(actor_clock::time_point x) {
mgrs.emplace_back(kvp.second);
std::sort(mgrs.begin(), mgrs.end());
auto e = std::unique(mgrs.begin(), mgrs.end());
auto idle = [=](const stream_manager_ptr& y) {
return y->idle();
};
auto idle = [=](const stream_manager_ptr& y) { return y->idle(); };
if (std::all_of(mgrs.begin(), e, idle)) {
CAF_LOG_DEBUG("suppress stream timeout");
return 0;
......@@ -616,9 +612,9 @@ scheduled_actor::categorize(mailbox_element& x) {
return message_category::internal;
}
case make_type_token<open_stream_msg>(): {
return handle_open_stream_msg(x) != im_skipped
? message_category::internal
: message_category::skipped;
return handle_open_stream_msg(x) != invoke_message_result::skipped
? message_category::internal
: message_category::skipped;
}
default:
return message_category::ordinary;
......@@ -635,25 +631,23 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
auto ordinary_invoke = [](ptr_t, behavior& f, mailbox_element& in) -> bool {
return f(in.content()) != none;
};
auto select_invoke_fun = [&]() -> fun_t {
return ordinary_invoke;
};
auto select_invoke_fun = [&]() -> fun_t { return ordinary_invoke; };
// Short-circuit awaited responses.
if (!awaited_responses_.empty()) {
auto invoke = select_invoke_fun();
auto& pr = awaited_responses_.front();
// skip all messages until we receive the currently awaited response
if (x.mid != pr.first)
return im_skipped;
return invoke_message_result::skipped;
auto f = std::move(pr.second);
awaited_responses_.pop_front();
if (!invoke(this, f, x)) {
// try again with error if first attempt failed
auto msg = make_message(make_error(sec::unexpected_response,
x.move_content_to_message()));
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
f(msg);
}
return im_success;
return invoke_message_result::consumed;
}
// Handle multiplexed responses.
if (x.mid.is_response()) {
......@@ -661,24 +655,24 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
auto mrh = multiplexed_responses_.find(x.mid);
// neither awaited nor multiplexed, probably an expired timeout
if (mrh == multiplexed_responses_.end())
return im_dropped;
return invoke_message_result::dropped;
auto bhvr = std::move(mrh->second);
multiplexed_responses_.erase(mrh);
if (!invoke(this, bhvr, x)) {
// try again with error if first attempt failed
auto msg = make_message(make_error(sec::unexpected_response,
x.move_content_to_message()));
auto msg = make_message(
make_error(sec::unexpected_response, x.move_content_to_message()));
bhvr(msg);
}
return im_success;
return invoke_message_result::consumed;
}
// Dispatch on the content of x.
switch (categorize(x)) {
case message_category::skipped:
return im_skipped;
return invoke_message_result::skipped;
case message_category::internal:
CAF_LOG_DEBUG("handled system message");
return im_success;
return invoke_message_result::consumed;
case message_category::ordinary: {
detail::default_invoke_result_visitor<scheduled_actor> visitor{this};
bool skipped = false;
......@@ -705,7 +699,8 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
};
if (bhvr_stack_.empty()) {
call_default_handler();
return !skipped ? im_success : im_skipped;
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
}
auto& bhvr = bhvr_stack_.back();
switch (bhvr(visitor, x.content())) {
......@@ -717,7 +712,8 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
case match_case::no_match:
call_default_handler();
}
return !skipped ? im_success : im_skipped;
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
}
}
// Unreachable.
......@@ -729,7 +725,7 @@ void scheduled_actor::consume(mailbox_element_ptr x) {
switch (consume(*x)) {
default:
break;
case im_skipped:
case invoke_message_result::skipped:
push_to_cache(std::move(x));
}
}
......@@ -743,9 +739,9 @@ bool scheduled_actor::activate(execution_unit* ctx) {
CAF_LOG_ERROR("activate called on a terminated actor");
return false;
}
# ifndef CAF_NO_EXCEPTIONS
#ifndef CAF_NO_EXCEPTIONS
try {
# endif // CAF_NO_EXCEPTIONS
#endif // CAF_NO_EXCEPTIONS
if (!getf(is_initialized_flag)) {
initialize();
if (finalize()) {
......@@ -754,21 +750,20 @@ bool scheduled_actor::activate(execution_unit* ctx) {
}
CAF_LOG_DEBUG("initialized actor:" << CAF_ARG(name()));
}
# ifndef CAF_NO_EXCEPTIONS
}
catch (...) {
#ifndef CAF_NO_EXCEPTIONS
} catch (...) {
CAF_LOG_ERROR("actor died during initialization");
auto eptr = std::current_exception();
quit(call_handler(exception_handler_, this, eptr));
finalize();
return false;
}
# endif // CAF_NO_EXCEPTIONS
#endif // CAF_NO_EXCEPTIONS
return true;
}
auto scheduled_actor::activate(execution_unit* ctx, mailbox_element& x)
-> activation_result {
-> activation_result {
CAF_LOG_TRACE(CAF_ARG(x));
if (!activate(ctx))
return activation_result::terminated;
......@@ -780,38 +775,36 @@ auto scheduled_actor::activate(execution_unit* ctx, mailbox_element& x)
auto scheduled_actor::reactivate(mailbox_element& x) -> activation_result {
CAF_LOG_TRACE(CAF_ARG(x));
# ifndef CAF_NO_EXCEPTIONS
#ifndef CAF_NO_EXCEPTIONS
try {
# endif // CAF_NO_EXCEPTIONS
#endif // CAF_NO_EXCEPTIONS
switch (consume(x)) {
case im_dropped:
case invoke_message_result::dropped:
return activation_result::dropped;
case im_success:
case invoke_message_result::consumed:
bhvr_stack_.cleanup();
if (finalize()) {
CAF_LOG_DEBUG("actor finalized");
return activation_result::terminated;
}
return activation_result::success;
case im_skipped:
case invoke_message_result::skipped:
return activation_result::skipped;
}
# ifndef CAF_NO_EXCEPTIONS
}
catch (std::exception& e) {
#ifndef CAF_NO_EXCEPTIONS
} catch (std::exception& e) {
CAF_LOG_INFO("actor died because of an exception, what: " << e.what());
static_cast<void>(e); // keep compiler happy when not logging
auto eptr = std::current_exception();
quit(call_handler(exception_handler_, this, eptr));
}
catch (...) {
} catch (...) {
CAF_LOG_INFO("actor died because of an unknown exception");
auto eptr = std::current_exception();
quit(call_handler(exception_handler_, this, eptr));
}
finalize();
return activation_result::terminated;
# endif // CAF_NO_EXCEPTIONS
#endif // CAF_NO_EXCEPTIONS
}
// -- behavior management ----------------------------------------------------
......@@ -1103,14 +1096,13 @@ scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
auto sres = call_handler(default_handler_, this, x);
switch (sres.flag) {
default:
CAF_LOG_DEBUG("default handler was called for open_stream_msg:"
<< osm.msg);
CAF_LOG_DEBUG(
"default handler was called for open_stream_msg:" << osm.msg);
fail(sec::stream_init_failed, "dropped open_stream_msg (no match)");
return im_dropped;
return invoke_message_result::dropped;
case rt_skip:
CAF_LOG_DEBUG("default handler skipped open_stream_msg:"
<< osm.msg);
return im_skipped;
CAF_LOG_DEBUG("default handler skipped open_stream_msg:" << osm.msg);
return invoke_message_result::skipped;
}
};
// Invoke behavior and dispatch on the result.
......@@ -1123,11 +1115,11 @@ 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: {
return im_success;
return invoke_message_result::consumed;
}
default:
CAF_LOG_DEBUG("behavior skipped open_stream_msg:" << osm.msg);
return im_skipped; // nop
return invoke_message_result::skipped; // nop
}
}
......
......@@ -60,13 +60,13 @@ void manager::detach(execution_unit*, bool invoke_disconnect_message) {
auto mptr = make_mailbox_element(nullptr, make_message_id(), {},
detach_message());
switch (raw_ptr->consume(*mptr)) {
case im_success:
case invoke_message_result::consumed:
raw_ptr->finalize();
break;
case im_skipped:
case invoke_message_result::skipped:
raw_ptr->push_to_cache(std::move(mptr));
break;
case im_dropped:
case invoke_message_result::dropped:
CAF_LOG_INFO("broker dropped disconnect message");
break;
}
......
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