[improve][broker] Optimize TripleLongPriorityQueue heap operations

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();
+                }
+            }
+        }
+    }
+}

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.
  *
  * <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 void add(long n1, long n2, long n3) {
             array.increaseCapacity();
         }
 
-        put(tuplesCount, n1, n2, n3);
-        siftUp(tuplesCount);
+        siftUp(tuplesCount, n1, n2, n3);
         ++tuplesCount;
     }
 
@@ -134,9 +152,17 @@ public long peekN3() {
      */
     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 @@ 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);
     }
 }