diff --git a/src/java/org/apache/cassandra/cache/JMXAggregatingCache.java b/src/java/org/apache/cassandra/cache/JMXAggregatingCache.java deleted file mode 100644 index 3e8296c781..0000000000 --- a/src/java/org/apache/cassandra/cache/JMXAggregatingCache.java +++ /dev/null @@ -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 cacheProviders; - - public JMXAggregatingCache(Iterable 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; - } -} diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index abd2fb2900..c49138bc00 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -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(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() - { - public Iterator iterator() - { - final Iterator iter = ssTables_.iterator(); - return new AbstractIterator() - { - @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(table, columnFamilyName + "RowCache", rowCacheSize); } public static ColumnFamilyStore createColumnFamilyStore(String table, String columnFamily) throws IOException diff --git a/src/java/org/apache/cassandra/io/SSTableReader.java b/src/java/org/apache/cassandra/io/SSTableReader.java index 1c21788849..b90a090622 100644 --- a/src/java/org/apache/cassandra/io/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/SSTableReader.java @@ -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 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 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 private final MappedByteBuffer[] indexBuffers; private final MappedByteBuffer[] buffers; - - public static InstrumentedCache createKeyCache(String ksname, String cfname, long expectedKeys) - { - int keysToCache = DatabaseDescriptor.getKeysCachedFor(ksname, cfname, expectedKeys); - return new InstrumentedCache(keysToCache); - } - - private InstrumentedCache keyCache; + private InstrumentedCache, PositionSize> keyCache; SSTableReader(String filename, IPartitioner partitioner, List indexPositions, Map spannedIndexDataPositions, - BloomFilter bloomFilter, - InstrumentedCache 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 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 private SSTableReader(String filename, IPartitioner partitioner) throws IOException { - this(filename, partitioner, null, null, null, new InstrumentedCache(0)); + this(filename, partitioner, null, null, null); } public List getIndexPositions() @@ -233,6 +222,11 @@ public class SSTableReader extends SSTable implements Comparable 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 */ 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 unifiedKey = new Pair(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 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 { 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 return ColumnFamily.create(getTableName(), getColumnFamilyName()); } - public InstrumentedCache getKeyCache() - { - return keyCache; - } - public ICompactSerializer2 getColumnSerializer() { return DatabaseDescriptor.getColumnFamilyType(getTableName(), getColumnFamilyName()).equals("Standard") diff --git a/src/java/org/apache/cassandra/io/SSTableTracker.java b/src/java/org/apache/cassandra/io/SSTableTracker.java index 4e20980588..52028b5901 100644 --- a/src/java/org/apache/cassandra/io/SSTableTracker.java +++ b/src/java/org/apache/cassandra/io/SSTableTracker.java @@ -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 { + private static final Logger logger = Logger.getLogger(SSTableTracker.class); + private volatile Set 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, SSTable.PositionSize> keyCache; + + public SSTableTracker(String ksname, String cfname) { - this.sstables = Collections.emptySet(); + this.ksname = ksname; + this.cfname = cfname; + sstables = Collections.emptySet(); + keyCache = new JMXInstrumentedCache, SSTable.PositionSize>(ksname, cfname + "KeyCache", 0); } public synchronized void replace(Collection oldSSTables, Iterable replacements) throws IOException @@ -48,7 +64,7 @@ public class SSTableTracker implements Iterable 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 } 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 sstables) @@ -107,7 +132,7 @@ public class SSTableTracker implements Iterable 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 { totalSize.addAndGet(-size); } + + public JMXInstrumentedCache, SSTable.PositionSize> getKeyCache() + { + return keyCache; + } } diff --git a/src/java/org/apache/cassandra/io/SSTableWriter.java b/src/java/org/apache/cassandra/io/SSTableWriter.java index aeebf8ad02..12db44848f 100644 --- a/src/java/org/apache/cassandra/io/SSTableWriter.java +++ b/src/java/org/apache/cassandra/io/SSTableWriter.java @@ -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 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) diff --git a/test/unit/org/apache/cassandra/io/SSTableAccessor.java b/test/unit/org/apache/cassandra/io/SSTableAccessor.java deleted file mode 100644 index 7a297fe88a..0000000000 --- a/test/unit/org/apache/cassandra/io/SSTableAccessor.java +++ /dev/null @@ -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; - } -} diff --git a/test/unit/org/apache/cassandra/io/SSTableTest.java b/test/unit/org/apache/cassandra/io/SSTableTest.java index 059240469a..338045f97c 100644 --- a/test/unit/org/apache/cassandra/io/SSTableTest.java +++ b/test/unit/org/apache/cassandra/io/SSTableTest.java @@ -41,7 +41,7 @@ public class SSTableTest extends CleanupHelper TreeMap map = new TreeMap(); 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);