21671 direct io + readahead exception fix

CHANGES.txt

@@ -1,4 +1,6 @@
 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)
  * Guardrail configurations of zero are now treated as zero instead of unlimited (CASSANDRA-21517)

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

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

src/java/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBuffer.java → 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());
     }
-
-}
+}

src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java → 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.
+ * <p>
+ * 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<ByteBuffer> bufferSupplier;
-
-    private static final FastThreadLocal<Map<String, Block>> blockMap = new FastThreadLocal<>()
-    {
-        @Override
-        protected Map<String, Block> 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<ByteBuffer> bufferSupplier)
+    public ReadAheadBuffer(ChannelProxy channel, Supplier<ByteBuffer> bufferSupplier)
     {
         this.channel = channel;
         this.channelSize = channel.size();
@@ -70,41 +61,39 @@ public ThreadLocalReadAheadBuffer(ChannelProxy channel, Supplier<ByteBuffer> 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());
     }
 }