Commit a1fd7896 authored by Dominik Charousset's avatar Dominik Charousset

Integrate review feedback

parent daff3591
...@@ -57,6 +57,14 @@ public: ...@@ -57,6 +57,14 @@ public:
/// Enqueues a new message wrapped in a `mailbox_element` to the actor. /// Enqueues a new message wrapped in a `mailbox_element` to the actor.
/// This `enqueue` variant allows to define forwarding chains. /// This `enqueue` variant allows to define forwarding chains.
/// @returns `true` if the message has added to the mailbox, `false`
/// otherwise. In the latter case, the actor terminated and the
/// message has been dropped. Once this function returns `false`, it
/// returns `false` for all future invocations.
/// @note The returned value is purely informational and may be used to
/// discard actor handles early. Messages may still get dropped later
/// even if this function returns `true`. In particular when dealing
/// with remote actors.
virtual bool enqueue(mailbox_element_ptr what, execution_unit* host) = 0; virtual bool enqueue(mailbox_element_ptr what, execution_unit* host) = 0;
/// Attaches `ptr` to this actor. The actor will call `ptr->detach(...)` on /// Attaches `ptr` to this actor. The actor will call `ptr->detach(...)` on
......
...@@ -23,6 +23,10 @@ public: ...@@ -23,6 +23,10 @@ public:
virtual ~abstract_channel(); virtual ~abstract_channel();
/// Enqueues a new message without forwarding stack to the channel. /// Enqueues a new message without forwarding stack to the channel.
/// @returns `true` if the message has been dispatches successful, `false`
/// otherwise. In the latter case, the channel has been closed and
/// the message has been dropped. Once this function returns `false`,
/// it returns `false` for all future invocations.
virtual bool enqueue(strong_actor_ptr sender, message_id mid, message content, virtual bool enqueue(strong_actor_ptr sender, message_id mid, message content,
execution_unit* host = nullptr) execution_unit* host = nullptr)
= 0; = 0;
......
...@@ -227,6 +227,8 @@ namespace caf { ...@@ -227,6 +227,8 @@ namespace caf {
/// `action::state::waiting`. /// `action::state::waiting`.
template <class F> template <class F>
action make_action(F f, action::state init_state = action::state::scheduled) { action make_action(F f, action::state init_state = action::state::scheduled) {
CAF_ASSERT(init_state == action::state::scheduled
|| init_state == action::state::waiting);
using impl_t = detail::default_action_impl<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), init_state)};
} }
......
...@@ -67,39 +67,39 @@ public: ...@@ -67,39 +67,39 @@ public:
= 0; = 0;
/// Schedules an action for execution by an actor at a later time. /// Schedules an action for execution by an actor at a later time.
/// @param t The local time at which the action gets enqueued to the mailbox /// @param t The local time at which the action should get enqueued to the
/// of the worker. /// mailbox of the target.
/// @param f The action to schedule. /// @param f The action to schedule.
/// @param worker The actor that should run the action. /// @param target The actor that should run the action.
disposable schedule(time_point t, action f, strong_actor_ptr worker); disposable schedule(time_point t, action f, strong_actor_ptr target);
/// Schedules an action for periodic execution by an actor. /// Schedules an action for periodic execution by an actor.
/// @param first_run The local time at which the action gets enqueued to the /// @param first_run The local time at which the action should get enqueued to
/// mailbox of the worker for the first time. /// the mailbox of the target for the first time.
/// @param f The action to schedule. /// @param f The action to schedule.
/// @param worker The actor that should run the action. /// @param target The actor that should run the action.
/// @param period The time to wait between two messages to the actor. /// @param period The time to wait between two messages to the actor.
/// @param policy The strategy for dealing with a stalling worker. /// @param policy The strategy for dealing with a stalling target.
disposable schedule_periodically(time_point first_run, action f, disposable schedule_periodically(time_point first_run, action f,
strong_actor_ptr worker, strong_actor_ptr target,
duration_type period, stall_policy policy); duration_type period, stall_policy policy);
/// Schedules an action for execution by an actor at a later time. /// Schedules an action for execution by an actor at a later time.
/// @param worker The actor that should run the action. /// @param target The actor that should run the action.
/// @param f The action to schedule. /// @param f The action to schedule.
/// @param t The local time at which the action gets enqueued to the mailbox /// @param t The local time at which the action should get enqueued to the
/// of the worker. /// mailbox of the target.
disposable schedule(time_point t, action f, weak_actor_ptr worker); disposable schedule(time_point t, action f, weak_actor_ptr target);
/// Schedules an action for periodic execution by an actor. /// Schedules an action for periodic execution by an actor.
/// @param worker The actor that should run the action. /// @param first_run The local time at which the action should get enqueued to
/// @param policy The strategy for dealing with a stalling worker. /// the mailbox of the target for the first time.
/// @param f The action to schedule. /// @param f The action to schedule.
/// @param first_run The local time at which the action gets enqueued to the /// @param target The actor that should run the action.
/// mailbox of the worker for the first time.
/// @param period The time to wait between two messages to the actor. /// @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, disposable schedule_periodically(time_point first_run, action f,
weak_actor_ptr worker, duration_type period, weak_actor_ptr target, duration_type period,
stall_policy policy); stall_policy policy);
/// Schedules an arbitrary message to `receiver` for time point `t`. /// Schedules an arbitrary message to `receiver` for time point `t`.
......
...@@ -50,7 +50,7 @@ public: ...@@ -50,7 +50,7 @@ public:
disposable& operator=(disposable&&) noexcept = default; disposable& operator=(disposable&&) noexcept = default;
disposable& operator=(const disposable&) noexcept = default; disposable& operator=(const disposable&) noexcept = default;
/// Combines multiple disposable into a single disposable. The new disposable /// Combines multiple disposables into a single disposable. The new disposable
/// is disposed if all of its elements are disposed. Disposing the composite /// is disposed if all of its elements are disposed. Disposing the composite
/// disposes all elements individually. /// disposes all elements individually.
static disposable make_composite(std::vector<disposable> entries); static disposable make_composite(std::vector<disposable> entries);
......
...@@ -397,10 +397,10 @@ public: ...@@ -397,10 +397,10 @@ public:
// -- timeout management ----------------------------------------------------- // -- timeout management -----------------------------------------------------
/// Requests a new timeout for the current behavior and returns its ID. /// Requests a new timeout for the current behavior.
void set_receive_timeout(); void set_receive_timeout();
/// Requests a new timeout and returns its ID. /// Requests a new timeout.
void set_stream_timeout(actor_clock::time_point x); void set_stream_timeout(actor_clock::time_point x);
// -- message processing ----------------------------------------------------- // -- message processing -----------------------------------------------------
...@@ -571,7 +571,7 @@ public: ...@@ -571,7 +571,7 @@ public:
/// Runs `what` asynchronously at some point after `when`. /// Runs `what` asynchronously at some point after `when`.
/// @param when The local time until the actor waits before invoking the /// @param when The local time until the 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
/// additional wait time. Passing the current time or a past times /// additional wait time. Passing the current time or a past time
/// immediately schedules the action for execution. /// immediately schedules the action for execution.
/// @param what The action to invoke after waiting on the timeout. /// @param what The action to invoke after waiting on the timeout.
/// @returns A @ref disposable that allows the actor to cancel the action. /// @returns A @ref disposable that allows the actor to cancel the action.
......
...@@ -19,7 +19,8 @@ thread_safe_actor_clock::thread_safe_actor_clock() { ...@@ -19,7 +19,8 @@ thread_safe_actor_clock::thread_safe_actor_clock() {
disposable disposable
thread_safe_actor_clock::schedule_periodically(time_point first_run, action f, thread_safe_actor_clock::schedule_periodically(time_point first_run, action f,
duration_type period) { duration_type period) {
queue_.emplace_back(new schedule_entry{first_run, f, period}); auto ptr = schedule_entry_ptr{new schedule_entry{first_run, f, period}};
queue_.emplace_back(std::move(ptr));
return std::move(f).as_disposable(); return std::move(f).as_disposable();
} }
......
...@@ -66,20 +66,20 @@ struct fixture : test_coordinator_fixture<> { ...@@ -66,20 +66,20 @@ struct fixture : test_coordinator_fixture<> {
CAF_TEST_FIXTURE_SCOPE(timer_tests, fixture) CAF_TEST_FIXTURE_SCOPE(timer_tests, fixture)
CAF_TEST(run_delayed without dispose) { CAF_TEST(run_delayed without dispose) {
// Have AUT call t.set_receive_timeout(). // Have AUT call self->run_delayed().
self->send(aut, ok_atom_v); self->send(aut, ok_atom_v);
expect((ok_atom), from(self).to(aut).with(_)); expect((ok_atom), from(self).to(aut).with(_));
CAF_CHECK_EQUAL(t.schedule.size(), 1u); CAF_CHECK_EQUAL(t.schedule.size(), 1u);
// Advance time to send timeout message. // Advance time to trigger timeout.
t.advance_time(10s); t.advance_time(10s);
CAF_CHECK_EQUAL(t.schedule.size(), 0u); CAF_CHECK_EQUAL(t.schedule.size(), 0u);
// Have AUT receive the timeout. // Have AUT receive the action.
expect((action), to(aut)); expect((action), to(aut));
CAF_CHECK(state().run_delayed_called); CAF_CHECK(state().run_delayed_called);
} }
CAF_TEST(run_delayed with dispose before expire) { CAF_TEST(run_delayed with dispose before expire) {
// Have AUT call t.set_receive_timeout(). // Have AUT call self->run_delayed().
self->send(aut, ok_atom_v); self->send(aut, ok_atom_v);
expect((ok_atom), from(self).to(aut).with(_)); expect((ok_atom), from(self).to(aut).with(_));
state().pending.dispose(); state().pending.dispose();
...@@ -93,7 +93,7 @@ CAF_TEST(run_delayed with dispose before expire) { ...@@ -93,7 +93,7 @@ CAF_TEST(run_delayed with dispose before expire) {
} }
CAF_TEST(run_delayed with dispose after expire) { CAF_TEST(run_delayed with dispose after expire) {
// Have AUT call t.set_receive_timeout(). // Have AUT call self->run_delayed().
self->send(aut, ok_atom_v); self->send(aut, ok_atom_v);
expect((ok_atom), from(self).to(aut).with(_)); expect((ok_atom), from(self).to(aut).with(_));
CAF_CHECK_EQUAL(t.schedule.size(), 1u); CAF_CHECK_EQUAL(t.schedule.size(), 1u);
......
...@@ -99,19 +99,3 @@ SCENARIO("an actor system shuts down after the last actor terminates") { ...@@ -99,19 +99,3 @@ SCENARIO("an actor system shuts down after the last actor terminates") {
} }
} }
} }
// CAF_TEST(shutdown_delayed_send_loop) {
// CAF_MESSAGE("does sys shut down after spawning a detached actor that used "
// "a delayed send loop and was interrupted via exit message?");
// auto f = [](event_based_actor* self) -> behavior {
// self->delayed_send(self, 1ns, ok_atom_v);
// return {
// [=](ok_atom) {
// self->delayed_send(self, 1ns, ok_atom_v);
// },
// };
// };
// auto a = sys.spawn<detached>(f);
// auto g = detail::make_scope_guard(
// [&] { self->send_exit(a, exit_reason::user_shutdown); });
// }
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