Aggregate files from the same segment into a single Arena

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

@@ -102,6 +102,9 @@ 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();
+
   final int chunkSizePower;
 
   /**
@@ -199,20 +202,27 @@ 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), 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, 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 +266,15 @@ default IOException convertMapFailedIOException(
     }
   }
 
-  private static MMapIndexInputProvider lookupProvider() {
+  private static <A> MMapIndexInputProvider<A> lookupProvider() {
     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));
       try {
-        return (MMapIndexInputProvider) constr.invoke();
+        return (MMapIndexInputProvider<A>) constr.invoke();
       } catch (RuntimeException | Error e) {
         throw e;
       } catch (Throwable th) {

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

@@ -24,11 +24,15 @@
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.lucene.index.IndexFileNames;
 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;
 
@@ -37,7 +41,12 @@ final class MemorySegmentIndexInputProvider implements MMapDirectory.MMapIndexIn
   }
 
   @Override
-  public IndexInput openInput(Path path, IOContext context, int chunkSizePower, boolean preload)
+  public IndexInput openInput(
+      Path path,
+      IOContext context,
+      int chunkSizePower,
+      boolean preload,
+      ConcurrentHashMap<String, RefCountedSharedArena> arenas)
       throws IOException {
     final String resourceDescription = "MemorySegmentIndexInput(path=\"" + path.toString() + "\")";
 
@@ -46,7 +55,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(path, arenas);
     try (var fc = FileChannel.open(path, StandardOpenOption.READ)) {
       final long fileSize = fc.size();
       final IndexInput in =
@@ -125,4 +134,31 @@ private final MemorySegment[] map(
     }
     return segments;
   }
+
+  public ConcurrentHashMap<String, RefCountedSharedArena> attachment() {
+    return new ConcurrentHashMap<>();
+  }
+
+  /**
+   * Gets an arena for the give path, 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.
+   */
+  static Arena getSharedArena(Path p, ConcurrentHashMap<String, RefCountedSharedArena> arenas) {
+    String filename = p.getFileName().toString();
+    String segmentName = IndexFileNames.parseSegmentName(filename);
+    if (filename.length() == segmentName.length()) {
+      // no segment found; just use a shared segment
+      return Arena.ofShared();
+    }
+
+    while (true) {
+      var refCountedArena =
+          arenas.computeIfAbsent(
+              segmentName, s -> new RefCountedSharedArena(s, () -> arenas.remove(s)));
+      if (refCountedArena.acquire()) {
+        return refCountedArena;
+      }
+    }
+  }
 }

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

@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.store;
+
+import java.lang.foreign.Arena;
+import java.lang.foreign.MemorySegment;
+import java.util.concurrent.atomic.AtomicInteger;
+
+@SuppressWarnings("preview")
+final class RefCountedSharedArena implements Arena {
+
+  static final int OPEN = 0;
+  static final int CLOSED = -1;
+
+  private final String segmentName;
+  private final Runnable removeFromMap;
+  private final Arena arena;
+  private final AtomicInteger state;
+
+  RefCountedSharedArena(String segmentName, Runnable removeFromMap) {
+    this.segmentName = segmentName;
+    this.removeFromMap = removeFromMap;
+    this.arena = Arena.ofShared();
+    this.state = new AtomicInteger(OPEN);
+  }
+
+  // for debugging
+  String getSegmentName() {
+    return segmentName;
+  }
+
+  boolean acquire() {
+    int value;
+    while (true) {
+      value = state.get();
+      if (value < OPEN) {
+        return false;
+      }
+      if (state.compareAndSet(value, value + 1)) {
+        return true;
+      }
+    }
+  }
+
+  void release() {
+    int value;
+    while (true) {
+      value = state.get();
+      if (value <= OPEN) {
+        throw new IllegalStateException("already closed");
+      }
+      if (state.compareAndSet(value, value - 1)) {
+        if (value - 1 == OPEN && state.compareAndSet(OPEN, CLOSED)) {
+          removeFromMap.run();
+          arena.close();
+        }
+        return;
+      }
+    }
+  }
+
+  @Override
+  public void close() {
+    release();
+  }
+
+  @Override
+  public MemorySegment allocate(long byteSize, long byteAlignment) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public MemorySegment.Scope scope() {
+    return arena.scope();
+  }
+
+  @Override
+  public String toString() {
+    return "RefCountedArena[segmentName="
+        + segmentName
+        + ", value="
+        + state.get()
+        + ", arena="
+        + arena
+        + "]";
+  }
+}

lucene/core/src/test/org/apache/lucene/store/TestMMapDirectory.java

@@ -16,14 +16,21 @@
  */
 package org.apache.lucene.store;
 
+import static java.util.stream.Collectors.toList;
+
 import java.io.IOException;
 import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.Random;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.stream.IntStream;
 import org.apache.lucene.tests.store.BaseDirectoryTestCase;
 import org.apache.lucene.util.Constants;
 import org.apache.lucene.util.NamedThreadFactory;
@@ -172,4 +179,76 @@ static Object getAndUnwrap(Future<Object> future) throws Throwable {
       throw ee.getCause();
     }
   }
+
+  public void testArenas() throws Exception {
+    // First create a number of file name lists to test with.
+    var exts =
+        List.of(
+            ".si", ".cfs", ".cfe", ".dvd", ".dvm", "nvd", ".nvm", ".fdt", ".vec", ".vex", ".vemf");
+    var names =
+        IntStream.range(0, 50)
+            .mapToObj(i -> "_" + i)
+            .flatMap(s -> exts.stream().map(ext -> s + ext))
+            .collect(toList());
+    IntStream.range(0, 50).mapToObj(i -> "segment_" + i).forEach(names::add);
+    Collections.shuffle(names, random());
+
+    final int size = 6;
+    byte[] bytes = new byte[size];
+    random().nextBytes(bytes);
+
+    try (var dir = new MMapDirectory(createTempDir("testArenas"))) {
+      for (var name : names) {
+        try (IndexOutput out = dir.createOutput(name, IOContext.DEFAULT)) {
+          out.writeBytes(bytes, 0, bytes.length);
+        }
+      }
+
+      int nThreads = 10;
+      int perListSize = (names.size() + nThreads) / nThreads;
+      List<List<String>> nameLists =
+          IntStream.range(0, nThreads)
+              .mapToObj(
+                  i ->
+                      names.subList(
+                          perListSize * i, Math.min(perListSize * i + perListSize, names.size())))
+              .toList();
+
+      var threadFactory = new NamedThreadFactory("testArenas");
+      try (var executor = Executors.newFixedThreadPool(nThreads, threadFactory)) {
+        var tasks = nameLists.stream().map(l -> new IndicesOpenTask(l, dir)).toList();
+        var futures = tasks.stream().map(executor::submit).toList();
+        for (var future : futures) {
+          future.get();
+        }
+      }
+
+      if (!(dir.attachment instanceof ConcurrentHashMap<?, ?> map)) {
+        throw new AssertionError("unexpected attachment: " + dir.attachment);
+      }
+      assertEquals(0, map.size());
+    }
+  }
+
+  static class IndicesOpenTask implements Callable<Void> {
+    final List<String> names;
+    final Directory dir;
+
+    IndicesOpenTask(List<String> names, Directory dir) {
+      this.names = names;
+      this.dir = dir;
+    }
+
+    @Override
+    public Void call() throws Exception {
+      List<IndexInput> closeables = new ArrayList<>();
+      for (var name : names) {
+        closeables.add(dir.openInput(name, IOContext.DEFAULT));
+      }
+      for (IndexInput closeable : closeables) {
+        closeable.close();
+      }
+      return null;
+    }
+  }
 }