From 1b5c3f7400e3acda5a84abf158d3879e5d3afe15 Mon Sep 17 00:00:00 2001 From: Jonathan Haddad Date: Wed, 9 Sep 2026 08:13:00 -0700 Subject: [PATCH 1/3] CASSANDRA-21671: Initialise bufferSize when reusing a cached ThreadLocalReadAheadBuffer Block Block objects are cached in a static thread-local map keyed by file path and shared across all ThreadLocalReadAheadBuffer instances. bufferSize is a per-instance field initialised only inside the block.buffer == null branch of getBlock(). When a second instance, on the same thread and for the same file path, reused an already-allocated cached Block, that branch was skipped and its bufferSize stayed -1. fill() then computed Math.min(remaining, -1) == -1 and called ByteBuffer.limit(-1), throwing IllegalArgumentException: newLimit < 0 and aborting compaction. Move the bufferSize initialisation out of the block.buffer == null guard so any instance that observes a Block, fresh or reused, initialises bufferSize from the buffer capacity. Add a regression test that reproduces the two-instance, same-path, same-thread scenario. --- .../io/util/ThreadLocalReadAheadBuffer.java | 16 +++++- .../util/ThreadLocalReadAheadBufferTest.java | 52 +++++++++++++++++++ 2 files changed, 66 insertions(+), 2 deletions(-) diff --git a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java index ff59f7cf96df..4dddb4b7b3e3 100644 --- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java +++ b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java @@ -29,6 +29,7 @@ import org.apache.cassandra.utils.Closeable; import org.apache.cassandra.utils.memory.MemoryUtil; +import com.google.common.annotations.VisibleForTesting; import io.netty.util.concurrent.FastThreadLocal; public class ThreadLocalReadAheadBuffer implements Closeable @@ -73,6 +74,12 @@ public boolean hasBuffer() return block().buffer != null; } + @VisibleForTesting + int bufferSize() + { + return bufferSize; + } + public int remaining() { return getBlock().buffer.remaining(); @@ -90,9 +97,14 @@ private Block getBlock() { block.buffer = bufferSupplier.get(); block.buffer.clear(); - if (bufferSize == -1) - bufferSize = block.buffer.capacity(); } + // bufferSize is a per-instance field, but Block objects are cached in a static + // thread-local map keyed by file path and shared across instances. When this + // instance reuses a Block allocated by an earlier instance for the same path, + // block.buffer is already non-null, so bufferSize must still be initialised here; + // leaving it at -1 makes fill() call ByteBuffer.limit(-1) and abort compaction. + if (bufferSize == -1) + bufferSize = block.buffer.capacity(); return block; } diff --git a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java b/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java index 47bba7b6c534..8ae0633ca3e1 100644 --- a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java +++ b/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java @@ -94,6 +94,58 @@ public void testReadsLikeChannelProxy() .checkAssert(this::testReads); } + @Test + public void testReusedCachedBlockInitialisesBufferSize() throws CorruptBlockException + { + // Block objects are cached in a static thread-local map keyed by file path and + // shared across instances. A second instance on the same thread for the same path + // reuses the first instance's Block, so block.buffer is already non-null. If + // bufferSize is only initialised in the block.buffer == null branch, the second + // instance keeps bufferSize == -1 and fill() calls ByteBuffer.limit(-1). + try (ChannelProxy channel = new ChannelProxy(files[0])) + { + int bufferSize = new DataStorageSpec.IntKibibytesBound("256KiB").toBytes(); + + // Instance A allocates and populates the cached Block for this file path. + ThreadLocalReadAheadBuffer a = new ThreadLocalReadAheadBuffer(channel, bufferSize, BufferType.OFF_HEAP); + ThreadLocalReadAheadBuffer b = new ThreadLocalReadAheadBuffer(channel, bufferSize, BufferType.OFF_HEAP); + try + { + a.fill(0); + + // B must see A's already-populated Block; this proves the shared-cache + // reuse that the bug depends on actually happens on this thread and path. + Assert.assertTrue("B should reuse A's cached Block", b.hasBuffer()); + + int readSize = 100; + ByteBuffer expected = ByteBuffer.allocate(readSize); + channel.read(expected, 0); + expected.flip(); + + // Instance B reuses A's cached Block without allocating first. + ByteBuffer actual = ByteBuffer.allocate(readSize); + b.fill(0); + b.read(actual, readSize); + actual.flip(); + + Assert.assertEquals(expected, actual); + + // A reused Block self-corrects reads on each fill(), so byte equality alone + // passes for any positive bufferSize. Pin the exact invariant the fix + // restores: a reused instance initialises bufferSize from the buffer + // capacity, not -1 and not some other value. + Assert.assertEquals("reused instance must initialise bufferSize from capacity", + bufferSize, b.bufferSize()); + } + finally + { + // Keep A open while B runs so the shared Block stays cached; close both here. + b.close(); + a.close(); + } + } + } + protected void testReads(InputData propertyInputs) { try (ChannelProxy channel = new ChannelProxy(propertyInputs.file); From 732f6ae57c2da4d0672463e9e6b27fdcbca96522 Mon Sep 17 00:00:00 2001 From: Jonathan Haddad Date: Wed, 9 Sep 2026 09:06:48 -0700 Subject: [PATCH 2/3] CASSANDRA-21671: Free reused read-ahead slices by their root allocation Second defect in the same read-ahead Direct IO family as Bug #01. The static per-thread, per-path Block cache in ThreadLocalReadAheadBuffer is shared across instances with different buffer ownership: - DirectThreadLocalReadAheadBuffer allocates an aligned SLICE via BufferUtil.allocateDirectAligned (no Cleaner; attachment = backing DirectByteBuffer) and previously overrode cleanBuffer() to clean the attachment. - Base ThreadLocalReadAheadBuffer allocates an OWNED buffer (has a Cleaner) and cleanBuffer() calls MemoryUtil.clean(buffer) directly. CompressedChunkReader.Direct uses the Direct subclass; .Standard uses the base. Both can open over the same file path on the same thread and share the same cached Block. When a Direct instance populates the Block with a slice and a Standard instance over the same path reuses it (the reuse path Bug #01 exposed), the Standard instance frees the slice through the base cleanBuffer, calling MemoryUtil.clean(slice). The hardened MemoryUtil.clean rejects a buffer with no cleaner and a non-null attachment (a latent double-free pre-hardening), throwing IllegalArgumentException and aborting BTI compaction. Make cleanup type-safe in the base class: cleanBuffer() resolves a direct buffer with no cleaner but a ByteBuffer attachment to its root allocation before calling MemoryUtil.clean, so any instance frees any buffer correctly. This does not weaken MemoryUtil.clean's guard; it resolves to the root before calling it. The DirectThreadLocalReadAheadBuffer.cleanBuffer override is now redundant and is removed so both paths use the single correct base implementation. The Bug #01 bufferSize fix is unchanged. Add a regression test that reproduces the Direct-slice/base-free scenario. --- CHANGES.txt | 1 + .../DirectThreadLocalReadAheadBuffer.java | 14 ++----- .../io/util/ThreadLocalReadAheadBuffer.java | 16 ++++++++ .../util/ThreadLocalReadAheadBufferTest.java | 41 +++++++++++++++++++ 4 files changed, 61 insertions(+), 11 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 1d76b5886821..16e9e7ff0599 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha3 + * Fix ThreadLocalReadAheadBuffer shared per-path Block cache aborting BTI + Direct IO compactions (CASSANDRA-21671) * Guardrail configurations of zero are now treated as zero instead of unlimited (CASSANDRA-21517) * Support multi-cell columns in cursor compaction (CASSANDRA-21463) * Optimize authorization logic for BatchStatement and TransactionStatement when a single table is updated (CASSANDRA-21606) diff --git a/src/java/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBuffer.java b/src/java/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBuffer.java index 934e3620e114..dd82cf86313f 100644 --- a/src/java/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBuffer.java +++ b/src/java/org/apache/cassandra/io/util/DirectThreadLocalReadAheadBuffer.java @@ -24,9 +24,6 @@ import org.agrona.BufferUtil; import org.apache.cassandra.io.sstable.CorruptSSTableException; -import org.apache.cassandra.utils.memory.MemoryUtil; - -import sun.nio.ch.DirectBuffer; public final class DirectThreadLocalReadAheadBuffer extends ThreadLocalReadAheadBuffer { @@ -50,11 +47,6 @@ protected void loadBlock(ByteBuffer blockBuffer, long blockPosition, int sizeToR throw new CorruptSSTableException(null, channel.filePath()); } - @Override - protected void cleanBuffer(ByteBuffer buffer) - { - // Aligned buffers from BufferUtil.allocateDirectAligned are slices; clean the backing buffer (attachment) - MemoryUtil.clean((ByteBuffer) ((DirectBuffer) buffer).attachment()); - } - -} \ No newline at end of file + // cleanBuffer() is inherited: the base implementation resolves an aligned slice to its + // backing allocation, so both Direct and Standard read-ahead instances free correctly. +} diff --git a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java index 4dddb4b7b3e3..1d8cfa4c88dd 100644 --- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java +++ b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java @@ -31,6 +31,7 @@ import com.google.common.annotations.VisibleForTesting; import io.netty.util.concurrent.FastThreadLocal; +import sun.nio.ch.DirectBuffer; public class ThreadLocalReadAheadBuffer implements Closeable { @@ -178,6 +179,21 @@ public void clear(boolean deallocate) protected void cleanBuffer(ByteBuffer buffer) { + // Block objects are cached in a static thread-local map keyed by file path and shared + // across instances. A DirectThreadLocalReadAheadBuffer stores an aligned SLICE (no + // cleaner; attachment = the backing DirectByteBuffer) in the shared Block. A base + // instance that reuses that Block for the same path must not free the slice as if it + // owned its memory; MemoryUtil.clean() rejects that and it is a latent double-free. + // Resolve to the root allocation here so any instance frees any buffer correctly. + if (buffer != null && buffer.isDirect()) + { + DirectBuffer db = (DirectBuffer) buffer; + if (db.cleaner() == null && db.attachment() instanceof ByteBuffer) + { + MemoryUtil.clean((ByteBuffer) db.attachment()); + return; + } + } MemoryUtil.clean(buffer); } diff --git a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java b/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java index 8ae0633ca3e1..361855ca4a9f 100644 --- a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java +++ b/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java @@ -146,6 +146,47 @@ public void testReusedCachedBlockInitialisesBufferSize() throws CorruptBlockExce } } + @Test + public void testReusedSliceIsFreedByBaseInstance() throws CorruptBlockException + { + // The shared per-thread, per-path Block cache is used by instances with different + // buffer ownership. A DirectThreadLocalReadAheadBuffer stores an aligned slice + // (no cleaner; attachment = backing DirectByteBuffer) in the Block. A base + // ThreadLocalReadAheadBuffer over the same path reuses that slice and, on close, + // frees it through the base cleanup path. If that path assumes the buffer owns its + // memory it calls MemoryUtil.clean(slice), which throws. + File file = files[0]; + int blockSize = FileUtils.getFileBlockSize(file); + int bufferSize = blockSize * 64; + try (ChannelProxy directChannel = new ChannelProxy(file, ChannelProxy.IOMode.DIRECT); + ChannelProxy standardChannel = new ChannelProxy(file)) + { + // Instance A (Direct) puts an aligned slice into the shared cached Block. + DirectThreadLocalReadAheadBuffer a = new DirectThreadLocalReadAheadBuffer(directChannel, bufferSize, blockSize); + ThreadLocalReadAheadBuffer b = new ThreadLocalReadAheadBuffer(standardChannel, bufferSize, BufferType.OFF_HEAP); + try + { + a.allocateBuffer(); + a.fill(0); + + // Instance B (base) reuses A's slice and frees it via the base cleanup path. + // Before the fix this throws IllegalArgumentException from MemoryUtil.clean. + b.close(); + + // B.close() must have freed the slice AND removed the shared Block from the + // map. A therefore sees no cached buffer, which makes A.close() below a + // genuine no-op rather than a silent second free of the same slice. + Assert.assertFalse("base close must free and remove the shared Block", a.hasBuffer()); + } + finally + { + // B.close() removed the shared Block from the map, so this is a safe no-op + // and must not double-free or throw. + a.close(); + } + } + } + protected void testReads(InputData propertyInputs) { try (ChannelProxy channel = new ChannelProxy(propertyInputs.file); From 4e7491017d986de7df1f5a3ff8715f0d9cd53c81 Mon Sep 17 00:00:00 2001 From: Jonathan Haddad Date: Thu, 10 Sep 2026 11:06:48 -0700 Subject: [PATCH 3/3] CASSANDRA-21671: Fix checkstyle import order in ThreadLocalReadAheadBuffer --- .../apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java index 1d8cfa4c88dd..ad545e98c961 100644 --- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java +++ b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java @@ -23,13 +23,14 @@ 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 com.google.common.annotations.VisibleForTesting; import io.netty.util.concurrent.FastThreadLocal; import sun.nio.ch.DirectBuffer;