Utility methods for rounding to aligned size (#107908)

Add more utility methods for rounding to aligned size and use
these in computeRange and toPageAlignedSize.

x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/BlobCacheUtils.java

@@ -33,14 +33,32 @@ public class BlobCacheUtils {
     }
 
     /**
-     * Rounds the length up so that it is aligned on the next page size (defined by SharedBytes.PAGE_SIZE). For example
+     * Round down the size to the nearest aligned size <= size.
      */
-    public static long toPageAlignedSize(long length) {
-        int remainder = (int) (length % SharedBytes.PAGE_SIZE);
-        if (remainder > 0L) {
-            return length + (SharedBytes.PAGE_SIZE - remainder);
+    public static long roundDownToAlignedSize(long size, long alignment) {
+        assert size >= 0;
+        assert alignment > 0;
+        return size / alignment * alignment;
+    }
+
+    /**
+     * Round up the size to the nearest aligned size >= size
+     */
+    public static long roundUpToAlignedSize(long size, long alignment) {
+        assert size >= 0;
+        if (size == 0) {
+            return 0;
         }
-        return length;
+        assert alignment > 0;
+        return (((size - 1) / alignment) + 1) * alignment;
+    }
+
+    /**
+     * Rounds the length up so that it is aligned on the next page size (defined by SharedBytes.PAGE_SIZE).
+     */
+    public static long toPageAlignedSize(long length) {
+        int alignment = SharedBytes.PAGE_SIZE;
+        return roundUpToAlignedSize(length, alignment);
     }
 
     public static void throwEOF(long channelPos, long len) throws EOFException {
@@ -58,13 +76,13 @@ public class BlobCacheUtils {
 
     public static ByteRange computeRange(long rangeSize, long position, long size, long blobLength) {
         return ByteRange.of(
-            (position / rangeSize) * rangeSize,
-            Math.min((((position + size - 1) / rangeSize) + 1) * rangeSize, blobLength)
+            roundDownToAlignedSize(position, rangeSize),
+            Math.min(roundUpToAlignedSize(position + size, rangeSize), blobLength)
         );
     }
 
     public static ByteRange computeRange(long rangeSize, long position, long size) {
-        return ByteRange.of((position / rangeSize) * rangeSize, (((position + size - 1) / rangeSize) + 1) * rangeSize);
+        return ByteRange.of(roundDownToAlignedSize(position, rangeSize), roundUpToAlignedSize(position + size, rangeSize));
     }
 
     public static void ensureSlice(String sliceName, long sliceOffset, long sliceLength, IndexInput input) {

x-pack/plugin/blob-cache/src/test/java/org/elasticsearch/blobcache/BlobCacheUtilsTests.java

@@ -6,14 +6,16 @@
  */
 package org.elasticsearch.blobcache;
 
+import org.elasticsearch.blobcache.common.ByteRange;
 import org.elasticsearch.blobcache.shared.SharedBytes;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.test.ESTestCase;
-import org.hamcrest.Matchers;
 
 import java.io.EOFException;
 import java.nio.ByteBuffer;
 
+import static org.hamcrest.Matchers.equalTo;
+
 public class BlobCacheUtilsTests extends ESTestCase {
 
     public void testReadSafeThrows() {
@@ -25,6 +27,58 @@ public class BlobCacheUtilsTests extends ESTestCase {
     public void testToPageAlignedSize() {
         long value = randomLongBetween(0, Long.MAX_VALUE - SharedBytes.PAGE_SIZE);
         long expected = ((value - 1) / SharedBytes.PAGE_SIZE + 1) * SharedBytes.PAGE_SIZE;
-        assertThat(BlobCacheUtils.toPageAlignedSize(value), Matchers.equalTo(expected));
+        assertThat(BlobCacheUtils.toPageAlignedSize(value), equalTo(expected));
+        assertThat(BlobCacheUtils.toPageAlignedSize(value), equalTo(roundUpUsingRemainder(value, SharedBytes.PAGE_SIZE)));
+    }
+
+    public void testRoundUpToAlignment() {
+        assertThat(BlobCacheUtils.roundUpToAlignedSize(8, 4), equalTo(8L));
+        assertThat(BlobCacheUtils.roundUpToAlignedSize(9, 4), equalTo(12L));
+        assertThat(BlobCacheUtils.roundUpToAlignedSize(between(1, 4), 4), equalTo(4L));
+        long alignment = randomLongBetween(1, Long.MAX_VALUE / 2);
+        assertThat(BlobCacheUtils.roundUpToAlignedSize(0, alignment), equalTo(0L));
+        long value = randomLongBetween(0, Long.MAX_VALUE - alignment);
+        assertThat(BlobCacheUtils.roundUpToAlignedSize(value, alignment), equalTo(roundUpUsingRemainder(value, alignment)));
+    }
+
+    public void testRoundDownToAlignment() {
+        assertThat(BlobCacheUtils.roundDownToAlignedSize(8, 4), equalTo(8L));
+        assertThat(BlobCacheUtils.roundDownToAlignedSize(9, 4), equalTo(8L));
+        assertThat(BlobCacheUtils.roundDownToAlignedSize(between(0, 3), 4), equalTo(0L));
+        long alignment = randomLongBetween(1, Long.MAX_VALUE / 2);
+        long value = randomLongBetween(0, Long.MAX_VALUE);
+        assertThat(BlobCacheUtils.roundDownToAlignedSize(value, alignment), equalTo(roundDownUsingRemainder(value, alignment)));
+    }
+
+    public void testComputeRange() {
+        assertThat(BlobCacheUtils.computeRange(8, 8, 8), equalTo(ByteRange.of(8, 16)));
+        assertThat(BlobCacheUtils.computeRange(8, 9, 8), equalTo(ByteRange.of(8, 24)));
+        assertThat(BlobCacheUtils.computeRange(8, 8, 9), equalTo(ByteRange.of(8, 24)));
+
+        long large = randomLongBetween(24, 64);
+        assertThat(BlobCacheUtils.computeRange(8, 8, 8, large), equalTo(ByteRange.of(8, 16)));
+        assertThat(BlobCacheUtils.computeRange(8, 9, 8, large), equalTo(ByteRange.of(8, 24)));
+        assertThat(BlobCacheUtils.computeRange(8, 8, 9, large), equalTo(ByteRange.of(8, 24)));
+
+        long small = randomLongBetween(8, 16);
+        assertThat(BlobCacheUtils.computeRange(8, 8, 8, small), equalTo(ByteRange.of(8, small)));
+        assertThat(BlobCacheUtils.computeRange(8, 9, 8, small), equalTo(ByteRange.of(8, small)));
+        assertThat(BlobCacheUtils.computeRange(8, 8, 9, small), equalTo(ByteRange.of(8, small)));
+    }
+
+    private static long roundUpUsingRemainder(long value, long alignment) {
+        long remainder = value % alignment;
+        if (remainder > 0L) {
+            return value + (alignment - remainder);
+        }
+        return value;
+    }
+
+    private static long roundDownUsingRemainder(long value, long alignment) {
+        long remainder = value % alignment;
+        if (remainder > 0L) {
+            return value - remainder;
+        }
+        return value;
     }
 }