From 29898b89ccc4ec4d9a417d9ebbc2039adb7bdd8e Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Tue, 8 Mar 2011 21:34:18 +0000 Subject: [PATCH 01/12] update NEWS for 0.7.4 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1079555 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 2 +- NEWS.txt | 14 ++++++++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/CHANGES.txt b/CHANGES.txt index a7b7bfad20..0f301224a9 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,7 +1,7 @@ 0.7.4 * add nodetool join command (CASSANDRA-2160) * fix secondary indexes on pre-existing or streamed data (CASSANDRA-2244) - * initialize endpoing in gossiper earlier (CASSANDRA-2228) + * initialize endpoint in gossiper earlier (CASSANDRA-2228) * add ability to write to Cassandra from Pig (CASSANDRA-1828) * add rpc_[min|max]_threads (CASSANDRA-2176) * add CL.TWO, CL.THREE (CASSANDRA-2013) diff --git a/NEWS.txt b/NEWS.txt index b7451c461d..f608075a44 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -1,3 +1,17 @@ +0.7.4 +===== + +Upgrading +--------- + - Nothing specific to 0.7.4, but see 0.7.3 Upgrading if upgrading + from earlier than 0.7.1. + +Features +-------- + - Output to Pig is now supported as well as input + + + 0.7.3 ===== From 703fce390f5deff5c323ecc2e8126bac80eeed74 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 9 Mar 2011 01:05:48 +0000 Subject: [PATCH 02/12] correct docstring git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1079632 13f79535-47bb-0310-9956-ffa450edef68 --- .../java/org/apache/cassandra/hadoop/pig/CassandraStorage.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java index 7c79af3755..f5a837e78f 100644 --- a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java +++ b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java @@ -71,7 +71,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo } /** - * @param limit: number of rows to fetch at a time + * @param limit: number of columns to fetch in a slice */ public CassandraStorage(int limit) { From 929f430da9814056f6f3062b2e45593b5b92001d Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 9 Mar 2011 02:42:20 +0000 Subject: [PATCH 03/12] add more information to scrub errors git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1079645 13f79535-47bb-0310-9956-ffa450edef68 --- conf/log4j-server.properties | 2 +- src/java/org/apache/cassandra/db/CompactionManager.java | 2 +- src/java/org/apache/cassandra/io/sstable/IndexHelper.java | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/conf/log4j-server.properties b/conf/log4j-server.properties index cde670729f..f72e4c360a 100644 --- a/conf/log4j-server.properties +++ b/conf/log4j-server.properties @@ -18,7 +18,7 @@ # (%l is slower.) # output messages into a rolling log file as well as stdout -log4j.rootLogger=INFO,stdout,R +log4j.rootLogger=DEBUG,stdout,R # stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java index d042a44962..d8d38bdef3 100644 --- a/src/java/org/apache/cassandra/db/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/CompactionManager.java @@ -629,7 +629,7 @@ public class CompactionManager implements CompactionManagerMBean } else { - logger.warn("Row is unreadable; skipping to next"); + logger.warn("Row at " + dataStart + " is unreadable; skipping to next"); if (currentIndexKey != null) dataFile.seek(nextRowPositionFromIndex); badRows++; diff --git a/src/java/org/apache/cassandra/io/sstable/IndexHelper.java b/src/java/org/apache/cassandra/io/sstable/IndexHelper.java index d5f90d44e3..74d1f7c119 100644 --- a/src/java/org/apache/cassandra/io/sstable/IndexHelper.java +++ b/src/java/org/apache/cassandra/io/sstable/IndexHelper.java @@ -110,7 +110,7 @@ public class IndexHelper { int size = file.readInt(); if (size > maxSize || size <= 0) - throw new EOFException("bloom filter claims to be longer than entire row size"); + throw new EOFException("bloom filter claims to be " + size + " bytes, longer than entire row size " + maxSize); ByteBuffer bytes = file.readBytes(size); DataInputStream stream = new DataInputStream(ByteBufferUtil.inputStream(bytes)); From cb7a1e475d9f86d866bd2913e1d6b0b40df8b621 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 9 Mar 2011 02:42:49 +0000 Subject: [PATCH 04/12] revert log4j change, again git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1079647 13f79535-47bb-0310-9956-ffa450edef68 --- conf/log4j-server.properties | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/conf/log4j-server.properties b/conf/log4j-server.properties index f72e4c360a..cde670729f 100644 --- a/conf/log4j-server.properties +++ b/conf/log4j-server.properties @@ -18,7 +18,7 @@ # (%l is slower.) # output messages into a rolling log file as well as stdout -log4j.rootLogger=DEBUG,stdout,R +log4j.rootLogger=INFO,stdout,R # stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender From 135b0e0c08d1263e96ededd30713ff84b384b521 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 9 Mar 2011 02:55:38 +0000 Subject: [PATCH 05/12] teach tryComponentFromFilename about backups dir git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1079649 13f79535-47bb-0310-9956-ffa450edef68 --- src/java/org/apache/cassandra/io/sstable/SSTable.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/java/org/apache/cassandra/io/sstable/SSTable.java b/src/java/org/apache/cassandra/io/sstable/SSTable.java index 05d6212f0a..6b4d517622 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTable.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTable.java @@ -174,7 +174,7 @@ public abstract class SSTable } catch (Exception e) { - if (!"snapshots".equals(name)) + if (!"snapshots".equals(name) && !"backups".equals(name)) logger.warn("Invalid file '{}' in data directory {}.", name, dir); return null; } From 5947c5907258c81fce0ff8e8333fa9852ed31eaf Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 9 Mar 2011 02:58:26 +0000 Subject: [PATCH 06/12] add asserts to make sure we don't write zero-length rows; see CASSANDRA-2296 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1079650 13f79535-47bb-0310-9956-ffa450edef68 --- src/java/org/apache/cassandra/db/CompactionManager.java | 1 + src/java/org/apache/cassandra/io/LazilyCompactedRow.java | 5 ++++- src/java/org/apache/cassandra/io/PrecompactedRow.java | 1 + src/java/org/apache/cassandra/io/sstable/SSTableWriter.java | 4 +++- 4 files changed, 9 insertions(+), 2 deletions(-) diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java index d8d38bdef3..723663dad1 100644 --- a/src/java/org/apache/cassandra/db/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/CompactionManager.java @@ -1122,6 +1122,7 @@ public class CompactionManager implements CompactionManagerMBean public void write(PageCacheInformer out) throws IOException { + assert row.dataSize != 0; out.writeLong(row.dataSize); row.echoData(out); } diff --git a/src/java/org/apache/cassandra/io/LazilyCompactedRow.java b/src/java/org/apache/cassandra/io/LazilyCompactedRow.java index 92a95499d8..3e48d38a70 100644 --- a/src/java/org/apache/cassandra/io/LazilyCompactedRow.java +++ b/src/java/org/apache/cassandra/io/LazilyCompactedRow.java @@ -98,6 +98,7 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl if (rows.size() == 1 && !shouldPurge && rows.get(0).sstable.descriptor.isLatestVersion && !forceDeserialize) { SSTableIdentityIterator row = rows.get(0); + assert row.dataSize != 0; out.writeLong(row.dataSize); row.echoData(out); return; @@ -106,7 +107,9 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl DataOutputBuffer clockOut = new DataOutputBuffer(); ColumnFamily.serializer().serializeCFInfo(emptyColumnFamily, clockOut); - out.writeLong(headerBuffer.getLength() + clockOut.getLength() + columnSerializedSize); + long dataSize = headerBuffer.getLength() + clockOut.getLength() + columnSerializedSize; + assert dataSize != 0; + out.writeLong(dataSize); out.write(headerBuffer.getData(), 0, headerBuffer.getLength()); out.write(clockOut.getData(), 0, clockOut.getLength()); out.writeInt(columnCount); diff --git a/src/java/org/apache/cassandra/io/PrecompactedRow.java b/src/java/org/apache/cassandra/io/PrecompactedRow.java index 16b5ff7d9c..2899249e09 100644 --- a/src/java/org/apache/cassandra/io/PrecompactedRow.java +++ b/src/java/org/apache/cassandra/io/PrecompactedRow.java @@ -118,6 +118,7 @@ public class PrecompactedRow extends AbstractCompactedRow public void write(PageCacheInformer out) throws IOException { + assert buffer.getLength() != 0; out.writeLong(buffer.getLength()); List> pageCacheMarkers = buffer.getPageCacheMarkers(); diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java index a46be8b7b4..9205b46303 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java @@ -147,7 +147,9 @@ public class SSTableWriter extends SSTable // seek back and write the row size (not including the size Long itself) long endPosition = dataFile.getFilePointer(); dataFile.seek(sizePosition); - dataFile.writeLong(endPosition - (sizePosition + 8)); + long dataSize = endPosition - (sizePosition + 8); + assert dataSize != 0; + dataFile.writeLong(dataSize); // finally, reset for next row dataFile.seek(endPosition); afterAppend(decoratedKey, startPosition); From 55a1e170d2aad5ae29b4e0e64ca764d3719fe3e8 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 9 Mar 2011 03:27:48 +0000 Subject: [PATCH 07/12] improve asserts git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1079652 13f79535-47bb-0310-9956-ffa450edef68 --- src/java/org/apache/cassandra/db/CompactionManager.java | 2 +- src/java/org/apache/cassandra/io/LazilyCompactedRow.java | 4 ++-- src/java/org/apache/cassandra/io/PrecompactedRow.java | 2 +- src/java/org/apache/cassandra/io/sstable/SSTableWriter.java | 2 +- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java index 723663dad1..b84be781fe 100644 --- a/src/java/org/apache/cassandra/db/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/CompactionManager.java @@ -1122,7 +1122,7 @@ public class CompactionManager implements CompactionManagerMBean public void write(PageCacheInformer out) throws IOException { - assert row.dataSize != 0; + assert row.dataSize > 0; out.writeLong(row.dataSize); row.echoData(out); } diff --git a/src/java/org/apache/cassandra/io/LazilyCompactedRow.java b/src/java/org/apache/cassandra/io/LazilyCompactedRow.java index 3e48d38a70..8c26b772c1 100644 --- a/src/java/org/apache/cassandra/io/LazilyCompactedRow.java +++ b/src/java/org/apache/cassandra/io/LazilyCompactedRow.java @@ -98,7 +98,7 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl if (rows.size() == 1 && !shouldPurge && rows.get(0).sstable.descriptor.isLatestVersion && !forceDeserialize) { SSTableIdentityIterator row = rows.get(0); - assert row.dataSize != 0; + assert row.dataSize > 0; out.writeLong(row.dataSize); row.echoData(out); return; @@ -108,7 +108,7 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements IIterabl ColumnFamily.serializer().serializeCFInfo(emptyColumnFamily, clockOut); long dataSize = headerBuffer.getLength() + clockOut.getLength() + columnSerializedSize; - assert dataSize != 0; + assert dataSize > 0; out.writeLong(dataSize); out.write(headerBuffer.getData(), 0, headerBuffer.getLength()); out.write(clockOut.getData(), 0, clockOut.getLength()); diff --git a/src/java/org/apache/cassandra/io/PrecompactedRow.java b/src/java/org/apache/cassandra/io/PrecompactedRow.java index 2899249e09..02112a74ea 100644 --- a/src/java/org/apache/cassandra/io/PrecompactedRow.java +++ b/src/java/org/apache/cassandra/io/PrecompactedRow.java @@ -118,7 +118,7 @@ public class PrecompactedRow extends AbstractCompactedRow public void write(PageCacheInformer out) throws IOException { - assert buffer.getLength() != 0; + assert buffer.getLength() > 0; out.writeLong(buffer.getLength()); List> pageCacheMarkers = buffer.getPageCacheMarkers(); diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java index 9205b46303..067be8f9c3 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java @@ -148,7 +148,7 @@ public class SSTableWriter extends SSTable long endPosition = dataFile.getFilePointer(); dataFile.seek(sizePosition); long dataSize = endPosition - (sizePosition + 8); - assert dataSize != 0; + assert dataSize > 0; dataFile.writeLong(dataSize); // finally, reset for next row dataFile.seek(endPosition); From fc4502acf0d3a2d9bdf8f26cbe63b9309705bca1 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 9 Mar 2011 14:27:46 +0000 Subject: [PATCH 08/12] revert r1078803 (#1902) git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1079812 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 1 - conf/cassandra.yaml | 4 - .../org/apache/cassandra/config/Config.java | 4 - .../cassandra/config/DatabaseDescriptor.java | 5 - src/java/org/apache/cassandra/db/Column.java | 12 -- .../org/apache/cassandra/db/ColumnFamily.java | 7 - .../cassandra/db/ColumnFamilySerializer.java | 60 ++------- .../cassandra/db/ColumnFamilyStore.java | 50 +++++-- .../cassandra/db/CompactionManager.java | 29 ++-- src/java/org/apache/cassandra/db/IColumn.java | 8 -- .../org/apache/cassandra/db/RowMutation.java | 5 +- .../org/apache/cassandra/db/SuperColumn.java | 11 -- .../cassandra/db/commitlog/CommitLog.java | 2 +- .../db/commitlog/CommitLogSegment.java | 2 +- .../cassandra/io/AbstractCompactedRow.java | 19 +-- .../cassandra/io/CompactionIterator.java | 11 -- .../cassandra/io/LazilyCompactedRow.java | 5 +- .../apache/cassandra/io/PrecompactedRow.java | 47 +------ .../cassandra/io/sstable/CacheWriter.java | 2 +- .../cassandra/io/sstable/KeyIterator.java | 3 +- .../io/sstable/SSTableIdentityIterator.java | 95 ++----------- .../cassandra/io/sstable/SSTableReader.java | 12 +- .../cassandra/io/sstable/SSTableScanner.java | 26 +--- .../cassandra/io/sstable/SSTableWriter.java | 19 ++- .../io/util/BufferedRandomAccessFile.java | 79 +++-------- .../cassandra/io/util/DataOutputBuffer.java | 34 +---- .../cassandra/io/util/PageCacheInformer.java | 35 ----- .../cassandra/service/StorageService.java | 21 ++- .../org/apache/cassandra/utils/CLibrary.java | 127 ++---------------- .../cassandra/utils/PageCacheMetrics.java | 81 ----------- .../service/AntiEntropyServiceTest.java | 4 +- 31 files changed, 150 insertions(+), 670 deletions(-) delete mode 100644 src/java/org/apache/cassandra/io/util/PageCacheInformer.java delete mode 100644 src/java/org/apache/cassandra/utils/PageCacheMetrics.java 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 From 942a432cc367437c7dc36017d59bea9589fc8946 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 9 Mar 2011 14:32:25 +0000 Subject: [PATCH 09/12] revert prematurely-applied patch for #2295 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1079815 13f79535-47bb-0310-9956-ffa450edef68 --- .../cassandra/db/ColumnFamilyStore.java | 44 +++++-------------- .../cassandra/service/StorageService.java | 21 ++------- 2 files changed, 13 insertions(+), 52 deletions(-) diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 75dcf587c4..44ca6d51ef 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -32,7 +32,6 @@ 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; @@ -335,6 +334,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean { public void run() { + logger.info("Creating index {}.{}", table, indexedCfMetadata.cfName); try { forceBlockingFlush(); @@ -348,6 +348,7 @@ 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); } }; @@ -356,8 +357,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean public void buildSecondaryIndexes(Collection sstables, SortedSet columns) { - logger.info(String.format("Submitting index build of %s for data in %s", - metadata.comparator.getString(columns), StringUtils.join(sstables, ", "))); + logger.debug("Submitting index build to compactionmanager"); Table.IndexBuilder builder = table.createIndexBuilder(this, columns, new ReducingKeyIterator(sstables)); Future future = CompactionManager.instance.submitIndexBuild(this, builder); try @@ -374,7 +374,6 @@ 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. @@ -685,31 +684,26 @@ 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) - { - logger.debug("column family was dropped; no point in flushing"); - return null; - } + return null; // column family was dropped. no point in flushing. 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()); - // 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())) + icc.add(this); + for (ColumnFamilyStore indexCfs : indexedColumns.values()) { - if (!cfs.memtable.isClean()) - icc.add(cfs); + if (!indexCfs.memtable.isClean()) + icc.add(indexCfs); } final CountDownLatch latch = new CountDownLatch(icc.size()); for (ColumnFamilyStore cfs : icc) @@ -717,10 +711,6 @@ 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, @@ -764,17 +754,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean public Future forceFlush() { - // 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"); + if (memtable.isClean()) return null; - } return maybeSwitchMemtable(memtable, true); } @@ -1956,11 +1937,6 @@ 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); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index aba9850d87..14c2e25d65 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -31,7 +31,6 @@ 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; @@ -2196,28 +2195,14 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe ColumnFamilyStore largestByThroughput = null; for (ColumnFamilyStore cfs : ColumnFamilyStore.all()) { - 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); + if (largestByOps == null || cfs.getMemtableColumnsCount() > largestByOps.getMemtableColumnsCount()) largestByOps = cfs; - } - if (throughput > 0 && (largestByThroughput == null || throughput > largestByThroughput.getMemtableThroughputInMB())) - { - logger_.debug(throughput + " total throughput in " + cfs); + if (largestByThroughput == null || cfs.getMemtableThroughputInMB() > largestByThroughput.getMemtableThroughputInMB()) largestByThroughput = cfs; - } } if (largestByOps == null) { - logger_.info("Unable to reduce heap usage since there are no dirty column families"); + logger_.error("Unable to reduce heap usage since there are no column families defined"); return; } From 0087c38a268d314172f06bcd522210ed8f226409 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 9 Mar 2011 14:36:59 +0000 Subject: [PATCH 10/12] avoid writing empty rows when scrubbing tombstoned rows patch by jbellis; reviewed by slebresne for CASSANDRA-2296 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1079816 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 1 + .../cassandra/db/CompactionManager.java | 28 +++++++++++++++---- 2 files changed, 23 insertions(+), 6 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 0a80284e65..c9094d866b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -13,6 +13,7 @@ (CASSANDRA-2259) * move sample schema from cassandra.yaml to schema-sample.txt, a cli scripts (CASSANDRA-2007) + * avoid writing empty rows when scrubbing tombstoned rows (CASSANDRA-2296) 0.7.3 diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java index 755880460b..8156f9c1e8 100644 --- a/src/java/org/apache/cassandra/db/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/CompactionManager.java @@ -534,7 +534,7 @@ public class CompactionManager implements CompactionManagerMBean SSTableWriter writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, null); executor.beginCompaction(cfs.columnFamily, new ScrubInfo(dataFile, sstable)); - int goodRows = 0, badRows = 0; + int goodRows = 0, badRows = 0, emptyRows = 0; while (!dataFile.isEOF()) { @@ -588,8 +588,16 @@ public class CompactionManager implements CompactionManagerMBean if (dataSize > dataFile.length()) throw new IOError(new IOException("Impossible row size " + dataSize)); SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataStart, dataSize, true); - writer.append(getCompactedRow(row, cfs, sstable.descriptor, true)); - goodRows++; + AbstractCompactedRow compactedRow = getCompactedRow(row, cfs, sstable.descriptor, true); + if (compactedRow.isEmpty()) + { + emptyRows++; + } + else + { + writer.append(compactedRow); + goodRows++; + } if (!key.key.equals(currentIndexKey) || dataStart != dataStartFromIndex) logger.warn("Row scrubbed successfully but index file contains a different key or row size; consider rebuilding the index as described in http://www.mail-archive.com/user@cassandra.apache.org/msg03325.html"); } @@ -608,8 +616,16 @@ public class CompactionManager implements CompactionManagerMBean try { SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataStartFromIndex, dataSizeFromIndex, true); - writer.append(getCompactedRow(row, cfs, sstable.descriptor, true)); - goodRows++; + AbstractCompactedRow compactedRow = getCompactedRow(row, cfs, sstable.descriptor, true); + if (compactedRow.isEmpty()) + { + emptyRows++; + } + else + { + writer.append(compactedRow); + goodRows++; + } } catch (Throwable th2) { @@ -634,7 +650,7 @@ public class CompactionManager implements CompactionManagerMBean { SSTableReader newSstable = writer.closeAndOpenReader(sstable.maxDataAge); cfs.replaceCompactedSSTables(Arrays.asList(sstable), Arrays.asList(newSstable)); - logger.info("Scrub of " + sstable + " complete: " + goodRows + " rows in new sstable"); + logger.info("Scrub of " + sstable + " complete: " + goodRows + " rows in new sstable and " + emptyRows + " empty (tombstoned) rows dropped"); if (badRows > 0) logger.warn("Unable to recover " + badRows + " rows that were skipped. You can attempt manual recovery from the pre-scrub snapshot. You can also run nodetool repair to transfer the data from a healthy replica, if any"); } From 51d3b94da6ae672b1f2d330a206065fb9dbea5d8 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 9 Mar 2011 14:58:22 +0000 Subject: [PATCH 11/12] update CHANGES for CASSANDRA-2282 that got committed already (in r1079812 I think) git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1079821 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGES.txt b/CHANGES.txt index c9094d866b..ea087fb6ad 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -14,6 +14,8 @@ * move sample schema from cassandra.yaml to schema-sample.txt, a cli scripts (CASSANDRA-2007) * avoid writing empty rows when scrubbing tombstoned rows (CASSANDRA-2296) + * fix assertion error in range and index scans for CL < ALL + (CASSANDRA-2282) 0.7.3 From 5040f8efe1df960a56c181ad9920a23fc5f8b7cf Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 9 Mar 2011 16:57:14 +0000 Subject: [PATCH 12/12] add test for #2296 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1079882 13f79535-47bb-0310-9956-ffa450edef68 --- .../cassandra/db/CompactionManager.java | 5 ++++- .../org/apache/cassandra/db/ScrubTest.java | 20 +++++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java index 8156f9c1e8..9680c99c63 100644 --- a/src/java/org/apache/cassandra/db/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/CompactionManager.java @@ -657,7 +657,10 @@ public class CompactionManager implements CompactionManagerMBean else { cfs.markCompacted(Arrays.asList(sstable)); - logger.warn("No valid rows found while scrubbing " + sstable + "; it is marked for deletion now. If you want to attempt manual recovery, you can find a copy in the pre-scrub snapshot"); + if (badRows > 0) + logger.warn("No valid rows found while scrubbing " + sstable + "; it is marked for deletion now. If you want to attempt manual recovery, you can find a copy in the pre-scrub snapshot"); + else + logger.info("Scrub of " + sstable + " complete; looks like all " + emptyRows + " rows were tombstoned"); } } } diff --git a/test/unit/org/apache/cassandra/db/ScrubTest.java b/test/unit/org/apache/cassandra/db/ScrubTest.java index 534a3a7ba6..a0ce0f1693 100644 --- a/test/unit/org/apache/cassandra/db/ScrubTest.java +++ b/test/unit/org/apache/cassandra/db/ScrubTest.java @@ -46,6 +46,7 @@ public class ScrubTest extends CleanupHelper public String TABLE = "Keyspace1"; public String CF = "Standard1"; public String CF2 = "Super5"; + public String CF3 = "Standard2"; public String corruptSSTableName; @@ -128,6 +129,25 @@ public class ScrubTest extends CleanupHelper assertEquals(1, rows.size()); } + @Test + public void testScrubDeletedRow() throws IOException, ExecutionException, InterruptedException, ConfigurationException + { + CompactionManager.instance.disableAutoCompaction(); + Table table = Table.open(TABLE); + ColumnFamilyStore cfs = table.getColumnFamilyStore(CF3); + + RowMutation rm; + rm = new RowMutation(TABLE, ByteBufferUtil.bytes(1)); + ColumnFamily cf = ColumnFamily.create(TABLE, CF3); + cf.delete(0, 1); // expired tombstone + rm.add(cf); + rm.applyUnsafe(); + cfs.forceBlockingFlush(); + + CompactionManager.instance.performScrub(cfs); + assert cfs.getSSTables().isEmpty(); + } + @Test public void testScrubMultiRow() throws IOException, ExecutionException, InterruptedException, ConfigurationException {