natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

global key/row caches 

patch by Pavel Yaskevich; reviewed by Sylvain Lebresne for CASSANDRA-3143

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1222715 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Pavel Yaskevich 2011-12-23 16:09:05 +00:00
parent 6ea00a3047
commit 295aedb278
47 changed files with 1187 additions and 1032 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -28,7 +28,7 @@
* Use separate writer thread in SSTableSimpleUnsortedWriter (CASSANDRA-3619)
* fsync the directory after new sstable or commitlog segment are created (CASSANDRA-3250)
* fix minor issues reported by FindBugs (CASSANDRA-3658)
* global key/row caches (CASSANDRA-3143)
1.0.7
* add nodetool setstreamthroughput (CASSANDRA-3571)

4
NEWS.txt
View File

@ -31,7 +31,9 @@ Upgrading
Larger batches will continue to be accepted but will not be
durable. Consider setting durable_writes=false if you really
want to use such large batches.
- Make sure that global settings: key_cache_{size_in_mb, save_period}
and row_cache_{size_in_mb, save_period} in conf/cassandra.yaml are
used instead of per-ColumnFamily options.
1.0.6
=====

66
conf/cassandra.yaml
View File

@ -66,6 +66,72 @@ data_file_directories:
# commit log
commitlog_directory: /var/lib/cassandra/commitlog
# Maximum size of the key cache in memory.
#
# Each key cache hit saves 1 seek and each row cache hit saves 2 seeks at the
# minimum, sometimes more. The key cache is fairly tiny for the amount of
# time it saves, so it's worthwhile to use it at large numbers.
# The row cache saves even more time, but must store the whole values of
# its rows, so it is extremely space-intensive. It's best to only use the
# row cache if you have hot rows or static rows.
#
# NOTE: if you reduce the size, you may not get you hottest keys loaded on startup.
#
# Default value is 2 (call hold > 200000 keys). Set to 0 to disable key cache.
key_cache_size_in_mb: 2
# Duration in seconds after which Cassandra should
# safe the keys cache. Caches are saved to saved_caches_directory as
# specified in this configuration file.
#
# Saved caches greatly improve cold-start speeds, and is relatively cheap in
# terms of I/O for the key cache. Row cache saving is much more expensive and
# has limited use.
#
# Default is 14400 or 4 hours.
key_cache_save_period: 14400
# Number of keys from the key cache to save
# Disabled by default, meaning all keys are going to be saved
# key_cache_keys_to_save: 100
# Maximum size of the row cache in memory.
# NOTE: if you reduce the size, you may not get you hottest keys loaded on startup.
#
# Default value is 0, to disable row caching.
row_cache_size_in_mb: 0
# Duration in seconds after which Cassandra should
# safe the row cache. Caches are saved to saved_caches_directory as specified
# in this configuration file.
#
# Saved caches greatly improve cold-start speeds, and is relatively cheap in
# terms of I/O for the key cache. Row cache saving is much more expensive and
# has limited use.
#
# Default is 0 to disable saving the row cache.
row_cache_save_period: 0
# Number of keys from the row cache to save
# Disabled by default, meaning all keys are going to be saved
# row_cache_keys_to_save: 100
# The provider for the row cache to use.
#
# Supported values are: ConcurrentLinkedHashCacheProvider, SerializingCacheProvider
#
# SerializingCacheProvider serialises the contents of the row and stores
# it in native memory, i.e., off the JVM Heap. Serialized rows take
# significantly less memory than "live" rows in the JVM, so you can cache
# more rows in a given memory footprint. And storing the cache off-heap
# means you can use smaller heap sizes, reducing the impact of GC pauses.
#
# It is also valid to specify the fully-qualified class name to a class
# that implements org.apache.cassandra.cache.IRowCacheProvider.
#
# Defaults to SerializingCacheProvider
row_cache_provider: SerializingCacheProvider
# saved caches
saved_caches_directory: /var/lib/cassandra/saved_caches

5
doc/cql/CQL.textile
View File

@ -538,16 +538,11 @@ A number of optional keyword arguments can be supplied to control the configurat
|_. keyword|_. default|_. description|
|comparator|text|Determines the storage type of column names (which itself determines the sorting and validation of column names). Valid values are listed in the "Data Storage Types":#storageTypes table above.|
|comment|none|A free-form, human-readable comment.|
|row_cache_provider|SerializingCacheProvider if JNA is present, otherwise ConcurrentHashMapCacheProvider|A factory for the cache with which to back the row cache.|
|row_cache_size|0|Number of rows whose entire contents to cache in memory.|
|key_cache_size|200000|Number of keys per SSTable whose locations are kept in memory in "mostly LRU" order.|
|read_repair_chance|1.0|The probability with which read repairs should be invoked on non-quorum reads.|
|gc_grace_seconds|864000|Time to wait before garbage collecting tombstones (deletion markers).|
|default_validation|text|Determines the default storage type of column values (which itself determines the validation for column values). This option does not affect the types of columns which were defined in a @CREATE COLUMNFAMILY@ statement-- only new columns. Valid values are listed in the "Data Storage Types":#storageTypes table above.|
|min_compaction_threshold|4|Minimum number of SSTables needed to start a minor compaction.|
|max_compaction_threshold|32|Maximum number of SSTables allowed before a minor compaction is forced.|
|row_cache_save_period_in_seconds|0|Number of seconds between saving row caches.|
|key_cache_save_period_in_seconds|14400|Number of seconds between saving key caches.|
|replicate_on_write|false| |
h2. CREATE INDEX

8
interface/cassandra.thrift
View File

@ -46,7 +46,7 @@ namespace rb CassandraThrift
# for every edit that doesn't result in a change to major/minor.
#
# See the Semantic Versioning Specification (SemVer) http://semver.org.
const string VERSION = "19.22.1"
const string VERSION = "19.23.1"
#
@ -394,8 +394,6 @@ struct CfDef {
5: optional string comparator_type="BytesType",
6: optional string subcomparator_type,
8: optional string comment,
9: optional double row_cache_size=0,
11: optional double key_cache_size=200000,
12: optional double read_repair_chance=1.0,
13: optional list<ColumnDef> column_metadata,
14: optional i32 gc_grace_seconds,
@ -403,16 +401,12 @@ struct CfDef {
16: optional i32 id,
17: optional i32 min_compaction_threshold,
18: optional i32 max_compaction_threshold,
19: optional i32 row_cache_save_period_in_seconds,
20: optional i32 key_cache_save_period_in_seconds,
24: optional bool replicate_on_write,
25: optional double merge_shards_chance,
26: optional string key_validation_class,
27: optional string row_cache_provider,
28: optional binary key_alias,
29: optional string compaction_strategy,
30: optional map<string,string> compaction_strategy_options,
31: optional i32 row_cache_keys_to_save,
32: optional map<string,string> compression_options,
}

6
src/avro/internode.genavro
View File

@ -51,8 +51,6 @@ protocol InterNode {
union { string, null } comparator_type;
union { string, null } subcomparator_type;
union { string, null } comment;
union { double, null } row_cache_size;
union { double, null } key_cache_size;
union { double, null } read_repair_chance;
boolean replicate_on_write = false;
union { int, null } gc_grace_seconds;
@ -60,13 +58,9 @@ protocol InterNode {
union { null, string } key_validation_class = null;
union { null, int } min_compaction_threshold = null;
union { null, int } max_compaction_threshold = null;
union { int, null } row_cache_save_period_in_seconds = 0;
union { int, null } key_cache_save_period_in_seconds = 3600;
union { null, int } row_cache_keys_to_save = null;
union { null, double} merge_shards_chance = null;
union { int, null } id;
union { array<ColumnDef>, null } column_metadata;
union { string, null } row_cache_provider = "org.apache.cassandra.cache.ConcurrentLinkedHashCacheProvider";
union { null, bytes } key_alias = null;
union { null, string } compaction_strategy = null;
union { null, map<string> } compaction_strategy_options = null;

188
src/java/org/apache/cassandra/cache/AutoSavingCache.java vendored
View File

@ -20,68 +20,53 @@ package org.apache.cassandra.cache;
import java.io.*;
import java.nio.ByteBuffer;
import java.util.Set;
import java.util.TreeSet;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.compaction.CompactionInfo;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.SequentialWriter;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.WrappedRunnable;
import org.apache.cassandra.utils.Pair;
public abstract class AutoSavingCache<K, V> extends InstrumentingCache<K, V>
public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K, V>
{
private static final Logger logger = LoggerFactory.getLogger(AutoSavingCache.class);
/** True if a cache flush is currently executing: only one may execute at a time. */
public static final AtomicBoolean flushInProgress = new AtomicBoolean(false);
protected final String cfName;
protected final String tableName;
protected volatile ScheduledFuture<?> saveTask;
protected final ColumnFamilyStore.CacheType cacheType;
protected final CacheService.CacheType cacheType;
public AutoSavingCache(ICache<K, V> cache, String tableName, String cfName, ColumnFamilyStore.CacheType cacheType)
public AutoSavingCache(ICache<K, V> cache, CacheService.CacheType cacheType)
{
super(cache, tableName, cfName + cacheType);
this.tableName = tableName;
this.cfName = cfName;
super(cache);
this.cacheType = cacheType;
}
public abstract ByteBuffer translateKey(K key);
public abstract double getConfiguredCacheSize(CFMetaData cfm);
public int getAdjustedCacheSize(long expectedKeys)
public File getCachePath(String ksName, String cfName)
{
CFMetaData cfm = Schema.instance.getCFMetaData(tableName, cfName);
return (int)Math.min(FBUtilities.absoluteFromFraction(getConfiguredCacheSize(cfm), expectedKeys), Integer.MAX_VALUE);
}
public File getCachePath()
{
return DatabaseDescriptor.getSerializedCachePath(tableName, cfName, cacheType);
return DatabaseDescriptor.getSerializedCachePath(ksName, cfName, cacheType);
}
public Writer getWriter(int keysToSave)
{
return new Writer(tableName, cfName, keysToSave);
return new Writer(keysToSave);
}
public void scheduleSaving(int savePeriodInSeconds, final int keysToSave)
@ -107,14 +92,9 @@ public abstract class AutoSavingCache<K, V> extends InstrumentingCache<K, V>
}
}
public Future<?> submitWrite(int keysToSave)
public Set<DecoratedKey> readSaved(String ksName, String cfName)
{
return CompactionManager.instance.submitCacheWrite(getWriter(keysToSave));
}
public Set<DecoratedKey> readSaved()
{
File path = getCachePath();
File path = getCachePath(ksName, cfName);
Set<DecoratedKey> keys = new TreeSet<DecoratedKey>();
if (path.exists())
{
@ -139,14 +119,14 @@ public abstract class AutoSavingCache<K, V> extends InstrumentingCache<K, V>
catch (Exception e)
{
logger.info(String.format("unable to read entry #%s from saved cache %s; skipping remaining entries",
keys.size(), path.getAbsolutePath()), e);
keys.size(), path.getAbsolutePath()), e);
break;
}
keys.add(key);
}
if (logger.isDebugEnabled())
logger.debug(String.format("completed reading (%d ms; %d keys) saved cache %s",
System.currentTimeMillis() - start, keys.size(), path));
System.currentTimeMillis() - start, keys.size(), path));
}
catch (Exception e)
{
@ -160,36 +140,34 @@ public abstract class AutoSavingCache<K, V> extends InstrumentingCache<K, V>
return keys;
}
/**
* Resizes the cache based on a key estimate.
* Caller is in charge of synchronizing this correctly if needed
*/
public void updateCacheSize(long keys)
public Future<?> submitWrite(int keysToSave)
{
if (!isCapacitySetManually())
{
int cacheSize = getAdjustedCacheSize(keys);
if (cacheSize != getCapacity())
{
// update cache size for the new volume
if (logger.isDebugEnabled())
logger.debug(cacheType + " capacity for " + cfName + " is " + cacheSize);
updateCapacity(cacheSize);
}
}
return CompactionManager.instance.submitCacheWrite(getWriter(keysToSave));
}
public void reduceCacheSize()
{
if (getCapacity() > 0)
{
int newCapacity = (int) (DatabaseDescriptor.getReduceCacheCapacityTo() * size());
logger.warn(String.format("Reducing %s %s capacity from %d to %s to reduce memory pressure",
cfName, cacheType, getCapacity(), newCapacity));
int newCapacity = (int) (DatabaseDescriptor.getReduceCacheCapacityTo() * weightedSize());
logger.warn(String.format("Reducing %s capacity from %d to %s to reduce memory pressure",
cacheType, getCapacity(), newCapacity));
setCapacity(newCapacity);
}
}
public int estimateSizeToSave(Set<K> keys)
{
int bytes = 0;
for (K key : keys)
bytes += key.serializedSize();
return bytes;
}
public class Writer extends CompactionInfo.Holder
{
private final Set<K> keys;
@ -197,29 +175,27 @@ public abstract class AutoSavingCache<K, V> extends InstrumentingCache<K, V>
private final long estimatedTotalBytes;
private long bytesWritten;
private Writer(String ksname, String cfname, int keysToSave)
protected Writer(int keysToSave)
{
if (keysToSave >= getKeySet().size())
keys = getKeySet();
else
keys = hotKeySet(keysToSave);
long bytes = 0;
for (K key : keys)
bytes += translateKey(key).remaining();
// an approximation -- the keyset can change while saving
estimatedTotalBytes = bytes;
OperationType type;
if (cacheType == ColumnFamilyStore.CacheType.KEY_CACHE_TYPE)
// an approximation -- the keyset can change while saving
estimatedTotalBytes = estimateSizeToSave(keys);
OperationType type;
if (cacheType == CacheService.CacheType.KEY_CACHE)
type = OperationType.KEY_CACHE_SAVE;
else if (cacheType == ColumnFamilyStore.CacheType.ROW_CACHE_TYPE)
else if (cacheType == CacheService.CacheType.ROW_CACHE)
type = OperationType.ROW_CACHE_SAVE;
else
type = OperationType.UNKNOWN;
info = new CompactionInfo(this.hashCode(),
ksname,
cfname,
"Global",
cacheType.toString(),
type,
0,
estimatedTotalBytes);
@ -235,37 +211,83 @@ public abstract class AutoSavingCache<K, V> extends InstrumentingCache<K, V>
public void saveCache() throws IOException
{
long start = System.currentTimeMillis();
File path = getCachePath();
logger.debug("Deleting old {} files.", cacheType);
deleteOldCacheFiles();
if (keys.size() == 0 || estimatedTotalBytes == 0)
{
logger.debug("Deleting {} (cache is empty)");
path.delete();
logger.debug("Skipping {} save, cache is empty.", cacheType);
return;
}
logger.debug("Saving {}", path);
File tmpFile = File.createTempFile(path.getName(), null, path.getParentFile());
DataOutputStream out = SequentialWriter.open(tmpFile, true).stream;
long start = System.currentTimeMillis();
HashMap<Pair<String, String>, SequentialWriter> writers = new HashMap<Pair<String, String>, SequentialWriter>();
try
{
for (K key : keys)
for (CacheKey key : keys)
{
ByteBuffer bytes = translateKey(key);
ByteBufferUtil.writeWithLength(bytes, out);
Pair<String, String> path = key.getPathInfo();
SequentialWriter writer = writers.get(path);
if (writer == null)
{
writer = tempCacheFile(path);
writers.put(path, writer);
}
ByteBuffer bytes = key.serializeForStorage();
ByteBufferUtil.writeWithLength(bytes, writer.stream);
bytesWritten += bytes.remaining();
}
}
finally
{
out.close();
for (SequentialWriter writer : writers.values())
FileUtils.closeQuietly(writer);
}
for (Map.Entry<Pair<String, String>, SequentialWriter> info : writers.entrySet())
{
Pair<String, String> path = info.getKey();
SequentialWriter writer = info.getValue();
File tmpFile = new File(writer.getPath());
File cacheFile = getCachePath(path.left, path.right);
cacheFile.delete(); // ignore error if it didn't exist
if (!tmpFile.renameTo(cacheFile))
logger.error("Unable to rename " + tmpFile + " to " + cacheFile);
}
logger.info(String.format("Saved %s (%d items) in %d ms", cacheType, keys.size(), System.currentTimeMillis() - start));
}
private SequentialWriter tempCacheFile(Pair<String, String> pathInfo) throws IOException
{
File path = getCachePath(pathInfo.left, pathInfo.right);
File tmpFile = File.createTempFile(path.getName(), null, path.getParentFile());
return SequentialWriter.open(tmpFile, true);
}
private void deleteOldCacheFiles()
{
File savedCachesDir = new File(DatabaseDescriptor.getSavedCachesLocation());
if (savedCachesDir.exists() && savedCachesDir.isDirectory())
{
for (File file : savedCachesDir.listFiles())
{
if (file.isFile() && file.getName().endsWith(cacheType.toString()))
{
if (!file.delete())
logger.warn("Failed to delete {}", file.getAbsolutePath());
}
}
}
path.delete(); // ignore error if it didn't exist
if (!tmpFile.renameTo(path))
throw new IOException("Unable to rename " + tmpFile + " to " + path);
logger.info(String.format("Saved %s (%d items) in %d ms",
path.getName(), keys.size(), (System.currentTimeMillis() - start)));
}
}
}

48
src/java/org/apache/cassandra/cache/AutoSavingRowCache.java vendored
View File

@ -1,48 +0,0 @@
package org.apache.cassandra.cache;
/*
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*
*/
import java.nio.ByteBuffer;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
public class AutoSavingRowCache<K extends DecoratedKey, V> extends AutoSavingCache<K, V>
{
public AutoSavingRowCache(ICache<K, V> cache, String tableName, String cfName)
{
super(cache, tableName, cfName, ColumnFamilyStore.CacheType.ROW_CACHE_TYPE);
}
@Override
public double getConfiguredCacheSize(CFMetaData cfm)
{
return cfm == null ? CFMetaData.DEFAULT_ROW_CACHE_SIZE : cfm.getRowCacheSize();
}
@Override
public ByteBuffer translateKey(K key)
{
return key.key;
}
}

40
src/java/org/apache/cassandra/cache/InstrumentingCacheMBean.java → src/java/org/apache/cassandra/cache/CacheKey.java vendored
View File

@ -1,6 +1,4 @@
package org.apache.cassandra.cache;
/*
*
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
@ -8,35 +6,37 @@ package org.apache.cassandra.cache;
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
*
* http://www.apache.org/licenses/LICENSE-2.0
*
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*
*
*/
package org.apache.cassandra.cache;
import java.nio.ByteBuffer;
public interface InstrumentingCacheMBean
import org.apache.cassandra.utils.Pair;
public interface CacheKey
{
public int getCapacity();
public void setCapacity(int capacity);
public int getSize();
/** total request count since cache creation */
public long getRequests();
/** total cache hit count since cache creation */
public long getHits();
/**
* @return Serialized part of the key which should be persisted
*/
public ByteBuffer serializeForStorage();
/**
* hits / requests since the last time getHitRate was called. serious telemetry apps should not use this,
* and should instead track the deltas from getHits / getRequests themselves, since those will not be
* affected by multiple users calling it. Provided for convenience only.
* @return The size of the serialized key
*/
public double getRecentHitRate();
public int serializedSize();
/**
* @return The keyspace and ColumnFamily names to which this key belongs
*/
public Pair<String, String> getPathInfo();
}

50
src/java/org/apache/cassandra/cache/ConcurrentLinkedHashCache.java vendored
View File

@ -20,12 +20,11 @@ package org.apache.cassandra.cache;
*
*/
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import com.googlecode.concurrentlinkedhashmap.ConcurrentLinkedHashMap;
import com.googlecode.concurrentlinkedhashmap.Weigher;
import com.googlecode.concurrentlinkedhashmap.Weighers;
/** Wrapper so CLHM can implement ICache interface.
@ -40,14 +39,41 @@ public class ConcurrentLinkedHashCache<K, V> implements ICache<K, V>
this.map = map;
}
public static <K, V> ConcurrentLinkedHashCache<K, V> create(int capacity, String tableName, String cfname)
/**
* Initialize a cache with weigher = Weighers.singleton() and initial capacity 0
*
* @param capacity cache weighted capacity
*
* @param <K> key type
* @param <V> value type
*
* @return initialized cache
*/
public static <K, V> ConcurrentLinkedHashCache<K, V> create(int capacity)
{
return create(capacity, Weighers.<V>singleton());
}
/**
* Initialize a cache with initial capacity set to 0
*
* @param weightedCapacity cache weighted capacity
* @param weigher The weigher to use
*
* @param <K> key type
* @param <V> value type
*
* @return initialized cache
*/
public static <K, V> ConcurrentLinkedHashCache<K, V> create(int weightedCapacity, Weigher<V> weigher)
{
ConcurrentLinkedHashMap<K, V> map = new ConcurrentLinkedHashMap.Builder<K, V>()
.weigher(Weighers.<V>singleton())
.initialCapacity(capacity)
.maximumWeightedCapacity(capacity)
.weigher(weigher)
.initialCapacity(0)
.maximumWeightedCapacity(weightedCapacity)
.concurrencyLevel(DEFAULT_CONCURENCY_LEVEL)
.build();
return new ConcurrentLinkedHashCache<K, V>(map);
}
@ -71,6 +97,11 @@ public class ConcurrentLinkedHashCache<K, V> implements ICache<K, V>
return map.size();
}
public int weightedSize()
{
return map.weightedSize();
}
public void clear()
{
map.clear();
@ -101,6 +132,11 @@ public class ConcurrentLinkedHashCache<K, V> implements ICache<K, V>
return map.descendingKeySetWithLimit(n);
}
public boolean containsKey(K key)
{
return map.containsKey(key);
}
public boolean isPutCopying()
{
return false;

28
src/java/org/apache/cassandra/cache/ConcurrentLinkedHashCacheProvider.java vendored
View File

@ -20,15 +20,33 @@ package org.apache.cassandra.cache;
*
*/
import com.googlecode.concurrentlinkedhashmap.ConcurrentLinkedHashMap;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.DecoratedKey;
import com.googlecode.concurrentlinkedhashmap.Weigher;
import com.googlecode.concurrentlinkedhashmap.Weighers;
import org.github.jamm.MemoryMeter;
public class ConcurrentLinkedHashCacheProvider implements IRowCacheProvider
{
public ICache<DecoratedKey, ColumnFamily> create(int capacity, String tableName, String cfName)
public ICache<RowCacheKey, ColumnFamily> create(int capacity, boolean useMemoryWeigher)
{
return ConcurrentLinkedHashCache.create(capacity, tableName, cfName);
return ConcurrentLinkedHashCache.create(capacity, useMemoryWeigher
? createMemoryWeigher()
: Weighers.<ColumnFamily>singleton());
}
private static Weigher<ColumnFamily> createMemoryWeigher()
{
return new Weigher<ColumnFamily>()
{
final MemoryMeter meter = new MemoryMeter();
@Override
public int weightOf(ColumnFamily value)
{
return (int) Math.min(meter.measure(value), Integer.MAX_VALUE);
}
};
}
}

4
src/java/org/apache/cassandra/cache/ICache.java vendored
View File

@ -42,12 +42,16 @@ public interface ICache<K, V>
public int size();
public int weightedSize();
public void clear();
public Set<K> keySet();
public Set<K> hotKeySet(int n);
public boolean containsKey(K key);
/**
* @return true if the cache implementation inherently copies the cached values; otherwise,
* the caller should copy manually before caching shared values like Thrift ByteBuffers.

4
src/java/org/apache/cassandra/cache/IRowCacheProvider.java vendored
View File

@ -20,14 +20,12 @@ package org.apache.cassandra.cache;
*
*/
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.DecoratedKey;
/**
* Provides cache objects with a requested capacity.
*/
public interface IRowCacheProvider
{
public ICache<DecoratedKey, ColumnFamily> create(int capacity, String tableName, String cfName);
public ICache<RowCacheKey, ColumnFamily> create(int capacity, boolean useMemoryWeigher);
}

31
src/java/org/apache/cassandra/cache/InstrumentingCache.java vendored
View File

@ -20,18 +20,13 @@ package org.apache.cassandra.cache;
*
*/
import java.lang.management.ManagementFactory;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import javax.management.MBeanServer;
import javax.management.ObjectName;
/**
* Wraps an ICache in requests + hits tracking.
*/
public class InstrumentingCache<K, V> implements InstrumentingCacheMBean
public class InstrumentingCache<K, V>
{
private final AtomicLong requests = new AtomicLong(0);
private final AtomicLong hits = new AtomicLong(0);
@ -40,22 +35,9 @@ public class InstrumentingCache<K, V> implements InstrumentingCacheMBean
private volatile boolean capacitySetManually;
private final ICache<K, V> map;
public InstrumentingCache(ICache<K, V> map, String table, String name)
public InstrumentingCache(ICache<K, V> map)
{
this.map = map;
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
ObjectName mbeanName = new ObjectName("org.apache.cassandra.db:type=Caches,keyspace=" + table + ",cache=" + name);
// unregister any previous, as this may be a replacement.
if (mbs.isRegistered(mbeanName))
mbs.unregisterMBean(mbeanName);
mbs.registerMBean(this, mbeanName);
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
public void put(K key, V value)
@ -108,9 +90,9 @@ public class InstrumentingCache<K, V> implements InstrumentingCacheMBean
return map.size();
}
public int getSize()
public int weightedSize()
{
return size();
return map.weightedSize();
}
public long getHits()
@ -155,6 +137,11 @@ public class InstrumentingCache<K, V> implements InstrumentingCacheMBean
return map.hotKeySet(n);
}
public boolean containsKey(K key)
{
return map.containsKey(key);
}
public boolean isPutCopying()
{
return map.isPutCopying();

47
src/java/org/apache/cassandra/cache/AutoSavingKeyCache.java → src/java/org/apache/cassandra/cache/KeyCacheKey.java vendored
View File

@ -1,6 +1,4 @@
package org.apache.cassandra.cache;
/*
*
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
@ -8,43 +6,56 @@ package org.apache.cassandra.cache;
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
*
* http://www.apache.org/licenses/LICENSE-2.0
*
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*
*
*/
package org.apache.cassandra.cache;
import java.nio.ByteBuffer;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.Pair;
public class AutoSavingKeyCache<K extends Pair<Descriptor, DecoratedKey>, V> extends AutoSavingCache<K, V>
public class KeyCacheKey extends Pair<Descriptor, ByteBuffer> implements CacheKey
{
public AutoSavingKeyCache(ICache<K, V> cache, String tableName, String cfName)
public KeyCacheKey(Descriptor desc, ByteBuffer key)
{
super(cache, tableName, cfName, ColumnFamilyStore.CacheType.KEY_CACHE_TYPE);
super(desc, key);
}
@Override
public double getConfiguredCacheSize(CFMetaData cfm)
public ByteBuffer serializeForStorage()
{
return cfm == null ? CFMetaData.DEFAULT_KEY_CACHE_SIZE : cfm.getKeyCacheSize();
ByteBuffer bytes = ByteBuffer.allocate(serializedSize());
bytes.put(right.slice());
bytes.rewind();
return bytes;
}
@Override
public ByteBuffer translateKey(K key)
public Pair<String, String> getPathInfo()
{
return key.right.key;
return new Pair<String, String>(left.ksname, left.cfname);
}
public int serializedSize()
{
return right.remaining();
}
public String toString()
{
return String.format("KeyCacheKey(descriptor:%s, key:%s)", left, right);
}
}

96
src/java/org/apache/cassandra/cache/RowCacheKey.java vendored Normal file
View File

@ -0,0 +1,96 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*
*/
package org.apache.cassandra.cache;
import java.nio.ByteBuffer;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.Pair;
import org.apache.commons.lang.builder.HashCodeBuilder;
public class RowCacheKey implements CacheKey, Comparable<RowCacheKey>
{
public final int cfId;
public final ByteBuffer key;
public RowCacheKey(int cfId, DecoratedKey key)
{
this.cfId = cfId;
this.key = key.key;
}
public ByteBuffer serializeForStorage()
{
ByteBuffer bytes = ByteBuffer.allocate(serializedSize());
bytes.put(key.slice());
bytes.rewind();
return bytes;
}
public Pair<String, String> getPathInfo()
{
return Schema.instance.getCF(cfId);
}
public int serializedSize()
{
return key.remaining();
}
@Override
public int hashCode()
{
return new HashCodeBuilder(131, 56337)
.append(cfId)
.append(key).toHashCode();
}
@Override
public boolean equals(Object obj)
{
if (this == obj)
return true;
if (obj == null)
return false;
if (getClass() != obj.getClass())
return false;
RowCacheKey otherKey = (RowCacheKey) obj;
return cfId == otherKey.cfId && key.equals(otherKey.key);
}
@Override
public int compareTo(RowCacheKey otherKey)
{
return (cfId < otherKey.cfId) ? -1 : ((cfId == otherKey.cfId) ? ByteBufferUtil.compareUnsigned(key, otherKey.key) : 1);
}
@Override
public String toString()
{
return String.format("RowCacheKey(cfId:%d, key:%s)", cfId, key);
}
}

31
src/java/org/apache/cassandra/cache/SerializingCache.java vendored
View File

@ -28,11 +28,13 @@ import java.util.Set;
import com.googlecode.concurrentlinkedhashmap.ConcurrentLinkedHashMap;
import com.googlecode.concurrentlinkedhashmap.EvictionListener;
import com.googlecode.concurrentlinkedhashmap.Weigher;
import com.googlecode.concurrentlinkedhashmap.Weighers;
import org.apache.cassandra.io.ISerializer;
import org.apache.cassandra.io.util.MemoryInputStream;
import org.apache.cassandra.io.util.MemoryOutputStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -47,7 +49,7 @@ public class SerializingCache<K, V> implements ICache<K, V>
private final ConcurrentLinkedHashMap<K, FreeableMemory> map;
private final ISerializer<V> serializer;
public SerializingCache(int capacity, ISerializer<V> serializer, String tableName, String cfName)
public SerializingCache(int capacity, boolean useMemoryWeigher, ISerializer<V> serializer)
{
this.serializer = serializer;
@ -58,8 +60,11 @@ public class SerializingCache<K, V> implements ICache<K, V>
mem.unreference();
}
};
this.map = new ConcurrentLinkedHashMap.Builder<K, FreeableMemory>()
.weigher(Weighers.<FreeableMemory>singleton())
.weigher(useMemoryWeigher
? createMemoryWeigher()
: Weighers.<FreeableMemory>singleton())
.initialCapacity(capacity)
.maximumWeightedCapacity(capacity)
.concurrencyLevel(DEFAULT_CONCURENCY_LEVEL)
@ -67,6 +72,18 @@ public class SerializingCache<K, V> implements ICache<K, V>
.build();
}
private static Weigher<FreeableMemory> createMemoryWeigher()
{
return new Weigher<FreeableMemory>()
{
@Override
public int weightOf(FreeableMemory value)
{
return (int) Math.min(value.size(), Integer.MAX_VALUE);
}
};
}
private V deserialize(FreeableMemory mem)
{
try
@ -127,6 +144,11 @@ public class SerializingCache<K, V> implements ICache<K, V>
return map.size();
}
public int weightedSize()
{
return map.weightedSize();
}
public void clear()
{
map.clear();
@ -177,6 +199,11 @@ public class SerializingCache<K, V> implements ICache<K, V>
return map.descendingKeySetWithLimit(n);
}
public boolean containsKey(K key)
{
return map.containsKey(key);
}
public boolean isPutCopying()
{
return true;

5
src/java/org/apache/cassandra/cache/SerializingCacheProvider.java vendored
View File

@ -21,12 +21,11 @@ package org.apache.cassandra.cache;
*/
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.DecoratedKey;
public class SerializingCacheProvider implements IRowCacheProvider
{
public ICache<DecoratedKey, ColumnFamily> create(int capacity, String tableName, String cfName)
public ICache<RowCacheKey, ColumnFamily> create(int capacity, boolean useMemoryWeigher)
{
return new SerializingCache<DecoratedKey, ColumnFamily>(capacity, ColumnFamily.serializer(), tableName, cfName);
return new SerializingCache<RowCacheKey, ColumnFamily>(capacity, useMemoryWeigher, ColumnFamily.serializer());
}
}

29
src/java/org/apache/cassandra/cli/CliClient.java
View File

@ -1199,12 +1199,6 @@ public class CliClient
case COMMENT:
cfDef.setComment(CliUtils.unescapeSQLString(mValue));
break;
case ROWS_CACHED:
cfDef.setRow_cache_size(Double.parseDouble(mValue));
break;
case KEYS_CACHED:
cfDef.setKey_cache_size(Double.parseDouble(mValue));
break;
case READ_REPAIR_CHANCE:
double chance = Double.parseDouble(mValue);
@ -1226,15 +1220,6 @@ public class CliClient
break;
case MEMTABLE_THROUGHPUT:
break;
case ROW_CACHE_SAVE_PERIOD:
cfDef.setRow_cache_save_period_in_seconds(Integer.parseInt(mValue));
break;
case KEY_CACHE_SAVE_PERIOD:
cfDef.setKey_cache_save_period_in_seconds(Integer.parseInt(mValue));
break;
case ROW_CACHE_KEYS_TO_SAVE:
cfDef.setRow_cache_keys_to_save(Integer.parseInt(mValue));
break;
case DEFAULT_VALIDATION_CLASS:
cfDef.setDefault_validation_class(CliUtils.unescapeSQLString(mValue));
break;
@ -1247,9 +1232,6 @@ public class CliClient
case REPLICATE_ON_WRITE:
cfDef.setReplicate_on_write(Boolean.parseBoolean(mValue));
break;
case ROW_CACHE_PROVIDER:
cfDef.setRow_cache_provider(CliUtils.unescapeSQLString(mValue));
break;
case KEY_VALIDATION_CLASS:
cfDef.setKey_validation_class(CliUtils.unescapeSQLString(mValue));
break;
@ -1643,17 +1625,11 @@ public class CliClient
normaliseType(cfDef.default_validation_class, "org.apache.cassandra.db.marshal"));
writeAttr(sb, false, "key_validation_class",
normaliseType(cfDef.key_validation_class, "org.apache.cassandra.db.marshal"));
writeAttr(sb, false, "rows_cached", cfDef.row_cache_size);
writeAttr(sb, false, "row_cache_save_period", cfDef.row_cache_save_period_in_seconds);
writeAttr(sb, false, "row_cache_keys_to_save", cfDef.row_cache_keys_to_save);
writeAttr(sb, false, "keys_cached", cfDef.key_cache_size);
writeAttr(sb, false, "key_cache_save_period", cfDef.key_cache_save_period_in_seconds);
writeAttr(sb, false, "read_repair_chance", cfDef.read_repair_chance);
writeAttr(sb, false, "gc_grace", cfDef.gc_grace_seconds);
writeAttr(sb, false, "min_compaction_threshold", cfDef.min_compaction_threshold);
writeAttr(sb, false, "max_compaction_threshold", cfDef.max_compaction_threshold);
writeAttr(sb, false, "replicate_on_write", cfDef.replicate_on_write);
writeAttr(sb, false, "row_cache_provider", normaliseType(cfDef.row_cache_provider, "org.apache.cassandra.cache"));
writeAttr(sb, false, "compaction_strategy", cfDef.compaction_strategy);
if (!cfDef.compaction_strategy_options.isEmpty())
@ -1982,11 +1958,6 @@ public class CliClient
sessionState.out.printf(" Default column value validator: %s%n", cf_def.default_validation_class);
sessionState.out.printf(" Columns sorted by: %s%s%n", cf_def.comparator_type, cf_def.column_type.equals("Super") ? "/" + cf_def.subcomparator_type : "");
sessionState.out.printf(" Row cache size / save period in seconds / keys to save : %s/%s/%s%n",
cf_def.row_cache_size, cf_def.row_cache_save_period_in_seconds,
cf_def.row_cache_keys_to_save == Integer.MAX_VALUE ? "all" : cf_def.row_cache_keys_to_save);
sessionState.out.printf(" Row Cache Provider: %s%n", cf_def.getRow_cache_provider());
sessionState.out.printf(" Key cache size / save period in seconds: %s/%s%n", cf_def.key_cache_size, cf_def.key_cache_save_period_in_seconds);
sessionState.out.printf(" GC grace seconds: %s%n", cf_def.gc_grace_seconds);
sessionState.out.printf(" Compaction min/max thresholds: %s/%s%n", cf_def.min_compaction_threshold, cf_def.max_compaction_threshold);
sessionState.out.printf(" Read repair chance: %s%n", cf_def.read_repair_chance);

135
src/java/org/apache/cassandra/config/CFMetaData.java
View File

@ -28,7 +28,6 @@ import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.avro.util.Utf8;
import org.apache.cassandra.cache.IRowCacheProvider;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.compaction.AbstractCompactionStrategy;
import org.apache.cassandra.db.marshal.*;
@ -37,10 +36,7 @@ import org.apache.cassandra.db.migration.avro.ColumnDef;
import org.apache.cassandra.io.IColumnSerializer;
import org.apache.cassandra.io.compress.CompressionParameters;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.cache.ConcurrentLinkedHashCacheProvider;
import org.apache.cassandra.cache.SerializingCacheProvider;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -55,18 +51,12 @@ public final class CFMetaData
private static Logger logger = LoggerFactory.getLogger(CFMetaData.class);
public final static double DEFAULT_ROW_CACHE_SIZE = 0.0;
public final static double DEFAULT_KEY_CACHE_SIZE = 200000;
public final static double DEFAULT_READ_REPAIR_CHANCE = 0.1;
public final static boolean DEFAULT_REPLICATE_ON_WRITE = true;
public final static int DEFAULT_ROW_CACHE_SAVE_PERIOD_IN_SECONDS = 0;
public final static int DEFAULT_KEY_CACHE_SAVE_PERIOD_IN_SECONDS = 4 * 3600;
public final static int DEFAULT_ROW_CACHE_KEYS_TO_SAVE = Integer.MAX_VALUE;
public final static int DEFAULT_GC_GRACE_SECONDS = 864000;
public final static int DEFAULT_MIN_COMPACTION_THRESHOLD = 4;
public final static int DEFAULT_MAX_COMPACTION_THRESHOLD = 32;
public final static double DEFAULT_MERGE_SHARDS_CHANCE = 0.1;
public final static IRowCacheProvider DEFAULT_ROW_CACHE_PROVIDER = new SerializingCacheProvider();
public final static String DEFAULT_COMPACTION_STRATEGY_CLASS = "SizeTieredCompactionStrategy";
public final static ByteBuffer DEFAULT_KEY_NAME = ByteBufferUtil.bytes("KEY");
@ -106,8 +96,6 @@ public final class CFMetaData
//OPTIONAL
private String comment; // default none, for humans only
private double rowCacheSize; // default 0
private double keyCacheSize; // default 0.01
private double readRepairChance; // default 1.0 (always), chance [0.0,1.0] of read repair
private boolean replicateOnWrite; // default false
private int gcGraceSeconds; // default 864000 (ten days)
@ -115,13 +103,9 @@ public final class CFMetaData
private AbstractType keyValidator; // default BytesType (no-op), use comparator types
private int minCompactionThreshold; // default 4
private int maxCompactionThreshold; // default 32
private int rowCacheSavePeriodInSeconds; // default 0 (off)
private int keyCacheSavePeriodInSeconds; // default 3600 (1 hour)
private int rowCacheKeysToSave; // default max int (aka feature is off)
// mergeShardsChance is now obsolete, but left here so as to not break
// thrift compatibility
private double mergeShardsChance; // default 0.1, chance [0.0, 1.0] of merging old shards during replication
private IRowCacheProvider rowCacheProvider;
private ByteBuffer keyAlias; // default NULL
private Map<ByteBuffer, ColumnDefinition> column_metadata;
@ -131,8 +115,6 @@ public final class CFMetaData
private CompressionParameters compressionParameters;
public CFMetaData comment(String prop) { comment = enforceCommentNotNull(prop); return this;}
public CFMetaData rowCacheSize(double prop) {rowCacheSize = prop; return this;}
public CFMetaData keyCacheSize(double prop) {keyCacheSize = prop; return this;}
public CFMetaData readRepairChance(double prop) {readRepairChance = prop; return this;}
public CFMetaData replicateOnWrite(boolean prop) {replicateOnWrite = prop; return this;}
public CFMetaData gcGraceSeconds(int prop) {gcGraceSeconds = prop; return this;}
@ -140,13 +122,9 @@ public final class CFMetaData
public CFMetaData keyValidator(AbstractType prop) {keyValidator = prop; return this;}
public CFMetaData minCompactionThreshold(int prop) {minCompactionThreshold = prop; return this;}
public CFMetaData maxCompactionThreshold(int prop) {maxCompactionThreshold = prop; return this;}
public CFMetaData rowCacheSavePeriod(int prop) {rowCacheSavePeriodInSeconds = prop; return this;}
public CFMetaData keyCacheSavePeriod(int prop) {keyCacheSavePeriodInSeconds = prop; return this;}
public CFMetaData rowCacheKeysToSave(int prop) {rowCacheKeysToSave = prop; return this;}
public CFMetaData mergeShardsChance(double prop) {mergeShardsChance = prop; return this;}
public CFMetaData keyAlias(ByteBuffer prop) {keyAlias = prop; return this;}
public CFMetaData columnMetadata(Map<ByteBuffer,ColumnDefinition> prop) {column_metadata = prop; return this;}
public CFMetaData rowCacheProvider(IRowCacheProvider prop) { rowCacheProvider = prop; return this;}
public CFMetaData compactionStrategyClass(Class<? extends AbstractCompactionStrategy> prop) {compactionStrategyClass = prop; return this;}
public CFMetaData compactionStrategyOptions(Map<String, String> prop) {compactionStrategyOptions = prop; return this;}
public CFMetaData compressionParameters(CompressionParameters prop) {compressionParameters = prop; return this;}
@ -185,16 +163,12 @@ public final class CFMetaData
private void init()
{
// Set a bunch of defaults
rowCacheSize = DEFAULT_ROW_CACHE_SIZE;
keyCacheSize = DEFAULT_KEY_CACHE_SIZE;
rowCacheKeysToSave = DEFAULT_ROW_CACHE_KEYS_TO_SAVE;
readRepairChance = DEFAULT_READ_REPAIR_CHANCE;
replicateOnWrite = DEFAULT_REPLICATE_ON_WRITE;
gcGraceSeconds = DEFAULT_GC_GRACE_SECONDS;
minCompactionThreshold = DEFAULT_MIN_COMPACTION_THRESHOLD;
maxCompactionThreshold = DEFAULT_MAX_COMPACTION_THRESHOLD;
mergeShardsChance = DEFAULT_MERGE_SHARDS_CHANCE;
rowCacheProvider = DEFAULT_ROW_CACHE_PROVIDER;
// Defaults strange or simple enough to not need a DEFAULT_T for
defaultValidator = BytesType.instance;
@ -222,19 +196,15 @@ public final class CFMetaData
CFMetaData newCFMD = new CFMetaData(Table.SYSTEM_TABLE, cfName, type, comparator, subcc, cfId);
return newCFMD.comment(comment)
.keyCacheSize(0.01)
.readRepairChance(0)
.gcGraceSeconds(0)
.mergeShardsChance(0.0)
.rowCacheSavePeriod(0)
.keyCacheSavePeriod(0);
.mergeShardsChance(0.0);
}
public static CFMetaData newIndexMetadata(CFMetaData parent, ColumnDefinition info, AbstractType columnComparator)
{
return new CFMetaData(parent.ksName, parent.indexColumnFamilyName(info), ColumnFamilyType.Standard, columnComparator, null)
.keyValidator(info.getValidator())
.keyCacheSize(0.0)
.readRepairChance(0.0)
.gcGraceSeconds(parent.gcGraceSeconds)
.minCompactionThreshold(parent.minCompactionThreshold)
@ -256,17 +226,12 @@ public final class CFMetaData
private static CFMetaData copyOpts(CFMetaData newCFMD, CFMetaData oldCFMD)
{
return newCFMD.comment(oldCFMD.comment)
.rowCacheSize(oldCFMD.rowCacheSize)
.keyCacheSize(oldCFMD.keyCacheSize)
.readRepairChance(oldCFMD.readRepairChance)
.replicateOnWrite(oldCFMD.replicateOnWrite)
.gcGraceSeconds(oldCFMD.gcGraceSeconds)
.defaultValidator(oldCFMD.defaultValidator)
.minCompactionThreshold(oldCFMD.minCompactionThreshold)
.maxCompactionThreshold(oldCFMD.maxCompactionThreshold)
.rowCacheSavePeriod(oldCFMD.rowCacheSavePeriodInSeconds)
.keyCacheSavePeriod(oldCFMD.keyCacheSavePeriodInSeconds)
.rowCacheKeysToSave(oldCFMD.rowCacheKeysToSave)
.columnMetadata(oldCFMD.column_metadata)
.compactionStrategyClass(oldCFMD.compactionStrategyClass)
.compactionStrategyOptions(oldCFMD.compactionStrategyOptions)
@ -303,8 +268,6 @@ public final class CFMetaData
cf.subcomparator_type = new Utf8(subcolumnComparator.toString());
}
cf.comment = new Utf8(enforceCommentNotNull(comment));
cf.row_cache_size = rowCacheSize;
cf.key_cache_size = keyCacheSize;
cf.read_repair_chance = readRepairChance;
cf.replicate_on_write = replicateOnWrite;
cf.gc_grace_seconds = gcGraceSeconds;
@ -312,15 +275,11 @@ public final class CFMetaData
cf.key_validation_class = new Utf8(keyValidator.toString());
cf.min_compaction_threshold = minCompactionThreshold;
cf.max_compaction_threshold = maxCompactionThreshold;
cf.row_cache_save_period_in_seconds = rowCacheSavePeriodInSeconds;
cf.key_cache_save_period_in_seconds = keyCacheSavePeriodInSeconds;
cf.row_cache_keys_to_save = rowCacheKeysToSave;
cf.merge_shards_chance = mergeShardsChance;
cf.key_alias = keyAlias;
cf.column_metadata = new ArrayList<ColumnDef>(column_metadata.size());
for (ColumnDefinition cd : column_metadata.values())
cf.column_metadata.add(cd.toAvro());
cf.row_cache_provider = new Utf8(rowCacheProvider.getClass().getName());
cf.compaction_strategy = new Utf8(compactionStrategyClass.getName());
if (compactionStrategyOptions != null)
{
@ -373,24 +332,7 @@ public final class CFMetaData
// Isn't AVRO supposed to handle stuff like this?
if (cf.min_compaction_threshold != null) { newCFMD.minCompactionThreshold(cf.min_compaction_threshold); }
if (cf.max_compaction_threshold != null) { newCFMD.maxCompactionThreshold(cf.max_compaction_threshold); }
if (cf.row_cache_save_period_in_seconds != null) { newCFMD.rowCacheSavePeriod(cf.row_cache_save_period_in_seconds); }
if (cf.key_cache_save_period_in_seconds != null) { newCFMD.keyCacheSavePeriod(cf.key_cache_save_period_in_seconds); }
if (cf.row_cache_keys_to_save != null) { newCFMD.rowCacheKeysToSave(cf.row_cache_keys_to_save); }
if (cf.merge_shards_chance != null) { newCFMD.mergeShardsChance(cf.merge_shards_chance); }
if (cf.row_cache_provider != null)
{
try
{
newCFMD.rowCacheProvider(FBUtilities.newCacheProvider(cf.row_cache_provider.toString()));
}
catch (ConfigurationException e)
{
// default was already set upon newCFMD init
logger.warn("Unable to instantiate cache provider {}; using default {} instead",
cf.row_cache_provider,
DEFAULT_ROW_CACHE_PROVIDER);
}
}
if (cf.key_alias != null) { newCFMD.keyAlias(cf.key_alias); }
if (cf.compaction_strategy != null)
{
@ -420,8 +362,6 @@ public final class CFMetaData
}
return newCFMD.comment(cf.comment.toString())
.rowCacheSize(cf.row_cache_size)
.keyCacheSize(cf.key_cache_size)
.readRepairChance(cf.read_repair_chance)
.replicateOnWrite(cf.replicate_on_write)
.gcGraceSeconds(cf.gc_grace_seconds)
@ -435,17 +375,7 @@ public final class CFMetaData
{
return comment;
}
public double getRowCacheSize()
{
return rowCacheSize;
}
public double getKeyCacheSize()
{
return keyCacheSize;
}
public double getReadRepairChance()
{
return readRepairChance;
@ -486,26 +416,6 @@ public final class CFMetaData
return maxCompactionThreshold;
}
public int getRowCacheSavePeriodInSeconds()
{
return rowCacheSavePeriodInSeconds;
}
public int getKeyCacheSavePeriodInSeconds()
{
return keyCacheSavePeriodInSeconds;
}
public int getRowCacheKeysToSave()
{
return rowCacheKeysToSave;
}
public IRowCacheProvider getRowCacheProvider()
{
return rowCacheProvider;
}
public ByteBuffer getKeyName()
{
return keyAlias == null ? DEFAULT_KEY_NAME : keyAlias;
@ -545,8 +455,6 @@ public final class CFMetaData
.append(comparator, rhs.comparator)
.append(subcolumnComparator, rhs.subcolumnComparator)
.append(comment, rhs.comment)
.append(rowCacheSize, rhs.rowCacheSize)
.append(keyCacheSize, rhs.keyCacheSize)
.append(readRepairChance, rhs.readRepairChance)
.append(replicateOnWrite, rhs.replicateOnWrite)
.append(gcGraceSeconds, rhs.gcGraceSeconds)
@ -556,9 +464,6 @@ public final class CFMetaData
.append(maxCompactionThreshold, rhs.maxCompactionThreshold)
.append(cfId.intValue(), rhs.cfId.intValue())
.append(column_metadata, rhs.column_metadata)
.append(rowCacheSavePeriodInSeconds, rhs.rowCacheSavePeriodInSeconds)
.append(keyCacheSavePeriodInSeconds, rhs.keyCacheSavePeriodInSeconds)
.append(rowCacheKeysToSave, rhs.rowCacheKeysToSave)
.append(mergeShardsChance, rhs.mergeShardsChance)
.append(keyAlias, rhs.keyAlias)
.append(compactionStrategyClass, rhs.compactionStrategyClass)
@ -576,8 +481,6 @@ public final class CFMetaData
.append(comparator)
.append(subcolumnComparator)
.append(comment)
.append(rowCacheSize)
.append(keyCacheSize)
.append(readRepairChance)
.append(replicateOnWrite)
.append(gcGraceSeconds)
@ -587,9 +490,6 @@ public final class CFMetaData
.append(maxCompactionThreshold)
.append(cfId)
.append(column_metadata)
.append(rowCacheSavePeriodInSeconds)
.append(keyCacheSavePeriodInSeconds)
.append(rowCacheKeysToSave)
.append(mergeShardsChance)
.append(keyAlias)
.append(compactionStrategyClass)
@ -621,12 +521,6 @@ public final class CFMetaData
cf_def.setMin_compaction_threshold(CFMetaData.DEFAULT_MIN_COMPACTION_THRESHOLD);
if (!cf_def.isSetMax_compaction_threshold())
cf_def.setMax_compaction_threshold(CFMetaData.DEFAULT_MAX_COMPACTION_THRESHOLD);
if (!cf_def.isSetRow_cache_save_period_in_seconds())
cf_def.setRow_cache_save_period_in_seconds(CFMetaData.DEFAULT_ROW_CACHE_SAVE_PERIOD_IN_SECONDS);
if (!cf_def.isSetKey_cache_save_period_in_seconds())
cf_def.setKey_cache_save_period_in_seconds(CFMetaData.DEFAULT_KEY_CACHE_SAVE_PERIOD_IN_SECONDS);
if (!cf_def.isSetRow_cache_keys_to_save())
cf_def.setRow_cache_keys_to_save(CFMetaData.DEFAULT_ROW_CACHE_KEYS_TO_SAVE);
if (!cf_def.isSetMerge_shards_chance())
cf_def.setMerge_shards_chance(CFMetaData.DEFAULT_MERGE_SHARDS_CHANCE);
if (null == cf_def.compaction_strategy)
@ -655,11 +549,7 @@ public final class CFMetaData
if (cf_def.isSetGc_grace_seconds()) { newCFMD.gcGraceSeconds(cf_def.gc_grace_seconds); }
if (cf_def.isSetMin_compaction_threshold()) { newCFMD.minCompactionThreshold(cf_def.min_compaction_threshold); }
if (cf_def.isSetMax_compaction_threshold()) { newCFMD.maxCompactionThreshold(cf_def.max_compaction_threshold); }
if (cf_def.isSetRow_cache_save_period_in_seconds()) { newCFMD.rowCacheSavePeriod(cf_def.row_cache_save_period_in_seconds); }
if (cf_def.isSetKey_cache_save_period_in_seconds()) { newCFMD.keyCacheSavePeriod(cf_def.key_cache_save_period_in_seconds); }
if (cf_def.isSetRow_cache_keys_to_save()) { newCFMD.rowCacheKeysToSave(cf_def.row_cache_keys_to_save); }
if (cf_def.isSetMerge_shards_chance()) { newCFMD.mergeShardsChance(cf_def.merge_shards_chance); }
if (cf_def.isSetRow_cache_provider()) { newCFMD.rowCacheProvider(FBUtilities.newCacheProvider(cf_def.row_cache_provider)); }
if (cf_def.isSetKey_alias()) { newCFMD.keyAlias(cf_def.key_alias); }
if (cf_def.isSetKey_validation_class()) { newCFMD.keyValidator(TypeParser.parse(cf_def.key_validation_class)); }
if (cf_def.isSetCompaction_strategy())
@ -670,8 +560,6 @@ public final class CFMetaData
CompressionParameters cp = CompressionParameters.create(cf_def.compression_options);
return newCFMD.comment(cf_def.comment)
.rowCacheSize(cf_def.row_cache_size)
.keyCacheSize(cf_def.key_cache_size)
.readRepairChance(cf_def.read_repair_chance)
.replicateOnWrite(cf_def.replicate_on_write)
.defaultValidator(TypeParser.parse(cf_def.default_validation_class))
@ -712,8 +600,6 @@ public final class CFMetaData
validateMinMaxCompactionThresholds(cf_def);
comment = enforceCommentNotNull(cf_def.comment);
rowCacheSize = cf_def.row_cache_size;
keyCacheSize = cf_def.key_cache_size;
readRepairChance = cf_def.read_repair_chance;
replicateOnWrite = cf_def.replicate_on_write;
gcGraceSeconds = cf_def.gc_grace_seconds;
@ -721,12 +607,7 @@ public final class CFMetaData
keyValidator = TypeParser.parse(cf_def.key_validation_class);
minCompactionThreshold = cf_def.min_compaction_threshold;
maxCompactionThreshold = cf_def.max_compaction_threshold;
rowCacheSavePeriodInSeconds = cf_def.row_cache_save_period_in_seconds;
keyCacheSavePeriodInSeconds = cf_def.key_cache_save_period_in_seconds;
rowCacheKeysToSave = cf_def.row_cache_keys_to_save;
mergeShardsChance = cf_def.merge_shards_chance;
if (cf_def.row_cache_provider != null)
rowCacheProvider = FBUtilities.newCacheProvider(cf_def.row_cache_provider.toString());
keyAlias = cf_def.key_alias;
// adjust column definitions. figure out who is coming and going.
@ -843,8 +724,6 @@ public final class CFMetaData
def.setSubcomparator_type(subcolumnComparator.toString());
}
def.setComment(enforceCommentNotNull(comment));
def.setRow_cache_size(rowCacheSize);
def.setKey_cache_size(keyCacheSize);
def.setRead_repair_chance(readRepairChance);
def.setReplicate_on_write(replicateOnWrite);
def.setGc_grace_seconds(gcGraceSeconds);
@ -852,10 +731,6 @@ public final class CFMetaData
def.setKey_validation_class(keyValidator.toString());
def.setMin_compaction_threshold(minCompactionThreshold);
def.setMax_compaction_threshold(maxCompactionThreshold);
def.setRow_cache_save_period_in_seconds(rowCacheSavePeriodInSeconds);
def.setKey_cache_save_period_in_seconds(keyCacheSavePeriodInSeconds);
def.setRow_cache_keys_to_save(rowCacheKeysToSave);
def.setRow_cache_provider(rowCacheProvider.getClass().getName());
def.setMerge_shards_chance(mergeShardsChance);
def.setKey_alias(getKeyName());
List<org.apache.cassandra.thrift.ColumnDef> column_meta = new ArrayList<org.apache.cassandra.thrift.ColumnDef>(column_metadata.size());
@ -996,8 +871,6 @@ public final class CFMetaData
.append("comparator", comparator)
.append("subcolumncomparator", subcolumnComparator)
.append("comment", comment)
.append("rowCacheSize", rowCacheSize)
.append("keyCacheSize", keyCacheSize)
.append("readRepairChance", readRepairChance)
.append("replicateOnWrite", replicateOnWrite)
.append("gcGraceSeconds", gcGraceSeconds)
@ -1005,10 +878,6 @@ public final class CFMetaData
.append("keyValidator", keyValidator)
.append("minCompactionThreshold", minCompactionThreshold)
.append("maxCompactionThreshold", maxCompactionThreshold)
.append("rowCacheSavePeriodInSeconds", rowCacheSavePeriodInSeconds)
.append("keyCacheSavePeriodInSeconds", keyCacheSavePeriodInSeconds)
.append("rowCacheKeysToSave", rowCacheKeysToSave)
.append("rowCacheProvider", rowCacheProvider)
.append("mergeShardsChance", mergeShardsChance)
.append("keyAlias", keyAlias)
.append("column_metadata", column_metadata)

12
src/java/org/apache/cassandra/config/Config.java
View File

@ -20,6 +20,9 @@ package org.apache.cassandra.config;
*
*/
import org.apache.cassandra.cache.ConcurrentLinkedHashCacheProvider;
import org.apache.cassandra.cache.IRowCacheProvider;
import java.util.List;
@ -122,6 +125,15 @@ public class Config
public boolean incremental_backups = false;
public int memtable_flush_queue_size = 4;
public int key_cache_size_in_mb = 2;
public int key_cache_save_period = 14400;
public int key_cache_keys_to_save = Integer.MAX_VALUE;
public int row_cache_size_in_mb = 0;
public int row_cache_save_period = 0;
public int row_cache_keys_to_save = Integer.MAX_VALUE;
public String row_cache_provider = ConcurrentLinkedHashCacheProvider.class.getSimpleName();
public static enum CommitLogSync {
periodic,
batch

44
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -34,8 +34,8 @@ import org.apache.cassandra.auth.AllowAllAuthenticator;
import org.apache.cassandra.auth.AllowAllAuthority;
import org.apache.cassandra.auth.IAuthenticator;
import org.apache.cassandra.auth.IAuthority;
import org.apache.cassandra.cache.IRowCacheProvider;
import org.apache.cassandra.config.Config.RequestSchedulerId;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DefsTable;
import org.apache.cassandra.db.migration.Migration;
import org.apache.cassandra.dht.IPartitioner;
@ -47,6 +47,7 @@ import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.locator.SeedProvider;
import org.apache.cassandra.scheduler.IRequestScheduler;
import org.apache.cassandra.scheduler.NoScheduler;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.thrift.CassandraDaemon;
import org.apache.cassandra.utils.FBUtilities;
import org.yaml.snakeyaml.Loader;
@ -82,6 +83,8 @@ public class DatabaseDescriptor
private static RequestSchedulerId requestSchedulerId;
private static RequestSchedulerOptions requestSchedulerOptions;
private static IRowCacheProvider rowCacheProvider;
/**
* Inspect the classpath to find storage configuration file
*/
@ -413,6 +416,8 @@ public class DatabaseDescriptor
if (conf.initial_token != null)
partitioner.getTokenFactory().validate(conf.initial_token);
rowCacheProvider = FBUtilities.newCacheProvider(conf.row_cache_provider);
// Hardcoded system tables
KSMetaData systemMeta = KSMetaData.systemKeyspace();
Schema.instance.load(CFMetaData.StatusCf);
@ -919,7 +924,7 @@ public class DatabaseDescriptor
return conf.index_interval;
}
public static File getSerializedCachePath(String ksName, String cfName, ColumnFamilyStore.CacheType cacheType)
public static File getSerializedCachePath(String ksName, String cfName, CacheService.CacheType cacheType)
{
return new File(conf.saved_caches_directory + File.separator + ksName + "-" + cfName + "-" + cacheType);
}
@ -1022,4 +1027,39 @@ public class DatabaseDescriptor
{
return conf.commitlog_total_space_in_mb;
}
public static int getKeyCacheSizeInMB()
{
return conf.key_cache_size_in_mb;
}
public static int getKeyCacheSavePeriod()
{
return conf.key_cache_save_period;
}
public static int getKeyCacheKeysToSave()
{
return conf.key_cache_keys_to_save;
}
public static int getRowCacheSizeInMB()
{
return conf.row_cache_size_in_mb;
}
public static int getRowCacheSavePeriod()
{
return conf.row_cache_save_period;
}
public static int getRowCacheKeysToSave()
{
return conf.row_cache_keys_to_save;
}
public static IRowCacheProvider getRowCacheProvider()
{
return rowCacheProvider;
}
}

22
src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java
View File

@ -50,18 +50,13 @@ public class CreateColumnFamilyStatement
private static final String KW_COMPARATOR = "comparator";
private static final String KW_COMMENT = "comment";
private static final String KW_ROWCACHESIZE = "row_cache_size";
private static final String KW_KEYCACHESIZE = "key_cache_size";
private static final String KW_READREPAIRCHANCE = "read_repair_chance";
private static final String KW_GCGRACESECONDS = "gc_grace_seconds";
private static final String KW_DEFAULTVALIDATION = "default_validation";
private static final String KW_MINCOMPACTIONTHRESHOLD = "min_compaction_threshold";
private static final String KW_MAXCOMPACTIONTHRESHOLD = "max_compaction_threshold";
private static final String KW_ROWCACHESAVEPERIODSECS = "row_cache_save_period_in_seconds";
private static final String KW_KEYCACHESAVEPERIODSECS = "key_cache_save_period_in_seconds";
private static final String KW_REPLICATEONWRITE = "replicate_on_write";
private static final String KW_ROW_CACHE_PROVIDER = "row_cache_provider";
// Maps CQL short names to the respective Cassandra comparator/validator class names
public static final Map<String, String> comparators = new HashMap<String, String>();
private static final Set<String> keywords = new HashSet<String>();
@ -86,21 +81,21 @@ public class CreateColumnFamilyStatement
keywords.add(KW_COMPARATOR);
keywords.add(KW_COMMENT);
keywords.add(KW_ROWCACHESIZE);
keywords.add(KW_KEYCACHESIZE);
keywords.add(KW_READREPAIRCHANCE);
keywords.add(KW_GCGRACESECONDS);
keywords.add(KW_DEFAULTVALIDATION);
keywords.add(KW_MINCOMPACTIONTHRESHOLD);
keywords.add(KW_MAXCOMPACTIONTHRESHOLD);
keywords.add(KW_ROWCACHESAVEPERIODSECS);
keywords.add(KW_KEYCACHESAVEPERIODSECS);
keywords.add(KW_REPLICATEONWRITE);
keywords.add(KW_ROW_CACHE_PROVIDER);
obsoleteKeywords.add("row_cache_size");
obsoleteKeywords.add("key_cache_size");
obsoleteKeywords.add("row_cache_save_period_in_seconds");
obsoleteKeywords.add("key_cache_save_period_in_seconds");
obsoleteKeywords.add("memtable_throughput_in_mb");
obsoleteKeywords.add("memtable_operations_in_millions");
obsoleteKeywords.add("memtable_flush_after_mins");
obsoleteKeywords.add("row_cache_provider");
}
private final String name;
@ -287,20 +282,15 @@ public class CreateColumnFamilyStatement
null);
newCFMD.comment(properties.get(KW_COMMENT))
.rowCacheSize(getPropertyDouble(KW_ROWCACHESIZE, CFMetaData.DEFAULT_ROW_CACHE_SIZE))
.keyCacheSize(getPropertyDouble(KW_KEYCACHESIZE, CFMetaData.DEFAULT_KEY_CACHE_SIZE))
.readRepairChance(getPropertyDouble(KW_READREPAIRCHANCE, CFMetaData.DEFAULT_READ_REPAIR_CHANCE))
.replicateOnWrite(getPropertyBoolean(KW_REPLICATEONWRITE, CFMetaData.DEFAULT_REPLICATE_ON_WRITE))
.gcGraceSeconds(getPropertyInt(KW_GCGRACESECONDS, CFMetaData.DEFAULT_GC_GRACE_SECONDS))
.defaultValidator(getValidator())
.minCompactionThreshold(getPropertyInt(KW_MINCOMPACTIONTHRESHOLD, CFMetaData.DEFAULT_MIN_COMPACTION_THRESHOLD))
.maxCompactionThreshold(getPropertyInt(KW_MAXCOMPACTIONTHRESHOLD, CFMetaData.DEFAULT_MAX_COMPACTION_THRESHOLD))
.rowCacheSavePeriod(getPropertyInt(KW_ROWCACHESAVEPERIODSECS, CFMetaData.DEFAULT_ROW_CACHE_SAVE_PERIOD_IN_SECONDS))
.keyCacheSavePeriod(getPropertyInt(KW_KEYCACHESAVEPERIODSECS, CFMetaData.DEFAULT_KEY_CACHE_SAVE_PERIOD_IN_SECONDS))
.mergeShardsChance(0.0)
.columnMetadata(getColumns(comparator))
.keyValidator(TypeParser.parse(comparators.get(getKeyType())))
.rowCacheProvider(FBUtilities.newCacheProvider(getPropertyString(KW_ROW_CACHE_PROVIDER, CFMetaData.DEFAULT_ROW_CACHE_PROVIDER.getClass().getName())))
.keyAlias(keyAlias)
.validate();
}

239
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -25,11 +25,15 @@ import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.regex.Pattern;
import javax.management.*;
import com.google.common.collect.Iterables;
import org.apache.cassandra.db.compaction.LeveledManifest;
import org.apache.cassandra.service.CacheService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -123,32 +127,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
private volatile DefaultInteger minCompactionThreshold;
private volatile DefaultInteger maxCompactionThreshold;
private volatile AbstractCompactionStrategy compactionStrategy;
private volatile DefaultInteger rowCacheSaveInSeconds;
private volatile DefaultInteger keyCacheSaveInSeconds;
private volatile DefaultInteger rowCacheKeysToSave;
public static enum CacheType
{
KEY_CACHE_TYPE("KeyCache"),
ROW_CACHE_TYPE("RowCache");
public final String name;
private CacheType(String name)
{
this.name = name;
}
@Override
public String toString()
{
return name;
}
}
public final AutoSavingCache<Pair<Descriptor,DecoratedKey>, Long> keyCache;
public final AutoSavingCache<DecoratedKey, ColumnFamily> rowCache;
/** ratio of in-memory memtable size, to serialized size */
volatile double liveRatio = 1.0;
@ -166,18 +144,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
if (!maxCompactionThreshold.isModified())
for (ColumnFamilyStore cfs : concatWithIndexes())
cfs.maxCompactionThreshold = new DefaultInteger(metadata.getMaxCompactionThreshold());
if (!rowCacheSaveInSeconds.isModified())
rowCacheSaveInSeconds = new DefaultInteger(metadata.getRowCacheSavePeriodInSeconds());
if (!keyCacheSaveInSeconds.isModified())
keyCacheSaveInSeconds = new DefaultInteger(metadata.getKeyCacheSavePeriodInSeconds());
if (!rowCacheKeysToSave.isModified())
rowCacheKeysToSave = new DefaultInteger(metadata.getRowCacheKeysToSave());
maybeReloadCompactionStrategy();
updateCacheSizes();
scheduleCacheSaving(rowCacheSaveInSeconds.value(), keyCacheSaveInSeconds.value(), rowCacheKeysToSave.value());
indexManager.reload();
}
@ -203,14 +172,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
private ColumnFamilyStore(Table table, String columnFamilyName, IPartitioner partitioner, int generation, CFMetaData metadata)
{
assert metadata != null : "null metadata for " + table + ":" + columnFamilyName;
this.table = table;
columnFamily = columnFamilyName;
this.metadata = metadata;
this.minCompactionThreshold = new DefaultInteger(metadata.getMinCompactionThreshold());
this.maxCompactionThreshold = new DefaultInteger(metadata.getMaxCompactionThreshold());
this.rowCacheSaveInSeconds = new DefaultInteger(metadata.getRowCacheSavePeriodInSeconds());
this.keyCacheSaveInSeconds = new DefaultInteger(metadata.getKeyCacheSavePeriodInSeconds());
this.rowCacheKeysToSave = new DefaultInteger(metadata.getRowCacheKeysToSave());
this.partitioner = partitioner;
this.indexManager = new SecondaryIndexManager(this);
fileIndexGenerator.set(generation);
@ -218,14 +185,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
if (logger.isDebugEnabled())
logger.debug("Starting CFS {}", columnFamily);
ICache<Pair<Descriptor, DecoratedKey>, Long> kc = ConcurrentLinkedHashCache.create(0, table.name, columnFamilyName);
keyCache = new AutoSavingKeyCache<Pair<Descriptor, DecoratedKey>, Long>(kc, table.name, columnFamilyName);
ICache<DecoratedKey, ColumnFamily> rc = metadata.getRowCacheProvider().create(0, table.name, columnFamilyName);
rowCache = new AutoSavingRowCache<DecoratedKey, ColumnFamily>(rc, table.name, columnFamilyName);
// scan for sstables corresponding to this cf and load them
data = new DataTracker(this);
Set<DecoratedKey> savedKeys = keyCache.readSaved();
Set<DecoratedKey> savedKeys = CacheService.instance.keyCache.readSaved(table.name, columnFamily);
Set<Map.Entry<Descriptor, Set<Component>>> entries = files(table.name, columnFamilyName, false, false).entrySet();
data.addInitialSSTables(SSTableReader.batchOpen(entries, savedKeys, data, metadata, this.partitioner));
@ -402,39 +364,30 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
// must be called after all sstables are loaded since row cache merges all row versions
public void initCaches()
public void initRowCache()
{
long start = System.currentTimeMillis();
AutoSavingCache<RowCacheKey, ColumnFamily> rowCache = CacheService.instance.rowCache;
// results are sorted on read (via treeset) because there are few reads and many writes and reads only happen at startup
int cachedRowsRead = 0;
for (DecoratedKey key : rowCache.readSaved())
for (DecoratedKey key : rowCache.readSaved(table.name, columnFamily))
{
cacheRow(key);
if (cachedRowsRead++ > rowCache.getCapacity())
{
logger.debug(String.format("Stopped loading row cache after capacity %d was reached", rowCache.getCapacity()));
break;
}
cacheRow(metadata.cfId, key);
}
if (rowCache.size() > 0)
if (cachedRowsRead > 0)
logger.info(String.format("completed loading (%d ms; %d keys) row cache for %s.%s",
System.currentTimeMillis()-start,
rowCache.size(),
table.name,
columnFamily));
scheduleCacheSaving(metadata.getRowCacheSavePeriodInSeconds(), metadata.getKeyCacheSavePeriodInSeconds(), metadata.getRowCacheKeysToSave());
System.currentTimeMillis() - start,
cachedRowsRead,
table.name,
columnFamily));
}
public void scheduleCacheSaving(int rowCacheSavePeriodInSeconds, int keyCacheSavePeriodInSeconds, int rowCacheKeysToSave)
public AutoSavingCache<KeyCacheKey, Long> getKeyCache()
{
keyCache.scheduleSaving(keyCacheSavePeriodInSeconds, Integer.MAX_VALUE);
rowCache.scheduleSaving(rowCacheSavePeriodInSeconds, rowCacheKeysToSave);
}
public AutoSavingCache<Pair<Descriptor,DecoratedKey>, Long> getKeyCache()
{
return keyCache;
return CacheService.instance.keyCache;
}
/**
@ -520,7 +473,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
Set<Descriptor> currentDescriptors = new HashSet<Descriptor>();
// going to hold new SSTable view of the CFS containing old and new SSTables
Set<SSTableReader> sstables = new HashSet<SSTableReader>();
Set<DecoratedKey> savedKeys = keyCache.readSaved();
// get the max generation number, to prevent generation conflicts
int generation = 0;
@ -553,8 +505,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
descriptor));
logger.info("Initializing new SSTable {}", rawSSTable);
try
{
Set<DecoratedKey> savedKeys = CacheService.instance.keyCache.readSaved(descriptor.ksname, descriptor.cfname);
reader = SSTableReader.open(rawSSTable.getKey(), rawSSTable.getValue(), savedKeys, data, metadata, partitioner);
}
catch (IOException e)
@ -580,7 +534,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
logger.info("Loading new SSTables and building secondary indexes for " + table.name + "/" + columnFamily + ": " + sstables);
SSTableReader.acquireReferences(sstables);
data.addSSTables(sstables); // this will call updateCacheSizes() for us
data.addSSTables(sstables);
try
{
indexManager.maybeBuildSecondaryIndexes(sstables, indexManager.getIndexedColumns());
@ -756,13 +710,19 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public void updateRowCache(DecoratedKey key, ColumnFamily columnFamily)
{
if (rowCache.isPutCopying())
Integer cfId = Schema.instance.getId(table.name, this.columnFamily);
if (cfId == null)
return; // secondary index
RowCacheKey cacheKey = new RowCacheKey(cfId, key);
if (CacheService.instance.rowCache.isPutCopying())
{
invalidateCachedRow(key);
invalidateCachedRow(cacheKey);
}
else
{
ColumnFamily cachedRow = getRawCachedRow(key);
ColumnFamily cachedRow = getRawCachedRow(cacheKey);
if (cachedRow != null)
cachedRow.addAll(columnFamily, HeapAllocator.instance);
}
@ -1141,19 +1101,26 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
return (int) (System.currentTimeMillis() / 1000) - metadata.getGcGraceSeconds();
}
private ColumnFamily cacheRow(DecoratedKey key)
public ColumnFamily cacheRow(Integer cfId, DecoratedKey decoratedKey)
{
RowCacheKey key = new RowCacheKey(cfId, decoratedKey);
ColumnFamily cached;
if ((cached = rowCache.get(key)) == null)
if ((cached = CacheService.instance.rowCache.get(key)) == null)
{
// We force ThreadSafeSortedColumns because cached row will be accessed concurrently
cached = getTopLevelColumns(QueryFilter.getIdentityFilter(key, new QueryPath(columnFamily)), Integer.MIN_VALUE, true);
cached = getTopLevelColumns(QueryFilter.getIdentityFilter(decoratedKey, new QueryPath(columnFamily)),
Integer.MIN_VALUE,
true);
if (cached == null)
return null;
// avoid keeping a permanent reference to the original key buffer
rowCache.put(new DecoratedKey(key.token, ByteBufferUtil.clone(key.key)), cached);
CacheService.instance.rowCache.put(key, cached);
}
return cached;
}
@ -1164,7 +1131,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
long start = System.nanoTime();
try
{
if (rowCache.getCapacity() == 0)
if (CacheService.instance.rowCache.getCapacity() == 0)
{
ColumnFamily cf = getTopLevelColumns(filter, gcBefore, false);
@ -1176,7 +1143,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
return cf.isSuper() ? removeDeleted(cf, gcBefore) : removeDeletedCF(cf, gcBefore);
}
ColumnFamily cached = cacheRow(filter.key);
Integer cfId = Schema.instance.getId(table.name, this.columnFamily);
if (cfId == null)
return null; // secondary index
ColumnFamily cached = cacheRow(cfId, filter.key);
if (cached == null)
return null;
@ -1465,14 +1436,33 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
/** raw cached row -- does not fetch the row if it is not present. not counted in cache statistics. */
public ColumnFamily getRawCachedRow(DecoratedKey key)
{
return rowCache.getCapacity() == 0 ? null : rowCache.getInternal(key);
Integer cfId = Schema.instance.getId(table.name, this.columnFamily);
if (cfId == null)
return null; // secondary index
return getRawCachedRow(new RowCacheKey(cfId, key));
}
public ColumnFamily getRawCachedRow(RowCacheKey key)
{
return CacheService.instance.rowCache.getCapacity() == 0 ? null : CacheService.instance.rowCache.getInternal(key);
}
public void invalidateCachedRow(RowCacheKey key)
{
CacheService.instance.rowCache.remove(key);
}
public void invalidateCachedRow(DecoratedKey key)
{
rowCache.remove(key);
Integer cfId = Schema.instance.getId(table.name, this.columnFamily);
if (cfId == null)
return; // secondary index
invalidateCachedRow(new RowCacheKey(cfId, key));
}
public void forceMajorCompaction() throws InterruptedException, ExecutionException
@ -1480,36 +1470,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
CompactionManager.instance.performMaximal(this);
}
public void invalidateRowCache()
{
rowCache.clear();
}
public void invalidateKeyCache()
{
keyCache.clear();
}
public int getRowCacheCapacity()
{
return rowCache.getCapacity();
}
public int getKeyCacheCapacity()
{
return keyCache.getCapacity();
}
public int getRowCacheSize()
{
return rowCache.size();
}
public int getKeyCacheSize()
{
return keyCache.size();
}
public static Iterable<ColumnFamilyStore> all()
{
Iterable<ColumnFamilyStore>[] stores = new Iterable[Schema.instance.getTables().size()];
@ -1690,9 +1650,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
- get memsize
- get memops
- get/set memtime
- get/set rowCacheSavePeriodInSeconds
- get/set keyCacheSavePeriodInSeconds
- get/set rowCacheKeysToSave
*/
public AbstractCompactionStrategy getCompactionStrategy()
@ -1733,43 +1690,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
return getMinimumCompactionThreshold() <= 0 || getMaximumCompactionThreshold() <= 0;
}
public int getRowCacheSavePeriodInSeconds()
{
return rowCacheSaveInSeconds.value();
}
public void setRowCacheSavePeriodInSeconds(int rcspis)
{
if (rcspis < 0)
{
throw new RuntimeException("RowCacheSavePeriodInSeconds must be non-negative.");
}
this.rowCacheSaveInSeconds.set(rcspis);
scheduleCacheSaving(rowCacheSaveInSeconds.value(), keyCacheSaveInSeconds.value(), rowCacheKeysToSave.value());
}
public int getKeyCacheSavePeriodInSeconds()
{
return keyCacheSaveInSeconds.value();
}
public void setKeyCacheSavePeriodInSeconds(int kcspis)
{
if (kcspis < 0)
{
throw new RuntimeException("KeyCacheSavePeriodInSeconds must be non-negative.");
}
this.keyCacheSaveInSeconds.set(kcspis);
scheduleCacheSaving(rowCacheSaveInSeconds.value(), keyCacheSaveInSeconds.value(), rowCacheKeysToSave.value());
}
public int getRowCacheKeysToSave()
{
return rowCacheKeysToSave.value();
}
public void setRowCacheKeysToSave(int keysToSave)
{
this.rowCacheKeysToSave.set(keysToSave);
}
// End JMX get/set.
public long estimateKeys()
@ -1777,16 +1697,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
return data.estimatedKeys();
}
/**
* Resizes the key and row caches based on the current key estimate.
*/
public synchronized void updateCacheSizes()
{
long keys = estimateKeys();
keyCache.updateCacheSize(keys);
rowCache.updateCacheSize(keys);
}
public long[] getEstimatedRowSizeHistogram()
{
return data.getEstimatedRowSizeHistogram();
@ -1814,15 +1724,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
return columnFamily.split("\\.")[0];
}
/**
* sets each cache's maximum capacity to 75% of its current size
*/
public void reduceCacheSizes()
{
rowCache.reduceCacheSize();
keyCache.reduceCacheSize();
}
private ByteBuffer intern(ByteBuffer name)
{
ByteBuffer internedName = internedNames.get(name);

22
src/java/org/apache/cassandra/db/ColumnFamilyStoreMBean.java
View File

@ -22,8 +22,6 @@ import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutionException;
import org.apache.cassandra.config.ConfigurationException;
/**
* The MBean interface for ColumnFamilyStore
*/
@ -147,17 +145,6 @@ public interface ColumnFamilyStoreMBean
*/
public void forceMajorCompaction() throws ExecutionException, InterruptedException;
/**
* invalidate the key cache; for use after invalidating row cache
*/
public void invalidateKeyCache();
/**
* invalidate the row cache; for use after bulk loading via BinaryMemtable
*/
public void invalidateRowCache();
/**
* return the size of the smallest compacted row
* @return
@ -223,15 +210,6 @@ public interface ColumnFamilyStoreMBean
*/
public List<String> getBuiltIndexes();
public int getRowCacheSavePeriodInSeconds();
public void setRowCacheSavePeriodInSeconds(int rcspis);
public int getKeyCacheSavePeriodInSeconds();
public void setKeyCacheSavePeriodInSeconds(int kcspis);
public int getRowCacheKeysToSave();
public void setRowCacheKeysToSave(int keysToSave);
/**
* Scan through Keyspace/ColumnFamily's data directory
* determine which SSTables should be loaded and load them

12
src/java/org/apache/cassandra/db/DataTracker.java
View File

@ -27,12 +27,13 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import com.google.common.collect.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.cache.AutoSavingCache;
import org.apache.cassandra.cache.KeyCacheKey;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.notifications.INotification;
import org.apache.cassandra.notifications.INotificationConsumer;
@ -41,7 +42,6 @@ import org.apache.cassandra.notifications.SSTableListChangedNotification;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.IntervalTree.Interval;
import org.apache.cassandra.utils.IntervalTree.IntervalTree;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.WrappedRunnable;
public class DataTracker
@ -155,7 +155,6 @@ public class DataTracker
while (!view.compareAndSet(currentView, newView));
addNewSSTablesSize(Arrays.asList(sstable));
cfstore.updateCacheSizes();
notifyAdded(sstable);
incrementallyBackup(sstable);
@ -334,8 +333,6 @@ public class DataTracker
{
addNewSSTablesSize(replacements);
removeOldSSTablesSize(oldSSTables);
cfstore.updateCacheSizes();
}
private void addNewSSTablesSize(Iterable<SSTableReader> newSSTables)
@ -367,11 +364,6 @@ public class DataTracker
}
}
public AutoSavingCache<Pair<Descriptor,DecoratedKey>,Long> getKeyCache()
{
return cfstore.getKeyCache();
}
public long getLiveSize()
{
return liveSize.get();

1
src/java/org/apache/cassandra/db/DecoratedKey.java
View File

@ -23,7 +23,6 @@ import java.util.Comparator;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.dht.RingPosition;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.ByteBufferUtil;

4
src/java/org/apache/cassandra/db/Table.java
View File

@ -119,9 +119,9 @@ public class Table
tableInstance = new Table(table);
schema.storeTableInstance(tableInstance);
//table has to be constructed and in the cache before cacheRow can be called
// table has to be constructed and in the cache before cacheRow can be called
for (ColumnFamilyStore cfs : tableInstance.getColumnFamilyStores())
cfs.initCaches();
cfs.initRowCache();
}
}
}

2
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -937,8 +937,6 @@ public class CompactionManager implements CompactionManagerMBean
{
compactionLock.writeLock().unlock();
}
main.invalidateRowCache();
}
};

28
src/java/org/apache/cassandra/io/sstable/SSTableReader.java
View File

@ -29,9 +29,11 @@ import java.util.concurrent.*;
import com.google.common.base.Function;
import com.google.common.collect.Collections2;
import org.apache.cassandra.cache.KeyCacheKey;
import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.io.compress.CompressedRandomAccessReader;
import org.apache.cassandra.service.CacheService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -82,7 +84,7 @@ public class SSTableReader extends SSTable
private IndexSummary indexSummary;
private Filter bf;
private InstrumentingCache<Pair<Descriptor, DecoratedKey>, Long> keyCache;
private InstrumentingCache<KeyCacheKey, Long> keyCache;
private BloomFilterTracker bloomFilterTracker = new BloomFilterTracker();
@ -283,7 +285,7 @@ public class SSTableReader extends SSTable
{
if (tracker != null)
{
keyCache = tracker.getKeyCache();
keyCache = CacheService.instance.keyCache;
deletingTask.setTracker(tracker);
}
}
@ -321,6 +323,7 @@ public class SSTableReader extends SSTable
private void load(boolean recreatebloom, Set<DecoratedKey> keysToLoadInCache) throws IOException
{
boolean cacheLoading = keyCache != null && !keysToLoadInCache.isEmpty();
SegmentedFile.Builder ibuilder = SegmentedFile.getBuilder(DatabaseDescriptor.getIndexAccessMode());
SegmentedFile.Builder dbuilder = compression
? SegmentedFile.getCompressedBuilder()
@ -331,9 +334,6 @@ public class SSTableReader extends SSTable
DecoratedKey left = null, right = null;
try
{
if (keyCache != null && keyCache.getCapacity() - keyCache.size() < keysToLoadInCache.size())
keyCache.updateCapacity(keyCache.size() + keysToLoadInCache.size());
long indexSize = input.length();
long estimatedKeys = SSTable.estimateRowsFromIndex(input);
indexSummary = new IndexSummary(estimatedKeys);
@ -367,6 +367,7 @@ public class SSTableReader extends SSTable
bf.add(decoratedKey.key);
if (shouldAddEntry)
indexSummary.addEntry(decoratedKey, indexPosition);
// if key cache could be used and we have key already pre-loaded
if (cacheLoading && keysToLoadInCache.contains(decoratedKey))
cacheKey(decoratedKey, dataPosition);
}
@ -449,7 +450,7 @@ public class SSTableReader extends SSTable
{
return indexSummary.getIndexPositions().size() * DatabaseDescriptor.getIndexInterval();
}
/**
* @param ranges
* @return An estimate of the number of keys for given ranges in this SSTable.
@ -604,17 +605,19 @@ public class SSTableReader extends SSTable
public void cacheKey(DecoratedKey key, Long info)
{
if (keyCache == null)
return;
// avoid keeping a permanent reference to the original key buffer
DecoratedKey copiedKey = new DecoratedKey(key.token, ByteBufferUtil.clone(key.key));
keyCache.put(new Pair<Descriptor, DecoratedKey>(descriptor, copiedKey), info);
keyCache.put(new KeyCacheKey(descriptor, ByteBufferUtil.clone(key.key)), info);
}
public Long getCachedPosition(DecoratedKey key, boolean updateStats)
{
return getCachedPosition(new Pair<Descriptor, DecoratedKey>(descriptor, key), updateStats);
return getCachedPosition(new KeyCacheKey(descriptor, key.key), updateStats);
}
private Long getCachedPosition(Pair<Descriptor, DecoratedKey> unifiedKey, boolean updateStats)
private Long getCachedPosition(KeyCacheKey unifiedKey, boolean updateStats)
{
if (keyCache != null && keyCache.getCapacity() > 0)
return updateStats ? keyCache.get(unifiedKey) : keyCache.getInternal(unifiedKey);
@ -641,8 +644,7 @@ public class SSTableReader extends SSTable
if ((op == Operator.EQ || op == Operator.GE) && (key instanceof DecoratedKey))
{
DecoratedKey decoratedKey = (DecoratedKey)key;
Pair<Descriptor, DecoratedKey> unifiedKey = new Pair<Descriptor, DecoratedKey>(descriptor, decoratedKey);
Long cachedPosition = getCachedPosition(unifiedKey, true);
Long cachedPosition = getCachedPosition(new KeyCacheKey(descriptor, decoratedKey.key), true);
if (cachedPosition != null)
return cachedPosition;
}
@ -909,7 +911,7 @@ public class SSTableReader extends SSTable
return bloomFilterTracker.getRecentTruePositiveCount();
}
public InstrumentingCache<Pair<Descriptor,DecoratedKey>, Long> getKeyCache()
public InstrumentingCache<KeyCacheKey, Long> getKeyCache()
{
return keyCache;
}

15
src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java
View File

@ -24,15 +24,12 @@ import java.net.InetAddress;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.gms.Gossiper;
import org.apache.log4j.PropertyConfigurator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -41,6 +38,8 @@ import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.SystemTable;
import org.apache.cassandra.db.Table;
@ -48,7 +47,6 @@ import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.migration.Migration;
import org.apache.cassandra.utils.CLibrary;
import org.apache.cassandra.utils.Mx4jTool;
import org.apache.commons.lang.ArrayUtils;
import com.google.common.collect.Iterables;
@ -158,6 +156,9 @@ public abstract class AbstractCassandraDaemon implements CassandraDaemon
: String.format("Directory %s is not accessible.", dataDir);
}
if (CacheService.instance == null) // should never happen
throw new RuntimeException("Failed to initialize Cache Service.");
// check the system table to keep user from shooting self in foot by changing partitioner, cluster name, etc.
// we do a one-off scrub of the system table first; we can't load the list of the rest of the tables,
// until system table is opened.
@ -201,6 +202,12 @@ public abstract class AbstractCassandraDaemon implements CassandraDaemon
Table.open(table);
}
if (CacheService.instance.keyCache.size() > 0)
logger.info("completed pre-loading ({} keys) key cache.", CacheService.instance.keyCache.size());
if (CacheService.instance.rowCache.size() > 0)
logger.info("completed pre-loading ({} keys) row cache.", CacheService.instance.rowCache.size());
try
{
GCInspector.instance.start();

287
src/java/org/apache/cassandra/service/CacheService.java Normal file
View File

@ -0,0 +1,287 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service;
import java.lang.management.ManagementFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import org.apache.cassandra.cache.*;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.utils.FBUtilities;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class CacheService implements CacheServiceMBean
{
private static final Logger logger = LoggerFactory.getLogger(CacheService.class);
public static final String MBEAN_NAME = "org.apache.cassandra.db:type=Caches";
public static final int AVERAGE_KEY_CACHE_ROW_SIZE = 48;
public static enum CacheType
{
KEY_CACHE("KeyCache"),
ROW_CACHE("RowCache");
private final String name;
private CacheType(String typeName)
{
name = typeName;
}
public String toString()
{
return name;
}
}
public final static CacheService instance = new CacheService();
public final AutoSavingCache<KeyCacheKey, Long> keyCache;
public final AutoSavingCache<RowCacheKey, ColumnFamily> rowCache;
private int rowCacheSavePeriod;
private int keyCacheSavePeriod;
private CacheService()
{
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
{
mbs.registerMBean(this, new ObjectName(MBEAN_NAME));
}
catch (Exception e)
{
throw new RuntimeException(e);
}
rowCacheSavePeriod = DatabaseDescriptor.getRowCacheSavePeriod();
keyCacheSavePeriod = DatabaseDescriptor.getKeyCacheSavePeriod();
keyCache = initKeyCache();
rowCache = initRowCache();
}
/**
* We can use Weighers.singleton() because Long can't be leaking memory
* @return auto saving cache object
*/
private AutoSavingCache<KeyCacheKey, Long> initKeyCache()
{
logger.info("Initializing key cache with capacity of {} MBs.", DatabaseDescriptor.getKeyCacheSizeInMB());
int keyCacheInMemoryCapacity = DatabaseDescriptor.getKeyCacheSizeInMB() * 1024 * 1024;
// as values are constant size we can use singleton weigher
// where 48 = 40 bytes (average size of the key) + 8 bytes (size of value)
ICache<KeyCacheKey, Long> kc = ConcurrentLinkedHashCache.create(keyCacheInMemoryCapacity / AVERAGE_KEY_CACHE_ROW_SIZE);
AutoSavingCache<KeyCacheKey, Long> keyCache = new AutoSavingCache<KeyCacheKey, Long>(kc, CacheType.KEY_CACHE);
int keyCacheKeysToSave = DatabaseDescriptor.getKeyCacheKeysToSave();
logger.info("Scheduling key cache save to each {} seconds (going to save {} keys).",
keyCacheSavePeriod,
keyCacheKeysToSave == Integer.MAX_VALUE ? "all" : keyCacheKeysToSave);
keyCache.scheduleSaving(keyCacheSavePeriod, keyCacheKeysToSave);
return keyCache;
}
/**
* @return initialized row cache
*/
private AutoSavingCache<RowCacheKey, ColumnFamily> initRowCache()
{
logger.info("Initializing row cache with capacity of {} MBs and provider {}",
DatabaseDescriptor.getRowCacheSizeInMB(),
DatabaseDescriptor.getRowCacheProvider().getClass().getName());
int rowCacheInMemoryCapacity = DatabaseDescriptor.getRowCacheSizeInMB() * 1024 * 1024;
// cache object
ICache<RowCacheKey, ColumnFamily> rc = DatabaseDescriptor.getRowCacheProvider().create(rowCacheInMemoryCapacity, true);
AutoSavingCache<RowCacheKey, ColumnFamily> rowCache = new AutoSavingCache<RowCacheKey, ColumnFamily>(rc, CacheType.ROW_CACHE);
int rowCacheKeysToSave = DatabaseDescriptor.getRowCacheKeysToSave();
logger.info("Scheduling row cache save to each {} seconds (going to save {} keys).",
rowCacheSavePeriod,
rowCacheKeysToSave == Integer.MAX_VALUE ? "all" : rowCacheKeysToSave);
rowCache.scheduleSaving(rowCacheSavePeriod, rowCacheKeysToSave);
return rowCache;
}
public long getKeyCacheHits()
{
return keyCache.getHits();
}
public long getRowCacheHits()
{
return rowCache.getHits();
}
public long getKeyCacheRequests()
{
return keyCache.getRequests();
}
public long getRowCacheRequests()
{
return rowCache.getRequests();
}
public double getKeyCacheRecentHitRate()
{
return keyCache.getRecentHitRate();
}
public double getRowCacheRecentHitRate()
{
return rowCache.getRecentHitRate();
}
public int getRowCacheSavePeriodInSeconds()
{
return rowCacheSavePeriod;
}
public void setRowCacheSavePeriodInSeconds(int rcspis)
{
if (rcspis < 0)
throw new RuntimeException("RowCacheSavePeriodInSeconds must be non-negative.");
rowCacheSavePeriod = rcspis;
rowCache.scheduleSaving(rowCacheSavePeriod, DatabaseDescriptor.getRowCacheKeysToSave());
}
public int getKeyCacheSavePeriodInSeconds()
{
return keyCacheSavePeriod;
}
public void setKeyCacheSavePeriodInSeconds(int kcspis)
{
if (kcspis < 0)
throw new RuntimeException("KeyCacheSavePeriodInSeconds must be non-negative.");
keyCacheSavePeriod = kcspis;
keyCache.scheduleSaving(keyCacheSavePeriod, DatabaseDescriptor.getKeyCacheKeysToSave());
}
public void invalidateKeyCache()
{
keyCache.clear();
}
public void invalidateRowCache()
{
rowCache.clear();
}
public int getRowCacheCapacityInBytes()
{
return rowCache.getCapacity();
}
public int getRowCacheCapacityInMB()
{
return getRowCacheCapacityInBytes() / 1024 / 1024;
}
public void setRowCacheCapacityInMB(int capacity)
{
if (capacity < 0)
throw new RuntimeException("capacity should not be negative.");
rowCache.setCapacity(capacity * 1024 * 1024);
}
public int getKeyCacheCapacityInBytes()
{
return keyCache.getCapacity() * AVERAGE_KEY_CACHE_ROW_SIZE;
}
public int getKeyCacheCapacityInMB()
{
return getKeyCacheCapacityInBytes() / 1024 / 1024;
}
public void setKeyCacheCapacityInMB(int capacity)
{
if (capacity < 0)
throw new RuntimeException("capacity should not be negative.");
keyCache.setCapacity(capacity * 1024 * 1024 / 48);
}
public int getRowCacheSize()
{
return rowCache.weightedSize();
}
public int getKeyCacheSize()
{
return keyCache.weightedSize() * AVERAGE_KEY_CACHE_ROW_SIZE;
}
public void reduceCacheSizes()
{
reduceRowCacheSize();
reduceKeyCacheSize();
}
public void reduceRowCacheSize()
{
rowCache.reduceCacheSize();
}
public void reduceKeyCacheSize()
{
keyCache.reduceCacheSize();
}
public void saveCaches() throws ExecutionException, InterruptedException
{
List<Future<?>> futures = new ArrayList<Future<?>>();
logger.debug("submitting cache saves");
futures.add(keyCache.submitWrite(DatabaseDescriptor.getKeyCacheKeysToSave()));
futures.add(rowCache.submitWrite(DatabaseDescriptor.getRowCacheKeysToSave()));
FBUtilities.waitOnFutures(futures);
logger.debug("cache saves completed");
}
}

73
src/java/org/apache/cassandra/service/CacheServiceMBean.java Normal file
View File

@ -0,0 +1,73 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service;
import java.util.concurrent.ExecutionException;
public interface CacheServiceMBean
{
public long getKeyCacheHits();
public long getRowCacheHits();
public long getKeyCacheRequests();
public long getRowCacheRequests();
public double getKeyCacheRecentHitRate();
public double getRowCacheRecentHitRate();
public int getRowCacheSavePeriodInSeconds();
public void setRowCacheSavePeriodInSeconds(int rcspis);
public int getKeyCacheSavePeriodInSeconds();
public void setKeyCacheSavePeriodInSeconds(int kcspis);
/**
* invalidate the key cache; for use after invalidating row cache
*/
public void invalidateKeyCache();
/**
* invalidate the row cache; for use after bulk loading via BinaryMemtable
*/
public void invalidateRowCache();
public int getRowCacheCapacityInMB();
public int getRowCacheCapacityInBytes();
public void setRowCacheCapacityInMB(int capacity);
public int getKeyCacheCapacityInMB();
public int getKeyCacheCapacityInBytes();
public void setKeyCacheCapacityInMB(int capacity);
public int getRowCacheSize();
public int getKeyCacheSize();
/**
* sets each cache's maximum capacity to "reduce_cache_capacity_to" of its current size
*/
public void reduceCacheSizes();
/**
* save row and key caches
*
* @throws ExecutionException when attempting to retrieve the result of a task that aborted by throwing an exception
* @throws InterruptedException when a thread is waiting, sleeping, or otherwise occupied, and the thread is interrupted, either before or during the activity.
*/
public void saveCaches() throws ExecutionException, InterruptedException;
}

2
src/java/org/apache/cassandra/service/GCInspector.java
View File

@ -138,7 +138,7 @@ public class GCInspector
{
cacheSizesReduced = true;
logger.warn("Heap is " + usage + " full. You may need to reduce memtable and/or cache sizes. Cassandra is now reducing cache sizes to free up memory. Adjust reduce_cache_sizes_at threshold in cassandra.yaml if you don't want Cassandra to do this automatically");
StorageService.instance.reduceCacheSizes();
CacheService.instance.reduceCacheSizes();
}
if (memoryUsed > DatabaseDescriptor.getFlushLargestMemtablesAt() * memoryMax)

35
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -1695,22 +1695,6 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
}
}
public void invalidateKeyCaches(String tableName, String... columnFamilies) throws IOException
{
for (ColumnFamilyStore cfStore : getValidColumnFamilies(tableName, columnFamilies))
{
cfStore.invalidateKeyCache();
}
}
public void invalidateRowCaches(String tableName, String... columnFamilies) throws IOException
{
for (ColumnFamilyStore cfStore : getValidColumnFamilies(tableName, columnFamilies))
{
cfStore.invalidateRowCache();
}
}
/**
* Takes the snapshot for the given tables. A snapshot name must be specified.
*
@ -2603,19 +2587,6 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
StorageProxy.truncateBlocking(keyspace, columnFamily);
}
public void saveCaches() throws ExecutionException, InterruptedException
{
List<Future<?>> futures = new ArrayList<Future<?>>();
logger_.debug("submitting cache saves");
for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
{
futures.add(cfs.keyCache.submitWrite(-1));
futures.add(cfs.rowCache.submitWrite(cfs.getRowCacheKeysToSave()));
}
FBUtilities.waitOnFutures(futures);
logger_.debug("cache saves completed");
}
public Map<Token, Float> getOwnership()
{
List<Token> sortedTokens = new ArrayList<Token>(getTokenToEndpointMap().keySet());
@ -2680,12 +2651,6 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
largest.forceFlush();
}
public void reduceCacheSizes()
{
for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
cfs.reduceCacheSizes();
}
/**
* Seed data to the endpoints that will be responsible for it at the future
*

6
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -303,9 +303,6 @@ public interface StorageServiceMBean
/** force hint delivery to an endpoint **/
public void deliverHints(String host) throws UnknownHostException;
/** save row and key caches */
public void saveCaches() throws ExecutionException, InterruptedException;
/**
* given a list of tokens (representing the nodes in the cluster), returns
* a mapping from "token -> %age of cluster owned by that token"
@ -343,9 +340,6 @@ public interface StorageServiceMBean
// to determine if thrift is running
public boolean isRPCServerRunning();
public void invalidateKeyCaches(String ks, String... cfs) throws IOException;
public void invalidateRowCaches(String ks, String... cfs) throws IOException;
// allows a node that have been started without joining the ring to join it
public void joinRing() throws IOException, org.apache.cassandra.config.ConfigurationException;
public boolean isJoined();

68
src/java/org/apache/cassandra/tools/NodeCmd.java
View File

@ -31,9 +31,9 @@ import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ExecutionException;
import org.apache.cassandra.service.CacheServiceMBean;
import org.apache.commons.cli.*;
import org.apache.cassandra.cache.InstrumentingCacheMBean;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutorMBean;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.db.ColumnFamilyStoreMBean;
@ -143,6 +143,8 @@ public class NodeCmd
addCmdHelp(header, "enablethrift", "Reenable thrift server");
addCmdHelp(header, "statusthrift", "Status of thrift server");
addCmdHelp(header, "gossipinfo", "Shows the gossip information for the cluster");
addCmdHelp(header, "invalidatekeycache", "Invalidate the key cache");
addCmdHelp(header, "invalidaterowcache", "Invalidate the row cache");
// One arg
addCmdHelp(header, "netstats [host]", "Print network information on provided host (connecting node by default)");
@ -160,9 +162,8 @@ public class NodeCmd
addCmdHelp(header, "cleanup [keyspace] [cfnames]", "Run cleanup on one or more column family");
addCmdHelp(header, "compact [keyspace] [cfnames]", "Force a (major) compaction on one or more column family");
addCmdHelp(header, "scrub [keyspace] [cfnames]", "Scrub (rebuild sstables for) one or more column family");
addCmdHelp(header, "upgradesstables [keyspace] [cfnames]", "Upgrade sstables for one or more column family");
addCmdHelp(header, "invalidatekeycache [keyspace] [cfnames]", "Invalidate the key cache of one or more column family");
addCmdHelp(header, "invalidaterowcache [keyspace] [cfnames]", "Invalidate the key cache of one or more column family");
addCmdHelp(header, "upgradesstables [keyspace] [cfnames]", "Scrub (rebuild sstables for) one or more column family");
addCmdHelp(header, "getcompactionthreshold <keyspace> <cfname>", "Print min and max compaction thresholds for a given column family");
addCmdHelp(header, "cfhistograms <keyspace> <cfname>", "Print statistic histograms for a given column family");
addCmdHelp(header, "refresh <keyspace> <cf-name>", "Load newly placed SSTables to the system without restart.");
@ -317,6 +318,28 @@ public class NodeCmd
// Exceptions
outs.printf("%-17s: %s%n", "Exceptions", probe.getExceptionCount());
CacheServiceMBean cacheService = probe.getCacheServiceMBean();
// Key Cache: Hits, Requests, RecentHitRate, SavePeriodInSeconds
outs.printf("%-17s: size %d (bytes), capacity %d (bytes), %d hits, %d requests, %.3f recent hit rate, %d save period in seconds%n",
"Key Cache",
cacheService.getKeyCacheSize(),
cacheService.getKeyCacheCapacityInBytes(),
cacheService.getKeyCacheHits(),
cacheService.getKeyCacheRequests(),
cacheService.getKeyCacheRecentHitRate(),
cacheService.getKeyCacheSavePeriodInSeconds());
// Row Cache: Hits, Requests, RecentHitRate, SavePeriodInSeconds
outs.printf("%-17s: size %d (bytes), capacity %d (bytes), %d hits, %d requests, %.3f recent hit rate, %d save period in seconds%n",
"Row Cache",
cacheService.getRowCacheSize(),
cacheService.getRowCacheCapacityInBytes(),
cacheService.getRowCacheHits(),
cacheService.getRowCacheRequests(),
cacheService.getRowCacheRecentHitRate(),
cacheService.getRowCacheSavePeriodInSeconds());
}
public void printReleaseVersion(PrintStream outs)
@ -501,31 +524,6 @@ public class NodeCmd
outs.println("\t\tBloom Filter False Postives: " + cfstore.getBloomFilterFalsePositives());
outs.println("\t\tBloom Filter False Ratio: " + String.format("%01.5f", cfstore.getRecentBloomFilterFalseRatio()));
outs.println("\t\tBloom Filter Space Used: " + cfstore.getBloomFilterDiskSpaceUsed());
InstrumentingCacheMBean keyCacheMBean = probe.getKeyCacheMBean(tableName, cfstore.getColumnFamilyName());
if (keyCacheMBean.getCapacity() > 0)
{
outs.println("\t\tKey cache capacity: " + keyCacheMBean.getCapacity());
outs.println("\t\tKey cache size: " + keyCacheMBean.getSize());
outs.println("\t\tKey cache hit rate: " + keyCacheMBean.getRecentHitRate());
}
else
{
outs.println("\t\tKey cache: disabled");
}
InstrumentingCacheMBean rowCacheMBean = probe.getRowCacheMBean(tableName, cfstore.getColumnFamilyName());
if (rowCacheMBean.getCapacity() > 0)
{
outs.println("\t\tRow cache capacity: " + rowCacheMBean.getCapacity());
outs.println("\t\tRow cache size: " + rowCacheMBean.getSize());
outs.println("\t\tRow cache hit rate: " + rowCacheMBean.getRecentHitRate());
}
else
{
outs.println("\t\tRow cache: disabled");
}
outs.println("\t\tCompacted row minimum size: " + cfstore.getMinRowSize());
outs.println("\t\tCompacted row maximum size: " + cfstore.getMaxRowSize());
outs.println("\t\tCompacted row mean size: " + cfstore.getMeanRowSize());
@ -710,14 +708,20 @@ public class NodeCmd
else { probe.removeToken(arguments[0]); }
break;
case INVALIDATEKEYCACHE :
probe.invalidateKeyCache();
break;
case INVALIDATEROWCACHE :
probe.invalidateRowCache();
break;
case CLEANUP :
case COMPACT :
case REPAIR :
case FLUSH :
case SCRUB :
case UPGRADESSTABLES :
case INVALIDATEKEYCACHE :
case INVALIDATEROWCACHE :
optionalKSandCFs(command, cmd, arguments, probe);
break;
@ -886,8 +890,6 @@ public class NodeCmd
else
probe.forceTableRepair(keyspace, columnFamilies);
break;
case INVALIDATEKEYCACHE : probe.invalidateKeyCaches(keyspace, columnFamilies); break;
case INVALIDATEROWCACHE : probe.invalidateRowCaches(keyspace, columnFamilies); break;
case FLUSH :
try { probe.forceTableFlush(keyspace, columnFamilies); }
catch (ExecutionException ee) { err(ee, "Error occured during flushing"); }

66
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -40,7 +40,6 @@ import javax.management.remote.JMXServiceURL;
import com.google.common.collect.Iterables;
import org.apache.cassandra.cache.InstrumentingCacheMBean;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutorMBean;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.db.ColumnFamilyStoreMBean;
@ -52,6 +51,8 @@ import org.apache.cassandra.gms.FailureDetectorMBean;
import org.apache.cassandra.locator.EndpointSnitchInfoMBean;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.MessagingServiceMBean;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.service.CacheServiceMBean;
import org.apache.cassandra.service.StorageServiceMBean;
import org.apache.cassandra.streaming.StreamingService;
import org.apache.cassandra.streaming.StreamingServiceMBean;
@ -80,6 +81,7 @@ public class NodeProbe
private StreamingServiceMBean streamProxy;
public MessagingServiceMBean msProxy;
private FailureDetectorMBean fdProxy;
private CacheServiceMBean cacheService;
/**
* Creates a NodeProbe using the specified JMX host, port, username, and password.
@ -156,6 +158,8 @@ public class NodeProbe
compactionProxy = JMX.newMBeanProxy(mbeanServerConn, name, CompactionManagerMBean.class);
name = new ObjectName(FailureDetector.MBEAN_NAME);
fdProxy = JMX.newMBeanProxy(mbeanServerConn, name, FailureDetectorMBean.class);
name = new ObjectName(CacheService.MBEAN_NAME);
cacheService = JMX.newMBeanProxy(mbeanServerConn, name, CacheServiceMBean.class);
} catch (MalformedObjectNameException e)
{
throw new RuntimeException(
@ -208,14 +212,14 @@ public class NodeProbe
ssProxy.forceTableRepairPrimaryRange(tableName, columnFamilies);
}
public void invalidateKeyCaches(String tableName, String... columnFamilies) throws IOException
public void invalidateKeyCache() throws IOException
{
ssProxy.invalidateKeyCaches(tableName, columnFamilies);
cacheService.invalidateKeyCache();
}
public void invalidateRowCaches(String tableName, String... columnFamilies) throws IOException
public void invalidateRowCache() throws IOException
{
ssProxy.invalidateRowCaches(tableName, columnFamilies);
cacheService.invalidateRowCache();
}
public void drain() throws IOException, InterruptedException, ExecutionException
@ -263,6 +267,20 @@ public class NodeProbe
return ssProxy.getOwnership();
}
public CacheServiceMBean getCacheServiceMBean()
{
String cachePath = "org.apache.cassandra.db:type=Caches";
try
{
return JMX.newMBeanProxy(mbeanServerConn, new ObjectName(cachePath), CacheServiceMBean.class);
}
catch (MalformedObjectNameException e)
{
throw new RuntimeException(e);
}
}
public Iterator<Map.Entry<String, ColumnFamilyStoreMBean>> getColumnFamilyStoreMBeanProxies()
{
try
@ -284,32 +302,6 @@ public class NodeProbe
return compactionProxy;
}
public InstrumentingCacheMBean getKeyCacheMBean(String tableName, String cfName)
{
String keyCachePath = "org.apache.cassandra.db:type=Caches,keyspace=" + tableName + ",cache=" + cfName + "KeyCache";
try
{
return JMX.newMBeanProxy(mbeanServerConn, new ObjectName(keyCachePath), InstrumentingCacheMBean.class);
}
catch (MalformedObjectNameException e)
{
throw new RuntimeException(e);
}
}
public InstrumentingCacheMBean getRowCacheMBean(String tableName, String cfName)
{
String rowCachePath = "org.apache.cassandra.db:type=Caches,keyspace=" + tableName + ",cache=" + cfName + "RowCache";
try
{
return JMX.newMBeanProxy(mbeanServerConn, new ObjectName(rowCachePath), InstrumentingCacheMBean.class);
}
catch (MalformedObjectNameException e)
{
throw new RuntimeException(e);
}
}
public String getToken()
{
return ssProxy.getToken();
@ -439,14 +431,10 @@ public class NodeProbe
{
try
{
String keyCachePath = "org.apache.cassandra.db:type=Caches,keyspace=" + tableName + ",cache=" + cfName + "KeyCache";
InstrumentingCacheMBean keyCacheMBean = JMX.newMBeanProxy(mbeanServerConn, new ObjectName(keyCachePath), InstrumentingCacheMBean.class);
keyCacheMBean.setCapacity(keyCacheCapacity);
String rowCachePath = "org.apache.cassandra.db:type=Caches,keyspace=" + tableName + ",cache=" + cfName + "RowCache";
InstrumentingCacheMBean rowCacheMBean = null;
rowCacheMBean = JMX.newMBeanProxy(mbeanServerConn, new ObjectName(rowCachePath), InstrumentingCacheMBean.class);
rowCacheMBean.setCapacity(rowCacheCapacity);
String keyCachePath = "org.apache.cassandra.db:type=Caches";
CacheServiceMBean cacheMBean = JMX.newMBeanProxy(mbeanServerConn, new ObjectName(keyCachePath), CacheServiceMBean.class);
cacheMBean.setKeyCacheCapacityInMB(keyCacheCapacity);
cacheMBean.setRowCacheCapacityInMB(rowCacheCapacity);
}
catch (MalformedObjectNameException e)
{

34
src/java/org/apache/cassandra/utils/StatusLogger.java
View File

@ -30,13 +30,19 @@ import javax.management.ObjectName;
import com.google.common.collect.Iterables;
import org.apache.cassandra.cache.AutoSavingCache;
import org.apache.cassandra.cache.KeyCacheKey;
import org.apache.cassandra.cache.RowCacheKey;
import org.apache.cassandra.db.ColumnFamily;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutorMBean;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.CacheService;
public class StatusLogger
{
@ -81,15 +87,33 @@ public class StatusLogger
logger.info(String.format("%-25s%10s%10s",
"MessagingService", "n/a", pendingCommands + "," + pendingResponses));
// Global key/row cache information
AutoSavingCache<KeyCacheKey, Long> keyCache = CacheService.instance.keyCache;
AutoSavingCache<RowCacheKey, ColumnFamily> rowCache = CacheService.instance.rowCache;
int keyCacheKeysToSave = DatabaseDescriptor.getKeyCacheKeysToSave();
int rowCacheKeysToSave = DatabaseDescriptor.getRowCacheKeysToSave();
logger.info(String.format("%-25s%10s%25s%25s%65s", "Cache Type", "Size", "Capacity", "KeysToSave", "Provider"));
logger.info(String.format("%-25s%10s%25s%25s%65s", "KeyCache",
keyCache.weightedSize(),
keyCache.getCapacity(),
keyCacheKeysToSave == Integer.MAX_VALUE ? "all" : keyCacheKeysToSave,
""));
logger.info(String.format("%-25s%10s%25s%25s%65s", "RowCache",
rowCache.weightedSize(),
rowCache.getCapacity(),
rowCacheKeysToSave == Integer.MAX_VALUE ? "all" : rowCacheKeysToSave,
DatabaseDescriptor.getRowCacheProvider().getClass().getName()));
// per-CF stats
logger.info(String.format("%-25s%20s%20s%20s", "ColumnFamily", "Memtable ops,data", "Row cache size/cap", "Key cache size/cap"));
logger.info(String.format("%-25s%20s", "ColumnFamily", "Memtable ops,data"));
for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
{
logger.info(String.format("%-25s%20s%20s%20s",
logger.info(String.format("%-25s%20s",
cfs.table.name + "." + cfs.columnFamily,
cfs.getMemtableColumnsCount() + "," + cfs.getMemtableDataSize(),
cfs.getRowCacheSize() + "/" + cfs.getRowCacheCapacity(),
cfs.getKeyCacheSize() + "/" + cfs.getKeyCacheCapacity()));
cfs.getMemtableColumnsCount() + "," + cfs.getMemtableDataSize()));
}
}
}

116
src/resources/org/apache/cassandra/cli/CliHelp.yaml
View File

@ -450,26 +450,6 @@ commands:
See http://wiki.apache.org/Cassandra/DistributedDeletes
- keys_cached: Maximum number of keys to cache in memory. Valid values are
either a double between 0 and 1 (inclusive on both ends) denoting what
fraction should be cached. Or an absolute number of rows to cache.
Default value is 200000.
Each key cache hit saves 1 seek and each row cache hit saves 2 seeks at the
minimum, sometimes more. The key cache is fairly tiny for the amount of
time it saves, so it's worthwhile to use it at large numbers all the way
up to 1.0 (all keys cached). The row cache saves even more time, but must
store the whole values of its rows, so it is extremely space-intensive.
It's best to only use the row cache if you have hot rows or static rows.
- key_cache_save_period: Duration in seconds after which Cassandra should
safe the keys cache. Caches are saved to saved_caches_directory as
specified in conf/Cassandra.yaml. Default is 14400 or 4 hours.
Saved caches greatly improve cold-start speeds, and is relatively cheap in
terms of I/O for the key cache. Row cache saving is much more expensive and
has limited use.
- read_repair_chance: Probability (0.0-1.0) with which to perform read
repairs for any read operation. Default is 0.1.
@ -477,27 +457,6 @@ commands:
will not have any latency information from all the replicas to recognize
when one is performing worse than usual.
- rows_cached: Maximum number of rows whose entire contents we
cache in memory. Valid values are either a double between 0 and 1 (
inclusive on both ends) denoting what fraction should be cached. Or an
absolute number of rows to cache. Default value is 0, to disable row
caching.
Each key cache hit saves 1 seek and each row cache hit saves 2 seeks at the
minimum, sometimes more. The key cache is fairly tiny for the amount of
time it saves, so it's worthwhile to use it at large numbers all the way
up to 1.0 (all keys cached). The row cache saves even more time, but must
store the whole values of its rows, so it is extremely space-intensive.
It's best to only use the row cache if you have hot rows or static rows.
- row_cache_save_period: Duration in seconds after which Cassandra should
safe the row cache. Caches are saved to saved_caches_directory as specified
in conf/Cassandra.yaml. Default is 0 to disable saving the row cache.
Saved caches greatly improve cold-start speeds, and is relatively cheap in
terms of I/O for the key cache. Row cache saving is much more expensive and
has limited use.
- subcomparator: Validator to use to validate and compare sub column names
in this column family. Only applied to Super column families. Default is
BytesType, which is a straight forward lexical comparison of the bytes in
@ -537,23 +496,6 @@ commands:
- replicate_on_write: Replicate every counter update from the leader to the
follower replicas. Accepts the values true and false.
- row_cache_provider: The provider for the row cache to use for this
column family.
Supported values are:
- ConcurrentLinkedHashCacheProvider
- SerializingCacheProvider
It is also valid to specify the fully-qualified class name to a class
that implements org.apache.cassandra.cache.IRowCacheProvider.
row_cache_provider defaults to SerializingCacheProvider.
SerializingCacheProvider serialises the contents of the row and stores
it in native memory, i.e., off the JVM Heap. Serialized rows take
significantly less memory than "live" rows in the JVM, so you can cache
more rows in a given memory footprint. And storing the cache off-heap
means you can use smaller heap sizes, reducing the impact of GC pauses.
- compression_options: Options related to compression.
Options have the form {key:value}.
The main recognized options are:
@ -710,26 +652,6 @@ commands:
See http://wiki.apache.org/Cassandra/DistributedDeletes
- keys_cached: Maximum number of keys to cache in memory. Valid values are
either a double between 0 and 1 (inclusive on both ends) denoting what
fraction should be cached. Or an absolute number of rows to cache.
Default value is 200000.
Each key cache hit saves 1 seek and each row cache hit saves 2 seeks at the
minimum, sometimes more. The key cache is fairly tiny for the amount of
time it saves, so it's worthwhile to use it at large numbers all the way
up to 1.0 (all keys cached). The row cache saves even more time, but must
store the whole values of its rows, so it is extremely space-intensive.
It's best to only use the row cache if you have hot rows or static rows.
- key_cache_save_period: Duration in seconds after which Cassandra should
safe the keys cache. Caches are saved to saved_caches_directory as
specified in conf/Cassandra.yaml. Default is 14400 or 4 hours.
Saved caches greatly improve cold-start speeds, and is relatively cheap in
terms of I/O for the key cache. Row cache saving is much more expensive and
has limited use.
- read_repair_chance: Probability (0.0-1.0) with which to perform read
repairs for any read operation. Default is 0.1.
@ -737,27 +659,6 @@ commands:
will not have any latency information from all the replicas to recognize
when one is performing worse than usual.
- rows_cached: Maximum number of rows whose entire contents we
cache in memory. Valid values are either a double between 0 and 1 (
inclusive on both ends) denoting what fraction should be cached. Or an
absolute number of rows to cache. Default value is 0, to disable row
caching.
Each key cache hit saves 1 seek and each row cache hit saves 2 seeks at the
minimum, sometimes more. The key cache is fairly tiny for the amount of
time it saves, so it's worthwhile to use it at large numbers all the way
up to 1.0 (all keys cached). The row cache saves even more time, but must
store the whole values of its rows, so it is extremely space-intensive.
It's best to only use the row cache if you have hot rows or static rows.
- row_cache_save_period: Duration in seconds after which Cassandra should
safe the row cache. Caches are saved to saved_caches_directory as specified
in conf/Cassandra.yaml. Default is 0 to disable saving the row cache.
Saved caches greatly improve cold-start speeds, and is relatively cheap in
terms of I/O for the key cache. Row cache saving is much more expensive and
has limited use.
- subcomparator: Validator to use to validate and compare sub column names
in this column family. Only applied to Super column families. Default is
BytesType, which is a straight forward lexical comparison of the bytes in
@ -797,23 +698,6 @@ commands:
- replicate_on_write: Replicate every counter update from the leader to the
follower replicas. Accepts the values true and false.
- row_cache_provider: The provider for the row cache to use for this
column family.
Supported values are:
- ConcurrentLinkedHashCacheProvider
- SerializingCacheProvider
It is also valid to specify the fully-qualified class name to a class
that implements org.apache.cassandra.cache.IRowCacheProvider.
row_cache_provider defaults to SerializingCacheProvider.
SerializingCacheProvider serialises the contents of the row and stores
it in native memory, i.e., off the JVM Heap. Serialized rows take
significantly less memory than "live" rows in the JVM, so you can cache
more rows in a given memory footprint. And storing the cache off-heap
means you can use smaller heap sizes, reducing the impact of GC pauses.
- compression_options: Options related to compression.
Options have the form {key:value}.
The main recognized options are:

12
test/unit/org/apache/cassandra/CleanupHelper.java
View File

@ -58,6 +58,8 @@ public class CleanupHelper extends SchemaLoader
FileUtils.deleteRecursive(dir);
}
cleanupSavedCaches();
// clean up data directory which are stored as data directory/table/data files
for (String dirName : DatabaseDescriptor.getAllDataFileLocations())
{
@ -105,4 +107,14 @@ public class CleanupHelper extends SchemaLoader
store.getColumnFamily(key, path, ByteBufferUtil.EMPTY_BYTE_BUFFER, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, 1);
}
}
protected static void cleanupSavedCaches()
{
File cachesDir = new File(DatabaseDescriptor.getSavedCachesLocation());
if (!cachesDir.exists() || !cachesDir.isDirectory())
return;
FileUtils.delete(cachesDir.listFiles());
}
}

24
test/unit/org/apache/cassandra/SchemaLoader.java
View File

@ -144,8 +144,7 @@ public class SchemaLoader
"StandardInteger1",
st,
IntegerType.instance,
null)
.keyCacheSize(0),
null),
new CFMetaData(ks1,
"Counter1",
st,
@ -212,8 +211,7 @@ public class SchemaLoader
"Super5",
su,
TimeUUIDType.instance,
bytes)
.keyCacheSize(0)));
bytes)));
// Keyspace 5
schema.add(KSMetaData.testMetadata(ks5,
@ -233,20 +231,16 @@ public class SchemaLoader
schema.add(KSMetaData.testMetadata(ks_kcs,
simple,
opts_rf1,
standardCFMD(ks_kcs, "Standard1")
.keyCacheSize(0.5),
standardCFMD(ks_kcs, "Standard2")
.keyCacheSize(1.0),
standardCFMD(ks_kcs, "Standard3")
.keyCacheSize(1.0)));
standardCFMD(ks_kcs, "Standard1"),
standardCFMD(ks_kcs, "Standard2"),
standardCFMD(ks_kcs, "Standard3")));
// RowCacheSpace
schema.add(KSMetaData.testMetadata(ks_rcs,
simple,
opts_rf1,
standardCFMD(ks_rcs, "CFWithoutCache"),
standardCFMD(ks_rcs, "CachedCF")
.rowCacheSize(100)));
standardCFMD(ks_rcs, "CachedCF")));
schema.add(KSMetaData.testMetadataNotDurable(ks_nocommit,
simple,
@ -273,15 +267,15 @@ public class SchemaLoader
private static CFMetaData standardCFMD(String ksName, String cfName)
{
return new CFMetaData(ksName, cfName, ColumnFamilyType.Standard, BytesType.instance, null).keyCacheSize(0);
return new CFMetaData(ksName, cfName, ColumnFamilyType.Standard, BytesType.instance, null);
}
private static CFMetaData superCFMD(String ksName, String cfName, AbstractType subcc)
{
return superCFMD(ksName, cfName, BytesType.instance, subcc).keyCacheSize(0);
return superCFMD(ksName, cfName, BytesType.instance, subcc);
}
private static CFMetaData superCFMD(String ksName, String cfName, AbstractType cc, AbstractType subcc)
{
return new CFMetaData(ksName, cfName, ColumnFamilyType.Super, cc, subcc).keyCacheSize(0);
return new CFMetaData(ksName, cfName, ColumnFamilyType.Super, cc, subcc);
}
private static CFMetaData indexCFMD(String ksName, String cfName, final Boolean withIdxType) throws ConfigurationException
{

4
test/unit/org/apache/cassandra/cache/CacheProviderTest.java vendored
View File

@ -105,7 +105,7 @@ public class CacheProviderTest extends SchemaLoader
@Test
public void testHeapCache() throws InterruptedException
{
ICache<String, ColumnFamily> cache = ConcurrentLinkedHashCache.create(CAPACITY, tableName, cfName);
ICache<String, ColumnFamily> cache = ConcurrentLinkedHashCache.create(CAPACITY);
ColumnFamily cf = createCF();
simpleCase(cf, cache);
concurrentCase(cf, cache);
@ -114,7 +114,7 @@ public class CacheProviderTest extends SchemaLoader
@Test
public void testSerializingCache() throws InterruptedException
{
ICache<String, ColumnFamily> cache = new SerializingCache<String, ColumnFamily>(CAPACITY, ColumnFamily.serializer(), tableName, cfName);
ICache<String, ColumnFamily> cache = new SerializingCache<String, ColumnFamily>(CAPACITY, false, ColumnFamily.serializer());
ColumnFamily cf = createCF();
simpleCase(cf, cache);
concurrentCase(cf, cache);

18
test/unit/org/apache/cassandra/db/DefsTest.java
View File

@ -70,8 +70,6 @@ public class DefsTest extends CleanupHelper
// make sure some of the fields didn't get unexpected zeros put in during [de]serialize operations.
assert cd.min_compaction_threshold == null;
assert cd2.min_compaction_threshold == null;
assert cd.row_cache_save_period_in_seconds == null;
assert cd2.row_cache_save_period_in_seconds == null;
assert cd.compaction_strategy == null;
}
@ -100,16 +98,12 @@ public class DefsTest extends CleanupHelper
null);
cfm.comment("No comment")
.rowCacheSize(1.0)
.keyCacheSize(1.0)
.readRepairChance(0.5)
.replicateOnWrite(false)
.gcGraceSeconds(100000)
.defaultValidator(null)
.minCompactionThreshold(500)
.maxCompactionThreshold(500)
.rowCacheSavePeriod(500)
.keyCacheSavePeriod(500)
.mergeShardsChance(0.0)
.columnMetadata(indexes);
@ -521,7 +515,6 @@ public class DefsTest extends CleanupHelper
// updating certain fields should fail.
org.apache.cassandra.db.migration.avro.CfDef cf_def = cf.toAvro();
cf_def.row_cache_size = 43.3;
cf_def.column_metadata = new ArrayList<org.apache.cassandra.db.migration.avro.ColumnDef>();
cf_def.default_validation_class ="BytesType";
cf_def.min_compaction_threshold = 5;
@ -530,13 +523,7 @@ public class DefsTest extends CleanupHelper
// test valid operations.
cf_def.comment = "Modified comment";
new UpdateColumnFamily(cf_def).apply(); // doesn't get set back here.
cf_def.row_cache_size = 2d;
new UpdateColumnFamily(cf_def).apply();
cf_def.key_cache_size = 3d;
new UpdateColumnFamily(cf_def).apply();
cf_def.read_repair_chance = 0.23;
new UpdateColumnFamily(cf_def).apply();
@ -556,8 +543,6 @@ public class DefsTest extends CleanupHelper
// check the cumulative affect.
assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName).getComment().equals(cf_def.comment);
assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName).getRowCacheSize() == cf_def.row_cache_size;
assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName).getKeyCacheSize() == cf_def.key_cache_size;
assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName).getReadRepairChance() == cf_def.read_repair_chance;
assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName).getGcGraceSeconds() == cf_def.gc_grace_seconds;
assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName).getDefaultValidator() == UTF8Type.instance;
@ -677,7 +662,6 @@ public class DefsTest extends CleanupHelper
{
CFMetaData newCFMD = new CFMetaData(ks, cf, ColumnFamilyType.Standard, UTF8Type.instance, null);
newCFMD.comment(comment)
.keyCacheSize(1.0)
.readRepairChance(0.0)
.mergeShardsChance(0.0);

128
test/unit/org/apache/cassandra/db/KeyCacheTest.java
View File

@ -20,12 +20,17 @@ package org.apache.cassandra.db;
*
*/
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import org.apache.cassandra.cache.KeyCacheKey;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.thrift.ColumnParent;
import org.junit.AfterClass;
import org.junit.Test;
import org.apache.cassandra.CleanupHelper;
@ -43,18 +48,11 @@ public class KeyCacheTest extends CleanupHelper
private static final String TABLE1 = "KeyCacheSpace";
private static final String COLUMN_FAMILY1 = "Standard1";
private static final String COLUMN_FAMILY2 = "Standard2";
private static final String COLUMN_FAMILY3 = "Standard3";
@Test
public void testKeyCache50() throws IOException, ExecutionException, InterruptedException
@AfterClass
public static void cleanup()
{
testKeyCache(COLUMN_FAMILY1, 64);
}
@Test
public void testKeyCache100() throws IOException, ExecutionException, InterruptedException
{
testKeyCache(COLUMN_FAMILY2, 128);
cleanupSavedCaches();
}
@Test
@ -62,57 +60,48 @@ public class KeyCacheTest extends CleanupHelper
{
CompactionManager.instance.disableAutoCompaction();
ColumnFamilyStore store = Table.open(TABLE1).getColumnFamilyStore(COLUMN_FAMILY3);
ColumnFamilyStore store = Table.open(TABLE1).getColumnFamilyStore(COLUMN_FAMILY2);
// empty the cache
store.invalidateKeyCache();
assert store.getKeyCacheSize() == 0;
CacheService.instance.invalidateKeyCache();
assert CacheService.instance.keyCache.size() == 0;
// insert data and force to disk
insertData(TABLE1, COLUMN_FAMILY3, 0, 100);
insertData(TABLE1, COLUMN_FAMILY2, 0, 100);
store.forceBlockingFlush();
// populate the cache
readData(TABLE1, COLUMN_FAMILY3, 0, 100);
assertEquals(100, store.getKeyCacheSize());
readData(TABLE1, COLUMN_FAMILY2, 0, 100);
assertEquals(100, CacheService.instance.keyCache.size());
// really? our caches don't implement the map interface? (hence no .addAll)
Map<Pair<Descriptor, DecoratedKey>, Long> savedMap = new HashMap<Pair<Descriptor, DecoratedKey>, Long>();
for (Pair<Descriptor, DecoratedKey> k : store.getKeyCache().getKeySet())
Map<KeyCacheKey, Long> savedMap = new HashMap<KeyCacheKey, Long>();
for (KeyCacheKey k : CacheService.instance.keyCache.getKeySet())
{
savedMap.put(k, store.getKeyCache().get(k));
savedMap.put(k, CacheService.instance.keyCache.get(k));
}
// force the cache to disk
store.keyCache.submitWrite(Integer.MAX_VALUE).get();
CacheService.instance.keyCache.submitWrite(Integer.MAX_VALUE).get();
// empty the cache again to make sure values came from disk
store.invalidateKeyCache();
assert store.getKeyCacheSize() == 0;
// load the cache from disk. unregister the old mbean so we can recreate a new CFS object.
// but don't invalidate() the old CFS, which would nuke the data we want to try to load
store.unregisterMBean();
ColumnFamilyStore newStore = ColumnFamilyStore.createColumnFamilyStore(Table.open(TABLE1), COLUMN_FAMILY3);
assertEquals(100, newStore.getKeyCacheSize());
assertEquals(100, savedMap.size());
for (Map.Entry<Pair<Descriptor, DecoratedKey>, Long> entry : savedMap.entrySet())
{
assert newStore.getKeyCache().get(entry.getKey()).equals(entry.getValue());
}
CacheService.instance.invalidateKeyCache();
assert CacheService.instance.keyCache.size() == 0;
}
public void testKeyCache(String cfName, int expectedCacheSize) throws IOException, ExecutionException, InterruptedException
@Test
public void testKeyCache() throws IOException, ExecutionException, InterruptedException
{
CompactionManager.instance.disableAutoCompaction();
Table table = Table.open(TABLE1);
ColumnFamilyStore cfs = table.getColumnFamilyStore(cfName);
ColumnFamilyStore cfs = table.getColumnFamilyStore(COLUMN_FAMILY1);
// KeyCache should start at size 1 if we're caching X% of zero data.
int keyCacheSize = cfs.getKeyCacheCapacity();
assert keyCacheSize == 1 : keyCacheSize;
// just to make sure that everything is clean
CacheService.instance.invalidateKeyCache();
// KeyCache should start at size 0 if we're caching X% of zero data.
int keyCacheSize = CacheService.instance.keyCache.size();
assert keyCacheSize == 0 : keyCacheSize;
DecoratedKey key1 = Util.dk("key1");
DecoratedKey key2 = Util.dk("key2");
@ -120,28 +109,53 @@ public class KeyCacheTest extends CleanupHelper
// inserts
rm = new RowMutation(TABLE1, key1.key);
rm.add(new QueryPath(cfName, null, ByteBufferUtil.bytes("1")), ByteBufferUtil.EMPTY_BYTE_BUFFER, 0);
rm.add(new QueryPath(COLUMN_FAMILY1, null, ByteBufferUtil.bytes("1")), ByteBufferUtil.EMPTY_BYTE_BUFFER, 0);
rm.apply();
rm = new RowMutation(TABLE1, key2.key);
rm.add(new QueryPath(cfName, null, ByteBufferUtil.bytes("2")), ByteBufferUtil.EMPTY_BYTE_BUFFER, 0);
rm.add(new QueryPath(COLUMN_FAMILY1, null, ByteBufferUtil.bytes("2")), ByteBufferUtil.EMPTY_BYTE_BUFFER, 0);
rm.apply();
// deletes
rm = new RowMutation(TABLE1, key1.key);
rm.delete(new QueryPath(cfName, null, ByteBufferUtil.bytes("1")), 1);
rm.apply();
rm = new RowMutation(TABLE1, key2.key);
rm.delete(new QueryPath(cfName, null, ByteBufferUtil.bytes("2")), 1);
rm.apply();
// After a flush, the cache should expand to be X% of indices * INDEX_INTERVAL.
// to make sure we have SSTable
cfs.forceBlockingFlush();
keyCacheSize = cfs.getKeyCacheCapacity();
assert keyCacheSize == expectedCacheSize : keyCacheSize;
// After a compaction, the cache should expand to be X% of zero data.
// reads to cache key position
cfs.getColumnFamily(QueryFilter.getSliceFilter(key1,
new QueryPath(new ColumnParent(COLUMN_FAMILY1)),
ByteBufferUtil.EMPTY_BYTE_BUFFER,
ByteBufferUtil.EMPTY_BYTE_BUFFER,
false,
10));
cfs.getColumnFamily(QueryFilter.getSliceFilter(key2,
new QueryPath(new ColumnParent(COLUMN_FAMILY1)),
ByteBufferUtil.EMPTY_BYTE_BUFFER,
ByteBufferUtil.EMPTY_BYTE_BUFFER,
false,
10));
assert CacheService.instance.keyCache.size() == 2;
Util.compactAll(cfs).get();
keyCacheSize = cfs.getKeyCacheCapacity();
assert keyCacheSize == 1 : keyCacheSize;
keyCacheSize = CacheService.instance.keyCache.size();
// after compaction cache should have entries for
// new SSTables, if we had 2 keys in cache previously it should become 4
assert keyCacheSize == 4 : keyCacheSize;
// re-read same keys to verify that key cache didn't grow further
cfs.getColumnFamily(QueryFilter.getSliceFilter(key1,
new QueryPath(new ColumnParent(COLUMN_FAMILY1)),
ByteBufferUtil.EMPTY_BYTE_BUFFER,
ByteBufferUtil.EMPTY_BYTE_BUFFER,
false,
10));
cfs.getColumnFamily(QueryFilter.getSliceFilter(key2,
new QueryPath(new ColumnParent(COLUMN_FAMILY1)),
ByteBufferUtil.EMPTY_BYTE_BUFFER,
ByteBufferUtil.EMPTY_BYTE_BUFFER,
false,
10));
assert CacheService.instance.keyCache.size() == 4;
}
}

115
test/unit/org/apache/cassandra/db/RowCacheTest.java
View File

@ -19,21 +19,27 @@
package org.apache.cassandra.db;
import java.util.Collection;
import java.util.Set;
import org.junit.AfterClass;
import org.junit.Test;
import org.apache.cassandra.CleanupHelper;
import org.apache.cassandra.Util;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.db.filter.QueryPath;
public class RowCacheTest extends CleanupHelper
{
private String KEYSPACE = "RowCacheSpace";
private String COLUMN_FAMILY_WITH_CACHE = "CachedCF";
private String COLUMN_FAMILY_WITHOUT_CACHE = "CFWithoutCache";
private String COLUMN_FAMILY = "CachedCF";
@AfterClass
public static void cleanup()
{
cleanupSavedCaches();
}
@Test
public void testRowCache() throws Exception
@ -41,24 +47,25 @@ public class RowCacheTest extends CleanupHelper
CompactionManager.instance.disableAutoCompaction();
Table table = Table.open(KEYSPACE);
ColumnFamilyStore cachedStore = table.getColumnFamilyStore(COLUMN_FAMILY_WITH_CACHE);
ColumnFamilyStore noCacheStore = table.getColumnFamilyStore(COLUMN_FAMILY_WITHOUT_CACHE);
ColumnFamilyStore cachedStore = table.getColumnFamilyStore(COLUMN_FAMILY);
// empty the row cache
cachedStore.invalidateRowCache();
CacheService.instance.invalidateRowCache();
// set global row cache size to 1 MB
CacheService.instance.setRowCacheCapacityInMB(1);
// inserting 100 rows into both column families
insertData(KEYSPACE, COLUMN_FAMILY_WITH_CACHE, 0, 100);
insertData(KEYSPACE, COLUMN_FAMILY_WITHOUT_CACHE, 0, 100);
insertData(KEYSPACE, COLUMN_FAMILY, 0, 100);
// now reading rows one by one and checking if row change grows
for (int i = 0; i < 100; i++)
{
DecoratedKey key = Util.dk("key" + i);
QueryPath path = new QueryPath(COLUMN_FAMILY_WITH_CACHE, null, ByteBufferUtil.bytes("col" + i));
QueryPath path = new QueryPath(COLUMN_FAMILY, null, ByteBufferUtil.bytes("col" + i));
cachedStore.getColumnFamily(key, path, ByteBufferUtil.EMPTY_BYTE_BUFFER, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, 1);
assert cachedStore.getRowCacheSize() == i + 1;
assert CacheService.instance.rowCache.size() == i + 1;
assert cachedStore.getRawCachedRow(key) != null; // current key should be stored in the cache
// checking if column is read correctly after cache
@ -70,24 +77,17 @@ public class RowCacheTest extends CleanupHelper
assert columns.size() == 1;
assert column.name().equals(ByteBufferUtil.bytes("col" + i));
assert column.value().equals(ByteBufferUtil.bytes("val" + i));
path = new QueryPath(COLUMN_FAMILY_WITHOUT_CACHE, null, ByteBufferUtil.bytes("col" + i));
// row cache should not get populated for the second store
noCacheStore.getColumnFamily(key, path, ByteBufferUtil.EMPTY_BYTE_BUFFER, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, 1);
assert noCacheStore.getRowCacheSize() == 0;
}
// insert 10 more keys and check that row cache is still at store.getRowCacheCapacity()
insertData(KEYSPACE, COLUMN_FAMILY_WITH_CACHE, 100, 10);
// insert 10 more keys
insertData(KEYSPACE, COLUMN_FAMILY, 100, 10);
for (int i = 100; i < 110; i++)
{
DecoratedKey key = Util.dk("key" + i);
QueryPath path = new QueryPath(COLUMN_FAMILY_WITH_CACHE, null, ByteBufferUtil.bytes("col" + i));
QueryPath path = new QueryPath(COLUMN_FAMILY, null, ByteBufferUtil.bytes("col" + i));
cachedStore.getColumnFamily(key, path, ByteBufferUtil.EMPTY_BYTE_BUFFER, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, 1);
assert cachedStore.getRowCacheSize() == cachedStore.getRowCacheCapacity();
assert cachedStore.getRawCachedRow(key) != null; // cache should be populated with the latest rows read (old ones should be popped)
// checking if column is read correctly after cache
@ -101,85 +101,58 @@ public class RowCacheTest extends CleanupHelper
assert column.value().equals(ByteBufferUtil.bytes("val" + i));
}
// clear all 100 rows from the cache
int keysLeft = 99;
// clear 100 rows from the cache
int keysLeft = 109;
for (int i = 109; i >= 10; i--)
{
cachedStore.invalidateCachedRow(Util.dk("key" + i));
assert cachedStore.getRowCacheSize() == keysLeft;
assert CacheService.instance.rowCache.size() == keysLeft;
keysLeft--;
}
CacheService.instance.setRowCacheCapacityInMB(0);
}
@Test
public void testRowCacheLoad() throws Exception
{
rowCacheLoad(100, 100, Integer.MAX_VALUE, false);
CacheService.instance.setRowCacheCapacityInMB(1);
rowCacheLoad(100, Integer.MAX_VALUE, false);
CacheService.instance.setRowCacheCapacityInMB(0);
}
@Test
public void testRowCachePartialLoad() throws Exception
{
rowCacheLoad(100, 50, 50, false);
CacheService.instance.setRowCacheCapacityInMB(1);
rowCacheLoad(100, 50, true);
CacheService.instance.setRowCacheCapacityInMB(0);
}
@Test
public void testRowCacheCapacityLoad() throws Exception
{
// 60 is default from DatabaseDescriptor
rowCacheLoad(100, 60, Integer.MAX_VALUE, true);
}
public void rowCacheLoad(int totalKeys, int expectedKeys, int keysToSave, boolean reduceLoadCapacity) throws Exception
public void rowCacheLoad(int totalKeys, int keysToSave, boolean reduceLoadCapacity) throws Exception
{
CompactionManager.instance.disableAutoCompaction();
ColumnFamilyStore store = Table.open(KEYSPACE).getColumnFamilyStore(COLUMN_FAMILY_WITH_CACHE);
ColumnFamilyStore store = Table.open(KEYSPACE).getColumnFamilyStore(COLUMN_FAMILY);
// empty the cache
store.invalidateRowCache();
assert store.getRowCacheSize() == 0;
CacheService.instance.invalidateRowCache();
assert CacheService.instance.rowCache.size() == 0;
// insert data and fill the cache
insertData(KEYSPACE, COLUMN_FAMILY_WITH_CACHE, 0, totalKeys);
readData(KEYSPACE, COLUMN_FAMILY_WITH_CACHE, 0, totalKeys);
assert store.getRowCacheSize() == totalKeys;
insertData(KEYSPACE, COLUMN_FAMILY, 0, totalKeys);
readData(KEYSPACE, COLUMN_FAMILY, 0, totalKeys);
assert CacheService.instance.rowCache.size() == totalKeys;
// force the cache to disk
store.rowCache.submitWrite(keysToSave).get();
CacheService.instance.rowCache.submitWrite(keysToSave).get();
if (reduceLoadCapacity)
store.reduceCacheSizes();
CacheService.instance.reduceRowCacheSize();
// empty the cache again to make sure values came from disk
store.invalidateRowCache();
assert store.getRowCacheSize() == 0;
// load the cache from disk
store.initCaches();
assert store.getRowCacheSize() == expectedKeys;
// If we are loading less than the entire cache back, we can't
// be sure which rows we will get if all rows are equally hot.
int nulls = 0;
int nonNull = 0;
for (int i = 0; i < expectedKeys; i++)
{
// verify the correct data was found when we expect to get
// back the entire cache. Otherwise only make assertions
// about how many items are read back.
ColumnFamily row = store.getRawCachedRow(Util.dk("key" + i));
if (expectedKeys == totalKeys)
{
assert row != null;
assert row.getColumn(ByteBufferUtil.bytes("col" + i)).value().equals(ByteBufferUtil.bytes("val" + i));
}
if (row == null)
nulls++;
else
nonNull++;
}
assert nulls + nonNull == expectedKeys;
CacheService.instance.invalidateRowCache();
assert CacheService.instance.rowCache.size() == 0;
assert CacheService.instance.rowCache.readSaved(KEYSPACE, COLUMN_FAMILY).size() == (keysToSave == Integer.MAX_VALUE ? totalKeys : keysToSave);
}
}

3
test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java
View File

@ -40,6 +40,7 @@ import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.io.util.FileDataInput;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.MmappedSegmentedFile;
@ -167,7 +168,7 @@ public class SSTableReaderTest extends CleanupHelper
{
Table table = Table.open("Keyspace1");
ColumnFamilyStore store = table.getColumnFamilyStore("Standard2");
store.getKeyCache().setCapacity(100);
CacheService.instance.keyCache.setCapacity(100);
// insert data and compact to a single sstable
CompactionManager.instance.disableAutoCompaction();