Commit cdff618e authored by Marian Triebe's avatar Marian Triebe

Add general purpose cache for shared types (WIP)

relates #607
parent 6c0f3d70
......@@ -21,18 +21,13 @@
#define CAF_ACTOR_REGISTRY_HPP
#include <mutex>
#include <thread>
#include <atomic>
#include <cstdint>
#include <unordered_map>
#include <condition_variable>
#include "caf/fwd.hpp"
#include "caf/actor.hpp"
#include "caf/gp_cache.hpp"
#include "caf/abstract_actor.hpp"
#include "caf/actor_control_block.hpp"
#include "caf/detail/shared_spinlock.hpp"
namespace caf {
......@@ -42,7 +37,10 @@ namespace caf {
/// identify important actors independent from their ID at runtime.
/// Note that the registry does *not* contain all actors of an actor system.
/// The middleman registers actors as needed.
class actor_registry {
class actor_registry : public gp_cache<actor_id, strong_actor_ptr>,
public gp_cache<atom_value, strong_actor_ptr> {
using actor_id_cache = gp_cache<actor_id, strong_actor_ptr>;
using atom_value_cache = gp_cache<atom_value, strong_actor_ptr>;
public:
friend class actor_system;
......@@ -58,6 +56,15 @@ public:
/// leaving `reason` for future reference.
void erase(actor_id key);
/// Returns the actor associated with `key` or `invalid_actor`.
strong_actor_ptr get(atom_value id) const;
/// Associates given actor to `key`.
void put(atom_value key, strong_actor_ptr val);
/// Removes a name mapping.
void erase(atom_value key);
/// Increases running-actors-count by one.
void inc_running();
......@@ -71,40 +78,24 @@ public:
/// (must be either 0 or 1).
void await_running_count_equal(size_t expected) const;
/// Returns the actor associated with `key` or `invalid_actor`.
strong_actor_ptr get(atom_value key) const;
/// Associates given actor to `key`.
void put(atom_value key, strong_actor_ptr value);
/// Removes a name mapping.
void erase(atom_value key);
using name_map = std::unordered_map<atom_value, strong_actor_ptr>;
name_map named_actors() const;
/// @private
auto get_cache() -> decltype(gp_cache<atom_value, strong_actor_ptr>::get_cache()) {
return atom_value_cache::get_cache();
}
private:
// Starts this component.
/// Starts this component.
void start();
// Stops this component.
/// Stops this component.
void stop();
using entries = std::unordered_map<actor_id, strong_actor_ptr>;
actor_registry(actor_system& sys);
std::atomic<size_t> running_;
mutable std::mutex running_mtx_;
mutable std::condition_variable running_cv_;
mutable detail::shared_spinlock instances_mtx_;
entries entries_;
name_map named_entries_;
mutable detail::shared_spinlock named_entries_mtx_;
actor_system& system_;
};
......
......@@ -31,6 +31,7 @@
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/gp_cache.hpp"
#include "caf/actor_cast.hpp"
#include "caf/make_actor.hpp"
#include "caf/infer_handle.hpp"
......@@ -499,6 +500,9 @@ public:
/// @warning must be called by thread which is about to terminate
void thread_terminates();
///
gp_cache<node_id, intrusive_ptr<node_id::data>>& nid_cache() { return nid_cache_; }
/// @endcond
private:
......@@ -554,6 +558,7 @@ private:
actor_registry registry_;
group_manager groups_;
module_array modules_;
gp_cache<node_id, intrusive_ptr<node_id::data>> nid_cache_;
scoped_execution_unit dummy_execution_unit_;
bool await_actors_before_shutdown_;
// Stores SpawnServ, ConfigServ, and StreamServ
......
......@@ -30,6 +30,7 @@
#include "caf/fwd.hpp"
#include "caf/stream.hpp"
#include "caf/gp_cache.hpp"
#include "caf/thread_hook.hpp"
#include "caf/config_value.hpp"
#include "caf/config_option.hpp"
......@@ -234,6 +235,19 @@ public:
return *this;
}
///
template <class... Keys, class... Values>
actor_system_config& set_caches(detail::type_list<Keys...>,
detail::type_list<Values...>) {
// TODO: Implement me...
static_assert(detail::tl_size<
detail::type_list<Keys...>>::value
== detail::tl_size<
detail::type_list<Values...>
>::value, "count of keys and values does not match.");
return *this;
};
/// Stores whether the help text for this config object was
/// printed. If set to `true`, the application should not use
/// this config object to initialize an `actor_system` and
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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_GP_CACHE_HPP
#define CAF_GP_CACHE_HPP
#include <unordered_map>
#include "caf/fwd.hpp"
#include "caf/locks.hpp"
#include "caf/detail/shared_spinlock.hpp"
namespace caf {
/// Generel purpose cache for shared types
template <class Key, class Value>
class gp_cache {
public:
using map_type = std::unordered_map<Key, Value>;
using exclusive_guard = unique_lock<detail::shared_spinlock>;
using shared_guard = shared_lock<detail::shared_spinlock>;
friend class actor_system;
gp_cache() = default;
gp_cache(const gp_cache&) = delete;
~gp_cache() {
// nop
}
///
Value get(Key key) const {
shared_guard guard{mtx_};
auto i = map_.find(key);
return i != map_.end() ? i->second : nullptr;
}
///
bool put(Key key, Value value) {
exclusive_guard guard{mtx_};
return map_.emplace(key, std::move(value)).second;
}
///
void erase(Key key) {
exclusive_guard guard{mtx_};
map_.erase(key);
}
///
const map_type& get_cache() const {
exclusive_guard guard{mtx_};
return map_;
}
private:
map_type map_;
mutable detail::shared_spinlock mtx_;
};
} // namespace caf
#endif // CAF_GP_CACHE_HPP
......@@ -25,10 +25,13 @@
#include <cstdint>
#include <functional>
#include <iostream>
#include "caf/fwd.hpp"
#include "caf/none.hpp"
#include "caf/error.hpp"
#include "caf/config.hpp"
#include "caf/gp_cache.hpp"
#include "caf/ref_counted.hpp"
#include "caf/make_counted.hpp"
#include "caf/intrusive_ptr.hpp"
......@@ -128,6 +131,9 @@ public:
explicit node_id(intrusive_ptr<data> dataptr);
// TODO: Remove this...
gp_cache<node_id, intrusive_ptr<node_id::data>>& access_system(actor_system&);
template <class Inspector>
friend detail::enable_if_t<Inspector::reads_state,
typename Inspector::result_type>
......@@ -145,12 +151,22 @@ public:
data tmp;
// write changes to tmp back to x at scope exit
auto sg = detail::make_scope_guard([&] {
if (!tmp.valid())
if (!tmp.valid()) {
x.data_.reset();
else if (!x || !x.data_->unique())
return;
}
auto& nid_cache = x.access_system(f.context()->system());
if (!x || !x.data_->unique()) {
x.data_ = make_counted<data>(tmp);
else
*x.data_ = tmp;
nid_cache.put(x, x.data_);
} else {
auto cached = nid_cache.get(x);
if (!cached) {
*x.data_ = tmp;
nid_cache.put(x, x.data_);
} else
x.data_ = cached;
}
});
return f(meta::type_name("node_id"), tmp.pid_,
meta::hex_formatted(), tmp.host_);
......
......@@ -19,35 +19,12 @@
#include "caf/actor_registry.hpp"
#include <mutex>
#include <limits>
#include <stdexcept>
#include <unordered_map>
#include <unordered_set>
#include "caf/sec.hpp"
#include "caf/locks.hpp"
#include "caf/logger.hpp"
#include "caf/actor_cast.hpp"
#include "caf/attachable.hpp"
#include "caf/exit_reason.hpp"
#include "caf/actor_system.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/uniform_type_info_map.hpp"
#include "caf/detail/shared_spinlock.hpp"
namespace caf {
namespace {
using exclusive_guard = unique_lock<detail::shared_spinlock>;
using shared_guard = shared_lock<detail::shared_spinlock>;
} // namespace <anonymous>
actor_registry::~actor_registry() {
// nop
}
......@@ -57,24 +34,19 @@ actor_registry::actor_registry(actor_system& sys) : running_(0), system_(sys) {
}
strong_actor_ptr actor_registry::get(actor_id key) const {
shared_guard guard(instances_mtx_);
auto i = entries_.find(key);
if (i != entries_.end())
return i->second;
CAF_LOG_DEBUG("key invalid, assume actor no longer exists:" << CAF_ARG(key));
return nullptr;
auto ptr = actor_id_cache::get(key);
if (!ptr)
CAF_LOG_DEBUG("key invalid, assume actor no longer exists:"
<< CAF_ARG(key));
return ptr;
}
void actor_registry::put(actor_id key, strong_actor_ptr val) {
CAF_LOG_TRACE(CAF_ARG(key));
if (!val)
return;
{ // lifetime scope of guard
exclusive_guard guard(instances_mtx_);
if (!entries_.emplace(key, val).second)
return;
}
// attach functor without lock
if (!actor_id_cache::put(key, val))
return;
CAF_LOG_INFO("added actor:" << CAF_ARG(key));
actor_registry* reg = this;
val->get()->attach_functor([key, reg]() {
......@@ -83,8 +55,23 @@ void actor_registry::put(actor_id key, strong_actor_ptr val) {
}
void actor_registry::erase(actor_id key) {
exclusive_guard guard{instances_mtx_};
entries_.erase(key);
actor_id_cache::erase(key);
}
strong_actor_ptr actor_registry::get(atom_value id) const {
return atom_value_cache::get(id);
}
void actor_registry::put(atom_value key, strong_actor_ptr val) {
if (val)
val->get()->attach_functor([=] {
system_.registry().put(key, nullptr);
});
atom_value_cache::put(key, std::move(val));
}
void actor_registry::erase(atom_value key) {
atom_value_cache::erase(key);
}
void actor_registry::inc_running() {
......@@ -119,33 +106,6 @@ void actor_registry::await_running_count_equal(size_t expected) const {
}
}
strong_actor_ptr actor_registry::get(atom_value key) const {
shared_guard guard{named_entries_mtx_};
auto i = named_entries_.find(key);
if (i == named_entries_.end())
return nullptr;
return i->second;
}
void actor_registry::put(atom_value key, strong_actor_ptr value) {
if (value)
value->get()->attach_functor([=] {
system_.registry().put(key, nullptr);
});
exclusive_guard guard{named_entries_mtx_};
named_entries_.emplace(key, std::move(value));
}
void actor_registry::erase(atom_value key) {
exclusive_guard guard{named_entries_mtx_};
named_entries_.erase(key);
}
auto actor_registry::named_actors() const -> name_map {
shared_guard guard{named_entries_mtx_};
return named_entries_;
}
void actor_registry::start() {
// nop
}
......
......@@ -25,6 +25,7 @@
#include "caf/config.hpp"
#include "caf/node_id.hpp"
#include "caf/serializer.hpp"
#include "caf/actor_system.hpp"
#include "caf/deserializer.hpp"
#include "caf/make_counted.hpp"
#include "caf/string_algorithms.hpp"
......@@ -178,6 +179,10 @@ void node_id::swap(node_id& x) {
data_.swap(x.data_);
}
gp_cache<node_id, intrusive_ptr<node_id::data>>& node_id::access_system(actor_system& system) {
return system.nid_cache();
}
std::string to_string(const node_id& x) {
std::string result;
append_to_string(result, x);
......
......@@ -612,7 +612,7 @@ CAF_TEST(remote_actor_and_send) {
CAF_REQUIRE(!mpx()->has_pending_scribe(lo, 4242));
// build a fake server handshake containing the id of our first pseudo actor
CAF_MESSAGE("server handshake => client handshake + proxy announcement");
auto na = registry()->named_actors();
auto na = registry()->get_cache();
mock(jupiter().connection,
{basp::message_type::server_handshake, 0, 0, basp::version,
jupiter().id, none,
......
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