natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
This commit is contained in:
Szymon Miężał 2025-04-24 15:06:55 +02:00 committed by Brandon Williams
parent 765bfb9daf
commit 12c5495d2f
4 changed files with 93 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/java/org/apache/cassandra/io/util/FileHandle.java
View File

@ -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);
}
}

53
src/java/org/apache/cassandra/io/util/MmappedRegions.java
View File

@ -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)
if (length > 0)
{
assert length == 0 : "expected no length with metadata";
updateState(metadata);
}
else 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;
}

10
src/java/org/apache/cassandra/io/util/MmappedRegionsCache.java
View File

@ -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();
}

81
test/unit/org/apache/cassandra/io/util/MmappedRegionsTest.java
View File

@ -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)