Commit 8e28cda4 authored by Dominik Charousset's avatar Dominik Charousset

Reimplement thread-safe actor clock

parent d49879bb
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 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. *
******************************************************************************/
#pragma once
#include <memory>
namespace caf {
namespace detail {
template <class T, class... Ts>
std::unique_ptr<T> make_unique(Ts&&... xs) {
return std::unique_ptr<T>{new T(std::forward<Ts>(xs)...)};
}
} // namespace detail
} // namespace caf
......@@ -51,10 +51,6 @@ public:
/// @returns The number of triggered timeouts.
size_t trigger_timeouts();
/// Triggers all timeouts with timestamp <= now.
/// @returns The number of triggered timeouts.
size_t trigger_expired_timeouts();
/// Advances the time by `x` and dispatches timeouts and delayed messages.
/// @returns The number of triggered timeouts.
size_t advance_time(duration_type x);
......
......@@ -18,10 +18,14 @@
#pragma once
#include <mutex>
#include <array>
#include <atomic>
#include <condition_variable>
#include <memory>
#include <mutex>
#include "caf/abstract_actor.hpp"
#include "caf/detail/ringbuffer.hpp"
#include "caf/detail/simple_actor_clock.hpp"
namespace caf {
......@@ -29,9 +33,15 @@ namespace detail {
class thread_safe_actor_clock : public simple_actor_clock {
public:
// -- constants --------------------------------------------------------------
static constexpr size_t buffer_size = 64;
// -- member types -----------------------------------------------------------
using super = simple_actor_clock;
thread_safe_actor_clock();
// -- member functions -------------------------------------------------------
void set_ordinary_timeout(time_point t, abstract_actor* self,
atom_value type, uint64_t id) override;
......@@ -61,11 +71,14 @@ public:
void cancel_dispatch_loop();
private:
std::recursive_mutex mx_;
std::condition_variable_any cv_;
std::atomic<bool> done_;
void push(event* ptr);
/// Receives timer events from other threads.
detail::ringbuffer<unique_event_ptr, buffer_size> queue_;
/// Locally caches events for processing.
std::array<unique_event_ptr, buffer_size> events_;
};
} // namespace detail
} // namespace caf
......@@ -25,108 +25,40 @@
namespace caf {
namespace detail {
bool simple_actor_clock::ordinary_predicate::
operator()(const secondary_map::value_type& x) const noexcept {
auto ptr = get_if<ordinary_timeout>(&x.second->second);
return ptr != nullptr ? ptr->type == type : false;
simple_actor_clock::event::~event() {
// nop
}
bool simple_actor_clock::multi_predicate::
operator()(const secondary_map::value_type& x) const noexcept {
auto ptr = get_if<multi_timeout>(&x.second->second);
return ptr != nullptr ? ptr->type == type : false;
}
bool simple_actor_clock::request_predicate::
operator()(const secondary_map::value_type& x) const noexcept {
auto ptr = get_if<request_timeout>(&x.second->second);
return ptr != nullptr ? ptr->id == id : false;
}
void simple_actor_clock::visitor::operator()(ordinary_timeout& x) {
CAF_ASSERT(x.self != nullptr);
x.self->get()->eq_impl(make_message_id(), x.self, nullptr,
timeout_msg{x.type, x.id});
ordinary_predicate pred{x.type};
thisptr->drop_lookup(x.self->get(), pred);
}
void simple_actor_clock::visitor::operator()(multi_timeout& x) {
CAF_ASSERT(x.self != nullptr);
x.self->get()->eq_impl(make_message_id(), x.self, nullptr,
timeout_msg{x.type, x.id});
multi_predicate pred{x.type};
thisptr->drop_lookup(x.self->get(), pred);
}
void simple_actor_clock::visitor::operator()(request_timeout& x) {
CAF_ASSERT(x.self != nullptr);
x.self->get()->eq_impl(x.id, x.self, nullptr, sec::request_timeout);
request_predicate pred{x.id};
thisptr->drop_lookup(x.self->get(), pred);
}
void simple_actor_clock::visitor::operator()(actor_msg& x) {
x.receiver->enqueue(std::move(x.content), nullptr);
}
void simple_actor_clock::visitor::operator()(group_msg& x) {
x.target->eq_impl(make_message_id(), std::move(x.sender), nullptr,
std::move(x.content));
}
void simple_actor_clock::set_ordinary_timeout(time_point t, abstract_actor* self,
void simple_actor_clock::set_ordinary_timeout(time_point t,
abstract_actor* self,
atom_value type, uint64_t id) {
ordinary_predicate pred{type};
auto i = lookup(self, pred);
auto sptr = actor_cast<strong_actor_ptr>(self);
ordinary_timeout tmp{std::move(sptr), type, id};
if (i != actor_lookup_.end()) {
schedule_.erase(i->second);
i->second = schedule_.emplace(t, std::move(tmp));
} else {
auto j = schedule_.emplace(t, std::move(tmp));
actor_lookup_.emplace(self, j);
}
new_schedule_entry<ordinary_timeout>(t, self->ctrl(), type, id);
}
void simple_actor_clock::set_multi_timeout(time_point t, abstract_actor* self,
atom_value type, uint64_t id) {
auto sptr = actor_cast<strong_actor_ptr>(self);
multi_timeout tmp{std::move(sptr), type, id};
auto j = schedule_.emplace(t, std::move(tmp));
actor_lookup_.emplace(self, j);
new_schedule_entry<multi_timeout>(t, self->ctrl(), type, id);
}
void simple_actor_clock::set_request_timeout(time_point t, abstract_actor* self,
message_id id) {
request_predicate pred{id};
auto i = lookup(self, pred);
auto sptr = actor_cast<strong_actor_ptr>(self);
request_timeout tmp{std::move(sptr), id};
if (i != actor_lookup_.end()) {
schedule_.erase(i->second);
i->second = schedule_.emplace(t, std::move(tmp));
} else {
auto j = schedule_.emplace(t, std::move(tmp));
actor_lookup_.emplace(self, j);
}
new_schedule_entry<request_timeout>(t, self->ctrl(), id);
}
void simple_actor_clock::cancel_ordinary_timeout(abstract_actor* self,
atom_value type) {
ordinary_predicate pred{type};
cancel(self, pred);
ordinary_timeout_cancellation tmp{self->id(), type};
handle(tmp);
}
void simple_actor_clock::cancel_request_timeout(abstract_actor* self,
message_id id) {
request_predicate pred{id};
cancel(self, pred);
request_timeout_cancellation tmp{self->id(), id};
handle(tmp);
}
void simple_actor_clock::cancel_timeouts(abstract_actor* self) {
auto range = actor_lookup_.equal_range(self);
auto range = actor_lookup_.equal_range(self->id());
if (range.first == range.second)
return;
for (auto i = range.first; i != range.second; ++i)
......@@ -137,14 +69,14 @@ void simple_actor_clock::cancel_timeouts(abstract_actor* self) {
void simple_actor_clock::schedule_message(time_point t,
strong_actor_ptr receiver,
mailbox_element_ptr content) {
schedule_.emplace(t, actor_msg{std::move(receiver), std::move(content)});
new_schedule_entry<actor_msg>(t, std::move(receiver), std::move(content));
}
void simple_actor_clock::schedule_message(time_point t, group target,
strong_actor_ptr sender,
message content) {
schedule_.emplace(
t, group_msg{std::move(target), std::move(sender), std::move(content)});
new_schedule_entry<group_msg>(t, std::move(target), std::move(sender),
std::move(content));
}
void simple_actor_clock::cancel_all() {
......@@ -152,5 +84,119 @@ void simple_actor_clock::cancel_all() {
schedule_.clear();
}
void simple_actor_clock::ship(delayed_event& x) {
switch (x.subtype) {
case ordinary_timeout_type: {
auto& dref = static_cast<ordinary_timeout&>(x);
auto& self = dref.self;
self->get()->eq_impl(make_message_id(), self, nullptr,
timeout_msg{dref.type, dref.id});
break;
}
case multi_timeout_type: {
auto& dref = static_cast<ordinary_timeout&>(x);
auto& self = dref.self;
self->get()->eq_impl(make_message_id(), self, nullptr,
timeout_msg{dref.type, dref.id});
break;
}
case request_timeout_type: {
auto& dref = static_cast<request_timeout&>(x);
auto& self = dref.self;
self->get()->eq_impl(dref.id, self, nullptr, sec::request_timeout);
break;
}
case actor_msg_type: {
auto& dref = static_cast<actor_msg&>(x);
dref.receiver->enqueue(std::move(dref.content), nullptr);
break;
}
case group_msg_type: {
auto& dref = static_cast<group_msg&>(x);
dref.target->eq_impl(make_message_id(), std::move(dref.sender), nullptr,
std::move(dref.content));
break;
}
default:
break;
}
}
void simple_actor_clock::handle(const ordinary_timeout_cancellation& x) {
auto pred = [&](const actor_lookup_map::value_type& kvp) {
auto& y = *kvp.second->second;
return y.subtype == ordinary_timeout_type
&& x.type == static_cast<const ordinary_timeout&>(y).type;
};
cancel(x.aid, pred);
}
void simple_actor_clock::handle(const multi_timeout_cancellation& x) {
auto pred = [&](const actor_lookup_map::value_type& kvp) {
auto& y = *kvp.second->second;
if (y.subtype != ordinary_timeout_type)
return false;
auto& dref = static_cast<const multi_timeout&>(y);
return x.type == dref.type && x.id == dref.id;
};
cancel(x.aid, pred);
}
void simple_actor_clock::handle(const request_timeout_cancellation& x) {
auto pred = [&](const actor_lookup_map::value_type& kvp) {
auto& y = *kvp.second->second;
return y.subtype == request_timeout_type
&& x.id == static_cast<const request_timeout&>(y).id;
};
cancel(x.aid, pred);
}
void simple_actor_clock::handle(const timeouts_cancellation& x) {
auto range = actor_lookup_.equal_range(x.aid);
if (range.first == range.second)
return;
for (auto i = range.first; i != range.second; ++i)
schedule_.erase(i->second);
actor_lookup_.erase(range.first, range.second);
}
size_t simple_actor_clock::trigger_expired_timeouts() {
size_t result = 0;
auto t = now();
auto i = schedule_.begin();
auto e = schedule_.end();
while (i != e && i->first <= t) {
auto ptr = std::move(i->second);
auto backlink = ptr->backlink;
if (backlink != actor_lookup_.end())
actor_lookup_.erase(backlink);
i = schedule_.erase(i);
ship(*ptr);
++result;
}
return result;
}
void
simple_actor_clock::add_schedule_entry(time_point t,
std::unique_ptr<ordinary_timeout> x) {
auto aid = x->self->id();
auto type = x->type;
auto pred = [&](const actor_lookup_map::value_type& kvp) {
auto& y = *kvp.second->second;
return y.subtype == ordinary_timeout_type
&& static_cast<const ordinary_timeout&>(y).type == type;
};
auto i = lookup(aid, pred);
if (i != actor_lookup_.end()) {
schedule_.erase(i->second);
i->second = schedule_.emplace(t, std::move(x));
} else {
auto j = schedule_.emplace(t, std::move(x));
i = actor_lookup_.emplace(aid, j);
}
i->second->second->backlink = i;
}
} // namespace detail
} // namespace caf
......@@ -46,13 +46,16 @@ bool test_actor_clock::trigger_timeout() {
CAF_LOG_TRACE(CAF_ARG2("schedule.size", schedule_.size()));
if (schedule_.empty())
return false;
visitor f{this};
auto i = schedule_.begin();
auto tout = i->first;
if (tout > current_time)
current_time = tout;
visit(f, i->second);
auto ptr = std::move(i->second);
schedule_.erase(i);
auto backlink = ptr->backlink;
if (backlink != actor_lookup_.end())
actor_lookup_.erase(backlink);
ship(*ptr);
return true;
}
......@@ -60,28 +63,9 @@ size_t test_actor_clock::trigger_timeouts() {
CAF_LOG_TRACE(CAF_ARG2("schedule.size", schedule_.size()));
if (schedule_.empty())
return 0u;
visitor f{this};
auto result = schedule_.size();
for (auto& kvp : schedule_) {
auto tout = kvp.first;
if (tout > current_time)
current_time = tout;
visit(f, kvp.second);
}
schedule_.clear();
return result;
}
size_t test_actor_clock::trigger_expired_timeouts() {
CAF_LOG_TRACE(CAF_ARG2("schedule.size", schedule_.size()));
visitor f{this};
size_t result = 0;
auto i = schedule_.begin();
while (i != schedule_.end() && i->first <= current_time) {
while (trigger_timeout())
++result;
visit(f, i->second);
i = schedule_.erase(i);
}
return result;
}
......
......@@ -18,133 +18,148 @@
#include "caf/detail/thread_safe_actor_clock.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/logger.hpp"
#include "caf/sec.hpp"
#include "caf/system_messages.hpp"
namespace caf {
namespace detail {
namespace {
using guard_type = std::unique_lock<std::recursive_mutex>;
} // namespace <anonymous>
thread_safe_actor_clock::thread_safe_actor_clock() : done_(false) {
// nop
}
void thread_safe_actor_clock::set_ordinary_timeout(time_point t,
abstract_actor* self,
atom_value type,
uint64_t id) {
guard_type guard{mx_};
if (!done_) {
super::set_ordinary_timeout(t, self, type, id);
cv_.notify_all();
}
push(new ordinary_timeout(t, self->ctrl(), type, id));
}
void thread_safe_actor_clock::set_request_timeout(time_point t,
abstract_actor* self,
message_id id) {
guard_type guard{mx_};
if (!done_) {
super::set_request_timeout(t, self, id);
cv_.notify_all();
}
push(new request_timeout(t, self->ctrl(), id));
}
void thread_safe_actor_clock::set_multi_timeout(time_point t, abstract_actor* self,
void thread_safe_actor_clock::set_multi_timeout(time_point t,
abstract_actor* self,
atom_value type, uint64_t id) {
guard_type guard{mx_};
if (!done_) {
super::set_multi_timeout(t, self, type, id);
cv_.notify_all();
}
push(new multi_timeout(t, self->ctrl(), type, id));
}
void thread_safe_actor_clock::cancel_ordinary_timeout(abstract_actor* self,
atom_value type) {
guard_type guard{mx_};
if (!done_) {
super::cancel_ordinary_timeout(self, type);
cv_.notify_all();
}
push(new ordinary_timeout_cancellation(self->id(), type));
}
void thread_safe_actor_clock::cancel_request_timeout(abstract_actor* self,
message_id id) {
guard_type guard{mx_};
if (!done_) {
super::cancel_request_timeout(self, id);
cv_.notify_all();
}
push(new request_timeout_cancellation(self->id(), id));
}
void thread_safe_actor_clock::cancel_timeouts(abstract_actor* self) {
guard_type guard{mx_};
if (!done_) {
super::cancel_timeouts(self);
cv_.notify_all();
}
push(new timeouts_cancellation(self->id()));
}
void thread_safe_actor_clock::schedule_message(time_point t,
strong_actor_ptr receiver,
mailbox_element_ptr content) {
guard_type guard{mx_};
if (!done_) {
super::schedule_message(t, std::move(receiver), std::move(content));
cv_.notify_all();
}
push(new actor_msg(t, std::move(receiver), std::move(content)));
}
void thread_safe_actor_clock::schedule_message(time_point t, group target,
strong_actor_ptr sender,
message content) {
guard_type guard{mx_};
if (!done_) {
super::schedule_message(t, std::move(target), std::move(sender),
std::move(content));
cv_.notify_all();
}
push(new group_msg(t, std::move(target), std::move(sender),
std::move(content)));
}
void thread_safe_actor_clock::cancel_all() {
guard_type guard{mx_};
super::cancel_all();
cv_.notify_all();
push(new drop_all);
}
void thread_safe_actor_clock::run_dispatch_loop() {
visitor f{this};
guard_type guard{mx_};
while (done_ == false) {
// Wait for non-empty schedule.
// Note: The thread calling run_dispatch_loop() is guaranteed not to lock
// the mutex recursively. Otherwise, cv_.wait() or cv_.wait_until()
// would be unsafe, because wait operations call unlock() only once.
for (;;) {
// Wait until queue is non-empty.
if (schedule_.empty()) {
cv_.wait(guard);
queue_.wait_nonempty();
} else {
auto tout = schedule_.begin()->first;
cv_.wait_until(guard, tout);
auto t = schedule_.begin()->second->due;
if (!queue_.wait_nonempty(t)) {
// Handle timeout by shipping timed-out events and starting anew.
trigger_expired_timeouts();
continue;
}
// 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);
}
// Push all elements from the queue to the events buffer.
auto i = events_.begin();
auto e = queue_.get_all(i);
for (; i != e; ++i) {
auto& x = *i;
CAF_ASSERT(x != nullptr);
switch (x->subtype) {
case ordinary_timeout_cancellation_type: {
handle(static_cast<ordinary_timeout_cancellation&>(*x));
break;
}
case request_timeout_cancellation_type: {
handle(static_cast<request_timeout_cancellation&>(*x));
break;
}
case timeouts_cancellation_type: {
handle(static_cast<timeouts_cancellation&>(*x));
break;
}
case drop_all_type: {
schedule_.clear();
actor_lookup_.clear();
break;
}
case shutdown_type: {
schedule_.clear();
actor_lookup_.clear();
// Call it a day.
return;
}
case ordinary_timeout_type: {
auto dptr = static_cast<ordinary_timeout*>(x.release());
add_schedule_entry(std::unique_ptr<ordinary_timeout>{dptr});
break;
}
case multi_timeout_type: {
auto dptr = static_cast<multi_timeout*>(x.release());
add_schedule_entry(std::unique_ptr<multi_timeout>{dptr});
break;
}
case request_timeout_type: {
auto dptr = static_cast<request_timeout*>(x.release());
add_schedule_entry(std::unique_ptr<request_timeout>{dptr});
break;
}
case actor_msg_type: {
auto dptr = static_cast<actor_msg*>(x.release());
add_schedule_entry(std::unique_ptr<actor_msg>{dptr});
break;
}
case group_msg_type: {
auto dptr = static_cast<group_msg*>(x.release());
add_schedule_entry(std::unique_ptr<group_msg>{dptr});
break;
}
default: {
CAF_LOG_ERROR("unexpected event type");
break;
}
}
x.reset();
}
}
}
void thread_safe_actor_clock::cancel_dispatch_loop() {
guard_type guard{mx_};
done_ = true;
cv_.notify_all();
push(new shutdown);
}
void thread_safe_actor_clock::push(event* ptr) {
queue_.push_back(unique_event_ptr{ptr});
}
} // namespace detail
......
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