You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/09/10 23:56:33 UTC
(pulsar) 11/18: [improve][ml] Avoid capturing cache insertion arguments (#26524)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c655dc4fcbeb9036edcc9bcd6e75b09abff58552
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Sep 10 20:32:16 2026 +0300
[improve][ml] Avoid capturing cache insertion arguments (#26524)
(cherry picked from commit 78688ddce6c918c0390d5a8c423422c5063d47cf)
---
.../bookkeeper/mledger/impl/cache/RangeCache.java | 25 +++---
.../impl/cache/CacheEvictionCycleBenchmark.java | 88 ++++++++++++++++++++++
2 files changed, 102 insertions(+), 11 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCache.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCache.java
index 23cad445c84c..e9d62f2b05ad 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCache.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCache.java
@@ -72,22 +72,25 @@ class RangeCache {
if (!value.matchesPosition(key)) {
throw new IllegalArgumentException("Value '" + value + "' does not match key '" + key + "'");
}
- boolean added = RangeCacheEntryWrapper.withNewInstance(this, key, value, entryLength, newWrapper -> {
- if (entries.putIfAbsent(key, newWrapper) == null && removalQueue.addEntry(newWrapper)) {
- this.size.addAndGet(entryLength);
- return true;
- } else {
- // recycle the new wrapper as it was not used
- newWrapper.recycle();
- return false;
- }
- });
- return added;
+ return RangeCacheEntryWrapper.withNewInstance(this, key, value, entryLength, RangeCache::addEntry);
} finally {
value.release();
}
}
+ private static boolean addEntry(RangeCacheEntryWrapper newWrapper) {
+ // withNewInstance holds the wrapper's write lock while these initialized fields are used.
+ RangeCache cache = newWrapper.rangeCache;
+ if (cache.entries.putIfAbsent(newWrapper.key, newWrapper) == null && cache.removalQueue.addEntry(newWrapper)) {
+ cache.size.addAndGet(newWrapper.size);
+ return true;
+ } else {
+ // recycle the new wrapper as it was not used
+ newWrapper.recycle();
+ return false;
+ }
+ }
+
/**
* Insert to cache with entry length determined directly from the value.
* This method is used in tests.
diff --git a/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/cache/CacheEvictionCycleBenchmark.java b/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/cache/CacheEvictionCycleBenchmark.java
new file mode 100644
index 000000000000..563d49fa9044
--- /dev/null
+++ b/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/cache/CacheEvictionCycleBenchmark.java
@@ -0,0 +1,88 @@
+/*
+ * 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.bookkeeper.mledger.impl.cache;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.mledger.impl.EntryImpl;
+import org.apache.commons.lang3.tuple.Pair;
+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.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+
+/** Measures a complete insertion/eviction cycle, including allocation, using retained real entries. */
+@State(Scope.Thread)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Fork(2)
+public class CacheEvictionCycleBenchmark {
+ @Param({"100", "1000"})
+ public int batchSize;
+
+ @Param({"false", "true"})
+ public boolean requeue;
+
+ private RangeCacheRemovalQueue queue;
+ private RangeCache cache;
+ private EntryImpl[] entries;
+
+ @Setup
+ public void setup() {
+ queue = new RangeCacheRemovalQueue(1, false);
+ cache = new RangeCache(queue);
+ entries = new EntryImpl[batchSize];
+ for (int i = 0; i < batchSize; i++) {
+ entries[i] = EntryImpl.create(0, i, new byte[1], requeue ? 1 : 0);
+ }
+ }
+
+ @Benchmark
+ public long insertAndEvict() {
+ for (EntryImpl entry : entries) {
+ entry.retain();
+ if (!cache.put(entry.getPosition(), entry)) {
+ entry.release();
+ throw new IllegalStateException("Duplicate cache entry");
+ }
+ }
+ Pair<Integer, Long> removed = queue.evictLeastAccessedEntries(batchSize);
+ if (removed.getLeft() != batchSize || !queue.isEmpty()) {
+ throw new IllegalStateException("Incomplete eviction");
+ }
+ return removed.getRight();
+ }
+
+ @TearDown
+ public void tearDown() {
+ cache.clear();
+ for (EntryImpl entry : entries) {
+ entry.release();
+ }
+ }
+}