Batch short-read ID buffer replenishment to reduce monitor contention

[liulx20]

Summary

The Interactive v1 driver keeps recently observed person and message IDs in two shared bounded queues. Each queue is a Guava EvictingQueue wrapped by Queues.synchronizedQueue:

static Queue<Long> synchronizedCircularQueueBuffer( int bufferSize )
{
    return Queues.synchronizedQueue( EvictingQueue.<Long>create( bufferSize ) );
}

Both queues have capacity 1024. They are shared by all operation-executor workers. Producers replenish them from query results and short-query factories consume one ID with poll().

In v1.2.0, SQ2, SQ3, SQ7, and multiple complex-query branches add list results to these shared buffers one ID at a time. SQ3 is a representative example because it can return many friend IDs:

case LdbcShortQuery3PersonFriends.TYPE:
{
    List<LdbcShortQuery3PersonFriendsResult> typedResults =
            (List<LdbcShortQuery3PersonFriendsResult>) result;
    for ( int i = 0; i < typedResults.size(); i++ )
    {
        personIdBuffer.add( typedResults.get( i ).getPersonId() );
    }
    break;
}

Every add() acquires the monitor of its destination buffer. A list-valued result can contain hundreds of IDs, so one completed operation may acquire the same monitor hundreds of times. At high driver concurrency this creates a hot, highly contended critical section in the operation-completion path.

Affected source in v1.2.0:
LdbcSnbShortReadGenerator.java

Observed behavior

In the initial SF1000 run, the driver used 192 operation-executor workers and was already close to the schedule-audit limit, while the SUT process consumed only about 1400% CPU (approximately 14 logical cores). This was consistent with a possible client-side bottleneck.

During SF1000 experiments with 192--256 operation-executor workers, JFR lock profiles and repeated thread dumps showed worker stacks blocked in com.google.common.collect.Synchronized$SynchronizedCollection.add, called from ResultBufferReplenishFun.replenish.

A representative blocked stack was:

com.google.common.collect.Synchronized$SynchronizedCollection.add
org.ldbcouncil.snb.driver.workloads.interactive
    .LdbcSnbShortReadGenerator$ResultBufferReplenishFun.replenish
org.ldbcouncil.snb.driver.ChildOperationExecutor
org.ldbcouncil.snb.driver.ThreadPoolOperationExecutor

These samples identify the shared queue as a contention hotspot in the operation-completion path.

Why this should be fixed in the driver

The contention mechanism is in the driver; its severity depends on workload cardinality and concurrency. Contended monitor work consumes operation-executor capacity, reducing driver headroom and potentially causing schedule compliance to reflect a client-side bottleneck rather than SUT capacity.

Reproduction

Run an Interactive workload with a high worker count and list-valued buffer replenishment enabled, then take a JFR profile during the stable phase. SQ3 is a convenient reproducer because of its result cardinality:

jcmd <pid> JFR.start \
  name=short_read_buffer \
  settings=profile \
  duration=60s \
  filename=short-read-buffer.jfr

Inspect jdk.JavaMonitorEnter events and worker thread stacks for SynchronizedCollection.add or SynchronizedQueue.poll with LdbcSnbShortReadGenerator below them. For a quantitative reproduction, count IDs passed to personIdBuffer/messageIdBuffer and monitor-wait time per completed operation.

Proposed fix

Keep the existing queue implementation, capacity, eviction policy, and consumer code. For every list-valued replenish branch, materialize the IDs for each destination buffer before entering the queue monitor, then submit them with one
addAll call per buffer and result collection. The SQ3 diff below illustrates this pattern.

Queues.synchronizedQueue(...).addAll(...) holds the same monitor once for the whole collection. This changes the lock cost from one acquisition per returned ID to one acquisition per destination buffer for that result collection.

ID extraction and boxing happen while building the temporary list outside the monitor. The single synchronized addAll section then performs only the queue updates, avoiding repeated monitor acquisition and handoff between IDs.

Semantics

For a fixed initial state in isolated execution, batching offers the same IDs in the same per-buffer order and preserves capacity and oldest-entry eviction; the queue and consumer code remain unchanged. However, addAll makes each per-buffer batch atomic, narrowing producer/consumer interleavings and, for branches that write both person and message IDs, changing the relative timing between the two queues. We would like maintainers to confirm that this remains consistent with the intended workload semantics.

Example patch for SQ3

 import com.google.common.collect.EvictingQueue;
 import com.google.common.collect.Ordering;
 import com.google.common.collect.Queues;
+import java.util.ArrayList;

 case LdbcShortQuery3PersonFriends.TYPE:
 {
     List<LdbcShortQuery3PersonFriendsResult> typedResults =
             (List<LdbcShortQuery3PersonFriendsResult>) result;
-    for ( int i = 0; i < typedResults.size(); i++ )
-    {
-        personIdBuffer.add( typedResults.get( i ).getPersonId() );
-    }
+    List<Long> personIds = new ArrayList<>( typedResults.size() );
+    for ( int i = 0; i < typedResults.size(); i++ )
+    {
+        personIds.add( typedResults.get( i ).getPersonId() );
+    }
+    personIdBuffer.addAll( personIds );
     break;
 }

Additional metrics-buffer adjustment

The same profiles also showed occasional operation-executor stalls while publishing completion events to DisruptorSbeMetricsService. Its capacity is hard-coded to 1024. At high completion rates this covers only a few milliseconds
of events, so a short pause in the single metrics consumer can propagate backpressure to operation-executor workers. The captured path was MultiProducerSequencer.next -> RingBuffer.publishEvent ->DisruptorSbeMetricsServiceWriter.submitOperationResult.

Along with the result-buffer batching fix, we propose increasing only this capacity:

-        int bufferSize = 1024;
+        int bufferSize = 65536;

This increases burst capacity by 64 times. It does not drop or reorder metrics events and does not increase the consumer's sustained processing rate; it only absorbs longer bursts before producers must wait.

Would maintainers accept this small capacity adjustment together with the result-buffer batching change? Alternatively, the capacity could be exposed as a validated power-of-two driver configuration parameter. We would also appreciate guidance on whether this fix can be included in a v1.2.x maintenance release.