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/10/19 17:54:12 UTC
(pulsar) branch master updated: [improve][ml] Don't visit a ledger's other single-entry pending reads for a read of a single entry (#27005)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 3164faeeb0a6 [improve][ml] Don't visit a ledger's other single-entry pending reads for a read of a single entry (#27005)
3164faeeb0a6 is described below
commit 3164faeeb0a6ad17be876c25cf5c234783185700
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Oct 19 20:54:05 2026 +0300
[improve][ml] Don't visit a ledger's other single-entry pending reads for a read of a single entry (#27005)
---
.../mledger/impl/cache/PendingReadsManager.java | 48 ++++++++++++++++------
.../impl/cache/PendingReadsManagerTest.java | 33 +++++++++++++++
2 files changed, 69 insertions(+), 12 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManager.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManager.java
index d5a29fe1c442..d3719db70694 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManager.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManager.java
@@ -26,7 +26,6 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.IntSupplier;
import lombok.CustomLog;
@@ -89,8 +88,32 @@ public class PendingReadsManager {
.register();
private final RangeEntryCacheImpl rangeEntryCache;
- private final ConcurrentHashMap<Long, ConcurrentHashMap<PendingReadKey, PendingRead>> cachedPendingReads =
- new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, LedgerPendingReads> cachedPendingReads = new ConcurrentHashMap<>();
+
+ /**
+ * A ledger's pending reads, and among them those of more than one entry. A read of a single entry, such as each
+ * read of a dispatcher's replay, can only match a pending read of the same entry or be included in a read of more
+ * entries, so that it doesn't visit the ledger's other single-entry reads: with hundreds of them pending, a full
+ * visit for each made a replay's reads quadratic.
+ */
+ private static final class LedgerPendingReads {
+ final ConcurrentHashMap<PendingReadKey, PendingRead> all = new ConcurrentHashMap<>();
+ final ConcurrentHashMap<PendingReadKey, PendingRead> multiEntry = new ConcurrentHashMap<>();
+
+ void put(PendingReadKey key, PendingRead read) {
+ all.put(key, read);
+ if (key.size() > 1) {
+ multiEntry.put(key, read);
+ }
+ }
+
+ void remove(PendingReadKey key, PendingRead read) {
+ all.remove(key, read);
+ if (key.size() > 1) {
+ multiEntry.remove(key, read);
+ }
+ }
+ }
public PendingReadsManager(RangeEntryCacheImpl rangeEntryCache) {
this.rangeEntryCache = rangeEntryCache;
@@ -143,10 +166,10 @@ public class PendingReadsManager {
}
}
- private FindPendingReadOutcome findPendingRead(PendingReadKey key, ConcurrentMap<PendingReadKey,
- PendingRead> ledgerCache, AtomicBoolean created) {
+ private FindPendingReadOutcome findPendingRead(PendingReadKey key, LedgerPendingReads ledgerCache,
+ AtomicBoolean created) {
synchronized (ledgerCache) {
- PendingRead existing = ledgerCache.get(key);
+ PendingRead existing = ledgerCache.all.get(key);
if (existing != null) {
COUNT_PENDING_READS_MATCHED.inc(key.size());
COUNT_ENTRIES_NOTREAD_FROM_BK.inc(key.size());
@@ -156,7 +179,9 @@ public class PendingReadsManager {
FindPendingReadOutcome foundButMissingSomethingOnRight = null;
FindPendingReadOutcome foundButMissingSomethingOnBoth = null;
- for (Map.Entry<PendingReadKey, PendingRead> entry : ledgerCache.entrySet()) {
+ // a single entry is either the key of a pending read, or included in a read of more entries
+ Map<PendingReadKey, PendingRead> candidates = key.size() == 1 ? ledgerCache.multiEntry : ledgerCache.all;
+ for (Map.Entry<PendingReadKey, PendingRead> entry : candidates.entrySet()) {
PendingReadKey entryKey = entry.getKey();
if (entryKey.includes(key)) {
COUNT_PENDING_READS_MATCHED_INCLUDED.inc(key.size());
@@ -212,7 +237,7 @@ public class PendingReadsManager {
private class PendingRead {
final PendingReadKey key;
- final ConcurrentMap<PendingReadKey, PendingRead> ledgerCache;
+ final LedgerPendingReads ledgerCache;
final List<ReadEntriesCallbackWithContext> listeners = new ArrayList<>(1);
PendingReadState state = PendingReadState.INITIALISED;
@@ -222,8 +247,7 @@ public class PendingReadsManager {
COMPLETED
}
- public PendingRead(PendingReadKey key,
- ConcurrentMap<PendingReadKey, PendingRead> ledgerCache) {
+ public PendingRead(PendingReadKey key, LedgerPendingReads ledgerCache) {
this.key = key;
this.ledgerCache = ledgerCache;
}
@@ -337,8 +361,8 @@ public class PendingReadsManager {
IntSupplier expectedReadCount, final AsyncCallbacks.ReadEntriesCallback callback, Object ctx) {
final PendingReadKey key = new PendingReadKey(firstEntry, lastEntry);
- ConcurrentMap<PendingReadKey, PendingRead> pendingReadsForLedger =
- cachedPendingReads.computeIfAbsent(lh.getId(), (l) -> new ConcurrentHashMap<>());
+ LedgerPendingReads pendingReadsForLedger =
+ cachedPendingReads.computeIfAbsent(lh.getId(), (l) -> new LedgerPendingReads());
boolean listenerAdded = false;
while (!listenerAdded) {
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManagerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManagerTest.java
index 0d11958cdfa2..27b62d76b993 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManagerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/PendingReadsManagerTest.java
@@ -28,6 +28,7 @@ import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
import io.opentelemetry.api.OpenTelemetry;
@@ -288,6 +289,38 @@ public class PendingReadsManagerTest {
}
+ @Test
+ public void singleEntryReadsShareAnIncludingReadAndNotEachOther() throws Exception {
+ IntSupplier expectedReadCount = () -> 0;
+ PreparedReadFromStorage range = prepareReadFromStorage(lh, rangeEntryCache, 100, 199, expectedReadCount);
+ PreparedReadFromStorage single1 = prepareReadFromStorage(lh, rangeEntryCache, 300, 300, expectedReadCount);
+ PreparedReadFromStorage single2 = prepareReadFromStorage(lh, rangeEntryCache, 305, 305, expectedReadCount);
+
+ CapturingReadEntriesCallback rangeCallback = new CapturingReadEntriesCallback();
+ pendingReadsManager.readEntries(lh, 100, 199, NO_MAX_SIZE_LIMIT, expectedReadCount, rangeCallback, CTX);
+ // as a dispatcher's replay reads single entries: one inside the pending range read, two outside of it
+ CapturingReadEntriesCallback included = new CapturingReadEntriesCallback();
+ pendingReadsManager.readEntries(lh, 150, 150, NO_MAX_SIZE_LIMIT, expectedReadCount, included, CTX2);
+ CapturingReadEntriesCallback separate1 = new CapturingReadEntriesCallback();
+ pendingReadsManager.readEntries(lh, 300, 300, NO_MAX_SIZE_LIMIT, expectedReadCount, separate1, CTX);
+ CapturingReadEntriesCallback separate2 = new CapturingReadEntriesCallback();
+ pendingReadsManager.readEntries(lh, 305, 305, NO_MAX_SIZE_LIMIT, expectedReadCount, separate2, CTX);
+
+ range.storageReadCompleted();
+ single1.storageReadCompleted();
+ single2.storageReadCompleted();
+
+ for (CapturingReadEntriesCallback callback : List.of(rangeCallback, included, separate1, separate2)) {
+ callback.get();
+ }
+ verifyRange(rangeCallback.entries, 100, 199);
+ // the included single entry came from the range read
+ verifyRange(included.entries, 150, 150);
+ assertNull(entryRangeReadCount.get(Pair.of(150L, 150L)));
+ verifyRange(separate1.entries, 300, 300);
+ verifyRange(separate2.entries, 305, 305);
+ }
+
@Test
public void simpleConcurrentReadIncluding() throws Exception {