99a15de9e5
The logger class constructor registers itself with the logger registry, in order to enable dynamically setting log levels. However, since thread_local variables may be (and are) initialized at the time of first use, when the program starts up no loggers are registered. Fix by making loggers global, not thread_local. This requires that the registry use locking to prevent registration happening on different threads from corrupting the registry. Note that technically global variables can also be initialized at the point of first use, and there is no portable way for classes to self-register. However this is the best we can do.
133 lines
3.8 KiB
C++
133 lines
3.8 KiB
C++
/*
|
|
* Copyright 2015 Cloudius Systems
|
|
*/
|
|
|
|
#include "row_cache.hh"
|
|
#include "core/memory.hh"
|
|
#include "core/do_with.hh"
|
|
#include "core/future-util.hh"
|
|
|
|
static logging::logger logger("cache");
|
|
|
|
cache_tracker& global_cache_tracker() {
|
|
static thread_local cache_tracker instance;
|
|
return instance;
|
|
}
|
|
|
|
cache_tracker::cache_tracker()
|
|
: _reclaimer([this] {
|
|
logger.warn("Clearing cache from reclaimer hook");
|
|
// FIXME: perform incremental eviction. We should first switch to a
|
|
// compacting memory allocator to avoid problems with memory
|
|
// fragmentation.
|
|
clear();
|
|
})
|
|
{ }
|
|
|
|
cache_tracker::~cache_tracker() {
|
|
clear();
|
|
}
|
|
|
|
void cache_tracker::clear() {
|
|
_lru.clear_and_dispose(std::default_delete<cache_entry>());
|
|
}
|
|
|
|
void cache_tracker::touch(cache_entry& e) {
|
|
_lru.erase(_lru.iterator_to(e));
|
|
_lru.push_front(e);
|
|
}
|
|
|
|
void cache_tracker::insert(cache_entry& entry) {
|
|
_lru.push_front(entry);
|
|
}
|
|
|
|
// Reader which populates the cache using data from the delegate.
|
|
class populating_reader {
|
|
row_cache& _cache;
|
|
mutation_reader _delegate;
|
|
public:
|
|
populating_reader(row_cache& cache, mutation_reader delegate)
|
|
: _cache(cache)
|
|
, _delegate(std::move(delegate))
|
|
{ }
|
|
|
|
future<mutation_opt> operator()() {
|
|
return _delegate().then([this] (mutation_opt&& mo) {
|
|
if (mo) {
|
|
_cache.populate(*mo);
|
|
}
|
|
return std::move(mo);
|
|
});
|
|
}
|
|
};
|
|
|
|
mutation_reader
|
|
row_cache::make_reader(const query::partition_range& range) {
|
|
if (range.is_singular()) {
|
|
const query::ring_position& pos = range.start_value();
|
|
|
|
if (!pos.has_key()) {
|
|
warn(unimplemented::cause::RANGE_QUERIES);
|
|
return populating_reader(*this, _underlying(range));
|
|
}
|
|
|
|
const dht::decorated_key& dk = pos.as_decorated_key();
|
|
auto i = _partitions.find(dk, cache_entry::compare(_schema));
|
|
if (i != _partitions.end()) {
|
|
cache_entry& e = *i;
|
|
_tracker.touch(e);
|
|
++_stats.hits;
|
|
return make_reader_returning(mutation(_schema, dk, e.partition()));
|
|
} else {
|
|
++_stats.misses;
|
|
return populating_reader(*this, _underlying(range));
|
|
}
|
|
}
|
|
|
|
warn(unimplemented::cause::RANGE_QUERIES);
|
|
return populating_reader(*this, _underlying(range));
|
|
}
|
|
|
|
row_cache::~row_cache() {
|
|
_partitions.clear_and_dispose(std::default_delete<cache_entry>());
|
|
}
|
|
|
|
void row_cache::populate(const mutation& m) {
|
|
populate(mutation(m));
|
|
}
|
|
|
|
void row_cache::populate(mutation&& m) {
|
|
auto i = _partitions.lower_bound(m.decorated_key(), cache_entry::compare(_schema));
|
|
if (i == _partitions.end()) {
|
|
cache_entry* entry = new cache_entry(m.decorated_key(), std::move(m.partition()));
|
|
_tracker.insert(*entry);
|
|
_partitions.insert(i, *entry);
|
|
} else {
|
|
cache_entry& entry = *i;
|
|
_tracker.touch(entry);
|
|
entry.partition().apply(*m.schema(), m.partition());
|
|
}
|
|
}
|
|
|
|
future<> row_cache::update(mutation_reader reader) {
|
|
return do_with(std::move(reader), [this] (mutation_reader& r) {
|
|
return consume(r, [this] (mutation&& m) {
|
|
auto i = _partitions.find(m.decorated_key(), cache_entry::compare(_schema));
|
|
// If cache doesn't contain the entry we cannot insert it because the mutation may be incomplete.
|
|
if (i != _partitions.end()) {
|
|
cache_entry& entry = *i;
|
|
_tracker.touch(entry);
|
|
entry.partition().apply(*m.schema(), std::move(m.partition()));
|
|
}
|
|
return stop_iteration::no;
|
|
});
|
|
});
|
|
}
|
|
|
|
row_cache::row_cache(schema_ptr s, mutation_source fallback_factory, cache_tracker& tracker)
|
|
: _tracker(tracker)
|
|
, _schema(std::move(s))
|
|
, _partitions(cache_entry::compare(_schema))
|
|
, _underlying(std::move(fallback_factory))
|
|
{ }
|