[improve][broker] Optimize TripleLongPriorityQueue heap operations

[nodece]

Motivation

TripleLongPriorityQueue is the core data structure for Pulsar's delayed message delivery, storing (deliverAt, ledgerId, entryId) tuples in a min-heap backed by SegmentedLongArray (direct memory, 16MB segments). CPU profiling shows siftUp/siftDown consuming ~37% of hot thread time during delayed message processing.

The original implementation has two performance issues:

1. Redundant readLong calls: compare(tupleIdx1, tupleIdx2) reads 6 longs per comparison, but the current node's values are already known from the prior iteration — they're re-read from SegmentedLongArray unnecessarily.
2. Swap-based sift: Each heap layer swap performs 6 readLong + 6 writeLong, but a hole-based approach only needs 3 writeLong per layer.

Each readLong on SegmentedLongArray incurs 3 layers of indirection (division + ArrayList.get() + ByteBuf.getLong()), so reducing readLong calls directly translates to throughput gains.

Modifications

Rewrote siftUp and siftDown using the hole-based algorithm with register-cached values:

• siftUp(tupleIdx, n1, n2, n3): Passes the inserted element's values as parameters. Each layer only reads the parent's 3 longs (was: 6 longs for self + parent), then writes the parent down. The displaced value lives in local variables and is written once at the final position.
• siftDown(tupleIdx, val0, val1, val2): Pop reads the last tuple and passes it directly. Each layer reads left + right child's 3 longs each (was: 6 longs for self + left + right), promotes the smaller child, and writes the displaced value at the leaf.
• add(): Removed redundant pre-write — siftUp writes the tuple at its final position.
• pop(): Reads last tuple, passes to siftDown(0, n1, n2, n3). Early return for single-element case.
• Removed compare(), swap(), put() methods.

Per-layer readLong reduction:

Operation	Before	After
siftUp per layer	12	3
siftDown per layer	24	6

This is a 4x reduction in SegmentedLongArray readLong calls per sift layer. Since each readLong involves 3 levels of indirection (segment lookup + ArrayList.get + ByteBuf.getLong), this reduces both instruction count and cache pressure.

Benchmark Results

TestNG micro-benchmark (add-then-pop, single-threaded):

Dataset size	Before	After	Improvement
50K	38,690 us	34,220 us	11.5%
500K	525,517 us	461,174 us	12.2%
2M	2,136,791 us	1,704,437 us	20.2%

JMH benchmark (add-then-pop, @Fork(2), @Warmup(3), @Measurement(5)):

Dataset size	Before	After	Improvement
50K	38,672 us	24,556 us	36.5%
500K	505,803 us	316,047 us	37.5%
2M	2,376,329 us	1,401,276 us	41.0%

The add-then-pop pattern (bulk recovery scenario) shows significant improvement because the heap stays in CPU cache during the pop phase, and the reduced readLong calls directly translate to fewer cache line fetches.

JMH realistic workload benchmark (interleaved batch add/pop, simulating steady-state delayed delivery):

The interleaved pattern — where messages are added in small batches (~500) and popped in small batches, matching Pulsar's actual getScheduledMessages behavior — shows roughly equivalent performance between old and new. This is expected: in the interleaved pattern, the heap is smaller and cache miss patterns are similar between both algorithms. The benchmark's data generation and Blackhole.consume overhead also dilute the algorithmic difference.

The code simplification (removing compare(), swap(), put()) is the primary benefit of this change, with performance improvement as a secondary benefit that manifests most clearly in bulk-operation scenarios.

Further optimization

The remaining performance gap is dominated by SegmentedLongArray's per-readLong indirection cost. Replacing SegmentedLongArray with a flat long[] eliminates the segment lookup entirely — each readLong becomes a single array load with hardware prefetcher support. Benchmark comparison shows long[]-backed implementation is ~3x faster than SegmentedLongArray at the same heap operations. This could be considered as a follow-up if the direct memory trade-off (GC pressure vs. off-heap) is acceptable for the delayed delivery use case.

Testing

• All existing tests pass including testCompareWithSamePrefix (validates correct ordering on tuples with shared prefixes).
• Added testDifferentialRandomPriorityQueue — randomized differential test against java.util.PriorityQueue<long[]> with 10 trials, ~15K interleaved add/pop per trial, 10% same-prefix tuples.
• Added JMH benchmark in microbench module with three workload scenarios.

[dao-jun]

LGTM

[lhotari]

Didn't go through deeply, but one question about swapping.

One possible optimization for swapping would be to read 3 longs in one shot and just writing it to the other location in one shot, using a ByteBuf directly. For comparisons, reading and comparing the keys alone should be sufficient.

[lhotari]

Benchmark comparison shows long[]-backed implementation is ~3x faster than SegmentedLongArray at the same heap operations. This could be considered as a follow-up if the direct memory trade-off (GC pressure vs. off-heap) is acceptable for the delayed delivery use case.

could you add the benchmarks to the microbench module?

[lhotari]

I performed a local review with Claude Code and these were the findings:

1. [QUALITY] Dead writes in add()
   The three array.writeLong(arrayIdx, …) calls before siftUp(tuplesCount, n1, n2, n3) are redundant. siftUp never reads the slot at tupleIdx (it only reads parents) and unconditionally writes (n1, n2, n3) at the final hole position — which equals arrayIdx when no movement occurs. Dropping them saves 3 writeLong calls per add(), exactly the kind of saving this PR is after. (SegmentedLongArray.writeLong is a plain setLong with no size/writer-index tracking, so nothing depends on the pre-write.)

2. [QUALITY] Single-element pop() does wasted I/O
   pop() reads the last (= only) tuple and siftDown writes it back to slot 0, which is now beyond tuplesCount. Harmless (the old code's swap(0, 0) was worse), but an early return when tuplesCount == 0 after the decrement would skip 3 reads + 3 writes. Very minor.

3. [QUALITY] No test changes for a heap-algorithm rewrite
   Existing coverage (testLargeQueue, testCompareWithSamePrefix) does exercise ordering, but for a from-scratch sift rewrite a randomized differential test against java.util.PriorityQueue<long[]> (interleaved add/pop, duplicates, same-prefix tuples) would be cheap insurance. Worth considering, not blocking.

[nodece]

Benchmark comparison shows long[]-backed implementation is ~3x faster than SegmentedLongArray at the same heap operations. This could be considered as a follow-up if the direct memory trade-off (GC pressure vs. off-heap) is acceptable for the delayed delivery use case.

could you add the benchmarks to the microbench module?

I will make a new PR to implement this.

[lhotari]

good work @nodece

[lhotari]

@nodece please add the algorithm description to the PR description as well.