Adding asynchronous fetching for DirectIO directory (#134803)

One significant cost of DirectIO is simply waiting for bytes to be read
in a path dedicate for compute. 

This change adds "prefetch" capabilities to DirectIO by allowing to
prefetch particular file positions. For simplicity, I have it always
prefetch a DirectIO page. Initially I did a bunch of work to allow
prefetching multiple pages (e.g. more than 8192 bytes), but this greatly
complicated the implementation. I think this can be added as a follow
up.

Here are some benchmarks for vectors. Note, the recall difference
indicates I am doing something wrong right now. I am thinking I have a
couple off-by-one errors and I am still investigating.

Opening as a draft until I can figure out this weird bug (and of course,
remove all my extraneous changes used for testing this thing)...This is
labeled as 9.2, but I would be very surprise if it actually lands there.

This PR:

```
index_name                      index_type  visit_percentage(%)  latency(ms)  net_cpu_time(ms)  avg_cpu_count     QPS  recall   visited  filter_selectivity
------------------------------  ----------  -------------------  -----------  ----------------  -------------  ------  ------  --------  ------------------
cohere-wikipedia-docs-768d.vec         ivf                 1.00         5.57              0.00           0.00  179.53    0.92  87397.37                1.00
```

Baseline DirectIO:

```
index_name                      index_type  visit_percentage(%)  latency(ms)  net_cpu_time(ms)  avg_cpu_count     QPS  recall   visited  filter_selectivity
------------------------------  ----------  -------------------  -----------  ----------------  -------------  ------  ------  --------  ------------------
cohere-wikipedia-docs-768d.vec         ivf                 1.00         8.12              0.00           0.00  123.15    0.94  87397.37                1.00
```

Baseline MMAP (when many floating points can still just reside in
memory):

```
index_name                      index_type  visit_percentage(%)  latency(ms)  net_cpu_time(ms)  avg_cpu_count     QPS  recall   visited  filter_selectivity
------------------------------  ----------  -------------------  -----------  ----------------  -------------  ------  ------  --------  ------------------
cohere-wikipedia-docs-768d.vec         ivf                 1.00         3.58              0.00           0.00  279.33    0.94  87397.37                1.00
```

Author: benwtrent

qa/vector/src/main/java/module-info.java

@@ -18,5 +18,6 @@
     requires org.elasticsearch.logging;
     requires java.management;
     requires jdk.management;
+    requires org.apache.lucene.misc;
     requires org.elasticsearch.gpu;
 }

qa/vector/src/main/java/org/elasticsearch/test/knn/KnnIndexer.java

@@ -34,16 +34,11 @@
 import org.apache.lucene.index.VectorSimilarityFunction;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.FSDirectory;
-import org.apache.lucene.store.IOContext;
-import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.store.MMapDirectory;
-import org.apache.lucene.store.NIOFSDirectory;
 import org.apache.lucene.store.NativeFSLockFactory;
 import org.apache.lucene.util.PrintStreamInfoStream;
 import org.elasticsearch.common.io.Channels;
-import org.elasticsearch.core.IOUtils;
-import org.elasticsearch.index.store.LuceneFilesExtensions;
-import org.elasticsearch.index.store.Store;
+import org.elasticsearch.index.store.FsDirectoryFactory;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -233,7 +228,7 @@ public boolean isEnabled(String component) {
     static Directory getDirectory(Path indexPath) throws IOException {
         Directory dir = FSDirectory.open(indexPath);
         if (dir instanceof MMapDirectory mmapDir) {
-            return new HybridDirectory(mmapDir);
+            return new FsDirectoryFactory.HybridDirectory(NativeFSLockFactory.INSTANCE, mmapDir, 64);
         }
         return dir;
     }
@@ -375,64 +370,4 @@ synchronized void next(byte[] dest) throws IOException {
             bytes.get(dest);
         }
     }
-
-    // Copy of Elastic's HybridDirectory which extends NIOFSDirectory and uses MMapDirectory for certain files.
-    static final class HybridDirectory extends NIOFSDirectory {
-        private final MMapDirectory delegate;
-
-        HybridDirectory(MMapDirectory delegate) throws IOException {
-            super(delegate.getDirectory(), NativeFSLockFactory.INSTANCE);
-            this.delegate = delegate;
-        }
-
-        @Override
-        public IndexInput openInput(String name, IOContext context) throws IOException {
-            if (useDelegate(name, context)) {
-                // we need to do these checks on the outer directory since the inner doesn't know about pending deletes
-                ensureOpen();
-                ensureCanRead(name);
-                // we switch the context here since mmap checks for the READONCE context by identity
-                context = context == Store.READONCE_CHECKSUM ? IOContext.READONCE : context;
-                // we only use the mmap to open inputs. Everything else is managed by the NIOFSDirectory otherwise
-                // we might run into trouble with files that are pendingDelete in one directory but still
-                // listed in listAll() from the other. We on the other hand don't want to list files from both dirs
-                // and intersect for perf reasons.
-                return delegate.openInput(name, context);
-            } else {
-                return super.openInput(name, context);
-            }
-        }
-
-        @Override
-        public void close() throws IOException {
-            IOUtils.close(super::close, delegate);
-        }
-
-        private static String getExtension(String name) {
-            // Unlike FileSwitchDirectory#getExtension, we treat `tmp` as a normal file extension, which can have its own rules for mmaping.
-            final int lastDotIndex = name.lastIndexOf('.');
-            if (lastDotIndex == -1) {
-                return "";
-            } else {
-                return name.substring(lastDotIndex + 1);
-            }
-        }
-
-        static boolean useDelegate(String name, IOContext ioContext) {
-            if (ioContext == Store.READONCE_CHECKSUM) {
-                // If we're just reading the footer for the checksum then mmap() isn't really necessary, and it's desperately inefficient
-                // if pre-loading is enabled on this file.
-                return false;
-            }
-
-            final LuceneFilesExtensions extension = LuceneFilesExtensions.fromExtension(getExtension(name));
-            if (extension == null || extension.shouldMmap() == false) {
-                // Other files are either less performance-sensitive (e.g. stored field index, norms metadata)
-                // or are large and have a random access pattern and mmap leads to page cache trashing
-                // (e.g. stored fields and term vectors).
-                return false;
-            }
-            return true;
-        }
-    }
 }

server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java

@@ -179,6 +179,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
                 IndexModule.INDEX_RECOVERY_TYPE_SETTING,
                 IndexModule.INDEX_QUERY_CACHE_ENABLED_SETTING,
                 FsDirectoryFactory.INDEX_LOCK_FACTOR_SETTING,
+                FsDirectoryFactory.ASYNC_PREFETCH_LIMIT,
                 EngineConfig.INDEX_CODEC_SETTING,
                 IndexMetadata.SETTING_WAIT_FOR_ACTIVE_SHARDS,
                 IndexSettings.DEFAULT_PIPELINE,