Commit ecf7df22 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'pr/1368'

parents 21316bb3 21cddd4c
...@@ -343,6 +343,8 @@ caf_add_component( ...@@ -343,6 +343,8 @@ caf_add_component(
response_handle response_handle
response_promise response_promise
result result
run_delayed
run_scheduled
save_inspector save_inspector
scheduled_actor scheduled_actor
serial_reply serial_reply
......
...@@ -647,6 +647,33 @@ public: ...@@ -647,6 +647,33 @@ public:
return run_scheduled(time_point_cast<duration_t>(when), make_action(what)); return run_scheduled(time_point_cast<duration_t>(when), make_action(what));
} }
/// Runs `what` asynchronously at some point after `when` if the actor still
/// exists. The callback is going to hold a weak reference to the actor, i.e.,
/// does not prevent the actor to become unreachable.
/// @param when The local time until the actor waits before invoking the
/// action. Due to scheduling delays, there will always be some
/// additional wait time. Passing the current time or a past time
/// immediately schedules the action for execution.
/// @param what The action to invoke after waiting on the timeout.
/// @returns A @ref disposable that allows the actor to cancel the action.
template <class Duration, class F>
disposable run_scheduled_weak(
std::chrono::time_point<std::chrono::system_clock, Duration> when, F what) {
using std::chrono::time_point_cast;
return run_scheduled_weak(time_point_cast<timespan>(when),
make_action(what));
}
/// @copydoc run_scheduled_weak
template <class Duration, class F>
disposable run_scheduled_weak(
std::chrono::time_point<actor_clock::clock_type, Duration> when, F what) {
using std::chrono::time_point_cast;
using duration_t = actor_clock::duration_type;
return run_scheduled_weak(time_point_cast<duration_t>(when),
make_action(what));
}
/// Runs `what` asynchronously after the `delay`. /// Runs `what` asynchronously after the `delay`.
/// @param delay Minimum amount of time that actor waits before invoking the /// @param delay Minimum amount of time that actor waits before invoking the
/// action. Due to scheduling delays, there will always be some /// action. Due to scheduling delays, there will always be some
...@@ -659,6 +686,21 @@ public: ...@@ -659,6 +686,21 @@ public:
return run_delayed(duration_cast<timespan>(delay), make_action(what)); return run_delayed(duration_cast<timespan>(delay), make_action(what));
} }
/// Runs `what` asynchronously after the `delay` if the actor still exists.
/// The callback is going to hold a weak reference to the actor, i.e., does
/// not prevent the actor to become unreachable.
/// @param delay Minimum amount of time that actor waits before invoking the
/// action. Due to scheduling delays, there will always be some
/// additional wait time.
/// @param what The action to invoke after the delay.
/// @returns A @ref disposable that allows the actor to cancel the action.
template <class Rep, class Period, class F>
disposable
run_delayed_weak(std::chrono::duration<Rep, Period> delay, F what) {
using std::chrono::duration_cast;
return run_delayed_weak(duration_cast<timespan>(delay), make_action(what));
}
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
/// Returns `true` if the actor has a behavior, awaits responses, or /// Returns `true` if the actor has a behavior, awaits responses, or
...@@ -744,7 +786,10 @@ private: ...@@ -744,7 +786,10 @@ private:
disposable run_scheduled(timestamp when, action what); disposable run_scheduled(timestamp when, action what);
disposable run_scheduled(actor_clock::time_point when, action what); disposable run_scheduled(actor_clock::time_point when, action what);
disposable run_scheduled_weak(timestamp when, action what);
disposable run_scheduled_weak(actor_clock::time_point when, action what);
disposable run_delayed(timespan delay, action what); disposable run_delayed(timespan delay, action what);
disposable run_delayed_weak(timespan delay, action what);
// -- caf::flow bindings ----------------------------------------------------- // -- caf::flow bindings -----------------------------------------------------
......
...@@ -885,11 +885,28 @@ disposable scheduled_actor::run_scheduled(actor_clock::time_point when, ...@@ -885,11 +885,28 @@ disposable scheduled_actor::run_scheduled(actor_clock::time_point when,
return clock().schedule(when, std::move(what), strong_actor_ptr{ctrl()}); return clock().schedule(when, std::move(what), strong_actor_ptr{ctrl()});
} }
disposable scheduled_actor::run_delayed(timespan delay, action what) { disposable scheduled_actor::run_scheduled_weak(timestamp when, action what) {
CAF_ASSERT(what.ptr() != nullptr);
CAF_LOG_TRACE(CAF_ARG(when));
auto delay = when - make_timestamp();
return run_scheduled_weak(clock().now() + delay, std::move(what));
}
disposable scheduled_actor::run_scheduled_weak(actor_clock::time_point when,
action what) {
CAF_ASSERT(what.ptr() != nullptr); CAF_ASSERT(what.ptr() != nullptr);
CAF_LOG_TRACE(CAF_ARG(when));
return clock().schedule(when, std::move(what), weak_actor_ptr{ctrl()});
}
disposable scheduled_actor::run_delayed(timespan delay, action what) {
CAF_LOG_TRACE(CAF_ARG(delay));
return run_scheduled(clock().now() + delay, std::move(what));
}
disposable scheduled_actor::run_delayed_weak(timespan delay, action what) {
CAF_LOG_TRACE(CAF_ARG(delay)); CAF_LOG_TRACE(CAF_ARG(delay));
return clock().schedule(clock().now() + delay, std::move(what), return run_scheduled_weak(clock().now() + delay, std::move(what));
strong_actor_ptr{ctrl()});
} }
// -- caf::flow bindings ------------------------------------------------------- // -- caf::flow bindings -------------------------------------------------------
......
// 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 run_delayed
#include "caf/event_based_actor.hpp"
#include "core-test.hpp"
using namespace caf;
using namespace std::literals;
namespace {
using fixture = test_coordinator_fixture<>;
behavior dummy_behavior() {
return {
[](int) {},
};
}
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("run_delayed triggers an action after a relative timeout") {
GIVEN("a scheduled actor") {
WHEN("the actor schedules an action with run_delayed") {
THEN("the action triggers after the relative timeout") {
auto called = std::make_shared<bool>(false);
auto aut = sys.spawn([called](event_based_actor* self) {
self->run_delayed(1s, [called] { *called = true; });
return dummy_behavior();
});
sched.run();
CHECK(!*called);
advance_time(1s);
sched.run();
CHECK(*called);
}
AND_THEN("disposing the pending timeout cancels the action") {
auto called = std::make_shared<bool>(false);
auto pending = disposable{};
auto aut = sys.spawn([called, &pending](event_based_actor* self) {
pending = self->run_delayed(1s, [called] { *called = true; });
return dummy_behavior();
});
sched.run();
CHECK(!*called);
pending.dispose();
advance_time(1s);
sched.run();
CHECK(!*called);
}
}
}
}
SCENARIO("run_delayed_weak triggers an action after a relative timeout") {
GIVEN("a scheduled actor") {
WHEN("the actor schedules an action with run_delayed") {
THEN("the action triggers after the relative timeout for live actors") {
auto called = std::make_shared<bool>(false);
auto aut = sys.spawn([called](event_based_actor* self) {
self->run_delayed_weak(1s, [called] { *called = true; });
return dummy_behavior();
});
sched.run();
CHECK(!*called);
advance_time(1s);
sched.run();
CHECK(*called);
}
AND_THEN("no action triggers for terminated actors") {
auto called = std::make_shared<bool>(false);
sys.spawn([called](event_based_actor* self) {
self->run_delayed_weak(1s, [called] { *called = true; });
return dummy_behavior();
});
sched.run(); // Note: actor cleaned up after this line.
CHECK(!*called);
advance_time(1s);
sched.run();
CHECK(!*called);
}
AND_THEN("disposing the pending timeout cancels the action") {
auto called = std::make_shared<bool>(false);
auto pending = disposable{};
auto aut = sys.spawn([called, &pending](event_based_actor* self) {
pending = self->run_delayed_weak(1s, [called] { *called = true; });
return dummy_behavior();
});
sched.run();
CHECK(!*called);
pending.dispose();
advance_time(1s);
sched.run();
CHECK(!*called);
}
}
}
}
END_FIXTURE_SCOPE()
// 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 run_scheduled
#include "caf/event_based_actor.hpp"
#include "core-test.hpp"
using namespace caf;
using namespace std::literals;
namespace {
using fixture = test_coordinator_fixture<>;
behavior dummy_behavior() {
return {
[](int) {},
};
}
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("run_scheduled triggers an action after a relative timeout") {
GIVEN("a scheduled actor") {
WHEN("the actor schedules an action with run_scheduled") {
THEN("the action triggers after the relative timeout") {
auto called = std::make_shared<bool>(false);
auto aut = sys.spawn([called](event_based_actor* self) {
auto now = self->clock().now();
self->run_scheduled(now + 1s, [called] { *called = true; });
return dummy_behavior();
});
sched.run();
CHECK(!*called);
advance_time(1s);
sched.run();
CHECK(*called);
}
AND_THEN("disposing the pending timeout cancels the action") {
auto called = std::make_shared<bool>(false);
auto pending = disposable{};
auto aut = sys.spawn([called, &pending](event_based_actor* self) {
auto now = self->clock().now();
pending = self->run_scheduled(now + 1s, [called] { *called = true; });
return dummy_behavior();
});
sched.run();
CHECK(!*called);
pending.dispose();
advance_time(1s);
sched.run();
CHECK(!*called);
}
}
}
}
SCENARIO("run_scheduled_weak triggers an action after a relative timeout") {
GIVEN("a scheduled actor") {
WHEN("the actor schedules an action with run_scheduled") {
THEN("the action triggers after the relative timeout for live actors") {
auto called = std::make_shared<bool>(false);
auto aut = sys.spawn([called](event_based_actor* self) {
auto now = self->clock().now();
self->run_scheduled_weak(now + 1s, [called] { *called = true; });
return dummy_behavior();
});
sched.run();
CHECK(!*called);
advance_time(1s);
sched.run();
CHECK(*called);
}
AND_THEN("no action triggers for terminated actors") {
auto called = std::make_shared<bool>(false);
sys.spawn([called](event_based_actor* self) {
auto now = self->clock().now();
self->run_scheduled_weak(now + 1s, [called] { *called = true; });
return dummy_behavior();
});
sched.run(); // Note: actor cleaned up after this line.
CHECK(!*called);
advance_time(1s);
sched.run();
CHECK(!*called);
}
AND_THEN("disposing the pending timeout cancels the action") {
auto called = std::make_shared<bool>(false);
auto pending = disposable{};
auto aut = sys.spawn([called, &pending](event_based_actor* self) {
auto now = self->clock().now();
pending = self->run_scheduled_weak(now + 1s,
[called] { *called = true; });
return dummy_behavior();
});
sched.run();
CHECK(!*called);
pending.dispose();
advance_time(1s);
sched.run();
CHECK(!*called);
}
}
}
}
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