Jeremiah Jordan created CASSANDRA-21644:
-------------------------------------------
Summary: Concurrent SAI vector inserts fail with "Number of
outstanding pooled objects has gone beyond the limit" since CASSANDRA-21160
Key: CASSANDRA-21644
URL: https://issues.apache.org/jira/browse/CASSANDRA-21644
Project: Apache Cassandra
Issue Type: Bug
Components: Feature/Vector Search
Reporter: Jeremiah Jordan
Since CASSANDRA-21160 removed {{synchronized}} from
{{{}VectorMemoryIndex.add(){}}}, concurrent inserts into a table with an SAI
vector index fail once more than {{Runtime.availableProcessors() + 1}} threads
are inside {{GraphIndexBuilder.addGraphNode()}} on the same memtable graph.
jvector 1.0.2 caps the pooled scratch objects per {{GraphIndexBuilder}} at that
number and throws {{IllegalStateException}} rather than waiting. The mutation
fails with {{{}WriteFailureException{}}}. Before CASSANDRA-21160 all inserts
into a memtable graph were serialized on the {{VectorMemoryIndex}} monitor, so
the cap was unreachable.
Reproduces with the default configuration ({{{}SkipListMemtable{}}},
{{{}concurrent_writes: 32{}}}) on JDK 11 and JDK 25.
h3. Symptom
Client:
{code:java}
WriteFailureException: Operation failed - received 0 responses and 1 failures:
UNKNOWN
{code}
Server (captured from a unit test; everything from {{StorageProxy.mutate}} down
is the production write path):
{code:java}
ERROR [pool-2-thread-2] StorageProxy.java:2044 - Failed to apply mutation
locally :
java.lang.RuntimeException: Number of outstanding pooled objects has gone
beyond the limit of 3 for ks: cql_test_keyspace, table: table_testsiftsmall_00
at
org.apache.cassandra.db.ColumnFamilyStore.apply(ColumnFamilyStore.java:1552)
at
org.apache.cassandra.db.CassandraTableWriteHandler.write(CassandraTableWriteHandler.java:38)
at org.apache.cassandra.db.Keyspace.applyInternal(Keyspace.java:579)
at org.apache.cassandra.db.Keyspace.apply(Keyspace.java:434)
at org.apache.cassandra.db.Mutation.apply(Mutation.java:297)
at
org.apache.cassandra.service.StorageProxy$4.runMayThrow(StorageProxy.java:2037)
at
org.apache.cassandra.service.StorageProxy$LocalMutationRunnable.run(StorageProxy.java:3209)
at
org.apache.cassandra.concurrent.SEPExecutor.maybeExecuteImmediately(SEPExecutor.java:218)
at
org.apache.cassandra.concurrent.Stage.maybeExecuteImmediately(Stage.java:130)
at
org.apache.cassandra.service.StorageProxy.performLocally(StorageProxy.java:2026)
at
org.apache.cassandra.service.StorageProxy.sendToHintedReplicas(StorageProxy.java:1919)
at
org.apache.cassandra.service.StorageProxy.performWrite(StorageProxy.java:1752)
at
org.apache.cassandra.service.StorageProxy.mutate(StorageProxy.java:993)
...
Caused by: java.lang.IllegalStateException: Number of outstanding pooled
objects has gone beyond the limit of 3
at
io.github.jbellis.jvector.util.PoolingSupport$ThreadPooling.get(PoolingSupport.java:120)
at
io.github.jbellis.jvector.graph.GraphIndexBuilder.addGraphNode(GraphIndexBuilder.java:162)
at
org.apache.cassandra.index.sai.disk.v1.vector.OnHeapGraph.add(OnHeapGraph.java:191)
{code}
The limit is 3 because unit tests run with {{{}-XX:ActiveProcessorCount=2{}}}.
h3. Cause
* jvector 1.0.2 {{PoolingSupport$ThreadPooling}} is constructed with {{{}limit
= Runtime.getRuntime().availableProcessors() + 1{}}}; {{get()}} throws when the
queue is empty and {{limit}} objects are already outstanding. Each
{{GraphIndexBuilder}} owns three such pools ({{{}graphSearcher{}}},
{{{}naturalScratch{}}}, {{{}concurrentScratch{}}}). {{addGraphNode()}} takes
one object from each and holds it for the duration of the call, so at most
{{availableProcessors + 1}} threads can be inside {{addGraphNode()}} per
builder. The limit is not configurable: the builder uses the no-argument
factory and keeps the pools in private fields. {{OnHeapGraph}} creates one
builder per memtable index.
* Commit 448d98ce31 (CASSANDRA-21160) removed {{synchronized}} from
{{{}VectorMemoryIndex.add(){}}}. Only {{updateKeyBounds()}} remains
synchronized, and it runs after {{{}graph.add(){}}}. {{MemtableIndex}} and
{{MemtableIndexManager}} hold no lock.
* {{SkipListMemtable.put()}} locks per partition, so inserts to different
partitions reach {{OnHeapGraph.add()}} concurrently. {{concurrent_writes}}
defaults to 32, and {{Stage.maybeExecuteImmediately}} runs local mutations on
the calling thread, so the number of threads in the builder is not bounded by
{{concurrent_writes}} either. An 8-core node fails with 10 concurrent
inserters. {{TrieMemtable}} bounds it (shards =
{{{}getAvailableProcessors(){}}}, indexing under the shard write lock) but is
not the default.
h3. Why CI does not catch it
* {{VectorMemoryIndexTest}} (added by CASSANDRA-21160) uses exactly
{{availableProcessors}} writer threads.
* {{VectorSiftSmallTest}} on trunk uses {{{}IntStream.parallel(){}}}: common
pool parallelism {{procs - 1}} plus the calling thread = {{{}procs{}}}.
* Unit tests run with {{{}-XX:ActiveProcessorCount=2{}}}, so the limit is 3
and no test uses more than 2 or 3 writers.
h3. Reproduction
Insert into a vector-indexed table from {{availableProcessors + 2}} or more
threads concurrently. On the CASSANDRA-21171 branch, {{VectorSiftSmallTest}}
with a 4-thread fixed pool fails on every run on JDK 11 and JDK 25. The
following test, in the style of the existing concurrent tests in
{{{}VectorMemoryIndexTest{}}}, reproduces it deterministically:
{code:java}
@Test
public void testConcurrentAddsExceedingJVectorPoolCap() throws Exception
{
// jvector's limit is availableProcessors + 1; use twice that many writers
int numThreads = 2 * (Runtime.getRuntime().availableProcessors() + 1);
int vectorsPerThread = 500;
Memtable memtable = Mockito.mock(Memtable.class);
VectorMemoryIndex memtableIndex = new VectorMemoryIndex(index, memtable);
ExecutorService executor = Executors.newFixedThreadPool(numThreads);
CyclicBarrier barrier = new CyclicBarrier(numThreads);
List<Future<?>> futures = new ArrayList<>();
for (int t = 0; t < numThreads; t++)
{
final int threadId = t;
futures.add(executor.submit(() -> {
barrier.await();
for (int i = 0; i < vectorsPerThread; i++)
{
int pk = threadId * vectorsPerThread + i;
DecoratedKey key =
cfs.metadata().partitioner.decorateKey(Int32Type.instance.decompose(pk));
memtableIndex.add(key, Clustering.EMPTY,
randomVector(dimensionCount));
}
return null;
}));
}
executor.shutdown();
assertTrue(executor.awaitTermination(60, TimeUnit.SECONDS));
for (Future<?> f : futures)
{
try
{
f.get();
}
catch (ExecutionException e)
{
// IllegalStateException: Number of outstanding pooled objects has
gone beyond the limit of N
fail("Worker thread threw during concurrent add(): " +
e.getCause());
}
}
// then search the full ring and check the inserted keys come back with
finite, positive scores
}
{code}
h3. Possible fixes
* Bound the number of threads inside {{builder.addGraphNode()}} in
{{OnHeapGraph}} to {{{}availableProcessors + 1{}}}, for example with a
semaphore, so that additional writers wait instead of failing. Graph insertion
is CPU-bound, so this should keep the throughput gain from CASSANDRA-21160,
which came from going from one writer to one per core. Everything else in
{{VectorMemoryIndex}} and {{OnHeapGraph}} would stay lock-free.
* Restore {{synchronized}} on {{{}VectorMemoryIndex.add(){}}}. This reverts
CASSANDRA-21160 and its measured throughput gain.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
