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

Unify key cache per CF under SSTableTracker, hooked in w/ setTrackedBy replacing addFinalizingReference. 

patch by Stu Hood and jbellis for CASSANDRA-801

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/branches/cassandra-0.6@916958 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-02-27 15:10:33 +00:00
parent ac7b875bd6
commit 197f1ecbfe
7 changed files with 94 additions and 236 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

99
src/java/org/apache/cassandra/cache/JMXAggregatingCache.java vendored
View File

@ -1,99 +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.
*
*/
public class JMXAggregatingCache implements JMXAggregatingCacheMBean
{
private final Iterable<IAggregatableCacheProvider> cacheProviders;
public JMXAggregatingCache(Iterable<IAggregatableCacheProvider> caches, String table, String name)
{
this.cacheProviders = caches;
AbstractCache.registerMBean(this, table, name);
}
public int getCapacity()
{
int capacity = 0;
for (IAggregatableCacheProvider cacheProvider : cacheProviders)
{
capacity += cacheProvider.getCache().getCapacity();
}
return capacity;
}
public void setCapacity(int capacity)
{
long totalObjects = 0;
for (IAggregatableCacheProvider cacheProvider : cacheProviders)
{
totalObjects += cacheProvider.getObjectCount();
}
for (IAggregatableCacheProvider cacheProvider : cacheProviders)
{
double ratio = ((double)cacheProvider.getObjectCount()) / totalObjects;
cacheProvider.getCache().setCapacity((int)(capacity * ratio));
}
}
public int getSize()
{
int size = 0;
for (IAggregatableCacheProvider cacheProvider : cacheProviders)
{
size += cacheProvider.getCache().getSize();
}
return size;
}
public long getRequests()
{
long requests = 0;
for (IAggregatableCacheProvider cacheProvider : cacheProviders)
{
requests += cacheProvider.getCache().getRequests();
}
return requests;
}
public long getHits()
{
long hits = 0;
for (IAggregatableCacheProvider cacheProvider : cacheProviders)
{
hits += cacheProvider.getCache().getHits();
}
return hits;
}
public double getRecentHitRate()
{
int n = 0;
double rate = 0;
for (IAggregatableCacheProvider cacheProvider : cacheProviders)
{
rate += cacheProvider.getCache().getRecentHitRate();
n++;
}
return rate / n;
}
}

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

@ -18,51 +18,47 @@
package org.apache.cassandra.db;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.Closeable;
import java.lang.management.ManagementFactory;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.collect.AbstractIterator;
import com.google.common.collect.Iterables;
import org.apache.cassandra.cache.IAggregatableCacheProvider;
import org.apache.cassandra.cache.InstrumentedCache;
import org.apache.cassandra.cache.JMXAggregatingCache;
import org.apache.cassandra.cache.JMXInstrumentedCache;
import org.apache.log4j.Logger;
import org.apache.commons.collections.IteratorUtils;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import com.google.common.collect.Iterators;
import org.apache.cassandra.cache.InstrumentedCache;
import org.apache.cassandra.cache.JMXInstrumentedCache;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.commitlog.CommitLogSegment;
import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.*;
import org.apache.cassandra.io.SSTable;
import org.apache.cassandra.io.SSTableReader;
import org.apache.cassandra.io.SSTableScanner;
import org.apache.cassandra.io.SSTableTracker;
import org.apache.cassandra.io.util.FileUtils;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.commons.collections.IteratorUtils;
import com.google.common.collect.Iterators;
import com.google.common.base.Predicate;
public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
@ -190,46 +186,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
sstables.add(sstable);
}
ssTables_ = new SSTableTracker();
ssTables_ = new SSTableTracker(table, columnFamilyName);
ssTables_.add(sstables);
int cacheSize = DatabaseDescriptor.getRowsCachedFor(table, columnFamilyName, ssTables_.estimatedKeys());
int rowCacheSize = DatabaseDescriptor.getRowsCachedFor(table, columnFamilyName, ssTables_.estimatedKeys());
if (logger_.isDebugEnabled())
logger_.debug("row cache capacity for " + columnFamilyName + " is " + cacheSize);
rowCache = new JMXInstrumentedCache<String, ColumnFamily>(table, columnFamilyName + "RowCache", cacheSize);
// we don't need to keep a reference to the key cache aggregator, just create it so it registers itself w/ JMX
new JMXAggregatingCache(new Iterable<IAggregatableCacheProvider>()
{
public Iterator<IAggregatableCacheProvider> iterator()
{
final Iterator<SSTableReader> iter = ssTables_.iterator();
return new AbstractIterator<IAggregatableCacheProvider>()
{
@Override
protected IAggregatableCacheProvider computeNext()
{
if (!iter.hasNext())
return endOfData();
return new IAggregatableCacheProvider()
{
SSTableReader sstable = iter.next();
public InstrumentedCache getCache()
{
return sstable.getKeyCache();
}
public long getObjectCount()
{
return sstable.getIndexPositions().size() * SSTableReader.indexInterval();
}
};
}
};
}
}, table, columnFamilyName + "KeyCache");
logger_.debug("row cache capacity for " + columnFamilyName + " is " + rowCacheSize);
rowCache = new JMXInstrumentedCache<String, ColumnFamily>(table, columnFamilyName + "RowCache", rowCacheSize);
}
public static ColumnFamilyStore createColumnFamilyStore(String table, String columnFamily) throws IOException

54
src/java/org/apache/cassandra/io/SSTableReader.java
View File

@ -32,6 +32,7 @@ import org.apache.commons.lang.StringUtils;
import org.apache.cassandra.cache.InstrumentedCache;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.utils.BloomFilter;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.*;
@ -108,8 +109,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
public static SSTableReader open(String dataFileName) throws IOException
{
return open(dataFileName,
StorageService.getPartitioner());
return open(dataFileName, StorageService.getPartitioner());
}
public static SSTableReader open(String dataFileName, IPartitioner partitioner) throws IOException
@ -122,8 +122,6 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
sstable.loadIndexFile();
sstable.loadBloomFilter();
long expectedKeys = (sstable.getIndexPositions().size() + 1) * INDEX_INTERVAL;
sstable.keyCache = createKeyCache(parseTableName(dataFileName), parseColumnFamilyName(dataFileName), expectedKeys);
if (logger.isDebugEnabled())
logger.debug("INDEX LOAD TIME for " + dataFileName + ": " + (System.currentTimeMillis() - start) + " ms.");
@ -135,24 +133,15 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
private final MappedByteBuffer[] indexBuffers;
private final MappedByteBuffer[] buffers;
public static InstrumentedCache<DecoratedKey, PositionSize> createKeyCache(String ksname, String cfname, long expectedKeys)
{
int keysToCache = DatabaseDescriptor.getKeysCachedFor(ksname, cfname, expectedKeys);
return new InstrumentedCache<DecoratedKey, PositionSize>(keysToCache);
}
private InstrumentedCache<DecoratedKey, PositionSize> keyCache;
private InstrumentedCache<Pair<String, DecoratedKey>, PositionSize> keyCache;
SSTableReader(String filename,
IPartitioner partitioner,
List<KeyPosition> indexPositions, Map<KeyPosition, PositionSize> spannedIndexDataPositions,
BloomFilter bloomFilter,
InstrumentedCache<DecoratedKey, PositionSize> keyCache)
throws IOException
BloomFilter bloomFilter)
throws IOException
{
super(filename, partitioner);
assert keyCache != null;
if (DatabaseDescriptor.getIndexAccessMode() == DatabaseDescriptor.DiskAccessMode.mmap)
{
@ -192,13 +181,13 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
this.indexPositions = indexPositions;
this.spannedIndexDataPositions = spannedIndexDataPositions;
this.bf = bloomFilter;
this.keyCache = keyCache;
}
public void addFinalizingReference(SSTableTracker tracker)
public void setTrackedBy(SSTableTracker tracker)
{
phantomReference = new SSTableDeletingReference(tracker, this, finalizerQueue);
finalizers.add(phantomReference);
keyCache = tracker.getKeyCache();
}
private static MappedByteBuffer mmap(String filename, long start, int size) throws IOException
@ -225,7 +214,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
private SSTableReader(String filename, IPartitioner partitioner) throws IOException
{
this(filename, partitioner, null, null, null, new InstrumentedCache<DecoratedKey, PositionSize>(0));
this(filename, partitioner, null, null, null);
}
public List<KeyPosition> getIndexPositions()
@ -233,6 +222,11 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
return indexPositions;
}
public long estimatedKeys()
{
return indexPositions.size() * INDEX_INTERVAL;
}
void loadBloomFilter() throws IOException
{
DataInputStream stream = new DataInputStream(new FileInputStream(filterFilename()));
@ -320,21 +314,29 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
*/
public PositionSize getPosition(DecoratedKey decoratedKey) throws IOException
{
// first, check bloom filter
if (!bf.isPresent(partitioner.convertToDiskFormat(decoratedKey)))
return null;
if (keyCache.getCapacity() > 0)
// next, the key cache
Pair<String, DecoratedKey> unifiedKey = new Pair<String, DecoratedKey>(path, decoratedKey);
if (keyCache != null && keyCache.getCapacity() > 0)
{
PositionSize cachedPosition = keyCache.get(decoratedKey);
PositionSize cachedPosition = keyCache.get(unifiedKey);
if (cachedPosition != null)
{
return cachedPosition;
}
}
// next, see if the sampled index says it's impossible for the key to be present
KeyPosition sampledPosition = getIndexScanPosition(decoratedKey);
if (sampledPosition == null)
{
return null;
}
// handle exact sampled index hit
if (spannedIndexDataPositions != null)
{
PositionSize info = spannedIndexDataPositions.get(sampledPosition);
@ -342,6 +344,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
return info;
}
// scan the on-disk index, starting at the nearest sampled position
long p = sampledPosition.position;
FileDataInput input;
if (indexBuffers == null)
@ -383,8 +386,8 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
{
info = new PositionSize(position, length() - position);
}
if (keyCache.getCapacity() > 0)
keyCache.put(decoratedKey, info);
if (keyCache != null && keyCache.getCapacity() > 0)
keyCache.put(unifiedKey, info);
return info;
}
if (v > 0)
@ -502,11 +505,6 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
return ColumnFamily.create(getTableName(), getColumnFamilyName());
}
public InstrumentedCache<DecoratedKey, PositionSize> getKeyCache()
{
return keyCache;
}
public ICompactSerializer2<IColumn> getColumnSerializer()
{
return DatabaseDescriptor.getColumnFamilyType(getTableName(), getColumnFamilyName()).equals("Standard")

38
src/java/org/apache/cassandra/io/SSTableTracker.java
View File

@ -25,16 +25,32 @@ import java.util.*;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.cassandra.cache.JMXInstrumentedCache;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.utils.Pair;
import org.apache.log4j.Logger;
public class SSTableTracker implements Iterable<SSTableReader>
{
private static final Logger logger = Logger.getLogger(SSTableTracker.class);
private volatile Set<SSTableReader> sstables;
private final AtomicLong liveSize = new AtomicLong();
private final AtomicLong totalSize = new AtomicLong();
public SSTableTracker()
private final String ksname;
private final String cfname;
private final JMXInstrumentedCache<Pair<String, DecoratedKey>, SSTable.PositionSize> keyCache;
public SSTableTracker(String ksname, String cfname)
{
this.sstables = Collections.<SSTableReader>emptySet();
this.ksname = ksname;
this.cfname = cfname;
sstables = Collections.emptySet();
keyCache = new JMXInstrumentedCache<Pair<String, DecoratedKey>, SSTable.PositionSize>(ksname, cfname + "KeyCache", 0);
}
public synchronized void replace(Collection<SSTableReader> oldSSTables, Iterable<SSTableReader> replacements) throws IOException
@ -48,7 +64,7 @@ public class SSTableTracker implements Iterable<SSTableReader>
long size = sstable.bytesOnDisk();
liveSize.addAndGet(size);
totalSize.addAndGet(size);
sstable.addFinalizingReference(this);
sstable.setTrackedBy(this);
}
for (SSTableReader sstable : oldSSTables)
@ -60,6 +76,15 @@ public class SSTableTracker implements Iterable<SSTableReader>
}
sstables = Collections.unmodifiableSet(sstablesNew);
int keyCacheSize = DatabaseDescriptor.getKeysCachedFor(ksname, cfname, estimatedKeys());
if (keyCacheSize != keyCache.getCapacity())
{
// update cache size for the new key volume
if (logger.isDebugEnabled())
logger.debug("key cache capacity for " + cfname + " is " + keyCacheSize);
keyCache.setCapacity(keyCacheSize);
}
}
public synchronized void add(Iterable<SSTableReader> sstables)
@ -107,7 +132,7 @@ public class SSTableTracker implements Iterable<SSTableReader>
long n = 0;
for (SSTableReader sstable : this)
{
n += sstable.getIndexPositions().size() * SSTableReader.INDEX_INTERVAL;
n += sstable.estimatedKeys();
}
return n;
}
@ -126,5 +151,10 @@ public class SSTableTracker implements Iterable<SSTableReader>
{
totalSize.addAndGet(-size);
}
public JMXInstrumentedCache<Pair<String, DecoratedKey>, SSTable.PositionSize> getKeyCache()
{
return keyCache;
}
}

15
src/java/org/apache/cassandra/io/SSTableWriter.java
View File

@ -21,21 +21,23 @@ package org.apache.cassandra.io;
*/
import java.io.*;
import java.io.DataOutputStream;
import java.io.FileOutputStream;
import java.io.IOError;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import org.apache.log4j.Logger;
import org.apache.cassandra.cache.InstrumentedCache;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.BloomFilter;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
import org.apache.cassandra.io.util.DataOutputBuffer;
public class SSTableWriter extends SSTable
{
@ -151,8 +153,7 @@ public class SSTableWriter extends SSTable
rename(filterFilename());
path = rename(path); // important to do this last since index & filter file names are derived from it
InstrumentedCache<DecoratedKey, PositionSize> keyCache = SSTableReader.createKeyCache(getTableName(), getColumnFamilyName(), keysWritten);
return new SSTableReader(path, partitioner, indexPositions, spannedIndexDataPositions, bf, keyCache);
return new SSTableReader(path, partitioner, indexPositions, spannedIndexDataPositions, bf);
}
static String rename(String tmpFilename)

35
test/unit/org/apache/cassandra/io/SSTableAccessor.java
View File

@ -1,35 +0,0 @@
/*
* 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.io;
import java.io.IOException;
import org.apache.cassandra.dht.IPartitioner;
public class SSTableAccessor
{
public static SSTableReader getSSTableReader(String filename, IPartitioner<?> partitioner)
throws IOException
{
SSTableReader sstable = new SSTableReader(filename, partitioner, null, null, null, SSTableReader.createKeyCache(0));
sstable.loadBloomFilter();
sstable.loadIndexFile();
return sstable;
}
}

4
test/unit/org/apache/cassandra/io/SSTableTest.java
View File

@ -41,7 +41,7 @@ public class SSTableTest extends CleanupHelper
TreeMap<String, byte[]> map = new TreeMap<String,byte[]>();
map.put(key, bytes);
SSTableReader ssTable = SSTableUtils.writeRawSSTable("table", "singlewrite", map);
SSTableReader ssTable = SSTableUtils.writeRawSSTable("Keyspace1", "Standard1", map);
// verify
verifySingle(ssTable, bytes, key);
@ -69,7 +69,7 @@ public class SSTableTest extends CleanupHelper
}
// write
SSTableReader ssTable = SSTableUtils.writeRawSSTable("table", "manywrites", map);
SSTableReader ssTable = SSTableUtils.writeRawSSTable("Keyspace1", "Standard2", map);
// verify
verifyMany(ssTable, map);