diff --git a/CHANGES.txt b/CHANGES.txt index 0f301224a9..0a80284e65 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -7,7 +7,6 @@ * add CL.TWO, CL.THREE (CASSANDRA-2013) * avoid exporting an un-requested row in sstable2json, when exporting a key that does not exist (CASSANDRA-2168) - * track and migrate cached pages during compaction (CASSANDRA-1902) * add incremental_backups option (CASSANDRA-1872) * add configurable row limit to Pig loadfunc (CASSANDRA-2276) * validate column values in batches as well as single-Column inserts diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 4060301ff8..1b2317b2df 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -251,10 +251,6 @@ in_memory_compaction_limit_in_mb: 64 # key caches. compaction_preheat_key_cache: true -# When set to true, this setting lets cassandra inform the OS to pre-cache -# popular data in newly compacted files. This requires Linux and JNA be installed. -enable_page_cache_migration: false - # Time to wait for a reply from other nodes before failing the command rpc_timeout_in_ms: 10000 diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 520bab4d79..169e28320c 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -111,10 +111,6 @@ public class Config public int hinted_handoff_throttle_delay_in_ms = 0; public boolean compaction_preheat_key_cache = true; - // make this configurable as its being releases in a maintenance release - // TODO: remove in 0.8 - public Boolean enable_page_cache_migration = false; - public boolean incremental_backups = false; public static enum CommitLogSync { diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index a57ae2e7cd..e2ec96de13 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -1214,11 +1214,6 @@ public class DatabaseDescriptor throw new ConfigurationException("memtable_flush_after_mins must be greater than 0."); } - public static boolean isPageCaheMigrationEnabled() - { - return conf.enable_page_cache_migration; - } - public static boolean incrementalBackupsEnabled() { return conf.incremental_backups; diff --git a/src/java/org/apache/cassandra/db/Column.java b/src/java/org/apache/cassandra/db/Column.java index acb618f5c7..9ba3820d9a 100644 --- a/src/java/org/apache/cassandra/db/Column.java +++ b/src/java/org/apache/cassandra/db/Column.java @@ -46,7 +46,6 @@ public class Column implements IColumn return new ColumnSerializer(); } - protected boolean isInPageCache; protected final ByteBuffer name; protected final ByteBuffer value; protected final long timestamp; @@ -69,7 +68,6 @@ public class Column implements IColumn this.name = name; this.value = value; this.timestamp = timestamp; - isInPageCache = false; } public ByteBuffer name() @@ -234,15 +232,5 @@ public class Column implements IColumn { return !isMarkedForDelete(); } - - public boolean isInPageCache() - { - return isInPageCache; - } - - public void setIsInPageCache(boolean isInPageCache) - { - this.isInPageCache = isInPageCache; - } } diff --git a/src/java/org/apache/cassandra/db/ColumnFamily.java b/src/java/org/apache/cassandra/db/ColumnFamily.java index 1436c54b48..9a06cc6e41 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamily.java +++ b/src/java/org/apache/cassandra/db/ColumnFamily.java @@ -215,13 +215,6 @@ public class ColumnFamily implements IColumnContainer, IIterableColumns IColumn oldColumn; while ((oldColumn = columns.putIfAbsent(name, column)) != null) { - // migrate any page cache info (prefer cached) - if (oldColumn.isInPageCache() || column.isInPageCache()) - { - oldColumn.setIsInPageCache(true); - column.setIsInPageCache(true); - } - if (oldColumn instanceof SuperColumn) { ((SuperColumn) oldColumn).putColumn(column); diff --git a/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java b/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java index 1a898e428c..b4961e907c 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java @@ -21,7 +21,9 @@ package org.apache.cassandra.db; */ -import java.io.*; +import java.io.DataInput; +import java.io.DataOutput; +import java.io.IOException; import java.util.Collection; import org.slf4j.Logger; @@ -29,8 +31,6 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.io.ICompactSerializer2; -import org.apache.cassandra.io.util.PageCacheInformer; -import org.apache.cassandra.utils.PageCacheMetrics; public class ColumnFamilySerializer implements ICompactSerializer2 { @@ -70,16 +70,11 @@ public class ColumnFamilySerializer implements ICompactSerializer2 { throw new RuntimeException(e); } - serializeForSSTable(columnFamily, dos); } public int serializeForSSTable(ColumnFamily columnFamily, DataOutput dos) { - PageCacheInformer pci = dos instanceof PageCacheInformer - ? (PageCacheInformer) dos - : null; - try { serializeCFInfo(columnFamily, dos); @@ -89,16 +84,8 @@ public class ColumnFamilySerializer implements ICompactSerializer2 dos.writeInt(count); for (IColumn column : columns) { - long startAt = pci != null ? pci.getCurrentPosition() : -1; - columnFamily.getColumnSerializer().serialize(column, dos); - - //Track the section of serialized data that should - //be included in the page cache (compaction) - if (column.isInPageCache() && pci != null) - pci.keepCacheWindow(startAt); } - return count; } catch (IOException e) @@ -130,49 +117,18 @@ public class ColumnFamilySerializer implements ICompactSerializer2 throw new UnserializableColumnFamilyException("Couldn't find cfId=" + cfId, cfId); ColumnFamily cf = ColumnFamily.create(cfId); deserializeFromSSTableNoColumns(cf, dis); - deserializeColumns(dis, cf, null); + deserializeColumns(dis, cf); return cf; } - public boolean deserializeColumns(DataInput dis, ColumnFamily cf, PageCacheMetrics pageCacheMetrics) throws IOException + public void deserializeColumns(DataInput dis, ColumnFamily cf) throws IOException { - int size = dis.readInt(); - - boolean hasColumnsInPageCache = false; - - - if (pageCacheMetrics != null && dis instanceof RandomAccessFile) + for (int i = 0; i < size; ++i) { - RandomAccessFile raf = (RandomAccessFile) dis; - - for (int i = 0; i < size; ++i) - { - long startAt = raf.getFilePointer(); - - IColumn column = cf.getColumnSerializer().deserialize(dis); - - long endAt = raf.getFilePointer(); - - column.setIsInPageCache(pageCacheMetrics.isRangeInCache(startAt, endAt)); - - if(!hasColumnsInPageCache) - hasColumnsInPageCache = column.isInPageCache(); - - cf.addColumn(column); - } + IColumn column = cf.getColumnSerializer().deserialize(dis); + cf.addColumn(column); } - else - { - for (int i = 0; i < size; ++i) - { - IColumn column = cf.getColumnSerializer().deserialize(dis); - - cf.addColumn(column); - } - } - - return hasColumnsInPageCache; } public ColumnFamily deserializeFromSSTableNoColumns(ColumnFamily cf, DataInput input) throws IOException diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 7270b0678b..75dcf587c4 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -32,6 +32,7 @@ import javax.management.ObjectName; import com.google.common.collect.Iterables; import org.apache.commons.collections.IteratorUtils; +import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -334,7 +335,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean { public void run() { - logger.info("Creating index {}.{}", table, indexedCfMetadata.cfName); try { forceBlockingFlush(); @@ -348,7 +348,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean throw new AssertionError(e); } buildSecondaryIndexes(getSSTables(), FBUtilities.singleton(info.name)); - logger.info("Index {} complete", indexedCfMetadata.cfName); SystemTable.setIndexBuilt(table.name, indexedCfMetadata.cfName); } }; @@ -357,7 +356,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean public void buildSecondaryIndexes(Collection sstables, SortedSet columns) { - logger.debug("Submitting index build to compactionmanager"); + logger.info(String.format("Submitting index build of %s for data in %s", + metadata.comparator.getString(columns), StringUtils.join(sstables, ", "))); Table.IndexBuilder builder = table.createIndexBuilder(this, columns, new ReducingKeyIterator(sstables)); Future future = CompactionManager.instance.submitIndexBuild(this, builder); try @@ -374,6 +374,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean { throw new RuntimeException(e); } + logger.info("Index build of " + metadata.comparator.getString(columns) + " complete"); } // called when dropping or renaming a CF. Performs mbean housekeeping and invalidates CFS to other operations. @@ -684,26 +685,31 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean try { if (oldMemtable.isFrozen()) + { + logger.debug("memtable is already frozen; another thread must be flushing it"); return null; + } boolean isDropped = isIndex() ? DatabaseDescriptor.getCFMetaData(table.name, getParentColumnfamily()) == null : DatabaseDescriptor.getCFMetaData(metadata.cfId) == null; if (isDropped) - return null; // column family was dropped. no point in flushing. + { + logger.debug("column family was dropped; no point in flushing"); + return null; + } assert memtable == oldMemtable; memtable.freeze(); final CommitLogSegment.CommitLogContext ctx = writeCommitLog ? CommitLog.instance.getContext() : null; - logger.info("switching in a fresh Memtable for " + columnFamily + " at " + ctx); // submit the memtable for any indexed sub-cfses, and our own. List icc = new ArrayList(indexedColumns.size()); - icc.add(this); - for (ColumnFamilyStore indexCfs : indexedColumns.values()) + // don't assume that this.memtable is dirty; forceFlush can bring us here during index build even if it is not + for (ColumnFamilyStore cfs : Iterables.concat(Collections.singleton(this), indexedColumns.values())) { - if (!indexCfs.memtable.isClean()) - icc.add(indexCfs); + if (!cfs.memtable.isClean()) + icc.add(cfs); } final CountDownLatch latch = new CountDownLatch(icc.size()); for (ColumnFamilyStore cfs : icc) @@ -711,6 +717,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean submitFlush(cfs.memtable, latch); cfs.memtable = new Memtable(cfs); } + // we marked our memtable as frozen as part of the concurrency control, + // so even if there was nothing to flush we need to switch it out + if (!icc.contains(this)) + memtable = new Memtable(this); // when all the memtables have been written, including for indexes, mark the flush in the commitlog header. // a second executor makes sure the onMemtableFlushes get called in the right order, @@ -754,8 +764,17 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean public Future forceFlush() { - if (memtable.isClean()) + // during index build, 2ary index memtables can be dirty even if parent is not. if so, + // we want flushLargestMemtables to flush the 2ary index ones too. + boolean clean = true; + for (ColumnFamilyStore cfs : Iterables.concat(Collections.singleton(this), getIndexColumnFamilyStores())) + clean &= cfs.memtable.isClean(); + + if (clean) + { + logger.debug("forceFlush requested but everything is clean"); return null; + } return maybeSwitchMemtable(memtable, true); } @@ -1937,6 +1956,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean return indexedColumns.get(column); } + public Collection getIndexColumnFamilyStores() + { + return indexedColumns.values(); + } + public ColumnFamily newIndexedColumnFamily(ByteBuffer column) { return ColumnFamily.create(indexedColumns.get(column).metadata); @@ -2161,11 +2185,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean public SSTableWriter createFlushWriter(long estimatedRows) throws IOException { - return new SSTableWriter(getFlushPath(), estimatedRows, metadata, partitioner, false); + return new SSTableWriter(getFlushPath(), estimatedRows, metadata, partitioner); } - public SSTableWriter createCompactionWriter(long estimatedRows, String location, boolean migratePageCache) throws IOException + public SSTableWriter createCompactionWriter(long estimatedRows, String location) throws IOException { - return new SSTableWriter(getTempSSTablePath(location), estimatedRows, metadata, partitioner, migratePageCache); + return new SSTableWriter(getTempSSTablePath(location), estimatedRows, metadata, partitioner); } } diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java index b84be781fe..755880460b 100644 --- a/src/java/org/apache/cassandra/db/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/CompactionManager.java @@ -46,7 +46,6 @@ import org.apache.cassandra.io.*; import org.apache.cassandra.io.sstable.*; import org.apache.cassandra.io.util.BufferedRandomAccessFile; import org.apache.cassandra.io.util.FileUtils; -import org.apache.cassandra.io.util.PageCacheInformer; import org.apache.cassandra.service.AntiEntropyService; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.ByteBufferUtil; @@ -390,7 +389,6 @@ public class CompactionManager implements CompactionManagerMBean assert sstables != null; Table table = cfs.table; - if (DatabaseDescriptor.isSnapshotBeforeCompaction()) table.snapshot("compact-" + cfs.columnFamily); @@ -400,11 +398,9 @@ public class CompactionManager implements CompactionManagerMBean assert sstable.descriptor.cfname.equals(cfs.columnFamily); String compactionFileLocation = table.getDataFileLocation(cfs.getExpectedCompactedFileSize(sstables)); - // If the compaction file path is null that means we have no space left for this compaction. // try again w/o the largest one. List smallerSSTables = new ArrayList(sstables); - while (compactionFileLocation == null && smallerSSTables.size() > 1) { logger.warn("insufficient space to compact all requested files " + StringUtils.join(smallerSSTables, ", ")); @@ -416,7 +412,6 @@ public class CompactionManager implements CompactionManagerMBean logger.error("insufficient space to compact even the two smallest files, aborting"); return 0; } - sstables = smallerSSTables; // new sstables from flush can be added during a compaction, but only the compaction can remove them, @@ -428,9 +423,9 @@ public class CompactionManager implements CompactionManagerMBean long totalkeysWritten = 0; // TODO the int cast here is potentially buggy - int expectedBloomFilterSize = Math.max(DatabaseDescriptor.getIndexInterval(), (int) SSTableReader.getApproximateKeyCount(sstables)); + int expectedBloomFilterSize = Math.max(DatabaseDescriptor.getIndexInterval(), (int)SSTableReader.getApproximateKeyCount(sstables)); if (logger.isDebugEnabled()) - logger.debug("Expected bloom filter size : " + expectedBloomFilterSize); + logger.debug("Expected bloom filter size : " + expectedBloomFilterSize); SSTableWriter writer; CompactionIterator ci = new CompactionIterator(cfs, sstables, gcBefore, major); // retain a handle so we can call close() @@ -450,13 +445,11 @@ public class CompactionManager implements CompactionManagerMBean return 0; } - writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation, ci.hasRowsInPageCache()); + writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation); while (nni.hasNext()) { AbstractCompactedRow row = nni.next(); - long position = writer.append(row); - totalkeysWritten++; if (DatabaseDescriptor.getPreheatKeyCache()) @@ -486,7 +479,7 @@ public class CompactionManager implements CompactionManagerMBean long dTime = System.currentTimeMillis() - startTime; long startsize = SSTable.getTotalBytes(sstables); long endsize = ssTable.length(); - double ratio = (double) endsize / (double) startsize; + double ratio = (double)endsize / (double)startsize; logger.info(String.format("Compacted to %s. %,d to %,d (~%d%% of original) bytes for %,d keys. Time: %,dms.", writer.getFilename(), startsize, endsize, (int) (ratio * 100), totalkeysWritten, dTime)); return sstables.size(); @@ -539,7 +532,7 @@ public class CompactionManager implements CompactionManagerMBean assert firstRowPositionFromIndex == 0 : firstRowPositionFromIndex; } - SSTableWriter writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, null, false); + SSTableWriter writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, null); executor.beginCompaction(cfs.columnFamily, new ScrubInfo(dataFile, sstable)); int goodRows = 0, badRows = 0; @@ -594,7 +587,7 @@ public class CompactionManager implements CompactionManagerMBean throw new IOError(new IOException("Unable to read row key from data file")); if (dataSize > dataFile.length()) throw new IOError(new IOException("Impossible row size " + dataSize)); - SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataStart, dataSize, null, true); + SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataStart, dataSize, true); writer.append(getCompactedRow(row, cfs, sstable.descriptor, true)); goodRows++; if (!key.key.equals(currentIndexKey) || dataStart != dataStartFromIndex) @@ -614,7 +607,7 @@ public class CompactionManager implements CompactionManagerMBean key = SSTableReader.decodeKey(sstable.partitioner, sstable.descriptor, currentIndexKey); try { - SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataStartFromIndex, dataSizeFromIndex, null, true); + SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataStartFromIndex, dataSizeFromIndex, true); writer.append(getCompactedRow(row, cfs, sstable.descriptor, true)); goodRows++; } @@ -699,7 +692,7 @@ public class CompactionManager implements CompactionManagerMBean SSTableIdentityIterator row = (SSTableIdentityIterator) scanner.next(); if (Range.isTokenInRanges(row.getKey().token, ranges)) { - writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, writer, row.hasRowsInPageCache()); + writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, writer); writer.append(getCompactedRow(row, cfs, sstable.descriptor, false)); totalkeysWritten++; } @@ -768,13 +761,13 @@ public class CompactionManager implements CompactionManagerMBean : new PrecompactedRow(cfs, Arrays.asList(row), false, getDefaultGcBefore(cfs), forceDeserialize); } - private SSTableWriter maybeCreateWriter(ColumnFamilyStore cfs, String compactionFileLocation, int expectedBloomFilterSize, SSTableWriter writer, boolean migrateCachedPages) + private SSTableWriter maybeCreateWriter(ColumnFamilyStore cfs, String compactionFileLocation, int expectedBloomFilterSize, SSTableWriter writer) throws IOException { if (writer == null) { FileUtils.createDirectory(compactionFileLocation); - writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation, migrateCachedPages); + writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation); } return writer; } @@ -1120,7 +1113,7 @@ public class CompactionManager implements CompactionManagerMBean this.row = row; } - public void write(PageCacheInformer out) throws IOException + public void write(DataOutput out) throws IOException { assert row.dataSize > 0; out.writeLong(row.dataSize); diff --git a/src/java/org/apache/cassandra/db/IColumn.java b/src/java/org/apache/cassandra/db/IColumn.java index 1f07f1298f..61bb6783c3 100644 --- a/src/java/org/apache/cassandra/db/IColumn.java +++ b/src/java/org/apache/cassandra/db/IColumn.java @@ -55,12 +55,4 @@ public interface IColumn * supercolumn deleted-at time. */ boolean isLive(); - - /** - * Used to identify columns during compaction that are in the os page cache - * so that they can be re-cached in new SSTables - * @return - */ - boolean isInPageCache(); - void setIsInPageCache(boolean isInPageCache); } diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java index 4344553185..5392a92c88 100644 --- a/src/java/org/apache/cassandra/db/RowMutation.java +++ b/src/java/org/apache/cassandra/db/RowMutation.java @@ -18,7 +18,10 @@ package org.apache.cassandra.db; -import java.io.*; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; import java.nio.ByteBuffer; import java.util.*; import java.util.concurrent.ExecutionException; diff --git a/src/java/org/apache/cassandra/db/SuperColumn.java b/src/java/org/apache/cassandra/db/SuperColumn.java index acbc51d062..8f73646428 100644 --- a/src/java/org/apache/cassandra/db/SuperColumn.java +++ b/src/java/org/apache/cassandra/db/SuperColumn.java @@ -53,7 +53,6 @@ public class SuperColumn implements IColumn, IColumnContainer private ConcurrentSkipListMap columns_; private AtomicInteger localDeletionTime = new AtomicInteger(Integer.MIN_VALUE); private AtomicLong markedForDeleteAt = new AtomicLong(Long.MIN_VALUE); - private boolean isInPageCache = false; public SuperColumn(ByteBuffer name, AbstractType comparator) { @@ -307,16 +306,6 @@ public class SuperColumn implements IColumn, IColumnContainer { throw new UnsupportedOperationException("This operation is unsupported on super columns."); } - - public boolean isInPageCache() - { - return isInPageCache; - } - - public void setIsInPageCache(boolean isInPageCache) - { - this.isInPageCache = isInPageCache; - } } class SuperColumnSerializer implements ICompactSerializer2 diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java index b2cd5cda85..3d5ad19e11 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java @@ -173,7 +173,7 @@ public class CommitLog for (File file : clogs) { int bufferSize = (int) Math.min(Math.max(file.length(), 1), 32 * 1024 * 1024); - BufferedRandomAccessFile reader = new BufferedRandomAccessFile(new File(file.getAbsolutePath()), "r", bufferSize, true, false); + BufferedRandomAccessFile reader = new BufferedRandomAccessFile(new File(file.getAbsolutePath()), "r", bufferSize, true); try { diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java index 0a1d7b8d8c..4bc12bdcaf 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java @@ -73,7 +73,7 @@ public class CommitLogSegment private static BufferedRandomAccessFile createWriter(String file) throws IOException { - return new BufferedRandomAccessFile(new File(file), "rw", 128 * 1024, true, false); + return new BufferedRandomAccessFile(new File(file), "rw", 128 * 1024, true); } public CommitLogSegment.CommitLogContext write(RowMutation rowMutation, Object serializedRow) throws IOException diff --git a/src/java/org/apache/cassandra/io/AbstractCompactedRow.java b/src/java/org/apache/cassandra/io/AbstractCompactedRow.java index 77ff784697..eaaa157219 100644 --- a/src/java/org/apache/cassandra/io/AbstractCompactedRow.java +++ b/src/java/org/apache/cassandra/io/AbstractCompactedRow.java @@ -21,11 +21,11 @@ package org.apache.cassandra.io; */ +import java.io.DataOutput; import java.io.IOException; import java.security.MessageDigest; import org.apache.cassandra.db.DecoratedKey; -import org.apache.cassandra.io.util.PageCacheInformer; /** * a CompactedRow is an object that takes a bunch of rows (keys + columnfamilies) @@ -35,7 +35,6 @@ import org.apache.cassandra.io.util.PageCacheInformer; public abstract class AbstractCompactedRow { public final DecoratedKey key; - protected boolean hasColumnsInPageCache = false; public AbstractCompactedRow(DecoratedKey key) { @@ -45,7 +44,7 @@ public abstract class AbstractCompactedRow /** * write the row (size + column index + filter + column data, but NOT row key) to @param out */ - public abstract void write(PageCacheInformer out) throws IOException; + public abstract void write(DataOutput out) throws IOException; /** * update @param digest with the data bytes of the row (not including row key or row size) @@ -61,18 +60,4 @@ public abstract class AbstractCompactedRow * @return the number of columns in the row */ public abstract int columnCount(); - - /** - * @return if any columns in this row are in the OS Page Cache - */ - public boolean hasColumnsInPageCache() - { - return hasColumnsInPageCache; - } - - public void setHasColumnsInPageCache(boolean hasColumnsInPageCache) - { - this.hasColumnsInPageCache = hasColumnsInPageCache; - } - } diff --git a/src/java/org/apache/cassandra/io/CompactionIterator.java b/src/java/org/apache/cassandra/io/CompactionIterator.java index 6ebc5b0b3c..88271e5515 100644 --- a/src/java/org/apache/cassandra/io/CompactionIterator.java +++ b/src/java/org/apache/cassandra/io/CompactionIterator.java @@ -48,7 +48,6 @@ implements Closeable, ICompactionInfo public static final int FILE_BUFFER_SIZE = 1024 * 1024; protected final List rows = new ArrayList(); - private final ColumnFamilyStore cfs; private final int gcBefore; private final boolean major; @@ -56,7 +55,6 @@ implements Closeable, ICompactionInfo private long totalBytes; private long bytesRead; private long row; - private boolean hasRowsInPageCache; public CompactionIterator(ColumnFamilyStore cfs, Iterable sstables, int gcBefore, boolean major) throws IOException { @@ -99,9 +97,6 @@ implements Closeable, ICompactionInfo public void reduce(SSTableIdentityIterator current) { rows.add(current); - - if(current.hasRowsInPageCache()) - hasRowsInPageCache = true; } protected AbstractCompactedRow getReduced() @@ -130,7 +125,6 @@ implements Closeable, ICompactionInfo protected AbstractCompactedRow getCompactedRow() { long rowSize = 0; - for (SSTableIdentityIterator row : rows) { rowSize += row.dataSize; @@ -168,11 +162,6 @@ implements Closeable, ICompactionInfo return bytesRead; } - public boolean hasRowsInPageCache() - { - return hasRowsInPageCache; - } - public String getTaskType() { return major ? "Major" : "Minor"; diff --git a/src/java/org/apache/cassandra/io/LazilyCompactedRow.java b/src/java/org/apache/cassandra/io/LazilyCompactedRow.java index 8c26b772c1..82f359d4ac 100644 --- a/src/java/org/apache/cassandra/io/LazilyCompactedRow.java +++ b/src/java/org/apache/cassandra/io/LazilyCompactedRow.java @@ -38,7 +38,8 @@ import org.apache.cassandra.db.IColumn; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.io.sstable.SSTable; import org.apache.cassandra.io.sstable.SSTableIdentityIterator; -import org.apache.cassandra.io.util.*; +import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.io.util.IIterableColumns; import org.apache.cassandra.utils.ReducingIterator; /** @@ -93,7 +94,7 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl iter = null; } - public void write(PageCacheInformer out) throws IOException + public void write(DataOutput out) throws IOException { if (rows.size() == 1 && !shouldPurge && rows.get(0).sstable.descriptor.isLatestVersion && !forceDeserialize) { diff --git a/src/java/org/apache/cassandra/io/PrecompactedRow.java b/src/java/org/apache/cassandra/io/PrecompactedRow.java index 02112a74ea..f02c58ce39 100644 --- a/src/java/org/apache/cassandra/io/PrecompactedRow.java +++ b/src/java/org/apache/cassandra/io/PrecompactedRow.java @@ -21,6 +21,7 @@ package org.apache.cassandra.io; */ +import java.io.DataOutput; import java.io.IOError; import java.io.IOException; import java.security.MessageDigest; @@ -37,8 +38,6 @@ import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.io.sstable.SSTable; import org.apache.cassandra.io.sstable.SSTableIdentityIterator; import org.apache.cassandra.io.util.DataOutputBuffer; -import org.apache.cassandra.io.util.PageCacheInformer; -import org.apache.cassandra.utils.Pair; /** * PrecompactedRow merges its rows in its constructor in memory. @@ -91,11 +90,7 @@ public class PrecompactedRow extends AbstractCompactedRow { cf.addAll(thisCF); } - - if (row.hasColumnsInPageCache()) - this.hasColumnsInPageCache = true; } - ColumnFamily cfPurged = shouldPurge ? ColumnFamilyStore.removeDeleted(cf, gcBefore) : cf; if (cfPurged == null) return; @@ -116,47 +111,11 @@ public class PrecompactedRow extends AbstractCompactedRow } } - public void write(PageCacheInformer out) throws IOException + public void write(DataOutput out) throws IOException { assert buffer.getLength() > 0; out.writeLong(buffer.getLength()); - - List> pageCacheMarkers = buffer.getPageCacheMarkers(); - - if(pageCacheMarkers == null) - { - out.write(buffer.getData(), 0, buffer.getLength()); - return; - } - - // Step through each page cache window and inform the - // output writer to respect these... - long startingPosition = out.getCurrentPosition(); - int bufferOffset = 0; - for (Pair window : pageCacheMarkers) - { - // write out any data before the window - if (window.left > (out.getCurrentPosition() - startingPosition)) - { - out.write(buffer.getData(), bufferOffset, window.left - bufferOffset); - bufferOffset = window.left; - } - - long startingAt = out.getCurrentPosition(); - - assert (bufferOffset + window.right) <= buffer.getLength() : ""+(bufferOffset + window.right)+" > "+buffer.getLength(); - - out.write(buffer.getData(), bufferOffset, window.right); - out.keepCacheWindow(startingAt); - - bufferOffset += window.right; - } - - // Write everything else - if (bufferOffset < buffer.getLength()) - { - out.write(buffer.getData(), bufferOffset, buffer.getLength() - bufferOffset); - } + out.write(buffer.getData(), 0, buffer.getLength()); } public void update(MessageDigest digest) diff --git a/src/java/org/apache/cassandra/io/sstable/CacheWriter.java b/src/java/org/apache/cassandra/io/sstable/CacheWriter.java index 8342d76bb9..b3ab1ce622 100644 --- a/src/java/org/apache/cassandra/io/sstable/CacheWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/CacheWriter.java @@ -75,7 +75,7 @@ public class CacheWriter implements ICompactionInfo logger.debug("Saving {}", path); File tmpFile = File.createTempFile(path.getName(), null, path.getParentFile()); - BufferedRandomAccessFile out = new BufferedRandomAccessFile(tmpFile, "rw", BufferedRandomAccessFile.DEFAULT_BUFFER_SIZE, true, false); + BufferedRandomAccessFile out = new BufferedRandomAccessFile(tmpFile, "rw", BufferedRandomAccessFile.DEFAULT_BUFFER_SIZE, true); try { for (K key : keys) diff --git a/src/java/org/apache/cassandra/io/sstable/KeyIterator.java b/src/java/org/apache/cassandra/io/sstable/KeyIterator.java index 7bd9eb8b5f..4f3ca43ab0 100644 --- a/src/java/org/apache/cassandra/io/sstable/KeyIterator.java +++ b/src/java/org/apache/cassandra/io/sstable/KeyIterator.java @@ -47,8 +47,7 @@ public class KeyIterator extends AbstractIterator implements Itera in = new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_INDEX)), "r", BufferedRandomAccessFile.DEFAULT_BUFFER_SIZE, - true, - false); + true); } catch (IOException e) { diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java index fb994cc5ea..fe73e51a63 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java @@ -35,8 +35,7 @@ import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.IColumn; import org.apache.cassandra.db.columniterator.IColumnIterator; import org.apache.cassandra.io.util.BufferedRandomAccessFile; -import org.apache.cassandra.io.util.PageCacheInformer; -import org.apache.cassandra.utils.PageCacheMetrics; +import org.apache.cassandra.utils.Filter; public class SSTableIdentityIterator implements Comparable, IColumnIterator { @@ -48,9 +47,6 @@ public class SSTableIdentityIterator implements Comparable= pageSize - int chunkSize = (int) (finishedAt - dataStart) / 128; - - chunkSize = chunkSize >= pageCacheMetrics.pageSize ? chunkSize : pageCacheMetrics.pageSize; - - long chunkStart = 0; - long chunkEnd = 0; - boolean isChunkInPageCache = false; - - while (file.getFilePointer() < finishedAt) - { - - // Mark chunks that have cached pages - // So we can migrate them - if (file.getFilePointer() >= chunkEnd) - { - if (isChunkInPageCache) - { - out.keepCacheWindow(out.getCurrentPosition() - chunkSize); - } - - chunkStart = file.getFilePointer(); - chunkEnd = chunkStart + chunkSize; - - if(chunkEnd > finishedAt) - chunkEnd = finishedAt; - - isChunkInPageCache = pageCacheMetrics.isRangeInCache(chunkStart, chunkEnd); - } - - - out.write(file.readByte()); - } - - if (isChunkInPageCache) - out.keepCacheWindow(out.getCurrentPosition() - (file.getFilePointer() - chunkStart)); + out.write(file.readByte()); } } @@ -237,9 +173,7 @@ public class SSTableIdentityIterator implements Comparable BufferedRandomAccessFile input = new BufferedRandomAccessFile(new File(descriptor.filenameFor(Component.PRIMARY_INDEX)), "r", BufferedRandomAccessFile.DEFAULT_BUFFER_SIZE, - true, - false); - + true); try { if (keyCache != null && keyCache.getCapacity() - keyCache.getSize() < keysToLoadInCache.size()) @@ -297,11 +295,9 @@ public class SSTableReader extends SSTable implements Comparable break; boolean shouldAddEntry = indexSummary.shouldAddEntry(); - ByteBuffer key = (shouldAddEntry || cacheLoading || recreatebloom) - ? ByteBufferUtil.readWithShortLength(input) - : ByteBufferUtil.skipShortLength(input); - + ? ByteBufferUtil.readWithShortLength(input) + : ByteBufferUtil.skipShortLength(input); long dataPosition = input.readLong(); if (key != null) { @@ -553,7 +549,7 @@ public class SSTableReader extends SSTable implements Comparable } /** - * SSTableScanner that avoids polluting the page cache + * Direct I/O SSTableScanner * @param bufferSize Buffer size in bytes for this Scanner. * @return A Scanner for seeking over the rows of the SSTable. */ diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java index 3061728812..fa7075ba83 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java @@ -19,7 +19,10 @@ package org.apache.cassandra.io.sstable; -import java.io.*; +import java.io.Closeable; +import java.io.File; +import java.io.IOError; +import java.io.IOException; import java.util.Arrays; import java.util.Iterator; @@ -31,8 +34,6 @@ import org.apache.cassandra.db.columniterator.IColumnIterator; import org.apache.cassandra.db.filter.QueryFilter; import org.apache.cassandra.io.util.BufferedRandomAccessFile; import org.apache.cassandra.utils.ByteBufferUtil; -import org.apache.cassandra.utils.CLibrary; -import org.apache.cassandra.utils.PageCacheMetrics; public class SSTableScanner implements Iterator, Closeable @@ -41,9 +42,6 @@ public class SSTableScanner implements Iterator, Closeable private final BufferedRandomAccessFile file; private final SSTableReader sstable; - - private final PageCacheMetrics pageCacheMetrics; - private IColumnIterator row; private boolean exhausted = false; private Iterator iterator; @@ -54,25 +52,15 @@ public class SSTableScanner implements Iterator, Closeable */ SSTableScanner(SSTableReader sstable, int bufferSize, boolean skipCache) { - PageCacheMetrics pageCacheMetrics = null; - try { - File sstableFile = new File(sstable.getFilename()); - file = new BufferedRandomAccessFile(sstableFile, "r", bufferSize, skipCache, false); - - if (skipCache) - { - pageCacheMetrics = CLibrary.getCachedPages(file); - } + this.file = new BufferedRandomAccessFile(new File(sstable.getFilename()), "r", bufferSize, skipCache); } catch (IOException e) { throw new IOError(e); } - this.sstable = sstable; - this.pageCacheMetrics = pageCacheMetrics; } /** @@ -91,7 +79,6 @@ public class SSTableScanner implements Iterator, Closeable } this.sstable = sstable; this.filter = filter; - this.pageCacheMetrics = null; } public void close() throws IOException @@ -189,7 +176,8 @@ public class SSTableScanner implements Iterator, Closeable if (filter == null) { - return row = new SSTableIdentityIterator(sstable, file, key, dataStart, dataSize, pageCacheMetrics); + row = new SSTableIdentityIterator(sstable, file, key, dataStart, dataSize); + return row; } else { diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java index 067be8f9c3..6f8959bb86 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java @@ -56,16 +56,16 @@ public class SSTableWriter extends SSTable private IndexWriter iwriter; private SegmentedFile.Builder dbuilder; - public final BufferedRandomAccessFile dataFile; + private final BufferedRandomAccessFile dataFile; private DecoratedKey lastWrittenKey; private FileMark dataMark; public SSTableWriter(String filename, long keyCount) throws IOException { - this(filename, keyCount, DatabaseDescriptor.getCFMetaData(Descriptor.fromFilename(filename)), StorageService.getPartitioner(), false); + this(filename, keyCount, DatabaseDescriptor.getCFMetaData(Descriptor.fromFilename(filename)), StorageService.getPartitioner()); } - public SSTableWriter(String filename, long keyCount, CFMetaData metadata, IPartitioner partitioner, boolean migratePageCache) throws IOException + public SSTableWriter(String filename, long keyCount, CFMetaData metadata, IPartitioner partitioner) throws IOException { super(Descriptor.fromFilename(filename), new HashSet(Arrays.asList(Component.DATA, Component.FILTER, Component.PRIMARY_INDEX, Component.STATS)), @@ -73,10 +73,9 @@ public class SSTableWriter extends SSTable partitioner, SSTable.defaultRowHistogram(), SSTable.defaultColumnHistogram()); - - iwriter = new IndexWriter(descriptor, partitioner, keyCount, !migratePageCache); //when we migrate pages we cache the index + iwriter = new IndexWriter(descriptor, partitioner, keyCount); dbuilder = SegmentedFile.getBuilder(DatabaseDescriptor.getDiskAccessMode()); - dataFile = new BufferedRandomAccessFile(new File(getFilename()), "rw", DatabaseDescriptor.getInMemoryCompactionLimit(), true, migratePageCache); + dataFile = new BufferedRandomAccessFile(new File(getFilename()), "rw", DatabaseDescriptor.getInMemoryCompactionLimit(), true); } public void mark() @@ -255,7 +254,7 @@ public class SSTableWriter extends SSTable cfs = Table.open(desc.ksname).getColumnFamilyStore(desc.cfname); try { - dfile = new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_DATA)), "r", 8 * 1024 * 1024, true, false); + dfile = new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_DATA)), "r", 8 * 1024 * 1024, true); } catch (IOException e) { @@ -280,7 +279,7 @@ public class SSTableWriter extends SSTable try { estimatedRows = SSTable.estimateRowsFromData(desc, dfile); - iwriter = new IndexWriter(desc, StorageService.getPartitioner(), estimatedRows, true); + iwriter = new IndexWriter(desc, StorageService.getPartitioner(), estimatedRows); } catch(IOException e) { @@ -367,11 +366,11 @@ public class SSTableWriter extends SSTable public final BloomFilter bf; private FileMark mark; - IndexWriter(Descriptor desc, IPartitioner part, long keyCount, boolean skipCache) throws IOException + IndexWriter(Descriptor desc, IPartitioner part, long keyCount) throws IOException { this.desc = desc; this.partitioner = part; - indexFile = new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_INDEX)), "rw", 8 * 1024 * 1024, skipCache, false); + indexFile = new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_INDEX)), "rw", 8 * 1024 * 1024, true); builder = SegmentedFile.getBuilder(DatabaseDescriptor.getIndexAccessMode()); summary = new IndexSummary(keyCount); bf = BloomFilter.getFilter(keyCount, 15); diff --git a/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java b/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java index 4157c59733..372d76ea6b 100644 --- a/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java +++ b/src/java/org/apache/cassandra/io/util/BufferedRandomAccessFile.java @@ -26,7 +26,6 @@ import java.nio.channels.ClosedChannelException; import java.nio.channels.FileChannel; import org.apache.cassandra.utils.CLibrary; -import org.apache.log4j.Logger; /** * A BufferedRandomAccessFile is like a @@ -39,9 +38,9 @@ import org.apache.log4j.Logger; * overridden here relies on the implementation of those methods in the * superclass. */ -public class BufferedRandomAccessFile extends RandomAccessFile implements FileDataInput, PageCacheInformer +public class BufferedRandomAccessFile extends RandomAccessFile implements FileDataInput { - private static final Logger logger = Logger.getLogger(BufferedRandomAccessFile.class); + private static final long MAX_BYTES_IN_PAGE_CACHE = (long) Math.pow(2, 27); // 128mb // absolute filesystem path to the file private final String filePath; @@ -73,23 +72,12 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa // file descriptor private int fd; - // keep *at most* this much data in the page cache from this file at a time - private static final long MAX_BYTES_IN_PAGE_CACHE = (long) Math.pow(2, 27); // 128mb - // skip cache - used for commit log and sstable writing w/ posix_fadvise private final boolean skipCache; - // used for page cache migration in compaction - private final boolean pageCacheMigrate; - - // tracks the number of bytes read or written since last page cache flush private long bytesSinceCacheFlush = 0; - - // tracks the lowest seen file position since the last page cache flush - // this is needed because the posix_fadvise call takes a offset and length private long minBufferOffset = Long.MAX_VALUE; - /* * Open a new BufferedRandomAccessFile on the file named * name in mode mode, which should be "r" for @@ -117,15 +105,14 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa public BufferedRandomAccessFile(File file, String mode, int bufferSize) throws IOException { - this(file, mode, bufferSize, false, false); + this(file, mode, bufferSize, false); } - public BufferedRandomAccessFile(File file, String mode, int bufferSize, boolean skipCache, boolean pageCacheMigrate) throws IOException + public BufferedRandomAccessFile(File file, String mode, int bufferSize, boolean skipCache) throws IOException { super(file, mode); this.skipCache = skipCache; - this.pageCacheMigrate = pageCacheMigrate; channel = super.getChannel(); filePath = file.getAbsolutePath(); @@ -150,15 +137,10 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa channel.force(true); // true, because file length counts as // "meta-data" - if (skipCache && bytesSinceCacheFlush > 0) + if (skipCache) { - // clear remaining file from page cache - long startAt = 0; - - if (pageCacheMigrate) - startAt = minBufferOffset; - - CLibrary.trySkipCache(this.fd, startAt, 0); + // clear entire file from page cache + CLibrary.trySkipCache(this.fd, 0, 0); minBufferOffset = Long.MAX_VALUE; bytesSinceCacheFlush = 0; @@ -168,34 +150,6 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa } } - /** {@InheritDoc} */ - public long getCurrentPosition() - { - return getFilePointer(); - } - - /** @{InheritDoc} */ - public void keepCacheWindow(long startingOffset) - { - - if (!pageCacheMigrate) - return; - - if (minBufferOffset < startingOffset) - { - //Flush anything before start offset - CLibrary.trySkipCache(this.fd, minBufferOffset, startingOffset - 1); - - //jump ahead to position() so it doesn't get flushed in the range - minBufferOffset = bufferOffset; - bytesSinceCacheFlush = 0; - } - - if(logger.isDebugEnabled()) - logger.debug("Kept " + (getFilePointer() - startingOffset) + " bytes in page cache"); - } - - public void flush() throws IOException { if (isDirty) @@ -208,8 +162,10 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa if (skipCache) { - // we don't know when the data reaches disk since we aren't calling flush - // so we continue to clear pages we don't need from the first offset we see + // we don't know when the data reaches disk since we aren't + // calling flush + // so we continue to clear pages we don't need from the first + // offset we see // periodically we update this starting offset bytesSinceCacheFlush += validBufferBytes; @@ -218,7 +174,7 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa if (bytesSinceCacheFlush >= MAX_BYTES_IN_PAGE_CACHE) { - CLibrary.trySkipCache(this.fd, minBufferOffset, 0); + CLibrary.trySkipCache(this.fd, (int) minBufferOffset, 0); minBufferOffset = bufferOffset; bytesSinceCacheFlush = 0; } @@ -257,7 +213,7 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa bytesSinceCacheFlush += read; if (skipCache && bytesSinceCacheFlush >= MAX_BYTES_IN_PAGE_CACHE) { - CLibrary.trySkipCache(this.fd, minBufferOffset, 0); + CLibrary.trySkipCache(this.fd, (int) minBufferOffset, 0); bytesSinceCacheFlush = 0; minBufferOffset = Long.MAX_VALUE; } @@ -441,12 +397,7 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa if (skipCache && bytesSinceCacheFlush > 0) { - long startAt = 0; - - if(pageCacheMigrate) - startAt = minBufferOffset; - - CLibrary.trySkipCache(this.fd, startAt, 0); + CLibrary.trySkipCache(this.fd, 0, 0); } super.close(); @@ -492,7 +443,7 @@ public class BufferedRandomAccessFile extends RandomAccessFile implements FileDa public static BufferedRandomAccessFile getUncachingReader(String filename) throws IOException { - return new BufferedRandomAccessFile(new File(filename), "r", 8 * 1024 * 1024, true, false); + return new BufferedRandomAccessFile(new File(filename), "r", 8 * 1024 * 1024, true); } /** diff --git a/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java b/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java index e1ec7a4284..ac8a0ed05e 100644 --- a/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java +++ b/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java @@ -19,20 +19,14 @@ package org.apache.cassandra.io.util; import java.io.DataOutputStream; -import java.util.ArrayList; -import java.util.List; - -import org.apache.cassandra.utils.Pair; /** * An implementation of the DataOutputStream interface. This class is completely thread * unsafe. */ -public final class DataOutputBuffer extends DataOutputStream implements PageCacheInformer +public final class DataOutputBuffer extends DataOutputStream { - private List> pageCacheMarkers; - public DataOutputBuffer() { this(128); @@ -76,32 +70,6 @@ public final class DataOutputBuffer extends DataOutputStream implements PageCach { this.written = 0; buffer().reset(); - pageCacheMarkers = null; - return this; } - - /** {@InheritDoc} */ - public void keepCacheWindow(long startAt) - { - if (pageCacheMarkers == null) - pageCacheMarkers = new ArrayList>(); - - long endAt = getCurrentPosition(); - - assert startAt <= endAt; - - pageCacheMarkers.add(new Pair((int) startAt, (int) (endAt - startAt))); - } - - /** {@InheritDoc} */ - public long getCurrentPosition() - { - return getLength(); - } - - public final List> getPageCacheMarkers() - { - return pageCacheMarkers; - } } diff --git a/src/java/org/apache/cassandra/io/util/PageCacheInformer.java b/src/java/org/apache/cassandra/io/util/PageCacheInformer.java deleted file mode 100644 index 46e57c11e6..0000000000 --- a/src/java/org/apache/cassandra/io/util/PageCacheInformer.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.util; - -import java.io.DataOutput; - -public interface PageCacheInformer extends DataOutput -{ - /** - * Instruct the OS to keep a portion of the page cache from startAt to current position. - * this call is only appropriate for writes - * @param startAt the start of the window - */ - void keepCacheWindow(long startAt); - - /** - * @return the current position - */ - long getCurrentPosition(); -} diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 14c2e25d65..aba9850d87 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -31,6 +31,7 @@ import javax.management.ObjectName; import com.google.common.base.Charsets; import com.google.common.collect.HashMultimap; +import com.google.common.collect.Iterables; import com.google.common.collect.Multimap; import org.apache.cassandra.db.commitlog.CommitLog; @@ -2195,14 +2196,28 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe ColumnFamilyStore largestByThroughput = null; for (ColumnFamilyStore cfs : ColumnFamilyStore.all()) { - if (largestByOps == null || cfs.getMemtableColumnsCount() > largestByOps.getMemtableColumnsCount()) + long ops = 0; + long throughput = 0; + for (ColumnFamilyStore subordinate : Iterables.concat(Collections.singleton(cfs), cfs.getIndexColumnFamilyStores())) + { + ops += subordinate.getMemtableColumnsCount(); + throughput = subordinate.getMemtableThroughputInMB(); + } + + if (ops > 0 && (largestByOps == null || ops > largestByOps.getMemtableColumnsCount())) + { + logger_.debug(ops + " total ops in " + cfs); largestByOps = cfs; - if (largestByThroughput == null || cfs.getMemtableThroughputInMB() > largestByThroughput.getMemtableThroughputInMB()) + } + if (throughput > 0 && (largestByThroughput == null || throughput > largestByThroughput.getMemtableThroughputInMB())) + { + logger_.debug(throughput + " total throughput in " + cfs); largestByThroughput = cfs; + } } if (largestByOps == null) { - logger_.error("Unable to reduce heap usage since there are no column families defined"); + logger_.info("Unable to reduce heap usage since there are no dirty column families"); return; } diff --git a/src/java/org/apache/cassandra/utils/CLibrary.java b/src/java/org/apache/cassandra/utils/CLibrary.java index aefee7651c..3b0c7af86e 100644 --- a/src/java/org/apache/cassandra/utils/CLibrary.java +++ b/src/java/org/apache/cassandra/utils/CLibrary.java @@ -18,14 +18,10 @@ */ package org.apache.cassandra.utils; -import java.io.*; +import java.io.File; +import java.io.FileDescriptor; +import java.io.IOException; import java.lang.reflect.Field; -import java.nio.ByteBuffer; -import java.nio.channels.FileChannel.MapMode; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicLong; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -33,10 +29,6 @@ import org.slf4j.LoggerFactory; import com.sun.jna.LastErrorException; import com.sun.jna.Native; -import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.io.util.BufferedRandomAccessFile; -import org.apache.cassandra.io.util.FileUtils; - public final class CLibrary { private static Logger logger = LoggerFactory.getLogger(CLibrary.class); @@ -58,28 +50,11 @@ public final class CLibrary private static final int POSIX_FADV_DONTNEED = 4; /* fadvise.h */ private static final int POSIX_FADV_NOREUSE = 5; /* fadvise.h */ - private static native int mlockall(int flags) throws LastErrorException; - private static native int munlockall() throws LastErrorException; - - private static native int link(String from, String to) throws LastErrorException; - - // fcntl - manipulate file descriptor, `man 2 fcntl` - public static native int fcntl(int fd, int command, long flags) throws LastErrorException; - - // fadvice - public static native int posix_fadvise(int fd, long offset, long len, int flag) throws LastErrorException; - - public static native int mincore(ByteBuffer buf, int length, char[] vec) throws LastErrorException; - - private static native int getpagesize() throws LastErrorException; - public static Integer pageSize = null; - static { try { Native.register("c"); - pageSize = CLibrary.getPageSize(); } catch (NoClassDefFoundError e) { @@ -93,38 +68,19 @@ public final class CLibrary { logger.warn("Obsolete version of JNA present; unable to register C library. Upgrade to JNA 3.2.7 or later"); } - } - private static Integer getPageSize() - { - int ps = -1; + private static native int mlockall(int flags) throws LastErrorException; + private static native int munlockall() throws LastErrorException; - if (!DatabaseDescriptor.isPageCaheMigrationEnabled()) - return null; + private static native int link(String from, String to) throws LastErrorException; - try - { - ps = CLibrary.getpagesize(); - assert ps >= 0; // on error a value of -1 is returned and errno is set to indicate the error. - - logger.info("PageSize = " + ps); - } - catch (RuntimeException e) - { - if (!(e instanceof LastErrorException)) - throw e; - - logger.warn(String.format("getpagesize failed, errno (%d).", CLibrary.errno(e))); - } - catch (UnsatisfiedLinkError e) - { - // this will have already been logged by CLibrary, no need to repeat it - } - - return ps > 0 ? ps : null; - } + // fcntl - manipulate file descriptor, `man 2 fcntl` + public static native int fcntl(int fd, int command, long flags) throws LastErrorException; + // fadvice + public static native int posix_fadvise(int fd, int offset, int len, int flag) throws LastErrorException; + private static int errno(RuntimeException e) { assert e instanceof LastErrorException; @@ -233,9 +189,9 @@ public final class CLibrary } } - public static void trySkipCache(int fd, long offset, long len) + public static void trySkipCache(int fd, int offset, int len) { - if (fd < 0 || !DatabaseDescriptor.isPageCaheMigrationEnabled()) + if (fd < 0) return; try @@ -256,63 +212,6 @@ public final class CLibrary } } - public static PageCacheMetrics getCachedPages(BufferedRandomAccessFile braf) throws IOException - { - if (pageSize == null || pageSize == 0 || !DatabaseDescriptor.isPageCaheMigrationEnabled()) - return null; - - long length = braf.length(); - - PageCacheMetrics pageCacheMetrics = new PageCacheMetrics(pageSize, length); - - try - { - char[] pages = null; - - // MMap 2G chunks - for (long offset = 0; offset < length; offset += Integer.MAX_VALUE) - { - long limit = (offset + Integer.MAX_VALUE) > length ? (length - offset) : Integer.MAX_VALUE; - - ByteBuffer buf = braf.getChannel().map(MapMode.READ_ONLY, offset, limit); - - int numPages = (int) ((limit + pageSize - 1) / pageSize); - - if (pages == null || pages.length < numPages) - pages = new char[numPages]; - - int rc = mincore(buf, (int) limit, pages); - - if (rc != 0) - { - logger.warn(String.format("mincore failed, rc (%d).", rc)); - break; - } - - for (long i = 0, position = offset; position < limit; position += pageSize, i++) - { - if ((pages[(int) i] & 1) == 1) - { - pageCacheMetrics.setPage(position); - } - } - } - } - catch (RuntimeException e) - { - if (!(e instanceof LastErrorException)) - throw e; - - logger.warn(String.format("mincore failed, errno (%d).", CLibrary.errno(e))); - } - catch (UnsatisfiedLinkError e) - { - // this will have already been logged by CLibrary, no need to repeat it - } - - return pageCacheMetrics; - } - public static int tryFcntl(int fd, int command, int flags) { int result = -1; diff --git a/src/java/org/apache/cassandra/utils/PageCacheMetrics.java b/src/java/org/apache/cassandra/utils/PageCacheMetrics.java deleted file mode 100644 index a1a73d8f36..0000000000 --- a/src/java/org/apache/cassandra/utils/PageCacheMetrics.java +++ /dev/null @@ -1,81 +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.utils; - -import org.apache.cassandra.utils.obs.OpenBitSet; - -public class PageCacheMetrics -{ - private final OpenBitSet pageMap; - private final long fileLength; - public final int pageSize; - - public PageCacheMetrics(int pageSize, long fileLength) - { - assert pageSize > 0; - assert fileLength > 0; - - this.pageSize = pageSize; - this.fileLength = fileLength; - - long numPages = (fileLength + pageSize - 1) / pageSize; - pageMap = new OpenBitSet(numPages); - - assert isEmpty(); - } - - public boolean isEmpty() - { - return pageMap.cardinality() == 0; - } - - public void setPage(long position) - { - pageMap.fastSet(getPageNumber(position)); - } - - private long getPageNumber(long position) - { - assert position <= fileLength; - - return (position + pageSize - 1) / pageSize; - } - - /** - * Returns true if any page within this range is cached - * @param start starting position - * @param end the limit of the range - * @return true if given range is in the cache, false otherwise - */ - public boolean isRangeInCache(long start, long end) - { - long startPage = getPageNumber(start); - long endPage = getPageNumber(end); - - assert startPage <= endPage; - - for (long i = startPage; i < endPage; i++) - { - if (pageMap.fastGet(i)) - return true; - } - - return false; - } -} diff --git a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java index bdafef1459..7402233469 100644 --- a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java +++ b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java @@ -138,11 +138,11 @@ public class AntiEntropyServiceTest extends CleanupHelper // add a row with the minimum token validator.add(new PrecompactedRow(new DecoratedKey(min, ByteBufferUtil.bytes("nonsense!")), - new DataOutputBuffer())); + new DataOutputBuffer())); // and a row after it validator.add(new PrecompactedRow(new DecoratedKey(mid, ByteBufferUtil.bytes("inconceivable!")), - new DataOutputBuffer())); + new DataOutputBuffer())); validator.complete(); // confirm that the tree was validated