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

Merge branch 'cassandra-2.1' into trunk 

Conflicts:
	src/java/org/apache/cassandra/db/compaction/CompactionManager.java
	src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
	src/java/org/apache/cassandra/service/ActiveRepairService.java
	src/java/org/apache/cassandra/streaming/StreamSession.java
	src/java/org/apache/cassandra/streaming/StreamTransferTask.java
	src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java
	test/unit/org/apache/cassandra/SchemaLoader.java
	test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java
	test/unit/org/apache/cassandra/db/compaction/BlacklistingCompactionsTest.java
	test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
This commit is contained in:
Benedict Elliott Smith 2015-01-28 15:19:58 +00:00
parent b4b1bdd325 c75ee4160c
commit 9c4a776d20
33 changed files with 1112 additions and 438 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -57,6 +57,7 @@
2.1.3
* Safer Resource Management (CASSANDRA-7705)
* Make sure we compact highly overlapping cold sstables with
STCS (CASSANDRA-8635)
* Make sure we don't add tmplink files to the compaction

1
build.xml
View File

@ -1120,6 +1120,7 @@
<!-- Cassandra 3.0+ needs <jvmarg line="... ${additionalagent}" /> here! (not value=) -->
<jvmarg line="-javaagent:${basedir}/lib/jamm-0.3.0.jar ${additionalagent}" />
<jvmarg value="-ea"/>
<jvmarg value="-Dcassandra.debugrefcount=true"/>
<jvmarg value="-Xss256k"/>
<jvmarg value="-Dcassandra.memtable_row_overhead_computation_step=100"/>
<jvmarg value="-Dcassandra.test.use_prepared=${cassandra.test.use_prepared}"/>

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

@ -33,6 +33,7 @@ import com.google.common.base.*;
import com.google.common.collect.*;
import com.google.common.util.concurrent.*;
import org.apache.cassandra.io.FSWriteError;
import org.json.simple.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -58,7 +59,6 @@ import org.apache.cassandra.dht.*;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.compress.CompressionParameters;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.*;
@ -73,8 +73,8 @@ import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.StreamLockfile;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.concurrent.*;
import org.apache.cassandra.utils.TopKSampler.SamplerResult;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.memory.MemtableAllocator;
import com.clearspring.analytics.stream.Counter;
@ -769,16 +769,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
logger.info("Loading new SSTables and building secondary indexes for {}/{}: {}", keyspace.getName(), name, newSSTables);
SSTableReader.acquireReferences(newSSTables);
data.addSSTables(newSSTables);
try
try (Refs<SSTableReader> refs = Refs.ref(newSSTables))
{
data.addSSTables(newSSTables);
indexManager.maybeBuildSecondaryIndexes(newSSTables, indexManager.allIndexesNames());
}
finally
{
SSTableReader.releaseReferences(newSSTables);
}
logger.info("Done loading load new SSTables for {}/{}", keyspace.getName(), name);
}
@ -790,18 +786,14 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
Set<String> indexes = new HashSet<String>(Arrays.asList(idxNames));
Collection<SSTableReader> sstables = cfs.getSSTables();
try
try (Refs<SSTableReader> refs = Refs.ref(sstables))
{
cfs.indexManager.setIndexRemoved(indexes);
SSTableReader.acquireReferences(sstables);
logger.info(String.format("User Requested secondary index re-build for %s/%s indexes", ksName, cfName));
cfs.indexManager.maybeBuildSecondaryIndexes(sstables, indexes);
cfs.indexManager.setIndexBuilt(indexes);
}
finally
{
SSTableReader.releaseReferences(sstables);
}
}
public String getColumnFamilyName()
@ -1293,13 +1285,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
* @return sstables whose key range overlaps with that of the given sstables, not including itself.
* (The given sstables may or may not overlap with each other.)
*/
public Set<SSTableReader> getOverlappingSSTables(Collection<SSTableReader> sstables)
public Collection<SSTableReader> getOverlappingSSTables(Iterable<SSTableReader> sstables)
{
logger.debug("Checking for sstables overlapping {}", sstables);
// a normal compaction won't ever have an empty sstables list, but we create a skeleton
// compaction controller for streaming, and that passes an empty list.
if (sstables.isEmpty())
if (!sstables.iterator().hasNext())
return ImmutableSet.of();
DataTracker.SSTableIntervalTree tree = data.getView().intervalTree;
@ -1318,13 +1310,14 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
/**
* like getOverlappingSSTables, but acquires references before returning
*/
public Set<SSTableReader> getAndReferenceOverlappingSSTables(Collection<SSTableReader> sstables)
public Refs<SSTableReader> getAndReferenceOverlappingSSTables(Iterable<SSTableReader> sstables)
{
while (true)
{
Set<SSTableReader> overlapped = getOverlappingSSTables(sstables);
if (SSTableReader.acquireReferences(overlapped))
return overlapped;
Iterable<SSTableReader> overlapped = getOverlappingSSTables(sstables);
Refs<SSTableReader> refs = Refs.tryRef(overlapped);
if (refs != null)
return refs;
}
}
@ -1776,38 +1769,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
return removeDeletedCF(cf, gcBefore);
}
/**
* Get the current view and acquires references on all its sstables.
* This is a bit tricky because we must ensure that between the time we
* get the current view and the time we acquire the references the set of
* sstables hasn't changed. Otherwise we could get a view for which an
* sstable have been deleted in the meantime.
*
* At the end of this method, a reference on all the sstables of the
* returned view will have been acquired and must thus be released when
* appropriate.
*/
private DataTracker.View markCurrentViewReferenced()
{
while (true)
{
DataTracker.View currentView = data.getView();
if (SSTableReader.acquireReferences(currentView.sstables))
return currentView;
}
}
/**
* Get the current sstables, acquiring references on all of them.
* The caller is in charge of releasing the references on the sstables.
*
* See markCurrentViewReferenced() above.
*/
public Collection<SSTableReader> markCurrentSSTablesReferenced()
{
return markCurrentViewReferenced().sstables;
}
public Set<SSTableReader> getUnrepairedSSTables()
{
Set<SSTableReader> unRepairedSSTables = new HashSet<>(getSSTables());
@ -1834,13 +1795,14 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
return repairedSSTables;
}
public ViewFragment selectAndReference(Function<DataTracker.View, List<SSTableReader>> filter)
public RefViewFragment selectAndReference(Function<DataTracker.View, List<SSTableReader>> filter)
{
while (true)
{
ViewFragment view = select(filter);
if (view.sstables.isEmpty() || SSTableReader.acquireReferences(view.sstables))
return view;
Refs<SSTableReader> refs = Refs.tryRef(view.sstables);
if (refs != null)
return new RefViewFragment(view.sstables, view.memtables, refs);
}
}
@ -2235,9 +2197,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
Set<SSTableReader> snapshottedSSTables = new HashSet<>();
for (ColumnFamilyStore cfs : concatWithIndexes())
{
DataTracker.View currentView = cfs.markCurrentViewReferenced();
final JSONArray filesJSONArr = new JSONArray();
try
try (RefViewFragment currentView = cfs.selectAndReference(ALL_SSTABLES))
{
for (SSTableReader ssTable : currentView.sstables)
{
@ -2256,10 +2217,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
writeSnapshotManifest(filesJSONArr, snapshotName);
}
finally
{
SSTableReader.releaseReferences(currentView.sstables);
}
}
return snapshottedSSTables;
}
@ -2286,13 +2243,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
}
public List<SSTableReader> getSnapshotSSTableReader(String tag) throws IOException
public Refs<SSTableReader> getSnapshotSSTableReader(String tag) throws IOException
{
Map<Integer, SSTableReader> active = new HashMap<>();
for (SSTableReader sstable : data.getView().sstables)
active.put(sstable.descriptor.generation, sstable);
Map<Descriptor, Set<Component>> snapshots = directories.sstableLister().snapshots(tag).list();
List<SSTableReader> readers = new ArrayList<>(snapshots.size());
Refs<SSTableReader> refs = new Refs<>();
try
{
for (Map.Entry<Descriptor, Set<Component>> entries : snapshots.entrySet())
@ -2300,29 +2257,28 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
// Try acquire reference to an active sstable instead of snapshot if it exists,
// to avoid opening new sstables. If it fails, use the snapshot reference instead.
SSTableReader sstable = active.get(entries.getKey().generation);
if (sstable == null || !sstable.acquireReference())
if (sstable == null || !refs.tryRef(sstable))
{
if (logger.isDebugEnabled())
logger.debug("using snapshot sstable {}", entries.getKey());
sstable = SSTableReader.open(entries.getKey(), entries.getValue(), metadata, partitioner);
// This is technically not necessary since it's a snapshot but makes things easier
sstable.acquireReference();
refs.tryRef(sstable);
}
else if (logger.isDebugEnabled())
{
logger.debug("using active sstable {}", entries.getKey());
}
readers.add(sstable);
}
}
catch (IOException | RuntimeException e)
{
// In case one of the snapshot sstables fails to open,
// we must release the references to the ones we opened so far
SSTableReader.releaseReferences(readers);
refs.release();
throw e;
}
return readers;
return refs;
}
/**
@ -2465,37 +2421,27 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public Iterable<DecoratedKey> keySamples(Range<Token> range)
{
Collection<SSTableReader> sstables = markCurrentSSTablesReferenced();
try
try (RefViewFragment view = selectAndReference(ALL_SSTABLES))
{
Iterable<DecoratedKey>[] samples = new Iterable[sstables.size()];
Iterable<DecoratedKey>[] samples = new Iterable[view.sstables.size()];
int i = 0;
for (SSTableReader sstable: sstables)
for (SSTableReader sstable: view.sstables)
{
samples[i++] = sstable.getKeySamples(range);
}
return Iterables.concat(samples);
}
finally
{
SSTableReader.releaseReferences(sstables);
}
}
public long estimatedKeysForRange(Range<Token> range)
{
Collection<SSTableReader> sstables = markCurrentSSTablesReferenced();
try
try (RefViewFragment view = selectAndReference(ALL_SSTABLES))
{
long count = 0;
for (SSTableReader sstable : sstables)
for (SSTableReader sstable : view.sstables)
count += sstable.estimatedKeysForRanges(Collections.singleton(range));
return count;
}
finally
{
SSTableReader.releaseReferences(sstables);
}
}
/**
@ -2874,6 +2820,27 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
}
public static class RefViewFragment extends ViewFragment implements AutoCloseable
{
public final Refs<SSTableReader> refs;
public RefViewFragment(List<SSTableReader> sstables, Iterable<Memtable> memtables, Refs<SSTableReader> refs)
{
super(sstables, memtables);
this.refs = refs;
}
public void release()
{
refs.release();
}
public void close()
{
refs.release();
}
}
/**
* Returns the creation time of the oldest memtable not fully flushed yet.
*/
@ -2937,4 +2904,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
fileIndexGenerator.set(0);
}
public static final Function<DataTracker.View, List<SSTableReader>> ALL_SSTABLES = new Function<DataTracker.View, List<SSTableReader>>()
{
public List<SSTableReader> apply(DataTracker.View view)
{
return new ArrayList<>(view.sstables);
}
};
}

6
src/java/org/apache/cassandra/db/DataTracker.java
View File

@ -367,7 +367,7 @@ public class DataTracker
while (!view.compareAndSet(currentView, newView));
for (SSTableReader sstable : currentView.sstables)
if (!remaining.contains(sstable))
sstable.releaseReference();
sstable.sharedRef().release();
notifySSTablesChanged(remaining, Collections.<SSTableReader>emptySet(), OperationType.UNKNOWN);
}
@ -406,7 +406,7 @@ public class DataTracker
sstable.setTrackedBy(this);
for (SSTableReader sstable : oldSSTables)
sstable.releaseReference();
sstable.sharedRef().release();
}
private void removeSSTablesFromTracker(Collection<SSTableReader> oldSSTables)
@ -467,7 +467,7 @@ public class DataTracker
{
boolean firstToCompact = sstable.markObsolete();
assert tolerateCompacted || firstToCompact : sstable + " was already marked compacted";
sstable.releaseReference();
sstable.sharedRef().release();
}
}

2
src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java
View File

@ -372,7 +372,7 @@ public abstract class AbstractCompactionStrategy
if (uncheckedTombstoneCompaction)
return true;
Set<SSTableReader> overlaps = cfs.getOverlappingSSTables(Collections.singleton(sstable));
Collection<SSTableReader> overlaps = cfs.getOverlappingSSTables(Collections.singleton(sstable));
if (overlaps.isEmpty())
{
// there is no overlap, tombstones are safely droppable

20
src/java/org/apache/cassandra/db/compaction/CompactionController.java
View File

@ -33,6 +33,8 @@ import org.apache.cassandra.db.DataTracker;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.utils.AlwaysPresentFilter;
import org.apache.cassandra.utils.concurrent.Refs;
/**
* Manage compaction options.
*/
@ -42,8 +44,8 @@ public class CompactionController implements AutoCloseable
public final ColumnFamilyStore cfs;
private DataTracker.SSTableIntervalTree overlappingTree;
private Set<SSTableReader> overlappingSSTables;
private final Set<SSTableReader> compacting;
private Refs<SSTableReader> overlappingSSTables;
private final Iterable<SSTableReader> compacting;
public final int gcBefore;
@ -76,11 +78,13 @@ public class CompactionController implements AutoCloseable
private void refreshOverlaps()
{
if (this.overlappingSSTables != null)
SSTableReader.releaseReferences(overlappingSSTables);
overlappingSSTables.release();
Set<SSTableReader> overlapping = compacting == null ? null : cfs.getAndReferenceOverlappingSSTables(compacting);
this.overlappingSSTables = overlapping == null ? Collections.<SSTableReader>emptySet() : overlapping;
this.overlappingTree = overlapping == null ? null : DataTracker.buildIntervalTree(overlapping);
if (compacting == null)
overlappingSSTables = Refs.tryRef(Collections.<SSTableReader>emptyList());
else
overlappingSSTables = cfs.getAndReferenceOverlappingSSTables(compacting);
this.overlappingTree = DataTracker.buildIntervalTree(overlappingSSTables);
}
public Set<SSTableReader> getFullyExpiredSSTables()
@ -104,7 +108,7 @@ public class CompactionController implements AutoCloseable
* @param gcBefore
* @return
*/
public static Set<SSTableReader> getFullyExpiredSSTables(ColumnFamilyStore cfStore, Set<SSTableReader> compacting, Set<SSTableReader> overlapping, int gcBefore)
public static Set<SSTableReader> getFullyExpiredSSTables(ColumnFamilyStore cfStore, Iterable<SSTableReader> compacting, Iterable<SSTableReader> overlapping, int gcBefore)
{
logger.debug("Checking droppable sstables in {}", cfStore);
@ -187,6 +191,6 @@ public class CompactionController implements AutoCloseable
public void close()
{
SSTableReader.releaseReferences(overlappingSSTables);
overlappingSSTables.release();
}
}

24
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -84,6 +84,8 @@ import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.concurrent.Refs;
/**
* <p>
* A singleton which manages a private executor of ongoing compactions.
@ -373,7 +375,7 @@ public class CompactionManager implements CompactionManagerMBean
public Future<?> submitAntiCompaction(final ColumnFamilyStore cfs,
final Collection<Range<Token>> ranges,
final Collection<SSTableReader> sstables,
final Refs<SSTableReader> sstables,
final long repairedAt)
{
Runnable runnable = new WrappedRunnable() {
@ -384,18 +386,12 @@ public class CompactionManager implements CompactionManagerMBean
while (!success)
{
for (SSTableReader compactingSSTable : cfs.getDataTracker().getCompacting())
{
if (sstables.remove(compactingSSTable))
SSTableReader.releaseReferences(Arrays.asList(compactingSSTable));
}
sstables.releaseIfHolds(compactingSSTable);
Set<SSTableReader> compactedSSTables = new HashSet<>();
for (SSTableReader sstable : sstables)
{
if (sstable.isMarkedCompacted())
compactedSSTables.add(sstable);
}
sstables.removeAll(compactedSSTables);
SSTableReader.releaseReferences(compactedSSTables);
sstables.release(compactedSSTables);
success = sstables.isEmpty() || cfs.getDataTracker().markCompacting(sstables);
}
performAnticompaction(cfs, ranges, sstables, repairedAt);
@ -417,7 +413,7 @@ public class CompactionManager implements CompactionManagerMBean
*/
public void performAnticompaction(ColumnFamilyStore cfs,
Collection<Range<Token>> ranges,
Collection<SSTableReader> validatedForRepair,
Refs<SSTableReader> validatedForRepair,
long repairedAt) throws InterruptedException, IOException
{
logger.info("Starting anticompaction for {}.{} on {}/{} sstables", cfs.keyspace.getName(), cfs.getColumnFamilyName(), validatedForRepair.size(), cfs.getSSTables().size());
@ -458,13 +454,13 @@ public class CompactionManager implements CompactionManagerMBean
}
cfs.getDataTracker().notifySSTableRepairedStatusChanged(mutatedRepairStatuses);
cfs.getDataTracker().unmarkCompacting(Sets.union(nonAnticompacting, mutatedRepairStatuses));
SSTableReader.releaseReferences(Sets.union(nonAnticompacting, mutatedRepairStatuses));
validatedForRepair.release(Sets.union(nonAnticompacting, mutatedRepairStatuses));
if (!sstables.isEmpty())
doAntiCompaction(cfs, ranges, sstables, repairedAt);
}
finally
{
SSTableReader.releaseReferences(sstables);
validatedForRepair.release();
cfs.getDataTracker().unmarkCompacting(sstables);
}
@ -930,7 +926,7 @@ public class CompactionManager implements CompactionManagerMBean
if (!cfs.isValid())
return;
Collection<SSTableReader> sstables = null;
Refs<SSTableReader> sstables = null;
try
{
@ -1040,7 +1036,7 @@ public class CompactionManager implements CompactionManagerMBean
finally
{
if (sstables != null)
SSTableReader.releaseReferences(sstables);
sstables.release();
}
}

9
src/java/org/apache/cassandra/db/compaction/CompactionTask.java
View File

@ -221,7 +221,14 @@ public class CompactionTask extends AbstractCompactionTask
}
catch (Throwable t)
{
writer.abort();
try
{
writer.abort();
}
catch (Throwable t2)
{
t.addSuppressed(t2);
}
throw t;
}
finally

10
src/java/org/apache/cassandra/db/index/SecondaryIndex.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.db.index;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.Set;
@ -54,6 +53,8 @@ import org.apache.cassandra.io.sstable.ReducingKeyIterator;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.Refs;
/**
* Abstract base class for different types of secondary indexes.
*
@ -209,8 +210,7 @@ public abstract class SecondaryIndex
logger.info(String.format("Submitting index build of %s for data in %s",
getIndexName(), StringUtils.join(baseCfs.getSSTables(), ", ")));
Collection<SSTableReader> sstables = baseCfs.markCurrentSSTablesReferenced();
try
try (Refs<SSTableReader> sstables = baseCfs.selectAndReference(ColumnFamilyStore.ALL_SSTABLES).refs)
{
SecondaryIndexBuilder builder = new SecondaryIndexBuilder(baseCfs,
Collections.singleton(getIndexName()),
@ -220,10 +220,6 @@ public abstract class SecondaryIndex
forceBlockingFlush();
setIndexBuilt();
}
finally
{
SSTableReader.releaseReferences(sstables);
}
logger.info("Index build of {} complete", getIndexName());
}

1
src/java/org/apache/cassandra/io/sstable/SSTable.java
View File

@ -38,6 +38,7 @@ import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.concurrent.RefCounted;
import org.apache.cassandra.utils.memory.HeapAllocator;
import org.apache.cassandra.utils.Pair;

14
src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
View File

@ -39,6 +39,8 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.OutputHandler;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.Ref;
/**
* Cassandra SSTable bulk loader.
* Load an externally created sstable into a cluster.
@ -125,7 +127,7 @@ public class SSTableLoader implements StreamEventHandler
List<Pair<Long, Long>> sstableSections = sstable.getPositionsForRanges(tokenRanges);
long estimatedKeys = sstable.estimatedKeysForRanges(tokenRanges);
StreamSession.SSTableStreamingSections details = new StreamSession.SSTableStreamingSections(sstable, sstableSections, estimatedKeys, ActiveRepairService.UNREPAIRED_SSTABLE);
StreamSession.SSTableStreamingSections details = new StreamSession.SSTableStreamingSections(sstable, sstable.sharedRef(), sstableSections, estimatedKeys, ActiveRepairService.UNREPAIRED_SSTABLE);
streamingDetails.put(endpoint, details);
}
@ -171,15 +173,17 @@ public class SSTableLoader implements StreamEventHandler
continue;
List<StreamSession.SSTableStreamingSections> endpointDetails = new LinkedList<>();
List<Ref> refs = new ArrayList<>();
try
{
// transferSSTables assumes references have been acquired
for (StreamSession.SSTableStreamingSections details : streamingDetails.get(remote))
{
if (!details.sstable.acquireReference())
Ref ref = details.sstable.tryRef();
if (ref == null)
throw new IllegalStateException();
refs.add(ref);
endpointDetails.add(details);
}
@ -187,8 +191,8 @@ public class SSTableLoader implements StreamEventHandler
}
finally
{
for (StreamSession.SSTableStreamingSections details : endpointDetails)
details.sstable.releaseReference();
for (Ref ref : refs)
ref.release();
}
}
plan.listeners(this, listeners);

15
src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java
View File

@ -178,14 +178,14 @@ public class SSTableRewriter
public void abort()
{
switchWriter(null);
switchWriter(null, true);
moveStarts(null, Functions.forMap(originalStarts), true);
// remove already completed SSTables
for (SSTableReader sstable : finished)
{
sstable.markObsolete();
sstable.releaseReference();
sstable.sharedRef().release();
}
// abort the writers
@ -277,6 +277,11 @@ public class SSTableRewriter
}
public void switchWriter(SSTableWriter newWriter)
{
switchWriter(newWriter, false);
}
private void switchWriter(SSTableWriter newWriter, boolean abort)
{
if (writer == null)
{
@ -285,7 +290,7 @@ public class SSTableRewriter
}
// we leave it as a tmp file, but we open it and add it to the dataTracker
if (writer.getFilePointer() != 0)
if (writer.getFilePointer() != 0 && !abort)
{
SSTableReader reader = writer.finish(SSTableWriter.FinishType.EARLY, maxAge, -1);
replaceEarlyOpenedFile(currentlyOpenedEarly, reader);
@ -335,7 +340,7 @@ public class SSTableRewriter
private List<SSTableReader> finishAndMaybeThrow(long repairedAt, boolean throwEarly, boolean throwLate)
{
List<SSTableReader> newReaders = new ArrayList<>();
switchWriter(null);
switchWriter(null, false);
if (throwEarly)
throw new RuntimeException("exception thrown early in finish, for testing");
@ -379,7 +384,7 @@ public class SSTableRewriter
{
if (reader.getCurrentReplacement() == null)
reader.markObsolete();
reader.releaseReference();
reader.sharedRef().release();
}
}
else

465
src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
View File

@ -22,7 +22,6 @@ import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import com.google.common.annotations.VisibleForTesting;
@ -63,6 +62,8 @@ import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.utils.concurrent.Ref;
import org.apache.cassandra.utils.concurrent.RefCounted;
import static org.apache.cassandra.db.Directories.SECONDARY_INDEX_NAME_SEPARATOR;
@ -70,7 +71,7 @@ import static org.apache.cassandra.db.Directories.SECONDARY_INDEX_NAME_SEPARATOR
* SSTableReaders are open()ed by Keyspace.onStart; after that they are created by SSTableWriter.renameAndOpen.
* Do not re-call open() on existing SSTable files; use the references kept by ColumnFamilyStore post-start instead.
*/
public abstract class SSTableReader extends SSTable
public abstract class SSTableReader extends SSTable implements RefCounted
{
private static final Logger logger = LoggerFactory.getLogger(SSTableReader.class);
@ -133,7 +134,6 @@ public abstract class SSTableReader extends SSTable
protected final BloomFilterTracker bloomFilterTracker = new BloomFilterTracker();
protected final AtomicInteger references = new AtomicInteger(1);
// technically isCompacted is not necessary since it should never be unreferenced unless it is also compacted,
// but it seems like a good extra layer of protection against reference counting bugs to not delete data based on that alone
protected final AtomicBoolean isCompacted = new AtomicBoolean(false);
@ -145,17 +145,8 @@ public abstract class SSTableReader extends SSTable
protected final AtomicLong keyCacheHit = new AtomicLong(0);
protected final AtomicLong keyCacheRequest = new AtomicLong(0);
/**
* To support replacing this sstablereader with another object that represents that same underlying sstable, but with different associated resources,
* we build a linked-list chain of replacement, which we synchronise using a shared object to make maintenance of the list across multiple threads simple.
* On close we check if any of the closeable resources differ between any chains either side of us; any that are in neither of the adjacent links (if any) are closed.
* Once we've made this decision we remove ourselves from the linked list, so that anybody behind/ahead will compare against only other still opened resources.
*/
protected Object replaceLock = new Object();
protected SSTableReader replacedBy;
private SSTableReader replaces;
private SSTableDeletingTask deletingTask;
private Runnable runOnClose;
private final Tidier tidy = new Tidier();
private final RefCounted refCounted = RefCounted.Impl.get(tidy);
@VisibleForTesting
public RestorableMeter readMeter;
@ -355,7 +346,7 @@ public abstract class SSTableReader extends SSTable
sstable.ifile = ibuilder.complete(sstable.descriptor.filenameFor(Component.PRIMARY_INDEX));
sstable.dfile = dbuilder.complete(sstable.descriptor.filenameFor(Component.DATA));
sstable.bf = FilterFactory.AlwaysPresent;
sstable.tidy.setup(sstable);
return sstable;
}
@ -400,6 +391,7 @@ public abstract class SSTableReader extends SSTable
if (sstable.getKeyCache() != null)
logger.debug("key cache contains {}/{} keys", sstable.getKeyCache().size(), sstable.getKeyCache().getCapacity());
sstable.tidy.setup(sstable);
return sstable;
}
@ -510,7 +502,7 @@ public abstract class SSTableReader extends SSTable
this.rowIndexEntrySerializer = descriptor.version.getSSTableFormat().getIndexSerializer(metadata);
deletingTask = new SSTableDeletingTask(this);
tidy.deletingTask = new SSTableDeletingTask(this);
// Don't track read rates for tables in the system keyspace. Also don't track reads for special operations (like early open)
// this is to avoid overflowing the executor queue (see CASSANDRA-8066)
@ -546,117 +538,6 @@ public abstract class SSTableReader extends SSTable
return sum;
}
private void tidy(boolean release)
{
if (readMeterSyncFuture != null)
readMeterSyncFuture.cancel(false);
if (references.get() != 0)
{
throw new IllegalStateException("SSTable is not fully released (" + references.get() + " references)");
}
synchronized (replaceLock)
{
boolean closeBf = true, closeSummary = true, closeFiles = true, deleteFiles = isCompacted.get();
if (replacedBy != null)
{
closeBf = replacedBy.bf != bf;
closeSummary = replacedBy.indexSummary != indexSummary;
closeFiles = replacedBy.dfile != dfile;
// if the replacement sstablereader uses a different path, clean up our paths
deleteFiles = !dfile.path.equals(replacedBy.dfile.path);
}
if (replaces != null)
{
closeBf &= replaces.bf != bf;
closeSummary &= replaces.indexSummary != indexSummary;
closeFiles &= replaces.dfile != dfile;
deleteFiles &= !dfile.path.equals(replaces.dfile.path);
}
boolean deleteAll = false;
if (release && isCompacted.get())
{
assert replacedBy == null;
if (replaces != null && !deleteFiles)
{
replaces.replacedBy = null;
replaces.deletingTask = deletingTask;
replaces.markObsolete();
}
else
{
deleteAll = true;
}
}
else
{
if (replaces != null)
replaces.replacedBy = replacedBy;
if (replacedBy != null)
replacedBy.replaces = replaces;
}
scheduleTidy(closeBf, closeSummary, closeFiles, deleteFiles, deleteAll);
}
}
private void scheduleTidy(final boolean closeBf, final boolean closeSummary, final boolean closeFiles, final boolean deleteFiles, final boolean deleteAll)
{
if (references.get() != 0)
throw new IllegalStateException("SSTable is not fully released (" + references.get() + " references)");
final ColumnFamilyStore cfs = Schema.instance.getColumnFamilyStoreInstance(metadata.cfId);
final OpOrder.Barrier barrier;
if (cfs != null)
{
barrier = cfs.readOrdering.newBarrier();
barrier.issue();
}
else
barrier = null;
ScheduledExecutors.nonPeriodicTasks.execute(new Runnable()
{
public void run()
{
if (barrier != null)
barrier.await();
if (closeBf)
bf.close();
if (closeSummary)
indexSummary.close();
if (closeFiles)
{
ifile.cleanup();
dfile.cleanup();
}
if (runOnClose != null)
runOnClose.run();
if (deleteAll)
{
/**
* Do the OS a favour and suggest (using fadvice call) that we
* don't want to see pages of this SSTable in memory anymore.
*
* NOTE: We can't use madvice in java because it requires the address of
* the mapping, so instead we always open a file and run fadvice(fd, 0, 0) on it
*/
dropPageCache();
deletingTask.run();
}
else if (deleteFiles)
{
FileUtils.deleteWithConfirm(new File(dfile.path));
FileUtils.deleteWithConfirm(new File(ifile.path));
}
}
});
}
public boolean equals(Object that)
{
return that instanceof SSTableReader && ((SSTableReader) that).descriptor.equals(this.descriptor);
@ -679,7 +560,7 @@ public abstract class SSTableReader extends SSTable
public void setTrackedBy(DataTracker tracker)
{
deletingTask.setTracker(tracker);
tidy.deletingTask.setTracker(tracker);
// under normal operation we can do this at any time, but SSTR is also used outside C* proper,
// e.g. by BulkLoader, which does not initialize the cache. As a kludge, we set up the cache
// here when we know we're being wired into the rest of the server infrastructure.
@ -751,6 +632,7 @@ public abstract class SSTableReader extends SSTable
dfile = dbuilder.complete(descriptor.filenameFor(Component.DATA));
if (saveSummaryIfCreated && (recreateBloomFilter || !summaryLoaded)) // save summary information to disk
saveSummary(ibuilder, dbuilder);
tidy.setup(this);
}
/**
@ -904,26 +786,26 @@ public abstract class SSTableReader extends SSTable
public void setReplacedBy(SSTableReader replacement)
{
synchronized (replaceLock)
synchronized (tidy.replaceLock)
{
assert replacedBy == null;
replacedBy = replacement;
replacement.replaces = this;
replacement.replaceLock = replaceLock;
assert tidy.replacedBy == null;
tidy.replacedBy = replacement;
replacement.tidy.replaces = this;
replacement.tidy.replaceLock = tidy.replaceLock;
}
}
public SSTableReader cloneWithNewStart(DecoratedKey newStart, final Runnable runOnClose)
{
synchronized (replaceLock)
synchronized (tidy.replaceLock)
{
assert replacedBy == null;
assert tidy.replacedBy == null;
if (newStart.compareTo(this.first) > 0)
{
if (newStart.compareTo(this.last) > 0)
{
this.runOnClose = new Runnable()
this.tidy.runOnClose = new Runnable()
{
public void run()
{
@ -937,7 +819,7 @@ public abstract class SSTableReader extends SSTable
{
final long dataStart = getPosition(newStart, Operator.GE).position;
final long indexStart = getIndexScanPosition(newStart);
this.runOnClose = new Runnable()
this.tidy.runOnClose = new Runnable()
{
public void run()
{
@ -970,9 +852,9 @@ public abstract class SSTableReader extends SSTable
*/
public SSTableReader cloneWithNewSummarySamplingLevel(ColumnFamilyStore parent, int samplingLevel) throws IOException
{
synchronized (replaceLock)
synchronized (tidy.replaceLock)
{
assert replacedBy == null;
assert tidy.replacedBy == null;
int minIndexInterval = metadata.getMinIndexInterval();
int maxIndexInterval = metadata.getMaxIndexInterval();
@ -1480,36 +1362,6 @@ public abstract class SSTableReader extends SSTable
return dfile.onDiskLength;
}
public boolean acquireReference()
{
while (true)
{
int n = references.get();
if (n <= 0)
return false;
if (references.compareAndSet(n, n + 1))
return true;
}
}
@VisibleForTesting
public int referenceCount()
{
return references.get();
}
/**
* Release reference to this SSTableReader.
* If there is no one referring to this SSTable, and is marked as compacted,
* all resources are cleaned up and files are deleted eventually.
*/
public void releaseReference()
{
if (references.decrementAndGet() == 0)
tidy(true);
assert references.get() >= 0 : "Reference counter " + references.get() + " for " + dfile.path;
}
/**
* Mark the sstable as obsolete, i.e., compacted into newer sstables.
*
@ -1524,9 +1376,9 @@ public abstract class SSTableReader extends SSTable
if (logger.isDebugEnabled())
logger.debug("Marking {} compacted", getFilename());
synchronized (replaceLock)
synchronized (tidy.replaceLock)
{
assert replacedBy == null : getFilename();
assert tidy.replacedBy == null : getFilename();
}
return !isCompacted.getAndSet(true);
}
@ -1637,13 +1489,13 @@ public abstract class SSTableReader extends SSTable
public SSTableReader getCurrentReplacement()
{
synchronized (replaceLock)
synchronized (tidy.replaceLock)
{
SSTableReader cur = this, next = replacedBy;
SSTableReader cur = this, next = tidy.replacedBy;
while (next != null)
{
cur = next;
next = next.replacedBy;
next = next.tidy.replacedBy;
}
return cur;
}
@ -1828,76 +1680,6 @@ public abstract class SSTableReader extends SSTable
return keyCacheRequest.get();
}
/**
* @param sstables
* @return true if all desired references were acquired. Otherwise, it will unreference any partial acquisition, and return false.
*/
public static boolean acquireReferences(Iterable<SSTableReader> sstables)
{
SSTableReader failed = null;
for (SSTableReader sstable : sstables)
{
if (!sstable.acquireReference())
{
failed = sstable;
break;
}
}
if (failed == null)
return true;
for (SSTableReader sstable : sstables)
{
if (sstable == failed)
break;
sstable.releaseReference();
}
return false;
}
public static void releaseReferences(Iterable<SSTableReader> sstables)
{
for (SSTableReader sstable : sstables)
{
sstable.releaseReference();
}
}
private void dropPageCache()
{
dropPageCache(dfile.path);
dropPageCache(ifile.path);
}
private void dropPageCache(String filePath)
{
RandomAccessFile file = null;
try
{
file = new RandomAccessFile(filePath, "r");
int fd = CLibrary.getfd(file.getFD());
if (fd > 0)
{
if (logger.isDebugEnabled())
logger.debug(String.format("Dropping page cache of file %s.", filePath));
CLibrary.trySkipCache(fd, 0, 0);
}
}
catch (IOException e)
{
// we don't care if cache cleanup fails
}
finally
{
FileUtils.closeQuietly(file);
}
}
/**
* Increment the total row read count and read rate for this SSTable. This should not be incremented for range
* slice queries, row cache hits, or non-query reads, like compaction.
@ -1916,6 +1698,201 @@ public abstract class SSTableReader extends SSTable
}
}
public Ref tryRef()
{
return refCounted.tryRef();
}
public Ref sharedRef()
{
return refCounted.sharedRef();
}
private static final class Tidier implements Tidy
{
private String name;
private CFMetaData metadata;
// indexfile and datafile: might be null before a call to load()
private SegmentedFile ifile;
private SegmentedFile dfile;
private IndexSummary indexSummary;
private IFilter bf;
private AtomicBoolean isCompacted;
/**
* To support replacing this sstablereader with another object that represents that same underlying sstable, but with different associated resources,
* we build a linked-list chain of replacement, which we synchronise using a shared object to make maintenance of the list across multiple threads simple.
* On close we check if any of the closeable resources differ between any chains either side of us; any that are in neither of the adjacent links (if any) are closed.
* Once we've made this decision we remove ourselves from the linked list, so that anybody behind/ahead will compare against only other still opened resources.
*/
private Object replaceLock = new Object();
private SSTableReader replacedBy;
private SSTableReader replaces;
private SSTableDeletingTask deletingTask;
private Runnable runOnClose;
@VisibleForTesting
public RestorableMeter readMeter;
private volatile ScheduledFuture readMeterSyncFuture;
private void setup(SSTableReader reader)
{
name = reader.toString();
metadata = reader.metadata;
ifile = reader.ifile;
dfile = reader.dfile;
indexSummary = reader.indexSummary;
bf = reader.bf;
isCompacted = reader.isCompacted;
readMeterSyncFuture = reader.readMeterSyncFuture;
}
public String name()
{
return name;
}
private void dropPageCache()
{
dropPageCache(dfile.path);
dropPageCache(ifile.path);
}
private void dropPageCache(String filePath)
{
RandomAccessFile file = null;
try
{
file = new RandomAccessFile(filePath, "r");
int fd = CLibrary.getfd(file.getFD());
if (fd > 0)
{
if (logger.isDebugEnabled())
logger.debug(String.format("Dropping page cache of file %s.", filePath));
CLibrary.trySkipCache(fd, 0, 0);
}
}
catch (IOException e)
{
// we don't care if cache cleanup fails
}
finally
{
FileUtils.closeQuietly(file);
}
}
public void tidy()
{
if (readMeterSyncFuture != null)
readMeterSyncFuture.cancel(false);
synchronized (replaceLock)
{
boolean closeBf = true, closeSummary = true, closeFiles = true, deleteFiles = isCompacted.get();
if (replacedBy != null)
{
closeBf = replacedBy.bf != bf;
closeSummary = replacedBy.indexSummary != indexSummary;
closeFiles = replacedBy.dfile != dfile;
// if the replacement sstablereader uses a different path, clean up our paths
deleteFiles = !dfile.path.equals(replacedBy.dfile.path);
}
if (replaces != null)
{
closeBf &= replaces.bf != bf;
closeSummary &= replaces.indexSummary != indexSummary;
closeFiles &= replaces.dfile != dfile;
deleteFiles &= !dfile.path.equals(replaces.dfile.path);
}
boolean deleteAll = false;
if (isCompacted.get())
{
assert replacedBy == null;
if (replaces != null && !deleteFiles)
{
replaces.tidy.replacedBy = null;
replaces.tidy.deletingTask = deletingTask;
replaces.markObsolete();
}
else
{
deleteAll = true;
}
}
else
{
closeSummary &= indexSummary != null;
if (replaces != null)
replaces.tidy.replacedBy = replacedBy;
if (replacedBy != null)
replacedBy.tidy.replaces = replaces;
}
scheduleTidy(closeBf, closeSummary, closeFiles, deleteFiles, deleteAll);
}
}
private void scheduleTidy(final boolean closeBf, final boolean closeSummary, final boolean closeFiles, final boolean deleteFiles, final boolean deleteAll)
{
final ColumnFamilyStore cfs = Schema.instance.getColumnFamilyStoreInstance(metadata.cfId);
final OpOrder.Barrier barrier;
if (cfs != null)
{
barrier = cfs.readOrdering.newBarrier();
barrier.issue();
}
else
barrier = null;
ScheduledExecutors.nonPeriodicTasks.execute(new Runnable()
{
public void run()
{
if (barrier != null)
barrier.await();
if (closeBf)
bf.close();
if (closeSummary)
indexSummary.close();
if (closeFiles)
{
ifile.cleanup();
dfile.cleanup();
}
if (runOnClose != null)
runOnClose.run();
if (deleteAll)
{
/**
* Do the OS a favour and suggest (using fadvice call) that we
* don't want to see pages of this SSTable in memory anymore.
*
* NOTE: We can't use madvice in java because it requires the address of
* the mapping, so instead we always open a file and run fadvice(fd, 0, 0) on it
*/
dropPageCache();
deletingTask.run();
}
else if (deleteFiles)
{
FileUtils.deleteWithConfirm(new File(dfile.path));
FileUtils.deleteWithConfirm(new File(ifile.path));
}
}
});
}
}
public static abstract class Factory
{
public abstract SSTableReader open(final Descriptor descriptor,

4
src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java
View File

@ -355,7 +355,7 @@ public class BigTableWriter extends SSTableWriter
if (inclusiveUpperBoundOfReadableData == null)
{
// Prevent leaving tmplink files on disk
sstable.releaseReference();
sstable.sharedRef().release();
return null;
}
int offset = 2;
@ -367,7 +367,7 @@ public class BigTableWriter extends SSTableWriter
inclusiveUpperBoundOfReadableData = iwriter.getMaxReadableKey(offset++);
if (inclusiveUpperBoundOfReadableData == null)
{
sstable.releaseReference();
sstable.sharedRef().release();
return null;
}
}

26
src/java/org/apache/cassandra/service/ActiveRepairService.java
View File

@ -25,6 +25,7 @@ import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.ListeningExecutorService;
@ -54,6 +55,10 @@ import org.apache.cassandra.repair.RepairSession;
import org.apache.cassandra.repair.messages.*;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.cassandra.utils.concurrent.Ref;
import org.apache.cassandra.utils.concurrent.RefCounted;
import org.apache.cassandra.utils.concurrent.Refs;
/**
* ActiveRepairService is the starting point for manual "active" repairs.
@ -343,7 +348,7 @@ public class ActiveRepairService
return futures;
for (Map.Entry<UUID, ColumnFamilyStore> columnFamilyStoreEntry : prs.columnFamilyStores.entrySet())
{
Collection<SSTableReader> sstables = new HashSet<>(prs.getAndReferenceSSTables(columnFamilyStoreEntry.getKey()));
Refs<SSTableReader> sstables = prs.getAndReferenceSSTables(columnFamilyStoreEntry.getKey());
ColumnFamilyStore cfs = columnFamilyStoreEntry.getValue();
futures.add(CompactionManager.instance.submitAntiCompaction(cfs, successfulRanges, sstables, prs.repairedAt));
}
@ -399,10 +404,11 @@ public class ActiveRepairService
this.sstableMap.put(cfId, existingSSTables);
}
public synchronized Collection<SSTableReader> getAndReferenceSSTables(UUID cfId)
public synchronized Refs<SSTableReader> getAndReferenceSSTables(UUID cfId)
{
Set<SSTableReader> sstables = sstableMap.get(cfId);
Iterator<SSTableReader> sstableIterator = sstables.iterator();
ImmutableMap.Builder<SSTableReader, Ref> references = ImmutableMap.builder();
while (sstableIterator.hasNext())
{
SSTableReader sstable = sstableIterator.next();
@ -412,23 +418,25 @@ public class ActiveRepairService
}
else
{
if (!sstable.acquireReference())
Ref ref = sstable.tryRef();
if (ref == null)
sstableIterator.remove();
else
references.put(sstable, ref);
}
}
return sstables;
return new Refs<>(references.build());
}
public synchronized Set<SSTableReader> getAndReferenceSSTablesInRange(UUID cfId, Range<Token> range)
public synchronized Refs<SSTableReader> getAndReferenceSSTablesInRange(UUID cfId, Range<Token> range)
{
Collection<SSTableReader> allSSTables = getAndReferenceSSTables(cfId);
Set<SSTableReader> sstables = new HashSet<>();
for (SSTableReader sstable : allSSTables)
Refs<SSTableReader> sstables = getAndReferenceSSTables(cfId);
for (SSTableReader sstable : new ArrayList<>(sstables))
{
if (new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(Arrays.asList(range)))
sstables.add(sstable);
else
sstable.releaseReference();
sstables.release(sstable);
}
return sstables;
}

10
src/java/org/apache/cassandra/streaming/StreamReceiveTask.java
View File

@ -35,6 +35,8 @@ import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.format.SSTableWriter;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.Refs;
/**
* Task that manages receiving files for the session for certain ColumnFamily.
*/
@ -124,18 +126,12 @@ public class StreamReceiveTask extends StreamTask
lockfile.delete();
task.sstables.clear();
if (!SSTableReader.acquireReferences(readers))
throw new AssertionError("We shouldn't fail acquiring a reference on a sstable that has just been transferred");
try
try (Refs<SSTableReader> refs = Refs.ref(readers))
{
// add sstables and build secondary indexes
cfs.addSSTables(readers);
cfs.indexManager.maybeBuildSecondaryIndexes(readers, cfs.indexManager.allIndexesNames());
}
finally
{
SSTableReader.releaseReferences(readers);
}
task.session.taskCompleted(task);
}

28
src/java/org/apache/cassandra/streaming/StreamSession.java
View File

@ -45,6 +45,10 @@ import org.apache.cassandra.streaming.messages.*;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.RefCounted;
import org.apache.cassandra.utils.concurrent.Ref;
import org.apache.cassandra.utils.concurrent.Refs;
/**
* Handles the streaming a one or more section of one of more sstables to and from a specific
@ -274,7 +278,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
finally
{
for (SSTableStreamingSections release : sections)
release.sstable.releaseReference();
release.ref.release();
}
}
@ -296,7 +300,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
private List<SSTableStreamingSections> getSSTableSectionsForRanges(Collection<Range<Token>> ranges, Collection<ColumnFamilyStore> stores, long overriddenRepairedAt)
{
List<SSTableReader> sstables = new ArrayList<>();
Refs<SSTableReader> refs = new Refs<>();
try
{
for (ColumnFamilyStore cfStore : stores)
@ -304,16 +308,16 @@ public class StreamSession implements IEndpointStateChangeSubscriber
List<AbstractBounds<RowPosition>> rowBoundsList = new ArrayList<>(ranges.size());
for (Range<Token> range : ranges)
rowBoundsList.add(range.toRowBounds());
ColumnFamilyStore.ViewFragment view = cfStore.selectAndReference(cfStore.viewFilter(rowBoundsList));
sstables.addAll(view.sstables);
refs.addAll(cfStore.selectAndReference(cfStore.viewFilter(rowBoundsList)).refs);
}
List<SSTableStreamingSections> sections = new ArrayList<>(sstables.size());
for (SSTableReader sstable : sstables)
List<SSTableStreamingSections> sections = new ArrayList<>(refs.size());
for (SSTableReader sstable : refs)
{
long repairedAt = overriddenRepairedAt;
if (overriddenRepairedAt == ActiveRepairService.UNREPAIRED_SSTABLE)
repairedAt = sstable.getSSTableMetadata().repairedAt;
sections.add(new SSTableStreamingSections(sstable,
sections.add(new SSTableStreamingSections(sstable, refs.get(sstable),
sstable.getPositionsForRanges(ranges),
sstable.estimatedKeysForRanges(ranges),
repairedAt));
@ -322,7 +326,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
}
catch (Throwable t)
{
SSTableReader.releaseReferences(sstables);
refs.release();
throw t;
}
}
@ -336,7 +340,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
if (details.sections.isEmpty())
{
// A reference was acquired on the sstable and we won't stream it
details.sstable.releaseReference();
details.ref.release();
iter.remove();
continue;
}
@ -348,7 +352,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
task = new StreamTransferTask(this, cfId);
transfers.put(cfId, task);
}
task.addTransferFile(details.sstable, details.estimatedKeys, details.sections, details.repairedAt);
task.addTransferFile(details.sstable, details.ref, details.estimatedKeys, details.sections, details.repairedAt);
iter.remove();
}
}
@ -356,13 +360,15 @@ public class StreamSession implements IEndpointStateChangeSubscriber
public static class SSTableStreamingSections
{
public final SSTableReader sstable;
public final Ref ref;
public final List<Pair<Long, Long>> sections;
public final long estimatedKeys;
public final long repairedAt;
public SSTableStreamingSections(SSTableReader sstable, List<Pair<Long, Long>> sections, long estimatedKeys, long repairedAt)
public SSTableStreamingSections(SSTableReader sstable, Ref ref, List<Pair<Long, Long>> sections, long estimatedKeys, long repairedAt)
{
this.sstable = sstable;
this.ref = ref;
this.sections = sections;
this.estimatedKeys = estimatedKeys;
this.repairedAt = repairedAt;

10
src/java/org/apache/cassandra/streaming/StreamTransferTask.java
View File

@ -26,6 +26,8 @@ import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.streaming.messages.OutgoingFileMessage;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.Ref;
import org.apache.cassandra.utils.concurrent.RefCounted;
/**
* StreamTransferTask sends sections of SSTable files in certain ColumnFamily.
@ -47,10 +49,10 @@ public class StreamTransferTask extends StreamTask
super(session, cfId);
}
public synchronized void addTransferFile(SSTableReader sstable, long estimatedKeys, List<Pair<Long, Long>> sections, long repairedAt)
public synchronized void addTransferFile(SSTableReader sstable, Ref ref, long estimatedKeys, List<Pair<Long, Long>> sections, long repairedAt)
{
assert sstable != null && cfId.equals(sstable.metadata.cfId);
OutgoingFileMessage message = new OutgoingFileMessage(sstable, sequenceNumber.getAndIncrement(), estimatedKeys, sections, repairedAt, session.keepSSTableLevel());
OutgoingFileMessage message = new OutgoingFileMessage(sstable, ref, sequenceNumber.getAndIncrement(), estimatedKeys, sections, repairedAt, session.keepSSTableLevel());
files.put(message.header.sequenceNumber, message);
totalSize += message.header.size();
}
@ -71,7 +73,7 @@ public class StreamTransferTask extends StreamTask
OutgoingFileMessage file = files.remove(sequenceNumber);
if (file != null)
file.sstable.releaseReference();
file.ref.release();
signalComplete = files.isEmpty();
}
@ -92,7 +94,7 @@ public class StreamTransferTask extends StreamTask
timeoutTasks.clear();
for (OutgoingFileMessage file : files.values())
file.sstable.releaseReference();
file.ref.release();
}
public synchronized int getTotalNumberOfFiles()

10
src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java
View File

@ -30,6 +30,8 @@ import org.apache.cassandra.streaming.compress.CompressedStreamWriter;
import org.apache.cassandra.streaming.compress.CompressionInfo;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.Ref;
/**
* OutgoingFileMessage is used to transfer the part(or whole) of a SSTable data file.
*/
@ -57,13 +59,15 @@ public class OutgoingFileMessage extends StreamMessage
}
};
public FileMessageHeader header;
public SSTableReader sstable;
public final FileMessageHeader header;
public final SSTableReader sstable;
public final Ref ref;
public OutgoingFileMessage(SSTableReader sstable, int sequenceNumber, long estimatedKeys, List<Pair<Long, Long>> sections, long repairedAt, boolean keepSSTableLevel)
public OutgoingFileMessage(SSTableReader sstable, Ref ref, int sequenceNumber, long estimatedKeys, List<Pair<Long, Long>> sections, long repairedAt, boolean keepSSTableLevel)
{
super(Type.FILE);
this.sstable = sstable;
this.ref = ref;
CompressionInfo compressionInfo = null;
if (sstable.compression)

2
src/java/org/apache/cassandra/tools/StandaloneScrubber.java
View File

@ -119,7 +119,7 @@ public class StandaloneScrubber
// Remove the sstable (it's been copied by scrub and snapshotted)
sstable.markObsolete();
sstable.releaseReference();
sstable.sharedRef().release();
}
catch (Exception e)
{

134
src/java/org/apache/cassandra/utils/concurrent/Ref.java Normal file
View File

@ -0,0 +1,134 @@
package org.apache.cassandra.utils.concurrent;
import java.lang.ref.PhantomReference;
import java.lang.ref.ReferenceQueue;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* A single managed reference to a RefCounted object
*/
public final class Ref
{
static final Logger logger = LoggerFactory.getLogger(Ref.class);
static final boolean DEBUG_ENABLED = System.getProperty("cassandra.debugrefcount", "false").equalsIgnoreCase("true");
final State state;
Ref(RefCountedImpl.GlobalState state, boolean isSharedRef)
{
this.state = new State(state, this, RefCountedImpl.referenceQueue, isSharedRef);
}
/**
* Must be called exactly once, when the logical operation for which this Ref was created has terminated.
* Failure to abide by this contract will result in an error (eventually) being reported, assuming a
* hard reference to the resource it managed is not leaked.
*/
public void release()
{
state.release(false);
}
/**
* A convenience method for reporting:
* @return the number of currently extant references globally, including the shared reference
*/
public int globalCount()
{
return state.globalState.count();
}
// similar to RefCountedState, but tracks only the management of each unique ref created to the managed object
// ensures it is only released once, and that it is always released
static final class State extends PhantomReference<Ref>
{
final Debug debug = DEBUG_ENABLED ? new Debug() : null;
final boolean isSharedRef;
final RefCountedImpl.GlobalState globalState;
private volatile int released;
private static final AtomicIntegerFieldUpdater<State> releasedUpdater = AtomicIntegerFieldUpdater.newUpdater(State.class, "released");
public State(final RefCountedImpl.GlobalState globalState, Ref reference, ReferenceQueue<? super Ref> q, boolean isSharedRef)
{
super(reference, q);
this.globalState = globalState;
this.isSharedRef = isSharedRef;
globalState.register(this);
}
void release(boolean leak)
{
if (!releasedUpdater.compareAndSet(this, 0, 1))
{
if (!leak)
{
String id = this.toString();
logger.error("BAD RELEASE: attempted to release a{} reference ({}) that has already been released", isSharedRef ? " shared" : "", id);
if (DEBUG_ENABLED)
debug.log(id);
throw new IllegalStateException("Attempted to release a reference that has already been released");
}
return;
}
globalState.release(this);
if (leak)
{
String id = this.toString();
if (isSharedRef)
logger.error("LEAK DETECTED: the shared reference ({}) to {} was not released before the object was garbage collected", id, globalState);
else
logger.error("LEAK DETECTED: a reference ({}) to {} was not released before the reference was garbage collected", id, globalState);
if (DEBUG_ENABLED)
debug.log(id);
}
else if (DEBUG_ENABLED)
{
debug.deallocate();
}
}
}
static final class Debug
{
String allocateThread, deallocateThread;
StackTraceElement[] allocateTrace, deallocateTrace;
Debug()
{
Thread thread = Thread.currentThread();
allocateThread = thread.toString();
allocateTrace = thread.getStackTrace();
}
synchronized void deallocate()
{
Thread thread = Thread.currentThread();
deallocateThread = thread.toString();
deallocateTrace = thread.getStackTrace();
}
synchronized void log(String id)
{
logger.error("Allocate trace {}:\n{}", id, print(allocateThread, allocateTrace));
if (deallocateThread != null)
logger.error("Deallocate trace {}:\n{}", id, print(deallocateThread, deallocateTrace));
}
String print(String thread, StackTraceElement[] trace)
{
StringBuilder sb = new StringBuilder();
sb.append(thread.toString());
sb.append("\n");
for (StackTraceElement element : trace)
{
sb.append("\tat ");
sb.append(element );
sb.append("\n");
}
return sb.toString();
}
}
}

94
src/java/org/apache/cassandra/utils/concurrent/RefCounted.java Normal file
View File

@ -0,0 +1,94 @@
/*
* 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.concurrent;
import java.lang.ref.PhantomReference;
import java.lang.ref.ReferenceQueue;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterators;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.NamedThreadFactory;
/**
* An object that needs ref counting does the following:
* - defines a Tidy object that will cleanup once it's gone,
* (this must retain no references to the object we're tracking (only its resources and how to clean up))
* - implements RefCounted
* - encapsulates a RefCounted.Impl, to which it proxies all calls to RefCounted behaviours
* - ensures no external access to the encapsulated Impl, and permits no references to it to leak
* - users must ensure no references to the sharedRef leak, or are retained outside of a method scope either.
* (to ensure the sharedRef is collected with the object, so that leaks may be detected and corrected)
*
* This class' functionality is achieved by what may look at first glance like a complex web of references,
* but boils down to:
*
* Target --> Impl --> sharedRef --> [RefState] <--> RefCountedState --> Tidy
* ^ ^
* | |
* Ref ----------------------------------- |
* |
* Global -------------------------------------------------
*
* So that, if Target is collected, Impl is collected and, hence, so is sharedRef.
*
* Once ref or sharedRef are collected, the paired RefState's release method is called, which if it had
* not already been called will update RefCountedState and log an error.
*
* Once the RefCountedState has been completely released, the Tidy method is called and it removes the global reference
* to itself so it may also be collected.
*/
public interface RefCounted
{
/**
* @return the a new Ref() to the managed object, incrementing its refcount, or null if it is already released
*/
public Ref tryRef();
/**
* @return the shared Ref that is created at instantiation of the RefCounted instance.
* Once released, if no other refs are extant the object will be tidied; references to
* this object should never be retained outside of a method's scope
*/
public Ref sharedRef();
public static interface Tidy
{
void tidy();
String name();
}
public static class Impl
{
public static RefCounted get(Tidy tidy)
{
return new RefCountedImpl(tidy);
}
}
}

132
src/java/org/apache/cassandra/utils/concurrent/RefCountedImpl.java Normal file
View File

@ -0,0 +1,132 @@
package org.apache.cassandra.utils.concurrent;
import java.lang.ref.ReferenceQueue;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.cassandra.concurrent.NamedThreadFactory;
// default implementation; can be hidden and proxied (like we do for SSTableReader)
final class RefCountedImpl implements RefCounted
{
private final Ref sharedRef;
private final GlobalState state;
public RefCountedImpl(Tidy tidy)
{
this.state = new GlobalState(tidy);
sharedRef = new Ref(this.state, true);
globallyExtant.add(this.state);
}
/**
* see {@link RefCounted#tryRef()}
*/
public Ref tryRef()
{
return state.ref() ? new Ref(state, false) : null;
}
/**
* see {@link RefCounted#sharedRef()}
*/
public Ref sharedRef()
{
return sharedRef;
}
// the object that manages the actual cleaning up; this does not reference the RefCounted.Impl
// so that we can detect when references are lost to the resource itself, and still cleanup afterwards
// the Tidy object MUST not contain any references to the object we are managing
static final class GlobalState
{
// we need to retain a reference to each of the PhantomReference instances
// we are using to track individual refs
private final ConcurrentLinkedQueue<Ref.State> locallyExtant = new ConcurrentLinkedQueue<>();
// the number of live refs
private final AtomicInteger counts = new AtomicInteger();
// the object to call to cleanup when our refs are all finished with
private final Tidy tidy;
GlobalState(Tidy tidy)
{
this.tidy = tidy;
}
void register(Ref.State ref)
{
locallyExtant.add(ref);
}
// increment ref count if not already tidied, and return success/failure
boolean ref()
{
while (true)
{
int cur = counts.get();
if (cur < 0)
return false;
if (counts.compareAndSet(cur, cur + 1))
return true;
}
}
// release a single reference, and cleanup if no more are extant
void release(Ref.State ref)
{
locallyExtant.remove(ref);
if (-1 == counts.decrementAndGet())
{
globallyExtant.remove(this);
tidy.tidy();
}
}
int count()
{
return 1 + counts.get();
}
public String toString()
{
return tidy.name();
}
}
private static final Set<GlobalState> globallyExtant = Collections.newSetFromMap(new ConcurrentHashMap<GlobalState, Boolean>());
static final ReferenceQueue<Object> referenceQueue = new ReferenceQueue<>();
private static final ExecutorService EXEC = Executors.newFixedThreadPool(1, new NamedThreadFactory("Reference-Reaper"));
static
{
EXEC.execute(new Runnable()
{
public void run()
{
try
{
while (true)
{
Object obj = referenceQueue.remove();
if (obj instanceof Ref.State)
{
((Ref.State) obj).release(true);
}
}
}
catch (InterruptedException e)
{
}
finally
{
EXEC.execute(this);
}
}
});
}
}

219
src/java/org/apache/cassandra/utils/concurrent/Refs.java Normal file
View File

@ -0,0 +1,219 @@
package org.apache.cassandra.utils.concurrent;
import java.util.*;
import com.google.common.base.Throwables;
import com.google.common.collect.Iterators;
/**
* A collection of managed Ref references to RefCounted objects, and the objects they are referencing.
* Care MUST be taken when using this collection, as if a permanent reference to it leaks we will not
* be alerted to a lack of reference release.
*
* All of the java.util.Collection operations that modify the collection are unsupported.
*/
public final class Refs<T extends RefCounted> extends AbstractCollection<T> implements AutoCloseable
{
private final Map<T, Ref> references;
public Refs()
{
this.references = new HashMap<>();
}
public Refs(Map<T, Ref> references)
{
this.references = new HashMap<>(references);
}
/**
* Release ALL of the references held by this Refs collection
*/
public void release()
{
try
{
release(references.values());
}
finally
{
references.clear();
}
}
/**
* See {@link Refs#release()}
*/
public void close()
{
release();
}
/**
* @param referenced the object we have a Ref to
* @return the Ref to said object
*/
public Ref get(T referenced)
{
return references.get(referenced);
}
/**
* @param referenced the object we have a Ref to
*/
public void release(T referenced)
{
Ref ref = references.remove(referenced);
if (ref == null)
throw new IllegalStateException("This Refs collection does not hold a reference to " + referenced);
ref.release();
}
/**
* Release the retained Ref to the provided object, if held, return false otherwise
* @param referenced the object we retain a Ref to
* @return return true if we held a reference to the object, and false otherwise
*/
public boolean releaseIfHolds(T referenced)
{
Ref ref = references.remove(referenced);
if (ref != null)
ref.release();
return ref != null;
}
/**
* Release a retained Ref to all of the provided objects; if any is not held, an exception will be thrown
* @param release
*/
public void release(Collection<T> release)
{
List<Ref> refs = new ArrayList<>();
List<T> notPresent = null;
for (T obj : release)
{
Ref ref = references.remove(obj);
if (ref == null)
{
if (notPresent == null)
notPresent = new ArrayList<>();
notPresent.add(obj);
}
else
{
refs.add(ref);
}
}
IllegalStateException notPresentFail = null;
if (notPresent != null)
{
notPresentFail = new IllegalStateException("Could not release references to " + notPresent
+ " as references to these objects were not held");
notPresentFail.fillInStackTrace();
}
try
{
release(refs);
}
catch (Throwable t)
{
if (notPresentFail != null)
t.addSuppressed(notPresentFail);
}
if (notPresentFail != null)
throw notPresentFail;
}
/**
* Attempt to take a reference to the provided object; if it has already been released, null will be returned
* @param t object to acquire a reference to
* @return true iff success
*/
public boolean tryRef(T t)
{
Ref ref = t.tryRef();
if (ref == null)
return false;
ref = references.put(t, ref);
if (ref != null)
ref.release(); // release dup
return true;
}
public Iterator<T> iterator()
{
return Iterators.unmodifiableIterator(references.keySet().iterator());
}
public int size()
{
return references.size();
}
/**
* Merge two sets of references, ensuring only one reference is retained between the two sets
*/
public Refs<T> addAll(Refs<T> add)
{
List<Ref> overlap = new ArrayList<>();
for (Map.Entry<T, Ref> e : add.references.entrySet())
{
if (this.references.containsKey(e.getKey()))
overlap.add(e.getValue());
else
this.references.put(e.getKey(), e.getValue());
}
add.references.clear();
release(overlap);
return this;
}
/**
* Acquire a reference to all of the provided objects, or none
*/
public static <T extends RefCounted> Refs<T> tryRef(Iterable<T> reference)
{
HashMap<T, Ref> refs = new HashMap<>();
for (T rc : reference)
{
Ref ref = rc.tryRef();
if (ref == null)
{
release(refs.values());
return null;
}
refs.put(rc, ref);
}
return new Refs<T>(refs);
}
public static <T extends RefCounted> Refs<T> ref(Iterable<T> reference)
{
Refs<T> refs = tryRef(reference);
if (refs != null)
return refs;
throw new IllegalStateException();
}
private static void release(Iterable<Ref> refs)
{
Throwable fail = null;
for (Ref ref : refs)
{
try
{
ref.release();
}
catch (Throwable t)
{
if (fail == null)
fail = t;
else
fail.addSuppressed(t);
}
}
if (fail != null)
throw Throwables.propagate(fail);
}
}

11
test/unit/org/apache/cassandra/SchemaLoader.java
View File

@ -21,6 +21,8 @@ import java.io.File;
import java.nio.ByteBuffer;
import java.util.*;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -58,6 +60,15 @@ public class SchemaLoader
startGossiper();
}
@After
public void leakDetect() throws InterruptedException
{
System.gc();
System.gc();
System.gc();
Thread.sleep(10);
}
public static void prepareServer()
{
// Cleanup first

9
test/unit/org/apache/cassandra/db/KeyCacheTest.java
View File

@ -43,6 +43,7 @@ import org.apache.cassandra.locator.SimpleStrategy;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.concurrent.Refs;
import static org.junit.Assert.assertEquals;
public class KeyCacheTest
@ -169,8 +170,9 @@ public class KeyCacheTest
assertKeyCacheSize(2, KEYSPACE1, COLUMN_FAMILY1);
Set<SSTableReader> readers = cfs.getDataTracker().getSSTables();
for (SSTableReader reader : readers)
reader.acquireReference();
Refs<SSTableReader> refs = Refs.tryRef(readers);
if (refs == null)
throw new IllegalStateException();
Util.compactAll(cfs, Integer.MAX_VALUE).get();
// after compaction cache should have entries for new SSTables,
@ -178,8 +180,7 @@ public class KeyCacheTest
// if we had 2 keys in cache previously it should become 4
assertKeyCacheSize(4, KEYSPACE1, COLUMN_FAMILY1);
for (SSTableReader reader : readers)
reader.releaseReference();
refs.release();
Uninterruptibles.sleepUninterruptibly(10, TimeUnit.MILLISECONDS);;
while (ScheduledExecutors.nonPeriodicTasks.getActiveCount() + ScheduledExecutors.nonPeriodicTasks.getQueue().size() > 0);

26
test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java
View File

@ -17,6 +17,8 @@
*/
package org.apache.cassandra.db.compaction;
import junit.framework.Assert;
import org.apache.cassandra.utils.concurrent.Refs;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@ -89,9 +91,9 @@ public class AntiCompactionTest
Range<Token> range = new Range<Token>(new BytesToken("0".getBytes()), new BytesToken("4".getBytes()));
List<Range<Token>> ranges = Arrays.asList(range);
SSTableReader.acquireReferences(sstables);
Refs<SSTableReader> refs = Refs.ref(sstables);
long repairedAt = 1000;
CompactionManager.instance.performAnticompaction(store, ranges, sstables, repairedAt);
CompactionManager.instance.performAnticompaction(store, ranges, refs, repairedAt);
assertEquals(2, store.getSSTables().size());
int repairedKeys = 0;
@ -119,7 +121,7 @@ public class AntiCompactionTest
for (SSTableReader sstable : store.getSSTables())
{
assertFalse(sstable.isMarkedCompacted());
assertEquals(1, sstable.referenceCount());
assertEquals(1, sstable.sharedRef().globalCount());
}
assertEquals(0, store.getDataTracker().getCompacting().size());
assertEquals(repairedKeys, 4);
@ -137,8 +139,7 @@ public class AntiCompactionTest
long origSize = s.bytesOnDisk();
Range<Token> range = new Range<Token>(new BytesToken(ByteBufferUtil.bytes(0)), new BytesToken(ByteBufferUtil.bytes(500)));
Collection<SSTableReader> sstables = cfs.getSSTables();
SSTableReader.acquireReferences(sstables);
CompactionManager.instance.performAnticompaction(cfs, Arrays.asList(range), sstables, 12345);
CompactionManager.instance.performAnticompaction(cfs, Arrays.asList(range), Refs.tryRef(sstables), 12345);
long sum = 0;
for (SSTableReader x : cfs.getSSTables())
sum += x.bytesOnDisk();
@ -206,9 +207,10 @@ public class AntiCompactionTest
Range<Token> range = new Range<Token>(new BytesToken("0".getBytes()), new BytesToken("4".getBytes()));
List<Range<Token>> ranges = Arrays.asList(range);
SSTableReader.acquireReferences(sstables);
Refs<SSTableReader> refs = Refs.tryRef(sstables);
Assert.assertNotNull(refs);
long repairedAt = 1000;
CompactionManager.instance.performAnticompaction(store, ranges, sstables, repairedAt);
CompactionManager.instance.performAnticompaction(store, ranges, refs, repairedAt);
/*
Anticompaction will be anti-compacting 10 SSTables but will be doing this two at a time
so there will be no net change in the number of sstables
@ -239,6 +241,7 @@ public class AntiCompactionTest
assertEquals(repairedKeys, 40);
assertEquals(nonRepairedKeys, 60);
}
@Test
public void shouldMutateRepairedAt() throws InterruptedException, IOException
{
@ -248,12 +251,11 @@ public class AntiCompactionTest
Range<Token> range = new Range<Token>(new BytesToken("0".getBytes()), new BytesToken("9999".getBytes()));
List<Range<Token>> ranges = Arrays.asList(range);
SSTableReader.acquireReferences(sstables);
CompactionManager.instance.performAnticompaction(store, ranges, sstables, 1);
CompactionManager.instance.performAnticompaction(store, ranges, Refs.tryRef(sstables), 1);
assertThat(store.getSSTables().size(), is(1));
assertThat(Iterables.get(store.getSSTables(), 0).isRepaired(), is(true));
assertThat(Iterables.get(store.getSSTables(), 0).referenceCount(), is(1));
assertThat(Iterables.get(store.getSSTables(), 0).sharedRef().globalCount(), is(1));
assertThat(store.getDataTracker().getCompacting().size(), is(0));
}
@ -275,8 +277,8 @@ public class AntiCompactionTest
Range<Token> range = new Range<Token>(new BytesToken("-10".getBytes()), new BytesToken("-1".getBytes()));
List<Range<Token>> ranges = Arrays.asList(range);
SSTableReader.acquireReferences(sstables);
CompactionManager.instance.performAnticompaction(store, ranges, sstables, 0);
Refs<SSTableReader> refs = Refs.ref(sstables);
CompactionManager.instance.performAnticompaction(store, ranges, refs, 0);
assertThat(store.getSSTables().size(), is(10));
assertThat(Iterables.get(store.getSSTables(), 0).isRepaired(), is(false));

11
test/unit/org/apache/cassandra/db/compaction/BlacklistingCompactionsTest.java
View File

@ -27,6 +27,8 @@ import java.util.HashSet;
import java.util.Set;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@ -48,6 +50,15 @@ public class BlacklistingCompactionsTest
private static final String KEYSPACE1 = "BlacklistingCompactionsTest";
private static final String CF_STANDARD1 = "Standard1";
@After
public void leakDetect() throws InterruptedException
{
System.gc();
System.gc();
System.gc();
Thread.sleep(10);
}
@BeforeClass
public static void defineSchema() throws ConfigurationException
{

2
test/unit/org/apache/cassandra/io/sstable/LegacySSTableTest.java
View File

@ -134,7 +134,7 @@ public class LegacySSTableTest
ranges.add(new Range<>(p.getMinimumToken(), p.getToken(ByteBufferUtil.bytes("100"))));
ranges.add(new Range<>(p.getToken(ByteBufferUtil.bytes("100")), p.getMinimumToken()));
ArrayList<StreamSession.SSTableStreamingSections> details = new ArrayList<>();
details.add(new StreamSession.SSTableStreamingSections(sstable,
details.add(new StreamSession.SSTableStreamingSections(sstable, sstable.tryRef(),
sstable.getPositionsForRanges(ranges),
sstable.estimatedKeysForRanges(ranges), sstable.getSSTableMetadata().repairedAt));
new StreamPlan("LegacyStreamingTest").transferFiles(FBUtilities.getBroadcastAddress(), details)

6
test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
View File

@ -230,8 +230,8 @@ public class SSTableRewriterTest extends SchemaLoader
s.setReplacedBy(s2);
s2.markObsolete();
s.releaseReference();
s2.releaseReference();
s.sharedRef().release();
s2.sharedRef().release();
writer.abort(false);
@ -731,7 +731,7 @@ public class SSTableRewriterTest extends SchemaLoader
for (SSTableReader sstable : cfs.getSSTables())
{
assertFalse(sstable.isMarkedCompacted());
assertEquals(1, sstable.referenceCount());
assertEquals(1, sstable.sharedRef().globalCount());
liveDescriptors.add(sstable.descriptor.generation);
}
for (File dir : cfs.directories.getCFDirectories())

2
test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
View File

@ -80,7 +80,7 @@ public class StreamTransferTaskTest
{
List<Range<Token>> ranges = new ArrayList<>();
ranges.add(new Range<>(sstable.first.getToken(), sstable.last.getToken()));
task.addTransferFile(sstable, 1, sstable.getPositionsForRanges(ranges), 0);
task.addTransferFile(sstable, sstable.sharedRef(), 1, sstable.getPositionsForRanges(ranges), 0);
}
assertEquals(2, task.getTotalNumberOfFiles());

18
test/unit/org/apache/cassandra/streaming/StreamingTransferTest.java
View File

@ -58,6 +58,7 @@ import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.CounterId;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.Refs;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
import static org.apache.cassandra.Util.cellname;
@ -235,15 +236,15 @@ public class StreamingTransferTest
private void transfer(SSTableReader sstable, List<Range<Token>> ranges) throws Exception
{
new StreamPlan("StreamingTransferTest").transferFiles(LOCAL, makeStreamingDetails(ranges, Arrays.asList(sstable))).execute().get();
new StreamPlan("StreamingTransferTest").transferFiles(LOCAL, makeStreamingDetails(ranges, Refs.tryRef(Arrays.asList(sstable)))).execute().get();
}
private Collection<StreamSession.SSTableStreamingSections> makeStreamingDetails(List<Range<Token>> ranges, Collection<SSTableReader> sstables)
private Collection<StreamSession.SSTableStreamingSections> makeStreamingDetails(List<Range<Token>> ranges, Refs<SSTableReader> sstables)
{
ArrayList<StreamSession.SSTableStreamingSections> details = new ArrayList<>();
for (SSTableReader sstable : sstables)
{
details.add(new StreamSession.SSTableStreamingSections(sstable,
details.add(new StreamSession.SSTableStreamingSections(sstable, sstables.get(sstable),
sstable.getPositionsForRanges(ranges),
sstable.estimatedKeysForRanges(ranges), sstable.getSSTableMetadata().repairedAt));
}
@ -407,9 +408,9 @@ public class StreamingTransferTest
ranges.add(new Range<>(p.getMinimumToken(), p.getToken(ByteBufferUtil.bytes("test"))));
ranges.add(new Range<>(p.getToken(ByteBufferUtil.bytes("transfer2")), p.getMinimumToken()));
// Acquiring references, transferSSTables needs it
sstable.acquireReference();
sstable2.acquireReference();
new StreamPlan("StreamingTransferTest").transferFiles(LOCAL, makeStreamingDetails(ranges, Arrays.asList(sstable, sstable2))).execute().get();
Refs<SSTableReader> refs = Refs.tryRef(Arrays.asList(sstable, sstable2));
assert refs != null;
new StreamPlan("StreamingTransferTest").transferFiles(LOCAL, makeStreamingDetails(ranges, refs)).execute().get();
// confirm that the sstables were transferred and registered and that 2 keys arrived
ColumnFamilyStore cfstore = Keyspace.open(keyspaceName).getColumnFamilyStore(cfname);
@ -460,10 +461,11 @@ public class StreamingTransferTest
ranges.add(new Range<>(secondtolast.getKey().getToken(), p.getMinimumToken()));
// Acquiring references, transferSSTables needs it
if (!SSTableReader.acquireReferences(ssTableReaders))
Refs<SSTableReader> refs = Refs.tryRef(ssTableReaders);
if (refs == null)
throw new AssertionError();
new StreamPlan("StreamingTransferTest").transferFiles(LOCAL, makeStreamingDetails(ranges, ssTableReaders)).execute().get();
new StreamPlan("StreamingTransferTest").transferFiles(LOCAL, makeStreamingDetails(ranges, refs)).execute().get();
// check that only two keys were transferred
for (Map.Entry<DecoratedKey,String> entry : Arrays.asList(first, last))

85
test/unit/org/apache/cassandra/utils/concurrent/RefCountedTest.java Normal file
View File

@ -0,0 +1,85 @@
/*
* 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.concurrent;
import org.junit.Test;
import junit.framework.Assert;
public class RefCountedTest
{
private static final class Tidier implements RefCounted.Tidy
{
boolean tidied;
public void tidy()
{
tidied = true;
}
public String name()
{
return "test tidy";
}
}
@Test
public void testLeak() throws InterruptedException
{
Tidier tidier = new Tidier();
RefCounted obj = RefCounted.Impl.get(tidier);
obj.tryRef();
obj.sharedRef().release();
System.gc();
System.gc();
Thread.sleep(1000);
Assert.assertTrue(tidier.tidied);
}
@Test
public void testSeriousLeak() throws InterruptedException
{
Tidier tidier = new Tidier();
RefCounted.Impl.get(tidier);
System.gc();
System.gc();
System.gc();
System.gc();
Thread.sleep(1000);
Assert.assertTrue(tidier.tidied);
}
@Test
public void testDoubleRelease() throws InterruptedException
{
Tidier tidier = null;
try
{
tidier = new Tidier();
RefCounted obj = RefCounted.Impl.get(tidier);
obj.sharedRef().release();
obj.sharedRef().release();
Assert.assertTrue(false);
}
catch (Exception e)
{
}
}
}