Cleanup some FrozenCacheService Range Handling (#70068)

Two spots here: For one the calculation of `distToStart` in 3 spots was needlessly complex.
Also, we were redundantly acquiring and then right-away releasing file regions in `populateAndRead`
even though we were not going to use them in any way, thus incorrectly recording access to the region.

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/cache/FrozenCacheService.java

@@ -563,13 +563,14 @@ public class FrozenCacheService implements Releasable {
             throw new AlreadyClosedException("File chunk is evicted");
         }
 
-        public StepListener<Integer> populateAndRead(
+        StepListener<Integer> populateAndRead(
             final ByteRange rangeToWrite,
             final ByteRange rangeToRead,
             final RangeAvailableHandler reader,
             final RangeMissingHandler writer,
             final Executor executor
         ) {
+            assert rangeToRead.length() > 0;
             final StepListener<Integer> listener = new StepListener<>();
             Releasable decrementRef = null;
             try {
@@ -582,11 +583,6 @@ public class FrozenCacheService implements Releasable {
                 final SharedBytes.IO fileChannel = sharedBytes.getFileChannel(sharedBytesPos);
                 listener.whenComplete(integer -> fileChannel.decRef(), e -> fileChannel.decRef());
                 final ActionListener<Void> rangeListener = rangeListener(rangeToRead, reader, listener, fileChannel);
-                if (rangeToRead.length() == 0L) {
-                    // nothing to read, skip
-                    rangeListener.onResponse(null);
-                    return listener;
-                }
                 final List<SparseFileTracker.Gap> gaps = tracker.waitForRange(rangeToWrite, rangeToRead, rangeListener);
 
                 for (SparseFileTracker.Gap gap : gaps) {
@@ -721,29 +717,33 @@ public class FrozenCacheService implements Releasable {
             StepListener<Integer> stepListener = null;
             final long writeStart = rangeToWrite.start();
             final long readStart = rangeToRead.start();
-            for (int i = getRegion(rangeToWrite.start()); i <= getEndingRegion(rangeToWrite.end()); i++) {
-                final int region = i;
+            for (int region = getRegion(rangeToWrite.start()); region <= getEndingRegion(rangeToWrite.end()); region++) {
                 final ByteRange subRangeToWrite = mapSubRangeToRegion(rangeToWrite, region);
                 final ByteRange subRangeToRead = mapSubRangeToRegion(rangeToRead, region);
+                if (subRangeToRead.length() == 0L) {
+                    // nothing to read, skip
+                    if (stepListener == null) {
+                        stepListener = new StepListener<>();
+                        stepListener.onResponse(0);
+                    }
+                    continue;
+                }
                 final CacheFileRegion fileRegion = get(cacheKey, length, region);
+                final long regionStart = getRegionStart(region);
+                final long writeOffset = writeStart - regionStart;
+                final long readOffset = readStart - regionStart;
                 final StepListener<Integer> lis = fileRegion.populateAndRead(
                     subRangeToWrite,
                     subRangeToRead,
                     (channel, channelPos, relativePos, length) -> {
-                        final long distanceToStart = region == getRegion(readStart)
-                            ? relativePos - getRegionRelativePosition(readStart)
-                            : getRegionStart(region) + relativePos - readStart;
                         assert regionOwners[fileRegion.sharedBytesPos].get() == fileRegion;
                         assert channelPos >= fileRegion.physicalStartOffset() && channelPos + length <= fileRegion.physicalEndOffset();
-                        return reader.onRangeAvailable(channel, channelPos, distanceToStart, length);
+                        return reader.onRangeAvailable(channel, channelPos, relativePos - readOffset, length);
                     },
                     (channel, channelPos, relativePos, length, progressUpdater) -> {
-                        final long distanceToStart = region == getRegion(writeStart)
-                            ? relativePos - getRegionRelativePosition(writeStart)
-                            : getRegionStart(region) + relativePos - writeStart;
                         assert regionOwners[fileRegion.sharedBytesPos].get() == fileRegion;
                         assert channelPos >= fileRegion.physicalStartOffset() && channelPos + length <= fileRegion.physicalEndOffset();
-                        writer.fillCacheRange(channel, channelPos, distanceToStart, length, progressUpdater);
+                        writer.fillCacheRange(channel, channelPos, relativePos - writeOffset, length, progressUpdater);
                     },
                     executor
                 );
@@ -762,18 +762,18 @@ public class FrozenCacheService implements Releasable {
         public StepListener<Integer> readIfAvailableOrPending(final ByteRange rangeToRead, final RangeAvailableHandler reader) {
             StepListener<Integer> stepListener = null;
             final long start = rangeToRead.start();
-            for (int i = getRegion(rangeToRead.start()); i <= getEndingRegion(rangeToRead.end()); i++) {
-                final int region = i;
+            for (int region = getRegion(start); region <= getEndingRegion(rangeToRead.end()); region++) {
                 final ByteRange subRangeToRead = mapSubRangeToRegion(rangeToRead, region);
                 final CacheFileRegion fileRegion = get(cacheKey, length, region);
+                final long readOffset = start - getRegionStart(region);
                 final StepListener<Integer> lis = fileRegion.readIfAvailableOrPending(
                     subRangeToRead,
-                    (channel, channelPos, relativePos, length) -> {
-                        final long distanceToStart = region == getRegion(start)
-                            ? relativePos - getRegionRelativePosition(start)
-                            : getRegionStart(region) + relativePos - start;
-                        return reader.onRangeAvailable(channel, channelPos, distanceToStart, length);
-                    }
+                    (channel, channelPos, relativePos, length) -> reader.onRangeAvailable(
+                        channel,
+                        channelPos,
+                        relativePos - readOffset,
+                        length
+                    )
                 );
                 if (lis == null) {
                     return null;