diff --git a/gigamap/gigamap/src/main/java/org/eclipse/store/gigamap/types/GigaIndices.java b/gigamap/gigamap/src/main/java/org/eclipse/store/gigamap/types/GigaIndices.java index 857d7ed7..d6a2ac8f 100644 --- a/gigamap/gigamap/src/main/java/org/eclipse/store/gigamap/types/GigaIndices.java +++ b/gigamap/gigamap/src/main/java/org/eclipse/store/gigamap/types/GigaIndices.java @@ -635,6 +635,7 @@ public final boolean remove(final Class> categoryType) private boolean internalRemoveGroup(final Class categoryType) { + final IndexGroup.Internal found; synchronized(this.parentMap()) { if(this.parent.isReadOnly()) @@ -644,59 +645,120 @@ private boolean internalRemoveGroup(final Class categoryType) ); } - IndexGroup.Internal found = null; + IndexGroup.Internal match = null; // exact class match is checked first, then the general (e.g. interface) type, mirroring #get. for(final IndexGroup.Internal indexGroup : this.indexGroups) { if(indexGroup.getClass() == categoryType) { - found = indexGroup; + match = indexGroup; break; } } - if(found == null) + if(match == null) { for(final IndexGroup.Internal indexGroup : this.indexGroups) { if(categoryType.isAssignableFrom(indexGroup.getClass())) { - found = indexGroup; + match = indexGroup; break; } } } - if(found == null) + if(match == null) { return false; } - if(found instanceof BitmapIndices) + if(match instanceof BitmapIndices) { throw new IllegalArgumentException( "The bitmap index group cannot be removed (it hosts the unique and custom constraints); " + "use BitmapIndices#removeIndex(String) to remove individual bitmap indices." ); } + found = match; + } - // release resources - if(found instanceof Closeable) + // Release resources outside the parentMap monitor: a group's close() may acquire its own internal + // lock (e.g. a vector index's builder write-lock) while a background task holds that lock and waits + // for the parentMap monitor, so closing under the monitor could dead-lock. Drop the group from the + // registry only after close() succeeds, so a failing close leaves it registered (and re-closeable) + // rather than detached-but-unclosed. + if(found instanceof Closeable) + { + try { - try + ((Closeable)found).close(); + } + catch(final IOException e) + { + throw new RuntimeException( + "Failed to close index group \"" + found.getClass().getName() + "\" while removing it.", + e + ); + } + } + + synchronized(this.parentMap()) + { + this.indexGroups.removeOne(found); + this.markStateChangeInstance(); + } + return true; + } + + /** + * Closes every {@link Closeable} index group, releasing background threads and auxiliary + * resources, without removing the groups or mutating persistent state. Invoked when the + * owning storage is shut down (see {@link GigaMap.Default#releaseOnShutdown()}). Best-effort: a + * failure of one group does not prevent the others from being closed; the first failure is + * rethrown (with the rest suppressed) once all groups have been attempted. + *

+ * The closeable groups are collected under the {@code parentMap} monitor but closed outside + * of it: a group's {@code close()} may acquire its own internal lock (e.g. a vector index's builder + * write-lock) while a background task holds that lock and waits for the {@code parentMap} monitor, + * so holding the monitor across {@code close()} would dead-lock. + */ + final void closeAllGroups() + { + final BulkList closeables = BulkList.New(); + synchronized(this.parentMap()) + { + for(final IndexGroup.Internal indexGroup : this.indexGroups) + { + if(indexGroup instanceof Closeable) { - ((Closeable)found).close(); + closeables.add((Closeable)indexGroup); } - catch(final IOException e) + } + } + + RuntimeException problem = null; + for(final Closeable closeable : closeables) + { + try + { + closeable.close(); + } + catch(final Exception e) + { + if(problem == null) { - throw new RuntimeException( - "Failed to close index group \"" + found.getClass().getName() - + "\" while removing it.", + problem = new RuntimeException( + "Failed to close one or more index groups on storage shutdown.", e ); } + else + { + problem.addSuppressed(e); + } } - - this.indexGroups.removeOne(found); - this.markStateChangeInstance(); - return true; + } + if(problem != null) + { + throw problem; } } diff --git a/gigamap/gigamap/src/main/java/org/eclipse/store/gigamap/types/GigaMap.java b/gigamap/gigamap/src/main/java/org/eclipse/store/gigamap/types/GigaMap.java index 743ff363..29a1d3ad 100644 --- a/gigamap/gigamap/src/main/java/org/eclipse/store/gigamap/types/GigaMap.java +++ b/gigamap/gigamap/src/main/java/org/eclipse/store/gigamap/types/GigaMap.java @@ -1421,7 +1421,7 @@ public static GigaMap New( // Unpersistable to force a custom type handler due to required initialization call. public class Default - implements Internal, EntityResolver, Unpersistable, PersistenceCommitListener + implements Internal, EntityResolver, Unpersistable, PersistenceCommitListener, PersistenceShutdownReleasable { static BinaryTypeHandler> provideTypeHandler() { @@ -1615,6 +1615,16 @@ public final GigaIndices index() { return this.indices; } + + @Override + public final void releaseOnShutdown() + { + // Closes Closeable index groups (e.g. vector indices: stops their background threads and + // frees off-heap/auxiliary-disk resources) when the owning storage is shut down, so callers + // need not close indices explicitly. Does not mutate the persistent graph; indices are + // transient and rebuilt on the next storage start. + this.indices.closeAllGroups(); + } @Override public final GigaConstraints constraints() diff --git a/gigamap/jvector/src/main/java/org/eclipse/store/gigamap/jvector/BackgroundTaskManager.java b/gigamap/jvector/src/main/java/org/eclipse/store/gigamap/jvector/BackgroundTaskManager.java index 097dc9f8..cd81d76a 100644 --- a/gigamap/jvector/src/main/java/org/eclipse/store/gigamap/jvector/BackgroundTaskManager.java +++ b/gigamap/jvector/src/main/java/org/eclipse/store/gigamap/jvector/BackgroundTaskManager.java @@ -17,6 +17,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.lang.ref.WeakReference; import java.util.List; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicBoolean; @@ -43,6 +44,14 @@ class BackgroundTaskManager { private static final Logger LOG = LoggerFactory.getLogger(BackgroundTaskManager.class); + /** + * Interval of the liveness watchdog that self-terminates the executor once the index + * (the {@link Callback}) has been garbage-collected. Kept short so an abandoned index's + * daemon thread is reclaimed promptly, independent of the (possibly long) optimization + * and persistence intervals. Each tick is only a weak {@code get()} plus a flag check. + */ + private static final long WATCHDOG_INTERVAL_MS = 1_000L; + // ======================================================================== // Indexing Operations // ======================================================================== @@ -151,9 +160,16 @@ interface Callback // Instance fields // ======================================================================== - private final Callback callback ; - private final String name ; - private final ScheduledExecutorService executor ; + /** + * The index is held weakly: the executor's scheduled tasks (method references on this + * manager) would otherwise strongly pin the index — and through it the whole HNSW graph — for + * as long as the daemon thread lives. Holding it weakly lets an abandoned index (dropped + * without {@code close()}) become collectable; the liveness watchdog then self-terminates the + * executor once the referent is gone. Mirrors {@code EvictionManager.IntervalThread}. + */ + private final WeakReference callbackRef ; + private final String name ; + private final ScheduledExecutorService executor ; // Indexing queue and dedup flag private final ConcurrentLinkedQueue indexingQueue ; @@ -170,6 +186,9 @@ interface Callback private final int persistenceMinChanges ; private ScheduledFuture persistenceTask ; + // Liveness watchdog: self-terminates the executor when the index has been abandoned (GC'd) + private ScheduledFuture watchdogTask ; + private volatile boolean shutdown = false; // ======================================================================== @@ -188,8 +207,8 @@ interface Callback final int persistenceMinChanges ) { - this.callback = callback; - this.name = name ; + this.callbackRef = new WeakReference<>(callback); + this.name = name ; this.executor = Executors.newSingleThreadScheduledExecutor(r -> { @@ -240,6 +259,16 @@ interface Callback { LOG.info("Eventual indexing enabled for index '{}'", name); } + + // Always run the liveness watchdog. In eventual-indexing-only mode there is no recurring + // optimization/persistence task, so without this the idle executor thread would survive + // forever after the index is abandoned. The watchdog guarantees teardown in every mode. + this.watchdogTask = this.executor.scheduleAtFixedRate( + this::checkLiveness, + WATCHDOG_INTERVAL_MS, + WATCHDOG_INTERVAL_MS, + TimeUnit.MILLISECONDS + ); } // ======================================================================== @@ -373,17 +402,8 @@ void shutdown(final boolean drainPending, final boolean optimizePending, final b { this.shutdown = true; - // Cancel scheduled tasks - if(this.optimizationTask != null) - { - this.optimizationTask.cancel(false); - this.optimizationTask = null; - } - if(this.persistenceTask != null) - { - this.persistenceTask.cancel(false); - this.persistenceTask = null; - } + // Cancel scheduled tasks (optimization, persistence, watchdog) + this.cancelScheduledTasks(); // Perform final work if requested if(drainPending || optimizePending || persistPending) @@ -426,6 +446,76 @@ void shutdown(final boolean drainPending, final boolean optimizePending, final b LOG.info("Background task manager shutdown for '{}'", this.name); } + /** + * Cancels the recurring scheduled tasks (optimization, persistence, watchdog) without + * touching the executor itself. Shared by {@link #shutdown} and {@link #selfTerminate}. + */ + private void cancelScheduledTasks() + { + if(this.optimizationTask != null) + { + this.optimizationTask.cancel(false); + this.optimizationTask = null; + } + if(this.persistenceTask != null) + { + this.persistenceTask.cancel(false); + this.persistenceTask = null; + } + if(this.watchdogTask != null) + { + this.watchdogTask.cancel(false); + this.watchdogTask = null; + } + } + + /** + * Resolves the weakly-held index. If it has been garbage-collected — meaning the index was + * abandoned without {@code close()} — this self-terminates the manager so the daemon thread + * and executor are reclaimed, and returns {@code null}. Callers running on the executor + * thread must skip their work when this returns {@code null}. + * + * @return the live {@link Callback}, or {@code null} if the index has been collected + */ + private Callback liveCallback() + { + final Callback cb = this.callbackRef.get(); + if(cb == null && !this.shutdown) + { + this.selfTerminate(); + } + return cb; + } + + /** + * Tears the manager down from within the executor thread after the index was abandoned. + *

+ * Unlike {@link #shutdown}, this must not call {@code awaitTermination} — it runs on the very + * thread being shut down, so awaiting would dead-lock. It only cancels the recurring tasks, + * drops pending work and calls {@link ExecutorService#shutdown()}; the current task then + * returns and the daemon thread ends, making the executor and this manager collectable. + */ + private void selfTerminate() + { + this.shutdown = true; + this.cancelScheduledTasks(); + this.indexingQueue.clear(); + this.executor.shutdown(); // no awaitTermination: we are on the executor thread + LOG.info("Background task manager self-terminated for abandoned index '{}'", this.name); + } + + /** + * Liveness watchdog body. Resolving the weak reference self-terminates the manager when the + * index has been collected (see {@link #liveCallback()}). + */ + private void checkLiveness() + { + if(!this.shutdown) + { + this.liveCallback(); + } + } + // ======================================================================== // Internal methods — all run on the executor thread // ======================================================================== @@ -461,12 +551,18 @@ private void processIndexingBatch() */ private void processAllPendingIndexingOps() { + final Callback cb = this.liveCallback(); + if(cb == null) + { + return; // index abandoned; liveCallback() has already self-terminated the manager + } + IndexingOperation op; while((op = this.indexingQueue.poll()) != null) { try { - op.execute(this.callback); + op.execute(cb); } catch(final Exception e) { @@ -491,6 +587,12 @@ private void runOptimizationIfDirty() return; } + final Callback cb = this.liveCallback(); + if(cb == null) + { + return; // index abandoned; liveCallback() has already self-terminated the manager + } + LOG.debug("Background optimizing index '{}' with {} changes", this.name, this.optimizationChangeCount.get()); @@ -499,7 +601,7 @@ private void runOptimizationIfDirty() // Drain pending indexing ops inline (same thread, no deadlock) this.processAllPendingIndexingOps(); - this.callback.doOptimize(); + cb.doOptimize(); this.optimizationChangeCount.set(0); this.optimizationCount.incrementAndGet(); @@ -528,6 +630,12 @@ private void runPersistenceIfDirty() return; } + final Callback cb = this.liveCallback(); + if(cb == null) + { + return; // index abandoned; liveCallback() has already self-terminated the manager + } + LOG.debug("Background persisting index '{}' with {} changes", this.name, this.persistenceChangeCount.get()); @@ -536,7 +644,7 @@ private void runPersistenceIfDirty() // Drain pending indexing ops inline (same thread, no deadlock) this.processAllPendingIndexingOps(); - this.callback.doPersistToDisk(); + cb.doPersistToDisk(); this.persistenceChangeCount.set(0); @@ -564,13 +672,21 @@ private void finalShutdownWork( this.processAllPendingIndexingOps(); } + // Callback may be null if the index was concurrently collected; skip the graph-touching + // work in that case (nothing to persist/optimize once the index is gone). + final Callback cb = this.callbackRef.get(); + if(cb == null) + { + return; + } + if(optimizePending && this.optimizationChangeCount.get() > 0) { LOG.info("Optimizing pending changes for '{}' before shutdown ({} changes)", this.name, this.optimizationChangeCount.get()); try { - this.callback.doOptimize(); + cb.doOptimize(); this.optimizationChangeCount.set(0); this.optimizationCount.incrementAndGet(); } @@ -586,7 +702,7 @@ private void finalShutdownWork( this.name, this.persistenceChangeCount.get()); try { - this.callback.doPersistToDisk(); + cb.doPersistToDisk(); this.persistenceChangeCount.set(0); } catch(final Exception e) diff --git a/gigamap/jvector/src/main/java/org/eclipse/store/gigamap/jvector/DiskIndexManager.java b/gigamap/jvector/src/main/java/org/eclipse/store/gigamap/jvector/DiskIndexManager.java index 1ab9a232..6fa42392 100644 --- a/gigamap/jvector/src/main/java/org/eclipse/store/gigamap/jvector/DiskIndexManager.java +++ b/gigamap/jvector/src/main/java/org/eclipse/store/gigamap/jvector/DiskIndexManager.java @@ -398,43 +398,44 @@ private void writeIndexWithFusedPQ( new InlineVectors.State(ravv.getVector(nodeId)) ); - // Get a view for FusedPQ state creation - final var view = index.getView(); - suppliers.put(FeatureId.FUSED_PQ, nodeId -> - new FusedPQ.State(view, pqVectors, nodeId) - ); + // Get a view for FusedPQ state creation. Try-with-resources guarantees the view is + // closed even if ordinal mapping or the writer throws (the view is heap-only, so a + // leak on the throwing path would otherwise go unnoticed). + try(final var view = index.getView()) + { + suppliers.put(FeatureId.FUSED_PQ, nodeId -> + new FusedPQ.State(view, pqVectors, nodeId) + ); - // Preserve graph ordinals on disk (identity map, not the default sequentialRenumbering) - // so on-disk node ids stay equal to the source entity ids the integration keys on. - final Map ordinalMap = identityOrdinalMap(index); + // Preserve graph ordinals on disk (identity map, not the default sequentialRenumbering) + // so on-disk node ids stay equal to the source entity ids the integration keys on. + final Map ordinalMap = identityOrdinalMap(index); - if(this.parallelOnDiskWrite) - { - try(final OnDiskParallelGraphIndexWriter writer = new OnDiskParallelGraphIndexWriter.Builder(index, graphPath) - .withParallelDirectBuffers(true) - .withMap(ordinalMap) - .with(inlineVectors) - .with(fusedPQ) - .build()) + if(this.parallelOnDiskWrite) { - writer.write(suppliers); + try(final OnDiskParallelGraphIndexWriter writer = new OnDiskParallelGraphIndexWriter.Builder(index, graphPath) + .withParallelDirectBuffers(true) + .withMap(ordinalMap) + .with(inlineVectors) + .with(fusedPQ) + .build()) + { + writer.write(suppliers); + } } - } - else - { - try(final OnDiskGraphIndexWriter writer = new OnDiskGraphIndexWriter.Builder(index, graphPath) - .withMap(ordinalMap) - .with(inlineVectors) - .with(fusedPQ) - .build()) + else { - writer.write(suppliers); + try(final OnDiskGraphIndexWriter writer = new OnDiskGraphIndexWriter.Builder(index, graphPath) + .withMap(ordinalMap) + .with(inlineVectors) + .with(fusedPQ) + .build()) + { + writer.write(suppliers); + } } } - // Close the view after writing - view.close(); - LOG.info("Wrote index '{}' with FusedPQ compression ({} nodes, parallel={})", this.name, index.size(0), this.parallelOnDiskWrite); } diff --git a/gigamap/jvector/src/main/java/org/eclipse/store/gigamap/jvector/VectorIndices.java b/gigamap/jvector/src/main/java/org/eclipse/store/gigamap/jvector/VectorIndices.java index eba4a1bf..99f62688 100644 --- a/gigamap/jvector/src/main/java/org/eclipse/store/gigamap/jvector/VectorIndices.java +++ b/gigamap/jvector/src/main/java/org/eclipse/store/gigamap/jvector/VectorIndices.java @@ -14,6 +14,7 @@ * #L% */ +import org.eclipse.serializer.collections.BulkList; import org.eclipse.serializer.collections.EqHashTable; import org.eclipse.serializer.collections.types.XGettingTable; import org.eclipse.serializer.collections.types.XIterable; @@ -331,6 +332,7 @@ public final VectorIndex get(final String name) @Override public boolean removeIndex(final String name) { + final VectorIndex.Internal index; synchronized(this.parentMap()) { if(this.parent.isReadOnly()) @@ -339,32 +341,48 @@ public boolean removeIndex(final String name) "Cannot remove vector index \"" + name + "\": the GigaMap is read-only." ); } - final VectorIndex.Internal index = this.vectorIndices.get(name); + index = this.vectorIndices.get(name); if(index == null) { return false; } - index.close(); // release background/disk/searcher resources before dropping the index + } + // Close the index outside the parentMap monitor: index.close() acquires the builder write-lock, + // while a background persist holds that lock and waits for the parentMap monitor — holding the + // monitor across close() would dead-lock. Drop the index from the registry only after close() + // succeeds, so a failing close leaves it registered (and re-closeable) rather than + // detached-but-unclosed. + index.close(); + synchronized(this.parentMap()) + { this.vectorIndices.removeFor(name); this.markStateChangeInstance(); this.parent.internalReportIndexGroupStateChange(this); - return true; } + return true; } /** * Closes all contained vector indices, releasing their native resources. Invoked when the - * whole vector index group is removed via {@link GigaIndices#remove(IndexCategory)}. + * whole vector index group is removed via {@link GigaIndices#remove(IndexCategory)} and on storage + * shutdown. + *

+ * The indices are collected under the {@code parentMap} monitor but closed outside of it: + * {@code index.close()} acquires the builder write-lock, while a background persist holds that lock + * and waits for the {@code parentMap} monitor, so holding the monitor across {@code close()} would + * dead-lock. */ @Override public void close() { + final BulkList> toClose; synchronized(this.parentMap()) { - for(final VectorIndex.Internal index : this.vectorIndices.values()) - { - index.close(); - } + toClose = BulkList.New(this.vectorIndices.values()); + } + for(final VectorIndex.Internal index : toClose) + { + index.close(); } } diff --git a/gigamap/jvector/src/test/java/org/eclipse/store/gigamap/jvector/VectorIndexAbandonmentLeakTest.java b/gigamap/jvector/src/test/java/org/eclipse/store/gigamap/jvector/VectorIndexAbandonmentLeakTest.java new file mode 100644 index 00000000..9ccc6b71 --- /dev/null +++ b/gigamap/jvector/src/test/java/org/eclipse/store/gigamap/jvector/VectorIndexAbandonmentLeakTest.java @@ -0,0 +1,306 @@ +package org.eclipse.store.gigamap.jvector; + +/*- + * #%L + * EclipseStore GigaMap JVector + * %% + * Copyright (C) 2023 - 2026 MicroStream Software + * %% + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + * #L% + */ + +import org.eclipse.store.gigamap.types.GigaMap; +import org.eclipse.store.storage.embedded.types.EmbeddedStorage; +import org.eclipse.store.storage.embedded.types.EmbeddedStorageManager; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +import java.lang.ref.WeakReference; +import java.nio.file.Path; +import java.time.Duration; +import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; + +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Regression tests for the JVector background-index lifecycle: a background-enabled {@link VectorIndex} + * that is abandoned without an explicit {@code close()} must not pin its object graph nor leak its daemon + * thread, and closing indices on storage shutdown or removal must not dead-lock. + *

+ * The manager holds the index through a {@link WeakReference} and a liveness watchdog self-terminates + * the executor once the index has been collected, so a dropped index becomes fully reclaimable + * (thread + graph) even with no storage in play. The storage-backed case additionally verifies that + * shutting down the {@link EmbeddedStorageManager} closes the index groups without an explicit + * {@code close()}. + */ +@Tag("slow") +class VectorIndexAbandonmentLeakTest +{ + private static final String BG_THREAD_PREFIX = "VectorIndex-Background-"; + + static final class Doc + { + final float[] embedding; + + Doc() + { + this.embedding = null; // for deserialization + } + + Doc(final float[] embedding) + { + this.embedding = embedding; + } + } + + static class EmbeddingVectorizer extends Vectorizer + { + @Override + public float[] vectorize(final Doc entity) + { + return entity.embedding; + } + + @Override + public boolean isEmbedded() + { + return true; + } + } + + private static VectorIndexConfiguration plainConfig() + { + return VectorIndexConfiguration.builder() + .dimension(3) + .similarityFunction(VectorSimilarityFunction.COSINE) + .build(); + } + + private static VectorIndexConfiguration backgroundConfig() + { + return VectorIndexConfiguration.builder() + .dimension(3) + .similarityFunction(VectorSimilarityFunction.COSINE) + .eventualIndexing(true) + .optimizationIntervalMs(50L) + .minChangesBetweenOptimizations(1) + .build(); + } + + // On-disk index + background persistence: index.close() runs doPersistToDisk on the background thread, + // which is what turns "close under the parentMap monitor" into a deadlock. + private static VectorIndexConfiguration onDiskBackgroundConfig(final Path indexDir) + { + return VectorIndexConfiguration.builder() + .dimension(3) + .similarityFunction(VectorSimilarityFunction.COSINE) + .onDisk(true) + .indexDirectory(indexDir) + .persistenceIntervalMs(600_000L) // background persistence enabled; interval too long to tick in-test + .build(); + } + + private static void addDocs(final GigaMap map) + { + map.add(new Doc(new float[]{1.0f, 0.0f, 0.0f})); + map.add(new Doc(new float[]{0.0f, 1.0f, 0.0f})); + } + + private static int backgroundThreadCount() + { + return (int)Thread.getAllStackTraces().keySet().stream() + .filter(t -> t.getName().startsWith(BG_THREAD_PREFIX)) + .count(); + } + + /** + * Builds a background-enabled vector index on a fresh in-memory map, forces it to initialize, then + * lets every strong reference (map, group, index) go out of scope on return. Only a + * {@link WeakReference} to the index survives. + */ + private static WeakReference buildAndAbandon(final VectorIndexConfiguration config) + { + final GigaMap map = GigaMap.New(); + final VectorIndices vi = map.index().register(VectorIndices.Category()); + vi.add("emb", config, new EmbeddingVectorizer()); + addDocs(map); + vi.get("emb").search(new float[]{1.0f, 0.0f, 0.0f}, 2); // force lazy initialization + return new WeakReference<>(vi.get("emb")); + } + + /** + * Repeatedly triggers GC and waits until {@code condition} holds or the timeout elapses. + * + * @return {@code true} if the condition became true within the timeout + */ + private static boolean awaitGc(final BooleanSupplier condition, final long timeoutMs) throws InterruptedException + { + final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMs); + while(System.nanoTime() < deadline) + { + System.gc(); + if(condition.getAsBoolean()) + { + return true; + } + Thread.sleep(100L); + } + return condition.getAsBoolean(); + } + + @Test + @Timeout(value = 120, unit = TimeUnit.SECONDS) + void abandonedBackgroundIndexIsCollectable() throws Exception + { + final WeakReference backgroundRef = buildAndAbandon(backgroundConfig()); + final WeakReference plainRef = buildAndAbandon(plainConfig()); + + assertTrue(awaitGc(() -> plainRef.get() == null, 30_000L), + "control: a vector index with no background feature must be collectable once abandoned"); + assertTrue(awaitGc(() -> backgroundRef.get() == null, 30_000L), + "abandoned background-enabled vector index must not be pinned by its BackgroundTaskManager tasks"); + assertNull(backgroundRef.get()); + } + + @Test + @Timeout(value = 120, unit = TimeUnit.SECONDS) + void abandonedVectorIndicesReleaseBackgroundThreads() throws Exception + { + final int n = 20; + final int before = backgroundThreadCount(); + + for(int i = 0; i < n; i++) + { + buildAndAbandon(backgroundConfig()); + } + + // The liveness watchdog self-terminates each abandoned index's executor once it is collected. + final boolean released = awaitGc(() -> backgroundThreadCount() <= before, 60_000L); + + assertTrue(released, + "abandoning " + n + " background-enabled vector indices without close() leaked " + + (backgroundThreadCount() - before) + " '" + BG_THREAD_PREFIX + "' threads; the " + + "BackgroundTaskManager must self-terminate when its index is garbage-collected"); + } + + @Test + @Timeout(value = 120, unit = TimeUnit.SECONDS) + void indicesClosedOnStorageShutdown(@TempDir final Path dir) throws Exception + { + final int before = backgroundThreadCount(); + + final EmbeddedStorageManager storage = EmbeddedStorage.start(dir); + final GigaMap map = GigaMap.New(); + storage.setRoot(map); + final VectorIndices vi = map.index().register(VectorIndices.Category()); + vi.add("emb", backgroundConfig(), new EmbeddingVectorizer()); + addDocs(map); + vi.get("emb").search(new float[]{1.0f, 0.0f, 0.0f}, 2); // force init + spawn the background thread + storage.storeRoot(); + + assertTrue(backgroundThreadCount() > before, "precondition: background thread must be running"); + + // Shut down storage WITHOUT ever calling index.close(); the shutdown hook must close the groups. + storage.shutdown(); + + assertTrue(backgroundThreadCount() <= before, + "EmbeddedStorageManager.shutdown() must close vector index groups so no '" + + BG_THREAD_PREFIX + "' thread survives"); + } + + @Test + @Timeout(value = 120, unit = TimeUnit.SECONDS) + void storageShutdownDoesNotDeadlockWithOnDiskBackgroundPersist(@TempDir final Path dir) throws Exception + { + final int before = backgroundThreadCount(); + + // On-disk background index with unsaved changes and the default persistOnShutdown=true: on shutdown + // the background thread runs doPersistToDisk, which takes the builder write-lock and then the + // parentMap monitor. If the close path still held the monitor across index.close() (which also wants + // the write-lock), shutdown would dead-lock permanently. + final EmbeddedStorageManager storage = EmbeddedStorage.start(dir.resolve("storage")); + final GigaMap map = GigaMap.New(); + storage.setRoot(map); + final VectorIndices vi = map.index().register(VectorIndices.Category()); + vi.add("emb", onDiskBackgroundConfig(dir.resolve("index")), new EmbeddingVectorizer()); + addDocs(map); // leaves unsaved (dirty) index changes + vi.get("emb").search(new float[]{1.0f, 0.0f, 0.0f}, 2); // force init + spawn the background thread + storage.storeRoot(); + + assertTrue(backgroundThreadCount() > before, "precondition: background thread must be running"); + + // Preemptive timeout: a hard monitor deadlock cannot be interrupted, so run shutdown on a separate + // thread that is abandoned on timeout, turning a regression into a clean failure instead of a hang. + assertTimeoutPreemptively(Duration.ofSeconds(30), () -> { storage.shutdown(); }, + "storage.shutdown() dead-locked closing the on-disk background index"); + + assertTrue(backgroundThreadCount() <= before, + "shutdown must close the on-disk background index without dead-locking"); + } + + @Test + @Timeout(value = 120, unit = TimeUnit.SECONDS) + void removeIndexDoesNotDeadlockWithOnDiskBackgroundPersist(@TempDir final Path dir) throws Exception + { + final int before = backgroundThreadCount(); + + try(final EmbeddedStorageManager storage = EmbeddedStorage.start(dir.resolve("storage"))) + { + final GigaMap map = GigaMap.New(); + storage.setRoot(map); + final VectorIndices vi = map.index().register(VectorIndices.Category()); + vi.add("emb", onDiskBackgroundConfig(dir.resolve("index")), new EmbeddingVectorizer()); + addDocs(map); + vi.get("emb").search(new float[]{1.0f, 0.0f, 0.0f}, 2); // force init + spawn the background thread + storage.storeRoot(); + + assertTrue(backgroundThreadCount() > before, "precondition: background thread must be running"); + + // removeIndex() closes the index; it must not hold the parentMap monitor across close(). + assertTimeoutPreemptively(Duration.ofSeconds(30), () -> { vi.removeIndex("emb"); }, + "removeIndex() dead-locked closing the on-disk background index"); + + assertNull(vi.get("emb"), "index must be removed"); + assertTrue(backgroundThreadCount() <= before, "removeIndex must close the background index"); + } + } + + @Test + @Timeout(value = 120, unit = TimeUnit.SECONDS) + void removeGroupDoesNotDeadlockWithOnDiskBackgroundPersist(@TempDir final Path dir) throws Exception + { + final int before = backgroundThreadCount(); + + try(final EmbeddedStorageManager storage = EmbeddedStorage.start(dir.resolve("storage"))) + { + final GigaMap map = GigaMap.New(); + storage.setRoot(map); + final VectorIndices vi = map.index().register(VectorIndices.Category()); + vi.add("emb", onDiskBackgroundConfig(dir.resolve("index")), new EmbeddingVectorizer()); + addDocs(map); + vi.get("emb").search(new float[]{1.0f, 0.0f, 0.0f}, 2); // force init + spawn the background thread + storage.storeRoot(); + + assertTrue(backgroundThreadCount() > before, "precondition: background thread must be running"); + + // Whole-group removal closes each contained index; it must not hold the parentMap monitor across + // close() (internalRemoveGroup must detach under the monitor but close outside it). + assertTimeoutPreemptively(Duration.ofSeconds(30), + () -> { map.index().remove(VectorIndices.Category()); }, + "index group removal dead-locked closing the on-disk background index"); + + assertNull(map.index().get(VectorIndices.Category()), "group must be removed"); + assertTrue(backgroundThreadCount() <= before, "group removal must close the background index"); + } + } +} diff --git a/storage/embedded/src/main/java/org/eclipse/store/storage/embedded/types/EmbeddedStorageManager.java b/storage/embedded/src/main/java/org/eclipse/store/storage/embedded/types/EmbeddedStorageManager.java index 3c40a43f..4b475bc5 100644 --- a/storage/embedded/src/main/java/org/eclipse/store/storage/embedded/types/EmbeddedStorageManager.java +++ b/storage/embedded/src/main/java/org/eclipse/store/storage/embedded/types/EmbeddedStorageManager.java @@ -19,6 +19,7 @@ import java.nio.ByteBuffer; import java.nio.file.Path; import java.util.Arrays; +import java.util.ArrayList; import java.util.List; import java.util.function.Predicate; @@ -30,12 +31,14 @@ import org.eclipse.serializer.monitoring.MonitoringManager; import org.eclipse.serializer.persistence.binary.types.Binary; import org.eclipse.serializer.persistence.types.Persistence; +import org.eclipse.serializer.persistence.types.PersistenceAcceptor; import org.eclipse.serializer.persistence.types.PersistenceManager; import org.eclipse.serializer.persistence.types.PersistenceObjectRegistry; import org.eclipse.serializer.persistence.types.PersistenceRootReference; import org.eclipse.serializer.persistence.types.PersistenceRoots; import org.eclipse.serializer.persistence.types.PersistenceRootsProvider; import org.eclipse.serializer.persistence.types.PersistenceRootsView; +import org.eclipse.serializer.persistence.types.PersistenceShutdownReleasable; import org.eclipse.serializer.persistence.types.PersistenceTypeDictionaryExporter; import org.eclipse.serializer.persistence.types.Storer; import org.eclipse.serializer.persistence.types.Unpersistable; @@ -464,6 +467,10 @@ private void initialize() @Override public final synchronized boolean shutdown() { + // Release shutdown-aware instances (e.g. GigaMaps closing their vector index groups) while + // the object registry is still populated and the storage system is still up. + this.releaseShutdownParticipants(); + LazyReferenceManager.get().removeController(this); final boolean result = this.storageSystem.shutdown(); if(!result) @@ -480,6 +487,49 @@ public final synchronized boolean shutdown() return true; } + /** + * Invokes {@link PersistenceShutdownReleasable#releaseOnShutdown()} on every registered instance + * that opts into shutdown participation. Enumerates the object registry, which holds only + * already-materialized instances, so no {@code Lazy} subgraph is force-loaded. Best-effort: a + * failing participant is logged and skipped so it cannot abort shutdown of the rest or of storage. + *

+ * Participants are collected during the registry iteration and released afterwards, outside of it: + * {@code releaseOnShutdown()} may block on application locks and await background-task termination, + * which must not happen while the object registry is being iterated. + */ + private void releaseShutdownParticipants() + { + final List participants = new ArrayList<>(); + try + { + final PersistenceAcceptor collector = (objectId, instance) -> + { + if(instance instanceof PersistenceShutdownReleasable) + { + participants.add((PersistenceShutdownReleasable)instance); + } + }; + this.connectionFoundation.getObjectRegistry().iterateEntries(collector); + } + catch(final Throwable t) + { + logger.error("Error while collecting storage shutdown participants", t); + } + + for(final PersistenceShutdownReleasable participant : participants) + { + try + { + participant.releaseOnShutdown(); + } + catch(final Throwable t) + { + logger.error("Error releasing shutdown participant of type {}", + participant.getClass().getName(), t); + } + } + } + @Override public final boolean isAcceptingTasks() {