Commit 59d040ce authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Get rid of detail::shared_spinlock

parent 3fcdd263
......@@ -132,7 +132,6 @@ caf_add_component(
src/detail/private_thread_pool.cpp
src/detail/ripemd_160.cpp
src/detail/set_thread_name.cpp
src/detail/shared_spinlock.cpp
src/detail/stringification_inspector.cpp
src/detail/sync_request_bouncer.cpp
src/detail/test_actor_clock.cpp
......
......@@ -4,11 +4,7 @@
#pragma once
#include <functional>
#include <memory>
#include "caf/detail/core_export.hpp"
#include "caf/detail/shared_spinlock.hpp"
#include "caf/extend.hpp"
#include "caf/fwd.hpp"
#include "caf/mailbox_element.hpp"
......@@ -17,6 +13,10 @@
#include "caf/mixin/subscriber.hpp"
#include "caf/scheduled_actor.hpp"
#include <functional>
#include <memory>
#include <shared_mutex>
namespace caf {
template <>
......@@ -45,9 +45,6 @@ public:
/// Required by `spawn` for type deduction.
using behavior_type = behavior;
/// A shared lockable.
using lock_type = detail::shared_spinlock;
/// Delegates incoming messages to user-defined event loop.
using enqueue_handler = std::function<void(mailbox_element_ptr)>;
......@@ -92,7 +89,7 @@ private:
on_exit_handler on_exit_;
// guards access to handler_
lock_type lock_;
std::shared_mutex lock_;
};
} // namespace caf
......@@ -4,18 +4,17 @@
#pragma once
#include <functional>
#include <vector>
#include "caf/actor.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/shared_spinlock.hpp"
#include "caf/detail/split_join.hpp"
#include "caf/execution_unit.hpp"
#include "caf/locks.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/monitorable_actor.hpp"
#include <functional>
#include <mutex>
#include <vector>
namespace caf {
/// An actor poool is a lightweight abstraction for a set of workers.
......@@ -42,10 +41,11 @@ namespace caf {
/// @experimental
class CAF_CORE_EXPORT actor_pool : public monitorable_actor {
public:
using uplock = upgrade_lock<detail::shared_spinlock>;
using actor_vec = std::vector<actor>;
using factory = std::function<actor()>;
using policy = std::function<void(actor_system&, uplock&, const actor_vec&,
using guard_type = std::unique_lock<std::mutex>;
using policy
= std::function<void(actor_system&, guard_type&, const actor_vec&,
mailbox_element_ptr&, execution_unit*)>;
/// Returns a simple round robin dispatching policy.
......@@ -99,14 +99,13 @@ protected:
void on_cleanup(const error& reason) override;
private:
bool filter(upgrade_lock<detail::shared_spinlock>&,
const strong_actor_ptr& sender, message_id mid, message& msg,
execution_unit* eu);
bool filter(guard_type&, const strong_actor_ptr& sender, message_id mid,
message& msg, execution_unit* eu);
// call without workers_mtx_ held
void quit(execution_unit* host);
detail::shared_spinlock workers_mtx_;
std::mutex workers_mtx_;
std::vector<actor> workers_;
policy policy_;
exit_reason planned_reason_;
......
......@@ -9,7 +9,6 @@
#include "caf/abstract_actor.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/shared_spinlock.hpp"
#include "caf/monitorable_actor.hpp"
namespace caf {
......
......@@ -4,23 +4,23 @@
#pragma once
#include <atomic>
#include <condition_variable>
#include <cstdint>
#include <mutex>
#include <string>
#include <thread>
#include <unordered_map>
#include "caf/abstract_actor.hpp"
#include "caf/actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/shared_spinlock.hpp"
#include "caf/fwd.hpp"
#include "caf/telemetry/int_gauge.hpp"
#include <atomic>
#include <condition_variable>
#include <cstdint>
#include <mutex>
#include <shared_mutex>
#include <string>
#include <thread>
#include <unordered_map>
namespace caf {
/// A registry is used to associate actors to IDs or names. This allows a
......@@ -113,11 +113,11 @@ private:
mutable std::mutex running_mtx_;
mutable std::condition_variable running_cv_;
mutable detail::shared_spinlock instances_mtx_;
mutable std::shared_mutex instances_mtx_;
entries entries_;
name_map named_entries_;
mutable detail::shared_spinlock named_entries_mtx_;
mutable std::shared_mutex named_entries_mtx_;
actor_system& system_;
};
......
// 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.
#pragma once
#include <atomic>
#include <cstddef>
#include "caf/detail/core_export.hpp"
namespace caf::detail {
/// A spinlock implementation providing shared and exclusive locking.
class CAF_CORE_EXPORT shared_spinlock {
public:
shared_spinlock();
void lock();
void unlock();
bool try_lock();
void lock_shared();
void unlock_shared();
bool try_lock_shared();
void lock_upgrade();
void unlock_upgrade();
void unlock_upgrade_and_lock();
void unlock_and_lock_upgrade();
private:
std::atomic<long> flag_;
};
} // namespace caf::detail
......@@ -9,9 +9,6 @@
#include "caf/actor.hpp"
#include "caf/actor_system.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/locks.hpp"
#include "caf/detail/shared_spinlock.hpp"
namespace caf::detail {
......@@ -78,8 +75,7 @@ public:
// nop
}
void
operator()(actor_system& sys, upgrade_lock<detail::shared_spinlock>& ulock,
void operator()(actor_system& sys, std::unique_lock<std::shared_mutex>& ulock,
const std::vector<actor>& workers, mailbox_element_ptr& ptr,
execution_unit* host) {
if (!ptr->sender)
......
......@@ -7,7 +7,8 @@
#include "caf/actor.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/detail/core_export.hpp"
#include "caf/detail/shared_spinlock.hpp"
#include <shared_mutex>
namespace caf {
......@@ -32,7 +33,7 @@ private:
bool forward_msg(strong_actor_ptr sender, message_id mid, message msg,
const forwarding_stack* fwd = nullptr);
mutable detail::shared_spinlock broker_mtx_;
mutable std::shared_mutex broker_mtx_;
actor broker_;
};
......
// 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.
#pragma once
#include <mutex>
namespace caf {
template <class Lockable>
using unique_lock = std::unique_lock<Lockable>;
template <class SharedLockable>
class shared_lock {
public:
using lockable = SharedLockable;
explicit shared_lock(lockable& arg) : lockable_(&arg) {
lockable_->lock_shared();
}
~shared_lock() {
unlock();
}
bool owns_lock() const {
return lockable_ != nullptr;
}
void unlock() {
if (lockable_) {
lockable_->unlock_shared();
lockable_ = nullptr;
}
}
lockable* release() {
auto result = lockable_;
lockable_ = nullptr;
return result;
}
private:
lockable* lockable_;
};
template <class SharedLockable>
using upgrade_lock = shared_lock<SharedLockable>;
template <class UpgradeLockable>
class upgrade_to_unique_lock {
public:
using lockable = UpgradeLockable;
template <class LockType>
explicit upgrade_to_unique_lock(LockType& other) {
lockable_ = other.release();
if (lockable_)
lockable_->unlock_upgrade_and_lock();
}
~upgrade_to_unique_lock() {
unlock();
}
bool owns_lock() const {
return lockable_ != nullptr;
}
void unlock() {
if (lockable_) {
lockable_->unlock();
lockable_ = nullptr;
}
}
private:
lockable* lockable_;
};
} // namespace caf
......@@ -24,7 +24,6 @@
#include "caf/detail/pretty_type_name.hpp"
#include "caf/detail/ringbuffer.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/detail/shared_spinlock.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
......
......@@ -17,7 +17,6 @@
#include "caf/actor_system.hpp"
#include "caf/config.hpp"
#include "caf/default_attachable.hpp"
#include "caf/detail/shared_spinlock.hpp"
#include "caf/execution_unit.hpp"
#include "caf/logger.hpp"
#include "caf/mailbox_element.hpp"
......
......@@ -5,7 +5,6 @@
#include "caf/abstract_group.hpp"
#include "caf/actor_cast.hpp"
#include "caf/detail/shared_spinlock.hpp"
#include "caf/group.hpp"
#include "caf/group_manager.hpp"
#include "caf/group_module.hpp"
......
......@@ -2,7 +2,6 @@
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include "caf/locks.hpp"
#include "caf/actor_companion.hpp"
namespace caf {
......@@ -16,7 +15,7 @@ actor_companion::~actor_companion() {
}
void actor_companion::on_enqueue(enqueue_handler handler) {
std::lock_guard<lock_type> guard(lock_);
std::lock_guard guard{lock_};
on_enqueue_ = std::move(handler);
}
......@@ -26,7 +25,7 @@ void actor_companion::on_exit(on_exit_handler handler) {
bool actor_companion::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_ASSERT(ptr);
shared_lock<lock_type> guard(lock_);
std::shared_lock guard{lock_};
if (on_enqueue_) {
on_enqueue_(std::move(ptr));
return true;
......@@ -49,7 +48,7 @@ void actor_companion::launch(execution_unit*, bool, bool hide) {
void actor_companion::on_exit() {
enqueue_handler tmp;
{ // lifetime scope of guard
std::lock_guard<lock_type> guard(lock_);
std::unique_lock guard(lock_);
on_enqueue_.swap(tmp);
}
if (on_exit_)
......
......@@ -7,8 +7,8 @@
#include <atomic>
#include <random>
#include "caf/send.hpp"
#include "caf/default_attachable.hpp"
#include "caf/send.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
......@@ -22,7 +22,7 @@ actor_pool::policy actor_pool::round_robin() {
impl(const impl&) : pos_(0) {
// nop
}
void operator()(actor_system&, uplock& guard, const actor_vec& vec,
void operator()(actor_system&, guard_type& guard, const actor_vec& vec,
mailbox_element_ptr& ptr, execution_unit* host) {
CAF_ASSERT(!vec.empty());
actor selected = vec[pos_++ % vec.size()];
......@@ -36,7 +36,7 @@ actor_pool::policy actor_pool::round_robin() {
namespace {
void broadcast_dispatch(actor_system&, actor_pool::uplock&,
void broadcast_dispatch(actor_system&, actor_pool::guard_type&,
const actor_pool::actor_vec& vec,
mailbox_element_ptr& ptr, execution_unit* host) {
CAF_ASSERT(!vec.empty());
......@@ -59,12 +59,11 @@ actor_pool::policy actor_pool::random() {
impl(const impl&) : rd_() {
// nop
}
void operator()(actor_system&, uplock& guard, const actor_vec& vec,
void operator()(actor_system&, guard_type& guard, const actor_vec& vec,
mailbox_element_ptr& ptr, execution_unit* host) {
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
auto selected =
vec[dis_(rd_, decltype(dis_)::param_type(0, vec.size() - 1))];
unique_guard.unlock();
auto selected
= vec[dis_(rd_, decltype(dis_)::param_type(0, vec.size() - 1))];
guard.unlock();
selected->enqueue(std::move(ptr), host);
}
std::random_device rd_;
......@@ -95,14 +94,15 @@ actor actor_pool::make(execution_unit* eu, size_t num_workers,
auto res_addr = ptr->address();
for (size_t i = 0; i < num_workers; ++i) {
auto worker = fac();
worker->attach(default_attachable::make_monitor(worker.address(), res_addr));
worker->attach(
default_attachable::make_monitor(worker.address(), res_addr));
ptr->workers_.push_back(std::move(worker));
}
return res;
}
bool actor_pool::enqueue(mailbox_element_ptr what, execution_unit* eu) {
upgrade_lock<detail::shared_spinlock> guard{workers_mtx_};
guard_type guard{workers_mtx_};
if (filter(guard, what->sender, what->mid, what->payload, eu))
return false;
policy_(home_system(), guard, workers_, what, eu);
......@@ -110,8 +110,7 @@ bool actor_pool::enqueue(mailbox_element_ptr what, execution_unit* eu) {
}
actor_pool::actor_pool(actor_config& cfg)
: monitorable_actor(cfg),
planned_reason_(exit_reason::normal) {
: monitorable_actor(cfg), planned_reason_(exit_reason::normal) {
register_at_system();
}
......@@ -130,9 +129,8 @@ void actor_pool::on_cleanup(const error& reason) {
CAF_LOG_TERMINATE_EVENT(this, reason);
}
bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
const strong_actor_ptr& sender, message_id mid,
message& content, execution_unit* eu) {
bool actor_pool::filter(guard_type& guard, const strong_actor_ptr& sender,
message_id mid, message& content, execution_unit* eu) {
CAF_LOG_TRACE(CAF_ARG(mid) << CAF_ARG(content));
if (auto view = make_const_typed_message_view<exit_msg>(content)) {
// acquire second mutex as well
......@@ -141,9 +139,8 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
if (cleanup(std::move(reason), eu)) {
// send exit messages *always* to all workers and clear vector afterwards
// but first swap workers_ out of the critical section
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
workers_.swap(workers);
unique_guard.unlock();
guard.unlock();
for (auto& w : workers)
anon_send(w, content);
unregister_from_system();
......@@ -153,7 +150,6 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
if (auto view = make_const_typed_message_view<down_msg>(content)) {
// remove failed worker from pool
const auto& dm = get<0>(view);
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
auto last = workers_.end();
auto i = std::find(workers_.begin(), workers_.end(), dm.source);
CAF_LOG_DEBUG_IF(i == last, "received down message for an unknown worker");
......@@ -161,7 +157,7 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
workers_.erase(i);
if (workers_.empty()) {
planned_reason_ = exit_reason::out_of_workers;
unique_guard.unlock();
guard.unlock();
quit(eu);
}
return true;
......@@ -169,15 +165,13 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
if (auto view
= make_const_typed_message_view<sys_atom, put_atom, actor>(content)) {
const auto& worker = get<2>(view);
worker->attach(default_attachable::make_monitor(worker.address(),
address()));
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
worker->attach(
default_attachable::make_monitor(worker.address(), address()));
workers_.push_back(worker);
return true;
}
if (auto view
= make_const_typed_message_view<sys_atom, delete_atom, actor>(content)) {
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
auto& what = get<2>(view);
auto last = workers_.end();
auto i = std::find(workers_.begin(), last, what);
......@@ -190,7 +184,6 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
return true;
}
if (content.match_elements<sys_atom, delete_atom>()) {
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
for (auto& worker : workers_) {
default_attachable::observe_token tk{address(),
default_attachable::monitor};
......@@ -202,8 +195,8 @@ bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
if (content.match_elements<sys_atom, get_atom>()) {
auto cpy = workers_;
guard.unlock();
sender->enqueue(nullptr, mid.response_id(),
make_message(std::move(cpy)), eu);
sender->enqueue(nullptr, mid.response_id(), make_message(std::move(cpy)),
eu);
return true;
}
if (workers_.empty()) {
......
......@@ -12,10 +12,8 @@
#include "caf/actor_system.hpp"
#include "caf/attachable.hpp"
#include "caf/detail/shared_spinlock.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/exit_reason.hpp"
#include "caf/locks.hpp"
#include "caf/logger.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/sec.hpp"
......@@ -25,8 +23,8 @@ namespace caf {
namespace {
using exclusive_guard = unique_lock<detail::shared_spinlock>;
using shared_guard = shared_lock<detail::shared_spinlock>;
using exclusive_guard = std::unique_lock<std::shared_mutex>;
using shared_guard = std::shared_lock<std::shared_mutex>;
} // namespace
......
// 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/config.hpp"
#include <limits>
#include <thread>
#include "caf/detail/shared_spinlock.hpp"
#include "caf/detail/cas_weak.hpp"
namespace {
inline long min_long() {
return std::numeric_limits<long>::min();
}
} // namespace
namespace caf::detail {
shared_spinlock::shared_spinlock() : flag_(0) {
// nop
}
void shared_spinlock::lock() {
long v = flag_.load();
for (;;) {
if (v != 0) {
v = flag_.load();
} else if (cas_weak(&flag_, &v, min_long())) {
return;
}
// else: next iteration
}
}
void shared_spinlock::lock_upgrade() {
lock_shared();
}
void shared_spinlock::unlock_upgrade() {
unlock_shared();
}
void shared_spinlock::unlock_upgrade_and_lock() {
unlock_shared();
lock();
}
void shared_spinlock::unlock_and_lock_upgrade() {
unlock();
lock_upgrade();
}
void shared_spinlock::unlock() {
flag_.store(0);
}
bool shared_spinlock::try_lock() {
long v = flag_.load();
return (v == 0) ? cas_weak(&flag_, &v, min_long()) : false;
}
void shared_spinlock::lock_shared() {
long v = flag_.load();
for (;;) {
if (v < 0) {
// std::this_thread::yield();
v = flag_.load();
} else if (cas_weak(&flag_, &v, v + 1)) {
return;
}
// else: next iteration
}
}
void shared_spinlock::unlock_shared() {
flag_.fetch_sub(1);
}
bool shared_spinlock::try_lock_shared() {
long v = flag_.load();
return (v >= 0) ? cas_weak(&flag_, &v, v + 1) : false;
}
} // namespace caf::detail
......@@ -2,15 +2,14 @@
// the main distribution directory for license terms and copyright or visit
// https://github.com/actor-framework/actor-framework/blob/master/LICENSE.
#include <utility>
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/locks.hpp"
#include "caf/logger.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/send.hpp"
#include <utility>
namespace caf {
forwarding_actor_proxy::forwarding_actor_proxy(actor_config& cfg, actor dest)
......@@ -30,7 +29,7 @@ bool forwarding_actor_proxy::forward_msg(strong_actor_ptr sender,
if (msg.match_elements<exit_msg>())
unlink_from(msg.get_as<exit_msg>(0).source);
forwarding_stack tmp;
shared_lock<detail::shared_spinlock> guard(broker_mtx_);
std::shared_lock guard{broker_mtx_};
if (broker_)
return broker_->enqueue(nullptr, make_message_id(),
make_message(forward_atom_v, std::move(sender),
......@@ -71,7 +70,7 @@ bool forwarding_actor_proxy::remove_backlink(abstract_actor* x) {
void forwarding_actor_proxy::kill_proxy(execution_unit* ctx, error rsn) {
actor tmp;
{ // lifetime scope of guard
std::unique_lock<detail::shared_spinlock> guard(broker_mtx_);
std::unique_lock guard{broker_mtx_};
broker_.swap(tmp); // manually break cycle
}
cleanup(std::move(rsn), ctx);
......
......@@ -8,7 +8,6 @@
#include "caf/detail/local_group_module.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/group.hpp"
#include "caf/locks.hpp"
#include "caf/message.hpp"
#include "caf/sec.hpp"
#include "caf/serializer.hpp"
......
......@@ -25,7 +25,6 @@
#include "caf/detail/set_thread_name.hpp"
#include "caf/intrusive/task_result.hpp"
#include "caf/local_actor.hpp"
#include "caf/locks.hpp"
#include "caf/message.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/term.hpp"
......
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