+ * 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 00000000..cc2b623e --- /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