Unverified Commit a965a4c9 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #1067

Fix unexpected responses to anon_send, close #1046
parents b7d4f4be a83e4230
...@@ -31,31 +31,36 @@ ...@@ -31,31 +31,36 @@
namespace caf::detail { namespace caf::detail {
template <class Self, class Sender, class Handle, class... Ts> template <class Self, class SelfHandle, class Handle, class... Ts>
void profiled_send(Self* self, Sender&& sender, const Handle& receiver, void profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
message_id msg_id, std::vector<strong_actor_ptr> stages, message_id msg_id, std::vector<strong_actor_ptr> stages,
execution_unit* context, Ts&&... xs) { execution_unit* context, Ts&&... xs) {
CAF_IGNORE_UNUSED(self); CAF_IGNORE_UNUSED(self);
if (receiver) { if (dst) {
auto element = make_mailbox_element(std::forward<Sender>(sender), msg_id, auto element = make_mailbox_element(std::forward<SelfHandle>(src), msg_id,
std::move(stages), std::move(stages),
std::forward<Ts>(xs)...); std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING(self, *element); CAF_BEFORE_SENDING(self, *element);
receiver->enqueue(std::move(element), context); dst->enqueue(std::move(element), context);
} }
} }
template <class Self, class Sender, class Handle, class... Ts> template <class Self, class SelfHandle, class Handle, class... Ts>
void profiled_send(Self* self, Sender&& sender, const Handle& receiver, void profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
actor_clock& clock, actor_clock::time_point timeout, actor_clock& clock, actor_clock::time_point timeout,
message_id msg_id, Ts&&... xs) { message_id msg_id, Ts&&... xs) {
CAF_IGNORE_UNUSED(self); CAF_IGNORE_UNUSED(self);
if (receiver) { if (dst) {
auto element = make_mailbox_element(std::forward<Sender>(sender), msg_id, if constexpr (std::is_same<Handle, group>::value) {
no_stages, std::forward<Ts>(xs)...); clock.schedule_message(timeout, dst, std::forward<SelfHandle>(src),
CAF_BEFORE_SENDING_SCHEDULED(self, timeout, *element); make_message(std::forward<Ts>(xs)...));
clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(receiver), } else {
std::move(element)); auto element = make_mailbox_element(std::forward<SelfHandle>(src), msg_id,
no_stages, std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING_SCHEDULED(self, timeout, *element);
clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(dst),
std::move(element));
}
} }
} }
......
...@@ -94,7 +94,7 @@ public: ...@@ -94,7 +94,7 @@ public:
static_assert(response_type_unbox<signatures_of_t<Dest>, token>::valid, static_assert(response_type_unbox<signatures_of_t<Dest>, token>::valid,
"receiver does not accept given message"); "receiver does not accept given message");
auto self = dptr(); auto self = dptr();
detail::profiled_send(self, self->ctrl(), dest, make_message_id(P), {}, detail::profiled_send(self, nullptr, dest, make_message_id(P), {},
self->context(), std::forward<Ts>(xs)...); self->context(), std::forward<Ts>(xs)...);
} }
...@@ -172,8 +172,8 @@ public: ...@@ -172,8 +172,8 @@ public:
detail::type_list<detail::strip_and_convert_t<Ts>...> args_token; detail::type_list<detail::strip_and_convert_t<Ts>...> args_token;
type_check(dest, args_token); type_check(dest, args_token);
auto self = dptr(); auto self = dptr();
detail::profiled_send(self, self->ctrl(), dest, self->system().clock(), detail::profiled_send(self, nullptr, dest, self->system().clock(), timeout,
timeout, make_message_id(P), std::forward<Ts>(xs)...); make_message_id(P), std::forward<Ts>(xs)...);
} }
template <message_priority P = message_priority::normal, class Dest = actor, template <message_priority P = message_priority::normal, class Dest = actor,
...@@ -187,40 +187,11 @@ public: ...@@ -187,40 +187,11 @@ public:
auto self = dptr(); auto self = dptr();
auto& clock = self->system().clock(); auto& clock = self->system().clock();
auto timeout = clock.now() + rel_timeout; auto timeout = clock.now() + rel_timeout;
detail::profiled_send(self, self->ctrl(), dest, clock, timeout, detail::profiled_send(self, nullptr, dest, clock, timeout,
make_message_id(P), std::forward<Ts>(xs)...); make_message_id(P), std::forward<Ts>(xs)...);
} }
template <class Rep = int, class Period = std::ratio<1>, class... Ts>
void delayed_anon_send(const group& dest,
std::chrono::duration<Rep, Period> rtime, Ts&&... xs) {
delayed_anon_send_impl(dest, rtime, std::forward<Ts>(xs)...);
}
private: private:
template <class Dest, class... Ts>
void scheduled_send_impl(message_id mid, const Dest& dest, actor_clock& clock,
actor_clock::time_point timeout, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
detail::type_list<detail::strip_and_convert_t<Ts>...> args_token;
type_check(dest, args_token);
auto self = dptr();
detail::profiled_send(self, self->ctrl(), dest, clock, timeout, mid,
std::forward<Ts>(xs)...);
}
template <class Dest, class... Ts>
void
scheduled_anon_send_impl(message_id mid, const Dest& dest, actor_clock& clock,
actor_clock::time_point timeout, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
detail::type_list<detail::strip_and_convert_t<Ts>...> args_token;
type_check(dest, args_token);
auto self = dptr();
detail::profiled_send(self, nullptr, dest, clock, timeout, mid,
std::forward<Ts>(xs)...);
}
template <class Dest, class ArgTypes> template <class Dest, class ArgTypes>
static void type_check(const Dest&, ArgTypes) { static void type_check(const Dest&, ArgTypes) {
static_assert(!statically_typed<Subtype>() || statically_typed<Dest>(), static_assert(!statically_typed<Subtype>() || statically_typed<Dest>(),
......
...@@ -31,7 +31,7 @@ using std::chrono::seconds; ...@@ -31,7 +31,7 @@ using std::chrono::seconds;
namespace { namespace {
behavior testee_impl(event_based_actor* self) { behavior testee_impl(event_based_actor* self) {
self->set_default_handler(drop); self->set_default_handler(reflect);
return {[] { return {[] {
// nop // nop
}}; }};
...@@ -41,6 +41,8 @@ struct fixture : test_coordinator_fixture<> { ...@@ -41,6 +41,8 @@ struct fixture : test_coordinator_fixture<> {
group grp; group grp;
actor testee; actor testee;
std::string hello = "hello world";
fixture() { fixture() {
grp = sys.groups().anonymous(); grp = sys.groups().anonymous();
testee = sys.spawn_in_group(grp, testee_impl); testee = sys.spawn_in_group(grp, testee_impl);
...@@ -55,28 +57,48 @@ struct fixture : test_coordinator_fixture<> { ...@@ -55,28 +57,48 @@ struct fixture : test_coordinator_fixture<> {
CAF_TEST_FIXTURE_SCOPE(sender_tests, fixture) CAF_TEST_FIXTURE_SCOPE(sender_tests, fixture)
CAF_TEST(delayed actor message) { CAF_TEST(delayed actor messages receive responses) {
self->delayed_send(testee, seconds(1), "hello world"); self->delayed_send(testee, seconds(1), hello);
sched.trigger_timeout(); sched.trigger_timeout();
expect((std::string), from(self).to(testee).with("hello world")); expect((std::string), from(self).to(testee).with(hello));
} expect((std::string), from(testee).to(self).with(hello));
self->scheduled_send(testee, self->clock().now() + seconds(1), hello);
CAF_TEST(delayed group message) {
self->delayed_send(grp, seconds(1), "hello world");
sched.trigger_timeout(); sched.trigger_timeout();
expect((std::string), from(self).to(testee).with("hello world")); expect((std::string), from(self).to(testee).with(hello));
expect((std::string), from(testee).to(self).with(hello));
} }
CAF_TEST(scheduled actor message) { CAF_TEST(delayed group message receive responses) {
self->scheduled_send(testee, self->clock().now() + seconds(1), "hello world"); self->delayed_send(grp, seconds(1), hello);
sched.trigger_timeout();
expect((std::string), from(self).to(testee).with(hello));
expect((std::string), from(testee).to(self).with(hello));
self->scheduled_send(grp, self->clock().now() + seconds(1), hello);
sched.trigger_timeout(); sched.trigger_timeout();
expect((std::string), from(self).to(testee).with("hello world")); expect((std::string), from(self).to(testee).with(hello));
expect((std::string), from(testee).to(self).with(hello));
} }
CAF_TEST(scheduled group message) { CAF_TEST(anonymous messages receive no response) {
self->scheduled_send(grp, self->clock().now() + seconds(1), "hello world"); self->anon_send(testee, hello);
expect((std::string), to(testee).with(hello));
disallow((std::string), from(testee).to(self).with(hello));
self->delayed_anon_send(testee, seconds(1), hello);
sched.trigger_timeout();
expect((std::string), to(testee).with(hello));
disallow((std::string), from(testee).to(self).with(hello));
self->scheduled_anon_send(testee, self->clock().now() + seconds(1), hello);
sched.trigger_timeout();
expect((std::string), to(testee).with(hello));
disallow((std::string), from(testee).to(self).with(hello));
self->delayed_anon_send(grp, seconds(1), hello);
sched.trigger_timeout();
expect((std::string), to(testee).with(hello));
disallow((std::string), from(testee).to(self).with(hello));
self->scheduled_anon_send(grp, self->clock().now() + seconds(1), hello);
sched.trigger_timeout(); sched.trigger_timeout();
expect((std::string), from(self).to(testee).with("hello world")); expect((std::string), to(testee).with(hello));
disallow((std::string), from(testee).to(self).with(hello));
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
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