From 01c138ed1101f826707719af82123e9472a6a6c4 Mon Sep 17 00:00:00 2001 From: Jeremiah Jordan Date: Thu, 3 Sep 2026 10:40:44 -0500 Subject: [PATCH 1/4] Add test for concurrent SAI vector inserts exceeding jvector's per-graph pool limit patch by Jeremiah Jordan; reviewed by XXX for CASSANDRA-21644 --- .../sai/memory/VectorMemoryIndexTest.java | 26 ++++++++++++++----- 1 file changed, 19 insertions(+), 7 deletions(-) diff --git a/test/unit/org/apache/cassandra/index/sai/memory/VectorMemoryIndexTest.java b/test/unit/org/apache/cassandra/index/sai/memory/VectorMemoryIndexTest.java index 3aa10bdd6b28..d730553c17b9 100644 --- a/test/unit/org/apache/cassandra/index/sai/memory/VectorMemoryIndexTest.java +++ b/test/unit/org/apache/cassandra/index/sai/memory/VectorMemoryIndexTest.java @@ -74,6 +74,7 @@ import org.apache.cassandra.index.sai.QueryContext; import org.apache.cassandra.index.sai.SAITester; import org.apache.cassandra.index.sai.StorageAttachedIndex; +import org.apache.cassandra.index.sai.disk.v1.vector.OnHeapGraph; import org.apache.cassandra.index.sai.disk.v1.vector.PrimaryKeyWithScore; import org.apache.cassandra.index.sai.plan.Expression; import org.apache.cassandra.index.sai.utils.PrimaryKey; @@ -102,6 +103,7 @@ public class VectorMemoryIndexTest extends SAITester private static final double RECALL_THRESHOLD = 0.9; private static final int VECTORS_PER_THREAD = 2000; + private static final int VECTORS_PER_THREAD_BEYOND_POOL_CAP = 500; // more writers, so fewer vectors each private ColumnFamilyStore cfs; private StorageAttachedIndex index; @@ -242,13 +244,24 @@ public void testExpectedNodesVisitedRespectsBounds() @Test public void testConcurrentAddsWithRandomVectors() throws Exception { - testConcurrentAddsAreEventuallyConsistent((threadId, i) -> randomVectorFromThreadLocal()); + testConcurrentAddsAreEventuallyConsistent(Runtime.getRuntime().availableProcessors(), VECTORS_PER_THREAD, (threadId, i) -> randomVectorFromThreadLocal()); } @Test public void testConcurrentAddsWithSharedVectors() throws Exception { - testConcurrentAddsAreEventuallyConsistent((threadId, i) -> makeSharedVector(i)); + testConcurrentAddsAreEventuallyConsistent(Runtime.getRuntime().availableProcessors(), VECTORS_PER_THREAD, (threadId, i) -> makeSharedVector(i)); + } + + /** + * More writers than jvector's GraphIndexBuilder can serve at once must wait, not fail the insert. The other + * concurrent tests use exactly availableProcessors writers, which never exceeds jvector's limit. + */ + @Test + public void testConcurrentAddsExceedingJVectorPoolCap() throws Exception + { + int numThreads = 2 * OnHeapGraph.MAX_CONCURRENT_GRAPH_INSERTS; + testConcurrentAddsAreEventuallyConsistent(numThreads, VECTORS_PER_THREAD_BEYOND_POOL_CAP, (threadId, i) -> randomVectorFromThreadLocal()); } /** @@ -262,13 +275,12 @@ public void testConcurrentAddsWithSharedVectors() throws Exception * After all writers complete, a full-ring search must return the vast majority of * inserted keys with valid scores, confirming no data was lost or corrupted. */ - private void testConcurrentAddsAreEventuallyConsistent(BiFunction vectorFactory) throws Exception + private void testConcurrentAddsAreEventuallyConsistent(int numThreads, int vectorsPerThread, BiFunction vectorFactory) throws Exception { Memtable memtable = Mockito.mock(Memtable.class); memtableIndex = new VectorMemoryIndex(index, memtable); - int numThreads = Runtime.getRuntime().availableProcessors(); - int totalInserted = numThreads * VECTORS_PER_THREAD; + int totalInserted = numThreads * vectorsPerThread; ExecutorService executor = Executors.newFixedThreadPool(numThreads); @@ -284,9 +296,9 @@ private void testConcurrentAddsAreEventuallyConsistent(BiFunction Date: Thu, 3 Sep 2026 10:40:44 -0500 Subject: [PATCH 2/4] Fix concurrent SAI vector inserts failing once jvector's per-graph pool limit is exceeded patch by Jeremiah Jordan; reviewed by XXX for CASSANDRA-21644 --- CHANGES.txt | 1 + .../index/sai/disk/v1/vector/OnHeapGraph.java | 22 ++++++++++++++++++- 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/CHANGES.txt b/CHANGES.txt index 800547a91bfd..39a3a3a69eef 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.0.10 + * Fix concurrent SAI vector inserts failing once jvector's per-graph pool limit is exceeded (CASSANDRA-21644) * Unwrap LongType properly when calculating min/max terms in V1SSTableIndex (CASSANDRA-21635) * Force repair should ignore min_repair_interval (CASSANDRA-21552) * Render SubnetGroups as JSON in system_views.settings (CASSANDRA-21579) diff --git a/src/java/org/apache/cassandra/index/sai/disk/v1/vector/OnHeapGraph.java b/src/java/org/apache/cassandra/index/sai/disk/v1/vector/OnHeapGraph.java index a823943a4e77..0afbc6917871 100644 --- a/src/java/org/apache/cassandra/index/sai/disk/v1/vector/OnHeapGraph.java +++ b/src/java/org/apache/cassandra/index/sai/disk/v1/vector/OnHeapGraph.java @@ -31,6 +31,8 @@ import java.util.function.Function; import java.util.stream.IntStream; +import com.google.common.annotations.VisibleForTesting; + import org.cliffc.high_scale_lib.NonBlockingHashMap; import org.cliffc.high_scale_lib.NonBlockingHashMapLong; import org.slf4j.Logger; @@ -63,6 +65,7 @@ import org.apache.cassandra.io.util.SequentialWriter; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.CloseableIterator; +import org.apache.cassandra.utils.concurrent.Semaphore; import org.apache.lucene.util.StringHelper; public class OnHeapGraph @@ -71,6 +74,13 @@ public class OnHeapGraph public static final int MIN_PQ_ROWS = 1024; + /** + * jvector's {@link GraphIndexBuilder#addGraphNode} throws if more than this many threads are inside it at once + * (see {@code PoolingSupport}). Uses {@link Runtime} rather than {@code DatabaseDescriptor} because that is what jvector reads. + */ + @VisibleForTesting + public static final int MAX_CONCURRENT_GRAPH_INSERTS = Runtime.getRuntime().availableProcessors() + 1; + private final RamAwareVectorValues vectorValues; private final GraphIndexBuilder builder; private final VectorType vectorType; @@ -79,6 +89,8 @@ public class OnHeapGraph private final NonBlockingHashMapLong> postingsByOrdinal; private final NonBlockingHashMap vectorsByKey; private final AtomicInteger nextOrdinal = new AtomicInteger(); + // memtable inserts are concurrent (CASSANDRA-21160); excess writers wait here rather than fail in jvector + private final Semaphore graphInsertPermits = Semaphore.newSemaphore(MAX_CONCURRENT_GRAPH_INSERTS); private volatile boolean hasDeletions; private String source; @@ -187,7 +199,15 @@ public long add(ByteBuffer term, T key, InvalidVectorBehavior behavior) : ((CompactionVectorValues) vectorValues).add(ordinal, term); bytesUsed += VectorPostings.emptyBytesUsed() + VectorPostings.bytesPerPosting(); postingsByOrdinal.put(ordinal, postings); - bytesUsed += builder.addGraphNode(ordinal, vectorValues); + graphInsertPermits.acquireThrowUncheckedOnInterrupt(1); + try + { + bytesUsed += builder.addGraphNode(ordinal, vectorValues); + } + finally + { + graphInsertPermits.release(1); + } return bytesUsed; } else From 0b11c79961578a62caf7f2e2d1432ac2b6c57dde Mon Sep 17 00:00:00 2001 From: Jeremiah Jordan Date: Thu, 3 Sep 2026 11:00:34 -0500 Subject: [PATCH 3/4] Skip the graph insert semaphore when the memtable already limits concurrent writers Review update: TrieMemtable and the locking ShardedSkipListMemtable serialize updates per shard, so when they have no more shards than jvector's per-builder limit there is nothing for the semaphore to bound. Add Memtable.limitsConcurrentWritesTo(int), false by default, and let OnHeapGraph leave the semaphore out when it is true. patch by Jeremiah Jordan; reviewed by XXX for CASSANDRA-21644 --- .../cassandra/db/memtable/Memtable.java | 11 ++ .../cassandra/db/memtable/Memtable_API.md | 4 + .../db/memtable/ShardedSkipListMemtable.java | 5 + .../cassandra/db/memtable/TrieMemtable.java | 7 ++ .../index/sai/disk/v1/vector/OnHeapGraph.java | 17 ++- .../MemtableConcurrentWriteLimitTest.java | 102 ++++++++++++++++++ 6 files changed, 142 insertions(+), 4 deletions(-) create mode 100644 test/unit/org/apache/cassandra/db/memtable/MemtableConcurrentWriteLimitTest.java diff --git a/src/java/org/apache/cassandra/db/memtable/Memtable.java b/src/java/org/apache/cassandra/db/memtable/Memtable.java index d4f0dedb995e..dac1cd5ccd8e 100644 --- a/src/java/org/apache/cassandra/db/memtable/Memtable.java +++ b/src/java/org/apache/cassandra/db/memtable/Memtable.java @@ -196,6 +196,17 @@ interface Owner */ long put(PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group opGroup); + /** + * Whether this memtable guarantees that no more than {@code maxWriters} threads are inside {@link #put} at once, + * and therefore that no more than that many threads concurrently apply the update's effect to the {@code indexer}. + * Memtables that serialize writes per shard can return true when they have at most {@code maxWriters} shards. + * Indexes whose implementation has a limit on concurrent writers can skip their own bounding when this is true. + */ + default boolean limitsConcurrentWritesTo(int maxWriters) + { + return false; + } + // Read operations are provided by the UnfilteredSource interface. // Statistics diff --git a/src/java/org/apache/cassandra/db/memtable/Memtable_API.md b/src/java/org/apache/cassandra/db/memtable/Memtable_API.md index 70f8f0b6053a..8e62b2f5bdbf 100644 --- a/src/java/org/apache/cassandra/db/memtable/Memtable_API.md +++ b/src/java/org/apache/cassandra/db/memtable/Memtable_API.md @@ -135,6 +135,10 @@ implementations) are provided as the `AbstractMemtable` (statistics tracking), ` commit log span tracking) and `AbstractAllocatorMemtable` (adds memory management via the `Allocator` class, together with flush triggering on memory use and time interval expiration). +A memtable that serializes writes, for example per shard, can say so through `limitsConcurrentWritesTo`. Secondary +indexes whose implementation limits the number of concurrent writers use this to skip their own bounding when the +memtable already provides it. The default answer is false. + The memtable API also gives the memtable some control over flushing and the functioning of the commit log. The former is there to permit memtables that operate long-term and/or can handle some events internally, without a need to flush. The latter enables memtables that have an internal durability mechanism, such as ones using persistent memory or a diff --git a/src/java/org/apache/cassandra/db/memtable/ShardedSkipListMemtable.java b/src/java/org/apache/cassandra/db/memtable/ShardedSkipListMemtable.java index 92cdbbad9fe0..955e6b0d5294 100644 --- a/src/java/org/apache/cassandra/db/memtable/ShardedSkipListMemtable.java +++ b/src/java/org/apache/cassandra/db/memtable/ShardedSkipListMemtable.java @@ -494,6 +494,11 @@ public long put(PartitionUpdate update, UpdateTransaction indexer, OpOrder.Group } } + @Override + public boolean limitsConcurrentWritesTo(int maxWriters) + { + return boundaries.shardCount() <= maxWriters; + } } public static Factory factory(Map optionsCopy) diff --git a/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java b/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java index 83b02db06a0c..f652c4675f1d 100644 --- a/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java +++ b/src/java/org/apache/cassandra/db/memtable/TrieMemtable.java @@ -174,6 +174,13 @@ public void discard() } } + @Override + public boolean limitsConcurrentWritesTo(int maxWriters) + { + // each shard applies one update at a time under its write lock + return boundaries.shardCount() <= maxWriters; + } + /** * Should only be called by ColumnFamilyStore.apply via Keyspace.apply, which supplies the appropriate * OpOrdering. diff --git a/src/java/org/apache/cassandra/index/sai/disk/v1/vector/OnHeapGraph.java b/src/java/org/apache/cassandra/index/sai/disk/v1/vector/OnHeapGraph.java index 0afbc6917871..62b7c4dfa41d 100644 --- a/src/java/org/apache/cassandra/index/sai/disk/v1/vector/OnHeapGraph.java +++ b/src/java/org/apache/cassandra/index/sai/disk/v1/vector/OnHeapGraph.java @@ -31,6 +31,8 @@ import java.util.function.Function; import java.util.stream.IntStream; +import javax.annotation.Nullable; + import com.google.common.annotations.VisibleForTesting; import org.cliffc.high_scale_lib.NonBlockingHashMap; @@ -89,8 +91,10 @@ public class OnHeapGraph private final NonBlockingHashMapLong> postingsByOrdinal; private final NonBlockingHashMap vectorsByKey; private final AtomicInteger nextOrdinal = new AtomicInteger(); - // memtable inserts are concurrent (CASSANDRA-21160); excess writers wait here rather than fail in jvector - private final Semaphore graphInsertPermits = Semaphore.newSemaphore(MAX_CONCURRENT_GRAPH_INSERTS); + // memtable inserts are concurrent (CASSANDRA-21160); excess writers wait here rather than fail in jvector. + // Null when the memtable itself already limits writers to that many, e.g. a TrieMemtable with few enough shards. + @Nullable + private final Semaphore graphInsertPermits; private volatile boolean hasDeletions; private String source; @@ -119,6 +123,9 @@ public OnHeapGraph(AbstractType termComparator, IndexWriterConfig indexWriter postingsMap = new ConcurrentSkipListMap<>(Arrays::compare); postingsByOrdinal = new NonBlockingHashMapLong<>(); vectorsByKey = memtable != null ? new NonBlockingHashMap<>() : null; + graphInsertPermits = memtable != null && memtable.limitsConcurrentWritesTo(MAX_CONCURRENT_GRAPH_INSERTS) + ? null + : Semaphore.newSemaphore(MAX_CONCURRENT_GRAPH_INSERTS); builder = new GraphIndexBuilder<>(vectorValues, VectorEncoding.FLOAT32, @@ -199,14 +206,16 @@ public long add(ByteBuffer term, T key, InvalidVectorBehavior behavior) : ((CompactionVectorValues) vectorValues).add(ordinal, term); bytesUsed += VectorPostings.emptyBytesUsed() + VectorPostings.bytesPerPosting(); postingsByOrdinal.put(ordinal, postings); - graphInsertPermits.acquireThrowUncheckedOnInterrupt(1); + if (graphInsertPermits != null) + graphInsertPermits.acquireThrowUncheckedOnInterrupt(1); try { bytesUsed += builder.addGraphNode(ordinal, vectorValues); } finally { - graphInsertPermits.release(1); + if (graphInsertPermits != null) + graphInsertPermits.release(1); } return bytesUsed; } diff --git a/test/unit/org/apache/cassandra/db/memtable/MemtableConcurrentWriteLimitTest.java b/test/unit/org/apache/cassandra/db/memtable/MemtableConcurrentWriteLimitTest.java new file mode 100644 index 000000000000..76ce3d01e1c4 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/memtable/MemtableConcurrentWriteLimitTest.java @@ -0,0 +1,102 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.cassandra.db.memtable; + +import java.util.LinkedHashMap; +import java.util.Map; + +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.config.Config; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.config.InheritingClass; +import org.apache.cassandra.cql3.CQLTester; +import org.apache.cassandra.db.ColumnFamilyStore; + +import static org.apache.cassandra.db.memtable.AbstractShardedMemtable.SHARDS_OPTION; +import static org.apache.cassandra.db.memtable.ShardedSkipListMemtable.LOCKING_OPTION; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +/** + * {@link Memtable#limitsConcurrentWritesTo} must be true only for memtables that serialize writes per shard, and + * only when the shard count is within the limit asked about. + */ +public class MemtableConcurrentWriteLimitTest extends CQLTester +{ + private static final int SHARDS = 4; + + // Overrides CQLTester.setUpClass so the memtable configurations are registered before the server is prepared + @BeforeClass + public static void setUpClass() + { + prePrepareServer(); + + LinkedHashMap memtableConfig = new LinkedHashMap<>(); + memtableConfig.put("skiplist", new InheritingClass(null, SkipListMemtable.class.getName(), Map.of())); + memtableConfig.put("trie", new InheritingClass(null, TrieMemtable.class.getName(), Map.of(SHARDS_OPTION, String.valueOf(SHARDS)))); + memtableConfig.put("sharded", new InheritingClass(null, ShardedSkipListMemtable.class.getName(), Map.of(SHARDS_OPTION, String.valueOf(SHARDS)))); + memtableConfig.put("sharded_locking", new InheritingClass(null, ShardedSkipListMemtable.class.getName(), Map.of(SHARDS_OPTION, String.valueOf(SHARDS), LOCKING_OPTION, "true"))); + DatabaseDescriptor.getRawConfig().memtable = new Config.MemtableOptions(); + DatabaseDescriptor.getRawConfig().memtable.configurations = memtableConfig; + + prepareServer(); + } + + @Test + public void unshardedMemtableDoesNotLimitWriters() + { + assertFalse(memtableFor("skiplist").limitsConcurrentWritesTo(Integer.MAX_VALUE)); + } + + @Test + public void shardedSkipListWithoutLockingDoesNotLimitWriters() + { + assertFalse(memtableFor("sharded").limitsConcurrentWritesTo(Integer.MAX_VALUE)); + } + + @Test + public void trieMemtableLimitsWritersToShardCount() + { + assertLimitsToShardCount(memtableFor("trie")); + } + + @Test + public void lockingShardedSkipListLimitsWritersToShardCount() + { + assertLimitsToShardCount(memtableFor("sharded_locking")); + } + + private void assertLimitsToShardCount(Memtable memtable) + { + // the memtable may get fewer shards than requested, e.g. if local ranges cannot be split that finely + int shards = getCurrentColumnFamilyStore().localRangeSplits(SHARDS).shardCount(); + assertTrue(memtable.limitsConcurrentWritesTo(shards)); + assertTrue(memtable.limitsConcurrentWritesTo(shards + 1)); + assertFalse(memtable.limitsConcurrentWritesTo(shards - 1)); + } + + private Memtable memtableFor(String memtableConfig) + { + createTable("CREATE TABLE %s (pk int PRIMARY KEY, v int) WITH memtable = '" + memtableConfig + '\''); + ColumnFamilyStore cfs = getCurrentColumnFamilyStore(); + return cfs.getTracker().getView().getCurrentMemtable(); + } +} From ed31aa3143436d3c6103ee5bcf4f3a4cf5691276 Mon Sep 17 00:00:00 2001 From: Jeremiah Jordan Date: Fri, 4 Sep 2026 17:56:41 -0500 Subject: [PATCH 4/4] Cover both graph insert paths in VectorMemoryIndexTest Review update: the mocked memtable always answered false to limitsConcurrentWritesTo(), so the tests never took the path where OnHeapGraph relies on the memtable and creates no semaphore. Have the mock answer according to the number of writer threads each test actually uses, so the tests within jvector's limit skip the semaphore and the test beyond it exercises it. The memtable implementations themselves are covered by MemtableConcurrentWriteLimitTest. Also the doc nit. patch by Jeremiah Jordan; reviewed by XXX for CASSANDRA-21644 --- .../cassandra/db/memtable/Memtable_API.md | 2 +- .../sai/memory/VectorMemoryIndexTest.java | 30 ++++++++++++------- 2 files changed, 20 insertions(+), 12 deletions(-) diff --git a/src/java/org/apache/cassandra/db/memtable/Memtable_API.md b/src/java/org/apache/cassandra/db/memtable/Memtable_API.md index 8e62b2f5bdbf..ae1eda1ca7ba 100644 --- a/src/java/org/apache/cassandra/db/memtable/Memtable_API.md +++ b/src/java/org/apache/cassandra/db/memtable/Memtable_API.md @@ -135,7 +135,7 @@ implementations) are provided as the `AbstractMemtable` (statistics tracking), ` commit log span tracking) and `AbstractAllocatorMemtable` (adds memory management via the `Allocator` class, together with flush triggering on memory use and time interval expiration). -A memtable that serializes writes, for example per shard, can say so through `limitsConcurrentWritesTo`. Secondary +A memtable that serializes writes, for example per shard, can say so through `limitsConcurrentWritesTo()`. Secondary indexes whose implementation limits the number of concurrent writers use this to skip their own bounding when the memtable already provides it. The default answer is false. diff --git a/test/unit/org/apache/cassandra/index/sai/memory/VectorMemoryIndexTest.java b/test/unit/org/apache/cassandra/index/sai/memory/VectorMemoryIndexTest.java index d730553c17b9..91ea00ad75b7 100644 --- a/test/unit/org/apache/cassandra/index/sai/memory/VectorMemoryIndexTest.java +++ b/test/unit/org/apache/cassandra/index/sai/memory/VectorMemoryIndexTest.java @@ -140,8 +140,7 @@ public void setup() throws Throwable public void randomQueryTest() throws Exception { // A non-null memtable tells it to track the mapping from primary key to vector, needed for brute force search - Memtable memtable = Mockito.mock(Memtable.class); - memtableIndex = new VectorMemoryIndex(index, memtable); + memtableIndex = new VectorMemoryIndex(index, mockMemtable(1)); for (int row = 0; row < getRandom().nextIntBetween(1000, 5000); row++) { @@ -255,7 +254,9 @@ public void testConcurrentAddsWithSharedVectors() throws Exception /** * More writers than jvector's GraphIndexBuilder can serve at once must wait, not fail the insert. The other - * concurrent tests use exactly availableProcessors writers, which never exceeds jvector's limit. + * concurrent tests use exactly availableProcessors writers, within jvector's limit, so their memtable reports that + * it bounds writers and OnHeapGraph skips its semaphore; this test's memtable cannot make that promise, so + * OnHeapGraph bounds the writers itself. */ @Test public void testConcurrentAddsExceedingJVectorPoolCap() throws Exception @@ -277,8 +278,7 @@ public void testConcurrentAddsExceedingJVectorPoolCap() throws Exception */ private void testConcurrentAddsAreEventuallyConsistent(int numThreads, int vectorsPerThread, BiFunction vectorFactory) throws Exception { - Memtable memtable = Mockito.mock(Memtable.class); - memtableIndex = new VectorMemoryIndex(index, memtable); + memtableIndex = new VectorMemoryIndex(index, mockMemtable(numThreads)); int totalInserted = numThreads * vectorsPerThread; @@ -388,11 +388,9 @@ public void testConcurrentAddsAndOrderBySharedVectors() throws Exception */ public void testConcurrentAddsAndOrderByNeverThrow(BiFunction vectorFactory) throws Exception { - Memtable memtable = Mockito.mock(Memtable.class); - memtableIndex = new VectorMemoryIndex(index, memtable); - int numWriterThreads = Runtime.getRuntime().availableProcessors(); int numReaderThreads = Runtime.getRuntime().availableProcessors(); + memtableIndex = new VectorMemoryIndex(index, mockMemtable(numWriterThreads)); int totalInserted = numWriterThreads * VECTORS_PER_THREAD; // Pre-seed enough rows that orderBy() always has a non-empty graph to search, @@ -552,11 +550,9 @@ public void testConcurrentAddsAndOrderResultsBySharedVectors() throws Exception */ private void testConcurrentAddsAndOrderResultsByNeverThrow(BiFunction vectorFactory) throws Exception { - Memtable memtable = Mockito.mock(Memtable.class); - memtableIndex = new VectorMemoryIndex(index, memtable); - int numWriterThreads = Runtime.getRuntime().availableProcessors(); int numReaderThreads = Runtime.getRuntime().availableProcessors(); + memtableIndex = new VectorMemoryIndex(index, mockMemtable(numWriterThreads)); int totalInserted = numWriterThreads * VECTORS_PER_THREAD; // Pre-seed rows so orderResultsBy() always has a non-empty [minimumKey, maximumKey] @@ -769,6 +765,18 @@ private void addRow(int pk, ByteBuffer value) keyMap.put(key, pk); } + /** + * A memtable that answers {@link Memtable#limitsConcurrentWritesTo} truthfully for the number of writer threads the + * test will use. With that many writers at or under {@link OnHeapGraph#MAX_CONCURRENT_GRAPH_INSERTS}, OnHeapGraph + * relies on the memtable and creates no semaphore; with more, it bounds the writers itself. + */ + private static Memtable mockMemtable(int writers) + { + Memtable memtable = Mockito.mock(Memtable.class); + Mockito.when(memtable.limitsConcurrentWritesTo(Mockito.anyInt())).thenAnswer(invocation -> writers <= (int) invocation.getArgument(0)); + return memtable; + } + private DecoratedKey makeKey(TableMetadata table, Integer partitionKey) { ByteBuffer key = table.partitionKeyType.fromString(partitionKey.toString());