This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch cassandra-5.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cassandra-5.0 by this push:
new de8fe4b5d9 Fix compressed scan read-ahead buffer Block leak and
chunk-cache pollution by scans
de8fe4b5d9 is described below
commit de8fe4b5d931c37a261912e11d994a6a45c9ef7e
Author: Jon Haddad <[email protected]>
AuthorDate: Mon Sep 14 19:23:44 2026 -0700
Fix compressed scan read-ahead buffer Block leak and chunk-cache pollution
by scans
A shared thread-local read-ahead buffer allowed concurrent scans of one
SSTable to collide and leak. Each compressed scan reader
now owns a per-scan buffer that it frees on close. Scans also polluted the
chunk cache with one-shot chunks that evicted hot data.
patch by Jon Haddad; reviewed by Caleb Rackliffe, Francisco Guerrero, and
Sam Lightfoot for CASSANDRA-21671
Co-authored-by: Jon Haddad <[email protected]>
Co-authored-by: Caleb Rackliffe <[email protected]>
Co-Authored-By: Francisco Guerrero <[email protected]>
Assisted-by: Claude Code:claude-opus-4-8
---
CHANGES.txt | 1 +
.../org/apache/cassandra/cache/ChunkCache.java | 3 +-
.../index/sai/StorageAttachedIndexBuilder.java | 2 +-
.../cassandra/index/sasi/SASIIndexBuilder.java | 2 +-
.../cassandra/io/sstable/format/SSTableReader.java | 6 +
.../io/sstable/format/SortedTableScrubber.java | 4 +-
.../io/sstable/format/SortedTableVerifier.java | 4 +-
.../io/sstable/format/bti/PartitionIterator.java | 2 +-
.../cassandra/io/util/CompressedChunkReader.java | 65 +++--
.../org/apache/cassandra/io/util/FileHandle.java | 22 +-
...alReadAheadBuffer.java => ReadAheadBuffer.java} | 130 +++++-----
.../cassandra/io/util/SimpleChunkReader.java | 2 +-
.../io/sstable/format/ForwardingSSTableReader.java | 12 +
.../cassandra/cache/ChunkCacheScanBypassTest.java | 225 +++++++++++++++++
.../io/util/CompressedChunkReaderTest.java | 187 ++++++++++----
.../cassandra/io/util/ReadAheadBufferTest.java | 276 +++++++++++++++++++++
.../io/util/ThreadLocalReadAheadBufferTest.java | 207 ----------------
17 files changed, 782 insertions(+), 368 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index fe6da8ff9f..dcac3d5e07 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
5.0.10
+ * Fix compressed scan read-ahead buffer Block leak and chunk-cache pollution
by scans (CASSANDRA-21671)
* A read on a replica that does not know the index yet fails with a
NullPointerException (CASSANDRA-21658)
* Release hints write buffer when the write executor shuts down
(CASSANDRA-21657)
* Fix concurrent SAI vector inserts failing once jvector's per-graph pool
limit is exceeded (CASSANDRA-21644)
diff --git a/src/java/org/apache/cassandra/cache/ChunkCache.java
b/src/java/org/apache/cassandra/cache/ChunkCache.java
index 5fb0bc6a2f..3454c5c333 100644
--- a/src/java/org/apache/cassandra/cache/ChunkCache.java
+++ b/src/java/org/apache/cassandra/cache/ChunkCache.java
@@ -261,7 +261,8 @@ public class ChunkCache
@Override
public Rebufferer instantiateRebufferer(boolean isScan)
{
- return this;
+ // Bypass the cache for scans, so they don't evict hot data with
chunks that may never be revisited.
+ return isScan ? source.instantiateRebufferer(isScan) : this;
}
@Override
diff --git
a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuilder.java
b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuilder.java
index dd94272700..c7ee124063 100644
--- a/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuilder.java
+++ b/src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuilder.java
@@ -142,7 +142,7 @@ public class StorageAttachedIndexBuilder extends
SecondaryIndexBuilder
return false;
}
- try (RandomAccessReader dataFile = sstable.openDataReader();
+ try (RandomAccessReader dataFile = sstable.openDataReaderForScan();
LifecycleTransaction txn =
LifecycleTransaction.offline(OperationType.INDEX_BUILD, sstable))
{
perSSTableFileLock = shouldWritePerSSTableFiles(sstable);
diff --git a/src/java/org/apache/cassandra/index/sasi/SASIIndexBuilder.java
b/src/java/org/apache/cassandra/index/sasi/SASIIndexBuilder.java
index 4e71538e41..c31f1aca16 100644
--- a/src/java/org/apache/cassandra/index/sasi/SASIIndexBuilder.java
+++ b/src/java/org/apache/cassandra/index/sasi/SASIIndexBuilder.java
@@ -80,7 +80,7 @@ class SASIIndexBuilder extends SecondaryIndexBuilder
SSTableReader sstable = e.getKey();
Map<ColumnMetadata, ColumnIndex> indexes = e.getValue();
- try (RandomAccessReader dataFile = sstable.openDataReader())
+ try (RandomAccessReader dataFile = sstable.openDataReaderForScan())
{
PerSSTableIndexWriter indexWriter =
SASIIndex.newWriter(keyValidator, sstable.descriptor, indexes,
OperationType.COMPACTION);
targetDirectory = indexWriter.getDescriptor().directory.path();
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 96010ded32..854a1b486e 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
@@ -1385,6 +1385,12 @@ public abstract class SSTableReader extends SSTable
implements UnfilteredSource,
return dfile.createReaderForScan();
}
+ public RandomAccessReader openDataReaderForScan(RateLimiter limiter)
+ {
+ assert limiter != null;
+ return dfile.createReaderForScan(limiter);
+ }
+
public void trySkipFileCacheBefore(DecoratedKey key)
{
long position = getPosition(key, SSTableReader.Operator.GE);
diff --git
a/src/java/org/apache/cassandra/io/sstable/format/SortedTableScrubber.java
b/src/java/org/apache/cassandra/io/sstable/format/SortedTableScrubber.java
index a83527509f..66a7a43108 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/SortedTableScrubber.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SortedTableScrubber.java
@@ -153,8 +153,8 @@ public abstract class SortedTableScrubber<R extends
SSTableReaderWithFilter> imp
// partition header (key or data size) is corrupt. (This means our
position in the index file will be one
// partition "ahead" of the data file.)
this.dataFile = transaction.isOffline()
- ? sstable.openDataReader()
- :
sstable.openDataReader(CompactionManager.instance.getRateLimiter());
+ ? sstable.openDataReaderForScan()
+ :
sstable.openDataReaderForScan(CompactionManager.instance.getRateLimiter());
this.scrubInfo = new ScrubInfo(dataFile, sstable,
fileAccessLock.readLock());
diff --git
a/src/java/org/apache/cassandra/io/sstable/format/SortedTableVerifier.java
b/src/java/org/apache/cassandra/io/sstable/format/SortedTableVerifier.java
index a45dc585a9..6abb1d42b4 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/SortedTableVerifier.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SortedTableVerifier.java
@@ -95,8 +95,8 @@ public abstract class SortedTableVerifier<R extends
SSTableReaderWithFilter> imp
this.fileAccessLock = new ReentrantReadWriteLock();
this.dataFile = isOffline
- ? sstable.openDataReader()
- :
sstable.openDataReader(CompactionManager.instance.getRateLimiter());
+ ? sstable.openDataReaderForScan()
+ :
sstable.openDataReaderForScan(CompactionManager.instance.getRateLimiter());
this.verifyInfo = new VerifyInfo(dataFile, sstable,
fileAccessLock.readLock());
this.options = options;
this.isOffline = isOffline;
diff --git
a/src/java/org/apache/cassandra/io/sstable/format/bti/PartitionIterator.java
b/src/java/org/apache/cassandra/io/sstable/format/bti/PartitionIterator.java
index 17d1041ee2..605262a9c4 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/bti/PartitionIterator.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/bti/PartitionIterator.java
@@ -232,7 +232,7 @@ class PartitionIterator extends
PartitionIndex.IndexPosIterator implements KeyRe
private void seekDataInput(long pos) throws IOException
{
if (dataInput == null)
- dataInput = dataFile.createReader(pos);
+ dataInput = dataFile.createReader(pos, true);
else
dataInput.seek(pos);
}
diff --git a/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java
b/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java
index b6b3c9a6a2..5156e263f0 100644
--- a/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java
+++ b/src/java/org/apache/cassandra/io/util/CompressedChunkReader.java
@@ -49,6 +49,14 @@ public abstract class CompressedChunkReader extends
AbstractReaderFileProxy impl
assert Integer.bitCount(metadata.chunkLength()) == 1; //must be a
power of two
}
+ protected CompressedChunkReader(CompressedChunkReader parent)
+ {
+ super(parent.channel, parent.metadata.dataLength);
+ this.metadata = parent.metadata;
+ this.maxCompressedLength = parent.maxCompressedLength;
+ this.crcCheckChanceSupplier = parent.crcCheckChanceSupplier;
+ }
+
protected CompressedChunkReader forScan()
{
return this;
@@ -97,25 +105,18 @@ public abstract class CompressedChunkReader extends
AbstractReaderFileProxy impl
protected interface CompressedReader extends Closeable
{
- default void allocateResources()
- {
- }
-
default void deallocateResources()
{
}
- default boolean allocated()
- {
- return false;
- }
-
default void close()
{
}
-
+ /**
+ * The returned buffer is only valid until the next call to read().
Callers must consume the data immediately.
+ */
ByteBuffer read(CompressionMetadata.Chunk chunk, boolean
shouldCheckCrc) throws CorruptBlockException;
}
@@ -157,13 +158,13 @@ public abstract class CompressedChunkReader extends
AbstractReaderFileProxy impl
{
private final ChannelProxy channel;
private final ThreadLocalByteBufferHolder bufferHolder;
- private final ThreadLocalReadAheadBuffer readAheadBuffer;
+ private final ReadAheadBuffer readAheadBuffer;
private ScanCompressedReader(ChannelProxy channel, CompressionMetadata
metadata, int readAheadBufferSize)
{
this.channel = channel;
this.bufferHolder = new
ThreadLocalByteBufferHolder(metadata.compressor().preferredBufferType());
- this.readAheadBuffer = new ThreadLocalReadAheadBuffer(channel,
readAheadBufferSize, metadata.compressor().preferredBufferType());
+ this.readAheadBuffer = new ReadAheadBuffer(channel,
readAheadBufferSize, metadata.compressor().preferredBufferType());
}
@Override
@@ -198,24 +199,13 @@ public abstract class CompressedChunkReader extends
AbstractReaderFileProxy impl
return compressed;
}
- @Override
- public void allocateResources()
- {
- readAheadBuffer.allocateBuffer();
- }
-
@Override
public void deallocateResources()
{
- readAheadBuffer.clear(true);
+ readAheadBuffer.close();
}
@Override
- public boolean allocated()
- {
- return readAheadBuffer.hasBuffer();
- }
-
public void close()
{
readAheadBuffer.close();
@@ -224,26 +214,35 @@ public abstract class CompressedChunkReader extends
AbstractReaderFileProxy impl
public static class Standard extends CompressedChunkReader
{
-
private final CompressedReader reader;
private final CompressedReader scanReader;
+ private final int readAheadBufferSize;
public Standard(ChannelProxy channel, CompressionMetadata metadata,
Supplier<Double> crcCheckChanceSupplier)
{
super(channel, metadata, crcCheckChanceSupplier);
- reader = new RandomAccessCompressedReader(channel, metadata);
+ this.reader = new RandomAccessCompressedReader(channel, metadata);
+ this.scanReader = null;
+ int size = DatabaseDescriptor.getCompressedReadAheadBufferSize();
+ this.readAheadBufferSize = (size > 0 && size >
metadata.chunkLength()) ? size : 0;
+ }
- int readAheadBufferSize =
DatabaseDescriptor.getCompressedReadAheadBufferSize();
- scanReader = (readAheadBufferSize > 0 && readAheadBufferSize >
metadata.chunkLength())
- ? new ScanCompressedReader(channel, metadata,
readAheadBufferSize) : null;
+ private Standard(Standard parent, CompressedReader scanReader)
+ {
+ super(parent);
+ this.reader = parent.reader;
+ this.scanReader = scanReader;
+ this.readAheadBufferSize = 0;
}
+ @Override
protected CompressedChunkReader forScan()
{
- if (scanReader != null)
- scanReader.allocateResources();
+ if (readAheadBufferSize == 0)
+ return this;
- return this;
+ ScanCompressedReader scan = new ScanCompressedReader(channel,
metadata, readAheadBufferSize);
+ return new Standard(this, scan);
}
@Override
@@ -265,7 +264,7 @@ public abstract class CompressedChunkReader extends
AbstractReaderFileProxy impl
CompressionMetadata.Chunk chunk = metadata.chunkFor(position);
boolean shouldCheckCrc = shouldCheckCrc();
- CompressedReader readFrom = (scanReader != null &&
scanReader.allocated()) ? scanReader : reader;
+ CompressedReader readFrom = scanReader != null ? scanReader :
reader;
if (chunk.length < maxCompressedLength)
{
ByteBuffer compressed = readFrom.read(chunk,
shouldCheckCrc);
diff --git a/src/java/org/apache/cassandra/io/util/FileHandle.java
b/src/java/org/apache/cassandra/io/util/FileHandle.java
index 7e4b214128..b9e3f233fa 100644
--- a/src/java/org/apache/cassandra/io/util/FileHandle.java
+++ b/src/java/org/apache/cassandra/io/util/FileHandle.java
@@ -136,7 +136,12 @@ public class FileHandle extends SharedCloseableImpl
public RandomAccessReader createReaderForScan()
{
- return createReader(null, true);
+ return createReaderForScan(null);
+ }
+
+ public RandomAccessReader createReaderForScan(RateLimiter limiter)
+ {
+ return createReader(limiter, true);
}
/**
@@ -151,14 +156,19 @@ public class FileHandle extends SharedCloseableImpl
return createReader(limiter, false);
}
- public RandomAccessReader createReader(RateLimiter limiter, boolean
forScan)
+ public RandomAccessReader createReader(RateLimiter limiter, boolean isScan)
{
- return new RandomAccessReader(instantiateRebufferer(limiter, forScan));
+ return new RandomAccessReader(instantiateRebufferer(limiter, isScan));
}
public FileDataInput createReader(long position)
{
- RandomAccessReader reader = createReader();
+ return createReader(position, false);
+ }
+
+ public FileDataInput createReader(long position, boolean isScan)
+ {
+ RandomAccessReader reader = createReader(null, isScan);
try
{
reader.seek(position);
@@ -199,9 +209,9 @@ public class FileHandle extends SharedCloseableImpl
return instantiateRebufferer(limiter, false);
}
- public Rebufferer instantiateRebufferer(RateLimiter limiter, boolean
forScan)
+ public Rebufferer instantiateRebufferer(RateLimiter limiter, boolean
isScan)
{
- Rebufferer rebufferer =
rebuffererFactory.instantiateRebufferer(forScan);
+ Rebufferer rebufferer =
rebuffererFactory.instantiateRebufferer(isScan);
if (limiter != null)
rebufferer = new LimitingRebufferer(rebufferer, limiter,
DiskOptimizationStrategy.MAX_BUFFER_SIZE);
diff --git
a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java
b/src/java/org/apache/cassandra/io/util/ReadAheadBuffer.java
similarity index 53%
rename from
src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java
rename to src/java/org/apache/cassandra/io/util/ReadAheadBuffer.java
index 5e78697686..2e85757275 100644
--- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java
+++ b/src/java/org/apache/cassandra/io/util/ReadAheadBuffer.java
@@ -19,39 +19,31 @@
package org.apache.cassandra.io.util;
import java.nio.ByteBuffer;
-import java.util.HashMap;
-import java.util.Map;
-import io.netty.util.concurrent.FastThreadLocal;
+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;
-public final class ThreadLocalReadAheadBuffer
-{
- private static class Block
- {
- ByteBuffer buffer = null;
- int index = -1;
- }
+import javax.annotation.concurrent.NotThreadSafe;
+/**
+ * A read-ahead buffer for sequential scans of a single file.
+ */
+@NotThreadSafe
+public class ReadAheadBuffer implements Closeable
+{
private final ChannelProxy channel;
-
private final BufferType bufferType;
-
- private static final FastThreadLocal<Map<String, Block>> blockMap = new
FastThreadLocal<>()
- {
- @Override
- protected Map<String, Block> initialValue()
- {
- return new HashMap<>();
- }
- };
-
- private final int bufferSize;
private final long channelSize;
+ private final int bufferSize;
+
+ private ByteBuffer buffer;
+ private int index = -1;
+ private boolean released;
- public ThreadLocalReadAheadBuffer(ChannelProxy channel, int bufferSize,
BufferType bufferType)
+ public ReadAheadBuffer(ChannelProxy channel, int bufferSize, BufferType
bufferType)
{
this.channel = channel;
this.channelSize = channel.size();
@@ -61,42 +53,50 @@ public final class ThreadLocalReadAheadBuffer
public boolean hasBuffer()
{
- return block().buffer != null;
+ return buffer != null;
+ }
+
+ @VisibleForTesting
+ int bufferSize()
+ {
+ return bufferSize;
}
- /**
- * Safe to call only if {@link #hasBuffer()} is true
- */
public int remaining()
{
- return getBlock().buffer.remaining();
+ return allocatedBuffer().remaining();
}
- public void allocateBuffer()
+ @VisibleForTesting
+ void allocateBuffer()
{
- getBlock();
+ getBuffer();
}
- private Block getBlock()
+ private ByteBuffer getBuffer()
{
- Block block = block();
- if (block.buffer == null)
+ if (released)
+ throw new IllegalStateException("read-ahead buffer for " +
channel.filePath() + " has been released");
+
+ if (buffer == null)
{
- block.buffer = bufferType.allocate(bufferSize);
- block.buffer.clear();
+ buffer = bufferType.allocate(bufferSize);
+ buffer.clear();
}
- return block;
+ return buffer;
}
- private Block block()
+ private ByteBuffer allocatedBuffer()
{
- return blockMap.get().computeIfAbsent(channel.filePath(), k -> new
Block());
+ if (buffer == null)
+ throw new IllegalStateException("read-ahead buffer for " +
channel.filePath() + " is not allocated");
+
+ 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);
@@ -105,14 +105,11 @@ public final class ThreadLocalReadAheadBuffer
long remaining = channelSize - blockPosition;
int sizeToRead = (int) Math.min(remaining, bufferSize);
- if (block.index != blockNo)
+ if (index != blockNo)
{
blockBuffer.flip();
- blockBuffer.limit(sizeToRead);
- if (channel.read(blockBuffer, blockPosition) != sizeToRead)
- throw new CorruptSSTableException(null, channel.filePath());
-
- block.index = blockNo;
+ loadBlock(blockBuffer, blockPosition, sizeToRead);
+ index = blockNo;
}
blockBuffer.flip();
@@ -120,10 +117,16 @@ public final class ThreadLocalReadAheadBuffer
blockBuffer.position((int) (position - blockPosition));
}
+ protected void loadBlock(ByteBuffer blockBuffer, long blockPosition, int
sizeToRead) throws CorruptBlockException
+ {
+ blockBuffer.limit(sizeToRead);
+ if (channel.read(blockBuffer, blockPosition) != sizeToRead)
+ throw new CorruptBlockException(channel.filePath(), blockPosition,
sizeToRead);
+ }
+
public int read(ByteBuffer dest, int length)
{
- Block block = getBlock();
- ByteBuffer blockBuffer = block.buffer;
+ ByteBuffer blockBuffer = allocatedBuffer();
ByteBuffer tmp = blockBuffer.duplicate();
tmp.limit(tmp.position() + length);
dest.put(tmp);
@@ -132,29 +135,16 @@ public final class ThreadLocalReadAheadBuffer
return length;
}
- public void clear(boolean deallocate)
+ @Override
+ public void close()
{
- // avoid calling block() here to reduce unintended allocations
- Block block = blockMap.get().get(channel.filePath());
- if (block == null)
+ released = true;
+ if (buffer == null)
return;
- block.index = -1;
- if (block.buffer == null)
- return;
-
- ByteBuffer blockBuffer = block.buffer;
- blockBuffer.clear();
- if (deallocate)
- {
- FileUtils.clean(blockBuffer);
- block.buffer = null;
- }
- }
-
- public void close()
- {
- clear(true);
- blockMap.get().remove(channel.filePath());
+ index = -1;
+ buffer.clear();
+ FileUtils.clean(buffer);
+ buffer = null;
}
}
diff --git a/src/java/org/apache/cassandra/io/util/SimpleChunkReader.java
b/src/java/org/apache/cassandra/io/util/SimpleChunkReader.java
index fec1216bf4..3daa19c37a 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 @@ class SimpleChunkReader extends AbstractReaderFileProxy
implements ChunkReader
}
@Override
- public Rebufferer instantiateRebufferer(boolean forScan)
+ public Rebufferer instantiateRebufferer(boolean isScan)
{
if (Integer.bitCount(bufferSize) == 1)
return new BufferManagingRebufferer.Aligned(this);
diff --git
a/test/distributed/org/apache/cassandra/io/sstable/format/ForwardingSSTableReader.java
b/test/distributed/org/apache/cassandra/io/sstable/format/ForwardingSSTableReader.java
index 710a426647..0d538d731a 100644
---
a/test/distributed/org/apache/cassandra/io/sstable/format/ForwardingSSTableReader.java
+++
b/test/distributed/org/apache/cassandra/io/sstable/format/ForwardingSSTableReader.java
@@ -537,6 +537,18 @@ public abstract class ForwardingSSTableReader extends
SSTableReader
return delegate.openDataReader();
}
+ @Override
+ public RandomAccessReader openDataReaderForScan()
+ {
+ return delegate.openDataReaderForScan();
+ }
+
+ @Override
+ public RandomAccessReader openDataReaderForScan(RateLimiter limiter)
+ {
+ return delegate.openDataReaderForScan(limiter);
+ }
+
@Override
public void trySkipFileCacheBefore(DecoratedKey key)
{
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 0000000000..8a25967827
--- /dev/null
+++ b/test/unit/org/apache/cassandra/cache/ChunkCacheScanBypassTest.java
@@ -0,0 +1,225 @@
+/*
+ * 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.junit.After;
+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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ChunkCacheScanBypassTest
+{
+ /** The minimum legal read-ahead buffer size. */
+ private static final int READ_AHEAD_BUFFER_SIZE_IN_KB = 256;
+
+ static
+ {
+ DatabaseDescriptor.clientInitialization();
+ // Client initialization leaves the cache size at zero, which would
make a point read spin forever.
+ DatabaseDescriptor.getRawConfig().file_cache_size = new
DataStorageSpec.IntMebibytesBound(64);
+ }
+
+ private final int originalReadAheadBufferSizeInKb =
DatabaseDescriptor.getCompressedReadAheadBufferSizeInKB();
+
+ @After
+ public void restoreReadAhead()
+ {
+
DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(originalReadAheadBufferSizeInKb);
+ }
+
+ @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);
+
+ Rebufferer scan = factory.instantiateRebufferer(true);
+ assertThat(scan).isNotSameAs(factory);
+ try
+ {
+ rebufferWholeFile(scan, reader.fileLength(),
reader.chunkSize());
+ assertThat(cache.size()).as("a scan must not populate the
chunk cache").isZero();
+ }
+ finally
+ {
+ // Only release the scan's own buffer. The shared reader stays
open for the point read below.
+ scan.closeReader();
+ }
+
+ cache.close();
+
+ Rebufferer point = factory.instantiateRebufferer(false);
+ assertThat(point).isSameAs(factory);
+ rebufferWholeFile(point, reader.fileLength(), reader.chunkSize());
+ assertThat(cache.size()).as("a point read must populate the chunk
cache").isGreaterThan(0);
+ }
+ finally
+ {
+ cache.close();
+ }
+ }
+
+ /**
+ * A scan bypasses the cache, but it must still batch its reads through
its own read-ahead buffer.
+ */
+ @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();
+ }
+
+ assertThat((long) reads.get()).as("the scan must batch reads
through the read-ahead buffer").isLessThan(chunkCount);
+ assertThat(cache.size()).as("a scan must not populate the chunk
cache").isZero();
+ }
+ finally
+ {
+ cache.close();
+ }
+ }
+
+ @Test
+ public void scanBypassesCacheEvenWithoutReadAhead() throws Exception
+ {
+ // 0 disables read-ahead outright.
+ assertScanBypassesCache("/chunk_cache_scan_no_readahead", 0,
CompressionParams.snappy(4096, 1.1));
+
+ // Standard requires the buffer to be strictly larger than a chunk, so
a chunk length equal to the
+ // read-ahead buffer size also gets no read-ahead, even with the
setting untouched.
+ assertScanBypassesCache("/chunk_cache_scan_big_chunks",
READ_AHEAD_BUFFER_SIZE_IN_KB,
CompressionParams.snappy(READ_AHEAD_BUFFER_SIZE_IN_KB * 1024, 1.1));
+ }
+
+ private void assertScanBypassesCache(String fileNamePrefix, int
readAheadBufferSizeInKb, CompressionParams params) throws Exception
+ {
+ FileSystems.newGlobalInMemoryFileSystem();
+ File f = new File(fileNamePrefix + ".db");
+ CompressionMetadata metadata = writeCompressedFile(f, new
File(fileNamePrefix + ".offset"), new File(fileNamePrefix + ".digest"),
200_000, params);
+
DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(readAheadBufferSizeInKb);
+
+ 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);
+ Rebufferer scan = factory.instantiateRebufferer(true);
+ assertThat(scan).as("a scan must bypass the chunk
cache").isNotSameAs(factory);
+
+ rebufferWholeFile(scan, reader.fileLength(), reader.chunkSize());
+ assertThat(cache.size()).as("a scan must not populate the chunk
cache").isZero();
+ }
+ finally
+ {
+ cache.close();
+ }
+ }
+
+ private static ChunkCache newChunkCache() throws Exception
+ {
+ Constructor<ChunkCache> 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
+ {
+ return writeCompressedFile(f, offsets, digest, longsToWrite,
CompressionParams.snappy(4096, 1.1));
+ }
+
+ private static CompressionMetadata writeCompressedFile(File f, File
offsets, File digest, long longsToWrite, CompressionParams params) throws
Exception
+ {
+ SequentialWriterOption option =
SequentialWriterOption.newBuilder().finishOnClose(false).bufferSize(1 <<
10).build();
+
+ 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);
+ }
+ }
+
+ 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);
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java
b/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java
index 6d1cc35821..7b705483ad 100644
--- a/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java
+++ b/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java
@@ -18,8 +18,26 @@
package org.apache.cassandra.io.util;
+import java.nio.ByteBuffer;
+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;
+import org.junit.Assert;
+import org.junit.Test;
+
import accord.utils.Gen;
import accord.utils.Gens;
+
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ClusteringComparator;
import org.apache.cassandra.io.compress.CompressedSequentialWriter;
@@ -28,16 +46,6 @@ import
org.apache.cassandra.io.filesystem.ListenableFileSystem;
import org.apache.cassandra.io.sstable.CorruptSSTableException;
import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
import org.apache.cassandra.schema.CompressionParams;
-import org.assertj.core.api.Assertions;
-
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.nio.ByteBuffer;
-import java.nio.channels.FileChannel;
-import java.nio.file.StandardOpenOption;
-import java.nio.file.Files;
-import java.util.concurrent.atomic.AtomicInteger;
import static accord.utils.Property.qt;
@@ -49,19 +57,18 @@ public class CompressedChunkReaderTest
}
@Test
- public void scanReaderReadsLessThanRAReader()
+ public void scanReaderIssuesFewerPhysicalReadsThanRandomAccessReader()
{
var optionGen = options();
var paramsGen = params();
var lengthGen = Gens.longs().between(1, 1 << 16);
- qt().withSeed(-1871070464864118891L).forAll(Gens.random(), optionGen,
paramsGen).check((rs, option, params) -> {
+
+ qt().forAll(Gens.random(), optionGen, paramsGen).check((rs, option,
params) -> {
ListenableFileSystem fs =
FileSystems.newGlobalInMemoryFileSystem();
- File f = new File("/file.db");
+ File f = new File("/file.bin");
AtomicInteger reads = new AtomicInteger();
- fs.onPostRead(f.path::equals, (p, c, pos, dst, r) -> {
- reads.incrementAndGet();
- });
+ fs.onPostRead(f.path::equals, (p, c, pos, dst, r) ->
reads.incrementAndGet());
long length = lengthGen.nextLong(rs);
CompressionMetadata metadata1, metadata2;
try (CompressedSequentialWriter writer = new
CompressedSequentialWriter(f, new File("/file.offset"), new
File("/file.digest"), option, params, new MetadataCollector(new
ClusteringComparator())))
@@ -74,14 +81,14 @@ public class CompressedChunkReaderTest
metadata2 = writer.open(0);
}
- doReads(f, metadata1, length, true);
+ doReads(f, metadata1, length, false);
+ int randomAccessReads = reads.getAndSet(0);
+
+ doReads(f, metadata2, length, true);
int scanReads = reads.getAndSet(0);
- doReads(f, metadata2, length, false);
- int raReads = reads.getAndSet(0);
-
if (Files.size(f.toPath()) >
DatabaseDescriptor.getCompressedReadAheadBufferSize())
- Assert.assertTrue(scanReads < raReads);
+ Assert.assertTrue(scanReads <= randomAccessReads);
});
}
@@ -89,33 +96,42 @@ public class CompressedChunkReaderTest
{
ByteBuffer buffer = ByteBuffer.allocateDirect(metadata.chunkLength());
- try (ChannelProxy channel = new ChannelProxy(f);
- CompressedChunkReader reader = new
CompressedChunkReader.Standard(channel, metadata, () -> 1.1);
- metadata)
+ try (ChannelProxy channel = new ChannelProxy(f))
{
- if (useReadAhead)
- reader.forScan();
-
- long offset = 0;
- long maxOffset = length * Long.BYTES;
- do
+ try (CompressedChunkReader reader = new
CompressedChunkReader.Standard(channel, metadata, () -> 1d);
+ metadata)
{
- reader.readChunk(offset, buffer);
- for (long expected = offset / Long.BYTES;
buffer.hasRemaining(); expected++)
-
Assertions.assertThat(buffer.getLong()).isEqualTo(expected);
+ CompressedChunkReader scanReader = useReadAhead ?
reader.forScan() : reader;
+ try
+ {
+ 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
{
FileUtils.clean(buffer);
- }}
+ }
+ }
private static Gen<SequentialWriterOption> options()
{
- Gen<Integer> bufferSizes = Gens.constant(1 << 10); //.pickInt(1 << 4,
1 << 10, 1 << 15);
+ Gen<Integer> bufferSizes = Gens.constant(1 << 10);
return rs -> SequentialWriterOption.newBuilder()
.finishOnClose(false)
.bufferSize(bufferSizes.next(rs))
@@ -181,15 +197,100 @@ public class CompressedChunkReaderTest
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
{
FileUtils.clean(buffer);
}
}
-}
\ No newline at end of file
+
+ /*
+ * Two or more concurrent scans of one file must not corrupt each other or
double-free a shared buffer.
+ */
+ @Test(timeout = 60_000)
+ public void concurrentScansOfOneReaderAreIndependent() throws Exception
+ {
+
+ SequentialWriterOption writerOption =
SequentialWriterOption.newBuilder().finishOnClose(false).bufferSize(1 <<
10).build();
+ 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);
+ }
+
+ // Minium legal buffer size. (This should span many blocks over the
test file.)
+ DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(256);
+
+ 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<Future<?>> 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();
+ FileUtils.clean(buffer);
+ }
+ }));
+ }
+
+ for (Future<?> future : futures)
+ future.get(45, TimeUnit.SECONDS);
+ }
+ finally
+ {
+ pool.shutdownNow();
+ }
+ }
+ }
+}
diff --git a/test/unit/org/apache/cassandra/io/util/ReadAheadBufferTest.java
b/test/unit/org/apache/cassandra/io/util/ReadAheadBufferTest.java
new file mode 100644
index 0000000000..b8182135aa
--- /dev/null
+++ b/test/unit/org/apache/cassandra/io/util/ReadAheadBufferTest.java
@@ -0,0 +1,276 @@
+/*
+ * 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;
+
+
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.List;
+import java.util.Random;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.quicktheories.WithQuickTheories;
+import org.quicktheories.core.Gen;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.DataStorageSpec;
+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.Pair;
+
+import static java.lang.Math.max;
+import static
org.apache.cassandra.config.CassandraRelevantProperties.JAVA_IO_TMPDIR;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class ReadAheadBufferTest implements WithQuickTheories
+{
+ private static final Logger logger =
LoggerFactory.getLogger(ReadAheadBufferTest.class);
+
+ private static final int BUFFER_SIZE = new
DataStorageSpec.IntKibibytesBound("256KiB").toBytes();
+ private static final int NUM_FILES = 5;
+ protected static final File[] files = new File[NUM_FILES];
+ protected static Integer seed;
+
+ @BeforeClass
+ public static void setup()
+ {
+ seed = new Random().nextInt();
+ logger.info("Seed: {}", seed);
+
+ Random seeded = new Random(seed);
+ for (int i = 0; i < NUM_FILES; i++)
+ {
+ int size = seeded.nextInt((Integer.MAX_VALUE - 1) / 8);
+ files[i] = writeFile(seed, size);
+ }
+ }
+
+ @AfterClass
+ public static void cleanup()
+ {
+ for (File f : files)
+ {
+ try
+ {
+ f.delete();
+ }
+ catch (Exception e)
+ {
+ // ignore
+ }
+ }
+ }
+
+ @Test
+ public void testLastBlockReads()
+ {
+
qt().withFixedSeed(seed).forAll(lastBlockReads()).checkAssert(this::testReads);
+ }
+
+ @Test
+ public void testReadsLikeChannelProxy()
+ {
+ qt().withFixedSeed(seed).forAll(reads()).checkAssert(this::testReads);
+ }
+
+ @Test
+ public void allocateInitialisesBufferSizeFromCapacity()
+ {
+ try (ChannelProxy channel = new ChannelProxy(files[0]))
+ {
+ try (ReadAheadBuffer buffer = new ReadAheadBuffer(channel,
BUFFER_SIZE, BufferType.OFF_HEAP))
+ {
+ assertFalse(buffer.hasBuffer());
+
+ buffer.allocateBuffer();
+
+ // Allocation must record the actual buffer size, not leave it
unset.
+ assertTrue(buffer.hasBuffer());
+ assertEquals("allocate must initialise bufferSize from
capacity", BUFFER_SIZE, buffer.bufferSize());
+ }
+ }
+ }
+
+ @Test
+ public void independentInstancesDoNotShareBuffer() throws
CorruptBlockException
+ {
+
+ File file = files[0];
+ try (ChannelProxy channel = new ChannelProxy(file))
+ {
+ try (ReadAheadBuffer a = new ReadAheadBuffer(channel, BUFFER_SIZE,
BufferType.OFF_HEAP); ReadAheadBuffer b = new ReadAheadBuffer(channel,
BUFFER_SIZE, BufferType.OFF_HEAP))
+ {
+ a.fill(0);
+
+ // Advance b to a different block if the file is large enough;
else keep it on block 0.
+ long secondBlock = BUFFER_SIZE;
+ 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();
+
+ assertEquals(expected, actual);
+ }
+ }
+ }
+
+ @Test
+ public void closeFreesBufferAndIsIdempotent()
+ {
+ try (ChannelProxy channel = new ChannelProxy(files[0]))
+ {
+ ReadAheadBuffer buffer = new ReadAheadBuffer(channel, BUFFER_SIZE,
BufferType.OFF_HEAP);
+ buffer.allocateBuffer();
+ assertTrue(buffer.hasBuffer());
+
+ buffer.close();
+ assertFalse("close must free the owned buffer",
buffer.hasBuffer());
+
+ // A second close must not double-free.
+ buffer.close();
+ assertFalse(buffer.hasBuffer());
+ }
+ }
+
+ protected void testReads(InputData propertyInputs)
+ {
+ try (ChannelProxy channel = new ChannelProxy(propertyInputs.file);
+ ReadAheadBuffer rab = new ReadAheadBuffer(channel, BUFFER_SIZE,
BufferType.OFF_HEAP); )
+ {
+ for (Pair<Long, Integer> read : propertyInputs.positionsAndLengths)
+ {
+ testRead(read, channel, rab);
+ }
+ }
+ }
+
+ protected static void testRead(Pair<Long, Integer> read, ChannelProxy
bufferedChannel, ReadAheadBuffer rab)
+ {
+ int readSize = Math.min(read.right, (int) (bufferedChannel.size() -
read.left));
+ ByteBuffer buf1 = ByteBuffer.allocate(readSize);
+ bufferedChannel.read(buf1, read.left);
+
+ ByteBuffer buf2 = ByteBuffer.allocate(readSize);
+ try
+ {
+ int copied = 0;
+ while (copied < readSize)
+ {
+ rab.fill(read.left + copied);
+ int leftToRead = readSize - copied;
+ if (rab.remaining() >= leftToRead)
+ copied += rab.read(buf2, leftToRead);
+ else
+ copied += rab.read(buf2, rab.remaining());
+ }
+ }
+ catch (CorruptSSTableException | CorruptBlockException e)
+ {
+ throw new RuntimeException(e);
+ }
+
+ assertEquals(buf1, buf2);
+ }
+
+ protected Gen<InputData> reads()
+ {
+ return arbitrary().pick(List.of(files))
+ .flatMap((file) ->
+ lists().of(longs().between(0,
fileSize(file)).zip(integers().between(1, 100), Pair::create))
+ .ofSizeBetween(5, 10)
+ .map(positionsAndLengths -> new
InputData(file, positionsAndLengths)));
+ }
+
+ protected Gen<InputData> lastBlockReads()
+ {
+ return arbitrary().pick(List.of(files))
+ .flatMap((file) ->
+ lists().of(longs().between(max(0,
fileSize(file) - BUFFER_SIZE), fileSize(file)).zip(integers().between(1, 100),
Pair::create))
+ .ofSizeBetween(5, 10)
+ .map(positionsAndLengths -> new
InputData(file, positionsAndLengths)));
+ }
+
+ private long fileSize(File file)
+ {
+ try
+ {
+ return Files.size(file.toPath());
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected static class InputData
+ {
+
+ protected final File file;
+ protected final List<Pair<Long, Integer>> positionsAndLengths;
+
+ public InputData(File file, List<Pair<Long, Integer>>
positionsAndLengths)
+ {
+ this.file = file;
+ this.positionsAndLengths = positionsAndLengths;
+ }
+ }
+
+ private static File writeFile(int seed, int length)
+ {
+ String fileName = "data+" + length + ".bin";
+
+ byte[] dataChunk = new byte[4096 * 8];
+ java.util.Random random = new Random(seed);
+ int writtenData = 0;
+
+ File file = new File(JAVA_IO_TMPDIR.getString(), fileName);
+ try (FileOutputStream fos = new FileOutputStream(file.toJavaIOFile()))
+ {
+ while (writtenData < length)
+ {
+ random.nextBytes(dataChunk);
+ int toWrite = Math.min((length - writtenData),
dataChunk.length);
+ fos.write(dataChunk, 0, toWrite);
+ writtenData += toWrite;
+ }
+ fos.flush();
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+
+ return file;
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java
b/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java
deleted file mode 100644
index 2d29504d6f..0000000000
--- a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java
+++ /dev/null
@@ -1,207 +0,0 @@
-/*
- * 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;
-
-
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.nio.file.Files;
-import java.util.List;
-import java.util.Random;
-
-import org.junit.AfterClass;
-import org.junit.Assert;
-import org.junit.BeforeClass;
-import org.junit.Test;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.config.DataStorageSpec;
-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.Pair;
-import org.quicktheories.WithQuickTheories;
-import org.quicktheories.core.Gen;
-
-import static java.lang.Math.max;
-import static
org.apache.cassandra.config.CassandraRelevantProperties.JAVA_IO_TMPDIR;
-
-public class ThreadLocalReadAheadBufferTest implements WithQuickTheories
-{
- private static final int numFiles = 5;
- private static final File[] files = new File[numFiles];
- private static final Logger logger =
LoggerFactory.getLogger(ThreadLocalReadAheadBufferTest.class);
- private static Integer seed;
-
- @BeforeClass
- public static void setup()
- {
- seed = new Random().nextInt();
- logger.info("Seed: {}", seed);
-
- for (int i = 0; i < numFiles; i++)
- {
- int size = new Random(seed).nextInt((Integer.MAX_VALUE - 1) / 8);
- files[i] = writeFile(seed, size);
- }
- }
-
- @AfterClass
- public static void cleanup()
- {
- for (File f : files)
- {
- try
- {
- f.delete();
- }
- catch (Exception e)
- {
- // ignore
- }
- }
- }
-
- @Test
- public void testLastBlockReads()
- {
- qt().withFixedSeed(seed).forAll(lastBlockReads())
- .checkAssert(this::testReads);
- }
-
- @Test
- public void testReadsLikeChannelProxy()
- {
-
- qt().withFixedSeed(seed).forAll(reads())
- .checkAssert(this::testReads);
- }
-
- private void testReads(InputData propertyInputs)
- {
- try (ChannelProxy channel = new ChannelProxy(propertyInputs.file))
- {
- ThreadLocalReadAheadBuffer trlab = new
ThreadLocalReadAheadBuffer(channel, new
DataStorageSpec.IntKibibytesBound("256KiB").toBytes(), BufferType.OFF_HEAP);
- for (Pair<Long, Integer> read : propertyInputs.positionsAndLengths)
- {
- int readSize = Math.min(read.right,(int) (channel.size() -
read.left));
- ByteBuffer buf1 = ByteBuffer.allocate(readSize);
- channel.read(buf1, read.left);
-
- ByteBuffer buf2 = ByteBuffer.allocate(readSize);
- try
- {
- int copied = 0;
- while (copied < readSize) {
- trlab.fill(read.left + copied);
- int leftToRead = readSize - copied;
- if (trlab.remaining() >= leftToRead)
- copied += trlab.read(buf2, leftToRead);
- else
- copied += trlab.read(buf2, trlab.remaining());
- }
- }
- catch (CorruptSSTableException | CorruptBlockException e)
- {
- throw new RuntimeException(e);
- }
-
- Assert.assertEquals(buf1, buf2);
- }
- }
- }
-
- private Gen<InputData> reads()
- {
- return arbitrary().pick(List.of(files))
- .flatMap((file) ->
- lists().of(longs().between(0,
fileSize(file)).zip(integers().between(1, 100), Pair::create))
- .ofSizeBetween(5, 10)
- .map(positionsAndLengths -> new
InputData(file, positionsAndLengths)));
-
- }
-
- private Gen<InputData> lastBlockReads()
- {
- int blockSize = new
DataStorageSpec.IntKibibytesBound("256KiB").toBytes();
- return arbitrary().pick(List.of(files))
- .flatMap((file) ->
- lists().of(longs().between(max(0,
fileSize(file) - blockSize), fileSize(file)).zip(integers().between(1, 100),
Pair::create))
- .ofSizeBetween(5, 10)
- .map(positionsAndLengths -> new
InputData(file, positionsAndLengths)));
-
- }
-
- // need this becasue generators don't handle the IOException
- private long fileSize(File file)
- {
- try
- {
- return Files.size(file.toPath());
- } catch (IOException e)
- {
- throw new RuntimeException(e);
- }
- }
-
- private static class InputData
- {
-
- private final File file;
- private final List<Pair<Long, Integer>> positionsAndLengths;
-
- public InputData(File file, List<Pair<Long, Integer>>
positionsAndLengths)
- {
- this.file = file;
- this.positionsAndLengths = positionsAndLengths;
- }
- }
-
- private static File writeFile(int seed, int length)
- {
- String fileName = "data+" + length + ".bin";
-
- byte[] dataChunk = new byte[4096 * 8];
- java.util.Random random = new Random(seed);
- int writtenData = 0;
-
- File file = new File(JAVA_IO_TMPDIR.getString(), fileName);
- try (FileOutputStream fos = new FileOutputStream(file.toJavaIOFile()))
- {
- while (writtenData < length)
- {
- random.nextBytes(dataChunk);
- int toWrite = Math.min((length - writtenData),
dataChunk.length);
- fos.write(dataChunk, 0, toWrite);
- writtenData += toWrite;
- }
- fos.flush();
- }
- catch (IOException e)
- {
- throw new RuntimeException(e);
- }
-
- return file;
- }
-
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
