From e92f972b52dacb43a84e7000f2cb654a382b180d Mon Sep 17 00:00:00 2001 From: Brett Wooldridge Date: Fri, 4 Sep 2026 08:18:49 +0900 Subject: [PATCH] fix: keep the collection state check outside the factory-wide lock CollectionFactory.getCollection holds one lock for the whole factory and, while holding it, calls isDropped() and isOpen() on the registered collection. Both take that collection's read lock. While a long write holds a collection's write lock (an index rebuild, a large remove(filter), an update on a big document) the one caller asking for that collection blocks inside the factory lock, and from then on every getCollection call for every other collection queues behind it. Observed on a production system: one thread rebuilding an index inside update() for over three hours, 349 other threads parked in CollectionFactory.getCollection on the factory lock, most of them wanting unrelated collections. The registry is now read under the factory read lock, the usability check runs with no factory lock held, and the factory write lock is taken only to create or replace an entry, with a re-check for a concurrent replacement. Callers of the busy collection still wait on it, as they should; callers of other collections no longer wait at all. CollectionFactoryConvoyTest parks a remove(filter) inside its filter under the collection's write lock through the public API, confirms that a getCollection for that collection waits, and asserts that getCollection for a registered and for a brand new collection both return promptly. It fails on main with a TimeoutException and passes with this change. --- .../no2/collection/CollectionFactory.java | 45 ++++++-- .../CollectionFactoryConvoyTest.java | 105 ++++++++++++++++++ 2 files changed, 141 insertions(+), 9 deletions(-) create mode 100644 nitrite/src/test/java/org/dizitart/no2/collection/CollectionFactoryConvoyTest.java diff --git a/nitrite/src/main/java/org/dizitart/no2/collection/CollectionFactory.java b/nitrite/src/main/java/org/dizitart/no2/collection/CollectionFactory.java index e573a7bcb..0fd274f7a 100644 --- a/nitrite/src/main/java/org/dizitart/no2/collection/CollectionFactory.java +++ b/nitrite/src/main/java/org/dizitart/no2/collection/CollectionFactory.java @@ -52,6 +52,14 @@ public CollectionFactory(LockService lockService) { /** * Gets or creates a collection. + *

+ * The factory-wide lock is held only while the registry is read or changed. Whether an + * already registered collection is still usable is decided outside it, because + * {@link NitriteCollection#isDropped()} and {@link NitriteCollection#isOpen()} take that + * collection's own read lock: while a long write (an index rebuild, a large + * {@code remove(filter)}) holds the collection's write lock, a caller asking for that + * collection has to wait for it, but with the factory lock held across the wait every + * caller asking for any collection was queued behind it as well. * * @param name the name * @param nitriteConfig the nitrite config @@ -62,24 +70,43 @@ public NitriteCollection getCollection(String name, NitriteConfig nitriteConfig, notNull(nitriteConfig, "Configuration is null while creating collection"); notEmpty(name, "Collection name is null or empty"); + NitriteCollection registered = getRegistered(name); + if (registered != null && isUsable(registered)) { + return registered; + } + Lock lock = lockService.getWriteLock(this.getClass().getName()); try { lock.lock(); - if (collectionMap.containsKey(name)) { - NitriteCollection collection = collectionMap.get(name); - if (collection.isDropped() || !collection.isOpen()) { - collectionMap.remove(name); - return createCollection(name, nitriteConfig, writeCatalogue); - } - return collectionMap.get(name); - } else { - return createCollection(name, nitriteConfig, writeCatalogue); + NitriteCollection current = collectionMap.get(name); + if (current != null && current != registered && isUsable(current)) { + // another caller replaced it while this one was checking the old instance + return current; } + + if (current != null) { + collectionMap.remove(name); + } + return createCollection(name, nitriteConfig, writeCatalogue); + } finally { + lock.unlock(); + } + } + + private NitriteCollection getRegistered(String name) { + Lock lock = lockService.getReadLock(this.getClass().getName()); + try { + lock.lock(); + return collectionMap.get(name); } finally { lock.unlock(); } } + private static boolean isUsable(NitriteCollection collection) { + return !collection.isDropped() && collection.isOpen(); + } + private NitriteCollection createCollection(String name, NitriteConfig nitriteConfig, boolean writeCatalog) { NitriteStore store = nitriteConfig.getNitriteStore(); diff --git a/nitrite/src/test/java/org/dizitart/no2/collection/CollectionFactoryConvoyTest.java b/nitrite/src/test/java/org/dizitart/no2/collection/CollectionFactoryConvoyTest.java new file mode 100644 index 000000000..cc2b623e1 --- /dev/null +++ b/nitrite/src/test/java/org/dizitart/no2/collection/CollectionFactoryConvoyTest.java @@ -0,0 +1,105 @@ +/* + * Copyright (c) 2017-2020. Nitrite author or authors. + * + * 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 org.dizitart.no2.collection; + +import org.dizitart.no2.Nitrite; +import org.dizitart.no2.filters.Filter; +import org.dizitart.no2.integration.Retry; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +/** + * A long write on one collection must not stall {@code getCollection} for every other + * collection. + *

+ * {@link CollectionFactory#getCollection} decides whether a registered collection is still + * usable by calling {@code isDropped()} and {@code isOpen()}, which take that collection's + * read lock. While a write holds the collection's write lock, a caller asking for that + * collection waits on it, which is expected. The factory used to make that check while + * holding its own factory-wide lock, so that one waiting caller also blocked every caller + * asking for any other collection until the write finished. + */ +public class CollectionFactoryConvoyTest { + @Rule + public Retry retry = new Retry(3); + + private Nitrite db; + private ExecutorService executor; + + @Before + public void setUp() { + db = Nitrite.builder().openOrCreate(); + executor = Executors.newCachedThreadPool(); + } + + @After + public void tearDown() { + executor.shutdownNow(); + db.close(); + } + + @Test + public void testWriteOnOneCollectionDoesNotBlockGetCollectionOfAnother() throws Exception { + NitriteCollection busy = db.getCollection("busy"); + busy.insert(Document.createDocument("key", 1)); + NitriteCollection other = db.getCollection("other"); + + // a remove(filter) evaluates the filter under the collection's write lock; park it there + CountDownLatch inFilter = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + Filter parked = element -> { + inFilter.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return false; + }; + Future writer = executor.submit(() -> busy.remove(parked)); + assertTrue("writer never entered the filter", inFilter.await(5, TimeUnit.SECONDS)); + + // a caller asking for the busy collection has to wait for that write; that is expected + Future waitingForBusy = executor.submit(() -> db.getCollection("busy")); + Thread.sleep(200); + assertTrue("caller for the busy collection should be waiting", !waitingForBusy.isDone()); + + // ... but callers asking for other collections, registered or new, must not queue behind it + Future registered = executor.submit(() -> db.getCollection("other")); + Future created = executor.submit(() -> db.getCollection("fresh")); + try { + assertSame(other, registered.get(2, TimeUnit.SECONDS)); + assertTrue(created.get(2, TimeUnit.SECONDS).isOpen()); + } finally { + release.countDown(); + } + + writer.get(5, TimeUnit.SECONDS); + assertSame(busy, waitingForBusy.get(5, TimeUnit.SECONDS)); + } +}