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/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..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,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 a823943a4e77..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,10 @@ 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; import org.cliffc.high_scale_lib.NonBlockingHashMapLong; import org.slf4j.Logger; @@ -63,6 +67,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 +76,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 +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. + // 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; @@ -107,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, @@ -187,7 +206,17 @@ 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); + if (graphInsertPermits != null) + graphInsertPermits.acquireThrowUncheckedOnInterrupt(1); + try + { + bytesUsed += builder.addGraphNode(ordinal, vectorValues); + } + finally + { + if (graphInsertPermits != null) + graphInsertPermits.release(1); + } return bytesUsed; } else 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(); + } +} 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..91ea00ad75b7 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; @@ -138,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++) { @@ -242,13 +243,26 @@ 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, 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 + { + int numThreads = 2 * OnHeapGraph.MAX_CONCURRENT_GRAPH_INSERTS; + testConcurrentAddsAreEventuallyConsistent(numThreads, VECTORS_PER_THREAD_BEYOND_POOL_CAP, (threadId, i) -> randomVectorFromThreadLocal()); } /** @@ -262,13 +276,11 @@ 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); + memtableIndex = new VectorMemoryIndex(index, mockMemtable(numThreads)); - 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 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, @@ -540,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] @@ -757,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());