diff --git a/libs/concurrent-queue/src/jmh/java/org/opensearch/common/queue/LockablePoolBenchmark.java b/libs/concurrent-queue/src/jmh/java/org/opensearch/common/queue/LockablePoolBenchmark.java new file mode 100644 index 0000000000000..38f2df3f5fa18 --- /dev/null +++ b/libs/concurrent-queue/src/jmh/java/org/opensearch/common/queue/LockablePoolBenchmark.java @@ -0,0 +1,253 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.common.queue; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Group; +import org.openjdk.jmh.annotations.GroupThreads; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Warmup; +import org.openjdk.jmh.infra.Blackhole; + +import java.util.LinkedList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.LockSupport; +import java.util.concurrent.locks.ReentrantLock; + +/** + * JMH benchmark for {@link LockablePool} measuring: + * + */ +@Fork(2) +@Warmup(iterations = 2, time = 3) +@Measurement(iterations = 3, time = 5) +@State(Scope.Benchmark) +@SuppressWarnings("unused") +public class LockablePoolBenchmark { + + @Param({ "4", "8" }) + int concurrency; + + private LockablePool pool; + + @Setup(Level.Iteration) + public void setup() { + AtomicInteger counter = new AtomicInteger(0); + pool = new LockablePool<>(() -> new PoolEntry(counter.getAndIncrement()), LinkedList::new, concurrency); + // Pre-warm the pool with more entries to populate multiple stripes + for (int i = 0; i < concurrency * 4; i++) { + PoolEntry e = pool.getAndLock(); + pool.releaseAndUnlock(e); + } + } + + // ── Mixed workload: writers + periodic refresh (1s interval) ── + + @Benchmark + @Group("mixed_7w_1r") + @GroupThreads(7) + @BenchmarkMode(Mode.Throughput) + @OutputTimeUnit(TimeUnit.MICROSECONDS) + public void writers_7w1r(Blackhole bh) { + PoolEntry e = pool.getAndLock(); + bh.consume(simulateWork(e)); + pool.releaseAndUnlock(e); + } + + @Benchmark + @Group("mixed_7w_1r") + @GroupThreads(1) + @BenchmarkMode(Mode.Throughput) + @OutputTimeUnit(TimeUnit.MICROSECONDS) + public List refresh_7w1r() throws InterruptedException { + Thread.sleep(1000); + return pool.checkoutAll(); + } + + @Benchmark + @Group("mixed_3w_1r") + @GroupThreads(3) + @BenchmarkMode(Mode.Throughput) + @OutputTimeUnit(TimeUnit.MICROSECONDS) + public void writers_3w1r(Blackhole bh) { + PoolEntry e = pool.getAndLock(); + bh.consume(simulateWork(e)); + pool.releaseAndUnlock(e); + } + + @Benchmark + @Group("mixed_3w_1r") + @GroupThreads(1) + @BenchmarkMode(Mode.Throughput) + @OutputTimeUnit(TimeUnit.MICROSECONDS) + public List refresh_3w1r() throws InterruptedException { + Thread.sleep(1000); + return pool.checkoutAll(); + } + + // ── Aggressive refresh: 10ms interval to stress checkoutAll Phase 3 ── + // This makes checkoutAll fire ~100x/sec instead of 1x/sec, amplifying + // the difference between per-item remove (old) and bulk removeIf (new). + + @Benchmark + @Group("aggressive_7w_1r") + @GroupThreads(7) + @BenchmarkMode(Mode.Throughput) + @OutputTimeUnit(TimeUnit.MICROSECONDS) + public void writers_aggressive_7w1r(Blackhole bh) { + PoolEntry e = pool.getAndLock(); + bh.consume(simulateWork(e)); + pool.releaseAndUnlock(e); + } + + @Benchmark + @Group("aggressive_7w_1r") + @GroupThreads(1) + @BenchmarkMode(Mode.Throughput) + @OutputTimeUnit(TimeUnit.MICROSECONDS) + public List refresh_aggressive_7w1r() { + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10)); + return pool.checkoutAll(); + } + + @Benchmark + @Group("aggressive_3w_1r") + @GroupThreads(3) + @BenchmarkMode(Mode.Throughput) + @OutputTimeUnit(TimeUnit.MICROSECONDS) + public void writers_aggressive_3w1r(Blackhole bh) { + PoolEntry e = pool.getAndLock(); + bh.consume(simulateWork(e)); + pool.releaseAndUnlock(e); + } + + @Benchmark + @Group("aggressive_3w_1r") + @GroupThreads(1) + @BenchmarkMode(Mode.Throughput) + @OutputTimeUnit(TimeUnit.MICROSECONDS) + public List refresh_aggressive_3w1r() { + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10)); + return pool.checkoutAll(); + } + + // ── Writer latency during refresh contention (sample mode) ── + // Measures per-operation latency distribution to capture tail latency + // spikes caused by checkoutAll holding the pool lock. + + @Benchmark + @Group("latency_7w_1r") + @GroupThreads(7) + @BenchmarkMode(Mode.SampleTime) + @OutputTimeUnit(TimeUnit.NANOSECONDS) + public void writers_latency_7w1r(Blackhole bh) { + PoolEntry e = pool.getAndLock(); + bh.consume(simulateWork(e)); + pool.releaseAndUnlock(e); + } + + @Benchmark + @Group("latency_7w_1r") + @GroupThreads(1) + @BenchmarkMode(Mode.SampleTime) + @OutputTimeUnit(TimeUnit.NANOSECONDS) + public List refresh_latency_7w1r() { + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10)); + return pool.checkoutAll(); + } + + // ── Isolated: pure writer throughput (no refresh contention) ── + + @Benchmark + @Group("writers_only_4t") + @GroupThreads(4) + @BenchmarkMode(Mode.Throughput) + @OutputTimeUnit(TimeUnit.MICROSECONDS) + public void writersOnly_4t(Blackhole bh) { + PoolEntry e = pool.getAndLock(); + bh.consume(simulateWork(e)); + pool.releaseAndUnlock(e); + } + + @Benchmark + @Group("writers_only_8t") + @GroupThreads(8) + @BenchmarkMode(Mode.Throughput) + @OutputTimeUnit(TimeUnit.MICROSECONDS) + public void writersOnly_8t(Blackhole bh) { + PoolEntry e = pool.getAndLock(); + bh.consume(simulateWork(e)); + pool.releaseAndUnlock(e); + } + + // ── Isolated checkoutAll throughput (no writers) ── + // Directly measures checkoutAll cost with a pre-populated pool. + + @Benchmark + @BenchmarkMode(Mode.AverageTime) + @OutputTimeUnit(TimeUnit.NANOSECONDS) + public List checkoutAll_isolated() { + // Re-populate pool before each checkout + for (int i = 0; i < concurrency; i++) { + PoolEntry e = pool.getAndLock(); + pool.releaseAndUnlock(e); + } + return pool.checkoutAll(); + } + + private static long simulateWork(PoolEntry entry) { + long result = entry.hashCode(); + for (int i = 0; i < 20; i++) { + result ^= (result << 13); + result ^= (result >> 7); + result ^= (result << 17); + } + return result; + } + + static final class PoolEntry implements Lockable { + final int id; + private final ReentrantLock lock = new ReentrantLock(); + + PoolEntry(int id) { + this.id = id; + } + + @Override + public void lock() { + lock.lock(); + } + + @Override + public boolean tryLock() { + return lock.tryLock(); + } + + @Override + public void unlock() { + lock.unlock(); + } + } +} diff --git a/libs/concurrent-queue/src/main/java/org/opensearch/common/queue/ConcurrentQueue.java b/libs/concurrent-queue/src/main/java/org/opensearch/common/queue/ConcurrentQueue.java index f7856c9464f54..b37d73eabc333 100644 --- a/libs/concurrent-queue/src/main/java/org/opensearch/common/queue/ConcurrentQueue.java +++ b/libs/concurrent-queue/src/main/java/org/opensearch/common/queue/ConcurrentQueue.java @@ -137,4 +137,24 @@ boolean remove(T entry) { } return false; } + + /** + * Removes all entries matching the given predicate in a single pass across all stripes. + * Each stripe is locked once, and all matching entries within that stripe are removed + * before moving to the next stripe. + * + * @param predicate the condition for removal + */ + void removeIf(Predicate predicate) { + for (int i = 0; i < concurrency; ++i) { + final Lock lock = locks[i]; + final Queue queue = queues[i]; + lock.lock(); + try { + queue.removeIf(predicate); + } finally { + lock.unlock(); + } + } + } } diff --git a/libs/concurrent-queue/src/main/java/org/opensearch/common/queue/LockableConcurrentQueue.java b/libs/concurrent-queue/src/main/java/org/opensearch/common/queue/LockableConcurrentQueue.java index 8d03a73fde08e..75005b4a88f99 100644 --- a/libs/concurrent-queue/src/main/java/org/opensearch/common/queue/LockableConcurrentQueue.java +++ b/libs/concurrent-queue/src/main/java/org/opensearch/common/queue/LockableConcurrentQueue.java @@ -10,6 +10,7 @@ import java.util.Queue; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Predicate; import java.util.function.Supplier; /** @@ -66,6 +67,15 @@ public boolean remove(T entry) { return queue.remove(entry); } + /** + * Removes all entries matching the given predicate in a single pass across all stripes. + * + * @param predicate the condition for removal + */ + public void removeIf(Predicate predicate) { + queue.removeIf(predicate); + } + /** * Add an entry to the queue and unlock it, in that order. * diff --git a/libs/concurrent-queue/src/main/java/org/opensearch/common/queue/LockablePool.java b/libs/concurrent-queue/src/main/java/org/opensearch/common/queue/LockablePool.java index 505c92abd5833..ddc24b364c46e 100644 --- a/libs/concurrent-queue/src/main/java/org/opensearch/common/queue/LockablePool.java +++ b/libs/concurrent-queue/src/main/java/org/opensearch/common/queue/LockablePool.java @@ -81,30 +81,53 @@ public void releaseAndUnlock(T item) { /** * Lock and checkout all items from the pool. + *

+ * Phase 1: Snapshot the items set under the pool lock. + * Phase 2: Lock each item outside the monitor to avoid holding it while blocking on in-flight operations. + * Phase 3: Remove checked-out items from the set and bulk-remove from the queue in a single pass. * * @return unmodifiable list of all items locked by current thread * @throws IllegalStateException if the pool is closed */ public List checkoutAll() { ensureOpen(); - List lockedItems = new ArrayList<>(); - List checkedOutItems = new ArrayList<>(); - for (T item : this) { + + // Phase 1: Snapshot + List snapshot; + synchronized (this) { + if (items.isEmpty()) { + return Collections.emptyList(); + } + snapshot = new ArrayList<>(items.size()); + snapshot.addAll(items); + } + + // Phase 2: Lock outside monitor + for (T item : snapshot) { item.lock(); - lockedItems.add(item); } + + // Phase 3: Process + bulk cleanup + List checkedOutItems = new ArrayList<>(snapshot.size()); synchronized (this) { - for (T item : lockedItems) { + Set toRemoveFromQueue = Collections.newSetFromMap(new IdentityHashMap<>(snapshot.size())); + + for (T item : snapshot) { try { - if (isRegistered(item) && items.remove(item)) { - availableItems.remove(item); + if (items.remove(item)) { + toRemoveFromQueue.add(item); checkedOutItems.add(item); } } finally { item.unlock(); } } + + if (toRemoveFromQueue.isEmpty() == false) { + availableItems.removeIf(toRemoveFromQueue::contains); + } } + return Collections.unmodifiableList(checkedOutItems); } diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDocumentInput.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDocumentInput.java index 75d7b37368f0c..5f095eec3eb07 100644 --- a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDocumentInput.java +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeDocumentInput.java @@ -13,6 +13,7 @@ import org.opensearch.index.engine.dataformat.DocumentInput; import org.opensearch.index.mapper.MappedFieldType; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Objects; @@ -47,7 +48,7 @@ public CompositeDocumentInput( ) { this.primaryFormat = Objects.requireNonNull(primaryFormat, "primaryFormat must not be null"); this.primaryDocumentInput = Objects.requireNonNull(primaryDocumentInput, "primaryDocumentInput must not be null"); - this.secondaryDocumentInputs = Map.copyOf( + this.secondaryDocumentInputs = Collections.unmodifiableMap( Objects.requireNonNull(secondaryDocumentInputs, "secondaryDocumentInputs must not be null") ); } diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java index b1080cbe9a63c..7f2d02a3d2b5d 100644 --- a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeIndexingExecutionEngine.java @@ -31,7 +31,7 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Collection; -import java.util.LinkedHashMap; +import java.util.IdentityHashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; @@ -275,7 +275,7 @@ public void deleteFiles(Map> filesToDelete) throws IO @Override public CompositeDocumentInput newDocumentInput() { DocumentInput primaryInput = primaryEngine.newDocumentInput(); - Map> secondaryInputMap = new LinkedHashMap<>(); + Map> secondaryInputMap = new IdentityHashMap<>(); for (IndexingExecutionEngine engine : secondaryEngines) { secondaryInputMap.put(engine.getDataFormat(), engine.newDocumentInput()); } diff --git a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeWriter.java b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeWriter.java index 309494adf21e9..63c68dbbea0cd 100644 --- a/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeWriter.java +++ b/sandbox/plugins/composite-engine/src/main/java/org/opensearch/composite/CompositeWriter.java @@ -21,8 +21,8 @@ import org.opensearch.index.engine.exec.WriterFileSet; import java.io.IOException; -import java.util.AbstractMap; -import java.util.LinkedHashMap; +import java.util.Collections; +import java.util.IdentityHashMap; import java.util.Map; import java.util.Optional; import java.util.concurrent.atomic.AtomicReference; @@ -44,7 +44,8 @@ class CompositeWriter implements Writer, Lockable { private static final Logger logger = LogManager.getLogger(CompositeWriter.class); - private final Map.Entry>> primaryWriter; + private final DataFormat primaryFormat; + private final Writer> primaryWriter; private final Map>> secondaryWritersByFormat; private final ReentrantLock lock; private final long writerGeneration; @@ -87,16 +88,14 @@ enum WriterState { this.writerGeneration = writerGeneration; IndexingExecutionEngine primaryDelegate = engine.getPrimaryDelegate(); - this.primaryWriter = new AbstractMap.SimpleImmutableEntry<>( - primaryDelegate.getDataFormat(), - (Writer>) primaryDelegate.createWriter(writerGeneration) - ); + this.primaryFormat = primaryDelegate.getDataFormat(); + this.primaryWriter = (Writer>) primaryDelegate.createWriter(writerGeneration); - Map>> secondaries = new LinkedHashMap<>(); + Map>> secondaries = new IdentityHashMap<>(); for (IndexingExecutionEngine delegate : engine.getSecondaryDelegates()) { secondaries.put(delegate.getDataFormat(), (Writer>) delegate.createWriter(writerGeneration)); } - this.secondaryWritersByFormat = Map.copyOf(secondaries); + this.secondaryWritersByFormat = Collections.unmodifiableMap(secondaries); this.rowIdGenerator = new RowIdGenerator(CompositeWriter.class.getName()); } @@ -106,11 +105,11 @@ public WriteResult addDoc(CompositeDocumentInput doc) throws IOException { throw new IllegalStateException("Cannot add document to writer in state " + state.get()); } // Write to primary first - WriteResult primaryResult = primaryWriter.getValue().addDoc(doc.getPrimaryInput()); + WriteResult primaryResult = primaryWriter.addDoc(doc.getPrimaryInput()); switch (primaryResult) { - case WriteResult.Success s -> logger.trace("Successfully added document in primary format [{}]", primaryWriter.getKey().name()); + case WriteResult.Success s -> logger.trace("Successfully added document in primary format [{}]", primaryFormat.name()); case WriteResult.Failure f -> { - logger.debug("Failed to add document in primary format [{}]", primaryWriter.getKey().name()); + logger.debug("Failed to add document in primary format [{}]", primaryFormat.name()); return primaryResult; } } @@ -141,8 +140,8 @@ public WriteResult addDoc(CompositeDocumentInput doc) throws IOException { public FileInfos flush() throws IOException { FileInfos.Builder builder = FileInfos.builder(); // Flush primary - Optional primaryWfs = primaryWriter.getValue().flush().getWriterFileSet(primaryWriter.getKey()); - primaryWfs.ifPresent(writerFileSet -> builder.putWriterFileSet(primaryWriter.getKey(), writerFileSet)); + Optional primaryWfs = primaryWriter.flush().getWriterFileSet(primaryFormat); + primaryWfs.ifPresent(writerFileSet -> builder.putWriterFileSet(primaryFormat, writerFileSet)); // Flush secondaries for (Writer> writer : secondaryWritersByFormat.values()) { FileInfos fileInfos = writer.flush(); @@ -156,7 +155,7 @@ public FileInfos flush() throws IOException { @Override public void sync() throws IOException { - primaryWriter.getValue().sync(); + primaryWriter.sync(); for (Writer> writer : secondaryWritersByFormat.values()) { writer.sync(); } @@ -164,7 +163,7 @@ public void sync() throws IOException { @Override public void close() throws IOException { - primaryWriter.getValue().close(); + primaryWriter.close(); for (Writer> writer : secondaryWritersByFormat.values()) { writer.close(); }