Commit 10e876d1 authored by Dominik Charousset's avatar Dominik Charousset

Return a disposable for scheduled and delayed send

parent ea9fb5d4
......@@ -16,6 +16,7 @@ is based on [Keep a Changelog](https://keepachangelog.com).
unreachable if other actors no longer reference it.
- Typed actors that use a `typed_actor_pointer` can now access the
`run_{delayed,scheduled}` member functions.
- Scheduled and delayed sends now returns a disposable (#1362).
### Fixed
......
......@@ -332,6 +332,7 @@ caf_add_component(
message_builder
message_id
message_lifetime
messaging
metaprogramming
mixin.requester
mixin.sender
......
......@@ -10,6 +10,7 @@
#include "caf/actor_clock.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/actor_profiler.hpp"
#include "caf/disposable.hpp"
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/message_id.hpp"
......@@ -33,22 +34,23 @@ void profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
}
template <class Self, class SelfHandle, class Handle, class... Ts>
void profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
actor_clock& clock, actor_clock::time_point timeout,
[[maybe_unused]] message_id msg_id, Ts&&... xs) {
disposable profiled_send(Self* self, SelfHandle&& src, const Handle& dst,
actor_clock& clock, actor_clock::time_point timeout,
[[maybe_unused]] message_id msg_id, Ts&&... xs) {
if (dst) {
if constexpr (std::is_same<Handle, group>::value) {
clock.schedule_message(timeout, dst, std::forward<SelfHandle>(src),
make_message(std::forward<Ts>(xs)...));
return clock.schedule_message(timeout, dst, std::forward<SelfHandle>(src),
make_message(std::forward<Ts>(xs)...));
} else {
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));
return clock.schedule_message(timeout, actor_cast<strong_actor_ptr>(dst),
std::move(element));
}
} else {
self->home_system().base_metrics().rejected_messages->inc();
return {};
}
}
......
......@@ -99,7 +99,7 @@ public:
/// passed already).
template <message_priority P = message_priority::normal, 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>
scheduled_send(const Dest& dest, actor_clock::time_point timeout,
Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
......@@ -109,15 +109,16 @@ public:
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, self->system().clock(),
timeout, make_message_id(P), std::forward<Ts>(xs)...);
return detail::profiled_send(self, self->ctrl(), dest,
self->system().clock(), timeout,
make_message_id(P), std::forward<Ts>(xs)...);
}
/// Sends a message at given time point (or immediately if `timeout` has
/// passed already).
template <class... Ts>
void scheduled_send(const group& dest, actor_clock::time_point timeout,
Ts&&... xs) {
disposable scheduled_send(const group& dest, actor_clock::time_point timeout,
Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
static_assert((detail::sendable<Ts> && ...),
"at least one type has no ID, "
......@@ -128,17 +129,17 @@ public:
auto self = dptr();
if (dest) {
auto& clock = self->system().clock();
clock.schedule_message(timeout, dest, self->ctrl(),
make_message(std::forward<Ts>(xs)...));
} else {
self->home_system().base_metrics().rejected_messages->inc();
return clock.schedule_message(timeout, dest, self->ctrl(),
make_message(std::forward<Ts>(xs)...));
}
self->home_system().base_metrics().rejected_messages->inc();
return {};
}
/// Sends a message after a relative timeout.
template <message_priority P = message_priority::normal, class Rep = int,
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,
Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
......@@ -150,15 +151,16 @@ public:
auto self = dptr();
auto& clock = self->system().clock();
auto timeout = clock.now() + rel_timeout;
detail::profiled_send(self, self->ctrl(), dest, clock, timeout,
make_message_id(P), std::forward<Ts>(xs)...);
return detail::profiled_send(self, self->ctrl(), dest, clock, timeout,
make_message_id(P), std::forward<Ts>(xs)...);
}
/// Sends a message after a relative timeout.
template <class Rep = int, class Period = std::ratio<1>, class Dest = actor,
class... Ts>
void delayed_send(const group& dest, std::chrono::duration<Rep, Period> rtime,
Ts&&... xs) {
disposable delayed_send(const group& dest,
std::chrono::duration<Rep, Period> rtime,
Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
static_assert((detail::sendable<Ts> && ...),
"at least one type has no ID, "
......@@ -166,19 +168,21 @@ public:
static_assert(!statically_typed<Subtype>(),
"statically typed actors are not allowed to send to groups");
// TODO: consider whether it's feasible to track messages to groups
auto self = dptr();
if (dest) {
auto self = dptr();
auto& clock = self->system().clock();
auto timeout = clock.now() + rtime;
clock.schedule_message(timeout, dest, self->ctrl(),
make_message(std::forward<Ts>(xs)...));
return clock.schedule_message(timeout, dest, self->ctrl(),
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,
class... Ts>
void scheduled_anon_send(const Dest& dest, actor_clock::time_point timeout,
Ts&&... xs) {
disposable scheduled_anon_send(const Dest& dest,
actor_clock::time_point timeout, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
static_assert((detail::sendable<Ts> && ...),
"at least one type has no ID, "
......@@ -186,15 +190,16 @@ public:
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, self->system().clock(), timeout,
make_message_id(P), std::forward<Ts>(xs)...);
return detail::profiled_send(self, nullptr, dest, self->system().clock(),
timeout, make_message_id(P),
std::forward<Ts>(xs)...);
}
template <message_priority P = message_priority::normal, class Dest = actor,
class Rep = int, class Period = std::ratio<1>, class... Ts>
void delayed_anon_send(const Dest& dest,
std::chrono::duration<Rep, Period> rel_timeout,
Ts&&... xs) {
disposable delayed_anon_send(const Dest& dest,
std::chrono::duration<Rep, Period> rel_timeout,
Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
static_assert((detail::sendable<Ts> && ...),
"at least one type has no ID, "
......@@ -204,8 +209,8 @@ public:
auto self = dptr();
auto& clock = self->system().clock();
auto timeout = clock.now() + rel_timeout;
detail::profiled_send(self, nullptr, dest, clock, timeout,
make_message_id(P), std::forward<Ts>(xs)...);
return detail::profiled_send(self, nullptr, dest, clock, timeout,
make_message_id(P), std::forward<Ts>(xs)...);
}
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