Quicker shared cache file preallocation (#79447)

Reworks preallocation of the shared_cache file, which was very slow on Windows.

Instead of manually filling the file with 0's, this new approach uses the RandomAccessFile.setLength() method, which
can quickly allocate a file of the given size (tested that this took only seconds to preallocate TB-size file on Windows).

Author: ywelsch

x-pack/plugin/searchable-snapshots/preallocate/src/main/java/org/elasticsearch/xpack/searchablesnapshots/preallocate/LinuxPreallocator.java

@@ -16,7 +16,7 @@
 final class LinuxPreallocator implements Preallocator {
 
     @Override
-    public boolean available() {
+    public boolean useNative() {
         return Natives.NATIVES_AVAILABLE;
     }
 

x-pack/plugin/searchable-snapshots/preallocate/src/main/java/org/elasticsearch/xpack/searchablesnapshots/preallocate/MacOsPreallocator.java

@@ -20,7 +20,7 @@
 final class MacOsPreallocator implements Preallocator {
 
     @Override
-    public boolean available() {
+    public boolean useNative() {
         return Natives.NATIVES_AVAILABLE;
     }
 

x-pack/plugin/searchable-snapshots/preallocate/src/main/java/org/elasticsearch/xpack/searchablesnapshots/preallocate/UnsupportedPreallocator.java renamed to x-pack/plugin/searchable-snapshots/preallocate/src/main/java/org/elasticsearch/xpack/searchablesnapshots/preallocate/NoNativePreallocator.java

@@ -7,10 +7,10 @@
 
 package org.elasticsearch.xpack.searchablesnapshots.preallocate;
 
-final class UnsupportedPreallocator implements Preallocator {
+final class NoNativePreallocator implements Preallocator {
 
     @Override
-    public boolean available() {
+    public boolean useNative() {
         return false;
     }
 

x-pack/plugin/searchable-snapshots/preallocate/src/main/java/org/elasticsearch/xpack/searchablesnapshots/preallocate/Preallocate.java

@@ -11,10 +11,12 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.lucene.util.Constants;
+import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.core.SuppressForbidden;
 
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.io.RandomAccessFile;
 import java.lang.reflect.Field;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -31,35 +33,53 @@ public static void preallocate(final Path cacheFile, final long fileSize) throws
         } else if (Constants.MAC_OS_X) {
             preallocate(cacheFile, fileSize, new MacOsPreallocator());
         } else {
-            preallocate(cacheFile, fileSize, new UnsupportedPreallocator());
+            preallocate(cacheFile, fileSize, new NoNativePreallocator());
         }
     }
 
     @SuppressForbidden(reason = "need access to fd on FileOutputStream")
     private static void preallocate(final Path cacheFile, final long fileSize, final Preallocator prealloactor) throws IOException {
-        if (prealloactor.available() == false) {
-            logger.warn("failed to pre-allocate cache file [{}] as native methods are not available", cacheFile);
-        }
         boolean success = false;
-        try (FileOutputStream fileChannel = new FileOutputStream(cacheFile.toFile())) {
-            long currentSize = fileChannel.getChannel().size();
-            if (currentSize < fileSize) {
-                final Field field = AccessController.doPrivileged(new FileDescriptorFieldAction(fileChannel));
-                final int errno = prealloactor.preallocate((int) field.get(fileChannel.getFD()), currentSize, fileSize - currentSize);
-                if (errno == 0) {
+        try {
+            if (prealloactor.useNative()) {
+                try (FileOutputStream fileChannel = new FileOutputStream(cacheFile.toFile())) {
+                    long currentSize = fileChannel.getChannel().size();
+                    if (currentSize < fileSize) {
+                        logger.info("pre-allocating cache file [{}] ({}) using native methods", cacheFile, new ByteSizeValue(fileSize));
+                        final Field field = AccessController.doPrivileged(new FileDescriptorFieldAction(fileChannel));
+                        final int errno = prealloactor.preallocate(
+                            (int) field.get(fileChannel.getFD()),
+                            currentSize,
+                            fileSize - currentSize
+                        );
+                        if (errno == 0) {
+                            success = true;
+                            logger.debug("pre-allocated cache file [{}] using native methods", cacheFile);
+                        } else {
+                            logger.warn(
+                                "failed to pre-allocate cache file [{}] using native methods, errno: [{}], error: [{}]",
+                                cacheFile,
+                                errno,
+                                prealloactor.error(errno)
+                            );
+                        }
+                    }
+                } catch (final Exception e) {
+                    logger.warn(new ParameterizedMessage("failed to pre-allocate cache file [{}] using native methods", cacheFile), e);
+                }
+            }
+            // even if allocation was successful above, verify again here
+            try (RandomAccessFile raf = new RandomAccessFile(cacheFile.toFile(), "rw")) {
+                if (raf.length() != fileSize) {
+                    logger.info("pre-allocating cache file [{}] ({}) using setLength method", cacheFile, new ByteSizeValue(fileSize));
+                    raf.setLength(fileSize);
                     success = true;
-                    logger.debug("pre-allocated cache file [{}] using native methods", cacheFile);
-                } else {
-                    logger.warn(
-                        "failed to pre-allocate cache file [{}] using native methods, errno: [{}], error: [{}]",
-                        cacheFile,
-                        errno,
-                        prealloactor.error(errno)
-                    );
+                    logger.debug("pre-allocated cache file [{}] using setLength method", cacheFile);
                 }
+            } catch (final Exception e) {
+                logger.warn(new ParameterizedMessage("failed to pre-allocate cache file [{}] using setLength method", cacheFile), e);
+                throw e;
             }
-        } catch (final Exception e) {
-            logger.warn(new ParameterizedMessage("failed to pre-allocate cache file [{}] using native methods", cacheFile), e);
         } finally {
             if (success == false) {
                 // if anything goes wrong, delete the potentially created file to not waste disk space

x-pack/plugin/searchable-snapshots/preallocate/src/main/java/org/elasticsearch/xpack/searchablesnapshots/preallocate/Preallocator.java

@@ -17,7 +17,7 @@ interface Preallocator {
      *
      * @return true if native methods are available, otherwise false
      */
-    boolean available();
+    boolean useNative();
 
     /**
      * Pre-allocate a file of given current size to the specified size using the given file descriptor.

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

@@ -9,7 +9,6 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.elasticsearch.common.io.Channels;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.core.AbstractRefCounted;
@@ -61,27 +60,8 @@ public class SharedBytes extends AbstractRefCounted {
         if (fileSize > 0) {
             cacheFile = findCacheSnapshotCacheFilePath(environment, fileSize);
             Preallocate.preallocate(cacheFile, fileSize);
-            // TODO: maybe make this faster by allocating a larger direct buffer if this is too slow for very large files
-            // We fill either the full file or the bytes between its current size and the desired size once with zeros to fully allocate
-            // the file up front
-            final ByteBuffer fillBytes = ByteBuffer.allocate(Channels.WRITE_CHUNK_SIZE);
             this.fileChannel = FileChannel.open(cacheFile, OPEN_OPTIONS);
-            long written = fileChannel.size();
-            if (fileSize < written) {
-                logger.info("creating shared snapshot cache file [size={}, path={}]", fileSize, cacheFile);
-            } else if (fileSize == written) {
-                logger.debug("reusing existing shared snapshot cache file [size={}, path={}]", fileSize, cacheFile);
-            }
-            fileChannel.position(written);
-            while (written < fileSize) {
-                final int toWrite = Math.toIntExact(Math.min(fileSize - written, Channels.WRITE_CHUNK_SIZE));
-                fillBytes.position(0).limit(toWrite);
-                Channels.writeToChannel(fillBytes, fileChannel);
-                written += toWrite;
-            }
-            if (written > fileChannel.size()) {
-                fileChannel.truncate(fileSize);
-            }
+            assert this.fileChannel.size() == fileSize : "expected file size " + fileSize + " but was " + fileChannel.size();
         } else {
             this.fileChannel = null;
             for (Path path : environment.nodeDataPaths()) {

x-pack/plugin/searchable-snapshots/src/test/java/org/elasticsearch/xpack/searchablesnapshots/cache/shared/FrozenCacheServiceTests.java

@@ -10,6 +10,7 @@
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsException;
+import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.RatioValue;
 import org.elasticsearch.common.unit.RelativeByteSizeValue;
@@ -346,4 +347,33 @@ private static CacheKey generateCacheKey() {
             randomAlphaOfLength(10)
         );
     }
+
+    public void testCacheSizeChanges() throws IOException {
+        ByteSizeValue val1 = new ByteSizeValue(randomIntBetween(1, 5), ByteSizeUnit.MB);
+        Settings settings = Settings.builder()
+            .put(NODE_NAME_SETTING.getKey(), "node")
+            .put(FrozenCacheService.SNAPSHOT_CACHE_SIZE_SETTING.getKey(), val1.getStringRep())
+            .put(FrozenCacheService.SNAPSHOT_CACHE_REGION_SIZE_SETTING.getKey(), new ByteSizeValue(size(100)).getStringRep())
+            .put("path.home", createTempDir())
+            .build();
+        final DeterministicTaskQueue taskQueue = new DeterministicTaskQueue();
+        try (
+            NodeEnvironment environment = new NodeEnvironment(settings, TestEnvironment.newEnvironment(settings));
+            FrozenCacheService cacheService = new FrozenCacheService(environment, settings, taskQueue.getThreadPool())
+        ) {
+            assertEquals(val1.getBytes(), cacheService.getStats().getSize());
+        }
+
+        ByteSizeValue val2 = new ByteSizeValue(randomIntBetween(1, 5), ByteSizeUnit.MB);
+        settings = Settings.builder()
+            .put(settings)
+            .put(FrozenCacheService.SNAPSHOT_CACHE_SIZE_SETTING.getKey(), val2.getStringRep())
+            .build();
+        try (
+            NodeEnvironment environment = new NodeEnvironment(settings, TestEnvironment.newEnvironment(settings));
+            FrozenCacheService cacheService = new FrozenCacheService(environment, settings, taskQueue.getThreadPool())
+        ) {
+            assertEquals(val2.getBytes(), cacheService.getStats().getSize());
+        }
+    }
 }