You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2026/06/15 10:46:46 UTC
(pulsar) branch master updated: [improve][broker] Optimize TripleLongPriorityQueue heap operations (#26010)
This is an automated email from the ASF dual-hosted git repository.
nodece pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 6e8c4967561 [improve][broker] Optimize TripleLongPriorityQueue heap operations (#26010)
6e8c4967561 is described below
commit 6e8c496756189934c364d20f6ee224f26ce5d5f2
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Mon Jun 15 18:46:38 2026 +0800
[improve][broker] Optimize TripleLongPriorityQueue heap operations (#26010)
---
.../TripleLongPriorityQueueBenchmark.java | 168 +++++++++++++++++++++
.../util/collections/TripleLongPriorityQueue.java | 156 ++++++++++++-------
.../collections/TripleLongPriorityQueueTest.java | 47 ++++++
3 files changed, 315 insertions(+), 56 deletions(-)
diff --git a/microbench/src/main/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueBenchmark.java b/microbench/src/main/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueBenchmark.java
new file mode 100644
index 00000000000..1e50ab600bf
--- /dev/null
+++ b/microbench/src/main/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueBenchmark.java
@@ -0,0 +1,168 @@
+/*
+ * 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.pulsar.common.util.collections;
+
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+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.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * JMH benchmarks for {@link TripleLongPriorityQueue} simulating Pulsar delayed delivery workloads.
+ *
+ * <p>Three scenarios matching real usage:
+ * <ul>
+ * <li>{@link #recoveryBulkAddThenPop} — snapshot recovery: bulk add all entries, then pop all.
+ * Cold cache, large heap. This is the <b>worst case</b> for the hole-based optimization
+ * because cache misses dominate over reduced readLong calls.</li>
+ * <li>{@link #interleavedAddPop} — steady-state delayed delivery: batch add (messages arriving
+ * between timer ticks), then batch pop (getScheduledMessages). Heap stays warm in cache.
+ * This is the <b>primary hot path</b>.</li>
+ * <li>{@link #steadyState} — constant-depth steady state: pre-fill queue, then alternating
+ * small add/pop batches. Simulates sustained throughput with ~10K queue depth.</li>
+ * </ul>
+ *
+ * <p>Build and run:
+ * <pre>
+ * ./gradlew :microbench:shadowJar
+ * java -jar microbench/build/libs/microbench-*-benchmarks.jar ".*TripleLongPriorityQueue.*"
+ * </pre>
+ */
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@Fork(2)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@State(Scope.Thread)
+public class TripleLongPriorityQueueBenchmark {
+
+ @Param({"50000", "500000", "2000000"})
+ int size;
+
+ /**
+ * Recovery scenario: bulk add all then pop all.
+ * Simulates snapshot recovery from BookKeeper — cold cache, large heap.
+ */
+ @Benchmark
+ public void recoveryBulkAddThenPop(Blackhole bh) {
+ try (TripleLongPriorityQueue pq = new TripleLongPriorityQueue()) {
+ long baseTs = System.currentTimeMillis();
+ Random rng = new Random(42);
+ for (int i = 0; i < size; i++) {
+ long n1 = baseTs + rng.nextLong(3_600_000);
+ long n2 = i / 1000;
+ long n3 = i;
+ pq.add(n1, n2, n3);
+ }
+ while (!pq.isEmpty()) {
+ bh.consume(pq.peekN1());
+ pq.pop();
+ }
+ }
+ }
+
+ /**
+ * Interleaved scenario: batch add then batch pop, repeated.
+ * Simulates steady-state delayed delivery — messages arrive continuously,
+ * getScheduledMessages pops in batches of ~500 when consumers are ready.
+ * Heap is warm in L2/L3 cache between operations.
+ */
+ @Benchmark
+ public void interleavedAddPop(Blackhole bh) {
+ try (TripleLongPriorityQueue pq = new TripleLongPriorityQueue()) {
+ Random rng = new Random(42);
+ long baseTs = System.currentTimeMillis();
+ int batchSize = 500;
+ int totalAdded = 0;
+
+ while (totalAdded < size) {
+ // Batch add: messages arriving between timer ticks
+ int addCount = Math.min(batchSize + rng.nextInt(500), size - totalAdded);
+ for (int i = 0; i < addCount; i++) {
+ long n1 = baseTs + rng.nextLong(3_600_000);
+ long n2 = (totalAdded + i) / 1000;
+ long n3 = totalAdded + i;
+ pq.add(n1, n2, n3);
+ }
+ totalAdded += addCount;
+
+ // Batch pop: getScheduledMessages delivering to consumers
+ int popCount = (int) Math.min(batchSize, pq.size());
+ for (int i = 0; i < popCount; i++) {
+ bh.consume(pq.peekN1());
+ pq.pop();
+ }
+ }
+ // Drain remaining
+ while (!pq.isEmpty()) {
+ bh.consume(pq.peekN1());
+ pq.pop();
+ }
+ }
+ }
+
+ /**
+ * Steady-state scenario: pre-fill queue, then alternating small add/pop.
+ * Simulates sustained throughput with constant ~10K queue depth.
+ * Heap stays hot in cache — this is where the readLong reduction matters most.
+ */
+ @Benchmark
+ public void steadyState(Blackhole bh) {
+ int steadyDepth = 10_000;
+ try (TripleLongPriorityQueue pq = new TripleLongPriorityQueue()) {
+ Random rng = new Random(42);
+ long baseTs = System.currentTimeMillis();
+ long seq = 0;
+
+ // Pre-fill
+ for (int i = 0; i < steadyDepth; i++) {
+ pq.add(baseTs + rng.nextLong(3_600_000), seq / 1000, seq);
+ seq++;
+ }
+
+ // Alternating add/pop to maintain steady depth
+ int ops = 0;
+ while (ops < size) {
+ // Small batch add
+ int addCount = 100 + rng.nextInt(100);
+ for (int i = 0; i < addCount && ops < size; i++) {
+ pq.add(baseTs + rng.nextLong(3_600_000), seq / 1000, seq);
+ seq++;
+ ops++;
+ }
+ // Small batch pop
+ int popCount = addCount - rng.nextInt(20);
+ for (int i = 0; i < popCount && !pq.isEmpty(); i++) {
+ bh.consume(pq.peekN1());
+ pq.pop();
+ }
+ }
+ }
+ }
+}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueue.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueue.java
index e75ae21ab85..cd878c64284 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueue.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueue.java
@@ -24,6 +24,25 @@ import static com.google.common.base.Preconditions.checkArgument;
* Provides a priority-queue implementation specialized on items composed by 3 longs.
*
* <p>This class is not thread safe and the items are stored in direct memory.
+ *
+ * <h3>Algorithm</h3>
+ *
+ * <p>This is a <b>binary min-heap</b> stored in a flat array, where each heap node occupies
+ * 3 consecutive longs (the tuple). The children of the node at index {@code i} are at
+ * {@code 2i + 1} and {@code 2i + 2}; the parent of node {@code i} is at {@code (i - 1) / 2}.
+ *
+ * <p>Both {@code siftUp} (on insert) and {@code siftDown} (on remove) use the
+ * <b>hole-based</b> (also called "bottom-up" or "Floyd's") optimization: instead of swapping
+ * the displaced element with its parent/child at each level, the displaced values are held in
+ * local variables (registers) and written only once at the final position. This reduces the
+ * number of array writes per sift layer from 6 (swap: 3 reads + 3 writes on each side) to 3
+ * (one directional write), and avoids re-reading the displaced element from the array on every
+ * comparison.
+ *
+ * <p>Comparison is lexicographic on (n1, n2, n3), using {@code Long.compare} at each level.
+ *
+ * @see <a href="https://en.wikipedia.org/wiki/Heapsort#Bottom-up_heapsort">Bottom-up heapsort
+ * (Wikipedia)</a>
*/
public class TripleLongPriorityQueue implements AutoCloseable {
private static final int DEFAULT_INITIAL_CAPACITY = 16;
@@ -94,8 +113,7 @@ public class TripleLongPriorityQueue implements AutoCloseable {
array.increaseCapacity();
}
- put(tuplesCount, n1, n2, n3);
- siftUp(tuplesCount);
+ siftUp(tuplesCount, n1, n2, n3);
++tuplesCount;
}
@@ -134,9 +152,17 @@ public class TripleLongPriorityQueue implements AutoCloseable {
*/
public void pop() {
checkArgument(tuplesCount != 0);
- swap(0, tuplesCount - 1);
- tuplesCount--;
- siftDown(0);
+
+ if (--tuplesCount == 0) {
+ return;
+ }
+
+ long lastBase = tuplesCount * ITEMS_COUNT;
+ long n1 = array.readLong(lastBase);
+ long n2 = array.readLong(lastBase + 1);
+ long n3 = array.readLong(lastBase + 2);
+
+ siftDown(0, n1, n2, n3);
shrinkCapacity();
}
@@ -188,81 +214,99 @@ public class TripleLongPriorityQueue implements AutoCloseable {
}
}
- private void siftUp(long tupleIdx) {
+ private void siftUp(long tupleIdx, long n1, long n2, long n3) {
+ long idx = tupleIdx * ITEMS_COUNT;
+
while (tupleIdx > 0) {
- long parentIdx = (tupleIdx - 1) / 2;
- if (compare(tupleIdx, parentIdx) >= 0) {
+ long parentIdx = (tupleIdx - 1) >>> 1;
+ long parentBase = parentIdx * ITEMS_COUNT;
+
+ long p0 = array.readLong(parentBase);
+ long p1 = array.readLong(parentBase + 1);
+ long p2 = array.readLong(parentBase + 2);
+
+ if (compareTuple(n1, n2, n3, p0, p1, p2) >= 0) {
break;
}
- swap(tupleIdx, parentIdx);
+ array.writeLong(idx, p0);
+ array.writeLong(idx + 1, p1);
+ array.writeLong(idx + 2, p2);
+
tupleIdx = parentIdx;
+ idx = parentBase;
}
+
+ array.writeLong(idx, n1);
+ array.writeLong(idx + 1, n2);
+ array.writeLong(idx + 2, n3);
}
- private void siftDown(long tupleIdx) {
- long half = tuplesCount / 2;
+ private void siftDown(long tupleIdx, long val0, long val1, long val2) {
+ long half = tuplesCount >>> 1;
+
+ long idx = tupleIdx * ITEMS_COUNT;
+
while (tupleIdx < half) {
- long left = 2 * tupleIdx + 1;
- long right = 2 * tupleIdx + 2;
+ long left = (tupleIdx << 1) + 1;
+ long right = left + 1;
- long swapIdx = tupleIdx;
+ long child = left;
+ long childBase = left * ITEMS_COUNT;
- if (compare(tupleIdx, left) > 0) {
- swapIdx = left;
- }
+ long child0 = array.readLong(childBase);
+ long child1 = array.readLong(childBase + 1);
+ long child2 = array.readLong(childBase + 2);
- if (right < tuplesCount && compare(swapIdx, right) > 0) {
- swapIdx = right;
- }
+ if (right < tuplesCount) {
+ long rightBase = right * ITEMS_COUNT;
- if (swapIdx == tupleIdx) {
- return;
- }
+ long right0 = array.readLong(rightBase);
+ long right1 = array.readLong(rightBase + 1);
+ long right2 = array.readLong(rightBase + 2);
- swap(tupleIdx, swapIdx);
- tupleIdx = swapIdx;
- }
- }
+ if (compareTuple(right0, right1, right2, child0, child1, child2) < 0) {
- private void put(long tupleIdx, long n1, long n2, long n3) {
- long idx = tupleIdx * ITEMS_COUNT;
- array.writeLong(idx, n1);
- array.writeLong(idx + 1, n2);
- array.writeLong(idx + 2, n3);
- }
+ child = right;
+ childBase = rightBase;
- private int compare(long tupleIdx1, long tupleIdx2) {
- long idx1 = tupleIdx1 * ITEMS_COUNT;
- long idx2 = tupleIdx2 * ITEMS_COUNT;
+ child0 = right0;
+ child1 = right1;
+ child2 = right2;
+ }
+ }
- int c1 = Long.compare(array.readLong(idx1), array.readLong(idx2));
- if (c1 != 0) {
- return c1;
- }
+ if (compareTuple(val0, val1, val2, child0, child1, child2) <= 0) {
+ break;
+ }
+
+ array.writeLong(idx, child0);
+ array.writeLong(idx + 1, child1);
+ array.writeLong(idx + 2, child2);
- int c2 = Long.compare(array.readLong(idx1 + 1), array.readLong(idx2 + 1));
- if (c2 != 0) {
- return c2;
+ tupleIdx = child;
+ idx = childBase;
}
- return Long.compare(array.readLong(idx1 + 2), array.readLong(idx2 + 2));
+ array.writeLong(idx, val0);
+ array.writeLong(idx + 1, val1);
+ array.writeLong(idx + 2, val2);
}
- private void swap(long tupleIdx1, long tupleIdx2) {
- long idx1 = tupleIdx1 * ITEMS_COUNT;
- long idx2 = tupleIdx2 * ITEMS_COUNT;
+ private static int compareTuple(
+ long a0, long a1, long a2,
+ long b0, long b1, long b2) {
- long tmp1 = array.readLong(idx1);
- long tmp2 = array.readLong(idx1 + 1);
- long tmp3 = array.readLong(idx1 + 2);
+ int c = Long.compare(a0, b0);
+ if (c != 0) {
+ return c;
+ }
- array.writeLong(idx1, array.readLong(idx2));
- array.writeLong(idx1 + 1, array.readLong(idx2 + 1));
- array.writeLong(idx1 + 2, array.readLong(idx2 + 2));
+ c = Long.compare(a1, b1);
+ if (c != 0) {
+ return c;
+ }
- array.writeLong(idx2, tmp1);
- array.writeLong(idx2 + 1, tmp2);
- array.writeLong(idx2 + 2, tmp3);
+ return Long.compare(a2, b2);
}
}
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java
index 79e5fc18d72..e2df71520fe 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueTest.java
@@ -20,8 +20,12 @@ package org.apache.pulsar.common.util.collections;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import java.util.Comparator;
+import java.util.PriorityQueue;
+import java.util.Random;
import org.testng.annotations.Test;
public class TripleLongPriorityQueueTest {
@@ -197,4 +201,47 @@ public class TripleLongPriorityQueueTest {
pq.add(i, i, i);
}
}
+
+ @Test
+ public void testDifferentialRandomPriorityQueue() {
+ Comparator<long[]> cmp = Comparator.comparingLong((long[] t) -> t[0])
+ .thenComparingLong(t -> t[1])
+ .thenComparingLong(t -> t[2]);
+
+ for (int trial = 0; trial < 10; trial++) {
+ Random rng = new Random(42 + trial);
+ PriorityQueue<long[]> oracle = new PriorityQueue<>(cmp);
+ try (TripleLongPriorityQueue pq = new TripleLongPriorityQueue()) {
+ int ops = 10_000 + rng.nextInt(10_000);
+ for (int i = 0; i < ops; i++) {
+ boolean doAdd = oracle.isEmpty() || rng.nextBoolean();
+ if (doAdd) {
+ // ~10% chance of same-prefix (small n1 range) to exercise tie-breaking
+ long n1 = rng.nextInt(100) < 10 ? rng.nextLong(20) : rng.nextLong(1_000_000);
+ long n2 = rng.nextLong(100);
+ long n3 = rng.nextLong(1_000_000);
+ oracle.add(new long[]{n1, n2, n3});
+ pq.add(n1, n2, n3);
+ } else {
+ long[] expected = oracle.poll();
+ assertNotNull(expected);
+ assertEquals(pq.peekN1(), expected[0], "n1 mismatch at op " + i);
+ assertEquals(pq.peekN2(), expected[1], "n2 mismatch at op " + i);
+ assertEquals(pq.peekN3(), expected[2], "n3 mismatch at op " + i);
+ pq.pop();
+ }
+ assertEquals(pq.size(), oracle.size(), "size mismatch at op " + i);
+ }
+ // drain remaining
+ while (!oracle.isEmpty()) {
+ long[] expected = oracle.poll();
+ assertEquals(pq.peekN1(), expected[0]);
+ assertEquals(pq.peekN2(), expected[1]);
+ assertEquals(pq.peekN3(), expected[2]);
+ pq.pop();
+ }
+ assertTrue(pq.isEmpty());
+ }
+ }
+ }
}