Unverified Commit 75f42ebc authored by Noir's avatar Noir Committed by GitHub

Merge branch 'master' into issue/1365

parents 67647891 d844748d
...@@ -16,6 +16,7 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -16,6 +16,7 @@ is based on [Keep a Changelog](https://keepachangelog.com).
unreachable if other actors no longer reference it. unreachable if other actors no longer reference it.
- Typed actors that use a `typed_actor_pointer` can now access the - Typed actors that use a `typed_actor_pointer` can now access the
`run_{delayed,scheduled}` member functions. `run_{delayed,scheduled}` member functions.
- Scheduled and delayed sends now return a disposable (#1362).
- Typed response handles received support for converting them to observable or - Typed response handles received support for converting them to observable or
single objects. single objects.
- Typed actors that use the type-erased pointer-view type received access to the - Typed actors that use the type-erased pointer-view type received access to the
......
...@@ -332,6 +332,7 @@ caf_add_component( ...@@ -332,6 +332,7 @@ caf_add_component(
message_builder message_builder
message_id message_id
message_lifetime message_lifetime
messaging
metaprogramming metaprogramming
mixin.requester mixin.requester
mixin.sender mixin.sender
......
...@@ -10,6 +10,7 @@ ...@@ -10,6 +10,7 @@
#include "caf/actor_clock.hpp" #include "caf/actor_clock.hpp"
#include "caf/actor_control_block.hpp" #include "caf/actor_control_block.hpp"
#include "caf/actor_profiler.hpp" #include "caf/actor_profiler.hpp"
#include "caf/disposable.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
...@@ -33,22 +34,23 @@ void profiled_send(Self* self, SelfHandle&& src, const Handle& dst, ...@@ -33,22 +34,23 @@ void profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
} }
template <class Self, class SelfHandle, class Handle, class... Ts> template <class Self, class SelfHandle, class Handle, class... Ts>
void profiled_send(Self* self, SelfHandle&& src, const Handle& dst, disposable 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,
[[maybe_unused]] message_id msg_id, Ts&&... xs) { [[maybe_unused]] message_id msg_id, Ts&&... xs) {
if (dst) { if (dst) {
if constexpr (std::is_same<Handle, group>::value) { if constexpr (std::is_same<Handle, group>::value) {
clock.schedule_message(timeout, dst, std::forward<SelfHandle>(src), return clock.schedule_message(timeout, dst, std::forward<SelfHandle>(src),
make_message(std::forward<Ts>(xs)...)); make_message(std::forward<Ts>(xs)...));
} else { } else {
auto element = make_mailbox_element(std::forward<SelfHandle>(src), msg_id, auto element = make_mailbox_element(std::forward<SelfHandle>(src), msg_id,
no_stages, std::forward<Ts>(xs)...); no_stages, std::forward<Ts>(xs)...);
CAF_BEFORE_SENDING_SCHEDULED(self, timeout, *element); CAF_BEFORE_SENDING_SCHEDULED(self, timeout, *element);
clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(dst), return clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(dst),
std::move(element)); std::move(element));
} }
} else { } else {
self->home_system().base_metrics().rejected_messages->inc(); self->home_system().base_metrics().rejected_messages->inc();
return {};
} }
} }
......
...@@ -99,7 +99,7 @@ public: ...@@ -99,7 +99,7 @@ public:
/// passed already). /// passed already).
template <message_priority P = message_priority::normal, class Dest = actor, template <message_priority P = message_priority::normal, class Dest = actor,
class... Ts> class... Ts>
detail::enable_if_t<!std::is_same<Dest, group>::value> detail::enable_if_t<!std::is_same<Dest, group>::value, disposable>
scheduled_send(const Dest& dest, actor_clock::time_point timeout, scheduled_send(const Dest& dest, actor_clock::time_point timeout,
Ts&&... xs) { Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
...@@ -109,15 +109,16 @@ public: ...@@ -109,15 +109,16 @@ 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(), return detail::profiled_send(self, self->ctrl(), dest,
timeout, make_message_id(P), std::forward<Ts>(xs)...); self->system().clock(), timeout,
make_message_id(P), std::forward<Ts>(xs)...);
} }
/// Sends a message at given time point (or immediately if `timeout` has /// Sends a message at given time point (or immediately if `timeout` has
/// passed already). /// passed already).
template <class... Ts> template <class... Ts>
void scheduled_send(const group& dest, actor_clock::time_point timeout, disposable scheduled_send(const group& dest, actor_clock::time_point timeout,
Ts&&... xs) { Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
static_assert((detail::sendable<Ts> && ...), static_assert((detail::sendable<Ts> && ...),
"at least one type has no ID, " "at least one type has no ID, "
...@@ -128,17 +129,17 @@ public: ...@@ -128,17 +129,17 @@ public:
auto self = dptr(); auto self = dptr();
if (dest) { if (dest) {
auto& clock = self->system().clock(); auto& clock = self->system().clock();
clock.schedule_message(timeout, dest, self->ctrl(), return clock.schedule_message(timeout, dest, self->ctrl(),
make_message(std::forward<Ts>(xs)...)); make_message(std::forward<Ts>(xs)...));
} else {
self->home_system().base_metrics().rejected_messages->inc();
} }
self->home_system().base_metrics().rejected_messages->inc();
return {};
} }
/// Sends a message after a relative timeout. /// Sends a message after a relative timeout.
template <message_priority P = message_priority::normal, class Rep = int, template <message_priority P = message_priority::normal, class Rep = int,
class Period = std::ratio<1>, class Dest = actor, class... Ts> class Period = std::ratio<1>, class Dest = actor, class... Ts>
detail::enable_if_t<!std::is_same<Dest, group>::value> detail::enable_if_t<!std::is_same<Dest, group>::value, disposable>
delayed_send(const Dest& dest, std::chrono::duration<Rep, Period> rel_timeout, delayed_send(const Dest& dest, std::chrono::duration<Rep, Period> rel_timeout,
Ts&&... xs) { Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
...@@ -150,15 +151,16 @@ public: ...@@ -150,15 +151,16 @@ 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, return detail::profiled_send(self, self->ctrl(), dest, clock, timeout,
make_message_id(P), std::forward<Ts>(xs)...); make_message_id(P), std::forward<Ts>(xs)...);
} }
/// Sends a message after a relative timeout. /// Sends a message after a relative timeout.
template <class Rep = int, class Period = std::ratio<1>, class Dest = actor, template <class Rep = int, class Period = std::ratio<1>, class Dest = actor,
class... Ts> class... Ts>
void delayed_send(const group& dest, std::chrono::duration<Rep, Period> rtime, disposable delayed_send(const group& dest,
Ts&&... xs) { std::chrono::duration<Rep, Period> rtime,
Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
static_assert((detail::sendable<Ts> && ...), static_assert((detail::sendable<Ts> && ...),
"at least one type has no ID, " "at least one type has no ID, "
...@@ -166,19 +168,21 @@ public: ...@@ -166,19 +168,21 @@ public:
static_assert(!statically_typed<Subtype>(), static_assert(!statically_typed<Subtype>(),
"statically typed actors are not allowed to send to groups"); "statically typed actors are not allowed to send to groups");
// TODO: consider whether it's feasible to track messages to groups // TODO: consider whether it's feasible to track messages to groups
auto self = dptr();
if (dest) { if (dest) {
auto self = dptr();
auto& clock = self->system().clock(); auto& clock = self->system().clock();
auto timeout = clock.now() + rtime; auto timeout = clock.now() + rtime;
clock.schedule_message(timeout, dest, self->ctrl(), return clock.schedule_message(timeout, dest, self->ctrl(),
make_message(std::forward<Ts>(xs)...)); make_message(std::forward<Ts>(xs)...));
} }
self->home_system().base_metrics().rejected_messages->inc();
return {};
} }
template <message_priority P = message_priority::normal, class Dest = actor, template <message_priority P = message_priority::normal, class Dest = actor,
class... Ts> class... Ts>
void scheduled_anon_send(const Dest& dest, actor_clock::time_point timeout, disposable scheduled_anon_send(const Dest& dest,
Ts&&... xs) { actor_clock::time_point timeout, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
static_assert((detail::sendable<Ts> && ...), static_assert((detail::sendable<Ts> && ...),
"at least one type has no ID, " "at least one type has no ID, "
...@@ -186,15 +190,16 @@ public: ...@@ -186,15 +190,16 @@ 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, nullptr, dest, self->system().clock(), timeout, return detail::profiled_send(self, nullptr, dest, self->system().clock(),
make_message_id(P), std::forward<Ts>(xs)...); timeout, 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,
class Rep = int, class Period = std::ratio<1>, class... Ts> class Rep = int, class Period = std::ratio<1>, class... Ts>
void delayed_anon_send(const Dest& dest, disposable delayed_anon_send(const Dest& dest,
std::chrono::duration<Rep, Period> rel_timeout, std::chrono::duration<Rep, Period> rel_timeout,
Ts&&... xs) { Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
static_assert((detail::sendable<Ts> && ...), static_assert((detail::sendable<Ts> && ...),
"at least one type has no ID, " "at least one type has no ID, "
...@@ -204,8 +209,8 @@ public: ...@@ -204,8 +209,8 @@ 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, nullptr, dest, clock, timeout, return detail::profiled_send(self, nullptr, dest, clock, timeout,
make_message_id(P), std::forward<Ts>(xs)...); make_message_id(P), std::forward<Ts>(xs)...);
} }
private: private:
......
// This file is part of CAF, the C++ Actor Framework. See the file LICENSE in
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#define CAF_SUITE messaging
#include "caf/event_based_actor.hpp"
#include "core-test.hpp"
using namespace caf;
using namespace std::literals;
using self_ptr = event_based_actor*;
namespace {
struct fixture : test_coordinator_fixture<> {
actor uut1;
actor uut2;
disposable dis;
bool had_message = false;
};
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("send transfers a message from one actor to another") {
GIVEN("two actors: uut1 and uut2") {
WHEN("sending a message from uu2 to uu1") {
THEN("uut1 calls the appropriate message handler") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int i) {
had_message = true;
CHECK_EQ(i, 42);
CHECK_EQ(self->current_sender(), uut2);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { self->send(uut1, 42); });
run();
CHECK(had_message);
}
}
}
}
SCENARIO("delayed_send transfers the message after a relative timeout") {
GIVEN("two actors: uut1 and uut2") {
WHEN("sending a message from uu2 to uu1") {
THEN("uut1 calls the appropriate message handler") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int i) {
had_message = true;
CHECK_EQ(i, 42);
CHECK_EQ(self->current_sender(), uut2);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
self->delayed_send(uut1, 1s, 42);
});
sched.run();
CHECK(!had_message);
advance_time(1s);
sched.run();
CHECK(had_message);
}
}
}
}
SCENARIO("scheduled_send transfers the message after an absolute timeout") {
GIVEN("two actors: uut1 and uut2") {
WHEN("sending a message from uu2 to uu1") {
THEN("uut1 calls the appropriate message handler") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int i) {
had_message = true;
CHECK_EQ(i, 42);
CHECK_EQ(self->current_sender(), uut2);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
auto timeout = self->clock().now() + 1s;
self->scheduled_send(uut1, timeout, 42);
});
sched.run();
CHECK(!had_message);
advance_time(1s);
sched.run();
CHECK(had_message);
}
}
}
}
SCENARIO("anon_send hides the sender of a message") {
GIVEN("two actors: uut1 and uut2") {
WHEN("sending a message from uu2 to uu1") {
THEN("uut1 calls the appropriate message handler") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int i) {
had_message = true;
CHECK_EQ(i, 42);
CHECK_EQ(self->current_sender(), nullptr);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { self->anon_send(uut1, 42); });
run();
CHECK(had_message);
}
}
}
}
SCENARIO("delayed_anon_send hides the sender of a message") {
GIVEN("two actors: uut1 and uut2") {
WHEN("sending a message from uu2 to uu1") {
THEN("uut1 calls the appropriate message handler") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int i) {
had_message = true;
CHECK_EQ(i, 42);
CHECK_EQ(self->current_sender(), nullptr);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
self->delayed_anon_send(uut1, 1s, 42);
});
sched.run();
CHECK(!had_message);
advance_time(1s);
sched.run();
CHECK(had_message);
}
}
}
}
SCENARIO("scheduled_anon_send hides the sender of a message") {
GIVEN("two actors: uut1 and uut2") {
WHEN("sending a message from uu2 to uu1") {
THEN("uut1 calls the appropriate message handler") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int i) {
had_message = true;
CHECK_EQ(i, 42);
CHECK_EQ(self->current_sender(), nullptr);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
auto timeout = self->clock().now() + 1s;
self->scheduled_anon_send(uut1, timeout, 42);
});
sched.run();
CHECK(!had_message);
advance_time(1s);
sched.run();
CHECK(had_message);
}
}
}
}
SCENARIO("a delayed message may be canceled before its timeout") {
GIVEN("two actors: uut1 and uut2") {
WHEN("when disposing the message of delayed_send before it arrives") {
THEN("uut1 receives no message") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int) {
had_message = true;
CHECK_EQ(self->current_sender(), uut2);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
dis = self->delayed_send(uut1, 1s, 42);
});
sched.run();
CHECK(!had_message);
dis.dispose();
advance_time(1s);
run();
CHECK(!had_message);
}
}
WHEN("when disposing the message of delayed_anon_send before it arrives") {
THEN("uut1 receives no message") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int) {
had_message = true;
CHECK_EQ(self->current_sender(), uut2);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
dis = self->delayed_anon_send(uut1, 1s, 42);
});
sched.run();
CHECK(!had_message);
dis.dispose();
advance_time(1s);
run();
CHECK(!had_message);
}
}
}
}
SCENARIO("a scheduled message may be canceled before its timeout") {
GIVEN("two actors: uut1 and uut2") {
WHEN("when disposing the message of scheduled_send before it arrives") {
THEN("uut1 receives no message") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int) {
had_message = true;
CHECK_EQ(self->current_sender(), uut2);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
auto timeout = self->clock().now() + 1s;
dis = self->scheduled_send(uut1, timeout, 42);
});
sched.run();
CHECK(!had_message);
dis.dispose();
advance_time(1s);
run();
CHECK(!had_message);
}
}
WHEN(
"when disposing the message of scheduled_anon_send before it arrives") {
THEN("uut1 receives no message") {
uut1 = sys.spawn([this](self_ptr self) -> behavior {
return {
[this, self](int) {
had_message = true;
CHECK_EQ(self->current_sender(), uut2);
},
[](float) { CAF_FAIL("float handler called"); },
};
});
uut2 = sys.spawn([this](self_ptr self) { //
auto timeout = self->clock().now() + 1s;
dis = self->scheduled_anon_send(uut1, timeout, 42);
});
sched.run();
CHECK(!had_message);
dis.dispose();
advance_time(1s);
run();
CHECK(!had_message);
}
}
}
}
END_FIXTURE_SCOPE()
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