Repository: cassandra Updated Branches: refs/heads/trunk d32a745b2 -> 10649b1ce
Fix compaction throttle Patch by tjake reviewed by Marcus Eriksson for CASSANDRA-12366 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/10649b1c Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/10649b1c Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/10649b1c Branch: refs/heads/trunk Commit: 10649b1ce5aba1683e024c0f9cf23bd9c42704b7 Parents: d32a745 Author: T Jake Luciani <j...@apache.org> Authored: Tue Aug 2 12:50:03 2016 -0400 Committer: T Jake Luciani <j...@apache.org> Committed: Wed Aug 10 13:57:18 2016 -0400 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../compaction/AbstractCompactionStrategy.java | 39 ++++++++++++++++++-- .../db/compaction/CompactionController.java | 3 +- .../db/compaction/CompactionManager.java | 19 +++++++++- .../cassandra/db/compaction/CompactionTask.java | 23 +++++++++++- .../compaction/LeveledCompactionStrategy.java | 27 ++++++++++++-- .../cassandra/io/sstable/ISSTableScanner.java | 2 + .../io/sstable/format/big/BigTableScanner.java | 35 ++++++++++++++++++ .../apache/cassandra/metrics/TableMetrics.java | 3 +- .../db/compaction/CompactionIteratorTest.java | 13 +++++++ 10 files changed, 152 insertions(+), 13 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/10649b1c/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 5738c30..5c1d286 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.10 + * Fix compaction throughput throttle (CASSANDRA-12366) * Delay releasing Memtable memory on flush until PostFlush has finished running (CASSANDRA-12358) * Cassandra stress should dump all setting on startup (CASSANDRA-11914) * Make it possible to compact a given token range (CASSANDRA-10643) http://git-wip-us.apache.org/repos/asf/cassandra/blob/10649b1c/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java index 48d5eb5..8454147 100644 --- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java @@ -23,7 +23,6 @@ import com.google.common.base.Throwables; import com.google.common.collect.ImmutableMap; import com.google.common.base.Predicate; import com.google.common.collect.Iterables; -import com.google.common.util.concurrent.RateLimiter; import org.apache.cassandra.db.Directories; import org.apache.cassandra.db.SerializationHeader; @@ -283,12 +282,11 @@ public abstract class AbstractCompactionStrategy @SuppressWarnings("resource") public ScannerList getScanners(Collection<SSTableReader> sstables, Collection<Range<Token>> ranges) { - RateLimiter limiter = CompactionManager.instance.getRateLimiter(); ArrayList<ISSTableScanner> scanners = new ArrayList<ISSTableScanner>(); try { for (SSTableReader sstable : sstables) - scanners.add(sstable.getScanner(ranges, limiter)); + scanners.add(sstable.getScanner(ranges, null)); } catch (Throwable t) { @@ -341,6 +339,41 @@ public abstract class AbstractCompactionStrategy this.scanners = scanners; } + public long getTotalBytesScanned() + { + long bytesScanned = 0L; + for (ISSTableScanner scanner : scanners) + bytesScanned += scanner.getBytesScanned(); + + return bytesScanned; + } + + public long getTotalCompressedSize() + { + long compressedSize = 0; + for (ISSTableScanner scanner : scanners) + compressedSize += scanner.getCompressedLengthInBytes(); + + return compressedSize; + } + + public double getCompressionRatio() + { + double compressed = 0.0; + double uncompressed = 0.0; + + for (ISSTableScanner scanner : scanners) + { + compressed += scanner.getCompressedLengthInBytes(); + uncompressed += scanner.getLengthInBytes(); + } + + if (compressed == uncompressed || uncompressed == 0) + return MetadataCollector.NO_COMPRESSION_RATIO; + + return compressed / uncompressed; + } + public void close() { Throwable t = null; http://git-wip-us.apache.org/repos/asf/cassandra/blob/10649b1c/src/java/org/apache/cassandra/db/compaction/CompactionController.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionController.java b/src/java/org/apache/cassandra/db/compaction/CompactionController.java index d922d28..b34eee6 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionController.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionController.java @@ -72,8 +72,7 @@ public class CompactionController implements AutoCloseable public CompactionController(ColumnFamilyStore cfs, Set<SSTableReader> compacting, int gcBefore) { - this(cfs, compacting, gcBefore, - CompactionManager.instance.getRateLimiter(), + this(cfs, compacting, gcBefore, null, cfs.getCompactionStrategyManager().getCompactionParams().tombstoneOption()); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/10649b1c/src/java/org/apache/cassandra/db/compaction/CompactionManager.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index ac6c753..f3a5711 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -30,6 +30,7 @@ import javax.management.openmbean.TabularData; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.*; +import com.google.common.primitives.Ints; import com.google.common.util.concurrent.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -483,7 +484,7 @@ public class CompactionManager implements CompactionManagerMBean @Override protected CompactionController getCompactionController(Set<SSTableReader> toCompact) { - return new CompactionController(cfStore, toCompact, gcBefore, getRateLimiter(), tombstoneOption); + return new CompactionController(cfStore, toCompact, gcBefore, null, tombstoneOption); } }; task.setUserDefined(true); @@ -1079,15 +1080,22 @@ public class CompactionManager implements CompactionManagerMBean logger.info("Cleaning up {}", sstable); File compactionFileLocation = sstable.descriptor.directory; + RateLimiter limiter = getRateLimiter(); + double compressionRatio = sstable.getCompressionRatio(); + if (compressionRatio == MetadataCollector.NO_COMPRESSION_RATIO) + compressionRatio = 1.0; List<SSTableReader> finished; + int nowInSec = FBUtilities.nowInSeconds(); try (SSTableRewriter writer = SSTableRewriter.construct(cfs, txn, false, sstable.maxDataAge); - ISSTableScanner scanner = cleanupStrategy.getScanner(sstable, getRateLimiter()); + ISSTableScanner scanner = cleanupStrategy.getScanner(sstable, null); CompactionController controller = new CompactionController(cfs, txn.originals(), getDefaultGcBefore(cfs, nowInSec)); CompactionIterator ci = new CompactionIterator(OperationType.CLEANUP, Collections.singletonList(scanner), controller, nowInSec, UUIDGen.getTimeUUID(), metrics)) { writer.switchWriter(createWriter(cfs, compactionFileLocation, expectedBloomFilterSize, sstable.getSSTableMetadata().repairedAt, sstable, txn)); + long lastBytesScanned = 0; + while (ci.hasNext()) { @@ -1102,6 +1110,13 @@ public class CompactionManager implements CompactionManagerMBean if (writer.append(notCleaned) != null) totalkeysWritten++; + + long bytesScanned = scanner.getBytesScanned(); + + int lengthRead = (int) (Ints.checkedCast(bytesScanned - lastBytesScanned) * compressionRatio); + limiter.acquire(lengthRead + 1); + + lastBytesScanned = bytesScanned; } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/10649b1c/src/java/org/apache/cassandra/db/compaction/CompactionTask.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java index 86c8a8f..43a726b 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java @@ -27,6 +27,8 @@ import java.util.concurrent.TimeUnit; import com.google.common.base.Predicate; import com.google.common.collect.Iterables; import com.google.common.collect.Sets; +import com.google.common.primitives.Ints; +import com.google.common.util.concurrent.RateLimiter; import org.apache.cassandra.db.Directories; import org.apache.cassandra.db.compaction.writers.CompactionAwareWriter; @@ -41,6 +43,7 @@ import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.db.compaction.CompactionManager.CompactionExecutorStatsCollector; import org.apache.cassandra.db.lifecycle.LifecycleTransaction; +import org.apache.cassandra.io.sstable.metadata.MetadataCollector; import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.concurrent.Refs; @@ -149,14 +152,15 @@ public class CompactionTask extends AbstractCompactionTask logger.debug("Compacting ({}) {}", taskId, ssTableLoggerMsg); + RateLimiter limiter = CompactionManager.instance.getRateLimiter(); long start = System.nanoTime(); long startTime = System.currentTimeMillis(); long totalKeysWritten = 0; long estimatedKeys = 0; + long inputSizeBytes; try (CompactionController controller = getCompactionController(transaction.originals())) { Set<SSTableReader> actuallyCompact = Sets.difference(transaction.originals(), controller.getFullyExpiredSSTables()); - Collection<SSTableReader> newSStables; long[] mergedRowCounts; @@ -173,6 +177,12 @@ public class CompactionTask extends AbstractCompactionTask if (collector != null) collector.beginCompaction(ci); long lastCheckObsoletion = start; + inputSizeBytes = scanners.getTotalCompressedSize(); + double compressionRatio = scanners.getCompressionRatio(); + if (compressionRatio == MetadataCollector.NO_COMPRESSION_RATIO) + compressionRatio = 1.0; + + long lastBytesScanned = 0; if (!controller.cfs.getCompactionStrategyManager().isActive()) throw new CompactionInterruptedException(ci.getCompactionInfo()); @@ -188,6 +198,15 @@ public class CompactionTask extends AbstractCompactionTask if (writer.append(ci.next())) totalKeysWritten++; + + long bytesScanned = scanners.getTotalBytesScanned(); + + //Rate limit the scanners, and account for compression + int lengthRead = (int) (Ints.checkedCast(bytesScanned - lastBytesScanned) * compressionRatio); + limiter.acquire(lengthRead + 1); + + lastBytesScanned = bytesScanned; + if (System.nanoTime() - lastCheckObsoletion > TimeUnit.MINUTES.toNanos(1L)) { controller.maybeRefreshOverlaps(); @@ -213,7 +232,7 @@ public class CompactionTask extends AbstractCompactionTask long durationInNano = System.nanoTime() - start; long dTime = TimeUnit.NANOSECONDS.toMillis(durationInNano); - long startsize = SSTableReader.getTotalBytes(transaction.originals()); + long startsize = inputSizeBytes; long endsize = SSTableReader.getTotalBytes(newSStables); double ratio = (double) endsize / (double) startsize; http://git-wip-us.apache.org/repos/asf/cassandra/blob/10649b1c/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java index 287e387..c224f91 100644 --- a/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/LeveledCompactionStrategy.java @@ -265,7 +265,7 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy { // L0 makes no guarantees about overlapping-ness. Just create a direct scanner for each for (SSTableReader sstable : byLevel.get(level)) - scanners.add(sstable.getScanner(ranges, CompactionManager.instance.getRateLimiter())); + scanners.add(sstable.getScanner(ranges, null)); } else { @@ -322,9 +322,12 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy private final List<SSTableReader> sstables; private final Iterator<SSTableReader> sstableIterator; private final long totalLength; + private final long compressedLength; private ISSTableScanner currentScanner; private long positionOffset; + private SSTableReader currentSSTable; + private long totalBytesScanned = 0; public LeveledScanner(Collection<SSTableReader> sstables, Collection<Range<Token>> ranges) { @@ -333,6 +336,7 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy // add only sstables that intersect our range, and estimate how much data that involves this.sstables = new ArrayList<>(sstables.size()); long length = 0; + long cLength = 0; for (SSTableReader sstable : sstables) { this.sstables.add(sstable); @@ -343,13 +347,17 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy estKeysInRangeRatio = ((double) sstable.estimatedKeysForRanges(ranges)) / estimatedKeys; length += sstable.uncompressedLength() * estKeysInRangeRatio; + cLength += sstable.onDiskLength() * estKeysInRangeRatio; } totalLength = length; + compressedLength = cLength; Collections.sort(this.sstables, SSTableReader.sstableComparator); sstableIterator = this.sstables.iterator(); assert sstableIterator.hasNext(); // caller should check intersecting first - currentScanner = sstableIterator.next().getScanner(ranges, CompactionManager.instance.getRateLimiter()); + currentSSTable = sstableIterator.next(); + currentScanner = currentSSTable.getScanner(ranges, null); + } public static Collection<SSTableReader> intersecting(Collection<SSTableReader> sstables, Collection<Range<Token>> ranges) @@ -392,6 +400,8 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy return currentScanner.next(); positionOffset += currentScanner.getLengthInBytes(); + totalBytesScanned += currentScanner.getBytesScanned(); + currentScanner.close(); if (!sstableIterator.hasNext()) { @@ -399,7 +409,8 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy currentScanner = null; return endOfData(); } - currentScanner = sstableIterator.next().getScanner(ranges, CompactionManager.instance.getRateLimiter()); + currentSSTable = sstableIterator.next(); + currentScanner = currentSSTable.getScanner(ranges, null); } } @@ -419,6 +430,16 @@ public class LeveledCompactionStrategy extends AbstractCompactionStrategy return positionOffset + (currentScanner == null ? 0L : currentScanner.getCurrentPosition()); } + public long getCompressedLengthInBytes() + { + return compressedLength; + } + + public long getBytesScanned() + { + return currentScanner == null ? totalBytesScanned : totalBytesScanned + currentScanner.getBytesScanned(); + } + public String getBackingFiles() { return Joiner.on(", ").join(sstables); http://git-wip-us.apache.org/repos/asf/cassandra/blob/10649b1c/src/java/org/apache/cassandra/io/sstable/ISSTableScanner.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/io/sstable/ISSTableScanner.java b/src/java/org/apache/cassandra/io/sstable/ISSTableScanner.java index 7063057..2dff34e 100644 --- a/src/java/org/apache/cassandra/io/sstable/ISSTableScanner.java +++ b/src/java/org/apache/cassandra/io/sstable/ISSTableScanner.java @@ -28,6 +28,8 @@ import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator; public interface ISSTableScanner extends UnfilteredPartitionIterator { public long getLengthInBytes(); + public long getCompressedLengthInBytes(); public long getCurrentPosition(); + public long getBytesScanned(); public String getBackingFiles(); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/10649b1c/src/java/org/apache/cassandra/io/sstable/format/big/BigTableScanner.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableScanner.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableScanner.java index 66213a6..e465a02 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableScanner.java +++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableScanner.java @@ -62,6 +62,8 @@ public class BigTableScanner implements ISSTableScanner private final DataRange dataRange; private final RowIndexEntry.IndexSerializer rowIndexEntrySerializer; private final boolean isForThrift; + private long startScan = -1; + private long bytesScanned = 0; protected Iterator<UnfilteredRowIterator> iterator; @@ -223,6 +225,16 @@ public class BigTableScanner implements ISSTableScanner return dfile.getFilePointer(); } + public long getBytesScanned() + { + return bytesScanned; + } + + public long getCompressedLengthInBytes() + { + return sstable.onDiskLength(); + } + public String getBackingFiles() { return sstable.toString(); @@ -277,12 +289,16 @@ public class BigTableScanner implements ISSTableScanner { do { + if (startScan != -1) + bytesScanned += dfile.getFilePointer() - startScan; + // we're starting the first range or we just passed the end of the previous range if (!rangeIterator.hasNext()) return endOfData(); currentRange = rangeIterator.next(); seekToCurrentRangeStart(); + startScan = dfile.getFilePointer(); if (ifile.isEOF()) return endOfData(); @@ -325,14 +341,23 @@ public class BigTableScanner implements ISSTableScanner { protected UnfilteredRowIterator initializeIterator() { + + if (startScan != -1) + bytesScanned += dfile.getFilePointer() - startScan; + try { if (dataRange == null) { dfile.seek(currentEntry.position); + startScan = dfile.getFilePointer(); ByteBufferUtil.skipShortLength(dfile); // key return SSTableIdentityIterator.create(sstable, dfile, partitionKey()); } + else + { + startScan = dfile.getFilePointer(); + } ClusteringIndexFilter filter = dataRange.clusteringIndexFilter(partitionKey()); return sstable.iterator(dfile, partitionKey(), currentEntry, filter.getSlices(BigTableScanner.this.metadata()), columns, filter.isReversed(), isForThrift); @@ -382,6 +407,16 @@ public class BigTableScanner implements ISSTableScanner return 0; } + public long getBytesScanned() + { + return 0; + } + + public long getCompressedLengthInBytes() + { + return 0; + } + public String getBackingFiles() { return sstable.getFilename(); http://git-wip-us.apache.org/repos/asf/cassandra/blob/10649b1c/src/java/org/apache/cassandra/metrics/TableMetrics.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/metrics/TableMetrics.java b/src/java/org/apache/cassandra/metrics/TableMetrics.java index 7889077..d1d4a3d 100644 --- a/src/java/org/apache/cassandra/metrics/TableMetrics.java +++ b/src/java/org/apache/cassandra/metrics/TableMetrics.java @@ -36,6 +36,7 @@ import org.apache.cassandra.db.lifecycle.SSTableSet; import org.apache.cassandra.index.SecondaryIndexManager; import org.apache.cassandra.io.compress.CompressionMetadata; import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.io.sstable.metadata.MetadataCollector; import org.apache.cassandra.repair.SystemDistributedKeyspace; import org.apache.cassandra.utils.EstimatedHistogram; import org.apache.cassandra.utils.TopKSampler; @@ -819,7 +820,7 @@ public class TableMetrics dataLengthSum += compressionMetadata.dataLength; } } - return dataLengthSum != 0 ? compressedLengthSum / dataLengthSum : 0; + return dataLengthSum != 0 ? compressedLengthSum / dataLengthSum : MetadataCollector.NO_COMPRESSION_RATIO; } /** http://git-wip-us.apache.org/repos/asf/cassandra/blob/10649b1c/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java index 2189e15..129d1e7 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java @@ -39,6 +39,7 @@ import org.apache.cassandra.db.marshal.UTF8Type; import org.apache.cassandra.db.partitions.AbstractUnfilteredPartitionIterator; import org.apache.cassandra.db.rows.*; import org.apache.cassandra.io.sstable.ISSTableScanner; +import org.apache.cassandra.io.sstable.metadata.MetadataCollector; import org.apache.cassandra.schema.KeyspaceParams; public class CompactionIteratorTest @@ -369,6 +370,18 @@ public class CompactionIteratorTest } @Override + public long getBytesScanned() + { + return 0; + } + + @Override + public long getCompressedLengthInBytes() + { + return 0; + } + + @Override public String getBackingFiles() { return null;
