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 e75ae21ab8518..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. + * + *
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;
@@ -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);
}
}
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