diff --git a/CHANGES.txt b/CHANGES.txt index 095a90d64d19..0ae7fed53035 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha3 + * Give each compressed scan reader its own read-ahead buffer and keep the chunk cache for partition reads while one-shot scans bypass it (CASSANDRA-21671) * Report the number of rows replica filtering protection cached when it warns (CASSANDRA-21623) * Fix non-printable characters in Gossiper log for TOKENS (CASSANDRA-21417) * Fix compression dictionary training failing with "insufficient samples" on large-chunk tables (CASSANDRA-21666) diff --git a/src/java/org/apache/cassandra/cache/ChunkCache.java b/src/java/org/apache/cassandra/cache/ChunkCache.java index 66782ac48bae..c8c966b7d246 100644 --- a/src/java/org/apache/cassandra/cache/ChunkCache.java +++ b/src/java/org/apache/cassandra/cache/ChunkCache.java @@ -37,6 +37,7 @@ import org.apache.cassandra.io.util.ChannelProxy; import org.apache.cassandra.io.util.ChunkReader; import org.apache.cassandra.io.util.FileHandle; +import org.apache.cassandra.io.util.ReadPattern; import org.apache.cassandra.io.util.Rebufferer; import org.apache.cassandra.io.util.RebuffererFactory; import org.apache.cassandra.metrics.ChunkCacheMetrics; @@ -258,9 +259,15 @@ public void invalidate(long position) } @Override - public Rebufferer instantiateRebufferer(boolean isScan) + public Rebufferer instantiateRebufferer(ReadPattern pattern) { - return this; + // A SCAN (compaction, cursor compaction) reads each chunk once and must not pollute the cache with + // one-shot chunks that evict hot data. It does not use the cache, so delegate to the source: the scan + // bypasses the cache and uses its own read-ahead buffer instead. For an Mmap source this also bypasses + // the chunk cache, which is correct: mmap data already lives in the OS page cache, so the chunk cache + // only duplicates it. A PARTITION_READ still uses the cache, because a repeated partition-range query + // re-reads hot data. See CASSANDRA-21671. + return pattern.usesCache() ? this : source.instantiateRebufferer(pattern); } @Override diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java index ab3171200f17..284e3f05cfac 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java @@ -97,6 +97,7 @@ import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.io.util.FileUtils.DuplicateHardlinkException; import org.apache.cassandra.io.util.RandomAccessReader; +import org.apache.cassandra.io.util.ReadPattern; import org.apache.cassandra.metrics.RestorableMeter; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.schema.TableMetadataRef; @@ -1421,43 +1422,51 @@ public StatsMetadata getSSTableMetadata() public RandomAccessReader openDataReader() { - return openDataReaderInternal(null, null, false); + return openDataReaderInternal(null, null, ReadPattern.ROW_READ); } public RandomAccessReader openDataReader(RateLimiter limiter) { assert limiter != null; - return openDataReaderInternal(null, limiter, false); + return openDataReaderInternal(null, limiter, ReadPattern.ROW_READ); } public RandomAccessReader openDataReader(DiskAccessMode diskAccessMode) { - return openDataReaderInternal(diskAccessMode, null, false); + return openDataReaderInternal(diskAccessMode, null, ReadPattern.ROW_READ); } - public RandomAccessReader openDataReaderForScan() + /** + * A reader for a query that walks a range of partitions, such as a token-range query. It reads in order, but + * keeps the chunk cache because a repeated partition-range query re-reads hot data. See CASSANDRA-21671. + */ + public RandomAccessReader openDataReaderForPartitionRead() { - return openDataReaderInternal(null, null, true); + return openDataReaderInternal(null, null, ReadPattern.PARTITION_READ); } + /** + * A reader for a one-shot scan (compaction and similar). It reads each chunk once, so it bypasses the chunk + * cache and uses its own read-ahead buffer. See CASSANDRA-21671. + */ public RandomAccessReader openDataReaderForScan(DiskAccessMode diskAccessMode) { - return openDataReaderInternal(diskAccessMode, null, true); + return openDataReaderInternal(diskAccessMode, null, ReadPattern.SCAN); } private RandomAccessReader openDataReaderInternal(@Nullable DiskAccessMode diskAccessMode, @Nullable RateLimiter limiter, - boolean forScan) + ReadPattern pattern) { if (canReuseDfile(diskAccessMode)) - return dfile.createReader(limiter, forScan, OnReaderClose.RETAIN_FILE_OPEN); + return dfile.createReader(limiter, pattern, OnReaderClose.RETAIN_FILE_OPEN); FileHandle handle = dfile.toBuilder() .withDiskAccessMode(diskAccessMode) .complete(); try { - return handle.createReader(limiter, forScan, OnReaderClose.CLOSE_FILE); + return handle.createReader(limiter, pattern, OnReaderClose.CLOSE_FILE); } catch (Throwable t) { diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableScanner.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableScanner.java index 0b54fec2c5da..9089b9b8bc21 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableScanner.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableScanner.java @@ -75,7 +75,7 @@ protected SSTableScanner(S sstable, { assert sstable != null; - this.dfile = sstable.openDataReaderForScan(); + this.dfile = sstable.openDataReaderForPartitionRead(); this.sstable = sstable; this.columns = columns; this.dataRange = dataRange; diff --git a/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java b/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java index 034fbb71532f..45dbae729e42 100644 --- a/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java +++ b/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java @@ -39,6 +39,10 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl final CompressionMetadata metadata; final int maxCompressedLength; final DoubleSupplier crcCheckChanceSupplier; + // Read-ahead is on only when a scan buffer is configured and larger than one chunk; a smaller buffer cannot + // batch reads, so it adds no value. A value of 0 means "no read-ahead". A per-scan view (see forScan) never + // reads ahead itself, so it always reports 0. + final int readAheadBufferSize; protected CompressedChunkReader(ChannelProxy channel, CompressionMetadata metadata, DoubleSupplier crcCheckChanceSupplier) { @@ -46,9 +50,22 @@ protected CompressedChunkReader(ChannelProxy channel, CompressionMetadata metada this.metadata = metadata; this.maxCompressedLength = metadata.maxCompressedLength(); this.crcCheckChanceSupplier = crcCheckChanceSupplier; + int size = DatabaseDescriptor.getCompressedReadAheadBufferSize(); + this.readAheadBufferSize = (size > 0 && size > metadata.chunkLength()) ? size : 0; assert Integer.bitCount(metadata.chunkLength()) == 1; //must be a power of two } + // Copy constructor for a per-scan view. The view shares the parent's channel and metadata but never reads + // ahead itself, so its readAheadBufferSize is 0. + protected CompressedChunkReader(CompressedChunkReader parent) + { + super(parent.channel, parent.metadata.dataLength); + this.metadata = parent.metadata; + this.maxCompressedLength = parent.maxCompressedLength; + this.crcCheckChanceSupplier = parent.crcCheckChanceSupplier; + this.readAheadBufferSize = 0; + } + protected CompressedChunkReader forScan() { return this; @@ -90,9 +107,11 @@ public BufferType preferredBufferType() } @Override - public Rebufferer instantiateRebufferer(boolean isScan) + public Rebufferer instantiateRebufferer(ReadPattern pattern) { - return new BufferManagingRebufferer.Aligned(isScan ? forScan() : this); + // A read-ahead pattern (PARTITION_READ, SCAN) gets a per-scan view that owns its own read-ahead buffer. + // A ROW_READ reads through this shared reader with no read-ahead. + return new BufferManagingRebufferer.Aligned(pattern.readsAhead() ? forScan() : this); } protected interface CompressedReader extends Closeable @@ -206,10 +225,10 @@ private static class ScanCompressedReader implements CompressedReader private final ChannelProxy channel; private final ByteBufferHolder bufferHolder; - private final ThreadLocalReadAheadBuffer readAheadBuffer; + private final ReadAheadBuffer readAheadBuffer; private ScanCompressedReader(ChannelProxy channel, ByteBufferHolder bufferHolder, - ThreadLocalReadAheadBuffer readAheadBuffer) + ReadAheadBuffer readAheadBuffer) { this.channel = channel; this.bufferHolder = bufferHolder; @@ -278,19 +297,25 @@ public static class Direct extends CompressedChunkReader private final CompressedReader reader; private final CompressedReader scanReader; + private final int blockSize; public Direct(ChannelProxy channel, CompressionMetadata metadata, DoubleSupplier crcCheckChanceSupplier) { super(channel, metadata, crcCheckChanceSupplier); - int blockSize = FileUtils.getFileBlockSize(channel.file()); + this.blockSize = FileUtils.getFileBlockSize(channel.file()); this.reader = new DirectRandomAccessReader(channel, blockSize); + this.scanReader = null; + } - int readAheadBufferSize = DatabaseDescriptor.getCompressedReadAheadBufferSize(); - this.scanReader = (readAheadBufferSize > 0 && readAheadBufferSize > metadata.chunkLength()) - ? new ScanCompressedReader(channel, - new DirectThreadLocalByteBufferHolder(blockSize), - new DirectThreadLocalReadAheadBuffer(channel, readAheadBufferSize, blockSize)) - : null; + // Per-scan view. Each scan reader is single-threaded and owns its own read-ahead buffer, so no buffer is + // shared across threads. It shares the parent's random-access reader as a fallback; that reader's close() is + // a no-op, so the view frees only its own scan buffer. + private Direct(Direct parent, CompressedReader scanReader) + { + super(parent); + this.blockSize = parent.blockSize; + this.reader = parent.reader; + this.scanReader = scanReader; } @Override @@ -337,10 +362,14 @@ public void readChunk(long position, ByteBuffer uncompressed) @Override protected CompressedChunkReader forScan() { - if (scanReader != null) - scanReader.allocateResources(); - - return this; + if (readAheadBufferSize == 0) + return this; + + ScanCompressedReader scan = new ScanCompressedReader(channel, + new DirectThreadLocalByteBufferHolder(blockSize), + new DirectReadAheadBuffer(channel, readAheadBufferSize, blockSize)); + scan.allocateResources(); + return new Direct(this, scan); } @Override @@ -371,21 +400,28 @@ public Standard(ChannelProxy channel, CompressionMetadata metadata, DoubleSuppli { super(channel, metadata, crcCheckChanceSupplier); reader = new RandomAccessCompressedReader(channel, metadata); + this.scanReader = null; + } - int readAheadBufferSize = DatabaseDescriptor.getCompressedReadAheadBufferSize(); - scanReader = (readAheadBufferSize > 0 && readAheadBufferSize > metadata.chunkLength()) - ? new ScanCompressedReader(channel, - new ThreadLocalByteBufferHolder(metadata.compressor().preferredBufferType()), - new ThreadLocalReadAheadBuffer(channel, readAheadBufferSize, metadata.compressor().preferredBufferType())) : null; + // Per-scan view; see Direct for the ownership rationale. + private Standard(Standard parent, CompressedReader scanReader) + { + super(parent); + this.reader = parent.reader; + this.scanReader = scanReader; } @Override protected CompressedChunkReader forScan() { - if (scanReader != null) - scanReader.allocateResources(); - - return this; + if (readAheadBufferSize == 0) + return this; + + ScanCompressedReader scan = new ScanCompressedReader(channel, + new ThreadLocalByteBufferHolder(metadata.compressor().preferredBufferType()), + new ReadAheadBuffer(channel, readAheadBufferSize, metadata.compressor().preferredBufferType())); + scan.allocateResources(); + return new Standard(this, scan); } @Override diff --git a/src/java/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBuffer.java b/src/java/org/apache/cassandra/io/util/DirectReadAheadBuffer.java similarity index 83% rename from src/java/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBuffer.java rename to src/java/org/apache/cassandra/io/util/DirectReadAheadBuffer.java index 934e3620e114..5c745de020f4 100644 --- a/src/java/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBuffer.java +++ b/src/java/org/apache/cassandra/io/util/DirectReadAheadBuffer.java @@ -28,12 +28,12 @@ import sun.nio.ch.DirectBuffer; -public final class DirectThreadLocalReadAheadBuffer extends ThreadLocalReadAheadBuffer +public final class DirectReadAheadBuffer extends ReadAheadBuffer { private final int blockSize; - public DirectThreadLocalReadAheadBuffer(ChannelProxy channel, int bufferSize, int blockSize) + public DirectReadAheadBuffer(ChannelProxy channel, int bufferSize, int blockSize) { super(channel, () -> BufferUtil.allocateDirectAligned(BitUtil.align(bufferSize, blockSize), blockSize)); this.blockSize = blockSize; @@ -53,8 +53,8 @@ protected void loadBlock(ByteBuffer blockBuffer, long blockPosition, int sizeToR @Override protected void cleanBuffer(ByteBuffer buffer) { - // Aligned buffers from BufferUtil.allocateDirectAligned are slices; clean the backing buffer (attachment) + // BufferUtil.allocateDirectAligned returns an aligned slice with no cleaner; free the backing + // allocation through the attachment, matching DirectThreadLocalByteBufferHolder. MemoryUtil.clean((ByteBuffer) ((DirectBuffer) buffer).attachment()); } - -} \ No newline at end of file +} diff --git a/src/java/org/apache/cassandra/io/util/EmptyRebufferer.java b/src/java/org/apache/cassandra/io/util/EmptyRebufferer.java index 7f54a6b180f2..0d3f11b33aa4 100644 --- a/src/java/org/apache/cassandra/io/util/EmptyRebufferer.java +++ b/src/java/org/apache/cassandra/io/util/EmptyRebufferer.java @@ -64,7 +64,7 @@ public void closeReader() } @Override - public Rebufferer instantiateRebufferer(boolean isScan) + public Rebufferer instantiateRebufferer(ReadPattern pattern) { return this; } diff --git a/src/java/org/apache/cassandra/io/util/FileHandle.java b/src/java/org/apache/cassandra/io/util/FileHandle.java index 7f8c913b4b0a..79cbfc2aa510 100644 --- a/src/java/org/apache/cassandra/io/util/FileHandle.java +++ b/src/java/org/apache/cassandra/io/util/FileHandle.java @@ -205,23 +205,23 @@ public RandomAccessReader createReader() */ public RandomAccessReader createReader(RateLimiter limiter) { - return createReader(limiter, false); + return createReader(limiter, ReadPattern.ROW_READ); } - public RandomAccessReader createReader(RateLimiter limiter, boolean forScan) + public RandomAccessReader createReader(RateLimiter limiter, ReadPattern pattern) { - return createReader(limiter, forScan, OnReaderClose.RETAIN_FILE_OPEN); + return createReader(limiter, pattern, OnReaderClose.RETAIN_FILE_OPEN); } - public RandomAccessReader createReader(RateLimiter limiter, boolean forScan, OnReaderClose onReaderClose) + public RandomAccessReader createReader(RateLimiter limiter, ReadPattern pattern, OnReaderClose onReaderClose) { if (onReaderClose == OnReaderClose.CLOSE_FILE) { - return new RandomAccessReader.RandomAccessReaderWithOwnFile(instantiateRebufferer(limiter, forScan), this); + return new RandomAccessReader.RandomAccessReaderWithOwnFile(instantiateRebufferer(limiter, pattern), this); } else if (onReaderClose == OnReaderClose.RETAIN_FILE_OPEN) { - return new RandomAccessReader(instantiateRebufferer(limiter, forScan)); + return new RandomAccessReader(instantiateRebufferer(limiter, pattern)); } throw new IllegalArgumentException("Unknown close policy: " + onReaderClose); } @@ -266,12 +266,12 @@ public void dropPageCache(long before) public Rebufferer instantiateRebufferer(RateLimiter limiter) { - return instantiateRebufferer(limiter, false); + return instantiateRebufferer(limiter, ReadPattern.ROW_READ); } - public Rebufferer instantiateRebufferer(RateLimiter limiter, boolean forScan) + public Rebufferer instantiateRebufferer(RateLimiter limiter, ReadPattern pattern) { - Rebufferer rebufferer = rebuffererFactory.instantiateRebufferer(forScan); + Rebufferer rebufferer = rebuffererFactory.instantiateRebufferer(pattern); if (limiter != null) rebufferer = new LimitingRebufferer(rebufferer, limiter, DiskOptimizationStrategy.MAX_BUFFER_SIZE); diff --git a/src/java/org/apache/cassandra/io/util/MmapRebufferer.java b/src/java/org/apache/cassandra/io/util/MmapRebufferer.java index 884bc9718642..b653c86c98d8 100644 --- a/src/java/org/apache/cassandra/io/util/MmapRebufferer.java +++ b/src/java/org/apache/cassandra/io/util/MmapRebufferer.java @@ -41,8 +41,10 @@ public BufferHolder rebuffer(long position) } @Override - public Rebufferer instantiateRebufferer(boolean isScan) + public Rebufferer instantiateRebufferer(ReadPattern pattern) { + // Mmap serves reads from the OS page cache regardless of the pattern, so there is no chunk cache to + // bypass and no separate read-ahead buffer to allocate. return this; } diff --git a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java index 44b2f0c2fdd5..0bedceba4c33 100644 --- a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java +++ b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java @@ -385,7 +385,7 @@ public static RandomAccessReader open(File file) try { ChunkReader reader = new SimpleChunkReader(channel, -1, BufferType.OFF_HEAP, DEFAULT_BUFFER_SIZE); - Rebufferer rebufferer = reader.instantiateRebufferer(false); + Rebufferer rebufferer = reader.instantiateRebufferer(ReadPattern.ROW_READ); return new RandomAccessReaderWithOwnChannel(rebufferer); } catch (Throwable t) diff --git a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java b/src/java/org/apache/cassandra/io/util/ReadAheadBuffer.java similarity index 63% rename from src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java rename to src/java/org/apache/cassandra/io/util/ReadAheadBuffer.java index ff59f7cf96df..62d521ef8960 100644 --- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java +++ b/src/java/org/apache/cassandra/io/util/ReadAheadBuffer.java @@ -19,49 +19,40 @@ package org.apache.cassandra.io.util; import java.nio.ByteBuffer; -import java.util.HashMap; -import java.util.Map; import java.util.function.Supplier; +import com.google.common.annotations.VisibleForTesting; + import org.apache.cassandra.io.compress.BufferType; import org.apache.cassandra.io.compress.CorruptBlockException; import org.apache.cassandra.io.sstable.CorruptSSTableException; import org.apache.cassandra.utils.Closeable; import org.apache.cassandra.utils.memory.MemoryUtil; -import io.netty.util.concurrent.FastThreadLocal; - -public class ThreadLocalReadAheadBuffer implements Closeable +/** + * A read-ahead buffer for sequential scans of a single file. + *

+ * Each instance owns its buffer. An instance is used by one scan reader, which is single-threaded, so the buffer is + * never shared across threads. A scan reader allocates one of these on open and frees it on close. N scanners over N + * inputs give N buffers by construction. + */ +public class ReadAheadBuffer implements Closeable { - - private static class Block - { - ByteBuffer buffer = null; - int index = -1; - } - protected final ChannelProxy channel; private final Supplier bufferSupplier; - - private static final FastThreadLocal> blockMap = new FastThreadLocal<>() - { - @Override - protected Map initialValue() - { - return new HashMap<>(); - } - }; - - private volatile int bufferSize = -1; private final long channelSize; - public ThreadLocalReadAheadBuffer(ChannelProxy channel, int bufferSize, BufferType bufferType) + private ByteBuffer buffer; + private int index = -1; + private int bufferSize = -1; + + public ReadAheadBuffer(ChannelProxy channel, int bufferSize, BufferType bufferType) { this(channel, () -> bufferType.allocate(bufferSize)); } - public ThreadLocalReadAheadBuffer(ChannelProxy channel, Supplier bufferSupplier) + public ReadAheadBuffer(ChannelProxy channel, Supplier bufferSupplier) { this.channel = channel; this.channelSize = channel.size(); @@ -70,41 +61,39 @@ public ThreadLocalReadAheadBuffer(ChannelProxy channel, Supplier buf public boolean hasBuffer() { - return block().buffer != null; + return buffer != null; + } + + @VisibleForTesting + int bufferSize() + { + return bufferSize; } public int remaining() { - return getBlock().buffer.remaining(); + return getBuffer().remaining(); } public void allocateBuffer() { - getBlock(); + getBuffer(); } - private Block getBlock() + private ByteBuffer getBuffer() { - Block block = block(); - if (block.buffer == null) + if (buffer == null) { - block.buffer = bufferSupplier.get(); - block.buffer.clear(); - if (bufferSize == -1) - bufferSize = block.buffer.capacity(); + buffer = bufferSupplier.get(); + buffer.clear(); + bufferSize = buffer.capacity(); } - return block; - } - - private Block block() - { - return blockMap.get().computeIfAbsent(channel.filePath(), k -> new Block()); + return buffer; } public void fill(long position) throws CorruptBlockException { - Block block = getBlock(); - ByteBuffer blockBuffer = block.buffer; + ByteBuffer blockBuffer = getBuffer(); if (position >= channelSize) throw new CorruptBlockException(channel.filePath(), position, bufferSize); @@ -113,11 +102,11 @@ public void fill(long position) throws CorruptBlockException long remaining = channelSize - blockPosition; int sizeToRead = (int) Math.min(remaining, bufferSize); - if (block.index != blockNo) + if (index != blockNo) { blockBuffer.flip(); loadBlock(blockBuffer, blockPosition, sizeToRead); - block.index = blockNo; + index = blockNo; } blockBuffer.flip(); @@ -134,8 +123,7 @@ protected void loadBlock(ByteBuffer blockBuffer, long blockPosition, int sizeToR public int read(ByteBuffer dest, int length) { - Block block = getBlock(); - ByteBuffer blockBuffer = block.buffer; + ByteBuffer blockBuffer = getBuffer(); ByteBuffer tmp = blockBuffer.duplicate(); tmp.limit(tmp.position() + length); dest.put(tmp); @@ -146,21 +134,16 @@ public int read(ByteBuffer dest, int length) public void clear(boolean deallocate) { - // avoid calling block() here to reduce unintended allocations - Block block = blockMap.get().get(channel.filePath()); - if (block == null) - return; - - block.index = -1; - if (block.buffer == null) + if (buffer == null) return; - ByteBuffer blockBuffer = block.buffer; - blockBuffer.clear(); + index = -1; + buffer.clear(); if (deallocate) { - cleanBuffer(blockBuffer); - block.buffer = null; + cleanBuffer(buffer); + buffer = null; + bufferSize = -1; } } @@ -173,6 +156,5 @@ protected void cleanBuffer(ByteBuffer buffer) public void close() { clear(true); - blockMap.get().remove(channel.filePath()); } } diff --git a/src/java/org/apache/cassandra/io/util/ReadPattern.java b/src/java/org/apache/cassandra/io/util/ReadPattern.java new file mode 100644 index 000000000000..62032924d959 --- /dev/null +++ b/src/java/org/apache/cassandra/io/util/ReadPattern.java @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.io.util; + +/** + * How a reader accesses a file. A caller states the access pattern when it instantiates a rebufferer. Each + * rebufferer layer then reads only the decision it owns: {@link org.apache.cassandra.cache.ChunkCache} reads + * {@link #usesCache()}, and {@link CompressedChunkReader} reads {@link #readsAhead()}. + * + *

The pattern carries two orthogonal decisions, but the three values below are the only meaningful pairs. + * A pattern that neither caches nor reads ahead has no caller, so the enum does not offer it. See + * CASSANDRA-21671. + */ +public enum ReadPattern +{ + /** + * A read of specific rows: a single-partition read served by named or sliced clustering keys. It uses the + * chunk cache and does not read ahead. Repeated reads of hot data hit the cache. + */ + ROW_READ(true, false), + + /** + * A read that walks a range of partitions in order, such as a token-range query. Re-use is likely, so it + * keeps the chunk cache. It reads ahead only when the chunk cache is off. + */ + PARTITION_READ(true, true), + + /** + * An unbounded, one-shot read: compaction, cursor compaction, and similar callers. It reads each chunk once, + * so it must bypass the chunk cache to avoid evicting hot data with one-shot chunks. It reads ahead through + * its own buffer instead. + */ + SCAN(false, true); + + private final boolean usesCache; + private final boolean readsAhead; + + ReadPattern(boolean usesCache, boolean readsAhead) + { + this.usesCache = usesCache; + this.readsAhead = readsAhead; + } + + /** True if the reader should go through the chunk cache. */ + public boolean usesCache() + { + return usesCache; + } + + /** True if the reader should use its own read-ahead buffer instead of the chunk cache. */ + public boolean readsAhead() + { + return readsAhead; + } +} diff --git a/src/java/org/apache/cassandra/io/util/RebuffererFactory.java b/src/java/org/apache/cassandra/io/util/RebuffererFactory.java index 192fb8ea0cd8..345fd832a11e 100644 --- a/src/java/org/apache/cassandra/io/util/RebuffererFactory.java +++ b/src/java/org/apache/cassandra/io/util/RebuffererFactory.java @@ -28,5 +28,5 @@ */ public interface RebuffererFactory extends ReaderFileProxy { - Rebufferer instantiateRebufferer(boolean isScan); + Rebufferer instantiateRebufferer(ReadPattern pattern); } diff --git a/src/java/org/apache/cassandra/io/util/SimpleChunkReader.java b/src/java/org/apache/cassandra/io/util/SimpleChunkReader.java index fec1216bf4e4..68780a14c1d5 100644 --- a/src/java/org/apache/cassandra/io/util/SimpleChunkReader.java +++ b/src/java/org/apache/cassandra/io/util/SimpleChunkReader.java @@ -55,7 +55,7 @@ public BufferType preferredBufferType() } @Override - public Rebufferer instantiateRebufferer(boolean forScan) + public Rebufferer instantiateRebufferer(ReadPattern pattern) { if (Integer.bitCount(bufferSize) == 1) return new BufferManagingRebufferer.Aligned(this); diff --git a/test/unit/org/apache/cassandra/cache/ChunkCacheScanBypassTest.java b/test/unit/org/apache/cassandra/cache/ChunkCacheScanBypassTest.java new file mode 100644 index 000000000000..8a96d25735da --- /dev/null +++ b/test/unit/org/apache/cassandra/cache/ChunkCacheScanBypassTest.java @@ -0,0 +1,201 @@ +/* + * 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.cache; + +import java.lang.reflect.Constructor; +import java.util.concurrent.atomic.AtomicInteger; + +import org.assertj.core.api.Assertions; +import org.junit.Test; + +import org.apache.cassandra.config.DataStorageSpec; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ClusteringComparator; +import org.apache.cassandra.io.compress.CompressedSequentialWriter; +import org.apache.cassandra.io.compress.CompressionMetadata; +import org.apache.cassandra.io.filesystem.ListenableFileSystem; +import org.apache.cassandra.io.sstable.metadata.MetadataCollector; +import org.apache.cassandra.io.util.ChannelProxy; +import org.apache.cassandra.io.util.CompressedChunkReader; +import org.apache.cassandra.io.util.File; +import org.apache.cassandra.io.util.FileSystems; +import org.apache.cassandra.io.util.ReadPattern; +import org.apache.cassandra.io.util.Rebufferer; +import org.apache.cassandra.io.util.RebuffererFactory; +import org.apache.cassandra.io.util.SequentialWriterOption; +import org.apache.cassandra.schema.CompressionParams; +import org.apache.cassandra.utils.memory.BufferPool; +import org.apache.cassandra.utils.memory.BufferPools; + +/** + * A one-shot SCAN (compaction and similar) must bypass the chunk cache so it does not evict hot data with + * one-shot chunks. CachingRebufferer.instantiateRebufferer(pattern) delegates to the source for a SCAN, which + * routes the read through its own read-ahead buffer instead of the cache. A PARTITION_READ and a ROW_READ still + * use the cache, because both re-read hot data. See CASSANDRA-21671. + */ +public class ChunkCacheScanBypassTest +{ + static + { + DatabaseDescriptor.clientInitialization(); + // Client init leaves file_cache_size null, which gives the chunk cache a maximum weight of 0 and a + // 0-byte buffer pool. Set a real size before ChunkCache and the buffer pool load so the cache can + // actually hold entries; otherwise a point read spins forever in Caffeine's degenerate maintenance. + DatabaseDescriptor.getRawConfig().file_cache_size = new DataStorageSpec.IntMebibytesBound(64); + } + + // The shared ChunkCache.instance is null when the file cache is disabled (the default). Build a private + // instance directly so the test controls the cache regardless of the file_cache_enabled setting. + private static ChunkCache newChunkCache() throws Exception + { + Constructor ctor = ChunkCache.class.getDeclaredConstructor(BufferPool.class); + ctor.setAccessible(true); + return ctor.newInstance(BufferPools.forChunkCache()); + } + + private static CompressionMetadata writeCompressedFile(File f, File offsets, File digest, long longsToWrite) throws Exception + { + SequentialWriterOption option = SequentialWriterOption.newBuilder() + .finishOnClose(false) + .bufferSize(1 << 10) + .build(); + CompressionParams params = CompressionParams.snappy(4096, 1.1); + try (CompressedSequentialWriter writer = new CompressedSequentialWriter(f, offsets, digest, option, params, + new MetadataCollector(new ClusteringComparator()))) + { + for (long i = 0; i < longsToWrite; i++) + writer.writeLong(i); + + writer.sync(); + return writer.open(0); + } + } + + // Rebuffer every chunk-aligned position over the whole file and release each holder. This is enough to + // drive cache population (a point read) or a read-ahead scan; data correctness is covered elsewhere. + private static void rebufferWholeFile(Rebufferer rebufferer, long fileLength, int chunkSize) + { + long position = 0; + do + { + Rebufferer.BufferHolder holder = rebufferer.rebuffer(position); + holder.release(); + position += chunkSize; + } + while (position < fileLength); + } + + @Test(timeout = 60_000) + public void scanBypassesCacheWhilePointReadPopulatesIt() throws Exception + { + FileSystems.newGlobalInMemoryFileSystem(); + File f = new File("/chunk_cache_scan_bypass.db"); + CompressionMetadata metadata = writeCompressedFile(f, new File("/chunk_cache_scan_bypass.offset"), + new File("/chunk_cache_scan_bypass.digest"), 200_000); + DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(256); + + ChunkCache cache = newChunkCache(); + try (ChannelProxy channel = new ChannelProxy(f); + CompressedChunkReader reader = new CompressedChunkReader.Standard(channel, metadata, () -> 1d); + CompressionMetadata ignored = metadata) + { + RebuffererFactory factory = cache.wrap(reader); + + // A SCAN gets a distinct, non-caching rebufferer and must not add anything to the cache. + Rebufferer scan = factory.instantiateRebufferer(ReadPattern.SCAN); + Assertions.assertThat(scan).isNotSameAs(factory); + try + { + rebufferWholeFile(scan, reader.fileLength(), reader.chunkSize()); + Assertions.assertThat(cache.size()).as("a scan must not populate the chunk cache").isZero(); + } + finally + { + // Release the scan's read-ahead buffer only. close() would close the shared underlying reader, + // which the point read below still needs; the parent reader's try-with-resources closes it. + scan.closeReader(); + } + + // A PARTITION_READ reuses the caching rebufferer and populates the cache, so a repeated partition-range + // query re-reads hot data. This is the CASSANDRA-21671 regression: it must not bypass the cache. + Rebufferer partitionRead = factory.instantiateRebufferer(ReadPattern.PARTITION_READ); + Assertions.assertThat(partitionRead).isSameAs(factory); + rebufferWholeFile(partitionRead, reader.fileLength(), reader.chunkSize()); + Assertions.assertThat(cache.size()).as("a partition read must populate the chunk cache").isGreaterThan(0); + + cache.clear(); + + // A point read reuses the caching rebufferer and populates the cache, proving the cache works. + Rebufferer point = factory.instantiateRebufferer(ReadPattern.ROW_READ); + Assertions.assertThat(point).isSameAs(factory); + rebufferWholeFile(point, reader.fileLength(), reader.chunkSize()); + Assertions.assertThat(cache.size()).as("a point read must populate the chunk cache").isGreaterThan(0); + } + finally + { + cache.clear(); + } + } + + @Test(timeout = 60_000) + public void scanUnderCacheUsesReadAheadBuffer() throws Exception + { + ListenableFileSystem fs = FileSystems.newGlobalInMemoryFileSystem(); + File f = new File("/chunk_cache_scan_readahead.db"); + CompressionMetadata metadata = writeCompressedFile(f, new File("/chunk_cache_scan_readahead.offset"), + new File("/chunk_cache_scan_readahead.digest"), 200_000); + DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(256); + + AtomicInteger reads = new AtomicInteger(); + fs.onPostRead(f.toPath()::equals, (p, c, pos, dst, r) -> reads.incrementAndGet()); + + ChunkCache cache = newChunkCache(); + try (ChannelProxy channel = new ChannelProxy(f); + CompressedChunkReader reader = new CompressedChunkReader.Standard(channel, metadata, () -> 1d); + CompressionMetadata ignored = metadata) + { + RebuffererFactory factory = cache.wrap(reader); + + long fileLength = reader.fileLength(); + int chunkSize = reader.chunkSize(); + long chunkCount = (fileLength + chunkSize - 1) / chunkSize; + + Rebufferer scan = factory.instantiateRebufferer(ReadPattern.SCAN); + try + { + rebufferWholeFile(scan, fileLength, chunkSize); + } + finally + { + scan.closeReader(); + } + + // Read-ahead reads many chunks per physical read, so the scan issues far fewer reads than chunks. + Assertions.assertThat((long) reads.get()) + .as("the scan must batch reads through the read-ahead buffer") + .isLessThan(chunkCount); + // The scan must not have populated the cache either. + Assertions.assertThat(cache.size()).as("a scan must not populate the chunk cache").isZero(); + } + finally + { + cache.clear(); + } + } +} diff --git a/test/unit/org/apache/cassandra/io/compress/DirectCompressedSequentialWriterTest.java b/test/unit/org/apache/cassandra/io/compress/DirectCompressedSequentialWriterTest.java index 68833f1530a0..b638783f49b2 100644 --- a/test/unit/org/apache/cassandra/io/compress/DirectCompressedSequentialWriterTest.java +++ b/test/unit/org/apache/cassandra/io/compress/DirectCompressedSequentialWriterTest.java @@ -62,6 +62,7 @@ import org.apache.cassandra.io.util.FileHandle; import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.io.util.RandomAccessReader; +import org.apache.cassandra.io.util.ReadPattern; import org.apache.cassandra.io.util.SequentialWriter; import org.apache.cassandra.io.util.SequentialWriterOption; import org.apache.cassandra.metrics.StorageMetrics; @@ -436,7 +437,7 @@ public void testAbortAfterFlushCleansUpResources() throws Exception @Test public void testDirectMemoryIsCleanedOnClose() throws Exception { - // Sized to dominate baseline allocator noise; matches DirectThreadLocalReadAheadBufferTest. + // Sized to dominate baseline allocator noise; matches DirectReadAheadBufferTest. int bufferSize = 64 * 1024 * 1024; withDirectWriteBufferSize(bufferSize / 1024, () -> withTempDataFile("direct_mem_clean", (dataFile, metadataFile) -> @@ -946,7 +947,7 @@ private static long assertReadBack(OpenMoment moment, ReaderPath path, byte[] pa writer.finish(); try (CompressionMetadata md = CompressionMetadata.open(metadataFile, dataFile.length(), true); FileHandle fh = new FileHandle.Builder(dataFile).withCompressionMetadata(md).complete(); - RandomAccessReader reader = fh.createReader(null, path == ReaderPath.SCAN)) + RandomAccessReader reader = fh.createReader(null, path == ReaderPath.SCAN ? ReadPattern.SCAN : ReadPattern.ROW_READ)) { assertReadablePrefix(reader, payload, payload.length, moment, path, params, regime); } @@ -972,7 +973,7 @@ private static void readEarlyOpenAndFinish(DirectCompressedSequentialWriter writ BufferRegime regime) throws IOException { try (FileHandle fh = new FileHandle.Builder(dataFile).withCompressionMetadata(md).complete(); - RandomAccessReader reader = fh.createReader(null, path == ReaderPath.SCAN)) + RandomAccessReader reader = fh.createReader(null, path == ReaderPath.SCAN ? ReadPattern.SCAN : ReadPattern.ROW_READ)) { if (path == ReaderPath.SCAN) { diff --git a/test/unit/org/apache/cassandra/io/sstable/format/SSTableReaderDataReaderTest.java b/test/unit/org/apache/cassandra/io/sstable/format/SSTableReaderDataReaderTest.java index 49165b6577c4..500b856c1285 100644 --- a/test/unit/org/apache/cassandra/io/sstable/format/SSTableReaderDataReaderTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/format/SSTableReaderDataReaderTest.java @@ -216,11 +216,11 @@ public void testMultipleNewHandleReadersDoNotLeakResources() } @Test - public void testForScanReusesWithNullMode() + public void testForPartitionReadReusesWithNullMode() { SSTableReader sstable = createSSTable(CF_UNCOMPRESSED); - try (RandomAccessReader reader = sstable.openDataReaderForScan()) + try (RandomAccessReader reader = sstable.openDataReaderForPartitionRead()) { assertReaderSharesDfileChannel(sstable, reader); } diff --git a/test/unit/org/apache/cassandra/io/util/DirectCompressedChunkReaderTest.java b/test/unit/org/apache/cassandra/io/util/DirectCompressedChunkReaderTest.java index 0065efc9260c..dc3c7521a7be 100644 --- a/test/unit/org/apache/cassandra/io/util/DirectCompressedChunkReaderTest.java +++ b/test/unit/org/apache/cassandra/io/util/DirectCompressedChunkReaderTest.java @@ -245,36 +245,43 @@ private static void readAndVerifyChunks(File file, CompressionMetadata metadata, CompressedChunkReader reader = new CompressedChunkReader.Direct(channel, metadata, () -> 1d); metadata) { - if (forScan) - reader.forScan(); - - long currentFileOffset = 0; - long totalBytesRead = 0; - int currentChunkIndex = 0; - - while (totalBytesRead < totalBytesExpected) + // forScan() returns a per-scan reader that owns its read-ahead buffer; use it, then close it. + CompressedChunkReader scanReader = forScan ? reader.forScan() : reader; + try { - ByteBuffer currentExpectedChunk = expectedChunksData.get(currentChunkIndex); + long currentFileOffset = 0; + long totalBytesRead = 0; + int currentChunkIndex = 0; + + while (totalBytesRead < totalBytesExpected) + { + ByteBuffer currentExpectedChunk = expectedChunksData.get(currentChunkIndex); - readBuffer.clear(); - reader.readChunk(currentFileOffset, readBuffer); + readBuffer.clear(); + scanReader.readChunk(currentFileOffset, readBuffer); - Assert.assertTrue("Read buffer is empty unexpectedly at offset " + currentFileOffset, readBuffer.hasRemaining()); + Assert.assertTrue("Read buffer is empty unexpectedly at offset " + currentFileOffset, readBuffer.hasRemaining()); - int actualBytesRead = readBuffer.remaining(); - int expectedBytes = currentExpectedChunk.remaining(); + int actualBytesRead = readBuffer.remaining(); + int expectedBytes = currentExpectedChunk.remaining(); - Assert.assertTrue("Read buffer remaining (" + actualBytesRead + ") is less than expected (" + expectedBytes + ") at offset " + currentFileOffset, - actualBytesRead >= expectedBytes); + Assert.assertTrue("Read buffer remaining (" + actualBytesRead + ") is less than expected (" + expectedBytes + ") at offset " + currentFileOffset, + actualBytesRead >= expectedBytes); - int originalReadBufferLimit = readBuffer.limit(); - readBuffer.limit(expectedBytes); - Assert.assertEquals("Mismatched data at offset " + currentFileOffset, currentExpectedChunk, readBuffer); - readBuffer.limit(originalReadBufferLimit); + int originalReadBufferLimit = readBuffer.limit(); + readBuffer.limit(expectedBytes); + Assert.assertEquals("Mismatched data at offset " + currentFileOffset, currentExpectedChunk, readBuffer); + readBuffer.limit(originalReadBufferLimit); - totalBytesRead += expectedBytes; - currentFileOffset += metadata.chunkLength(); - currentChunkIndex++; + totalBytesRead += expectedBytes; + currentFileOffset += metadata.chunkLength(); + currentChunkIndex++; + } + } + finally + { + if (scanReader != reader) + scanReader.close(); } } finally diff --git a/test/unit/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBufferTest.java b/test/unit/org/apache/cassandra/io/util/DirectReadAheadBufferTest.java similarity index 88% rename from test/unit/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBufferTest.java rename to test/unit/org/apache/cassandra/io/util/DirectReadAheadBufferTest.java index 46b3091f1167..9d7019503301 100644 --- a/test/unit/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBufferTest.java +++ b/test/unit/org/apache/cassandra/io/util/DirectReadAheadBufferTest.java @@ -27,7 +27,7 @@ import org.apache.cassandra.utils.Pair; -public class DirectThreadLocalReadAheadBufferTest extends ThreadLocalReadAheadBufferTest +public class DirectReadAheadBufferTest extends ReadAheadBufferTest { @Test @@ -57,12 +57,12 @@ private void testReads(InputData propertyInputs, int bufferSize, int blockSize) try (ChannelProxy bufferedChannel = new ChannelProxy(propertyInputs.file); ChannelProxy directChannel = new ChannelProxy(propertyInputs.file, ChannelProxy.IOMode.DIRECT)) { - ThreadLocalReadAheadBuffer tlrab = new DirectThreadLocalReadAheadBuffer(directChannel, bufferSize, blockSize); + ReadAheadBuffer rab = new DirectReadAheadBuffer(directChannel, bufferSize, blockSize); for (Pair read : propertyInputs.positionsAndLengths) { - testRead(read, bufferedChannel, tlrab); + testRead(read, bufferedChannel, rab); } - tlrab.close(); + rab.close(); } } @@ -75,16 +75,16 @@ public void testDirectMemoryIsCleanedOnClose() try (ChannelProxy channel = new ChannelProxy(files[0], ChannelProxy.IOMode.DIRECT)) { - DirectThreadLocalReadAheadBuffer tlrab = - new DirectThreadLocalReadAheadBuffer(channel, bufferSize, blockSize); + DirectReadAheadBuffer rab = + new DirectReadAheadBuffer(channel, bufferSize, blockSize); // Force buffer allocation - tlrab.allocateBuffer(); + rab.allocateBuffer(); long memoryUsedBefore = directPool.getMemoryUsed(); // Close should clean the direct memory - tlrab.close(); + rab.close(); long memoryUsedAfter = directPool.getMemoryUsed(); @@ -110,4 +110,4 @@ private static BufferPoolMXBean getDirectBufferPool() throw new IllegalStateException("Direct buffer pool not found"); } -} \ No newline at end of file +} diff --git a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java b/test/unit/org/apache/cassandra/io/util/ReadAheadBufferTest.java similarity index 60% rename from test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java rename to test/unit/org/apache/cassandra/io/util/ReadAheadBufferTest.java index 47bba7b6c534..8529987b1b23 100644 --- a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java +++ b/test/unit/org/apache/cassandra/io/util/ReadAheadBufferTest.java @@ -44,10 +44,10 @@ import static java.lang.Math.max; import static org.apache.cassandra.config.CassandraRelevantProperties.JAVA_IO_TMPDIR; -public class ThreadLocalReadAheadBufferTest implements WithQuickTheories +public class ReadAheadBufferTest implements WithQuickTheories { private static final int numFiles = 5; - private static final Logger logger = LoggerFactory.getLogger(ThreadLocalReadAheadBufferTest.class); + private static final Logger logger = LoggerFactory.getLogger(ReadAheadBufferTest.class); protected static final File[] files = new File[numFiles]; protected static Integer seed; @@ -94,19 +94,105 @@ public void testReadsLikeChannelProxy() .checkAssert(this::testReads); } + @Test + public void allocateInitialisesBufferSizeFromCapacity() throws CorruptBlockException + { + int bufferSize = new DataStorageSpec.IntKibibytesBound("256KiB").toBytes(); + try (ChannelProxy channel = new ChannelProxy(files[0])) + { + ReadAheadBuffer buffer = new ReadAheadBuffer(channel, bufferSize, BufferType.OFF_HEAP); + try + { + // Buffer is lazily allocated; before allocation there is no buffer. + Assert.assertFalse(buffer.hasBuffer()); + + buffer.allocateBuffer(); + + // Ownership is per-instance: allocation must set bufferSize from the buffer capacity, + // not leave it at -1 (which would make fill() call ByteBuffer.limit(-1)). + Assert.assertTrue(buffer.hasBuffer()); + Assert.assertEquals("allocate must initialise bufferSize from capacity", + bufferSize, buffer.bufferSize()); + } + finally + { + buffer.close(); + } + } + } + + @Test + public void independentInstancesDoNotShareBuffer() throws CorruptBlockException + { + // Each ReadAheadBuffer owns its own buffer. Two instances over the same file, on the same + // thread, must not share state. Under the old static per-thread, per-path Block cache they + // shared one buffer, so advancing one instance to a different block clobbered the other's view. + File file = files[0]; + int bufferSize = new DataStorageSpec.IntKibibytesBound("256KiB").toBytes(); + try (ChannelProxy channel = new ChannelProxy(file)) + { + ReadAheadBuffer a = new ReadAheadBuffer(channel, bufferSize, BufferType.OFF_HEAP); + ReadAheadBuffer b = new ReadAheadBuffer(channel, bufferSize, BufferType.OFF_HEAP); + try + { + a.fill(0); + + // Advance b to a different block if the file is large enough; else keep it on block 0. + long secondBlock = bufferSize; + b.fill(channel.size() > secondBlock ? secondBlock : 0); + + // a must still read block 0. A shared buffer would return b's block here. + int readSize = Math.min(100, (int) channel.size()); + ByteBuffer expected = ByteBuffer.allocate(readSize); + channel.read(expected, 0); + expected.flip(); + + ByteBuffer actual = ByteBuffer.allocate(readSize); + a.read(actual, readSize); + actual.flip(); + + Assert.assertEquals(expected, actual); + } + finally + { + b.close(); + a.close(); + } + } + } + + @Test + public void closeFreesBufferAndIsIdempotent() throws CorruptBlockException + { + int bufferSize = new DataStorageSpec.IntKibibytesBound("256KiB").toBytes(); + try (ChannelProxy channel = new ChannelProxy(files[0])) + { + ReadAheadBuffer buffer = new ReadAheadBuffer(channel, bufferSize, BufferType.OFF_HEAP); + buffer.allocateBuffer(); + Assert.assertTrue(buffer.hasBuffer()); + + buffer.close(); + Assert.assertFalse("close must free the owned buffer", buffer.hasBuffer()); + + // A second close must not double-free. + buffer.close(); + Assert.assertFalse(buffer.hasBuffer()); + } + } + protected void testReads(InputData propertyInputs) { try (ChannelProxy channel = new ChannelProxy(propertyInputs.file); - ThreadLocalReadAheadBuffer tlrab = new ThreadLocalReadAheadBuffer(channel, new DataStorageSpec.IntKibibytesBound("256KiB").toBytes(), BufferType.OFF_HEAP); ) + ReadAheadBuffer rab = new ReadAheadBuffer(channel, new DataStorageSpec.IntKibibytesBound("256KiB").toBytes(), BufferType.OFF_HEAP); ) { for (Pair read : propertyInputs.positionsAndLengths) { - testRead(read, channel, tlrab); + testRead(read, channel, rab); } } } - protected static void testRead(Pair read, ChannelProxy bufferedChannel, ThreadLocalReadAheadBuffer tlrab) + protected static void testRead(Pair read, ChannelProxy bufferedChannel, ReadAheadBuffer rab) { int readSize = Math.min(read.right, (int) (bufferedChannel.size() - read.left)); ByteBuffer buf1 = ByteBuffer.allocate(readSize); @@ -118,12 +204,12 @@ protected static void testRead(Pair read, ChannelProxy bufferedCh int copied = 0; while (copied < readSize) { - tlrab.fill(read.left + copied); + rab.fill(read.left + copied); int leftToRead = readSize - copied; - if (tlrab.remaining() >= leftToRead) - copied += tlrab.read(buf2, leftToRead); + if (rab.remaining() >= leftToRead) + copied += rab.read(buf2, leftToRead); else - copied += tlrab.read(buf2, tlrab.remaining()); + copied += rab.read(buf2, rab.remaining()); } } catch (CorruptSSTableException | CorruptBlockException e) diff --git a/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java b/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java index 436a3e3aa23f..3fe16480b3a8 100644 --- a/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java +++ b/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java @@ -22,6 +22,13 @@ import java.nio.channels.FileChannel; import java.nio.file.Files; import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.assertj.core.api.Assertions; @@ -94,20 +101,27 @@ protected void doReads(File f, CompressionMetadata metadata, long length, boolea try (CompressedChunkReader reader = new CompressedChunkReader.Standard(channel, metadata, () -> 1d); metadata) { - if (useReadAhead) - reader.forScan(); - - long offset = 0; - long maxOffset = length * Long.BYTES; - do + // forScan() returns a per-scan reader that owns its read-ahead buffer; use it, then close it. + CompressedChunkReader scanReader = useReadAhead ? reader.forScan() : reader; + try { - reader.readChunk(offset, buffer); - for (long expected = offset / Long.BYTES; buffer.hasRemaining(); expected++) - Assertions.assertThat(buffer.getLong()).isEqualTo(expected); + long offset = 0; + long maxOffset = length * Long.BYTES; + do + { + scanReader.readChunk(offset, buffer); + for (long expected = offset / Long.BYTES; buffer.hasRemaining(); expected++) + Assertions.assertThat(buffer.getLong()).isEqualTo(expected); - offset += metadata.chunkLength(); + offset += metadata.chunkLength(); + } + while (offset < maxOffset); + } + finally + { + if (scanReader != reader) + scanReader.close(); } - while (offset < maxOffset); } } finally @@ -155,15 +169,99 @@ public void scanReaderShouldNotHangOnTruncatedFile() throws Exception CompressedChunkReader reader = new CompressedChunkReader.Standard(channel, metadata, () -> 1.1); metadata) { - reader.forScan(); - - Assertions.assertThatThrownBy(() -> reader.readChunk(lastChunkUncompressedStart, buffer)) - .as("readChunk() reading past truncated EOF via the scan path") - .isInstanceOf(CorruptSSTableException.class); + CompressedChunkReader scanReader = reader.forScan(); + try + { + Assertions.assertThatThrownBy(() -> scanReader.readChunk(lastChunkUncompressedStart, buffer)) + .as("readChunk() reading past truncated EOF via the scan path") + .isInstanceOf(CorruptSSTableException.class); + } + finally + { + if (scanReader != reader) + scanReader.close(); + } } finally { MemoryUtil.clean(buffer); } } + + @Test(timeout = 60_000) + public void concurrentScansOfOneReaderAreIndependent() throws Exception + { + // Two or more scanners over one SSTable (compaction plus an index build sharing the dfile) each open + // their own scan reader via forScan(). Under Option A each scan reader owns its read-ahead buffer, so + // concurrent scans of one file must not corrupt each other or double-free a shared buffer. + SequentialWriterOption writerOption = writerOption(1 << 10); + CompressionParams params = CompressionParams.snappy(4096, 1.1); + + FileSystems.newGlobalInMemoryFileSystem(); + File f = new File("/concurrent_scans.db"); + File offsets = new File("/concurrent_scans.offset"); + File digest = new File("/concurrent_scans.digest"); + + long longsToWrite = 200_000; // spans many read-ahead blocks + CompressionMetadata metadata; + try (CompressedSequentialWriter writer = new CompressedSequentialWriter(f, offsets, digest, writerOption, params, new MetadataCollector(new ClusteringComparator()))) + { + for (long i = 0; i < longsToWrite; i++) + writer.writeLong(i); + writer.sync(); + metadata = writer.open(0); + } + + DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(256); // minimum allowed; spans many blocks over the test file + + int threads = 4; + long maxOffset = longsToWrite * Long.BYTES; + try (ChannelProxy channel = new ChannelProxy(f); + CompressedChunkReader reader = new CompressedChunkReader.Standard(channel, metadata, () -> 1d); + metadata) + { + ExecutorService pool = Executors.newFixedThreadPool(threads); + CyclicBarrier barrier = new CyclicBarrier(threads); + List> futures = new ArrayList<>(); + try + { + for (int t = 0; t < threads; t++) + { + futures.add(pool.submit(() -> { + ByteBuffer buffer = ByteBuffer.allocateDirect(metadata.chunkLength()); + CompressedChunkReader scanReader = reader.forScan(); + try + { + barrier.await(); // start all scans together to maximise overlap + long offset = 0; + do + { + scanReader.readChunk(offset, buffer); + for (long expected = offset / Long.BYTES; buffer.hasRemaining(); expected++) + Assertions.assertThat(buffer.getLong()).isEqualTo(expected); + + offset += metadata.chunkLength(); + } + while (offset < maxOffset); + return null; + } + finally + { + if (scanReader != reader) + scanReader.close(); + MemoryUtil.clean(buffer); + } + })); + } + + // Propagate any assertion failure or corruption from the worker threads. + for (Future future : futures) + future.get(45, TimeUnit.SECONDS); + } + finally + { + pool.shutdownNow(); + } + } + } } \ No newline at end of file