From 34c7925ac7c2a1f0a50cd316cb1a6187503c31f6 Mon Sep 17 00:00:00 2001 From: Jon Haddad Date: Sun, 13 Sep 2026 17:43:40 -0700 Subject: [PATCH 1/2] CASSANDRA-21671: Give each compressed scan reader its own read-ahead buffer The compressed read-ahead buffer was owned by the single scanReader field of a shared CompressedChunkReader, so one buffer was reused across every reader of the file and across threads. To make that shared object safe, its state lived in a static per-thread, per-path Block map. That cross-instance, cross-thread reuse caused the reinit and double-free faults on this path. Own the read-ahead buffer on the per-scan reader instead. forScan() now returns a per-reader view that owns a plain buffer, allocated on open and freed on close. Each scan reader is single-threaded, so no buffer is shared across threads and no thread-local state is needed. N scanners over N inputs give N buffers by construction. Remove the static Block map, the bufferSize == -1 reinit, and the aligned slice resolution in cleanBuffer; each buffer class now frees the buffer type it allocates. Rename ThreadLocalReadAheadBuffer to ReadAheadBuffer and DirectThreadLocalReadAheadBuffer to DirectReadAheadBuffer, since the names no longer describe thread-local state. Add a concurrent-scan test that reads one file from several threads at once, plus per-instance ownership and lifecycle tests. --- CHANGES.txt | 1 + .../io/util/CompressedChunkReader.java | 71 +++++++--- ...Buffer.java => DirectReadAheadBuffer.java} | 10 +- ...dAheadBuffer.java => ReadAheadBuffer.java} | 100 ++++++-------- .../DirectCompressedSequentialWriterTest.java | 2 +- .../util/DirectCompressedChunkReaderTest.java | 53 +++---- ...st.java => DirectReadAheadBufferTest.java} | 18 +-- ...fferTest.java => ReadAheadBufferTest.java} | 104 ++++++++++++-- .../StandardCompressedChunkReaderTest.java | 130 +++++++++++++++--- 9 files changed, 345 insertions(+), 144 deletions(-) rename src/java/org/apache/cassandra/io/util/{DirectThreadLocalReadAheadBuffer.java => DirectReadAheadBuffer.java} (83%) rename src/java/org/apache/cassandra/io/util/{ThreadLocalReadAheadBuffer.java => ReadAheadBuffer.java} (63%) rename test/unit/org/apache/cassandra/io/util/{DirectThreadLocalReadAheadBufferTest.java => DirectReadAheadBufferTest.java} (88%) rename test/unit/org/apache/cassandra/io/util/{ThreadLocalReadAheadBufferTest.java => ReadAheadBufferTest.java} (60%) diff --git a/CHANGES.txt b/CHANGES.txt index 63d4533f91a3..47d510a5b8a2 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha3 + * Give each compressed scan reader its own read-ahead buffer instead of a shared per-thread cache (CASSANDRA-21671) * AccordExecutor improvements (CASSANDRA-21662) * Upgrade async-profiler to 4.5 (CASSANDRA-21630) * Guardrail configurations of zero are now treated as zero instead of unlimited (CASSANDRA-21517) diff --git a/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java b/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java index 034fbb71532f..7c7f551d06bd 100644 --- a/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java +++ b/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java @@ -206,10 +206,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 +278,30 @@ public static class Direct extends CompressedChunkReader private final CompressedReader reader; private final CompressedReader scanReader; + private final int blockSize; + private final int readAheadBufferSize; 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); - int readAheadBufferSize = DatabaseDescriptor.getCompressedReadAheadBufferSize(); - this.scanReader = (readAheadBufferSize > 0 && readAheadBufferSize > metadata.chunkLength()) - ? new ScanCompressedReader(channel, - new DirectThreadLocalByteBufferHolder(blockSize), - new DirectThreadLocalReadAheadBuffer(channel, readAheadBufferSize, blockSize)) - : null; + int size = DatabaseDescriptor.getCompressedReadAheadBufferSize(); + this.readAheadBufferSize = (size > 0 && size > metadata.chunkLength()) ? size : 0; + this.scanReader = 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.channel, parent.metadata, parent.crcCheckChanceSupplier); + this.blockSize = parent.blockSize; + this.reader = parent.reader; + this.readAheadBufferSize = 0; + this.scanReader = scanReader; } @Override @@ -337,10 +348,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 @@ -366,26 +381,38 @@ public static class Standard extends CompressedChunkReader private final CompressedReader reader; private final CompressedReader scanReader; + private final int readAheadBufferSize; public Standard(ChannelProxy channel, CompressionMetadata metadata, DoubleSupplier crcCheckChanceSupplier) { super(channel, metadata, crcCheckChanceSupplier); reader = new RandomAccessCompressedReader(channel, metadata); - 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; + int size = DatabaseDescriptor.getCompressedReadAheadBufferSize(); + this.readAheadBufferSize = (size > 0 && size > metadata.chunkLength()) ? size : 0; + this.scanReader = null; + } + + // Per-scan view; see Direct for the ownership rationale. + private Standard(Standard parent, CompressedReader scanReader) + { + super(parent.channel, parent.metadata, parent.crcCheckChanceSupplier); + this.reader = parent.reader; + this.readAheadBufferSize = 0; + 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/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/test/unit/org/apache/cassandra/io/compress/DirectCompressedSequentialWriterTest.java b/test/unit/org/apache/cassandra/io/compress/DirectCompressedSequentialWriterTest.java index 68833f1530a0..e536875d20d8 100644 --- a/test/unit/org/apache/cassandra/io/compress/DirectCompressedSequentialWriterTest.java +++ b/test/unit/org/apache/cassandra/io/compress/DirectCompressedSequentialWriterTest.java @@ -436,7 +436,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) -> 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 From 3143ff54b4915b08c951d0d961c1c358bba08d02 Mon Sep 17 00:00:00 2001 From: Jon Haddad Date: Sun, 13 Sep 2026 18:09:16 -0700 Subject: [PATCH 2/2] CASSANDRA-21671: Make scans bypass the chunk cache CachingRebufferer.instantiateRebufferer(isScan) returned this, so it dropped the isScan flag. A scan with the chunk cache on therefore populated the cache with one-shot chunks and evicted hot data. Delegate to the source when isScan is true, so the scan bypasses the cache and uses its own read-ahead buffer. 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. Add ChunkCacheScanBypassTest: a scan adds nothing to the cache while a point read populates it, and a scan under the cache still reads through the read-ahead buffer. --- CHANGES.txt | 1 + .../apache/cassandra/cache/ChunkCache.java | 6 +- .../cache/ChunkCacheScanBypassTest.java | 190 ++++++++++++++++++ 3 files changed, 196 insertions(+), 1 deletion(-) create mode 100644 test/unit/org/apache/cassandra/cache/ChunkCacheScanBypassTest.java diff --git a/CHANGES.txt b/CHANGES.txt index 47d510a5b8a2..dadc4c86a888 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha3 + * Make scans bypass the chunk cache so compaction does not evict hot data (CASSANDRA-21671) * Give each compressed scan reader its own read-ahead buffer instead of a shared per-thread cache (CASSANDRA-21671) * AccordExecutor improvements (CASSANDRA-21662) * Upgrade async-profiler to 4.5 (CASSANDRA-21630) diff --git a/src/java/org/apache/cassandra/cache/ChunkCache.java b/src/java/org/apache/cassandra/cache/ChunkCache.java index 66782ac48bae..2c2fea99025f 100644 --- a/src/java/org/apache/cassandra/cache/ChunkCache.java +++ b/src/java/org/apache/cassandra/cache/ChunkCache.java @@ -260,7 +260,11 @@ public void invalidate(long position) @Override public Rebufferer instantiateRebufferer(boolean isScan) { - return this; + // A scan (compaction, index build) reads each chunk once and must not pollute the cache with + // one-shot chunks that evict hot data. Delegate to the source so 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. + return isScan ? source.instantiateRebufferer(true) : this; } @Override 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..f8931e439c23 --- /dev/null +++ b/test/unit/org/apache/cassandra/cache/ChunkCacheScanBypassTest.java @@ -0,0 +1,190 @@ +/* + * 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.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 scan (compaction, index build) must bypass the chunk cache so it does not evict hot data with one-shot + * chunks. CachingRebufferer.instantiateRebufferer(isScan) must delegate to the source for a scan, which routes + * the scan through its own read-ahead buffer instead of the cache. 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(true); + 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 point read reuses the caching rebufferer and populates the cache, proving the cache works. + Rebufferer point = factory.instantiateRebufferer(false); + 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(true); + 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(); + } + } +}