You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jb...@apache.org on 2012/11/09 22:17:40 UTC
[2/6] git commit: acquire references on overlapped sstables patch by
jbellis; reviewed by vijay for CASSANDRA-4934
acquire references on overlapped sstables
patch by jbellis; reviewed by vijay for CASSANDRA-4934
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/f09a89f4
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/f09a89f4
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/f09a89f4
Branch: refs/heads/cassandra-1.2
Commit: f09a89f4cd13af2087fcc92f09f6cf1ee4785feb
Parents: e0a0eb7
Author: Jonathan Ellis <jb...@apache.org>
Authored: Fri Nov 9 10:06:18 2012 -0600
Committer: Jonathan Ellis <jb...@apache.org>
Committed: Fri Nov 9 15:17:11 2012 -0600
----------------------------------------------------------------------
CHANGES.txt | 3 +++
.../org/apache/cassandra/db/ColumnFamilyStore.java | 13 +++++++++++++
.../db/compaction/CompactionController.java | 14 +++++++++++---
.../cassandra/db/compaction/CompactionManager.java | 1 +
.../cassandra/db/compaction/CompactionTask.java | 2 ++
.../apache/cassandra/db/compaction/Scrubber.java | 4 ++++
.../cassandra/streaming/IncomingStreamReader.java | 4 ++++
7 files changed, 38 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f09a89f4/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index ba9e134..3b80885 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,9 +1,12 @@
1.2-rc1
+ * acquire references to overlapping sstables during compaction so bloom filter
+ doesn't get free'd prematurely (CASSANDRA-4934)
* Don't share slice query filter in CQL3 SelectStatement (CASSANDRA-4928)
* Separate tracing from Log4J (CASSANDRA-4861)
* Exclude gcable tombstones from merkle-tree computation (CASSANDRA-4905)
* Better printing of AbstractBounds for tracing (CASSANDRA-4931)
+
1.2-beta2
* fp rate of 1.0 disables BF entirely; LCS defaults to 1.0 (CASSANDRA-4876)
* off-heap bloom filters for row keys (CASSANDRA_4865)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f09a89f4/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 882a322..a91af8c 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -889,6 +889,19 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
return results;
}
+ /**
+ * like getOverlappingSSTables, but acquires references before returning
+ */
+ public Set<SSTableReader> getAndReferenceOverlappingSSTables(Collection<SSTableReader> sstables)
+ {
+ while (true)
+ {
+ Set<SSTableReader> overlapped = getOverlappingSSTables(sstables);
+ if (SSTableReader.acquireReferences(overlapped))
+ return overlapped;
+ }
+ }
+
/*
* Called after a BinaryMemtable flushes its in-memory data, or we add a file
* via bootstrap. This information is cached in the ColumnFamilyStore.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f09a89f4/src/java/org/apache/cassandra/db/compaction/CompactionController.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionController.java b/src/java/org/apache/cassandra/db/compaction/CompactionController.java
index 3b04833..2c8ddba 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionController.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionController.java
@@ -20,6 +20,7 @@ package org.apache.cassandra.db.compaction;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,6 +45,7 @@ public class CompactionController
public final ColumnFamilyStore cfs;
private final DataTracker.SSTableIntervalTree overlappingTree;
+ private final Set<SSTableReader> overlappingSSTables;
public final int gcBefore;
public final int mergeShardBefore;
@@ -66,7 +68,7 @@ public class CompactionController
{
this(cfs,
gcBefore,
- DataTracker.buildIntervalTree(cfs.getOverlappingSSTables(sstables)));
+ cfs.getAndReferenceOverlappingSSTables(sstables));
}
/**
@@ -79,7 +81,7 @@ public class CompactionController
private CompactionController(ColumnFamilyStore cfs,
int gcBefore,
- DataTracker.SSTableIntervalTree overlappingTree)
+ Set<SSTableReader> overlappingSSTables)
{
assert cfs != null;
this.cfs = cfs;
@@ -89,7 +91,8 @@ public class CompactionController
// add 5 minutes to be sure we're on the safe side in terms of thread safety (though we should be fine in our
// current 'stop all write during memtable switch' situation).
this.mergeShardBefore = (int) ((cfs.oldestUnflushedMemtable() + 5 * 3600) / 1000);
- this.overlappingTree = overlappingTree;
+ this.overlappingSSTables = overlappingSSTables == null ? Collections.<SSTableReader>emptySet() : overlappingSSTables;
+ overlappingTree = overlappingSSTables == null ? null : DataTracker.buildIntervalTree(overlappingSSTables);
}
public String getKeyspace()
@@ -169,4 +172,9 @@ public class CompactionController
{
throttle.throttle(currentBytes);
}
+
+ public void close()
+ {
+ SSTableReader.releaseReferences(overlappingSSTables);
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f09a89f4/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index 4f16f6a..5435a57 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@ -661,6 +661,7 @@ public class CompactionManager implements CompactionManagerMBean
}
finally
{
+ controller.close();
scanner.close();
metrics.finishCompaction(ci);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f09a89f4/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
index 575731f..3c676a6 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
@@ -199,6 +199,8 @@ public class CompactionTask extends AbstractCompactionTask
}
finally
{
+ controller.close();
+
try
{
iter.close();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f09a89f4/src/java/org/apache/cassandra/db/compaction/Scrubber.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/Scrubber.java b/src/java/org/apache/cassandra/db/compaction/Scrubber.java
index 3946c37..c6855e8 100644
--- a/src/java/org/apache/cassandra/db/compaction/Scrubber.java
+++ b/src/java/org/apache/cassandra/db/compaction/Scrubber.java
@@ -262,6 +262,10 @@ public class Scrubber implements Closeable
writer.abort();
throw Throwables.propagate(t);
}
+ finally
+ {
+ controller.close();
+ }
if (!outOfOrderRows.isEmpty())
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/f09a89f4/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java b/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
index 0859eaa..b4bea58 100644
--- a/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
+++ b/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
@@ -201,6 +201,10 @@ public class IncomingStreamReader
else
throw Throwables.propagate(e);
}
+ finally
+ {
+ controller.close();
+ }
}
private void retry()