Commit e06f1a6c authored by Dominik Charousset's avatar Dominik Charousset

Simplify action API, implement interval operator

parent a90ad94f
......@@ -146,6 +146,7 @@ caf_add_component(
src/event_based_actor.cpp
src/execution_unit.cpp
src/flow/coordinator.cpp
src/flow/observable_builder
src/flow/scoped_coordinator.cpp
src/flow/subscription.cpp
src/forwarding_actor_proxy.cpp
......@@ -286,6 +287,7 @@ caf_add_component(
flow.concat_map
flow.flat_map
flow.for_each
flow.interval
flow.merge
flow.observe_on
flow.prefix_and_tail
......
......@@ -25,23 +25,12 @@ public:
enum class state {
disposed, /// The action may no longer run.
scheduled, /// The action is scheduled for execution.
invoked, /// The action fired and needs rescheduling before running again.
waiting, /// The action waits for reschedule but didn't run yet.
};
/// Describes the result of an attempted state transition.
enum class transition {
success, /// Transition completed as expected.
disposed, /// No transition since the action has been disposed.
failure, /// No transition since preconditions did not hold.
};
/// Internal interface of `action`.
class CAF_CORE_EXPORT impl : public disposable::impl {
public:
virtual transition reschedule() = 0;
virtual transition run() = 0;
virtual void run() = 0;
virtual state current_state() const noexcept = 0;
......@@ -58,9 +47,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
explicit action(impl_ptr ptr) noexcept : pimpl_(std::move(ptr)) {
// nop
}
explicit action(impl_ptr ptr) noexcept;
action() noexcept = default;
......@@ -82,28 +69,18 @@ public:
return pimpl_->current_state() == state::scheduled;
}
[[nodiscard]] bool invoked() const {
return pimpl_->current_state() == state::invoked;
}
// -- mutators ---------------------------------------------------------------
/// Tries to transition from `scheduled` to `invoked`, running the body of the
/// internal function object as a side effect on success.
/// @return whether the transition took place.
transition run();
/// Triggers the action.
void run() {
pimpl_->run();
}
/// Cancel the action if it has not been invoked yet.
void dispose() {
pimpl_->dispose();
}
/// Tries setting the state from `invoked` back to `scheduled`.
/// @return whether the transition took place.
transition reschedule() {
return pimpl_->reschedule();
}
// -- conversion -------------------------------------------------------------
/// Returns a smart pointer to the implementation.
......@@ -143,8 +120,8 @@ struct default_action_impl : ref_counted, action::impl {
std::atomic<action::state> state_;
F f_;
default_action_impl(F fn, action::state init_state)
: state_(init_state), f_(std::move(fn)) {
default_action_impl(F fn)
: state_(action::state::scheduled), f_(std::move(fn)) {
// nop
}
......@@ -160,42 +137,12 @@ struct default_action_impl : ref_counted, action::impl {
return state_.load();
}
action::transition reschedule() override {
auto st = action::state::invoked;
for (;;) {
if (state_.compare_exchange_strong(st, action::state::scheduled))
return action::transition::success;
switch (st) {
case action::state::invoked:
case action::state::waiting:
break; // Try again.
case action::state::disposed:
return action::transition::disposed;
default:
return action::transition::failure;
}
}
}
action::transition run() override {
auto st = state_.load();
switch (st) {
case action::state::scheduled:
f_();
// No retry. If this action has been disposed while running, we stay
// in the state 'disposed'. We assume that only one thread may try to
// transition from scheduled to invoked, while other threads may only
// dispose the action.
if (state_.compare_exchange_strong(st, action::state::invoked)) {
return action::transition::success;
} else {
CAF_ASSERT(st == action::state::disposed);
return action::transition::disposed;
}
case action::state::disposed:
return action::transition::disposed;
default:
return action::transition::failure;
void run() override {
// Note: we do *not* set the state to disposed after running the function
// object. This allows the action to re-register itself when needed, e.g.,
// to implement time-based loops.
if (state_.load() == action::state::scheduled) {
f_();
}
}
......@@ -220,17 +167,12 @@ struct default_action_impl : ref_counted, action::impl {
namespace caf {
/// Convenience function for creating @ref action objects from a function
/// object.
/// Convenience function for creating an @ref action from a function object.
/// @param f The body for the action.
/// @param init_state either `action::state::scheduled` or
/// `action::state::waiting`.
template <class F>
action make_action(F f, action::state init_state = action::state::scheduled) {
CAF_ASSERT(init_state == action::state::scheduled
|| init_state == action::state::waiting);
action make_action(F f) {
using impl_t = detail::default_action_impl<F>;
return action{make_counted<impl_t>(std::move(f), init_state)};
return action{make_counted<impl_t>(std::move(f))};
}
} // namespace caf
......
......@@ -53,18 +53,7 @@ public:
/// @param f The action to schedule.
/// @note The action runs on the thread of the clock worker and thus must
/// complete within a very short time in order to not delay other work.
disposable schedule(time_point t, action f);
/// Schedules an action for periodic execution.
/// @param first_run The local time at which the action should run initially.
/// @param f The action to schedule.
/// @param period The time to wait between two runs. A non-positive period
/// disables periodic execution.
/// @note The action runs on the thread of the clock worker and thus must
/// complete within a very short time in order to not delay other work.
virtual disposable
schedule_periodically(time_point first_run, action f, duration_type period)
= 0;
virtual disposable schedule(time_point t, action f) = 0;
/// Schedules an action for execution by an actor at a later time.
/// @param t The local time at which the action should get enqueued to the
......@@ -73,17 +62,6 @@ public:
/// @param target The actor that should run the action.
disposable schedule(time_point t, action f, strong_actor_ptr target);
/// Schedules an action for periodic execution by an actor.
/// @param first_run The local time at which the action should get enqueued to
/// the mailbox of the target for the first time.
/// @param f The action to schedule.
/// @param target The actor that should run the action.
/// @param period The time to wait between two messages to the actor.
/// @param policy The strategy for dealing with a stalling target.
disposable schedule_periodically(time_point first_run, action f,
strong_actor_ptr target,
duration_type period, stall_policy policy);
/// Schedules an action for execution by an actor at a later time.
/// @param target The actor that should run the action.
/// @param f The action to schedule.
......@@ -91,17 +69,6 @@ public:
/// mailbox of the target.
disposable schedule(time_point t, action f, weak_actor_ptr target);
/// Schedules an action for periodic execution by an actor.
/// @param first_run The local time at which the action should get enqueued to
/// the mailbox of the target for the first time.
/// @param f The action to schedule.
/// @param target The actor that should run the action.
/// @param period The time to wait between two messages to the actor.
/// @param policy The strategy for dealing with a stalling target.
disposable schedule_periodically(time_point first_run, action f,
weak_actor_ptr target, duration_type period,
stall_policy policy);
/// Schedules an arbitrary message to `receiver` for time point `t`.
disposable schedule_message(time_point t, strong_actor_ptr receiver,
mailbox_element_ptr content);
......
......@@ -15,15 +15,6 @@ namespace caf::detail {
class CAF_CORE_EXPORT test_actor_clock : public actor_clock {
public:
// -- member types -----------------------------------------------------------
struct schedule_entry {
action f;
duration_type period;
};
using schedule_map = std::multimap<time_point, schedule_entry>;
// -- constructors, destructors, and assignment operators --------------------
test_actor_clock();
......@@ -32,17 +23,14 @@ public:
time_point now() const noexcept override;
disposable schedule_periodically(time_point first_run, action f,
duration_type period) override;
disposable schedule(time_point abs_time, action f) override;
// -- testing DSL API --------------------------------------------------------
/// Returns whether the actor clock has at least one pending timeout.
bool has_pending_timeout() const {
auto not_disposed = [](const auto& kvp) {
return !kvp.second.f.disposed();
};
return std::any_of(schedule.begin(), schedule.end(), not_disposed);
auto not_disposed = [](const auto& kvp) { return !kvp.second.disposed(); };
return std::any_of(actions.begin(), actions.end(), not_disposed);
}
/// Triggers the next pending timeout regardless of its timestamp. Sets
......@@ -65,7 +53,7 @@ public:
time_point current_time;
schedule_map schedule;
std::multimap<time_point, action> actions;
private:
bool try_trigger_once();
......
......@@ -25,11 +25,12 @@ public:
// -- member types -----------------------------------------------------------
using super = actor_clock;
/// Stores actions along with their scheduling period.
struct schedule_entry {
time_point t;
action f;
duration_type period;
};
/// @relates schedule_entry
......@@ -41,8 +42,9 @@ public:
// -- overrides --------------------------------------------------------------
disposable schedule_periodically(time_point first_run, action f,
duration_type period) override;
using super::schedule;
disposable schedule(time_point abs_time, action f) override;
// -- thread management ------------------------------------------------------
......
......@@ -52,6 +52,11 @@ public:
disposable& operator=(const disposable&) noexcept = default;
disposable& operator=(std::nullptr_t) noexcept {
pimpl_ = nullptr;
return *this;
}
// -- factories --------------------------------------------------------------
/// Combines multiple disposables into a single disposable. The new disposable
......
......@@ -4,14 +4,16 @@
#pragma once
#include <tuple>
#include "caf/action.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/flow/fwd.hpp"
#include "caf/flow/subscription.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/ref_counted.hpp"
#include "caf/timespan.hpp"
#include <chrono>
#include <tuple>
namespace caf::flow {
......@@ -20,12 +22,20 @@ namespace caf::flow {
/// objects since the coordinator guarantees synchronous execution.
class CAF_CORE_EXPORT coordinator {
public:
// -- friends ----------------------------------------------------------------
friend class subscription_impl;
// -- member types -----------------------------------------------------------
/// A time point of the monotonic clock.
using steady_time_point = std::chrono::steady_clock::time_point;
// -- constructors, destructors, and assignment operators --------------------
virtual ~coordinator();
/// Returns a factory object for new observable objects on this coordinator.
[[nodiscard]] observable_builder make_observable();
// -- reference counting -----------------------------------------------------
/// Increases the reference count of the coordinator.
virtual void ref_coordinator() const noexcept = 0;
......@@ -34,6 +44,33 @@ public:
/// if necessary.
virtual void deref_coordinator() const noexcept = 0;
friend void intrusive_ptr_add_ref(const coordinator* ptr) noexcept {
ptr->ref_coordinator();
}
friend void intrusive_ptr_release(const coordinator* ptr) noexcept {
ptr->deref_coordinator();
}
// -- conversions ------------------------------------------------------------
/// Returns a factory object for new observable objects on this coordinator.
[[nodiscard]] observable_builder make_observable();
// -- lifetime management ----------------------------------------------------
/// Asks the coordinator to keep its event loop running until `obj` becomes
/// disposed since it depends on external events or produces events that are
/// visible to outside observers.
virtual void watch(disposable what) = 0;
// -- time -------------------------------------------------------------------
/// Returns the current time on the monotonic clock of this coordinator.
virtual steady_time_point steady_time() = 0;
// -- scheduling of actions --------------------------------------------------
/// Schedules an action for execution on this coordinator. This member
/// function may get called from external sources or threads.
/// @thread-safe
......@@ -45,27 +82,42 @@ public:
return schedule(make_action(std::forward<F>(what)));
}
/// Schedules an action for execution from within the coordinator. May call
/// `schedule` for coordinators that use a single work queue.
virtual void post_internally(action what) = 0;
/// Delays execution of an action until all pending actions were executed. May
/// call `schedule`.
/// @param what The action for delayed execution.
virtual void delay(action what) = 0;
///@copydoc post_internally
///@copydoc delay
template <class F>
void post_internally_fn(F&& what) {
return post_internally(make_action(std::forward<F>(what)));
void delay_fn(F&& what) {
return delay(make_action(std::forward<F>(what)));
}
/// Asks the coordinator to keep its event loop running until `obj` becomes
/// disposed since it depends on external events or produces events that are
/// visible to outside observers.
virtual void watch(disposable what) = 0;
/// Delays execution of an action with an absolute timeout.
/// @param abs_time The absolute time when this action should take place.
/// @param what The action for delayed execution.
/// @returns a @ref disposable to cancel the pending timeout.
virtual disposable delay_until(steady_time_point abs_time, action what) = 0;
friend void intrusive_ptr_add_ref(const coordinator* ptr) noexcept {
ptr->ref_coordinator();
///@copydoc delay
template <class F>
disposable delay_until_fn(steady_time_point abs_time, F&& what) {
return delay_until(abs_time, make_action(std::forward<F>(what)));
}
friend void intrusive_ptr_release(const coordinator* ptr) noexcept {
ptr->deref_coordinator();
/// Delays execution of an action with a relative timeout.
/// @param rel_time The relative time when this action should take place.
/// @param what The action for delayed execution.
/// @returns a @ref disposable to cancel the pending timeout.
disposable delay_for(timespan rel_time, action what) {
return delay_until(steady_time() + rel_time, std::move(what));
}
///@copydoc delay_for
template <class F>
void delay_for_fn(timespan rel_time, F&& what) {
return delay_until(steady_time() + rel_time,
make_action(std::forward<F>(what)));
}
};
......
......@@ -89,14 +89,14 @@ public:
void request(size_t n) override {
if (src_)
ctx()->post_internally_fn([src = src_, snk = snk_, n] { //
ctx()->delay_fn([src = src_, snk = snk_, n] { //
src->on_request(snk.get(), n);
});
}
void cancel() override {
if (src_) {
ctx()->post_internally_fn([src = src_, snk = snk_] { //
ctx()->delay_fn([src = src_, snk = snk_] { //
src->on_cancel(snk.get());
});
src_.reset();
......
......@@ -6,9 +6,12 @@
#include "caf/async/spsc_buffer.hpp"
#include "caf/defaults.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/flow/coordinator.hpp"
#include "caf/flow/observable.hpp"
#include <cstdint>
namespace caf::flow {
// -- forward declarations -----------------------------------------------------
......@@ -25,6 +28,68 @@ class value_source;
template <class F>
class callable_source;
// -- special-purpose observable implementations -------------------------------
class interval_action;
class CAF_CORE_EXPORT interval_impl : public ref_counted,
public observable_impl<int64_t> {
public:
// -- member types -----------------------------------------------------------
using output_type = int64_t;
using super = observable_impl<int64_t>;
// -- friends ----------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(interval_impl)
friend class interval_action;
// -- constructors, destructors, and assignment operators --------------------
interval_impl(coordinator* ctx, timespan initial_delay, timespan period);
interval_impl(coordinator* ctx, timespan initial_delay, timespan period,
int64_t max_val);
~interval_impl() override;
// -- implementation of disposable::impl -------------------------------------
void dispose() override;
bool disposed() const noexcept override;
void ref_disposable() const noexcept override;
void deref_disposable() const noexcept override;
// -- implementation of observable<T>::impl ----------------------------------
coordinator* ctx() const noexcept override;
void on_request(observer_impl<int64_t>*, size_t) override;
void on_cancel(observer_impl<int64_t>*) override;
disposable subscribe(observer<int64_t> what) override;
private:
void fire(interval_action*);
coordinator* ctx_;
observer<int64_t> obs_;
disposable pending_;
timespan initial_delay_;
timespan period_;
coordinator::steady_time_point last_;
int64_t val_ = 0;
int64_t max_;
size_t demand_ = 0;
};
// -- builder interface --------------------------------------------------------
/// Factory for @ref observable objects.
......@@ -65,6 +130,17 @@ public:
[[nodiscard]] observable<T>
from_resource(async::consumer_resource<T> res) const;
template <class Rep, class Period>
[[nodiscard]] observable<int64_t>
interval(std::chrono::duration<Rep, Period> initial_delay,
std::chrono::duration<Rep, Period> period) {
// Intervals introduce a time-dependency, so we need to watch them in order
// to prevent actors from shutting down while timeouts are still pending.
auto ptr = make_counted<interval_impl>(ctx_, initial_delay, period);
ctx_->watch(ptr->as_disposable());
return observable<int64_t>{std::move(ptr)};
}
private:
explicit observable_builder(coordinator* ctx) : ctx_(ctx) {
// nop
......
......@@ -5,6 +5,7 @@
#pragma once
#include <condition_variable>
#include <map>
#include <mutex>
#include "caf/flow/coordinator.hpp"
......@@ -16,35 +17,55 @@ namespace caf::flow {
class CAF_CORE_EXPORT scoped_coordinator final : public ref_counted,
public coordinator {
public:
void run();
// -- factories --------------------------------------------------------------
void ref_coordinator() const noexcept override;
static intrusive_ptr<scoped_coordinator> make();
void deref_coordinator() const noexcept override;
// -- execution --------------------------------------------------------------
void schedule(action what) override;
void run();
void post_internally(action what) override;
// -- reference counting -----------------------------------------------------
void watch(disposable what) override;
void ref_coordinator() const noexcept override;
void deref_coordinator() const noexcept override;
friend void intrusive_ptr_add_ref(const scoped_coordinator* ptr) {
ptr->ref();
}
friend void intrusive_ptr_release(scoped_coordinator* ptr) {
friend void intrusive_ptr_release(const scoped_coordinator* ptr) {
ptr->deref();
}
static intrusive_ptr<scoped_coordinator> make();
// -- lifetime management ----------------------------------------------------
void watch(disposable what) override;
// -- time -------------------------------------------------------------------
steady_time_point steady_time() override;
// -- scheduling of actions --------------------------------------------------
void schedule(action what) override;
void delay(action what) override;
disposable delay_until(steady_time_point abs_time, action what) override;
private:
scoped_coordinator() = default;
action next(bool blocking);
void drop_disposed_flows();
std::vector<disposable> watched_disposables_;
std::multimap<steady_time_point, action> delayed_;
mutable std::mutex mtx_;
std::condition_variable cv_;
std::vector<action> actions_;
......
......@@ -483,13 +483,17 @@ public:
// -- caf::flow API ----------------------------------------------------------
steady_time_point steady_time() override;
void ref_coordinator() const noexcept override;
void deref_coordinator() const noexcept override;
void schedule(action what) override;
void post_internally(action what) override;
void delay(action what) override;
disposable delay_until(steady_time_point abs_time, action what) override;
void watch(disposable what) override;
......@@ -600,8 +604,7 @@ public:
disposable run_scheduled(
std::chrono::time_point<std::chrono::system_clock, Duration> when, F what) {
using std::chrono::time_point_cast;
return run_scheduled(time_point_cast<timespan>(when),
make_action(what, action::state::waiting));
return run_scheduled(time_point_cast<timespan>(when), make_action(what));
}
/// @copydoc run_scheduled
......@@ -611,8 +614,7 @@ public:
F what) {
using std::chrono::time_point_cast;
using duration_t = actor_clock::duration_type;
return run_scheduled(time_point_cast<duration_t>(when),
make_action(what, action::state::waiting));
return run_scheduled(time_point_cast<duration_t>(when), make_action(what));
}
/// Runs `what` asynchronously after the `delay`.
......@@ -624,8 +626,7 @@ public:
template <class Rep, class Period, class F>
disposable run_delayed(std::chrono::duration<Rep, Period> delay, F what) {
using std::chrono::duration_cast;
return run_delayed(duration_cast<timespan>(delay),
make_action(what, action::state::waiting));
return run_delayed(duration_cast<timespan>(delay), make_action(what));
}
// -- stream processing ------------------------------------------------------
......
......@@ -8,10 +8,8 @@
namespace caf {
action::transition action::run() {
CAF_LOG_TRACE("");
CAF_ASSERT(pimpl_ != nullptr);
return pimpl_->run();
action::action(impl_ptr ptr) noexcept : pimpl_(std::move(ptr)) {
// nop
}
} // namespace caf
......@@ -24,28 +24,19 @@ class action_decorator : public ref_counted, public action::impl {
public:
using state = action::state;
using transition = action::transition;
action_decorator(action::impl_ptr decorated, WorkerPtr worker,
actor_clock::stall_policy policy)
: decorated_(std::move(decorated)),
worker_(std::move(worker)),
policy_(policy) {
action_decorator(action::impl_ptr decorated, WorkerPtr worker)
: decorated_(std::move(decorated)), worker_(std::move(worker)) {
CAF_ASSERT(decorated_ != nullptr);
CAF_ASSERT(worker_ != nullptr);
}
void dispose() override {
if (decorated_) {
decorated_->dispose();
decorated_ = nullptr;
}
if (worker_)
worker_ = nullptr;
decorated_->dispose();
worker_ = nullptr;
}
bool disposed() const noexcept override {
return decorated_ ? decorated_->disposed() : true;
return decorated_->disposed();
}
void ref_disposable() const noexcept override {
......@@ -56,28 +47,23 @@ public:
deref();
}
transition reschedule() override {
// Always succeeds since we implicitly reschedule in do_run.
return transition::success;
}
transition run() override {
void run() override {
CAF_ASSERT(decorated_ != nullptr);
CAF_ASSERT(worker_ != nullptr);
if constexpr (std::is_same_v<WorkerPtr, weak_actor_ptr>) {
if (auto ptr = actor_cast<strong_actor_ptr>(worker_)) {
return do_run(ptr);
do_run(ptr);
} else {
dispose();
return transition::disposed;
decorated_->dispose();
}
} else {
return do_run(worker_);
do_run(worker_);
}
worker_ = nullptr;
}
state current_state() const noexcept override {
return decorated_ ? decorated_->current_state() : action::state::disposed;
return decorated_->current_state();
}
friend void intrusive_ptr_add_ref(const action_decorator* ptr) noexcept {
......@@ -89,44 +75,21 @@ public:
}
private:
transition do_run(strong_actor_ptr& ptr) {
switch (decorated_->reschedule()) {
case transition::disposed:
decorated_ = nullptr;
worker_ = nullptr;
return transition::disposed;
case transition::success:
if (ptr->enqueue(nullptr, make_message_id(),
make_message(action{decorated_}), nullptr)) {
return transition::success;
} else {
dispose();
return transition::disposed;
}
default:
if (policy_ == actor_clock::stall_policy::fail) {
ptr->enqueue(nullptr, make_message_id(),
make_message(make_error(sec::action_reschedule_failed)),
nullptr);
dispose();
return transition::failure;
} else {
return transition::success;
}
}
void do_run(strong_actor_ptr& ptr) {
ptr->enqueue(nullptr, make_message_id(), make_message(action{decorated_}),
nullptr);
}
action::impl_ptr decorated_;
WorkerPtr worker_;
actor_clock::stall_policy policy_;
};
template <class WorkerPtr>
action decorate(action f, WorkerPtr worker, actor_clock::stall_policy policy) {
action decorate(action f, WorkerPtr worker) {
CAF_ASSERT(f.ptr() != nullptr);
using impl_t = action_decorator<WorkerPtr>;
auto ptr = make_counted<impl_t>(std::move(f).as_intrusive_ptr(),
std::move(worker), policy);
std::move(worker));
return action{std::move(ptr)};
}
......@@ -145,44 +108,17 @@ actor_clock::time_point actor_clock::now() const noexcept {
}
disposable actor_clock::schedule(action f) {
return schedule_periodically(time_point{duration_type{0}}, std::move(f),
duration_type{0});
}
disposable actor_clock::schedule(time_point t, action f) {
return schedule_periodically(t, std::move(f), duration_type{0});
return schedule(time_point{duration_type{0}}, std::move(f));
}
disposable actor_clock::schedule(time_point t, action f,
strong_actor_ptr worker) {
return schedule_periodically(t, std::move(f), std::move(worker),
duration_type{0}, stall_policy::skip);
}
disposable actor_clock::schedule_periodically(time_point first_run, action f,
strong_actor_ptr worker,
duration_type period,
stall_policy policy) {
auto res = f.as_disposable();
auto g = decorate(std::move(f), std::move(worker), policy);
schedule_periodically(first_run, std::move(g), period);
return res;
return schedule(t, decorate(std::move(f), std::move(worker)));
}
disposable actor_clock::schedule(time_point t, action f,
weak_actor_ptr worker) {
return schedule_periodically(t, std::move(f), std::move(worker),
duration_type{0}, stall_policy::skip);
}
disposable actor_clock::schedule_periodically(time_point first_run, action f,
weak_actor_ptr worker,
duration_type period,
stall_policy policy) {
auto res = f.as_disposable();
auto g = decorate(std::move(f), std::move(worker), policy);
schedule_periodically(first_run, std::move(g), period);
return res;
return schedule(t, decorate(std::move(f), std::move(worker)));
}
disposable actor_clock::schedule_message(time_point t,
......
......@@ -13,11 +13,9 @@ test_actor_clock::test_actor_clock() : current_time(duration_type{1}) {
// time_point, because begin-of-epoch may have special meaning.
}
disposable test_actor_clock::schedule_periodically(time_point first_run,
action f,
duration_type period) {
disposable test_actor_clock::schedule(time_point abs_time, action f) {
CAF_ASSERT(f.ptr() != nullptr);
schedule.emplace(first_run, schedule_entry{f, period});
actions.emplace(abs_time, f);
return std::move(f).as_disposable();
}
......@@ -26,11 +24,11 @@ test_actor_clock::time_point test_actor_clock::now() const noexcept {
}
bool test_actor_clock::trigger_timeout() {
CAF_LOG_TRACE(CAF_ARG2("schedule.size", schedule.size()));
CAF_LOG_TRACE(CAF_ARG2("actions.size", actions.size()));
for (;;) {
if (schedule.empty())
if (actions.empty())
return false;
auto i = schedule.begin();
auto i = actions.begin();
auto t = i->first;
if (t > current_time)
current_time = t;
......@@ -40,8 +38,8 @@ bool test_actor_clock::trigger_timeout() {
}
size_t test_actor_clock::trigger_timeouts() {
CAF_LOG_TRACE(CAF_ARG2("schedule.size", schedule.size()));
if (schedule.empty())
CAF_LOG_TRACE(CAF_ARG2("actions.size", actions.size()));
if (actions.empty())
return 0u;
size_t result = 0;
while (trigger_timeout())
......@@ -50,33 +48,29 @@ size_t test_actor_clock::trigger_timeouts() {
}
size_t test_actor_clock::advance_time(duration_type x) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG2("schedule.size", schedule.size()));
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG2("actions.size", actions.size()));
CAF_ASSERT(x.count() >= 0);
current_time += x;
auto result = size_t{0};
while (!schedule.empty() && schedule.begin()->first <= current_time)
while (!actions.empty() && actions.begin()->first <= current_time)
if (try_trigger_once())
++result;
return result;
}
bool test_actor_clock::try_trigger_once() {
auto i = schedule.begin();
auto t = i->first;
if (t > current_time)
return false;
auto [f, period] = i->second;
schedule.erase(i);
if (f.run() == action::transition::success) {
if (period.count() > 0) {
auto next = t + period;
while (next <= current_time)
next += period;
schedule.emplace(next, schedule_entry{std::move(f), period});
for (;;) {
if (actions.empty())
return false;
auto i = actions.begin();
auto [t, f] = *i;
if (t > current_time)
return false;
actions.erase(i);
if (!f.disposed()) {
f.run();
return true;
}
return true;
} else {
return false;
}
}
......
......@@ -16,11 +16,8 @@ thread_safe_actor_clock::thread_safe_actor_clock() {
tbl_.reserve(buffer_size * 2);
}
disposable
thread_safe_actor_clock::schedule_periodically(time_point first_run, action f,
duration_type period) {
auto ptr = schedule_entry_ptr{new schedule_entry{first_run, f, period}};
queue_.emplace_back(std::move(ptr));
disposable thread_safe_actor_clock::schedule(time_point abs_time, action f) {
queue_.emplace_back(schedule_entry_ptr{new schedule_entry{abs_time, f}});
return std::move(f).as_disposable();
}
......@@ -29,6 +26,7 @@ void thread_safe_actor_clock::run() {
auto is_disposed = [](auto& x) { return !x || x->f.disposed(); };
auto by_timeout = [](auto& x, auto& y) { return x->t < y->t; };
while (running_) {
// Fetch additional scheduling requests from the queue.
if (tbl_.empty()) {
queue_.wait_nonempty();
queue_.get_all(std::back_inserter(tbl_));
......@@ -40,25 +38,16 @@ void thread_safe_actor_clock::run() {
std::sort(tbl_.begin(), tbl_.end(), by_timeout);
}
}
// Run all actions that timed out.
auto n = now();
for (auto i = tbl_.begin(); i != tbl_.end() && (*i)->t <= n; ++i) {
auto& entry = **i;
if (entry.f.run() == action::transition::success) {
if (entry.period.count() > 0) {
auto next = entry.t + entry.period;
while (next <= n) {
CAF_LOG_WARNING("clock lagging behind, skipping a tick!");
next += entry.period;
}
} else {
i->reset(); // Remove from tbl_ after the for-loop body.
}
} else {
i->reset(); // Remove from tbl_ after the for-loop body.
}
}
tbl_.erase(std::remove_if(tbl_.begin(), tbl_.end(), is_disposed),
tbl_.end());
auto i = tbl_.begin();
for (; i != tbl_.end() && (*i)->t <= n; ++i)
(*i)->f.run();
// Here, we have [begin, i) be the actions that were executed. Move any
// already disposed action also to the beginning so that we can erase them
// at once.
i = std::stable_partition(i, tbl_.end(), is_disposed);
tbl_.erase(tbl_.begin(), i);
}
}
......
// 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.
#include "caf/flow/observable_builder.hpp"
#include <limits>
namespace caf::flow {
class interval_action : public ref_counted, public action::impl {
public:
interval_action(intrusive_ptr<interval_impl> impl)
: state_(action::state::scheduled), impl_(std::move(impl)) {
// nop
}
void dispose() override {
state_ = action::state::disposed;
}
bool disposed() const noexcept override {
return state_.load() == action::state::disposed;
}
action::state current_state() const noexcept override {
return state_.load();
}
void run() override {
if (state_.load() == action::state::scheduled)
impl_->fire(this);
}
void ref_disposable() const noexcept override {
ref();
}
void deref_disposable() const noexcept override {
deref();
}
friend void intrusive_ptr_add_ref(const interval_action* ptr) noexcept {
ptr->ref();
}
friend void intrusive_ptr_release(const interval_action* ptr) noexcept {
ptr->deref();
}
private:
std::atomic<action::state> state_;
intrusive_ptr<interval_impl> impl_;
};
interval_impl::interval_impl(coordinator* ctx, timespan initial_delay,
timespan period)
: interval_impl(ctx, initial_delay, period,
std::numeric_limits<int64_t>::max()) {
// nop
}
interval_impl::interval_impl(coordinator* ctx, timespan initial_delay,
timespan period, int64_t max_val)
: ctx_(ctx), initial_delay_(initial_delay), period_(period), max_(max_val) {
CAF_ASSERT(max_val > 0);
}
interval_impl::~interval_impl() {
// nop
}
void interval_impl::dispose() {
if (obs_) {
obs_.on_complete();
obs_ = nullptr;
}
if (pending_) {
pending_.dispose();
pending_ = nullptr;
}
val_ = max_;
}
bool interval_impl::disposed() const noexcept {
return val_ == max_;
}
void interval_impl::ref_disposable() const noexcept {
this->ref();
}
void interval_impl::deref_disposable() const noexcept {
this->deref();
}
coordinator* interval_impl::ctx() const noexcept {
return ctx_;
}
void interval_impl::on_request(observer_impl<int64_t>* ptr, size_t n) {
if (obs_.ptr() == ptr) {
if (demand_ == 0 && !pending_) {
if (val_ == 0)
last_ = ctx_->steady_time() + initial_delay_;
else
last_ = ctx_->steady_time() + period_;
pending_ = ctx_->delay_until(last_,
action{make_counted<interval_action>(this)});
}
demand_ += n;
}
}
void interval_impl::on_cancel(observer_impl<int64_t>* ptr) {
if (obs_.ptr() == ptr) {
obs_ = nullptr;
pending_.dispose();
val_ = max_;
}
}
disposable interval_impl::subscribe(observer<int64_t> sink) {
if (obs_ || val_ == max_) {
sink.on_error(make_error(sec::invalid_observable));
return {};
} else {
obs_ = sink;
return super::do_subscribe(sink.ptr());
}
}
void interval_impl::fire(interval_action* act) {
if (obs_) {
--demand_;
obs_.on_next(make_span(&val_, 1));
if (++val_ == max_) {
obs_.on_complete();
obs_ = nullptr;
pending_ = nullptr;
} else if (demand_ > 0) {
auto now = ctx_->steady_time();
auto next = last_ + period_;
while (next <= now)
next += period_;
last_ = next;
pending_ = ctx_->delay_until(next, action{act});
}
}
}
} // namespace caf::flow
......@@ -6,23 +6,19 @@
namespace caf::flow {
// -- factories ----------------------------------------------------------------
intrusive_ptr<scoped_coordinator> scoped_coordinator::make() {
return {new scoped_coordinator, false};
}
// -- execution ----------------------------------------------------------------
void scoped_coordinator::run() {
auto next = [this](bool blocking) {
std::unique_lock guard{mtx_};
if (blocking) {
while (actions_.empty())
cv_.wait(guard);
} else if (actions_.empty()) {
return action{};
}
auto res = std::move(actions_[0]);
actions_.erase(actions_.begin());
return res;
};
for (;;) {
auto hdl = next(!watched_disposables_.empty());
if (hdl.ptr() != nullptr) {
hdl.run();
auto f = next(!watched_disposables_.empty());
if (f.ptr() != nullptr) {
f.run();
drop_disposed_flows();
} else {
return;
......@@ -30,6 +26,8 @@ void scoped_coordinator::run() {
}
}
// -- reference counting -------------------------------------------------------
void scoped_coordinator::ref_coordinator() const noexcept {
ref();
}
......@@ -38,6 +36,20 @@ void scoped_coordinator::deref_coordinator() const noexcept {
deref();
}
// -- lifetime management ------------------------------------------------------
void scoped_coordinator::watch(disposable what) {
watched_disposables_.emplace_back(std::move(what));
}
// -- time ---------------------------------------------------------------------
coordinator::steady_time_point scoped_coordinator::steady_time() {
return std::chrono::steady_clock::now();
}
// -- scheduling of actions ----------------------------------------------------
void scoped_coordinator::schedule(action what) {
std::unique_lock guard{mtx_};
actions_.emplace_back(std::move(what));
......@@ -45,16 +57,15 @@ void scoped_coordinator::schedule(action what) {
cv_.notify_all();
}
void scoped_coordinator::post_internally(action what) {
void scoped_coordinator::delay(action what) {
schedule(std::move(what));
}
void scoped_coordinator::watch(disposable what) {
watched_disposables_.emplace_back(std::move(what));
}
intrusive_ptr<scoped_coordinator> scoped_coordinator::make() {
return {new scoped_coordinator, false};
disposable scoped_coordinator::delay_until(steady_time_point abs_time,
action what) {
using namespace std::literals;
delayed_.emplace(abs_time, what);
return what.as_disposable();
}
void scoped_coordinator::drop_disposed_flows() {
......@@ -64,4 +75,41 @@ void scoped_coordinator::drop_disposed_flows() {
xs.erase(e, xs.end());
}
// -- queue and schedule access ------------------------------------------------
action scoped_coordinator::next(bool blocking) {
if (!delayed_.empty()) {
auto n = std::chrono::steady_clock::now();
auto i = delayed_.begin();
if (n >= i->first) {
auto res = std::move(i->second);
delayed_.erase(i);
return res;
}
auto tout = i->first;
std::unique_lock guard{mtx_};
while (actions_.empty()) {
if (cv_.wait_until(guard, tout) == std::cv_status::timeout) {
auto res = std::move(i->second);
delayed_.erase(i);
return res;
}
}
auto res = std::move(actions_[0]);
actions_.erase(actions_.begin());
return res;
} else {
std::unique_lock guard{mtx_};
if (blocking) {
while (actions_.empty())
cv_.wait(guard);
} else if (actions_.empty()) {
return action{};
}
auto res = std::move(actions_[0]);
actions_.erase(actions_.begin());
return res;
}
}
} // namespace caf::flow
......@@ -589,6 +589,10 @@ void scheduled_actor::set_stream_timeout(actor_clock::time_point x) {
// -- caf::flow API ------------------------------------------------------------
flow::coordinator::steady_time_point scheduled_actor::steady_time() {
return clock().now();
}
void scheduled_actor::ref_coordinator() const noexcept {
intrusive_ptr_add_ref(ctrl());
}
......@@ -601,10 +605,15 @@ void scheduled_actor::schedule(action what) {
enqueue(nullptr, make_message_id(), make_message(std::move(what)), nullptr);
}
void scheduled_actor::post_internally(action what) {
void scheduled_actor::delay(action what) {
actions_.emplace_back(std::move(what));
}
disposable scheduled_actor::delay_until(steady_time_point abs_time,
action what) {
return clock().schedule(abs_time, std::move(what), strong_actor_ptr{ctrl()});
}
// -- message processing -------------------------------------------------------
void scheduled_actor::add_awaited_response_handler(message_id response_id,
......
......@@ -27,7 +27,6 @@ SCENARIO("actions wrap function calls") {
CHECK(uut.scheduled());
uut.run();
CHECK(called);
CHECK(uut.invoked());
}
}
WHEN("disposing the action") {
......@@ -43,27 +42,13 @@ SCENARIO("actions wrap function calls") {
}
}
WHEN("running the action multiple times") {
THEN("any call after the first becomes a no-op") {
THEN("the action invokes its function that many times") {
auto n = 0;
auto uut = make_action([&n] { ++n; });
uut.run();
uut.run();
uut.run();
CHECK(uut.invoked());
CHECK_EQ(n, 1);
}
}
WHEN("re-scheduling an action after running it") {
THEN("then the lambda gets invoked twice") {
auto n = 0;
auto uut = make_action([&n] { ++n; });
uut.run();
uut.run();
CHECK_EQ(uut.reschedule(), action::transition::success);
uut.run();
uut.run();
CHECK(uut.invoked());
CHECK_EQ(n, 2);
CHECK_EQ(n, 3);
}
}
WHEN("converting an action to a disposable") {
......
......@@ -69,10 +69,10 @@ CAF_TEST(run_delayed without dispose) {
// Have AUT call self->run_delayed().
self->send(aut, ok_atom_v);
expect((ok_atom), from(self).to(aut).with(_));
CAF_CHECK_EQUAL(t.schedule.size(), 1u);
CAF_CHECK_EQUAL(t.actions.size(), 1u);
// Advance time to trigger timeout.
t.advance_time(10s);
CAF_CHECK_EQUAL(t.schedule.size(), 0u);
CAF_CHECK_EQUAL(t.actions.size(), 0u);
// Have AUT receive the action.
expect((action), to(aut));
CAF_CHECK(state().run_delayed_called);
......@@ -83,10 +83,10 @@ CAF_TEST(run_delayed with dispose before expire) {
self->send(aut, ok_atom_v);
expect((ok_atom), from(self).to(aut).with(_));
state().pending.dispose();
CAF_CHECK_EQUAL(t.schedule.size(), 1u);
CAF_CHECK_EQUAL(t.actions.size(), 1u);
// Advance time, but the clock drops the disposed callback.
t.advance_time(10s);
CAF_CHECK_EQUAL(t.schedule.size(), 0u);
CAF_CHECK_EQUAL(t.actions.size(), 0u);
// Have AUT receive the timeout.
disallow((action), to(aut));
CAF_CHECK(!state().run_delayed_called);
......@@ -96,10 +96,10 @@ CAF_TEST(run_delayed with dispose after expire) {
// Have AUT call self->run_delayed().
self->send(aut, ok_atom_v);
expect((ok_atom), from(self).to(aut).with(_));
CAF_CHECK_EQUAL(t.schedule.size(), 1u);
CAF_CHECK_EQUAL(t.actions.size(), 1u);
// Advance time to send timeout message.
t.advance_time(10s);
CAF_CHECK_EQUAL(t.schedule.size(), 0u);
CAF_CHECK_EQUAL(t.actions.size(), 0u);
// Have AUT receive the timeout but dispose it: turns into a nop.
state().pending.dispose();
expect((action), to(aut));
......@@ -113,10 +113,10 @@ CAF_TEST(delay_actor_message) {
t.schedule_message(n, autptr,
make_mailbox_element(autptr, make_message_id(), no_stages,
"foo"));
CAF_CHECK_EQUAL(t.schedule.size(), 1u);
CAF_CHECK_EQUAL(t.actions.size(), 1u);
// Advance time to send the message.
t.advance_time(10s);
CAF_CHECK_EQUAL(t.schedule.size(), 0u);
CAF_CHECK_EQUAL(t.actions.size(), 0u);
// Have AUT receive the message.
expect((std::string), from(aut).to(aut).with("foo"));
}
......@@ -130,10 +130,10 @@ CAF_TEST(delay_group_message) {
auto n = t.now() + 10s;
auto autptr = actor_cast<strong_actor_ptr>(aut);
t.schedule_message(n, std::move(grp), autptr, make_message("foo"));
CAF_CHECK_EQUAL(t.schedule.size(), 1u);
CAF_CHECK_EQUAL(t.actions.size(), 1u);
// Advance time to send the message.
t.advance_time(10s);
CAF_CHECK_EQUAL(t.schedule.size(), 0u);
CAF_CHECK_EQUAL(t.actions.size(), 0u);
// Have AUT receive the message.
expect((std::string), from(aut).to(aut).with("foo"));
// Kill AUT (necessary because the group keeps a reference around).
......
// 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 flow.interval
#include "caf/flow/observable_builder.hpp"
#include "core-test.hpp"
#include "caf/flow/scoped_coordinator.hpp"
using namespace std::literals;
using namespace caf;
namespace {
struct fixture : test_coordinator_fixture<> {
flow::scoped_coordinator_ptr ctx = flow::make_scoped_coordinator();
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(interval_tests, fixture)
SCENARIO("scoped coordinators wait on observable intervals") {
using i64_list = std::vector<int64_t>;
GIVEN("an observable interval") {
WHEN("an observer subscribes to it") {
THEN("the coordinator blocks the current thread for the delays") {
auto outputs = i64_list{};
ctx->make_observable()
.interval(50ms, 25ms)
.take(3)
.for_each([&outputs](int64_t x) { outputs.emplace_back(x); });
ctx->run();
CHECK_EQ(outputs, i64_list({0, 1, 2}));
}
}
}
}
SCENARIO("scheduled actors schedule observable intervals delays") {
using i64_list = std::vector<int64_t>;
GIVEN("an observable interval") {
WHEN("an observer subscribes to it") {
THEN("the actor uses the actor clock to schedule flow processing") {
auto outputs = i64_list{};
sys.spawn([&outputs](caf::event_based_actor* self) {
self->make_observable()
.interval(50ms, 25ms)
.take(3)
.for_each([&outputs](int64_t x) { outputs.emplace_back(x); });
});
CHECK(sched.clock().actions.empty());
sched.run();
CHECK_EQ(sched.clock().actions.size(), 1u);
advance_time(40ms);
sched.run();
CHECK_EQ(outputs, i64_list());
advance_time(10ms);
sched.run();
CHECK_EQ(outputs, i64_list({0}));
advance_time(20ms);
sched.run();
CHECK_EQ(outputs, i64_list({0}));
advance_time(10ms);
sched.run();
CHECK_EQ(outputs, i64_list({0, 1}));
advance_time(20ms);
sched.run();
CHECK_EQ(outputs, i64_list({0, 1, 2}));
run();
CHECK_EQ(outputs, i64_list({0, 1, 2}));
}
}
}
}
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