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 ff59f7cf96df..ad545e98c961 100644 --- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java +++ b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java @@ -23,6 +23,8 @@ 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; @@ -30,6 +32,7 @@ import org.apache.cassandra.utils.memory.MemoryUtil; import io.netty.util.concurrent.FastThreadLocal; +import sun.nio.ch.DirectBuffer; public class ThreadLocalReadAheadBuffer implements Closeable { @@ -73,6 +76,12 @@ public boolean hasBuffer() return block().buffer != null; } + @VisibleForTesting + int bufferSize() + { + return bufferSize; + } + public int remaining() { return getBlock().buffer.remaining(); @@ -90,9 +99,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; } @@ -166,6 +180,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 47bba7b6c534..361855ca4a9f 100644 --- a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java +++ b/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java @@ -94,6 +94,99 @@ 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(); + } + } + } + + @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);