Aggregate files from the same segment into a single Arena

lucene/CHANGES.txt

@@ -300,6 +300,13 @@ Optimizations
 * GITHUB#13582: Stop requiring MaxScoreBulkScorer's outer window from having at
   least INNER_WINDOW_SIZE docs. (Adrien Grand)
 
+* GITHUB#13570, GITHUB#13574, GITHUB#13535: Avoid performance degradation with closing shared Arenas.
+  Closing many individual index files can potentially lead to a degradation in execution performance.
+  Index files are mmapped one-to-one with the JDK's foreign shared Arena. The JVM deoptimizes the top
+  few frames of all threads when closing a shared Arena (see JDK-8335480). We mitigate this situation
+  by 1) using a confined Arena where appropriate, and 2) grouping files from the same segment to a
+  single shared Arena. (Chris Hegarty, Michael Gibney, Uwe Schindler)
+
 Changes in runtime behavior
 ---------------------
 

lucene/core/src/java/org/apache/lucene/store/MMapDirectory.java

@@ -16,14 +16,20 @@
  */
 package org.apache.lucene.store;
 
+import static org.apache.lucene.index.IndexFileNames.CODEC_FILE_PATTERN;
+
 import java.io.IOException;
 import java.lang.invoke.MethodHandles;
 import java.lang.invoke.MethodType;
 import java.nio.channels.ClosedChannelException; // javadoc @link
 import java.nio.file.Path;
 import java.util.Locale;
+import java.util.Optional;
 import java.util.concurrent.Future;
 import java.util.function.BiPredicate;
+import java.util.function.Function;
+import java.util.logging.Logger;
+import org.apache.lucene.index.IndexFileNames;
 import org.apache.lucene.util.Constants;
 
 /**
@@ -42,6 +48,11 @@
  * performance of searches on a cold page cache at the expense of slowing down opening an index. See
  * {@link #setPreload(BiPredicate)} for more details.
  *
+ * <p>This class supports grouping of files that are part of the same logical group. This is a hint
+ * that allows for better handling of resources. For example, individual files that are part of the
+ * same segment can be considered part of the same logical group. See {@link
+ * #setGroupingFunction(Function)} for more details.
+ *
  * <p>This class will use the modern {@link java.lang.foreign.MemorySegment} API available since
  * Java 21 which allows to safely unmap previously mmapped files after closing the {@link
  * IndexInput}s. There is no need to enable the "preview feature" of your Java version; it works out
@@ -83,6 +94,41 @@ public class MMapDirectory extends FSDirectory {
    */
   public static final BiPredicate<String, IOContext> NO_FILES = (filename, context) -> false;
 
+  /**
+   * This sysprop allows to control the total maximum number of mmapped files that can be associated
+   * with a single shared {@link java.lang.foreign.Arena foreign Arena}. For example, to set the max
+   * number of permits to 256, pass the following on the command line pass {@code
+   * -Dorg.apache.lucene.store.MMapDirectory.sharedArenaMaxPermits=256}. Setting a value of 1
+   * associates one file to one shared arena.
+   *
+   * @lucene.internal
+   */
+  public static final String SHARED_ARENA_MAX_PERMITS_SYSPROP =
+      "org.apache.lucene.store.MMapDirectory.sharedArenaMaxPermits";
+
+  /** Argument for {@link #setGroupingFunction(Function)} that configures no grouping. */
+  public static final Function<String, Optional<String>> NO_GROUPING = filename -> Optional.empty();
+
+  /** Argument for {@link #setGroupingFunction(Function)} that configures grouping by segment. */
+  public static final Function<String, Optional<String>> GROUP_BY_SEGMENT =
+      filename -> {
+        if (!CODEC_FILE_PATTERN.matcher(filename).matches()) {
+          return Optional.empty();
+        }
+        String groupKey = IndexFileNames.parseSegmentName(filename).substring(1);
+        try {
+          // keep the original generation (=0) in base group, later generations in extra group
+          if (IndexFileNames.parseGeneration(filename) > 0) {
+            groupKey += "-g";
+          }
+        } catch (
+            @SuppressWarnings("unused")
+            NumberFormatException unused) {
+          // does not confirm to the generation syntax, or trash
+        }
+        return Optional.of(groupKey);
+      };
+
   /**
    * Argument for {@link #setPreload(BiPredicate)} that configures files to be preloaded upon
    * opening them if they use the {@link ReadAdvice#RANDOM_PRELOAD} advice.
@@ -102,6 +148,11 @@ public class MMapDirectory extends FSDirectory {
    */
   public static final long DEFAULT_MAX_CHUNK_SIZE;
 
+  /** A provider specific context object or null, that will be passed to openInput. */
+  final Object attachment = PROVIDER.attachment();
+
+  private Function<String, Optional<String>> groupingFunction = GROUP_BY_SEGMENT;
+
   final int chunkSizePower;
 
   /**
@@ -184,6 +235,21 @@ public void setPreload(BiPredicate<String, IOContext> preload) {
     this.preload = preload;
   }
 
+  /**
+   * Configures a grouping function for files that are part of the same logical group. The gathering
+   * of files into a logical group is a hint that allows for better handling of resources.
+   *
+   * <p>By default, grouping is {@link #GROUP_BY_SEGMENT}. To disable, invoke this method with
+   * {@link #NO_GROUPING}.
+   *
+   * @param groupingFunction a function that accepts a file name and returns an optional group key.
+   *     If the optional is present, then its value is the logical group to which the file belongs.
+   *     Otherwise, the file name if not associated with any logical group.
+   */
+  public void setGroupingFunction(Function<String, Optional<String>> groupingFunction) {
+    this.groupingFunction = groupingFunction;
+  }
+
   /**
    * Returns the current mmap chunk size.
    *
@@ -199,20 +265,37 @@ public IndexInput openInput(String name, IOContext context) throws IOException {
     ensureOpen();
     ensureCanRead(name);
     Path path = directory.resolve(name);
-    return PROVIDER.openInput(path, context, chunkSizePower, preload.test(name, context));
+    return PROVIDER.openInput(
+        path,
+        context,
+        chunkSizePower,
+        preload.test(name, context),
+        groupingFunction.apply(name),
+        attachment);
   }
 
   // visible for tests:
-  static final MMapIndexInputProvider PROVIDER;
+  static final MMapIndexInputProvider<Object> PROVIDER;
 
-  interface MMapIndexInputProvider {
-    IndexInput openInput(Path path, IOContext context, int chunkSizePower, boolean preload)
+  interface MMapIndexInputProvider<A> {
+    IndexInput openInput(
+        Path path,
+        IOContext context,
+        int chunkSizePower,
+        boolean preload,
+        Optional<String> group,
+        A attachment)
         throws IOException;
 
     long getDefaultMaxChunkSize();
 
     boolean supportsMadvise();
 
+    /** An optional attachment of the provider, that will be passed to openInput. */
+    default A attachment() {
+      return null;
+    }
+
     default IOException convertMapFailedIOException(
         IOException ioe, String resourceDescription, long bufSize) {
       final String originalMessage;
@@ -256,15 +339,33 @@ default IOException convertMapFailedIOException(
     }
   }
 
-  private static MMapIndexInputProvider lookupProvider() {
+  private static int getSharedArenaMaxPermitsSysprop() {
+    int ret = 1024; // default value
+    try {
+      String str = System.getProperty(SHARED_ARENA_MAX_PERMITS_SYSPROP);
+      if (str != null) {
+        ret = Integer.parseInt(str);
+      }
+    } catch (@SuppressWarnings("unused") NumberFormatException | SecurityException ignored) {
+      Logger.getLogger(MMapDirectory.class.getName())
+          .warning(
+              "Cannot read sysprop "
+                  + SHARED_ARENA_MAX_PERMITS_SYSPROP
+                  + ", so the default value will be used.");
+    }
+    return ret;
+  }
+
+  private static <A> MMapIndexInputProvider<A> lookupProvider() {
+    final var maxPermits = getSharedArenaMaxPermitsSysprop();
     final var lookup = MethodHandles.lookup();
     try {
       final var cls = lookup.findClass("org.apache.lucene.store.MemorySegmentIndexInputProvider");
       // we use method handles, so we do not need to deal with setAccessible as we have private
       // access through the lookup:
-      final var constr = lookup.findConstructor(cls, MethodType.methodType(void.class));
+      final var constr = lookup.findConstructor(cls, MethodType.methodType(void.class, int.class));
       try {
-        return (MMapIndexInputProvider) constr.invoke();
+        return (MMapIndexInputProvider<A>) constr.invoke(maxPermits);
       } catch (RuntimeException | Error e) {
         throw e;
       } catch (Throwable th) {

lucene/core/src/java21/org/apache/lucene/store/MemorySegmentIndexInputProvider.java

@@ -24,20 +24,32 @@
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.logging.Logger;
 import org.apache.lucene.util.Constants;
 import org.apache.lucene.util.Unwrappable;
 
 @SuppressWarnings("preview")
-final class MemorySegmentIndexInputProvider implements MMapDirectory.MMapIndexInputProvider {
+final class MemorySegmentIndexInputProvider
+    implements MMapDirectory.MMapIndexInputProvider<
+        ConcurrentHashMap<String, RefCountedSharedArena>> {
 
   private final Optional<NativeAccess> nativeAccess;
+  private final int sharedArenaMaxPermits;
 
-  MemorySegmentIndexInputProvider() {
+  MemorySegmentIndexInputProvider(int maxPermits) {
     this.nativeAccess = NativeAccess.getImplementation();
+    this.sharedArenaMaxPermits = checkMaxPermits(maxPermits);
   }
 
   @Override
-  public IndexInput openInput(Path path, IOContext context, int chunkSizePower, boolean preload)
+  public IndexInput openInput(
+      Path path,
+      IOContext context,
+      int chunkSizePower,
+      boolean preload,
+      Optional<String> group,
+      ConcurrentHashMap<String, RefCountedSharedArena> arenas)
       throws IOException {
     final String resourceDescription = "MemorySegmentIndexInput(path=\"" + path.toString() + "\")";
 
@@ -46,7 +58,7 @@ public IndexInput openInput(Path path, IOContext context, int chunkSizePower, bo
 
     boolean success = false;
     final boolean confined = context == IOContext.READONCE;
-    final Arena arena = confined ? Arena.ofConfined() : Arena.ofShared();
+    final Arena arena = confined ? Arena.ofConfined() : getSharedArena(group, arenas);
     try (var fc = FileChannel.open(path, StandardOpenOption.READ)) {
       final long fileSize = fc.size();
       final IndexInput in =
@@ -125,4 +137,53 @@ private final MemorySegment[] map(
     }
     return segments;
   }
+
+  @Override
+  public ConcurrentHashMap<String, RefCountedSharedArena> attachment() {
+    return new ConcurrentHashMap<>();
+  }
+
+  private static int checkMaxPermits(int maxPermits) {
+    if (RefCountedSharedArena.validMaxPermits(maxPermits)) {
+      return maxPermits;
+    }
+    Logger.getLogger(MemorySegmentIndexInputProvider.class.getName())
+        .warning(
+            "Invalid value for sysprop "
+                + MMapDirectory.SHARED_ARENA_MAX_PERMITS_SYSPROP
+                + ", must be positive and <= 0x07FF. The default value will be used.");
+    return RefCountedSharedArena.DEFAULT_MAX_PERMITS;
+  }
+
+  /**
+   * Gets an arena for the given group, potentially aggregating files from the same segment into a
+   * single ref counted shared arena. A ref counted shared arena, if created will be added to the
+   * given arenas map.
+   */
+  private Arena getSharedArena(
+      Optional<String> group, ConcurrentHashMap<String, RefCountedSharedArena> arenas) {
+    if (group.isEmpty()) {
+      return Arena.ofShared();
+    }
+
+    String key = group.get();
+    var refCountedArena =
+        arenas.computeIfAbsent(
+            key, s -> new RefCountedSharedArena(s, () -> arenas.remove(s), sharedArenaMaxPermits));
+    if (refCountedArena.acquire()) {
+      return refCountedArena;
+    } else {
+      return arenas.compute(
+          key,
+          (s, v) -> {
+            if (v != null && v.acquire()) {
+              return v;
+            } else {
+              v = new RefCountedSharedArena(s, () -> arenas.remove(s), sharedArenaMaxPermits);
+              v.acquire(); // guaranteed to succeed
+              return v;
+            }
+          });
+    }
+  }
 }