diff --git a/docs/batched-vector-access.md b/docs/batched-vector-access.md index 0a3c585cb..c31673018 100644 --- a/docs/batched-vector-access.md +++ b/docs/batched-vector-access.md @@ -87,12 +87,14 @@ useDatasetPrefetching: true ``` Omitted or `false` preserves the resident path. The existing VM option -`-Djvector.dataset_loader=buffered` is an explicit override; `preload` overrides the -flag in the other direction. Neither runner changes a user's YAML files. +`-Djvector.dataset_loader=buffered` explicitly overrides the YAML flag; `preload` +overrides the flag in the other direction. Neither runner changes a user's YAML files. The harness uses file input for bulk consumption in quantized construction and existing disk storage. Uncompressed construction, or a search configuration that requires raw input rather than stored vectors/codes, still uses resident input. +The VM option does not bypass this resident-input routing. If an explicit +`buffered` request requires resident input, the harness logs the reason for using preload. A mixed configuration requiring resident input remains resident; split those cases into separate configurations to measure dataset prefetching independently. This is source routing, not a restriction on the vector-access API. @@ -267,6 +269,8 @@ reads and payload allocation, not physical device I/O. - Close cursors with try-with-resources, stop consumers, then close the root source. - Copies share the root's storage and I/O executor. Closing a copy affects that view only; closing the root invalidates all copies and releases the shared resources. +- If point reads are still running when the root closes, close waits for each read + in progress to finish; later point reads fail with `IllegalStateException`. - Direct payload buffers become eligible for JVM reclamation after references are released; closure does not promise immediate native-memory reclamation. diff --git a/docs/release notes/4.0.2/744.feature.md b/docs/release notes/4.0.2/744.feature.md index 9efe1e844..750d2cf28 100644 --- a/docs/release notes/4.0.2/744.feature.md +++ b/docs/release notes/4.0.2/744.feature.md @@ -6,7 +6,9 @@ of the original base dataset; existing vector sources remain supported. BenchYAML and AutoBenchYAML support `useDatasetPrefetching: true` in their index configuration. The flag defaults to false, preserving resident loading and scoring. Comparison-heavy uncompressed configurations retain resident input. The existing -`jvector.dataset_loader` VM option remains an explicit override. +`jvector.dataset_loader` VM option overrides the flag, while configurations +requiring resident raw vectors still preload; an explicit buffered request logs +the reason for that fallback. See [batched vector access](../../batched-vector-access.md) for ownership, resource limits, examples and the benchmark opt-in. Encoded vectors, training diff --git a/jvector-base/src/main/java/io/github/jbellis/jvector/disk/FvecFileVectorValues.java b/jvector-base/src/main/java/io/github/jbellis/jvector/disk/FvecFileVectorValues.java index 64b27d30c..8a1815d94 100644 --- a/jvector-base/src/main/java/io/github/jbellis/jvector/disk/FvecFileVectorValues.java +++ b/jvector-base/src/main/java/io/github/jbellis/jvector/disk/FvecFileVectorValues.java @@ -55,7 +55,9 @@ * has its own borrowed vector. * *

Close all cursors and stop consumers before closing the source. Closing the root - * releases all resources, including resources used by copies. No memory mapping or + * releases all resources, including resources used by copies. A point read already in + * progress finishes before its channel is closed; later point reads, including ones + * waiting for a reader, fail with {@link IllegalStateException}. No memory mapping or * OS cache eviction is performed. Retained training vectors are caller-owned memory. */ public final class FvecFileVectorValues implements BatchedVectorValues, AutoCloseable { @@ -463,6 +465,8 @@ void readPointInto(int ordinal, VectorFloat destination, int offset) { PointReader reader = null; try { reader = borrowPointReader(); + // Close waits for each lease, so a closed source can't reopen this reader's channel. + if (closed) throw new IllegalStateException("Vector source is closed"); pointHint.set(reader); var input = pointChannel(reader); var buffer = reader.buffer; @@ -626,10 +630,12 @@ void close() throws IOException { } IOException failure = null; for (PointReader reader : pointReaders) { + // Wait for an in-flight point read to finish; readers that lease afterwards see closed. + reader.permit.acquireUninterruptibly(); try { if (reader.channel != null) reader.channel.close(); } catch (IOException error) { if (failure == null) failure = error; else failure.addSuppressed(error); - } finally { reader.channel = null; reader.buffer = null; } + } finally { reader.channel = null; reader.buffer = null; reader.permit.release(); } } if (pointHint != null) pointHint.remove(); try { channel.close(); } diff --git a/jvector-base/src/main/java/io/github/jbellis/jvector/graph/similarity/BuildScoreProvider.java b/jvector-base/src/main/java/io/github/jbellis/jvector/graph/similarity/BuildScoreProvider.java index bda700d04..8e1f91cf8 100644 --- a/jvector-base/src/main/java/io/github/jbellis/jvector/graph/similarity/BuildScoreProvider.java +++ b/jvector-base/src/main/java/io/github/jbellis/jvector/graph/similarity/BuildScoreProvider.java @@ -110,11 +110,23 @@ static BuildScoreProvider randomAccessScoreProvider(RandomAccessVectorValues rav var vectorsCopy = ravv.threadLocalSupplier(); return new BuildScoreProvider() { + /// Exact scores against `node1`. + /// + /// For a value-shared source the anchor is read on the first comparison and kept as + /// a private copy, so creating a score function reads nothing, later reads through the + /// same views can't overwrite the anchor, and each comparison reads only `node2`. private ScoreFunction.ExactScoreFunction scoreFunctionFor(int node1) { if (ravv.isValueShared()) { - // Resolve both ordinals using separate borrowed views; no anchor escapes a read. - return node2 -> similarityFunction.compare(vectors.get().getVector(node1), - vectorsCopy.get().getVector(node2)); + return new ScoreFunction.ExactScoreFunction() { + private VectorFloat anchor; + + @Override + public float similarityTo(int node2) { + var a = anchor; + if (a == null) anchor = a = vectors.get().getVector(node1).copy(); + return similarityFunction.compare(a, vectorsCopy.get().getVector(node2)); + } + }; } var vector = vectors.get().getVector(node1); var vc = vectorsCopy.get(); diff --git a/jvector-examples/pom.xml b/jvector-examples/pom.xml index 088f60d8e..843c8ef25 100644 --- a/jvector-examples/pom.xml +++ b/jvector-examples/pom.xml @@ -15,6 +15,7 @@ 2.21.10 + @@ -23,6 +24,7 @@ maven-surefire-plugin false + ${argLine} ${project.parent.basedir} @@ -275,6 +277,9 @@ jdk20 + + --add-modules jdk.incubator.vector + io.github.jbellis @@ -395,6 +400,9 @@ jdk22 + + --add-modules jdk.incubator.vector --enable-native-access=ALL-UNNAMED + true diff --git a/jvector-examples/src/main/java/io/github/jbellis/jvector/example/benchmarks/datasets/BaseVectorLoading.java b/jvector-examples/src/main/java/io/github/jbellis/jvector/example/benchmarks/datasets/BaseVectorLoading.java index a01dd972f..79965a0ef 100644 --- a/jvector-examples/src/main/java/io/github/jbellis/jvector/example/benchmarks/datasets/BaseVectorLoading.java +++ b/jvector-examples/src/main/java/io/github/jbellis/jvector/example/benchmarks/datasets/BaseVectorLoading.java @@ -17,6 +17,7 @@ package io.github.jbellis.jvector.example.benchmarks.datasets; import java.util.Locale; +import java.util.logging.Logger; import io.github.jbellis.jvector.example.yaml.MultiConfig; import io.github.jbellis.jvector.graph.disk.feature.FeatureId; @@ -25,6 +26,8 @@ public enum BaseVectorLoading { BUFFERED, PRELOAD; + private static final Logger LOG = Logger.getLogger(BaseVectorLoading.class.getName()); + /** Existing VM option is an explicit override of the YAML flag. */ public static BaseVectorLoading forBenchmark(boolean useDatasetPrefetching) { String value = System.getProperty("jvector.dataset_loader", @@ -37,16 +40,25 @@ public static BaseVectorLoading forBenchmark(boolean useDatasetPrefetching) { public static BaseVectorLoading forBenchmark() { return forBenchmark(false); } - /** Bulk input mode applies only where the harness consumes quantized or stored vectors. */ + /** + * The VM option overrides the YAML flag. Configurations requiring resident raw + * vectors retain preload so construction and query scoring stay unchanged. + */ public static BaseVectorLoading forBenchmark(MultiConfig config) { var loading = forBenchmark(config.useDatasetPrefetching); if (loading == PRELOAD) return loading; - // Uncompressed construction still uses main's resident raw-vector source. if (config.construction.compression.stream().anyMatch(c -> "None".equals(c.type))) - return PRELOAD; + return residentInput("uncompressed construction"); boolean exactSearch = config.search.compression.stream().anyMatch(c -> "None".equals(c.type)); boolean needsRawInput = exactSearch && config.construction.getFeatureSets().stream().anyMatch(features -> !features.contains(FeatureId.FUSED_PQ)); - return needsRawInput ? PRELOAD : BUFFERED; + return needsRawInput ? residentInput("raw-input exact search") : BUFFERED; + } + + private static BaseVectorLoading residentInput(String reason) { + if ("buffered".equalsIgnoreCase(System.getProperty("jvector.dataset_loader"))) + LOG.warning("Using preload instead of jvector.dataset_loader=buffered for " + reason + + "; retaining resident input to preserve scoring behavior."); + return PRELOAD; } } diff --git a/jvector-tests/src/test/java/io/github/jbellis/jvector/disk/TestFvecCloseDuringPointReads.java b/jvector-tests/src/test/java/io/github/jbellis/jvector/disk/TestFvecCloseDuringPointReads.java new file mode 100644 index 000000000..459ea391f --- /dev/null +++ b/jvector-tests/src/test/java/io/github/jbellis/jvector/disk/TestFvecCloseDuringPointReads.java @@ -0,0 +1,119 @@ +/* + * Copyright DataStax, Inc. + * + * Licensed 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 io.github.jbellis.jvector.disk; + +import org.junit.Assume; +import org.junit.Test; + +import java.io.UncheckedIOException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.channels.ClosedChannelException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.SplittableRandom; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; + + +/// Closing the root while point reads are in flight must fail those reads cleanly, as the rest +/// of the source does ("Vector source is closed" or a closed-channel I/O error), and must leave +/// no file descriptors open on the input file. +/// +/// The race is timing-dependent, so the test repeats close-under-load until it sees a +/// violation or the round limit is reached. A pass is not proof of absence. +public class TestFvecCloseDuringPointReads { + private static final int ROWS = 10_000, DIMENSIONS = 16, READERS = 12, ROUNDS = 300; + + private static Path writeFvecs() throws Exception { + var file = Files.createTempFile("close-during-point-reads", ".fvecs"); + var bytes = ByteBuffer.allocate(ROWS * (DIMENSIONS + 1) * 4).order(ByteOrder.LITTLE_ENDIAN); + for (int row = 0; row < ROWS; row++) { + bytes.putInt(DIMENSIONS); + for (int d = 0; d < DIMENSIONS; d++) bytes.putFloat(row); + } + Files.write(file, bytes.array()); + return file; + } + + /// Expected failures after close: the source reports closure, or the leased channel was + /// closed underneath the read. + private static boolean isCleanClosureFailure(Throwable t) { + if (t instanceof IllegalStateException) return true; + return t instanceof UncheckedIOException && t.getCause() instanceof ClosedChannelException; + } + + private static long openDescriptorsOn(Path file) throws Exception { + var real = file.toRealPath(); + long count = 0; + try (var fds = Files.list(Path.of("/proc/self/fd"))) { + for (var fd : (Iterable) fds::iterator) { + try { if (Files.readSymbolicLink(fd).equals(real)) count++; } + catch (java.io.IOException ignored) { } // descriptor closed while listing + } + } + return count; + } + + @Test(timeout = 120_000) + public void closeUnderPointLoadFailsCleanlyAndLeaksNoDescriptors() throws Exception { + Assume.assumeTrue("needs /proc/self/fd", Files.isDirectory(Path.of("/proc/self/fd"))); + var file = writeFvecs(); + try { + for (int round = 0; round < ROUNDS; round++) { + var unexpected = new ConcurrentLinkedQueue(); + var root = FvecFileVectorValues.open(file); + var started = new CountDownLatch(READERS); + List readers = new ArrayList<>(); + for (int r = 0; r < READERS; r++) { + var view = root.copy(); + long seed = round * 1000L + r; + var thread = new Thread(() -> { + var random = new SplittableRandom(seed); + started.countDown(); + try { + while (true) view.getVector(random.nextInt(ROWS)); + } catch (Throwable t) { + if (!isCleanClosureFailure(t)) unexpected.add(t); + } + }, "point-reader-" + r); + readers.add(thread); + thread.start(); + } + started.await(); + Thread.sleep(1); + root.close(); + for (var thread : readers) thread.join(); + + long leaked = openDescriptorsOn(file); + if (!unexpected.isEmpty() || leaked != 0) { + var first = unexpected.peek(); + var error = new AssertionError("round " + round + ": " + unexpected.size() + + " point read(s) failed with an unexpected exception after close" + + (first == null ? "" : " (first: " + first + ")") + + "; " + leaked + " file descriptor(s) still open on the input"); + if (first != null) error.initCause(first); + throw error; + } + } + } finally { + Files.delete(file); + } + } +} diff --git a/jvector-tests/src/test/java/io/github/jbellis/jvector/example/TestDatasetLoaderOverride.java b/jvector-tests/src/test/java/io/github/jbellis/jvector/example/TestDatasetLoaderOverride.java new file mode 100644 index 000000000..a04c01959 --- /dev/null +++ b/jvector-tests/src/test/java/io/github/jbellis/jvector/example/TestDatasetLoaderOverride.java @@ -0,0 +1,98 @@ +/* + * Copyright DataStax, Inc. + * + * Licensed 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 io.github.jbellis.jvector.example; + +import io.github.jbellis.jvector.example.benchmarks.datasets.BaseVectorLoading; +import io.github.jbellis.jvector.example.yaml.ConstructionParameters; +import io.github.jbellis.jvector.example.yaml.MultiConfig; +import io.github.jbellis.jvector.example.yaml.SearchParameters; +import io.github.jbellis.jvector.example.yaml.Compression; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.logging.Handler; +import java.util.logging.LogRecord; +import java.util.logging.Logger; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +/** The VM override selects loading without changing resident construction/query routing. */ +public class TestDatasetLoaderOverride { + private static final String PROPERTY = "jvector.dataset_loader"; + private String previous; + private final List warnings = new ArrayList<>(); + private final Logger logger = Logger.getLogger(BaseVectorLoading.class.getName()); + private final Handler handler = new Handler() { + @Override public void publish(LogRecord record) { warnings.add(record.getMessage()); } + @Override public void flush() { } + @Override public void close() { } + }; + + @Before public void saveProperty() { + previous = System.getProperty(PROPERTY); + logger.addHandler(handler); + } + + @After public void restoreProperty() { + logger.removeHandler(handler); + if (previous == null) System.clearProperty(PROPERTY); + else System.setProperty(PROPERTY, previous); + } + + private static MultiConfig config(String build, String search) { + var config = new MultiConfig(); + config.construction = new ConstructionParameters(); + config.search = new SearchParameters(); + var bc = new Compression(); bc.type = build; + var sc = new Compression(); sc.type = search; + config.construction.compression = List.of(bc); + config.construction.fusedGraph = List.of(false); + config.construction.reranking = List.of("NVQ"); + config.search.compression = List.of(sc); + return config; + } + + @Test public void explicitBufferedPropertyRetainsExactSearchRoutingAndWarns() { + System.setProperty(PROPERTY, "buffered"); + assertEquals(BaseVectorLoading.PRELOAD, BaseVectorLoading.forBenchmark(config("PQ", "None"))); + assertTrue(warnings.stream().anyMatch(message -> message.contains("raw-input exact search"))); + } + + @Test public void explicitBufferedPropertyRetainsUncompressedConstructionAndWarns() { + System.setProperty(PROPERTY, "buffered"); + assertEquals(BaseVectorLoading.PRELOAD, BaseVectorLoading.forBenchmark(config("None", "PQ"))); + assertTrue(warnings.stream().anyMatch(message -> message.contains("uncompressed construction"))); + } + + @Test public void explicitBufferedPropertyOverridesDisabledFlagForQuantizedWorkloads() { + System.setProperty(PROPERTY, "buffered"); + assertEquals(BaseVectorLoading.BUFFERED, BaseVectorLoading.forBenchmark(config("PQ", "PQ"))); + assertTrue(warnings.isEmpty()); + } + + @Test public void explicitPreloadPropertyOverridesEnabledFlag() { + System.setProperty(PROPERTY, "preload"); + var config = config("PQ", "PQ"); + config.useDatasetPrefetching = true; + assertEquals(BaseVectorLoading.PRELOAD, BaseVectorLoading.forBenchmark(config)); + assertTrue(warnings.isEmpty()); + } +} diff --git a/jvector-tests/src/test/java/io/github/jbellis/jvector/example/TestDatasetPrefetching.java b/jvector-tests/src/test/java/io/github/jbellis/jvector/example/TestDatasetPrefetching.java index 5ecf83c9f..d94ae48a9 100644 --- a/jvector-tests/src/test/java/io/github/jbellis/jvector/example/TestDatasetPrefetching.java +++ b/jvector-tests/src/test/java/io/github/jbellis/jvector/example/TestDatasetPrefetching.java @@ -105,16 +105,19 @@ private MultiConfig config(String build, String search, boolean fused, String re public VectorCursor openRange(int a,int b){ranges.incrementAndGet();return VectorAccess.openRange(resident,a,b);} public VectorCursor openSelection(int[] ordinals,int a,int n){selections.incrementAndGet();return VectorAccess.openSelection(resident,ordinals,a,n);} }; + // A previous test run must not supply this run's training codebook. + var datasetName = "bulk-wiring-" + UUID.randomUUID(); DataSet ds = new DataSet() { public boolean useDatasetPrefetching(){return true;} public int getDimension(){return 17;} - public String getName(){return "bulk-wiring-"+System.identityHashCode(this);} + public String getName(){return datasetName;} public VectorSimilarityFunction getSimilarityFunction(){return VectorSimilarityFunction.DOT_PRODUCT;} public RandomAccessVectorValues getBaseRavv(){return input;} public List> getBaseVectors(){return vectors;} public List> getQueryVectors(){return List.of(vectors.get(0));} public List> getGroundTruth(){return List.of(List.of(0));} }; + var cacheFile = Path.of("pq_cache").resolve(new CompressorParameters.PQParameters(4,16,false,-1f).idStringFor(ds)); var work = Files.createTempDirectory("bulk-grid-wiring"); try { Grid.runOneGraph(OnDiskGraphIndexCache.initialize(work.resolve("saved")), List.of(EnumSet.of(FeatureId.INLINE_VECTORS),EnumSet.of(FeatureId.NVQ_VECTORS)), @@ -139,7 +142,10 @@ private MultiConfig config(String build, String search, boolean fused, String re } assertEquals(2,seen); } finally { - try(var paths=Files.walk(work)){for(var path:paths.sorted(Comparator.reverseOrder()).collect(java.util.stream.Collectors.toList()))Files.delete(path);} + try { Files.deleteIfExists(cacheFile); } + finally { + try(var paths=Files.walk(work)){for(var path:paths.sorted(Comparator.reverseOrder()).collect(java.util.stream.Collectors.toList()))Files.delete(path);} + } } } } diff --git a/jvector-tests/src/test/java/io/github/jbellis/jvector/graph/TestBuildScoreProviderAnchorReads.java b/jvector-tests/src/test/java/io/github/jbellis/jvector/graph/TestBuildScoreProviderAnchorReads.java new file mode 100644 index 000000000..da55e3a26 --- /dev/null +++ b/jvector-tests/src/test/java/io/github/jbellis/jvector/graph/TestBuildScoreProviderAnchorReads.java @@ -0,0 +1,79 @@ +/* + * Copyright DataStax, Inc. + * + * Licensed 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 io.github.jbellis.jvector.graph; + +import io.github.jbellis.jvector.disk.FvecFileVectorValues; +import io.github.jbellis.jvector.graph.similarity.BuildScoreProvider; +import io.github.jbellis.jvector.vector.VectorSimilarityFunction; +import org.junit.Test; + +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.junit.Assert.assertTrue; + +/// A full-precision build score function compares one fixed anchor against many nodes. For a +/// value-shared file source it should read the anchor once, not once per comparison. +public class TestBuildScoreProviderAnchorReads { + private static final int ROWS = 2000, DIMENSIONS = 16, COMPARISONS = 1000; + + private static Path writeFvecs() throws Exception { + var file = Files.createTempFile("anchor-reads", ".fvecs"); + var bytes = ByteBuffer.allocate(ROWS * (DIMENSIONS + 1) * 4).order(ByteOrder.LITTLE_ENDIAN); + for (int row = 0; row < ROWS; row++) { + bytes.putInt(DIMENSIONS); + for (int d = 0; d < DIMENSIONS; d++) bytes.putFloat(row + d / 32f); + } + Files.write(file, bytes.array()); + return file; + } + + private static void assertAnchorReadOnce(String kind, long reads) { + // One read per compared node, plus one for the anchor and a little slack. + long limit = COMPARISONS + 8; + assertTrue(kind + " made " + reads + " source reads for " + COMPARISONS + + " comparisons against one anchor; expected at most " + limit, reads <= limit); + } + + @Test public void diversityScoreFunctionReadsAnchorOnce() throws Exception { + var file = writeFvecs(); + try (var source = FvecFileVectorValues.open(file)) { + var bsp = BuildScoreProvider.randomAccessScoreProvider(source, VectorSimilarityFunction.EUCLIDEAN); + var sf = bsp.diversityScoreFunctionFor(0); + long before = source.statistics().reads; + for (int node = 1; node <= COMPARISONS; node++) sf.similarityTo(node); + assertAnchorReadOnce("diversityScoreFunctionFor", source.statistics().reads - before); + } finally { + Files.delete(file); + } + } + + @Test public void searchProviderReadsAnchorOnce() throws Exception { + var file = writeFvecs(); + try (var source = FvecFileVectorValues.open(file)) { + var bsp = BuildScoreProvider.randomAccessScoreProvider(source, VectorSimilarityFunction.EUCLIDEAN); + var sf = bsp.searchProviderFor(0).scoreFunction(); + long before = source.statistics().reads; + for (int node = 1; node <= COMPARISONS; node++) sf.similarityTo(node); + assertAnchorReadOnce("searchProviderFor", source.statistics().reads - before); + } finally { + Files.delete(file); + } + } +}