You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "lhotari (via GitHub)" <gi...@apache.org> on 2026/10/02 21:35:45 UTC
[PR] [improve][ml] Store the broker entry cache in pages of consecutive entry IDs [pulsar]
lhotari opened a new pull request, #26821:
URL: https://github.com/apache/pulsar/pull/26821
### Motivation
The broker's entry cache (`RangeCache`, one per managed ledger) kept every cached entry in a `ConcurrentSkipListMap<Position, RangeCacheEntryWrapper>`. A managed ledger inserts every entry that it adds to the cache on its ordered executor thread. So each insert searched a skip list of all the topic's cached entries, comparing positions through the `Position` interface level by level, even though the new entry always goes at the tail.
With many publishers to one topic at a high rate, that thread is the broker's limit. In a profile of the Pulsar Performance Testing Framework's max-rate IoT telemetry scenario at 128 B, the managed-ledger thread (`BookKeeperClientWorker-OrderedExecutor-N-0`) was 93–95 % busy. The entry cache insert was 18 % of its time, with `ConcurrentSkipListMap.putIfAbsent` at about 7 %.
The cache's access pattern fits a simpler structure. Entries are inserted in runs of consecutive entry IDs: a managed ledger adds its entries in order at the tail, and a catch-up read from storage inserts the consecutive entries that it read, at any earlier position, when other cursors are expected to read them (single-entry reads aren't cached). Entries are read in ranges within a ledger, and removed in ranges or by the eviction.
### Modifications
- **Pages of consecutive entry IDs:** `RangeCache` keeps pages of 64 consecutive entry IDs of one ledger in the `ConcurrentSkipListMap`, keyed by the ledger ID and the page index (`entryId >> 6`). Each page holds its entries in an `AtomicReferenceArray`.
- **Insert:** a `RangeCache.Inserter` remembers the page of its previous insert, so an insert in order usually costs a compare-and-set of an array slot, with no skip-list search; the skip list of pages is searched once for every 64 entries. An entry at any other position is inserted too, after a lookup of its page.
- **Inserters:** `RangeEntryCacheImpl` inserts the managed ledger's added entries, which arrive in order on its executor, with one inserter, and each read from storage, usually a batch of 100 consecutive entries, inserts its entries with a new one through the batch method `insert(List<? extends Entry>, boolean)`, which replaces `insert(Entry, boolean)`. A catch-up read and the managed ledger's adds therefore don't displace each other's pages, and no field is shared between their threads.
- **Lookup:** a page lookup and an array read.
- **Range reads and removals:** visit the pages of the range in order and the slots within each page.
- **Removing empty pages:** each page counts its entries. A page that loses its last entry is sealed and removed from the map. An insert reserves its slot in the count first, so it never adds an entry to a sealed page; when it finds one, it retries with a new page.
- **Unchanged:** the `RangeCacheEntryWrapper` locking, the identity checks that make sure each entry is removed and released exactly once, the removal queue and the eviction. Range iteration matches a slot's wrapper by the slot's ledger and entry ID instead of by the identity of the map's key instance.
- **Memory:** a full page takes about 6 bytes per entry, against about 32 bytes for a skip-list node and its share of the index nodes. A page with a single entry takes about 380 bytes. Since entries are inserted in runs, only the pages at the ends of a run are partly filled.
- **Benchmarks:**
- `RangeCacheInsertBenchmark`: a new JMH benchmark of the managed-ledger thread's pattern. It inserts at the tail, trims the head in batches as the cursors' progress does, and drains the removal queue of the removed entries as the eviction does.
- `RangeCacheCatchUpInsertBenchmark`: a new JMH benchmark of tail inserts on one thread while another inserts catch-up runs of 100 entries at random earlier positions.
- `CacheEvictionCycleBenchmark` inserts each batch with an inserter, as a read from storage does.
### Measurements
The [Pulsar Performance Testing Framework](https://github.com/apache/pulsar/tree/master/tests/performance)'s `iot-telemetry-max-rate` scenario: 500 gateways publish unbatched messages to one persistent topic as fast as they can, and 20 consumers receive them. The payload sizes were 128 B (the scenario's default), and 1 KB, 8 KB and 64 KB with larger measured message counts.
- **Host:** Intel i9-9980HK, 8 cores / 16 threads, fixed at 2.4 GHz with the framework's TuneD profile.
- **Both sides:** the bookies' journal on a tmpfs, and the write barriers of the containers' file system off (#26800), so that the laptop's disk isn't the limit. The broker's default `numIOThreads`.
- **Builds:** the same commit, with and without this change. The payload comparison below ran the first commit of this PR, whose single remembered page served the adds equally; the final code's 128 B result is under it.
- **Runs:** 3 runs per side and payload, interleaved.
- **Correctness:** every run reported no duplicates, ordering violations or invalid messages.
| Max rate, means of 3 | Master | This change | Change |
|---|---|---|---|
| 128 B throughput | 107,395 msg/s (104.4k–109.2k) | 113,818 msg/s (113.6k–114.0k) | +6.0 % |
| 128 B broker CPU per million messages | 55.5 s (54.0–57.5) | 51.8 s (51.4–52.2) | −6.7 % |
| 128 B publish p99 | 1,069 ms (1,026–1,117) | 1,016 ms (995–1,030) | −5.0 % |
| 1 KB throughput | 101,286 msg/s (100.7k–102.3k) | 107,578 msg/s (105.8k–108.5k) | +6.2 % |
| 1 KB broker CPU per million messages | 57.2 s (56.9–57.5) | 52.7 s (51.7–53.9) | −7.9 % |
| 1 KB publish p99 | 601 ms (585–619) | 584 ms (565–605) | −2.8 % |
| 8 KB throughput | 52,843 msg/s (52.7k–53.1k) | 52,047 msg/s (51.3k–53.0k) | −1.5 % |
| 64 KB throughput | 7,952 msg/s (7.88k–8.03k) | 7,918 msg/s (7.88k–7.98k) | −0.4 % |
- **128 B and 1 KB:** these sizes are limited by the managed-ledger thread. Throughput rises by about 6 %, the throughput ranges don't overlap, and the broker uses 7–8 % less CPU per message.
- **8 KB and 64 KB:** no difference beyond the spread between runs. These sizes are limited by the bookie's write-cache flush thread, not by the broker.
- **Latency:** at max rate, the publish latency is mostly the queueing of the gateways' 100,000 outstanding messages, so it follows the throughput.
- **Profiled pair at 128 B** (the final code, async-profiler, one run per side, 108,757 and 116,103 msg/s): the entry cache insert falls from 18.7 % to 11.9 % of the managed-ledger thread's CPU samples. The skip list's insert (`ConcurrentSkipListMap.doPut`, `addIndices` and `Position.compareTo`, 5.1 % of the thread) is gone; the rest of the insert is the wrapper and its lock, the page's slot and count, the removal queue and the cached entry's duplicate. The thread's largest items are now `PersistentTopic.addComplete` (15.9 %), the BookKeeper channel write (12.6 %), BookKeeper's digest and copy (9.8 %) and deduplication (9.7 %).
- **The final code at 128 B** (3 more interleaved runs per side): 109,343 msg/s (108.4k–110.7k) on master against 117,663 msg/s (116.9k–118.2k), +7.6 %, with broker CPU per million messages 54.2 s against 49.3 s, −9.0 %.
The summary chart shows the change at each payload size. The whiskers span the change between the slowest and the fastest runs of the two sides.
<a href="https://github.com/user-attachments/assets/0cc8a12a-4d3e-4168-8384-3372678fcb33"><img src="https://github.com/user-attachments/assets/0cc8a12a-4d3e-4168-8384-3372678fcb33" alt="Change in max-rate throughput, broker CPU per message and publish latency p99 by payload size" width="800"></a>
The A/B charts compare each side's median run: at 128 B, master at 108,585 msg/s and this change at 113,830 msg/s; at 1 KB, master at 100,892 msg/s and this change at 108,475 msg/s. Both runs of a chart share its axes.
**Throughput at 128 B, master (A) and this change (B)**
<a href="https://github.com/user-attachments/assets/47f7cec7-2716-4339-bfd7-aa396f1ea410"><img src="https://github.com/user-attachments/assets/47f7cec7-2716-4339-bfd7-aa396f1ea410" alt="Throughput over time at 128 B, master above and this change below" width="800"></a>
**Latency by percentile at 128 B, master (A) and this change (B)**
<a href="https://github.com/user-attachments/assets/5f090a7d-2010-41b3-bf04-71aad44f6359"><img src="https://github.com/user-attachments/assets/5f090a7d-2010-41b3-bf04-71aad44f6359" alt="Publish and end-to-end latency by percentile at 128 B, master and this change" width="800"></a>
**Throughput at 1 KB, master (A) and this change (B)**
<a href="https://github.com/user-attachments/assets/dd25663c-d931-4b0a-b194-a9bf14730672"><img src="https://github.com/user-attachments/assets/dd25663c-d931-4b0a-b194-a9bf14730672" alt="Throughput over time at 1 KB, master above and this change below" width="800"></a>
**Microbenchmarks** (JMH, the same host at the fixed frequency, average time per operation, final code):
| Benchmark | Master | This change | Change |
|---|---:|---:|---:|
| `RangeCacheInsertBenchmark`, 1,000 cached entries | 616 ns | 350 ns | −43 % |
| `RangeCacheInsertBenchmark`, 100,000 cached entries | 784 ns | 381 ns | −51 % |
| `RangeCacheCatchUpInsertBenchmark`, tail insert | 1,202 ns | 601 ns | −50 % |
| `RangeCacheCatchUpInsertBenchmark`, catch-up run of 100 entries | 93.3 µs | 40.6 µs | −57 % |
| `CacheEvictionCycleBenchmark`, 4 variants | 41.9–566 µs | 27.0–357 µs | −31 % to −45 % |
| `RangeCacheReadBenchmark`, 100 % hits, 100 / 1,000 entries | 4.27 / 40.4 µs | 3.66 / 36.2 µs | −14 % / −10 % |
| `RangeCacheReadBenchmark`, 10 % hits, 100 / 1,000 entries | 0.48 / 4.41 µs | 0.52 / 5.16 µs | +8 % / +17 % |
| `RangeCacheReadBenchmark`, no hits | 16 ns | 24 ns | +8 ns |
A range in which only a few entries are cached is slower, since the read visits the empty slots of its pages at about 0.8 ns each. The broker's cache is usually dense: the entries at the tail of a ledger are all cached as they're added.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
- `RangeCacheTest`:
- `rangesSpanPagesAndLedgers`: reads and removals across page and ledger boundaries, an exclusive end at a page boundary, and a range from before the first ledger, as invalidating consumed ledgers uses.
- `removesPagesWhenTheirEntriesAreEvicted`: eviction removes empty pages and releases the cache's references, and an entry can be added again where a page was removed.
- `concurrentInsertsReadsAndRemovals`: inserts with concurrent range reads, lookups, range removals and evictions. Afterwards, every entry has had the cache's reference released exactly once, and no page is left.
- `concurrentTailAndCatchUpInserts`: tail inserts with one inserter while two threads insert catch-up runs of 1–100 entries at random earlier positions, each run with an inserter of its own, with concurrent range reads, removals and evictions. Afterwards, the cache has released the reference of every entry that it inserted exactly once, and of no other entry, and no page is left.
- `inserterAddsToANewPageAfterItsPageWasRemoved`: an inserter whose page lost its last entry and was removed adds the next entry to a new page, and inserts entries before and after its page.
- The whole `managed-ledger` test suite, and the broker's `BrokerEntryCacheTest`, `MinimumBacklogCacheStrategyTest` and managed-ledger interceptor tests pass with the final code, and `./gradlew sanityCheck` compiles every module.
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
This change was prepared with the assistance of Claude Code (claude-opus-5-5); I have reviewed and verified it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Store the broker entry cache in pages of consecutive entry IDs [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26821:
URL: https://github.com/apache/pulsar/pull/26821#discussion_r4172036689
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java:
##########
@@ -315,6 +327,26 @@ public EntryReadCountHandler getReadCountHandler() {
return readCountHandler;
}
+ /**
+ * Takes the expected read count of an entry that was added at this entry's position while this one is cached.
+ * A cached entry's data is immutable and kept, but its expected read count is the latest addition's, such as a
+ * read from storage, which knows how many cursors are expected to read the entry; it overrides the count, also
+ * when it has no expected reads. A cached entry without expected reads takes the latest addition's handler.
+ *
+ * @param latest the read count handler of the latest addition, or null when it has no expected reads
+ */
+ public void updateExpectedReadCount(EntryReadCountHandler latest) {
+ EntryReadCountHandler current = readCountHandler;
+ if (latest == current) {
+ return;
+ }
+ if (current instanceof EntryReadCountHandlerImpl currentImpl) {
+ currentImpl.setExpectedReadCount(latest != null ? latest.getExpectedReadCount() : 0);
Review Comment:
Thanks, good catch. Fixed in 5154503: the cached entry now takes over the latest addition's read count handler instead of copying only its value, so releasing the returned entries counts their reads for the cached entry too, as for a freshly inserted entry. `testCachedEntriesAreEvictableAfterAFailedPartialReadRetriesTheWholeRange` covers the fallback: entry 0 is cached, the read of the missing entry 1 fails, the retry reads both from storage, and after the returned entries are released neither cached entry has expected reads. It fails on the previous head and passes now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Store the broker entry cache in pages of consecutive entry IDs [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #26821:
URL: https://github.com/apache/pulsar/pull/26821#discussion_r4171547439
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java:
##########
@@ -315,6 +327,26 @@ public EntryReadCountHandler getReadCountHandler() {
return readCountHandler;
}
+ /**
+ * Takes the expected read count of an entry that was added at this entry's position while this one is cached.
+ * A cached entry's data is immutable and kept, but its expected read count is the latest addition's, such as a
+ * read from storage, which knows how many cursors are expected to read the entry; it overrides the count, also
+ * when it has no expected reads. A cached entry without expected reads takes the latest addition's handler.
+ *
+ * @param latest the read count handler of the latest addition, or null when it has no expected reads
+ */
+ public void updateExpectedReadCount(EntryReadCountHandler latest) {
+ EntryReadCountHandler current = readCountHandler;
+ if (latest == current) {
+ return;
+ }
+ if (current instanceof EntryReadCountHandlerImpl currentImpl) {
+ currentImpl.setExpectedReadCount(latest != null ? latest.getExpectedReadCount() : 0);
Review Comment:
Could we keep the cached and returned entries on the same read-count handler instead of copying only its value? After a partial-cache read fails and retries the full range, releasing the returned entries decrements `latest`, but the cached count stays at 1 and delays eviction. I reproduced this on the new head; the previous head evicts both entries. Please add a regression test for this fallback path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][ml] Store the broker entry cache in pages of consecutive entry IDs [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui merged PR #26821:
URL: https://github.com/apache/pulsar/pull/26821
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org