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