Commit 2f308ccb authored by Dominik Charousset's avatar Dominik Charousset

Use new clock abstraction for timeouts and delays

parent 6d67d620
...@@ -110,6 +110,7 @@ set (LIBCAF_CORE_SRCS ...@@ -110,6 +110,7 @@ set (LIBCAF_CORE_SRCS
src/terminal_stream_scatterer.cpp src/terminal_stream_scatterer.cpp
src/test_actor_clock.cpp src/test_actor_clock.cpp
src/test_coordinator.cpp src/test_coordinator.cpp
src/thread_safe_actor_clock.cpp
src/timestamp.cpp src/timestamp.cpp
src/try_match.cpp src/try_match.cpp
src/type_erased_tuple.cpp src/type_erased_tuple.cpp
......
...@@ -32,6 +32,7 @@ ...@@ -32,6 +32,7 @@
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/make_actor.hpp" #include "caf/make_actor.hpp"
#include "caf/actor_clock.hpp"
#include "caf/infer_handle.hpp" #include "caf/infer_handle.hpp"
#include "caf/actor_config.hpp" #include "caf/actor_config.hpp"
#include "caf/spawn_options.hpp" #include "caf/spawn_options.hpp"
...@@ -475,10 +476,13 @@ public: ...@@ -475,10 +476,13 @@ public:
} }
/// Returns the configuration of this actor system. /// Returns the configuration of this actor system.
const actor_system_config& config() const { inline const actor_system_config& config() const {
return cfg_; return cfg_;
} }
/// Returns the system-wide clock.
actor_clock& clock() noexcept;
/// @cond PRIVATE /// @cond PRIVATE
/// Increases running-detached-threads-count by one. /// Increases running-detached-threads-count by one.
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_DETAIL_THREAD_SAFE_ACTOR_CLOCK_HPP
#define CAF_DETAIL_THREAD_SAFE_ACTOR_CLOCK_HPP
#include <mutex>
#include <atomic>
#include <condition_variable>
#include "caf/detail/simple_actor_clock.hpp"
namespace caf {
namespace detail {
class thread_safe_actor_clock : public simple_actor_clock {
public:
using super = simple_actor_clock;
thread_safe_actor_clock();
void set_receive_timeout(time_point t, abstract_actor* self,
uint32_t id) override;
void set_request_timeout(time_point t, abstract_actor* self,
message_id id) override;
void cancel_receive_timeout(abstract_actor* self) override;
void cancel_request_timeout(abstract_actor* self, message_id id) override;
void cancel_timeouts(abstract_actor* self) override;
void schedule_message(time_point t, strong_actor_ptr receiver,
mailbox_element_ptr content) override;
void schedule_message(time_point t, group target, strong_actor_ptr sender,
message content) override;
void run_dispatch_loop();
void cancel_dispatch_loop();
private:
std::mutex mx_;
std::condition_variable cv_;
std::atomic<bool> done_;
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_THREAD_SAFE_ACTOR_CLOCK_HPP
...@@ -206,6 +206,11 @@ public: ...@@ -206,6 +206,11 @@ public:
return context_->system(); return context_->system();
} }
/// Returns the clock of the actor system.
inline actor_clock& clock() const {
return home_system().clock();
}
/// @cond PRIVATE /// @cond PRIVATE
void monitor(abstract_actor* ptr); void monitor(abstract_actor* ptr);
......
...@@ -26,6 +26,7 @@ ...@@ -26,6 +26,7 @@
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/duration.hpp" #include "caf/duration.hpp"
#include "caf/no_stages.hpp"
#include "caf/response_type.hpp" #include "caf/response_type.hpp"
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/message_priority.hpp" #include "caf/message_priority.hpp"
...@@ -100,9 +101,10 @@ public: ...@@ -100,9 +101,10 @@ public:
dptr()->context(), std::forward<Ts>(xs)...); dptr()->context(), std::forward<Ts>(xs)...);
} }
template <message_priority P = message_priority::normal, template <message_priority P = message_priority::normal, class Rep = int,
class Dest = actor, class... Ts> class Period = std::ratio<1>, class Dest = actor, class... Ts>
void delayed_send(const Dest& dest, const duration& rtime, Ts&&... xs) { void delayed_send(const Dest& dest, std::chrono::duration<Rep, Period> rtime,
Ts&&... xs) {
using token = using token =
detail::type_list< detail::type_list<
typename detail::implicit_conversions< typename detail::implicit_conversions<
...@@ -136,23 +138,21 @@ public: ...@@ -136,23 +138,21 @@ public:
>::type >::type
>::valid, >::valid,
"this actor does not accept the response message"); "this actor does not accept the response message");
if (dest) if (dest) {
dptr()->system().scheduler().delayed_send( auto& clock = dptr()->system().clock();
rtime, dptr()->ctrl(), actor_cast<strong_actor_ptr>(dest), auto t = clock.now() + rtime;
message_id::make(P), make_message(std::forward<Ts>(xs)...)); auto me = make_mailbox_element(dptr()->ctrl(), message_id::make(P),
no_stages, std::forward<Ts>(xs)...);
clock.schedule_message(t, actor_cast<strong_actor_ptr>(dest),
std::move(me));
}
} }
template <message_priority P = message_priority::normal, template <message_priority P = message_priority::normal, class Rep = int,
class Rep = int, class Period = std::ratio<1>, class Period = std::ratio<1>, class Source = actor,
class Dest = actor, class... Ts> class Dest = actor, class... Ts>
void delayed_send(const Dest& dest, std::chrono::duration<Rep, Period> rtime, void delayed_anon_send(const Dest& dest,
Ts&&... xs) { std::chrono::duration<Rep, Period> rtime, Ts&&... xs) {
delayed_send(dest, duration{rtime}, std::forward<Ts>(xs)...);
}
template <message_priority P = message_priority::normal,
class Source = actor, class Dest = actor, class... Ts>
void delayed_anon_send(const Dest& dest, const duration& rtime, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
using token = using token =
detail::type_list< detail::type_list<
...@@ -164,19 +164,14 @@ public: ...@@ -164,19 +164,14 @@ public:
token token
>::valid, >::valid,
"receiver does not accept given message"); "receiver does not accept given message");
if (dest) if (dest) {
dptr()->system().scheduler().delayed_send( auto& clock = dptr()->system().clock();
rtime, nullptr, actor_cast<strong_actor_ptr>(dest), message_id::make(P), auto t = clock.now() + rtime;
make_message(std::forward<Ts>(xs)...)); auto me = make_mailbox_element(nullptr, message_id::make(P), no_stages,
} std::forward<Ts>(xs)...);
clock.schedule_message(t, actor_cast<strong_actor_ptr>(dest),
template <message_priority P = message_priority::normal, std::move(me));
class Rep = int, class Period = std::ratio<1>, }
class Source = actor, class Dest = actor, class... Ts>
void delayed_anon_send(const Dest& dest,
std::chrono::duration<Rep, Period> rtime,
Ts&&... xs) {
delayed_anon_send(dest, duration{rtime}, std::forward<Ts>(xs)...);
} }
private: private:
......
...@@ -30,6 +30,7 @@ ...@@ -30,6 +30,7 @@
#include "caf/duration.hpp" #include "caf/duration.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/actor_clock.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
namespace caf { namespace caf {
...@@ -43,7 +44,6 @@ class abstract_coordinator : public actor_system::module { ...@@ -43,7 +44,6 @@ class abstract_coordinator : public actor_system::module {
public: public:
enum utility_actor_id : size_t { enum utility_actor_id : size_t {
printer_id, printer_id,
timer_id,
max_id max_id
}; };
...@@ -54,11 +54,6 @@ public: ...@@ -54,11 +54,6 @@ public:
return actor_cast<actor>(utility_actors_[printer_id]); return actor_cast<actor>(utility_actors_[printer_id]);
} }
/// Returns a handle to the central timer actor.
inline actor timer() const {
return actor_cast<actor>(utility_actors_[timer_id]);
}
/// Returns the number of utility actors. /// Returns the number of utility actors.
inline size_t num_utility_actors() const { inline size_t num_utility_actors() const {
return utility_actors_.size(); return utility_actors_.size();
...@@ -67,16 +62,6 @@ public: ...@@ -67,16 +62,6 @@ public:
/// Puts `what` into the queue of a randomly chosen worker. /// Puts `what` into the queue of a randomly chosen worker.
virtual void enqueue(resumable* what) = 0; virtual void enqueue(resumable* what) = 0;
template <class Duration, class... Data>
void delayed_send(Duration rel_time, strong_actor_ptr from,
strong_actor_ptr to, message_id mid, message data) {
auto& dest = utility_actors_[timer_id];
dest->enqueue(nullptr, invalid_message_id,
make_message(duration{rel_time}, std::move(from),
std::move(to), mid, std::move(data)),
nullptr);
}
inline actor_system& system() { inline actor_system& system() {
return system_; return system_;
} }
...@@ -102,21 +87,26 @@ public: ...@@ -102,21 +87,26 @@ public:
static void cleanup_and_release(resumable*); static void cleanup_and_release(resumable*);
virtual actor_clock& clock() noexcept = 0;
protected: protected:
void stop_actors(); void stop_actors();
// ID of the worker receiving the next enqueue /// ID of the worker receiving the next enqueue (round-robin dispatch).
std::atomic<size_t> next_worker_; std::atomic<size_t> next_worker_;
// number of messages each actor is allowed to consume per resume /// Number of messages each actor is allowed to consume per resume.
size_t max_throughput_; size_t max_throughput_;
// configured number of workers /// Configured number of workers.
size_t num_workers_; size_t num_workers_;
/// Background workers, e.g., printer.
std::array<actor, max_id> utility_actors_; std::array<actor, max_id> utility_actors_;
/// Reference to the host system.
actor_system& system_; actor_system& system_;
}; };
} // namespace scheduler } // namespace scheduler
......
...@@ -29,6 +29,8 @@ ...@@ -29,6 +29,8 @@
#include "caf/scheduler/worker.hpp" #include "caf/scheduler/worker.hpp"
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/detail/thread_safe_actor_clock.hpp"
namespace caf { namespace caf {
namespace scheduler { namespace scheduler {
...@@ -68,6 +70,10 @@ protected: ...@@ -68,6 +70,10 @@ protected:
// start all workers now that all workers have been initialized // start all workers now that all workers have been initialized
for (auto& w : workers_) for (auto& w : workers_)
w->start(); w->start();
// launch thread for dispatching timeouts and delayed messages
timer_ = std::thread{[&] {
clock_.run_dispatch_loop();
}};
// run remaining startup code // run remaining startup code
super::start(); super::start();
} }
...@@ -127,19 +133,34 @@ protected: ...@@ -127,19 +133,34 @@ protected:
for (auto& w : workers_) for (auto& w : workers_)
policy_.foreach_resumable(w.get(), f); policy_.foreach_resumable(w.get(), f);
policy_.foreach_central_resumable(this, f); policy_.foreach_central_resumable(this, f);
// stop timer thread
clock_.cancel_dispatch_loop();
timer_.join();
} }
void enqueue(resumable* ptr) override { void enqueue(resumable* ptr) override {
policy_.central_enqueue(this, ptr); policy_.central_enqueue(this, ptr);
} }
detail::thread_safe_actor_clock& clock() noexcept override {
return clock_;
}
private: private:
// usually of size std::thread::hardware_concurrency() /// System-wide clock.
detail::thread_safe_actor_clock clock_;
/// Set of workers.
std::vector<std::unique_ptr<worker_type>> workers_; std::vector<std::unique_ptr<worker_type>> workers_;
// policy-specific data
/// Policy-specific data.
policy_data data_; policy_data data_;
// instance of our policy object
/// The policy object.
Policy policy_; Policy policy_;
/// Thread for managing timeouts and delayed messages.
std::thread timer_;
}; };
} // namespace scheduler } // namespace scheduler
......
...@@ -30,6 +30,8 @@ ...@@ -30,6 +30,8 @@
#include "caf/scheduled_actor.hpp" #include "caf/scheduled_actor.hpp"
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/scheduler/abstract_coordinator.hpp"
#include "caf/detail/test_actor_clock.hpp"
namespace caf { namespace caf {
namespace scheduler { namespace scheduler {
...@@ -142,6 +144,8 @@ public: ...@@ -142,6 +144,8 @@ public:
bool detaches_utility_actors() const override; bool detaches_utility_actors() const override;
detail::test_actor_clock& clock() noexcept override;
protected: protected:
void start() override; void start() override;
...@@ -152,6 +156,9 @@ protected: ...@@ -152,6 +156,9 @@ protected:
private: private:
void inline_all_enqueues_helper(); void inline_all_enqueues_helper();
/// Allows users to fake time at will.
detail::test_actor_clock clock_;
/// User-provided callback for triggering custom code in `enqueue`. /// User-provided callback for triggering custom code in `enqueue`.
std::function<void()> after_next_enqueue_; std::function<void()> after_next_enqueue_;
}; };
......
...@@ -53,85 +53,6 @@ namespace scheduler { ...@@ -53,85 +53,6 @@ namespace scheduler {
namespace { namespace {
using hrc = std::chrono::high_resolution_clock;
class timer_actor : public blocking_actor {
public:
explicit timer_actor(actor_config& cfg) : blocking_actor(cfg) {
// nop
}
struct delayed_msg {
strong_actor_ptr from;
strong_actor_ptr to;
message_id mid;
message msg;
};
void deliver(delayed_msg& dm) {
dm.to->enqueue(dm.from, dm.mid, std::move(dm.msg), nullptr);
}
template <class Map, class... Ts>
void insert_dmsg(Map& storage, const duration& d, Ts&&... xs) {
auto tout = hrc::now();
tout += d;
delayed_msg dmsg{std::forward<Ts>(xs)...};
storage.emplace(std::move(tout), std::move(dmsg));
}
void act() override {
// local state
accept_one_cond rc;
bool running = true;
std::multimap<hrc::time_point, delayed_msg> messages;
// our message handler
behavior nested{
[&](const duration& d, strong_actor_ptr& from,
strong_actor_ptr& to, message_id mid, message& msg) {
insert_dmsg(messages, d, std::move(from),
std::move(to), mid, std::move(msg));
},
[&](const exit_msg& dm) {
if (dm.reason) {
fail_state(dm.reason);
running = false;
}
}
};
auto bhvr = detail::make_blocking_behavior(
&nested,
others >> [&](message_view& x) -> result<message> {
std::cerr << "*** unexpected message in timer_actor: "
<< to_string(x.content()) << std::endl;
return sec::unexpected_message;
}
);
// loop until receiving an exit message
while (running) {
if (messages.empty()) {
// use regular receive as long as we don't have a pending timeout
receive_impl(rc, message_id::make(), bhvr);
} else {
auto tout = messages.begin()->first;
if (await_data(tout)) {
receive_impl(rc, message_id::make(), bhvr);
} else {
auto it = messages.begin();
while (it != messages.end() && (it->first) <= tout) {
deliver(it->second);
it = messages.erase(it);
}
}
}
}
}
const char* name() const override {
return "timer_actor";
}
};
using string_sink = std::function<void (std::string&&)>; using string_sink = std::function<void (std::string&&)>;
// the first value is the use count, the last ostream_handle that // the first value is the use count, the last ostream_handle that
...@@ -322,7 +243,6 @@ void abstract_coordinator::start() { ...@@ -322,7 +243,6 @@ void abstract_coordinator::start() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// launch utility actors // launch utility actors
static constexpr auto fs = hidden + detached; static constexpr auto fs = hidden + detached;
utility_actors_[timer_id] = system_.spawn<timer_actor, fs>();
utility_actors_[printer_id] = system_.spawn<printer_actor, fs>(); utility_actors_[printer_id] = system_.spawn<printer_actor, fs>();
} }
......
...@@ -221,11 +221,13 @@ actor_system::actor_system(actor_system_config& cfg) ...@@ -221,11 +221,13 @@ actor_system::actor_system(actor_system_config& cfg)
modules_[mod_ptr->id()].reset(mod_ptr); modules_[mod_ptr->id()].reset(mod_ptr);
} }
auto& sched = modules_[module::scheduler]; auto& sched = modules_[module::scheduler];
using test = scheduler::test_coordinator; using namespace scheduler;
using share = scheduler::coordinator<policy::work_sharing>; using policy::work_sharing;
using steal = scheduler::coordinator<policy::work_stealing>; using policy::work_stealing;
using profiled_share = scheduler::profiled_coordinator<policy::profiled<policy::work_sharing>>; using share = coordinator<work_sharing>;
using profiled_steal = scheduler::profiled_coordinator<policy::profiled<policy::work_stealing>>; using steal = coordinator<work_stealing>;
using profiled_share = profiled_coordinator<policy::profiled<work_sharing>>;
using profiled_steal = profiled_coordinator<policy::profiled<work_stealing>>;
// set scheduler only if not explicitly loaded by user // set scheduler only if not explicitly loaded by user
if (!sched) { if (!sched) {
enum sched_conf { enum sched_conf {
...@@ -262,7 +264,7 @@ actor_system::actor_system(actor_system_config& cfg) ...@@ -262,7 +264,7 @@ actor_system::actor_system(actor_system_config& cfg)
sched.reset(new profiled_share(*this)); sched.reset(new profiled_share(*this));
break; break;
case testing: case testing:
sched.reset(new test(*this)); sched.reset(new test_coordinator(*this));
} }
} }
// initialize state for each module and give each module the opportunity // initialize state for each module and give each module the opportunity
...@@ -402,6 +404,11 @@ void actor_system::await_all_actors_done() const { ...@@ -402,6 +404,11 @@ void actor_system::await_all_actors_done() const {
registry_.await_running_count_equal(0); registry_.await_running_count_equal(0);
} }
actor_clock& actor_system::clock() noexcept {
return scheduler().clock();
}
void actor_system::inc_detached_threads() { void actor_system::inc_detached_threads() {
++detached; ++detached;
} }
......
...@@ -65,8 +65,9 @@ void local_actor::request_response_timeout(const duration& d, message_id mid) { ...@@ -65,8 +65,9 @@ void local_actor::request_response_timeout(const duration& d, message_id mid) {
CAF_LOG_TRACE(CAF_ARG(d) << CAF_ARG(mid)); CAF_LOG_TRACE(CAF_ARG(d) << CAF_ARG(mid));
if (!d.valid()) if (!d.valid())
return; return;
system().scheduler().delayed_send(d, ctrl(), ctrl(), mid.response_id(), auto t = clock().now();
make_message(sec::request_timeout)); t += d;
clock().set_request_timeout(t, this, mid.response_id());
} }
void local_actor::monitor(abstract_actor* ptr) { void local_actor::monitor(abstract_actor* ptr) {
...@@ -185,6 +186,7 @@ bool local_actor::cleanup(error&& fail_state, execution_unit* host) { ...@@ -185,6 +186,7 @@ bool local_actor::cleanup(error&& fail_state, execution_unit* host) {
// tell registry we're done // tell registry we're done
unregister_from_system(); unregister_from_system();
monitorable_actor::cleanup(std::move(fail_state), host); monitorable_actor::cleanup(std::move(fail_state), host);
clock().cancel_timeouts(this);
CAF_LOG_TERMINATE_EVENT(this, fail_state); CAF_LOG_TERMINATE_EVENT(this, fail_state);
return true; return true;
} }
......
...@@ -297,14 +297,16 @@ uint32_t scheduled_actor::request_timeout(const duration& d) { ...@@ -297,14 +297,16 @@ uint32_t scheduled_actor::request_timeout(const duration& d) {
} }
setf(has_timeout_flag); setf(has_timeout_flag);
auto result = ++timeout_id_; auto result = ++timeout_id_;
auto msg = make_message(timeout_msg{++timeout_id_}); auto msg = make_message(timeout_msg{result});
CAF_LOG_TRACE("send new timeout_msg, " << CAF_ARG(timeout_id_)); CAF_LOG_TRACE("send new timeout_msg, " << CAF_ARG(timeout_id_));
if (d.is_zero()) if (d.is_zero()) {
// immediately enqueue timeout message if duration == 0s // immediately enqueue timeout message if duration == 0s
enqueue(ctrl(), invalid_message_id, std::move(msg), context()); enqueue(ctrl(), invalid_message_id, std::move(msg), context());
else } else {
system().scheduler().delayed_send(d, ctrl(), strong_actor_ptr(ctrl()), auto t = clock().now();
message_id::make(), std::move(msg)); t += d;
clock().set_receive_timeout(t, this, result);
}
return result; return result;
} }
......
...@@ -99,12 +99,14 @@ bool test_coordinator::detaches_utility_actors() const { ...@@ -99,12 +99,14 @@ bool test_coordinator::detaches_utility_actors() const {
return false; return false;
} }
detail::test_actor_clock& test_coordinator::clock() noexcept {
return clock_;
}
void test_coordinator::start() { void test_coordinator::start() {
dummy_worker worker{this}; dummy_worker worker{this};
actor_config cfg{&worker}; actor_config cfg{&worker};
auto& sys = system(); auto& sys = system();
utility_actors_[timer_id] = make_actor<dummy_timer, actor>(
sys.next_actor_id(), sys.node(), &sys, cfg, this);
utility_actors_[printer_id] = make_actor<dummy_printer, actor>( utility_actors_[printer_id] = make_actor<dummy_printer, actor>(
sys.next_actor_id(), sys.node(), &sys, cfg); sys.next_actor_id(), sys.node(), &sys, cfg);
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/detail/thread_safe_actor_clock.hpp"
namespace caf {
namespace detail {
namespace {
using guard_type = std::unique_lock<std::mutex>;
} // namespace <anonymous>
thread_safe_actor_clock::thread_safe_actor_clock() : done_(false) {
// nop
}
void thread_safe_actor_clock::set_receive_timeout(time_point t,
abstract_actor* self,
uint32_t id) {
guard_type guard{mx_};
super::set_receive_timeout(t, self, id);
cv_.notify_all();
}
void thread_safe_actor_clock::set_request_timeout(time_point t,
abstract_actor* self,
message_id id) {
guard_type guard{mx_};
super::set_request_timeout(t, self, id);
cv_.notify_all();
}
void thread_safe_actor_clock::cancel_receive_timeout(abstract_actor* self) {
guard_type guard{mx_};
super::cancel_receive_timeout(self);
cv_.notify_all();
}
void thread_safe_actor_clock::cancel_request_timeout(abstract_actor* self,
message_id id) {
guard_type guard{mx_};
super::cancel_request_timeout(self, id);
cv_.notify_all();
}
void thread_safe_actor_clock::cancel_timeouts(abstract_actor* self) {
guard_type guard{mx_};
super::cancel_timeouts(self);
cv_.notify_all();
}
void thread_safe_actor_clock::schedule_message(time_point t,
strong_actor_ptr receiver,
mailbox_element_ptr content) {
guard_type guard{mx_};
super::schedule_message(t, std::move(receiver), std::move(content));
cv_.notify_all();
}
void thread_safe_actor_clock::schedule_message(time_point t, group target,
strong_actor_ptr sender,
message content) {
guard_type guard{mx_};
super::schedule_message(t, std::move(target), std::move(sender),
std::move(content));
cv_.notify_all();
}
void thread_safe_actor_clock::run_dispatch_loop() {
visitor f{this};
guard_type guard{mx_};
while (done_ == false) {
// Wait for non-empty schedule.
if (schedule_.empty()) {
cv_.wait(guard);
} else {
auto tout = schedule_.begin()->first;
cv_.wait_until(guard, tout);
}
// Double-check whether schedule is non-empty and execute it.
if (!schedule_.empty()) {
auto t = now();
auto i = schedule_.begin();
while (i != schedule_.end() && i->first <= t) {
visit(f, i->second);
i = schedule_.erase(i);
}
}
}
}
void thread_safe_actor_clock::cancel_dispatch_loop() {
guard_type guard{mx_};
done_ = true;
cv_.notify_all();
}
} // namespace detail
} // namespace caf
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