Commit be8c3f34 authored by Dominik Charousset's avatar Dominik Charousset

Restrict skipping to the default handler

parent 13ac1a41
...@@ -3,17 +3,14 @@ ...@@ -3,17 +3,14 @@
using namespace caf; using namespace caf;
behavior server(event_based_actor* self) { behavior server(event_based_actor* self) {
self->set_default_handler(skip);
return { return {
[=](idle_atom, const actor& worker) { [=](idle_atom, const actor& worker) {
self->become( self->become(keep_behavior, [=](ping_atom atm) {
keep_behavior,
[=](ping_atom atm) {
self->delegate(worker, atm); self->delegate(worker, atm);
self->unbecome(); self->unbecome();
});
}, },
[=](idle_atom, const actor&) { return skip(); });
},
[=](ping_atom) { return skip(); },
}; };
} }
......
...@@ -28,7 +28,11 @@ namespace caf { ...@@ -28,7 +28,11 @@ namespace caf {
template <class F> template <class F>
struct catch_all { struct catch_all {
using fun_type = std::function<result<message>(message&)>; using fun_type = std::function<skippable_result(message&)>;
static_assert(std::is_convertible<F, fun_type>::value,
"catch-all handler must have signature "
"skippable_result (message&)");
F handler; F handler;
...@@ -41,10 +45,6 @@ struct catch_all { ...@@ -41,10 +45,6 @@ struct catch_all {
// nop // nop
} }
static_assert(std::is_convertible<F, fun_type>::value,
"catch-all handler must have signature "
"result<message> (message&)");
fun_type lift() const { fun_type lift() const {
return handler; return handler;
} }
......
...@@ -131,7 +131,6 @@ public: ...@@ -131,7 +131,6 @@ public:
template <size_t... Is> template <size_t... Is>
match_result invoke_impl(detail::invoke_result_visitor& f, message& msg, match_result invoke_impl(detail::invoke_result_visitor& f, message& msg,
std::index_sequence<Is...>) { std::index_sequence<Is...>) {
auto result = match_result::no_match;
auto dispatch = [&](auto& fun) { auto dispatch = [&](auto& fun) {
using fun_type = std::decay_t<decltype(fun)>; using fun_type = std::decay_t<decltype(fun)>;
using trait = get_callable_trait_t<fun_type>; using trait = get_callable_trait_t<fun_type>;
...@@ -141,18 +140,17 @@ public: ...@@ -141,18 +140,17 @@ public:
using fun_result = decltype(detail::apply_args(fun, xs)); using fun_result = decltype(detail::apply_args(fun, xs));
if constexpr (std::is_same<void, fun_result>::value) { if constexpr (std::is_same<void, fun_result>::value) {
detail::apply_args(fun, xs); detail::apply_args(fun, xs);
result = f.visit(unit) ? match_result::match : match_result::skip; f(unit);
} else { } else {
auto invoke_res = detail::apply_args(fun, xs); auto invoke_res = detail::apply_args(fun, xs);
result = f.visit(invoke_res) ? match_result::match f(invoke_res);
: match_result::skip;
} }
return true; return true;
} }
return false; return false;
}; };
static_cast<void>((dispatch(std::get<Is>(cases_)) || ...)); bool dispatched = (dispatch(std::get<Is>(cases_)) || ...);
return result; return dispatched ? match_result::match : match_result::no_match;
} }
void handle_timeout() override { void handle_timeout() override {
......
...@@ -35,7 +35,7 @@ public: ...@@ -35,7 +35,7 @@ public:
virtual ~blocking_behavior(); virtual ~blocking_behavior();
virtual result<message> fallback(message&); virtual skippable_result fallback(message&);
virtual timespan timeout(); virtual timespan timeout();
...@@ -54,7 +54,7 @@ public: ...@@ -54,7 +54,7 @@ public:
blocking_behavior_v2(blocking_behavior_v2&&) = default; blocking_behavior_v2(blocking_behavior_v2&&) = default;
result<message> fallback(message& x) override { skippable_result fallback(message& x) override {
return f.handler(x); return f.handler(x);
} }
}; };
...@@ -93,7 +93,7 @@ public: ...@@ -93,7 +93,7 @@ public:
blocking_behavior_v4(blocking_behavior_v4&&) = default; blocking_behavior_v4(blocking_behavior_v4&&) = default;
result<message> fallback(message& x) override { skippable_result fallback(message& x) override {
return f1.handler(x); return f1.handler(x);
} }
......
...@@ -27,7 +27,9 @@ namespace caf::detail { ...@@ -27,7 +27,9 @@ namespace caf::detail {
template <class Self> template <class Self>
class default_invoke_result_visitor : public invoke_result_visitor { class default_invoke_result_visitor : public invoke_result_visitor {
public: public:
inline default_invoke_result_visitor(Self* ptr) : self_(ptr) { using super = invoke_result_visitor;
default_invoke_result_visitor(Self* ptr) : self_(ptr) {
// nop // nop
} }
...@@ -35,6 +37,8 @@ public: ...@@ -35,6 +37,8 @@ public:
// nop // nop
} }
using super::operator();
void operator()(error& x) override { void operator()(error& x) override {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
delegate(x); delegate(x);
......
...@@ -56,7 +56,7 @@ public: ...@@ -56,7 +56,7 @@ public:
/// Called if the message handler returned any "ordinary" value. /// Called if the message handler returned any "ordinary" value.
virtual void operator()(message&) = 0; virtual void operator()(message&) = 0;
// -- on-the-fly type conversions -------------------------------------------- // -- extraction and conversions ---------------------------------------------
/// Wraps arbitrary values into a `message` and calls the visitor recursively. /// Wraps arbitrary values into a `message` and calls the visitor recursively.
template <class... Ts> template <class... Ts>
...@@ -79,6 +79,12 @@ public: ...@@ -79,6 +79,12 @@ public:
(*this)(const_cast<const unit_t&>(x)); (*this)(const_cast<const unit_t&>(x));
} }
/// Dispatches on the runtime-type of `x`.
template <class... Ts>
void operator()(result<Ts...>& res) {
caf::visit([this](auto& x) { (*this)(x); }, res);
}
// -- special-purpose handlers that don't produce results -------------------- // -- special-purpose handlers that don't produce results --------------------
void operator()(response_promise&) { void operator()(response_promise&) {
...@@ -119,46 +125,6 @@ public: ...@@ -119,46 +125,6 @@ public:
void operator()(make_stage_result<In, DownstreamManager, Ts...>&) { void operator()(make_stage_result<In, DownstreamManager, Ts...>&) {
// nop // nop
} }
// -- 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`.
bool visit(skip_t&) {
return false;
}
/// Returns `false`.
bool visit(const skip_t&) {
return false;
}
/// Returns `false` if `x != none`, otherwise calls the void handler and
/// returns `true`..
bool visit(optional<skip_t>& x) {
return static_cast<bool>(x);
}
/// Dispatches on the runtime-type of `x`.
template <class... Ts>
bool visit(result<Ts...>& res) {
auto f = detail::make_overload(
[this](auto& x) {
(*this)(x);
return true;
},
[this](delegated<Ts...>&) {
return true;
},
[this](skip_t&) { return false; });
return caf::visit(f, res);
}
}; };
} // namespace caf::detail } // namespace caf::detail
...@@ -87,7 +87,7 @@ struct CAF_CORE_EXPORT function_view_storage_catch_all { ...@@ -87,7 +87,7 @@ struct CAF_CORE_EXPORT function_view_storage_catch_all {
// nop // nop
} }
result<message> operator()(message& msg) { skippable_result operator()(message& msg) {
*storage_ = std::move(msg); *storage_ = std::move(msg);
return message{}; return message{};
} }
......
...@@ -135,6 +135,7 @@ class resumable; ...@@ -135,6 +135,7 @@ class resumable;
class scheduled_actor; class scheduled_actor;
class scoped_actor; class scoped_actor;
class serializer; class serializer;
class skip_t;
class stream_manager; class stream_manager;
class string_view; class string_view;
class tracing_data; class tracing_data;
...@@ -193,6 +194,7 @@ using ip_address = ipv6_address; ...@@ -193,6 +194,7 @@ using ip_address = ipv6_address;
using ip_endpoint = ipv6_endpoint; using ip_endpoint = ipv6_endpoint;
using ip_subnet = ipv6_subnet; using ip_subnet = ipv6_subnet;
using settings = dictionary<config_value>; using settings = dictionary<config_value>;
using skippable_result = variant<delegated<message>, message, error, skip_t>;
using stream_slot = uint16_t; using stream_slot = uint16_t;
using type_id_t = uint16_t; using type_id_t = uint16_t;
......
...@@ -26,9 +26,8 @@ namespace caf { ...@@ -26,9 +26,8 @@ namespace caf {
/// Denotes the invoke result of a ::behavior or ::message_handler. /// Denotes the invoke result of a ::behavior or ::message_handler.
enum class match_result { enum class match_result {
no_match,
match, match,
skip, no_match,
}; };
/// @relates match_result /// @relates match_result
......
...@@ -50,7 +50,7 @@ class result_base { ...@@ -50,7 +50,7 @@ class result_base {
public: public:
static_assert(sizeof...(Ts) > 0); static_assert(sizeof...(Ts) > 0);
using types = detail::type_list<delegated<Ts...>, message, error, skip_t>; using types = detail::type_list<delegated<Ts...>, message, error>;
result_base() = default; result_base() = default;
...@@ -71,10 +71,6 @@ public: ...@@ -71,10 +71,6 @@ public:
// nop // nop
} }
result_base(skip_t) : content_(skip) {
// nop
}
result_base(delegated<Ts...> x) : content_(x) { result_base(delegated<Ts...> x) : content_(x) {
// nop // nop
} }
...@@ -109,7 +105,7 @@ protected: ...@@ -109,7 +105,7 @@ protected:
// nop // nop
} }
variant<delegated<Ts...>, message, error, skip_t> content_; variant<delegated<Ts...>, message, error> content_;
}; };
// -- result<Ts...> and its specializations ------------------------------------ // -- result<Ts...> and its specializations ------------------------------------
...@@ -245,6 +241,11 @@ auto make_result(Ts&&... xs) { ...@@ -245,6 +241,11 @@ auto make_result(Ts&&... xs) {
return result<std::decay_t<Ts>...>(std::forward<Ts>(xs)...); return result<std::decay_t<Ts>...>(std::forward<Ts>(xs)...);
} }
// -- special type alias for a skippable result<message> -----------------------
/// Similar to `result<message>`, but also allows to *skip* a message.
using skippable_result = variant<delegated<message>, message, error, skip_t>;
// -- sum type access to result<Ts...> ----------------------------------------- // -- sum type access to result<Ts...> -----------------------------------------
template <class... Ts> template <class... Ts>
......
...@@ -84,21 +84,21 @@ namespace caf { ...@@ -84,21 +84,21 @@ namespace caf {
/// @relates scheduled_actor /// @relates scheduled_actor
/// Default handler function that sends the message back to the sender. /// Default handler function that sends the message back to the sender.
CAF_CORE_EXPORT result<message> reflect(scheduled_actor*, message&); CAF_CORE_EXPORT skippable_result reflect(scheduled_actor*, message&);
/// @relates scheduled_actor /// @relates scheduled_actor
/// Default handler function that sends /// Default handler function that sends
/// the message back to the sender and then quits. /// the message back to the sender and then quits.
CAF_CORE_EXPORT result<message> reflect_and_quit(scheduled_actor*, message&); CAF_CORE_EXPORT skippable_result reflect_and_quit(scheduled_actor*, message&);
/// @relates scheduled_actor /// @relates scheduled_actor
/// Default handler function that prints messages /// Default handler function that prints messages
/// message via `aout` and drops them afterwards. /// message via `aout` and drops them afterwards.
CAF_CORE_EXPORT result<message> print_and_drop(scheduled_actor*, message&); CAF_CORE_EXPORT skippable_result print_and_drop(scheduled_actor*, message&);
/// @relates scheduled_actor /// @relates scheduled_actor
/// Default handler function that simply drops messages. /// Default handler function that simply drops messages.
CAF_CORE_EXPORT result<message> drop(scheduled_actor*, message&); CAF_CORE_EXPORT skippable_result drop(scheduled_actor*, message&);
/// A cooperatively scheduled, event-based actor implementation. /// A cooperatively scheduled, event-based actor implementation.
class CAF_CORE_EXPORT scheduled_actor : public local_actor, class CAF_CORE_EXPORT scheduled_actor : public local_actor,
...@@ -189,7 +189,7 @@ public: ...@@ -189,7 +189,7 @@ public:
using pointer = scheduled_actor*; using pointer = scheduled_actor*;
/// Function object for handling unmatched messages. /// Function object for handling unmatched messages.
using default_handler = std::function<result<message>(pointer, message&)>; using default_handler = std::function<skippable_result(pointer, message&)>;
/// Function object for handling error messages. /// Function object for handling error messages.
using error_handler = std::function<void(pointer, error&)>; using error_handler = std::function<void(pointer, error&)>;
...@@ -215,8 +215,8 @@ public: ...@@ -215,8 +215,8 @@ public:
size_t max_throughput; size_t max_throughput;
/// Consumes upstream messages. /// Consumes upstream messages.
intrusive::task_result intrusive::task_result operator()(size_t, upstream_queue&,
operator()(size_t, upstream_queue&, mailbox_element&); mailbox_element&);
/// Consumes downstream messages. /// Consumes downstream messages.
intrusive::task_result intrusive::task_result
...@@ -730,8 +730,8 @@ public: ...@@ -730,8 +730,8 @@ public:
/// Closes all incoming stream traffic for a manager. Emits a drop message on /// Closes all incoming stream traffic for a manager. Emits a drop message on
/// each path if `reason == none` and a `forced_drop` message on each path /// each path if `reason == none` and a `forced_drop` message on each path
/// otherwise. /// otherwise.
virtual void virtual void erase_inbound_paths_later(const stream_manager* mgr,
erase_inbound_paths_later(const stream_manager* mgr, error reason); error reason);
// -- handling of stream messages -------------------------------------------- // -- handling of stream messages --------------------------------------------
......
...@@ -31,20 +31,13 @@ namespace caf { ...@@ -31,20 +31,13 @@ namespace caf {
/// skipping to the runtime. /// skipping to the runtime.
class CAF_CORE_EXPORT skip_t { class CAF_CORE_EXPORT skip_t {
public: public:
using fun = std::function<result<message>(scheduled_actor* self, message&)>; using fun = std::function<skippable_result(scheduled_actor* self, message&)>;
constexpr skip_t() { constexpr skip_t() {
// nop // nop
} }
constexpr skip_t operator()() const {
return *this;
}
operator fun() const; operator fun() const;
private:
static result<message> skip_fun_impl(scheduled_actor*, message&);
}; };
/// Tells the runtime system to skip a message when used as message /// Tells the runtime system to skip a message when used as message
......
...@@ -27,9 +27,7 @@ ...@@ -27,9 +27,7 @@
#include "caf/detail/typed_actor_util.hpp" #include "caf/detail/typed_actor_util.hpp"
namespace caf { namespace caf ::detail {
namespace detail {
// converts a list of replies_to<...>::with<...> elements to a list of // converts a list of replies_to<...>::with<...> elements to a list of
// lists containing the replies_to<...> half only // lists containing the replies_to<...> half only
...@@ -127,15 +125,30 @@ void static_check_typed_behavior_input() { ...@@ -127,15 +125,30 @@ void static_check_typed_behavior_input() {
"typed behavior (exact match needed)"); "typed behavior (exact match needed)");
} }
} // namespace detail } // namespace caf::detail
template <class... Sigs> namespace caf::mixin {
class typed_actor;
namespace mixin {
template <class, class, class> template <class, class, class>
class behavior_stack_based_impl; class behavior_stack_based_impl;
}
} // namespace caf::mixin
namespace caf {
/// Tag type for constructing a `typed_behavior` with an incomplete list of
/// message handlers, delegating to the default handler for all unmatched
/// inputs.
struct partial_behavior_init_t {};
constexpr partial_behavior_init_t partial_behavior_init
= partial_behavior_init_t{};
/// Empty struct tag for constructing from an untyped behavior.
struct unsafe_behavior_init_t {};
constexpr unsafe_behavior_init_t unsafe_behavior_init
= unsafe_behavior_init_t{};
template <class... Sigs> template <class... Sigs>
class typed_behavior { class typed_behavior {
...@@ -156,9 +169,6 @@ public: ...@@ -156,9 +169,6 @@ public:
/// Stores the template parameter pack in a type list. /// Stores the template parameter pack in a type list.
using signatures = detail::type_list<Sigs...>; using signatures = detail::type_list<Sigs...>;
/// Empty struct tag for constructing from an untyped behavior.
struct unsafe_init {};
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
typed_behavior(typed_behavior&&) = default; typed_behavior(typed_behavior&&) = default;
...@@ -182,11 +192,18 @@ public: ...@@ -182,11 +192,18 @@ public:
set(detail::make_behavior(std::move(x), std::move(xs)...)); set(detail::make_behavior(std::move(x), std::move(xs)...));
} }
typed_behavior(unsafe_init, behavior x) : bhvr_(std::move(x)) { template <class... Ts>
typed_behavior(partial_behavior_init_t, Ts... xs)
: typed_behavior(unsafe_behavior_init, behavior{std::move(xs)...}) {
// TODO: implement type checking.
}
typed_behavior(unsafe_behavior_init_t, behavior x) : bhvr_(std::move(x)) {
// nop // nop
} }
typed_behavior(unsafe_init, message_handler x) : bhvr_(std::move(x)) { typed_behavior(unsafe_behavior_init_t, message_handler x)
: bhvr_(std::move(x)) {
// nop // nop
} }
......
...@@ -196,24 +196,28 @@ blocking_actor::mailbox_visitor::operator()(mailbox_element& x) { ...@@ -196,24 +196,28 @@ blocking_actor::mailbox_visitor::operator()(mailbox_element& x) {
case match_result::no_match: { // Blocking actors can have fallback case match_result::no_match: { // Blocking actors can have fallback
// handlers for catch-all rules. // handlers for catch-all rules.
auto sres = bhvr.fallback(self->current_element_->payload); auto sres = bhvr.fallback(self->current_element_->payload);
if (!holds_alternative<skip_t>(sres)) { auto f = detail::make_overload(
visitor.visit(sres); [&](skip_t&) {
return check_if_done(); // Response handlers must get re-invoked with an error when
} // receiving an unexpected message.
}
// Response handlers must get re-invoked with an error when receiving an
// unexpected message.
if (mid.is_response()) { if (mid.is_response()) {
auto err = make_error(sec::unexpected_response, std::move(x.payload)); auto err = make_error(sec::unexpected_response,
mailbox_element tmp{std::move(x.sender), x.mid, std::move(x.stages), std::move(x.payload));
mailbox_element tmp{std::move(x.sender), x.mid,
std::move(x.stages),
make_message(std::move(err))}; make_message(std::move(err))};
self->current_element_ = &tmp; self->current_element_ = &tmp;
bhvr.nested(tmp.content()); bhvr.nested(tmp.content());
return check_if_done(); return check_if_done();
} }
CAF_ANNOTATE_FALLTHROUGH;
case match_result::skip:
return intrusive::task_result::skip; return intrusive::task_result::skip;
},
[&](auto& res) {
visitor(res);
return check_if_done();
});
return visit(f, sres);
}
} }
}; };
// Post-process the returned value from the function body. // Post-process the returned value from the function body.
......
...@@ -28,7 +28,7 @@ blocking_behavior::blocking_behavior(behavior& x) : nested(x) { ...@@ -28,7 +28,7 @@ blocking_behavior::blocking_behavior(behavior& x) : nested(x) {
// nop // nop
} }
result<message> blocking_behavior::fallback(message&) { skippable_result blocking_behavior::fallback(message&) {
return skip; return skip;
} }
......
...@@ -161,9 +161,9 @@ public: ...@@ -161,9 +161,9 @@ public:
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// instead of dropping "unexpected" messages, // instead of dropping "unexpected" messages,
// we simply forward them to our acquaintances // we simply forward them to our acquaintances
auto fwd = [=](scheduled_actor*, message& msg) -> result<message> { auto fwd = [=](scheduled_actor*, message& msg) -> skippable_result {
send_to_acquaintances(std::move(msg)); send_to_acquaintances(std::move(msg));
return message{}; return delegated<message>{};
}; };
set_default_handler(fwd); set_default_handler(fwd);
set_down_handler([=](down_msg& dm) { set_down_handler([=](down_msg& dm) {
...@@ -313,7 +313,7 @@ behavior proxy_broker::make_behavior() { ...@@ -313,7 +313,7 @@ behavior proxy_broker::make_behavior() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// instead of dropping "unexpected" messages, // instead of dropping "unexpected" messages,
// we simply forward them to our acquaintances // we simply forward them to our acquaintances
auto fwd = [=](local_actor*, message& msg) -> result<message> { auto fwd = [=](local_actor*, message& msg) -> skippable_result {
group_->send_all_subscribers(current_element_->sender, std::move(msg), group_->send_all_subscribers(current_element_->sender, std::move(msg),
context()); context());
return message{}; return message{};
......
...@@ -11,12 +11,10 @@ std::string to_string(match_result x) { ...@@ -11,12 +11,10 @@ std::string to_string(match_result x) {
switch(x) { switch(x) {
default: default:
return "???"; return "???";
case match_result::no_match:
return "no_match";
case match_result::match: case match_result::match:
return "match"; return "match";
case match_result::skip: case match_result::no_match:
return "skip"; return "no_match";
}; };
} }
......
...@@ -82,38 +82,27 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) { ...@@ -82,38 +82,27 @@ invoke_message_result raw_event_based_actor::consume(mailbox_element& x) {
} }
// handle everything else as ordinary message // handle everything else as ordinary message
detail::default_invoke_result_visitor<event_based_actor> visitor{this}; detail::default_invoke_result_visitor<event_based_actor> visitor{this};
bool skipped = false;
auto had_timeout = getf(has_timeout_flag); auto had_timeout = getf(has_timeout_flag);
if (had_timeout) if (had_timeout)
unsetf(has_timeout_flag); unsetf(has_timeout_flag);
// restore timeout at scope exit if message was skipped if (!bhvr_stack_.empty()) {
auto timeout_guard = detail::make_scope_guard([&] {
if (skipped && had_timeout)
setf(has_timeout_flag);
});
auto call_default_handler = [&] {
auto sres = call_handler(default_handler_, this, x.payload);
auto f = detail::make_overload([&](auto& x) { visitor.visit(x); },
[&](skip_t& x) { skipped = true; });
visit(f, sres);
};
if (bhvr_stack_.empty()) {
call_default_handler();
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
}
auto& bhvr = bhvr_stack_.back(); auto& bhvr = bhvr_stack_.back();
switch (bhvr(visitor, x.content())) { if (bhvr(visitor, x.content()) == match_result::match)
default: return invoke_message_result::consumed;
break;
case match_result::skip:
skipped = true;
break;
case match_result::no_match:
call_default_handler();
} }
return !skipped ? invoke_message_result::consumed auto sres = call_handler(default_handler_, this, x.payload);
: invoke_message_result::skipped; auto f = detail::make_overload(
[&](auto& x) {
visitor(x);
return invoke_message_result::consumed;
},
[&](skip_t& x) {
// Restore timeout if message was skipped.
if (had_timeout)
setf(has_timeout_flag);
return invoke_message_result::skipped;
});
return visit(f, sres);
}; };
// Post-process the returned value from the function body. // Post-process the returned value from the function body.
auto result = body(); auto result = body();
......
...@@ -36,26 +36,26 @@ namespace caf { ...@@ -36,26 +36,26 @@ namespace caf {
// -- related free functions --------------------------------------------------- // -- related free functions ---------------------------------------------------
result<message> reflect(scheduled_actor*, message& msg) { skippable_result reflect(scheduled_actor*, message& msg) {
return std::move(msg); return std::move(msg);
} }
result<message> reflect_and_quit(scheduled_actor* ptr, message& msg) { skippable_result reflect_and_quit(scheduled_actor* ptr, message& msg) {
error err = exit_reason::normal; error err = exit_reason::normal;
scheduled_actor::default_error_handler(ptr, err); scheduled_actor::default_error_handler(ptr, err);
return reflect(ptr, msg); return reflect(ptr, msg);
} }
result<message> print_and_drop(scheduled_actor* ptr, message& msg) { skippable_result print_and_drop(scheduled_actor* ptr, message& msg) {
CAF_LOG_WARNING("unexpected message:" << msg); CAF_LOG_WARNING("unexpected message:" << msg);
aout(ptr) << "*** unexpected message [id: " << ptr->id() aout(ptr) << "*** unexpected message [id: " << ptr->id()
<< ", name: " << ptr->name() << "]: " << to_string(msg) << ", name: " << ptr->name() << "]: " << to_string(msg)
<< std::endl; << std::endl;
return sec::unexpected_message; return make_error(sec::unexpected_message);
} }
result<message> drop(scheduled_actor*, message&) { skippable_result drop(scheduled_actor*, message&) {
return sec::unexpected_message; return make_error(sec::unexpected_message);
} }
// -- implementation details --------------------------------------------------- // -- implementation details ---------------------------------------------------
...@@ -67,7 +67,7 @@ void silently_ignore(scheduled_actor*, T&) { ...@@ -67,7 +67,7 @@ void silently_ignore(scheduled_actor*, T&) {
// nop // nop
} }
result<message> drop_after_quit(scheduled_actor* self, message&) { skippable_result drop_after_quit(scheduled_actor* self, message&) {
if (self->current_message_id().is_request()) if (self->current_message_id().is_request())
return make_error(sec::request_receiver_down); return make_error(sec::request_receiver_down);
return make_message(); return make_message();
...@@ -691,38 +691,26 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) { ...@@ -691,38 +691,26 @@ invoke_message_result scheduled_actor::consume(mailbox_element& x) {
return invoke_message_result::consumed; return invoke_message_result::consumed;
case message_category::ordinary: { case message_category::ordinary: {
detail::default_invoke_result_visitor<scheduled_actor> visitor{this}; detail::default_invoke_result_visitor<scheduled_actor> visitor{this};
bool skipped = false;
auto had_timeout = getf(has_timeout_flag); auto had_timeout = getf(has_timeout_flag);
if (had_timeout) if (had_timeout)
unsetf(has_timeout_flag); unsetf(has_timeout_flag);
// restore timeout at scope exit if message was skipped if (!bhvr_stack_.empty()) {
auto timeout_guard = detail::make_scope_guard([&] {
if (skipped && had_timeout)
setf(has_timeout_flag);
});
auto call_default_handler = [&] {
auto f = detail::make_overload([&](auto& val) { visitor.visit(val); },
[&](skip_t&) { skipped = true; });
auto sres = call_handler(default_handler_, this, x.payload);
visit(f, sres);
};
if (bhvr_stack_.empty()) {
call_default_handler();
return !skipped ? invoke_message_result::consumed
: invoke_message_result::skipped;
}
auto& bhvr = bhvr_stack_.back(); auto& bhvr = bhvr_stack_.back();
switch (bhvr(visitor, x.content())) { if (bhvr(visitor, x.content()) == match_result::match)
default: return invoke_message_result::consumed;
break;
case match_result::skip:
skipped = true;
break;
case match_result::no_match:
call_default_handler();
} }
return !skipped ? invoke_message_result::consumed auto sres = call_handler(default_handler_, this, x.payload);
: invoke_message_result::skipped; auto f = detail::make_overload(
[&](auto& x) {
visitor(x);
return invoke_message_result::consumed;
},
[&](skip_t&) {
if (had_timeout)
setf(has_timeout_flag);
return invoke_message_result::skipped;
});
return visit(f, sres);
} }
} }
// Unreachable. // Unreachable.
......
...@@ -23,10 +23,14 @@ ...@@ -23,10 +23,14 @@
namespace caf { namespace caf {
result<message> skip_t::skip_fun_impl(scheduled_actor*, message&) { namespace {
return skip();
skippable_result skip_fun_impl(scheduled_actor*, message&) {
return skip;
} }
} // namespace
skip_t::operator fun() const { skip_t::operator fun() const {
return skip_fun_impl; return skip_fun_impl;
} }
......
...@@ -43,9 +43,9 @@ CAF_TEST_FIXTURE_SCOPE(blocking_actor_tests, fixture) ...@@ -43,9 +43,9 @@ CAF_TEST_FIXTURE_SCOPE(blocking_actor_tests, fixture)
CAF_TEST(catch_all) { CAF_TEST(catch_all) {
self->send(self, 42); self->send(self, 42);
self->receive([](float) { CAF_FAIL("received unexpected float"); }, self->receive([](float) { CAF_FAIL("received unexpected float"); },
others >> [](message& msg) -> result<message> { others >> [](message& msg) -> skippable_result {
CAF_CHECK_EQUAL(to_string(msg), "(42)"); CAF_CHECK_EQUAL(to_string(msg), "(42)");
return sec::unexpected_message; return make_error(sec::unexpected_message);
}); });
self->receive( self->receive(
[](const error& err) { [](const error& err) {
...@@ -75,110 +75,4 @@ CAF_TEST(timeout_in_scoped_actor) { ...@@ -75,110 +75,4 @@ CAF_TEST(timeout_in_scoped_actor) {
CAF_CHECK(timeout_called); CAF_CHECK(timeout_called);
} }
// -- scoped_actors using skip -------------------------------------------------
using msg_t = int;
// send_order_t contains messages which are send to an actor in the same order
// as in vector
using send_order_t = std::vector<msg_t>;
// sequence_t contains a number of messages for processing by an actor with
// the information to skip the current message for later processing
using sequence_t = std::vector<std::pair<msg_t, bool>>;
using check_order_t = std::pair<send_order_t, sequence_t>;
behavior check_order_behavior_factory(local_actor*,
sequence_t::const_iterator* seq_it_ptr) {
return {
[=](int i) -> result<void> {
auto& seq_it = *seq_it_ptr;
CAF_CHECK_EQUAL(i, seq_it->first);
if (seq_it->second) {
CAF_MESSAGE("current: " << i << "; awaiting: " << seq_it->first
<< " SKIPPED");
++seq_it;
return skip();
} else {
CAF_MESSAGE("current: " << i << "; awaiting: " << seq_it->first
<< " OK");
++seq_it;
return unit;
}
}
};
}
void check_order_event_based_actor(const check_order_t& corder) {
actor_system_config cfg;
actor_system system{cfg};
auto& send_order = corder.first;
auto& sequence = corder.second;
auto seq_it = sequence.cbegin();
{
auto tmp = system.spawn(
[&](event_based_actor* self) {
self->set_default_handler(skip);
for(auto i : send_order) {
self->send(self, i);
}
self->become(
check_order_behavior_factory(self, &seq_it));
}
);
}
system.await_all_actors_done();
}
void check_order_scoped_actor(const check_order_t& corder) {
actor_system_config cfg;
actor_system system{cfg};
auto& send_order = corder.first;
auto& sequence = corder.second;
auto seq_it = begin(sequence);
scoped_actor self{system};
auto check_order_behavior =
check_order_behavior_factory(actor_cast<local_actor*>(self), &seq_it);
for(auto i : send_order) {
self->send(self, i);
}
while (seq_it != end(sequence)) {
self->receive(check_order_behavior);
}
}
CAF_TEST(skip_message) {
check_order_t a = {
{0, 1, 2, 3}, //recv_order = 0,1,2,3
{{0, false}, {1, false}, {2, false}, {3, false}}
};
check_order_t b = {
{3, 2, 1, 0}, //recv_order = 0,1,2,3
{{3, true}, {2, true}, {1, true}, {0, false},
{3, true}, {2, true}, {1, false},
{3, true}, {2, false},
{3, false}}
};
check_order_t c = {
{1, 0, 2}, //recv_order = 0,1,2
{{1, true}, {0, false},
{1, false},
{2, false}}
};
check_order_t d = {
{3, 1, 2, 0}, //recv_order = 0,1,2,3
{{3, true}, {1, true}, {2, true}, {0, false},
{3, true}, {1, false},
{3, true}, {2, false},
{3, false}}
};
check_order_event_based_actor(a);
check_order_event_based_actor(b);
check_order_event_based_actor(c);
check_order_event_based_actor(d);
check_order_scoped_actor(a);
check_order_scoped_actor(b);
check_order_scoped_actor(c);
check_order_scoped_actor(d);
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -165,34 +165,18 @@ CAF_TEST(typed_behavior_assignment) { ...@@ -165,34 +165,18 @@ CAF_TEST(typed_behavior_assignment) {
replies_to<double, double>::with<int, int>>; replies_to<double, double>::with<int, int>>;
// compatible handlers resulting in perfect match // compatible handlers resulting in perfect match
auto f1 = [=](int) { return 0.; }; auto f1 = [=](int) { return 0.; };
auto f2 = [=](double, double) { return std::make_tuple(0, 0); }; auto f2 = [=](double, double) -> result<int, int> { return {0, 0}; };
// compatible handlers using skip
auto g1 = [=](int) { return skip(); };
auto g2 = [=](double, double) { return skip(); };
// incompatbile handlers // incompatbile handlers
auto e1 = [=](int) { return 0.f; }; auto e1 = [=](int) { return 0.f; };
auto e2 = [=](double, double) { return std::make_tuple(0.f, 0.f); }; auto e2 = [=](double, double) { return std::make_tuple(0.f, 0.f); };
// omit one handler // omit one handler
CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(f1)); CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(f1));
CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(f2)); CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(f2));
CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(g1));
CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(g2));
CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(e1)); CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(e1));
CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(e2)); CAF_CHECK_EQUAL(bi_pair(false, -1), tb_assign<bh1>(e2));
// any valid alteration of (f1, f2, g1, g2) // any valid alteration of (f1, f2)
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(f1, f2)); CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(f1, f2));
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(f2, f1)); CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(f2, f1));
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(g1, g2));
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(g2, g1));
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(g1, f2));
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(f2, g1));
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(f1, g2));
CAF_CHECK_EQUAL(bi_pair(true, 2), tb_assign<bh1>(g2, f1));
// any invalid alteration of (f1, f2, g1, g2)
CAF_CHECK_EQUAL(bi_pair(false, 1), tb_assign<bh1>(f1, g1));
CAF_CHECK_EQUAL(bi_pair(false, 1), tb_assign<bh1>(g1, f1));
CAF_CHECK_EQUAL(bi_pair(false, 1), tb_assign<bh1>(f2, g2));
CAF_CHECK_EQUAL(bi_pair(false, 1), tb_assign<bh1>(g2, g2));
// any invalid alteration of (f1, f2, e1, e2) // any invalid alteration of (f1, f2, e1, e2)
CAF_CHECK_EQUAL(bi_pair(false, 1), tb_assign<bh1>(f1, e1)); CAF_CHECK_EQUAL(bi_pair(false, 1), tb_assign<bh1>(f1, e1));
CAF_CHECK_EQUAL(bi_pair(false, 1), tb_assign<bh1>(f1, e2)); CAF_CHECK_EQUAL(bi_pair(false, 1), tb_assign<bh1>(f1, e2));
......
...@@ -32,17 +32,10 @@ template<class T> ...@@ -32,17 +32,10 @@ template<class T>
void test_unit_void() { void test_unit_void() {
auto x = result<T>{}; auto x = result<T>{};
CAF_CHECK(holds_alternative<message>(x)); CAF_CHECK(holds_alternative<message>(x));
x = skip;
CAF_CHECK(holds_alternative<skip_t>(x));
} }
} // namespace anonymous } // namespace anonymous
CAF_TEST(skip) {
auto x = result<int>{skip};
CAF_CHECK(holds_alternative<skip_t>(x));
}
CAF_TEST(value) { CAF_TEST(value) {
auto x = result<int>{42}; auto x = result<int>{42};
CAF_REQUIRE(holds_alternative<message>(x)); CAF_REQUIRE(holds_alternative<message>(x));
......
...@@ -108,36 +108,33 @@ using event_testee_type ...@@ -108,36 +108,33 @@ using event_testee_type
class event_testee : public event_testee_type::base { class event_testee : public event_testee_type::base {
public: public:
event_testee(actor_config& cfg) : event_testee_type::base(cfg) { event_testee(actor_config& cfg) : event_testee_type::base(cfg) {
// nop set_default_handler(skip);
} }
behavior_type wait4string() { behavior_type wait4string() {
return { return {
partial_behavior_init,
[=](get_state_atom) { return "wait4string"; }, [=](get_state_atom) { return "wait4string"; },
[=](const string&) { become(wait4int()); }, [=](const string&) { become(wait4int()); },
[=](float) { return skip(); },
[=](int) { return skip(); },
}; };
} }
behavior_type wait4int() { behavior_type wait4int() {
return { return {
partial_behavior_init,
[=](get_state_atom) { return "wait4int"; }, [=](get_state_atom) { return "wait4int"; },
[=](int) -> int { [=](int) -> int {
become(wait4float()); become(wait4float());
return 42; return 42;
}, },
[=](float) { return skip(); },
[=](const string&) { return skip(); },
}; };
} }
behavior_type wait4float() { behavior_type wait4float() {
return { return {
partial_behavior_init,
[=](get_state_atom) { return "wait4float"; }, [=](get_state_atom) { return "wait4float"; },
[=](float) { become(wait4string()); }, [=](float) { become(wait4string()); },
[=](const string&) { return skip(); },
[=](int) { return skip(); },
}; };
} }
......
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