From a012dab401e96b1d759a5a9e68887dcb744577b3 Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Fri, 12 Jun 2026 18:55:32 +0800 Subject: [PATCH 1/3] [improve][broker] Optimize TripleLongPriorityQueue heap operations --- .../collections/TripleLongPriorityQueue.java | 138 +++++++++++------- 1 file changed, 82 insertions(+), 56 deletions(-) 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 e75ae21ab8518..d8d7491804af5 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 @@ -94,8 +94,10 @@ public void add(long n1, long n2, long n3) { array.increaseCapacity(); } - put(tuplesCount, n1, n2, n3); - siftUp(tuplesCount); + array.writeLong(arrayIdx, n1); + array.writeLong(arrayIdx + 1, n2); + array.writeLong(arrayIdx + 2, n3); + siftUp(tuplesCount, n1, n2, n3); ++tuplesCount; } @@ -134,9 +136,15 @@ public long peekN3() { */ public void pop() { checkArgument(tuplesCount != 0); - swap(0, tuplesCount - 1); - tuplesCount--; - siftDown(0); + + long lastIdx = --tuplesCount; + long lastBase = lastIdx * 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 +196,99 @@ private void shrinkCapacity() { } } - 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); } } From df406da131b35046b9c8bf85810d489634b4b39a Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Mon, 15 Jun 2026 12:16:22 +0800 Subject: [PATCH 2/3] Address comment --- .../TripleLongPriorityQueueBenchmark.java | 168 ++++++++++++++++++ .../collections/TripleLongPriorityQueue.java | 9 +- .../TripleLongPriorityQueueTest.java | 47 +++++ 3 files changed, 219 insertions(+), 5 deletions(-) create mode 100644 microbench/src/main/java/org/apache/pulsar/common/util/collections/TripleLongPriorityQueueBenchmark.java 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 0000000000000..1e50ab600bfae --- /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. + * + *

Three scenarios matching real usage: + *

+ * + *

Build and run: + * 

+ * ./gradlew :microbench:shadowJar
+ * java -jar microbench/build/libs/microbench-*-benchmarks.jar ".*TripleLongPriorityQueue.*"
+ *
+ */ +@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 d8d7491804af5..bfb45503cbaf5 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 @@ -94,9 +94,6 @@ public void add(long n1, long n2, long n3) { array.increaseCapacity(); } - array.writeLong(arrayIdx, n1); - array.writeLong(arrayIdx + 1, n2); - array.writeLong(arrayIdx + 2, n3); siftUp(tuplesCount, n1, n2, n3); ++tuplesCount; } @@ -137,9 +134,11 @@ public long peekN3() { public void pop() { checkArgument(tuplesCount != 0); - long lastIdx = --tuplesCount; - long lastBase = lastIdx * ITEMS_COUNT; + 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); 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 79e5fc18d72d0..e2df71520fe76 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 @@ 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 @@ private void triggerScaleOut(int initialCapacity, TripleLongPriorityQueue pq) { pq.add(i, i, i); } } + + @Test + public void testDifferentialRandomPriorityQueue() { + Comparator 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 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()); + } + } + } } From fa65f2546f8907720f5d4c16360d43e8c3f4c84e Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Mon, 15 Jun 2026 17:16:46 +0800 Subject: [PATCH 3/3] Add algorithm doc --- .../collections/TripleLongPriorityQueue.java | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) 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 bfb45503cbaf5..cd878c6428459 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 @@ * Provides a priority-queue implementation specialized on items composed by 3 longs. * *

This class is not thread safe and the items are stored in direct memory. + * + *

Algorithm

+ * + *

This is a binary min-heap 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}. + * + *

Both {@code siftUp} (on insert) and {@code siftDown} (on remove) use the + * hole-based (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. + * + *

Comparison is lexicographic on (n1, n2, n3), using {@code Long.compare} at each level. + * + * @see Bottom-up heapsort + * (Wikipedia) */ public class TripleLongPriorityQueue implements AutoCloseable { private static final int DEFAULT_INITIAL_CAPACITY = 16;