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

fix compaction throttling bursty-ness 

patch by yukim and jbellis for CASSANDRA-4316
This commit is contained in:
Jonathan Ellis 2013-04-30 14:09:25 -05:00
parent 927c4a4f98
commit 2b0797b24e
16 changed files with 151 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
1.2.5
* fix compaction throttling bursty-ness (CASSANDRA-4316)
* reduce memory consumption of IndexSummary (CASSANDRA-5506)
* remove per-row column name bloom filters (CASSANDRA-5492)
* Include fatal errors in trace events (CASSANDRA-5447)

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

@ -19,6 +19,7 @@ package org.apache.cassandra.db.compaction;
import java.util.*;
import com.google.common.util.concurrent.RateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -150,9 +151,10 @@ public abstract class AbstractCompactionStrategy
*/
public List<ICompactionScanner> getScanners(Collection<SSTableReader> sstables, Range<Token> range)
{
RateLimiter limiter = CompactionManager.instance.getRateLimiter();
ArrayList<ICompactionScanner> scanners = new ArrayList<ICompactionScanner>();
for (SSTableReader sstable : sstables)
scanners.add(sstable.getDirectScanner(range));
scanners.add(sstable.getDirectScanner(range, limiter));
return scanners;
}

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

@ -22,6 +22,8 @@ import java.util.Collections;
import java.util.List;
import java.util.Set;
import com.google.common.primitives.Ints;
import com.google.common.util.concurrent.RateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -36,6 +38,8 @@ import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.AlwaysPresentFilter;
import org.apache.cassandra.utils.Throttle;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.Throttle;
/**
* Manage compaction options.
@ -50,20 +54,6 @@ public class CompactionController
public final int gcBefore;
public final int mergeShardBefore;
private final Throttle throttle = new Throttle("Cassandra_Throttle", new Throttle.ThroughputFunction()
{
/** @return Instantaneous throughput target in bytes per millisecond. */
public int targetThroughput()
{
if (DatabaseDescriptor.getCompactionThroughputMbPerSec() < 1 || StorageService.instance.isBootstrapMode())
// throttling disabled
return 0;
// total throughput
int totalBytesPerMS = DatabaseDescriptor.getCompactionThroughputMbPerSec() * 1024 * 1024 / 1000;
// per stream throughput (target bytes per MS)
return totalBytesPerMS / Math.max(1, CompactionManager.instance.getActiveCompactions());
}
});
public CompactionController(ColumnFamilyStore cfs, Collection<SSTableReader> sstables, int gcBefore)
{
@ -176,11 +166,6 @@ public class CompactionController
return getCompactedRow(Collections.singletonList(row));
}
public void mayThrottle(long currentBytes)
{
throttle.throttle(currentBytes);
}
public void close()
{
SSTableReader.releaseReferences(overlappingSSTables);

12
src/java/org/apache/cassandra/db/compaction/CompactionIterable.java
View File

@ -78,14 +78,10 @@ public class CompactionIterable extends AbstractCompactionIterable
finally
{
rows.clear();
if ((row++ % 1000) == 0)
{
long n = 0;
for (ICompactionScanner scanner : scanners)
n += scanner.getCurrentPosition();
bytesRead = n;
controller.mayThrottle(bytesRead);
}
long n = 0;
for (ICompactionScanner scanner : scanners)
n += scanner.getCurrentPosition();
bytesRead = n;
}
}
}

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

@ -31,6 +31,7 @@ import com.google.common.base.Throwables;
import com.google.common.collect.ConcurrentHashMultiset;
import com.google.common.collect.Multiset;
import com.google.common.primitives.Longs;
import com.google.common.util.concurrent.RateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -118,6 +119,26 @@ public class CompactionManager implements CompactionManagerMBean
private final CompactionMetrics metrics = new CompactionMetrics(executor, validationExecutor);
private final Multiset<ColumnFamilyStore> compactingCF = ConcurrentHashMultiset.create();
private final RateLimiter compactionRateLimiter = RateLimiter.create(Double.MAX_VALUE);
/**
* Gets compaction rate limiter. When compaction_throughput_mb_per_sec is 0 or node is bootstrapping,
* this returns rate limiter with the rate of Double.MAX_VALUE bytes per second.
* Rate unit is bytes per sec.
*
* @return RateLimiter with rate limit set
*/
public RateLimiter getRateLimiter()
{
double currentThroughput = DatabaseDescriptor.getCompactionThroughputMbPerSec() * 1024 * 1024;
// if throughput is set to 0, throttling is disabled
if (currentThroughput == 0 || StorageService.instance.isBootstrapMode())
currentThroughput = Double.MAX_VALUE;
if (compactionRateLimiter.getRate() != currentThroughput)
compactionRateLimiter.setRate(currentThroughput);
return compactionRateLimiter;
}
/**
* @return A lock, for which acquisition means no compactions can run.
*/
@ -568,7 +589,7 @@ public class CompactionManager implements CompactionManagerMBean
if (compactionFileLocation == null)
throw new IOException("disk full");
SSTableScanner scanner = sstable.getDirectScanner();
SSTableScanner scanner = sstable.getDirectScanner(getRateLimiter());
long rowsRead = 0;
List<IColumn> indexedColumnsInRow = null;
@ -628,8 +649,6 @@ public class CompactionManager implements CompactionManagerMBean
}
}
}
if ((rowsRead++ % 1000) == 0)
controller.mayThrottle(scanner.getCurrentPosition());
}
if (writer != null)
newSstable = writer.closeAndOpenReader(sstable.maxDataAge);

7
src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java
View File

@ -24,6 +24,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
import com.google.common.collect.*;
import com.google.common.primitives.Doubles;
import com.google.common.util.concurrent.RateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -178,7 +179,7 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy implem
{
// L0 makes no guarantees about overlapping-ness. Just create a direct scanner for each
for (SSTableReader sstable : byLevel.get(level))
scanners.add(sstable.getDirectScanner(range));
scanners.add(sstable.getDirectScanner(range, CompactionManager.instance.getRateLimiter()));
}
else
{
@ -208,7 +209,7 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy implem
this.sstables = new ArrayList<SSTableReader>(sstables);
Collections.sort(this.sstables, SSTable.sstableComparator);
sstableIterator = this.sstables.iterator();
currentScanner = sstableIterator.next().getDirectScanner(range);
currentScanner = sstableIterator.next().getDirectScanner(range, CompactionManager.instance.getRateLimiter());
long length = 0;
for (SSTableReader sstable : sstables)
@ -233,7 +234,7 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy implem
currentScanner = null;
return endOfData();
}
currentScanner = sstableIterator.next().getDirectScanner(range);
currentScanner = sstableIterator.next().getDirectScanner(range, CompactionManager.instance.getRateLimiter());
}
}
catch (IOException e)

13
src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java
View File

@ -117,7 +117,6 @@ public class ParallelCompactionIterable extends AbstractCompactionIterable
private class Reducer extends MergeIterator.Reducer<RowContainer, CompactedRowContainer>
{
private final List<RowContainer> rows = new ArrayList<RowContainer>();
private int row = 0;
private final ThreadPoolExecutor executor = new DebuggableThreadPoolExecutor(FBUtilities.getAvailableProcessors(),
Integer.MAX_VALUE,
@ -137,14 +136,10 @@ public class ParallelCompactionIterable extends AbstractCompactionIterable
ParallelCompactionIterable.this.updateCounterFor(rows.size());
CompactedRowContainer compacted = getCompactedRow(rows);
rows.clear();
if ((row++ % 1000) == 0)
{
long n = 0;
for (ICompactionScanner scanner : scanners)
n += scanner.getCurrentPosition();
bytesRead = n;
controller.mayThrottle(bytesRead);
}
long n = 0;
for (ICompactionScanner scanner : scanners)
n += scanner.getCurrentPosition();
bytesRead = n;
return compacted;
}

8
src/java/org/apache/cassandra/db/compaction/Scrubber.java
View File

@ -45,8 +45,6 @@ public class Scrubber implements Closeable
private final RandomAccessReader indexFile;
private final ScrubInfo scrubInfo;
private long rowsRead;
private SSTableWriter writer;
private SSTableReader newSstable;
private SSTableReader newInOrderSstable;
@ -94,7 +92,9 @@ public class Scrubber implements Closeable
// we'll also loop through the index at the same time, using the position from the index to recover if the
// row header (key or data size) is corrupt. (This means our position in the index file will be one row
// "ahead" of the data file.)
this.dataFile = sstable.openDataReader(true);
this.dataFile = isOffline
? sstable.openDataReader(true)
: sstable.openDataReader(CompactionManager.instance.getRateLimiter());
this.indexFile = RandomAccessReader.open(new File(sstable.descriptor.filenameFor(Component.PRIMARY_INDEX)), true);
this.scrubInfo = new ScrubInfo(dataFile, sstable);
}
@ -249,8 +249,6 @@ public class Scrubber implements Closeable
badRows++;
}
}
if ((rowsRead++ % 1000) == 0)
controller.mayThrottle(dataFile.getFilePointer());
}
if (writer.getFilePointer() > 0)

2
src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java
View File

@ -70,7 +70,7 @@ public class CompressedRandomAccessReader extends RandomAccessReader
// raw checksum bytes
private final ByteBuffer checksumBytes = ByteBuffer.wrap(new byte[4]);
private CompressedRandomAccessReader(String dataFilePath, CompressionMetadata metadata, boolean skipIOCache, PoolingSegmentedFile owner) throws FileNotFoundException
protected CompressedRandomAccessReader(String dataFilePath, CompressionMetadata metadata, boolean skipIOCache, PoolingSegmentedFile owner) throws FileNotFoundException
{
super(new File(dataFilePath), metadata.chunkLength(), skipIOCache, owner);
this.metadata = metadata;

38
src/java/org/apache/cassandra/io/compress/CompressedThrottledReader.java Normal file
View File

@ -0,0 +1,38 @@
package org.apache.cassandra.io.compress;
import java.io.File;
import java.io.FileNotFoundException;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.cassandra.io.util.PoolingSegmentedFile;
import org.apache.cassandra.io.util.RandomAccessReader;
public class CompressedThrottledReader extends CompressedRandomAccessReader
{
private final RateLimiter limiter;
public CompressedThrottledReader(String file, CompressionMetadata metadata, RateLimiter limiter) throws FileNotFoundException
{
super(file, metadata, true, null);
this.limiter = limiter;
}
protected void reBuffer()
{
limiter.acquire(buffer.length);
super.reBuffer();
}
public static CompressedThrottledReader open(String file, CompressionMetadata metadata, RateLimiter limiter)
{
try
{
return new CompressedThrottledReader(file, metadata, limiter);
}
catch (FileNotFoundException e)
{
throw new RuntimeException(e);
}
}
}

6
src/java/org/apache/cassandra/io/sstable/SSTableBoundedScanner.java
View File

@ -20,6 +20,8 @@ package org.apache.cassandra.io.sstable;
import java.util.Arrays;
import java.util.Iterator;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.cassandra.db.columniterator.OnDiskAtomIterator;
import org.apache.cassandra.db.RowPosition;
import org.apache.cassandra.utils.Pair;
@ -32,9 +34,9 @@ public class SSTableBoundedScanner extends SSTableScanner
private final Iterator<Pair<Long, Long>> rangeIterator;
private Pair<Long, Long> currentRange;
SSTableBoundedScanner(SSTableReader sstable, boolean skipCache, Iterator<Pair<Long, Long>> rangeIterator)
SSTableBoundedScanner(SSTableReader sstable, boolean skipCache, Iterator<Pair<Long, Long>> rangeIterator, RateLimiter limiter)
{
super(sstable, skipCache);
super(sstable, skipCache, limiter);
this.rangeIterator = rangeIterator;
assert rangeIterator.hasNext(); // use EmptyCompactionScanner otherwise
currentRange = rangeIterator.next();

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

@ -24,6 +24,8 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.*;
import com.google.common.primitives.Longs;
import com.google.common.util.concurrent.RateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -46,6 +48,7 @@ import org.apache.cassandra.dht.LocalPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.compress.CompressedRandomAccessReader;
import org.apache.cassandra.io.compress.CompressedThrottledReader;
import org.apache.cassandra.io.compress.CompressionMetadata;
import org.apache.cassandra.io.util.*;
import org.apache.cassandra.service.CacheService;
@ -946,10 +949,10 @@ public class SSTableReader extends SSTable
* Direct I/O SSTableScanner
* @return A Scanner for seeking over the rows of the SSTable.
*/
public SSTableScanner getDirectScanner()
{
return new SSTableScanner(this, true);
}
public SSTableScanner getDirectScanner(RateLimiter limiter)
{
return new SSTableScanner(this, true, limiter);
}
/**
* Direct I/O SSTableScanner over a defined range of tokens.
@ -957,14 +960,14 @@ public class SSTableReader extends SSTable
* @param range the range of keys to cover
* @return A Scanner for seeking over the rows of the SSTable.
*/
public ICompactionScanner getDirectScanner(Range<Token> range)
public ICompactionScanner getDirectScanner(Range<Token> range, RateLimiter limiter)
{
if (range == null)
return getDirectScanner();
return getDirectScanner(limiter);
Iterator<Pair<Long, Long>> rangeIterator = getPositionsForRanges(Collections.singletonList(range)).iterator();
return rangeIterator.hasNext()
? new SSTableBoundedScanner(this, true, rangeIterator)
? new SSTableBoundedScanner(this, true, rangeIterator, limiter)
: new EmptyCompactionScanner(getFilename());
}
@ -1117,6 +1120,14 @@ public class SSTableReader extends SSTable
return sstableMetadata.ancestors;
}
public RandomAccessReader openDataReader(RateLimiter limiter)
{
assert limiter != null;
return compression
? CompressedThrottledReader.open(getFilename(), getCompressionMetadata(), limiter)
: ThrottledReader.open(new File(getFilename()), limiter);
}
public RandomAccessReader openDataReader(boolean skipIOCache)
{
return compression

7
src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
View File

@ -21,6 +21,8 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.Iterator;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.RowIndexEntry;
import org.apache.cassandra.db.RowPosition;
@ -45,10 +47,11 @@ public class SSTableScanner implements ICompactionScanner
/**
* @param sstable SSTable to scan.
* @param limiter
*/
SSTableScanner(SSTableReader sstable, boolean skipCache)
SSTableScanner(SSTableReader sstable, boolean skipCache, RateLimiter limiter)
{
this.dfile = sstable.openDataReader(skipCache);
this.dfile = limiter == null ? sstable.openDataReader(skipCache) : sstable.openDataReader(limiter);
this.ifile = sstable.openIndexReader(skipCache);
this.sstable = sstable;
this.filter = null;

35
src/java/org/apache/cassandra/io/util/ThrottledReader.java Normal file
View File

@ -0,0 +1,35 @@
package org.apache.cassandra.io.util;
import java.io.File;
import java.io.FileNotFoundException;
import com.google.common.util.concurrent.RateLimiter;
public class ThrottledReader extends RandomAccessReader
{
private final RateLimiter limiter;
protected ThrottledReader(File file, RateLimiter limiter) throws FileNotFoundException
{
super(file, RandomAccessReader.DEFAULT_BUFFER_SIZE, true, null);
this.limiter = limiter;
}
protected void reBuffer()
{
limiter.acquire(buffer.length);
super.reBuffer();
}
public static ThrottledReader open(File file, RateLimiter limiter)
{
try
{
return new ThrottledReader(file, limiter);
}
catch (FileNotFoundException e)
{
throw new RuntimeException(e);
}
}
}

5
src/java/org/apache/cassandra/tools/SSTableExport.java
View File

@ -32,6 +32,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Set;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.Option;
@ -349,7 +350,7 @@ public class SSTableExport
public static void export(Descriptor desc, PrintStream outs, Collection<String> toExport, String[] excludes) throws IOException
{
SSTableReader reader = SSTableReader.open(desc);
SSTableScanner scanner = reader.getDirectScanner();
SSTableScanner scanner = reader.getDirectScanner(null);
IPartitioner<?> partitioner = reader.partitioner;
@ -406,7 +407,7 @@ public class SSTableExport
SSTableIdentityIterator row;
SSTableScanner scanner = reader.getDirectScanner();
SSTableScanner scanner = reader.getDirectScanner(null);
outs.println("[");

4
test/unit/org/apache/cassandra/io/sstable/SSTableUtils.java
View File

@ -72,8 +72,8 @@ public class SSTableUtils
public static void assertContentEquals(SSTableReader lhs, SSTableReader rhs) throws IOException
{
SSTableScanner slhs = lhs.getDirectScanner();
SSTableScanner srhs = rhs.getDirectScanner();
SSTableScanner slhs = lhs.getDirectScanner(null);
SSTableScanner srhs = rhs.getDirectScanner(null);
while (slhs.hasNext())
{
OnDiskAtomIterator ilhs = slhs.next();