You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by br...@apache.org on 2025/05/06 11:42:37 UTC
(cassandra) branch trunk updated (0669db7a04 -> 12c5495d2f)
This is an automated email from the ASF dual-hosted git repository.
brandonwilliams pushed a change to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
from 0669db7a04 Write user docs for CEP-24 - Password validation / generation
new 6a81306876 Reading mmapped trie-index exceeding 2GiB results in exception
new 765bfb9daf Merge branch 'cassandra-5.0' into trunk
new 12c5495d2f Reading mmapped trie-index exceeding 2GiB results in exception
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/cassandra/io/util/FileHandle.java | 6 +-
.../apache/cassandra/io/util/MmappedRegions.java | 53 +++++++-------
.../cassandra/io/util/MmappedRegionsCache.java | 10 +--
.../cassandra/io/util/MmappedRegionsTest.java | 81 ++++++++++++++++------
4 files changed, 93 insertions(+), 57 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 01/02: Merge branch 'cassandra-5.0' into trunk
Posted by br...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
brandonwilliams pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 765bfb9daf234aa15d6a549b53d339831edbb8ec
Merge: 0669db7a04 6a81306876
Author: Brandon Williams <br...@apache.org>
AuthorDate: Tue May 6 06:39:12 2025 -0500
Merge branch 'cassandra-5.0' into trunk
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
(cassandra) 02/02: Reading mmapped trie-index exceeding 2GiB results in exception
Posted by br...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
brandonwilliams pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 12c5495d2f70e0775c6d8d1ffbd00123cc8923ed
Author: Szymon Miężał <sz...@datastax.com>
AuthorDate: Thu Apr 24 15:06:55 2025 +0200
Reading mmapped trie-index exceeding 2GiB results in exception
Memory-mapping is done in buffers of size less than 2GiB.
When these buffers aren't aligned to 4KiB and the trie-index file
spans many buffers then reading it results in going out of buffer
bounds.
This patch fixes it by making sure that the buffers are correctly
aligned.
patch by Szymon Miezal; reviewed by blambov and brandonwilliams for CASSANDRA-20351
---
.../org/apache/cassandra/io/util/FileHandle.java | 6 +-
.../apache/cassandra/io/util/MmappedRegions.java | 53 +++++++-------
.../cassandra/io/util/MmappedRegionsCache.java | 10 +--
.../cassandra/io/util/MmappedRegionsTest.java | 81 ++++++++++++++++------
4 files changed, 93 insertions(+), 57 deletions(-)
diff --git a/src/java/org/apache/cassandra/io/util/FileHandle.java b/src/java/org/apache/cassandra/io/util/FileHandle.java
index 67bfd239d6..7e4b214128 100644
--- a/src/java/org/apache/cassandra/io/util/FileHandle.java
+++ b/src/java/org/apache/cassandra/io/util/FileHandle.java
@@ -406,14 +406,14 @@ public class FileHandle extends SharedCloseableImpl
{
if (compressionMetadata != null)
{
- regions = mmappedRegionsCache != null ? mmappedRegionsCache.getOrCreate(channel, compressionMetadata)
+ regions = mmappedRegionsCache != null ? mmappedRegionsCache.getOrCreate(channel, compressionMetadata, bufferSize)
: MmappedRegions.map(channel, compressionMetadata);
rebuffererFactory = maybeCached(new CompressedChunkReader.Mmap(channel, compressionMetadata, regions, crcCheckChanceSupplier));
}
else
{
- regions = mmappedRegionsCache != null ? mmappedRegionsCache.getOrCreate(channel, length)
- : MmappedRegions.map(channel, length);
+ regions = mmappedRegionsCache != null ? mmappedRegionsCache.getOrCreate(channel, length, bufferSize)
+ : MmappedRegions.map(channel, length, bufferSize);
rebuffererFactory = new MmapRebufferer(channel, length, regions);
}
}
diff --git a/src/java/org/apache/cassandra/io/util/MmappedRegions.java b/src/java/org/apache/cassandra/io/util/MmappedRegions.java
index 0ab07b8d0f..578217e279 100644
--- a/src/java/org/apache/cassandra/io/util/MmappedRegions.java
+++ b/src/java/org/apache/cassandra/io/util/MmappedRegions.java
@@ -65,27 +65,22 @@ public class MmappedRegions extends SharedCloseableImpl
*/
private volatile State copy;
- private MmappedRegions(ChannelProxy channel, CompressionMetadata metadata, long length)
- {
- this(new State(channel), metadata, length);
- }
-
- private MmappedRegions(State state, CompressionMetadata metadata, long length)
+ private MmappedRegions(State state, long length, int chunkSize)
{
super(new Tidier(state));
-
this.state = state;
-
- if (metadata != null)
- {
- assert length == 0 : "expected no length with metadata";
- updateState(metadata);
- }
- else if (length > 0)
+ if (length > 0)
{
- updateState(length);
+ updateState(length, chunkSize);
}
+ this.copy = new State(state);
+ }
+ private MmappedRegions(State state, CompressionMetadata metadata)
+ {
+ super(new Tidier(state));
+ this.state = state;
+ updateState(metadata);
this.copy = new State(state);
}
@@ -97,7 +92,7 @@ public class MmappedRegions extends SharedCloseableImpl
public static MmappedRegions empty(ChannelProxy channel)
{
- return new MmappedRegions(channel, null, 0);
+ return new MmappedRegions(new State(channel), 0, 0);
}
/**
@@ -109,16 +104,16 @@ public class MmappedRegions extends SharedCloseableImpl
{
if (metadata == null)
throw new IllegalArgumentException("metadata cannot be null");
-
- return new MmappedRegions(channel, metadata, 0);
+ State state = new State(channel);
+ return new MmappedRegions(state, metadata);
}
- public static MmappedRegions map(ChannelProxy channel, long length)
+ public static MmappedRegions map(ChannelProxy channel, long length, int chunkSize)
{
if (length <= 0)
throw new IllegalArgumentException("Length must be positive");
-
- return new MmappedRegions(channel, null, length);
+ State state = new State(channel);
+ return new MmappedRegions(state, length, chunkSize);
}
/**
@@ -140,8 +135,10 @@ public class MmappedRegions extends SharedCloseableImpl
*
* @return {@code true} if new regions have been created
*/
- public boolean extend(long length)
+ public boolean extend(long length, int chunkSize)
{
+ // We cannot enforce length to be a multiple of chunkSize (at the very least the last extend on a file
+ // will not satisfy this), so we hope the caller knows what they are doing.
if (length < 0)
throw new IllegalArgumentException("Length must not be negative");
@@ -151,7 +148,7 @@ public class MmappedRegions extends SharedCloseableImpl
return false;
int initialRegions = state.last;
- updateState(length);
+ updateState(length, chunkSize);
copy = new State(state);
return state.last > initialRegions;
}
@@ -162,7 +159,7 @@ public class MmappedRegions extends SharedCloseableImpl
*
* @return {@code true} if new regions have been created
*/
- public boolean extend(CompressionMetadata compressionMetadata)
+ public boolean extend(CompressionMetadata compressionMetadata, int chunkSize)
{
assert !isCopy() : "Copies cannot be extended";
@@ -171,7 +168,7 @@ public class MmappedRegions extends SharedCloseableImpl
int initialRegions = state.last;
if (compressionMetadata.compressedFileLength - state.length <= MAX_SEGMENT_SIZE)
- updateState(compressionMetadata.compressedFileLength);
+ updateState(compressionMetadata.compressedFileLength, chunkSize);
else
updateState(compressionMetadata);
@@ -183,13 +180,15 @@ public class MmappedRegions extends SharedCloseableImpl
* Updates state by adding the remaining segments. It starts with the current state last segment end position and
* subsequently add new segments until all data up to the provided length are mapped.
*/
- private void updateState(long length)
+ private void updateState(long length, int chunkSize)
{
+ // make sure the regions span whole chunks
+ long maxSize = (long) (MAX_SEGMENT_SIZE / chunkSize) * chunkSize;
state.length = length;
long pos = state.getPosition();
while (pos < length)
{
- long size = Math.min(MAX_SEGMENT_SIZE, length - pos);
+ long size = Math.min(maxSize, length - pos);
state.add(pos, size);
pos += size;
}
diff --git a/src/java/org/apache/cassandra/io/util/MmappedRegionsCache.java b/src/java/org/apache/cassandra/io/util/MmappedRegionsCache.java
index dff9561f4f..e3ebc34609 100644
--- a/src/java/org/apache/cassandra/io/util/MmappedRegionsCache.java
+++ b/src/java/org/apache/cassandra/io/util/MmappedRegionsCache.java
@@ -45,12 +45,12 @@ public class MmappedRegionsCache implements AutoCloseable
* @param length length of the file
* @return a shared copy of the cached mmapped regions
*/
- public MmappedRegions getOrCreate(ChannelProxy channel, long length)
+ public MmappedRegions getOrCreate(ChannelProxy channel, long length, int bufferSize)
{
Preconditions.checkState(!closed);
- MmappedRegions regions = cache.computeIfAbsent(channel.file(), ignored -> MmappedRegions.map(channel, length));
+ MmappedRegions regions = cache.computeIfAbsent(channel.file(), ignored -> MmappedRegions.map(channel, length, bufferSize));
Preconditions.checkArgument(regions.isValid(channel));
- regions.extend(length);
+ regions.extend(length, bufferSize);
return regions.sharedCopy();
}
@@ -62,12 +62,12 @@ public class MmappedRegionsCache implements AutoCloseable
* @param metadata compression metadata of the file
* @return a shared copy of the cached mmapped regions
*/
- public MmappedRegions getOrCreate(ChannelProxy channel, CompressionMetadata metadata)
+ public MmappedRegions getOrCreate(ChannelProxy channel, CompressionMetadata metadata, int bufferSize)
{
Preconditions.checkState(!closed);
MmappedRegions regions = cache.computeIfAbsent(channel.file(), ignored -> MmappedRegions.map(channel, metadata));
Preconditions.checkArgument(regions.isValid(channel));
- regions.extend(metadata);
+ regions.extend(metadata, bufferSize);
return regions.sharedCopy();
}
diff --git a/test/unit/org/apache/cassandra/io/util/MmappedRegionsTest.java b/test/unit/org/apache/cassandra/io/util/MmappedRegionsTest.java
index e6b5dd0c09..af4c6f042d 100644
--- a/test/unit/org/apache/cassandra/io/util/MmappedRegionsTest.java
+++ b/test/unit/org/apache/cassandra/io/util/MmappedRegionsTest.java
@@ -114,10 +114,11 @@ public class MmappedRegionsTest
public void testTwoSegments() throws Exception
{
ByteBuffer buffer = allocateBuffer(2048);
+ int bufSize = 1024;
try (ChannelProxy channel = new ChannelProxy(writeFile("testTwoSegments", buffer));
MmappedRegions regions = MmappedRegions.empty(channel))
{
- regions.extend(1024);
+ regions.extend(1024, bufSize);
for (int i = 0; i < 1024; i++)
{
MmappedRegions.Region region = regions.floor(i);
@@ -126,7 +127,7 @@ public class MmappedRegionsTest
assertEquals(1024, region.end());
}
- regions.extend(2048);
+ regions.extend(2048, bufSize);
for (int i = 0; i < 2048; i++)
{
MmappedRegions.Region region = regions.floor(i);
@@ -149,14 +150,15 @@ public class MmappedRegionsTest
public void testSmallSegmentSize() throws Exception
{
MmappedRegions.MAX_SEGMENT_SIZE = 1024;
+ int bufSize = 1024;
ByteBuffer buffer = allocateBuffer(4096);
try (ChannelProxy channel = new ChannelProxy(writeFile("testSmallSegmentSize", buffer));
MmappedRegions regions = MmappedRegions.empty(channel))
{
- regions.extend(1024);
- regions.extend(2048);
- regions.extend(4096);
+ regions.extend(1024, bufSize);
+ regions.extend(2048, bufSize);
+ regions.extend(4096, bufSize);
final int SIZE = MmappedRegions.MAX_SEGMENT_SIZE;
for (int i = 0; i < buffer.capacity(); i++)
@@ -169,17 +171,45 @@ public class MmappedRegionsTest
}
}
+ @Test
+ public void testSizeIsChunkMultiple() throws Exception
+ {
+ final int oldMaxSegmentSize = MmappedRegions.MAX_SEGMENT_SIZE;
+ final int bufSize = 1024;
+ MmappedRegions.MAX_SEGMENT_SIZE = 2047;
+ ByteBuffer buffer = allocateBuffer(4096);
+ try(ChannelProxy channel = new ChannelProxy(writeFile("testSmallSegmentSize", buffer));
+ MmappedRegions regions = MmappedRegions.empty(channel))
+ {
+ regions.extend(1024, bufSize);
+ regions.extend(2048, bufSize);
+ regions.extend(4096, bufSize);
+ for (int i = 0; i < buffer.capacity(); i++)
+ {
+ MmappedRegions.Region region = regions.floor(i);
+ assertNotNull(region);
+ assertEquals(bufSize * (i / bufSize), region.offset());
+ assertEquals(bufSize + (bufSize * (i / bufSize)), region.end());
+ }
+ }
+ finally
+ {
+ MmappedRegions.MAX_SEGMENT_SIZE = oldMaxSegmentSize;
+ }
+ }
+
@Test
public void testAllocRegions() throws Exception
{
MmappedRegions.MAX_SEGMENT_SIZE = 1024;
ByteBuffer buffer = allocateBuffer(MmappedRegions.MAX_SEGMENT_SIZE * MmappedRegions.REGION_ALLOC_SIZE * 3);
+ int bufSize = 1024;
try (ChannelProxy channel = new ChannelProxy(writeFile("testAllocRegions", buffer));
MmappedRegions regions = MmappedRegions.empty(channel))
{
- regions.extend(buffer.capacity());
+ regions.extend(buffer.capacity(), bufSize);
final int SIZE = MmappedRegions.MAX_SEGMENT_SIZE;
for (int i = 0; i < buffer.capacity(); i++)
@@ -196,17 +226,18 @@ public class MmappedRegionsTest
public void testCopy() throws Exception
{
ByteBuffer buffer = allocateBuffer(128 * 1024);
+ int bufSize = 4096;
MmappedRegions snapshot;
ChannelProxy channelCopy;
try (ChannelProxy channel = new ChannelProxy(writeFile("testSnapshot", buffer));
- MmappedRegions regions = MmappedRegions.map(channel, buffer.capacity() / 4))
+ MmappedRegions regions = MmappedRegions.map(channel, buffer.capacity() / 4, bufSize))
{
// create 3 more segments, one per quater capacity
- regions.extend(buffer.capacity() / 2);
- regions.extend(3 * buffer.capacity() / 4);
- regions.extend(buffer.capacity());
+ regions.extend(buffer.capacity() / 2, bufSize);
+ regions.extend(3 * buffer.capacity() / 4, bufSize);
+ regions.extend(buffer.capacity(), bufSize);
// make a snapshot
snapshot = regions.sharedCopy();
@@ -238,6 +269,7 @@ public class MmappedRegionsTest
public void testCopyCannotExtend() throws Exception
{
ByteBuffer buffer = allocateBuffer(128 * 1024);
+ int bufSize = 1024;
MmappedRegions snapshot;
ChannelProxy channelCopy;
@@ -245,7 +277,7 @@ public class MmappedRegionsTest
try (ChannelProxy channel = new ChannelProxy(writeFile("testSnapshotCannotExtend", buffer));
MmappedRegions regions = MmappedRegions.empty(channel))
{
- regions.extend(buffer.capacity() / 2);
+ regions.extend(buffer.capacity() / 2, bufSize);
// make a snapshot
snapshot = regions.sharedCopy();
@@ -256,7 +288,7 @@ public class MmappedRegionsTest
try
{
- snapshot.extend(buffer.capacity());
+ snapshot.extend(buffer.capacity(), bufSize);
}
finally
{
@@ -269,12 +301,13 @@ public class MmappedRegionsTest
public void testExtendOutOfOrder() throws Exception
{
ByteBuffer buffer = allocateBuffer(4096);
+ int bufSize = 1024;
try (ChannelProxy channel = new ChannelProxy(writeFile("testExtendOutOfOrder", buffer));
MmappedRegions regions = MmappedRegions.empty(channel))
{
- regions.extend(4096);
- regions.extend(1024);
- regions.extend(2048);
+ regions.extend(4096, bufSize);
+ regions.extend(1024, bufSize);
+ regions.extend(2048, bufSize);
for (int i = 0; i < buffer.capacity(); i++)
{
@@ -290,10 +323,11 @@ public class MmappedRegionsTest
public void testNegativeExtend() throws Exception
{
ByteBuffer buffer = allocateBuffer(1024);
+ int bufSize = 1024;
try (ChannelProxy channel = new ChannelProxy(writeFile("testNegativeExtend", buffer));
MmappedRegions regions = MmappedRegions.empty(channel))
{
- regions.extend(-1);
+ regions.extend(-1, bufSize);
}
}
@@ -341,8 +375,9 @@ public class MmappedRegionsTest
public void testIllegalArgForMap1() throws Exception
{
ByteBuffer buffer = allocateBuffer(1024);
+ int bufSize = 1024;
try (ChannelProxy channel = new ChannelProxy(writeFile("testIllegalArgForMap1", buffer));
- MmappedRegions regions = MmappedRegions.map(channel, 0))
+ MmappedRegions regions = MmappedRegions.map(channel, 0, bufSize))
{
assertTrue(regions.isEmpty());
}
@@ -352,8 +387,9 @@ public class MmappedRegionsTest
public void testIllegalArgForMap2() throws Exception
{
ByteBuffer buffer = allocateBuffer(1024);
+ int bufSize = 1024;
try (ChannelProxy channel = new ChannelProxy(writeFile("testIllegalArgForMap2", buffer));
- MmappedRegions regions = MmappedRegions.map(channel, -1L))
+ MmappedRegions regions = MmappedRegions.map(channel, -1L, bufSize))
{
assertTrue(regions.isEmpty());
}
@@ -382,6 +418,7 @@ public class MmappedRegionsTest
{
MmappedRegions.MAX_SEGMENT_SIZE = maxSegmentSize << 10;
int size = Arrays.stream(writeSizes).sum() << 10;
+ int bufSize = 4096;
ByteBuffer buffer = allocateBuffer(size);
File f = FileUtils.createTempFile("testMapForCompressionMetadata", "1");
@@ -423,10 +460,10 @@ public class MmappedRegionsTest
writer.sync();
// verify that calling extend for the same (first iteration) or some previous metadata (further iterations) has no effect
- assertFalse(regions.extend(metadata));
+ assertFalse(regions.extend(metadata, bufSize));
logger.info("Checking extend on compressed chunk for range={} {}..{} / {}", idx, pos, pos + (writeSizes[idx] << 10), size);
- checkExtendOnCompressedChunks(f, writer, regions);
+ checkExtendOnCompressedChunks(f, writer, regions, bufSize);
pos += writeSizes[idx] << 10;
idx++;
}
@@ -434,12 +471,12 @@ public class MmappedRegionsTest
}
}
- private void checkExtendOnCompressedChunks(File f, CompressedSequentialWriter writer, MmappedRegions regions)
+ private void checkExtendOnCompressedChunks(File f, CompressedSequentialWriter writer, MmappedRegions regions, int bufSize)
{
int dataOffset;
try (CompressionMetadata metadata = writer.open(writer.getLastFlushOffset()))
{
- regions.extend(metadata);
+ regions.extend(metadata, bufSize);
assertFalse(regions.isEmpty());
dataOffset = 0;
while (dataOffset < metadata.dataLength)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org