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/05 18:42:21 UTC
(pulsar) 04/07: [improve][ml] Store the broker entry cache in pages of consecutive entry IDs (#26821)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e539bbd476915fbf3214c8443fabce7ceee89f0a
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sat Oct 3 12:51:25 2026 -0500
[improve][ml] Store the broker entry cache in pages of consecutive entry IDs (#26821)
(cherry picked from commit 2b6becbd69926768869f4ffb3b5929b1c777c991)
---
.../apache/bookkeeper/mledger/impl/EntryImpl.java | 35 +-
.../bookkeeper/mledger/impl/cache/RangeCache.java | 392 +++++++++++++++++----
.../mledger/impl/cache/RangeCacheEntryWrapper.java | 43 +--
.../mledger/impl/cache/RangeEntryCacheImpl.java | 51 ++-
.../mledger/impl/cache/RangeCacheTest.java | 298 ++++++++++++++++
.../impl/cache/RangeEntryCacheImplTest.java | 149 ++++++++
.../impl/cache/CacheEvictionCycleBenchmark.java | 4 +-
.../cache/RangeCacheCatchUpInsertBenchmark.java | 115 ++++++
.../impl/cache/RangeCacheInsertBenchmark.java | 109 ++++++
9 files changed, 1095 insertions(+), 101 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java
index 9fafec3c1dd7..bcfc22c4d978 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java
@@ -24,6 +24,8 @@ import io.netty.buffer.Unpooled;
import io.netty.util.Recycler;
import io.netty.util.Recycler.Handle;
import io.netty.util.ReferenceCounted;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
import lombok.CustomLog;
import lombok.Getter;
import lombok.Setter;
@@ -56,6 +58,16 @@ public final class EntryImpl extends AbstractCASReferenceCounted
private Position position;
ByteBuf data;
private EntryReadCountHandler readCountHandler;
+ private static final VarHandle READ_COUNT_HANDLER;
+
+ static {
+ try {
+ READ_COUNT_HANDLER = MethodHandles.lookup()
+ .findVarHandle(EntryImpl.class, "readCountHandler", EntryReadCountHandler.class);
+ } catch (ReflectiveOperationException e) {
+ throw new ExceptionInInitializerError(e);
+ }
+ }
private boolean decreaseReadCountOnRelease = true;
// Cache readers publish metadata lazily; entry copies must see a fully initialized instance.
@Getter @Setter
@@ -181,7 +193,7 @@ public final class EntryImpl extends AbstractCASReferenceCounted
entry.ledgerId = other.ledgerId;
entry.entryId = other.entryId;
entry.data = other.data.retainedDuplicate();
- entry.readCountHandler = other.readCountHandler;
+ entry.readCountHandler = other.getReadCountHandler();
entry.messageMetadata = other.messageMetadata;
entry.setRefCnt(1);
return entry;
@@ -282,8 +294,9 @@ public final class EntryImpl extends AbstractCASReferenceCounted
@Override
protected void deallocate() {
- if (decreaseReadCountOnRelease && readCountHandler != null) {
- readCountHandler.markRead();
+ EntryReadCountHandler handler = getReadCountHandler();
+ if (decreaseReadCountOnRelease && handler != null) {
+ handler.markRead();
}
// This method is called whenever the ref-count of the EntryImpl reaches 0, so that now we can recycle it
if (onDeallocate != null) {
@@ -312,7 +325,21 @@ public final class EntryImpl extends AbstractCASReferenceCounted
@Override
public EntryReadCountHandler getReadCountHandler() {
- return readCountHandler;
+ // a cached entry can take a later addition's handler, see updateExpectedReadCount
+ return (EntryReadCountHandler) READ_COUNT_HANDLER.getAcquire(this);
+ }
+
+ /**
+ * Takes the read count handler 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, also when it has none. The
+ * cached entry shares the latest addition's handler, as an inserted entry does, so that the reads of the entries
+ * that the addition returned, counted when they're released, count for the cached entry too.
+ *
+ * @param latest the read count handler of the latest addition, or null when it has no expected reads
+ */
+ public void updateExpectedReadCount(EntryReadCountHandler latest) {
+ READ_COUNT_HANDLER.setRelease(this, latest);
}
public void setDecreaseReadCountOnRelease(boolean enabled) {
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 5b84d13f2b2e..726cce1221d9 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
@@ -22,10 +22,11 @@ import io.netty.util.IllegalReferenceCountException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.function.Consumer;
import java.util.function.Function;
import lombok.CustomLog;
@@ -35,17 +36,31 @@ import org.apache.commons.lang3.tuple.Pair;
/**
* Special type of cache where get() and delete() operations can be done over a range of keys.
- * The implementation avoids locks and synchronization by relying on ConcurrentSkipListMap for storing the entries.
- * Since there are no locks, it's necessary to ensure that a single entry in the cache is removed exactly once.
+ *
+ * <p>The entries are stored in pages, each holding up to {@link #PAGE_SIZE} consecutive entry IDs of one ledger in an
+ * array, and the pages are kept in a ConcurrentSkipListMap ordered by ledger ID and page. Entries are inserted in
+ * order: a managed ledger adds its entries at the tail, and a read from storage inserts the consecutive entries that
+ * it read, at any earlier position. Each of them inserts with an {@link Inserter}, which remembers the page of its
+ * previous insert, so that an insert usually costs a write to an array slot instead of a search of a skip list of all
+ * the cached entries, and a new page is found with a search of the pages. A lookup is a page lookup and an array read,
+ * and a range is visited page by page. A page is removed from the map when its last entry is removed.
+ *
+ * <p>The implementation avoids locks and synchronization by relying on the ConcurrentSkipListMap and atomic array
+ * slots. Since there are no locks, it's necessary to ensure that a single entry in the cache is removed exactly once.
* Removing an entry multiple times could result in the entries of the cache being released multiple times,
- * even while they are still in use. This is prevented by using a custom wrapper around the value to store in the map
- * that ensures that the value is removed from the map only if the exact same instance is present in the map.
+ * even while they are still in use. This is prevented by using a custom wrapper around the value to store in the
+ * slots that ensures that the value is removed only if the exact same instance is present in the slot.
* There's also a check that ensures that the value matches the key. This is used to detect races without impacting
* consistency.
*/
@CustomLog
class RangeCache {
- private final ConcurrentNavigableMap<Position, RangeCacheEntryWrapper> entries;
+ static final int PAGE_SHIFT = 6;
+ // The number of entry IDs in a page, a power of two so that an entry's page and slot are a shift and a mask
+ static final int PAGE_SIZE = 1 << PAGE_SHIFT;
+ private static final int SLOT_MASK = PAGE_SIZE - 1;
+
+ private final ConcurrentNavigableMap<PageKey, Page> pages;
private final RangeCacheRemovalQueue removalQueue;
private final AtomicLong size; // Total size of values stored in cache
@@ -54,7 +69,7 @@ class RangeCache {
*/
public RangeCache(RangeCacheRemovalQueue removalQueue) {
this.removalQueue = removalQueue;
- this.entries = new ConcurrentSkipListMap<>();
+ this.pages = new ConcurrentSkipListMap<>();
this.size = new AtomicLong(0);
}
@@ -67,29 +82,158 @@ class RangeCache {
* @return whether the entry was inserted in the cache
*/
public boolean put(Position key, ReferenceCountedEntry value, int entryLength) {
- // retain value so that it's not released before we put it in the cache and calculate the weight
- value.retain();
- try {
- if (!value.matchesPosition(key)) {
- throw new IllegalArgumentException("Value '" + value + "' does not match key '" + key + "'");
+ return newInserter().put(key, value, entryLength);
+ }
+
+ /**
+ * Returns an inserter for entries that are inserted in order, such as a managed ledger's added entries or the
+ * entries of a read from storage.
+ */
+ public Inserter newInserter() {
+ return new Inserter();
+ }
+
+ /**
+ * Inserts entries into the cache, remembering the page of its previous insert, so that the next entry in order
+ * usually goes to the same page without a lookup. An entry at any other position is inserted too, after a lookup
+ * of its page. An inserter is used by one thread at a time.
+ */
+ public final class Inserter implements Function<RangeCacheEntryWrapper, Boolean> {
+ // the page of the previous insert; its fields that are used here are final
+ private Page page;
+
+ private Inserter() {
+ }
+
+ /**
+ * Insert.
+ *
+ * @param key the position of the entry
+ * @param value ref counted value with at least 1 ref to pass on the cache
+ * @param entryLength size of the entry in bytes
+ * @return whether the entry was inserted in the cache
+ */
+ public boolean put(Position key, ReferenceCountedEntry value, int entryLength) {
+ // retain value so that it's not released before we put it in the cache and calculate the weight
+ value.retain();
+ try {
+ if (!value.matchesPosition(key)) {
+ throw new IllegalArgumentException("Value '" + value + "' does not match key '" + key + "'");
+ }
+ return RangeCacheEntryWrapper.withNewInstance(RangeCache.this, key, value, entryLength, this);
+ } finally {
+ value.release();
}
- 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);
+ /**
+ * Runs the update on the entry that the cache has at the position, if it has one, and returns whether it has.
+ * Cached entries are immutable and never replaced, so an entry that the cache has doesn't need to be prepared
+ * for an insert, such as copied, but its mutable state, its expected read count, can be updated. The update
+ * doesn't count as an access of the entry for the eviction.
+ */
+ public boolean updateIfCached(Position key, Consumer<ReferenceCountedEntry> update) {
+ long ledgerId = key.getLedgerId();
+ long entryId = key.getEntryId();
+ long pageIndex = entryId >> PAGE_SHIFT;
+ Page current = page;
+ if (current == null || !current.covers(ledgerId, pageIndex)) {
+ current = pages.get(new PageKey(ledgerId, pageIndex));
+ if (current == null) {
+ return false;
+ }
+ page = current;
+ }
+ RangeCacheEntryWrapper wrapper = current.slots.get((int) (entryId & SLOT_MASK));
+ if (wrapper == null) {
+ return false;
+ }
+ ReferenceCountedEntry value =
+ getRetainedValueMatchingPosition(ledgerId, entryId, wrapper.getValue(ledgerId, entryId, false));
+ if (value == null) {
+ return false;
+ }
+ try {
+ update.accept(value);
+ } finally {
+ value.release();
+ }
return true;
- } else {
- // recycle the new wrapper as it was not used
- newWrapper.recycle();
- return false;
}
+
+ /**
+ * Adds the new wrapper to the cache. withNewInstance holds the wrapper's write lock while its initialized
+ * fields are used.
+ */
+ @Override
+ public Boolean apply(RangeCacheEntryWrapper newWrapper) {
+ if (addToPage(newWrapper) && removalQueue.addEntry(newWrapper)) {
+ size.addAndGet(newWrapper.size);
+ return true;
+ } else {
+ // recycle the new wrapper as it was not used
+ newWrapper.recycle();
+ return false;
+ }
+ }
+
+ /**
+ * Stores the wrapper in the slot of its key, unless the slot already has an entry.
+ */
+ private boolean addToPage(RangeCacheEntryWrapper wrapper) {
+ long ledgerId = wrapper.key.getLedgerId();
+ long entryId = wrapper.key.getEntryId();
+ long pageIndex = entryId >> PAGE_SHIFT;
+ int slot = (int) (entryId & SLOT_MASK);
+ while (true) {
+ Page current = page;
+ if (current == null || !current.covers(ledgerId, pageIndex)) {
+ current = getOrCreatePage(ledgerId, pageIndex);
+ page = current;
+ }
+ if (!current.reserveSlot()) {
+ // the page lost its last entry and is being removed, so the entry goes to a new page
+ removePage(current);
+ page = null;
+ continue;
+ }
+ if (current.slots.compareAndSet(slot, null, wrapper)) {
+ return true;
+ }
+ releaseSlot(current);
+ return false;
+ }
+ }
+ }
+
+ private Page getOrCreatePage(long ledgerId, long pageIndex) {
+ PageKey pageKey = new PageKey(ledgerId, pageIndex);
+ Page page = pages.get(pageKey);
+ if (page == null) {
+ Page newPage = new Page(pageKey);
+ page = pages.putIfAbsent(pageKey, newPage);
+ if (page == null) {
+ page = newPage;
+ }
+ }
+ return page;
+ }
+
+ /**
+ * Returns the page of the entry ID, or null when there isn't one.
+ */
+ private Page findPage(long ledgerId, long entryId) {
+ return pages.get(new PageKey(ledgerId, entryId >> PAGE_SHIFT));
+ }
+
+ private void releaseSlot(Page page) {
+ if (page.releaseSlot()) {
+ removePage(page);
+ }
+ }
+
+ private void removePage(Page page) {
+ pages.remove(page.key, page);
}
/**
@@ -104,7 +248,12 @@ class RangeCache {
}
public boolean exists(Position key) {
- return key != null ? entries.containsKey(key) : true;
+ return key != null ? getWrapper(key) != null : true;
+ }
+
+ private RangeCacheEntryWrapper getWrapper(Position key) {
+ Page page = findPage(key.getLedgerId(), key.getEntryId());
+ return page != null ? page.slots.get((int) (key.getEntryId() & SLOT_MASK)) : null;
}
/**
@@ -112,33 +261,30 @@ class RangeCache {
* The caller is responsible for releasing the reference.
*/
public ReferenceCountedEntry get(Position key) {
- return getValueFromWrapper(key, entries.get(key));
- }
-
- private ReferenceCountedEntry getValueFromWrapper(Position key, RangeCacheEntryWrapper valueWrapper) {
+ RangeCacheEntryWrapper valueWrapper = getWrapper(key);
if (valueWrapper == null) {
return null;
} else {
ReferenceCountedEntry value = valueWrapper.getValue(key);
- return getRetainedValueMatchingKey(key, value);
+ return getRetainedValueMatchingPosition(key.getLedgerId(), key.getEntryId(), value);
}
}
/**
* @apiNote the returned value must be released if it's not null
*/
- private ReferenceCountedEntry getValueMatchingEntry(Map.Entry<Position, RangeCacheEntryWrapper> entry) {
- ReferenceCountedEntry valueMatchingEntry =
- RangeCacheEntryWrapper.getValueMatchingMapEntry(entry);
- return getRetainedValueMatchingKey(entry.getKey(), valueMatchingEntry);
+ private static ReferenceCountedEntry getRetainedValueAt(RangeCacheEntryWrapper wrapper, long ledgerId,
+ long entryId) {
+ return getRetainedValueMatchingPosition(ledgerId, entryId, wrapper.getValue(ledgerId, entryId));
}
- // validates that the value matches the key and that the value has not been recycled
+ // validates that the value matches the position and that the value has not been recycled
// which are possible due to the lack of exclusive locks in the cache and the use of reference counted objects
/**
* @apiNote the returned value must be released if it's not null
*/
- private ReferenceCountedEntry getRetainedValueMatchingKey(Position key, ReferenceCountedEntry value) {
+ private static ReferenceCountedEntry getRetainedValueMatchingPosition(long ledgerId, long entryId,
+ ReferenceCountedEntry value) {
if (value == null) {
// the wrapper has been recycled and contains another key
return null;
@@ -151,7 +297,7 @@ class RangeCache {
}
// check that the value matches the key and that there's at least 2 references to it since
// the cache should be holding one reference and a new reference was just added in this method
- if (value.refCnt() > 1 && value.matchesPosition(key)) {
+ if (value.refCnt() > 1 && value.getLedgerId() == ledgerId && value.getEntryId() == entryId) {
return value;
} else {
// Value or IdentityWrapper was recycled and already contains another value
@@ -173,13 +319,12 @@ class RangeCache {
List<ReferenceCountedEntry> values = new ArrayList<>();
// Return the values of the entries found in cache
- for (Map.Entry<Position, RangeCacheEntryWrapper> entry : entries.subMap(first, true, last, true)
- .entrySet()) {
- ReferenceCountedEntry value = getValueMatchingEntry(entry);
+ forEachWrapperInRange(first, last, true, (wrapper, ledgerId, entryId) -> {
+ ReferenceCountedEntry value = getRetainedValueAt(wrapper, ledgerId, entryId);
if (value != null) {
values.add(value);
}
- }
+ });
return values;
}
@@ -189,9 +334,8 @@ class RangeCache {
* released afterwards, including when the callback throws. The visitor must retain entries it needs to keep.
*/
public void forEachInRange(Position first, Position last, Consumer<ReferenceCountedEntry> visitor) {
- for (Map.Entry<Position, RangeCacheEntryWrapper> entry : entries.subMap(first, true, last, true)
- .entrySet()) {
- ReferenceCountedEntry value = getValueMatchingEntry(entry);
+ forEachWrapperInRange(first, last, true, (wrapper, ledgerId, entryId) -> {
+ ReferenceCountedEntry value = getRetainedValueAt(wrapper, ledgerId, entryId);
if (value != null) {
try {
visitor.accept(value);
@@ -199,6 +343,38 @@ class RangeCache {
value.release();
}
}
+ });
+ }
+
+ private interface WrapperVisitor {
+ void visit(RangeCacheEntryWrapper wrapper, long ledgerId, long entryId);
+ }
+
+ /**
+ * Visits the occupied slots of the range in order.
+ */
+ private void forEachWrapperInRange(Position first, Position last, boolean lastInclusive,
+ WrapperVisitor visitor) {
+ long firstLedgerId = first.getLedgerId();
+ long firstEntryId = first.getEntryId();
+ long lastLedgerId = last.getLedgerId();
+ long lastEntryId = lastInclusive ? last.getEntryId() : last.getEntryId() - 1;
+ if (firstLedgerId > lastLedgerId || (firstLedgerId == lastLedgerId && firstEntryId > lastEntryId)) {
+ return;
+ }
+ PageKey firstPage = new PageKey(firstLedgerId, firstEntryId >> PAGE_SHIFT);
+ PageKey lastPage = new PageKey(lastLedgerId, lastEntryId >> PAGE_SHIFT);
+ for (Page page : pages.subMap(firstPage, true, lastPage, true).values()) {
+ int firstSlot = page.key.equals(firstPage) ? (int) (firstEntryId & SLOT_MASK) : 0;
+ int lastSlot = page.key.equals(lastPage) ? (int) (lastEntryId & SLOT_MASK) : PAGE_SIZE - 1;
+ long ledgerId = page.key.ledgerId();
+ long pageFirstEntryId = page.key.pageIndex() << PAGE_SHIFT;
+ for (int slot = firstSlot; slot <= lastSlot; slot++) {
+ RangeCacheEntryWrapper wrapper = page.slots.get(slot);
+ if (wrapper != null) {
+ visitor.visit(wrapper, ledgerId, pageFirstEntryId + slot);
+ }
+ }
}
}
@@ -215,18 +391,16 @@ class RangeCache {
.attr("lastInclusive", lastInclusive)
.log("Removing entries in range");
RangeCacheRemovalCounters counters = RangeCacheRemovalCounters.create();
- Map<Position, RangeCacheEntryWrapper> subMap = entries.subMap(first, true, last, lastInclusive);
- for (Map.Entry<Position, RangeCacheEntryWrapper> entry : subMap.entrySet()) {
- removeEntryWithWriteLock(entry.getKey(), entry.getValue(), counters);
- }
+ forEachWrapperInRange(first, last, lastInclusive,
+ (wrapper, ledgerId, entryId) -> removeEntryWithWriteLock(wrapper, ledgerId, entryId, counters));
return handleRemovalResult(counters);
}
- boolean removeEntryWithWriteLock(Position expectedKey, RangeCacheEntryWrapper entryWrapper,
- RangeCacheRemovalCounters counters) {
+ private boolean removeEntryWithWriteLock(RangeCacheEntryWrapper entryWrapper, long ledgerId, long entryId,
+ RangeCacheRemovalCounters counters) {
return entryWrapper.withWriteLock(e -> {
- if (e.key == null || e.key != expectedKey) {
- // entry has already been removed
+ if (e.key == null || e.rangeCache != this || e.key.compareTo(ledgerId, entryId) != 0) {
+ // entry has already been removed, and the wrapper may have been reused for another entry
return false;
}
return removeEntry(e.key, e.value, e, counters, false);
@@ -244,8 +418,8 @@ class RangeCache {
*/
boolean removeEntry(Position key, ReferenceCountedEntry value, RangeCacheEntryWrapper entryWrapper,
RangeCacheRemovalCounters counters, boolean updateSize) {
- // always remove the entry from the map
- entries.remove(key, entryWrapper);
+ // always remove the entry from its page
+ removeFromPage(key, entryWrapper);
if (value == null) {
// the wrapper has already been recycled and contains another key
return false;
@@ -284,6 +458,13 @@ class RangeCache {
}
}
+ private void removeFromPage(Position key, RangeCacheEntryWrapper entryWrapper) {
+ Page page = findPage(key.getLedgerId(), key.getEntryId());
+ if (page != null && page.slots.compareAndSet((int) (key.getEntryId() & SLOT_MASK), entryWrapper, null)) {
+ releaseSlot(page);
+ }
+ }
+
private Pair<Integer, Long> handleRemovalResult(RangeCacheRemovalCounters counters) {
size.addAndGet(-counters.removedSize);
Pair<Integer, Long> result = Pair.of(counters.removedEntries, counters.removedSize);
@@ -292,10 +473,21 @@ class RangeCache {
}
/**
- * Just for testing. Getting the number of entries is very expensive on the concurrent map
+ * Just for testing. Counts the entries of all the pages.
*/
protected long getNumberOfEntries() {
- return entries.size();
+ long numberOfEntries = 0;
+ for (Page page : pages.values()) {
+ numberOfEntries += Math.max(page.count, 0);
+ }
+ return numberOfEntries;
+ }
+
+ /**
+ * Just for testing. The number of pages in the cache.
+ */
+ int getNumberOfPages() {
+ return pages.size();
}
public long getSize() {
@@ -308,15 +500,89 @@ class RangeCache {
* @return size of removed entries
*/
public Pair<Integer, Long> clear() {
- log.debug().attr("numEntries", () -> entries.size()).attr("size", size.get()).log("Clearing the cache");
+ log.debug().attr("numPages", () -> pages.size()).attr("size", size.get()).log("Clearing the cache");
RangeCacheRemovalCounters counters = RangeCacheRemovalCounters.create();
- while (!Thread.currentThread().isInterrupted()) {
- Map.Entry<Position, RangeCacheEntryWrapper> entry = entries.firstEntry();
- if (entry == null) {
- break;
+ // Passes over the pages until one finds no entry, so that an entry that a concurrent insert added to a page
+ // that an earlier pass had visited is removed too
+ boolean found = true;
+ while (found && !Thread.currentThread().isInterrupted()) {
+ found = false;
+ for (Page page : pages.values()) {
+ long ledgerId = page.key.ledgerId();
+ long pageFirstEntryId = page.key.pageIndex() << PAGE_SHIFT;
+ for (int slot = 0; slot < PAGE_SIZE; slot++) {
+ RangeCacheEntryWrapper wrapper = page.slots.get(slot);
+ if (wrapper != null) {
+ found = true;
+ removeEntryWithWriteLock(wrapper, ledgerId, pageFirstEntryId + slot, counters);
+ }
+ }
}
- removeEntryWithWriteLock(entry.getKey(), entry.getValue(), counters);
}
return handleRemovalResult(counters);
}
+
+ /**
+ * The key of a page: the ledger ID and the entry ID divided by {@link #PAGE_SIZE}.
+ */
+ record PageKey(long ledgerId, long pageIndex) implements Comparable<PageKey> {
+ @Override
+ public int compareTo(PageKey other) {
+ int result = Long.compare(ledgerId, other.ledgerId);
+ return result != 0 ? result : Long.compare(pageIndex, other.pageIndex);
+ }
+ }
+
+ /**
+ * The slots of the consecutive entry IDs of a page, and the number of slots that have an entry. A page that loses
+ * its last entry is sealed, so that no entry is added to it while it's being removed from the map.
+ */
+ static final class Page {
+ private static final int SEALED = -1;
+ private static final AtomicIntegerFieldUpdater<Page> COUNT_UPDATER =
+ AtomicIntegerFieldUpdater.newUpdater(Page.class, "count");
+
+ final PageKey key;
+ final AtomicReferenceArray<RangeCacheEntryWrapper> slots = new AtomicReferenceArray<>(PAGE_SIZE);
+ // the number of slots that have an entry or are reserved for one, or SEALED
+ private volatile int count;
+
+ Page(PageKey key) {
+ this.key = key;
+ }
+
+ boolean covers(long ledgerId, long pageIndex) {
+ return key.ledgerId() == ledgerId && key.pageIndex() == pageIndex;
+ }
+
+ /**
+ * Reserves a slot for an entry to add, unless the page has been sealed.
+ */
+ boolean reserveSlot() {
+ while (true) {
+ int current = count;
+ if (current == SEALED) {
+ return false;
+ }
+ if (COUNT_UPDATER.compareAndSet(this, current, current + 1)) {
+ return true;
+ }
+ }
+ }
+
+ /**
+ * Releases a slot whose entry was removed or wasn't added, and seals the page when it was the last one.
+ *
+ * @return whether the page was sealed
+ */
+ boolean releaseSlot() {
+ while (true) {
+ int current = count;
+ int next = current == 1 ? SEALED : current - 1;
+ if (COUNT_UPDATER.compareAndSet(this, current, next)) {
+ return next == SEALED;
+ }
+ }
+ }
+ }
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
index 85664240e0b6..c6fe88d1e1c8 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
@@ -19,7 +19,6 @@
package org.apache.bookkeeper.mledger.impl.cache;
import io.netty.util.Recycler;
-import java.util.Map;
import java.util.concurrent.locks.StampedLock;
import java.util.function.Function;
import lombok.CustomLog;
@@ -27,9 +26,9 @@ import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.ReferenceCountedEntry;
/**
- * Wrapper around the value to store in Map. This is needed to ensure that a specific instance can be removed from
- * the map by calling the {@link Map#remove(Object, Object)} method. Certain race conditions could result in the
- * wrong value being removed from the map. The instances of this class are recycled to avoid creating new objects.
+ * Wrapper around the value to store in a {@link RangeCache} slot. This is needed to ensure that a specific instance
+ * can be removed from its slot with a compare-and-set. Certain race conditions could result in the wrong value being
+ * removed from the cache. The instances of this class are recycled to avoid creating new objects.
*/
@CustomLog
class RangeCacheEntryWrapper {
@@ -80,32 +79,24 @@ class RangeCacheEntryWrapper {
* match
*/
ReferenceCountedEntry getValue(Position key) {
- return getValueInternal(key, false);
+ return getValue(key.getLedgerId(), key.getEntryId());
}
/**
- * Get the value associated with the Map.Entry's key and value. Exact instance of the key is required to match.
+ * Get the value of the entry at the given ledger ID and entry ID, such as the position of a cache slot.
*
- * @param entry the entry which contains the key and {@link RangeCacheEntryWrapper} value to get the value from
- * @return the value associated with the key, or null if the value has already been recycled or the key does not
- * exactly match the same instance
+ * @return the value associated with the position, or null if the value has already been recycled or the wrapper
+ * holds another position
*/
- static ReferenceCountedEntry getValueMatchingMapEntry(Map.Entry<Position, RangeCacheEntryWrapper> entry) {
- return entry.getValue().getValueInternal(entry.getKey(), true);
+ ReferenceCountedEntry getValue(long ledgerId, long entryId) {
+ return getValue(ledgerId, entryId, true);
}
/**
- * Get the value associated with the key. Returns null if the key does not match the key associated with the
- * value.
- *
- * @param key the key to match
- * @param requireSameKeyInstance when true, the matching will be restricted to exactly the same instance of the
- * key as the one stored in the wrapper. This is used to avoid any races
- * when retrieving or removing the entries from the cache when the key and value
- * instances are available.
- * @return the value associated with the key, or null if the key does not match
+ * Get the value of the entry at the given position, marking the entry accessed for the eviction only when
+ * {@code markAccessed} is set.
*/
- private ReferenceCountedEntry getValueInternal(Position key, boolean requireSameKeyInstance) {
+ ReferenceCountedEntry getValue(long ledgerId, long entryId, boolean markAccessed) {
long stamp = lock.tryOptimisticRead();
Position localKey = this.key;
ReferenceCountedEntry localValue = this.value;
@@ -115,14 +106,14 @@ class RangeCacheEntryWrapper {
localValue = this.value;
lock.unlockRead(stamp);
}
- // check that the given key matches the key associated with the value in the entry
+ // check that the position matches the key associated with the value in the entry
// this is used to detect if the entry has already been recycled and contains another key
- // when requireSameKeyInstance is true, the key must be exactly the same instance as the one stored in the
- // entry to match
- if (localKey != key && (requireSameKeyInstance || localKey == null || !localKey.equals(key))) {
+ if (localKey == null || localKey.compareTo(ledgerId, entryId) != 0) {
return null;
}
- accessed = true;
+ if (markAccessed) {
+ accessed = true;
+ }
return localValue;
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
index 6375465a13eb..8e43221b231e 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
@@ -72,6 +72,8 @@ public class RangeEntryCacheImpl implements EntryCache {
private final Logger log;
private ManagedLedgerInterceptor interceptor;
private final RangeCache entries;
+ // inserts the managed ledger's added entries, which come in order on the managed ledger's executor
+ private final RangeCache.Inserter addedEntriesInserter;
private final boolean copyEntries;
private final PendingReadsManager pendingReadsManager;
@@ -99,6 +101,7 @@ public class RangeEntryCacheImpl implements EntryCache {
this.entryLengthFunction = entryLengthFunction;
this.interceptor = ml.getManagedLedgerInterceptor();
this.entries = new RangeCache(rangeCacheRemovalQueue);
+ this.addedEntriesInserter = entries.newInserter();
this.copyEntries = copyEntries;
log.debug("Initialized managed-ledger entry cache");
@@ -131,7 +134,30 @@ public class RangeEntryCacheImpl implements EntryCache {
@Override
public boolean insert(Entry entry) {
- return insert(entry, copyEntries);
+ return insert(entry, copyEntries, addedEntriesInserter);
+ }
+
+ /**
+ * Inserts consecutive entries in order, such as the entries of a read from storage, with an inserter of their own,
+ * so that they don't displace the page where the managed ledger adds its entries. Cached entries are immutable,
+ * so an entry that is already cached, such as one that an overlapping read inserted, is skipped before it's
+ * copied; the cached entry takes its expected read count.
+ *
+ * @param entriesToInsert the entries to cache, in order
+ * @param copy whether to copy the entries' data into cache owned buffers, always the case when the cache is
+ * configured to copy entries
+ * @return the number of entries that were inserted
+ */
+ public int insert(List<? extends Entry> entriesToInsert, boolean copy) {
+ RangeCache.Inserter inserter = entries.newInserter();
+ int inserted = 0;
+ for (Entry entry : entriesToInsert) {
+ if (!inserter.updateIfCached(entry.getPosition(), cached -> takeExpectedReadCount(cached, entry))
+ && insert(entry, copy, inserter)) {
+ inserted++;
+ }
+ }
+ return inserted;
}
/**
@@ -142,9 +168,10 @@ public class RangeEntryCacheImpl implements EntryCache {
* @param entry the entry to cache
* @param copy whether to copy the entry data into a cache owned buffer, always the case when the cache is
* configured to copy entries
+ * @param inserter the inserter of the entries that this entry is inserted in order with
* @return whether the entry was inserted
*/
- public boolean insert(Entry entry, boolean copy) {
+ private boolean insert(Entry entry, boolean copy, RangeCache.Inserter inserter) {
int entryLength = entryLengthFunction.getEntryLength(ml, entry);
log.debug().attr("position", entry.getPosition())
@@ -170,18 +197,28 @@ public class RangeEntryCacheImpl implements EntryCache {
EntryImpl.createWithRetainedDuplicate(position, cachedData, entry.getReadCountHandler(),
copy ? null : entry.getMessageMetadata());
cachedData.release();
- if (entries.put(position, cacheEntry, entryLength)) {
+ if (inserter.put(position, cacheEntry, entryLength)) {
totalAddedEntriesSize.add(entryLength);
totalAddedEntriesCount.increment();
manager.entryAdded(entryLength);
return true;
} else {
- // entry was not inserted into cache, we need to discard it
+ // the cache already has the entry, which takes the expected read count of this addition
+ inserter.updateIfCached(position, cached -> takeExpectedReadCount(cached, entry));
+ // entry was not inserted into cache, we need to discard it; nobody read the discarded duplicate, so its
+ // release doesn't count as a read of the expected read count that it shares with the added entry
+ cacheEntry.setDecreaseReadCountOnRelease(false);
cacheEntry.release();
return false;
}
}
+ private static void takeExpectedReadCount(ReferenceCountedEntry cached, Entry latest) {
+ if (cached instanceof EntryImpl cachedEntry) {
+ cachedEntry.updateExpectedReadCount(latest.getReadCountHandler());
+ }
+ }
+
private ByteBuf copyEntry(Entry entry) {
// Copy the entry into a buffer owned by the cache. The reason is that the incoming entry is retaining a buffer
// from netty, usually allocated in 64Kb chunks. So if we just retain the entry without copying it, we might
@@ -599,9 +636,9 @@ public class RangeEntryCacheImpl implements EntryCache {
}
entriesToReturn.add(entry);
totalSize += entry.getLength();
- if (expectedReadCountVal > 0) {
- insert(entry, copyEntries || batchRead);
- }
+ }
+ if (expectedReadCountVal > 0) {
+ insert(entriesToReturn, copyEntries || batchRead);
}
ml.getMbean().recordReadEntriesOpsCacheMisses(entriesToReturn.size(), totalSize);
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheTest.java
index c26fadbe72f1..60fa7e6a1409 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheTest.java
@@ -31,9 +31,16 @@ import io.netty.buffer.Unpooled;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Queue;
+import java.util.Random;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import lombok.Cleanup;
import org.apache.bookkeeper.mledger.Entry;
@@ -453,4 +460,295 @@ public class RangeCacheTest {
// the value should be found
assertEquals(value2.getData(), "129".getBytes());
}
+
+ @Test
+ public void rangesSpanPagesAndLedgers() {
+ RangeCache cache = new RangeCache(createRemovalQueue());
+ int pageSize = RangeCache.PAGE_SIZE;
+ // the last entries of a page, the first of the next, and the first entries of the next ledger
+ long[][] positions = {{1, pageSize - 2}, {1, pageSize - 1}, {1, pageSize}, {1, 3L * pageSize}, {2, 0}, {2, 1}};
+ for (long[] position : positions) {
+ putToCache(cache, position[0], position[1]);
+ }
+ assertEquals(cache.getNumberOfEntries(), positions.length);
+ // ledger 1's pages 0, 1 and 3, and ledger 2's page 0
+ assertEquals(cache.getNumberOfPages(), 4);
+
+ assertRange(cache, PositionFactory.create(1, pageSize - 1), PositionFactory.create(2, 0),
+ PositionFactory.create(1, pageSize - 1), PositionFactory.create(1, pageSize),
+ PositionFactory.create(1, 3L * pageSize), PositionFactory.create(2, 0));
+ assertTrue(cache.exists(PositionFactory.create(1, pageSize)));
+ assertFalse(cache.exists(PositionFactory.create(1, pageSize + 1)));
+
+ // the end of a removed range is exclusive also at a page boundary
+ assertEquals(cache.removeRange(PositionFactory.create(1, 0), PositionFactory.create(1, pageSize), false),
+ Pair.of(2, (long) (Long.toString(pageSize - 2).length() + Long.toString(pageSize - 1).length())));
+ assertNull(cache.get(PositionFactory.create(1, pageSize - 1)));
+ assertEquals(cache.getNumberOfPages(), 3);
+
+ // a range from before the first ledger, as invalidating the entries of the consumed ledgers does
+ assertEquals(cache.removeRange(PositionFactory.create(-1, 0), PositionFactory.create(2, 1), false).getLeft(),
+ 3);
+ assertRange(cache, PositionFactory.create(0, 0), PositionFactory.create(3, 0), PositionFactory.create(2, 1));
+ assertEquals(cache.getNumberOfPages(), 1);
+ cache.clear();
+ assertEquals(cache.getNumberOfPages(), 0);
+ assertEquals(cache.getSize(), 0);
+ }
+
+ @Test
+ public void removesPagesWhenTheirEntriesAreEvicted() {
+ RangeCacheRemovalQueue removalQueue = createRemovalQueue();
+ RangeCache cache = new RangeCache(removalQueue);
+ List<ReferenceCountedEntry> values = new ArrayList<>();
+ for (int i = 0; i < 3 * RangeCache.PAGE_SIZE; i++) {
+ ReferenceCountedEntry value = createCachedEntry(i, "x");
+ value.retain();
+ values.add(value);
+ assertTrue(cache.put(value.getPosition(), value));
+ }
+ assertEquals(cache.getNumberOfPages(), 3);
+ assertEquals(removalQueue.evictLeastAccessedEntries(RangeCache.PAGE_SIZE + 1),
+ Pair.of(RangeCache.PAGE_SIZE + 1, RangeCache.PAGE_SIZE + 1L));
+ assertEquals(cache.getNumberOfPages(), 2);
+ assertEquals(removalQueue.evictLeastAccessedEntries(Long.MAX_VALUE).getLeft(), 2 * RangeCache.PAGE_SIZE - 1);
+ assertEquals(cache.getNumberOfPages(), 0);
+ assertEquals(cache.getNumberOfEntries(), 0);
+ // the cache released its references, and an entry can be added to a removed page's range again
+ assertThat(values).allSatisfy(value -> assertEquals(value.refCnt(), 1));
+ putToCache(cache, 1, "1");
+ assertEquals(new String(releaseRetained(cache, createPosition(1)).getData()), "1");
+ cache.clear();
+ values.forEach(ReferenceCountedEntry::release);
+ }
+
+ @Test
+ public void inserterUpdatesTheCachedEntryAtAPosition() {
+ RangeCache cache = new RangeCache(createRemovalQueue());
+ RangeCache.Inserter inserter = cache.newInserter();
+ assertFalse(inserter.updateIfCached(createPosition(1), entry -> fail("Not cached")));
+ putToCache(cache, 1, "1");
+ // the inserter looks the page up, and then uses it; the update gets the cached entry, retained meanwhile
+ List<String> updated = new ArrayList<>();
+ assertTrue(inserter.updateIfCached(createPosition(1), entry -> {
+ assertEquals(entry.refCnt(), 2);
+ updated.add(new String(entry.getData()));
+ }));
+ assertThat(updated).containsExactly("1");
+ assertFalse(inserter.updateIfCached(createPosition(2), entry -> fail("Not cached")));
+ assertFalse(inserter.updateIfCached(PositionFactory.create(1, 1), entry -> fail("Not cached")));
+ cache.removeRange(createPosition(1), createPosition(1), true);
+ assertFalse(inserter.updateIfCached(createPosition(1), entry -> fail("Not cached")));
+ cache.clear();
+ }
+
+ @Test
+ public void inserterAddsToANewPageAfterItsPageWasRemoved() {
+ RangeCache cache = new RangeCache(createRemovalQueue());
+ RangeCache.Inserter inserter = cache.newInserter();
+ ReferenceCountedEntry first = createCachedEntry(0, "0");
+ assertTrue(inserter.put(first.getPosition(), first, first.getLength()));
+ // the page loses its only entry and is removed, while the inserter still remembers it
+ assertEquals(cache.removeRange(createPosition(0), createPosition(0), true).getLeft(), 1);
+ assertEquals(cache.getNumberOfPages(), 0);
+ ReferenceCountedEntry second = createCachedEntry(1, "1");
+ assertTrue(inserter.put(second.getPosition(), second, second.getLength()));
+ assertEquals(cache.getNumberOfPages(), 1);
+ assertEquals(new String(releaseRetained(cache, createPosition(1)).getData()), "1");
+ // an entry before the inserter's page, as a read from storage inserts, and one in a later page
+ ReferenceCountedEntry earlier = createCachedEntry(PositionFactory.create(-1, 5), "e");
+ assertTrue(inserter.put(earlier.getPosition(), earlier, earlier.getLength()));
+ ReferenceCountedEntry later = createCachedEntry(2 * RangeCache.PAGE_SIZE, "l");
+ assertTrue(inserter.put(later.getPosition(), later, later.getLength()));
+ assertRange(cache, PositionFactory.create(-1, 0), createPosition(1000), earlier.getPosition(),
+ createPosition(1), later.getPosition());
+ cache.clear();
+ assertEquals(cache.getNumberOfPages(), 0);
+ }
+
+ @Test
+ public void concurrentInsertsReadsAndRemovals() throws Exception {
+ RangeCacheRemovalQueue removalQueue = createRemovalQueue();
+ RangeCache cache = new RangeCache(removalQueue);
+ int numberOfEntries = 200_000;
+ int numberOfReaders = 4;
+ List<ReferenceCountedEntry> values = new ArrayList<>(numberOfEntries);
+ for (int i = 0; i < numberOfEntries; i++) {
+ ReferenceCountedEntry value = createCachedEntry(PositionFactory.create(i / 10_000, i % 10_000), "x");
+ // the test's own reference, to check that the cache releases each of its references exactly once
+ value.retain();
+ values.add(value);
+ }
+ AtomicBoolean done = new AtomicBoolean();
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ @Cleanup("shutdownNow")
+ ExecutorService executor = Executors.newFixedThreadPool(numberOfReaders + 2);
+ List<Future<?>> futures = new ArrayList<>();
+ for (int r = 0; r < numberOfReaders; r++) {
+ int reader = r;
+ futures.add(executor.submit(() -> {
+ Random random = new Random(reader);
+ while (!done.get()) {
+ int first = random.nextInt(numberOfEntries);
+ Position firstPosition = values.get(first).getPosition();
+ Position lastPosition = values.get(Math.min(first + 150, numberOfEntries - 1)).getPosition();
+ cache.forEachInRange(firstPosition, lastPosition, value -> {
+ if (value.refCnt() < 2 || value.getPosition().compareTo(firstPosition) < 0
+ || value.getPosition().compareTo(lastPosition) > 0) {
+ failure.compareAndSet(null, new AssertionError("Unexpected entry " + value));
+ }
+ });
+ ReferenceCountedEntry value = cache.get(firstPosition);
+ if (value != null) {
+ if (!value.matchesPosition(firstPosition)) {
+ failure.compareAndSet(null, new AssertionError("Unexpected entry " + value));
+ }
+ value.release();
+ }
+ }
+ }));
+ }
+ // the cursors' progress invalidates ranges while the eviction removes the oldest entries
+ futures.add(executor.submit(() -> {
+ Random random = new Random(-1);
+ while (!done.get()) {
+ Position last = values.get(random.nextInt(numberOfEntries)).getPosition();
+ cache.removeRange(PositionFactory.create(last.getLedgerId(), Math.max(0, last.getEntryId() - 200)),
+ last, random.nextBoolean());
+ }
+ }));
+ futures.add(executor.submit(() -> {
+ while (!done.get()) {
+ removalQueue.evictLeastAccessedEntries(1_000);
+ }
+ }));
+ try {
+ // the managed ledger adds its entries with one inserter, whose remembered page races with the removals
+ RangeCache.Inserter inserter = cache.newInserter();
+ for (ReferenceCountedEntry value : values) {
+ inserter.put(value.getPosition(), value, value.getLength());
+ }
+ } finally {
+ done.set(true);
+ }
+ for (Future<?> future : futures) {
+ future.get(30, TimeUnit.SECONDS);
+ }
+ assertNull(failure.get());
+
+ cache.clear();
+ removalQueue.evictLeastAccessedEntries(Long.MAX_VALUE);
+ assertEquals(cache.getSize(), 0);
+ assertEquals(cache.getNumberOfEntries(), 0);
+ assertEquals(cache.getNumberOfPages(), 0);
+ assertThat(values).allSatisfy(value -> assertEquals(value.refCnt(), 1));
+ values.forEach(ReferenceCountedEntry::release);
+ }
+
+ @Test
+ public void concurrentTailAndCatchUpInserts() throws Exception {
+ RangeCacheRemovalQueue removalQueue = createRemovalQueue();
+ RangeCache cache = new RangeCache(removalQueue);
+ long catchUpLedger = 1;
+ long tailLedger = 2;
+ int catchUpLedgerEntries = 5_000;
+ int tailEntries = 100_000;
+ // every entry offered to the cache, with the test's own reference, to check that the cache releases each of
+ // its references exactly once and doesn't keep an entry that it didn't insert
+ Queue<ReferenceCountedEntry> offered = new ConcurrentLinkedQueue<>();
+ AtomicBoolean done = new AtomicBoolean();
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ @Cleanup("shutdownNow")
+ ExecutorService executor = Executors.newFixedThreadPool(4);
+ List<Future<?>> futures = new ArrayList<>();
+ // catch-up reads insert runs of consecutive entries at random positions of the earlier ledger
+ for (int t = 0; t < 2; t++) {
+ int seed = t;
+ futures.add(executor.submit(() -> {
+ Random random = new Random(seed);
+ while (!done.get()) {
+ int first = random.nextInt(catchUpLedgerEntries - 100);
+ int count = 1 + random.nextInt(100);
+ // a read from storage inserts its consecutive entries with an inserter of its own
+ RangeCache.Inserter inserter = cache.newInserter();
+ for (int i = first; i < first + count; i++) {
+ offer(inserter, offered, PositionFactory.create(catchUpLedger, i));
+ }
+ Position firstPosition = PositionFactory.create(catchUpLedger, first);
+ Position lastPosition = PositionFactory.create(catchUpLedger, first + count - 1);
+ cache.forEachInRange(firstPosition, lastPosition, value -> {
+ if (value.refCnt() < 2 || value.getPosition().compareTo(firstPosition) < 0
+ || value.getPosition().compareTo(lastPosition) > 0) {
+ failure.compareAndSet(null, new AssertionError("Unexpected entry " + value));
+ }
+ });
+ if (random.nextInt(4) == 0) {
+ cache.removeRange(firstPosition, lastPosition, true);
+ }
+ }
+ }));
+ }
+ futures.add(executor.submit(() -> {
+ while (!done.get()) {
+ removalQueue.evictLeastAccessedEntries(1_000);
+ }
+ }));
+ try {
+ RangeCache.Inserter tailInserter = cache.newInserter();
+ for (int i = 0; i < tailEntries; i++) {
+ offer(tailInserter, offered, PositionFactory.create(tailLedger, i));
+ if (i % 1_000 == 999) {
+ // the cursors moved past the oldest tail entries
+ cache.removeRange(PositionFactory.create(tailLedger, 0),
+ PositionFactory.create(tailLedger, i - 500), false);
+ }
+ }
+ } finally {
+ done.set(true);
+ }
+ for (Future<?> future : futures) {
+ future.get(30, TimeUnit.SECONDS);
+ }
+ assertNull(failure.get());
+
+ cache.clear();
+ removalQueue.evictLeastAccessedEntries(Long.MAX_VALUE);
+ assertEquals(cache.getSize(), 0);
+ assertEquals(cache.getNumberOfEntries(), 0);
+ assertEquals(cache.getNumberOfPages(), 0);
+ assertThat(offered).allSatisfy(value -> assertEquals(value.refCnt(), 1));
+ offered.forEach(ReferenceCountedEntry::release);
+ }
+
+ // offers a new entry to the cache, which takes over its reference when it inserts it
+ private static void offer(RangeCache.Inserter inserter, Queue<ReferenceCountedEntry> offered, Position position) {
+ ReferenceCountedEntry value = createCachedEntry(position, "x");
+ value.retain();
+ offered.add(value);
+ if (!inserter.put(position, value, value.getLength())) {
+ value.release();
+ }
+ }
+
+ private void putToCache(RangeCache cache, long ledgerId, long entryId) {
+ Position position = PositionFactory.create(ledgerId, entryId);
+ assertTrue(cache.put(position, createCachedEntry(position, Long.toString(entryId))));
+ }
+
+ private static void assertRange(RangeCache cache, Position first, Position last, Position... expected) {
+ List<ReferenceCountedEntry> range = new ArrayList<>(cache.getRange(first, last));
+ try {
+ assertThat(range).map(Entry::getPosition).containsExactly(expected);
+ } finally {
+ range.forEach(ReferenceCountedEntry::release);
+ }
+ }
+
+ // gets the cached entry, releases the reference that get added and returns the entry
+ private static ReferenceCountedEntry releaseRetained(RangeCache cache, Position position) {
+ ReferenceCountedEntry value = cache.get(position);
+ assertNotNull(value);
+ value.release();
+ return value;
+ }
}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
index 4ef496f95cd7..708aff8c5a4a 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImplTest.java
@@ -176,6 +176,155 @@ public class RangeEntryCacheImplTest {
}
}
+ @Test
+ public void testCachedEntriesAreEvictableAfterAFailedPartialReadRetriesTheWholeRange() {
+ // entry 0 is cached with one expected read; entry 1 isn't cached
+ when(mockManagedLedger.getLastConfirmedEntry()).thenReturn(PositionFactory.create(1, 99));
+ when(mockManagedLedger.getExecutor()).thenReturn(mock(ThreadBoundExecutor.class));
+ EntryImpl added = EntryImpl.create(1, 0, "added-0".getBytes(StandardCharsets.UTF_8), 1);
+ added.setDecreaseReadCountOnRelease(false);
+ assertThat(rangeEntryCache.insert(added)).isTrue();
+ added.release();
+ // reading the missing entry 1 fails, and the retry reads both entries from storage, as a cursor's read would
+ List<LedgerEntryImpl> ledgerEntries = new ArrayList<>();
+ for (long entryId = 0; entryId <= 1; entryId++) {
+ ledgerEntries.add(LedgerEntryImpl.create(1L, entryId, 6,
+ Unpooled.wrappedBuffer(("read-" + entryId).getBytes(StandardCharsets.UTF_8))));
+ }
+ LedgerEntries read = mock(LedgerEntries.class);
+ when(read.iterator()).thenReturn(List.<LedgerEntry>copyOf(ledgerEntries).iterator());
+ when(lh.readAsync(0, 1)).thenReturn(CompletableFuture.completedFuture(read));
+ doAnswer(invocation -> {
+ AsyncCallbacks.ReadEntriesCallback callback = invocation.getArgument(5);
+ callback.readEntriesFailed(new ManagedLedgerException("Injected test failure"), invocation.getArgument(6));
+ return null;
+ }).when(pendingReadsManager).readEntries(any(), eq(1L), eq(1L), anyLong(), any(), any(), any());
+ doAnswer(invocation -> {
+ AsyncCallbacks.ReadEntriesCallback callback = invocation.getArgument(5);
+ rangeEntryCache.readFromStorage(lh, 0, 1, NO_MAX_SIZE_LIMIT, invocation.getArgument(4))
+ .whenComplete((entries, failure) -> {
+ if (failure != null) {
+ callback.readEntriesFailed(new ManagedLedgerException(failure), invocation.getArgument(6));
+ } else {
+ callback.readEntriesComplete(entries, invocation.getArgument(6));
+ }
+ });
+ return null;
+ }).when(pendingReadsManager).readEntries(any(), eq(0L), eq(1L), anyLong(), any(), any(), any());
+
+ CompletableFuture<List<Entry>> future = new CompletableFuture<>();
+ rangeEntryCache.asyncReadEntry(lh, 0, 1, NO_MAX_SIZE_LIMIT, () -> 1,
+ new AsyncCallbacks.ReadEntriesCallback() {
+ @Override
+ public void readEntriesComplete(List<Entry> entries, Object ctx) {
+ future.complete(entries);
+ }
+
+ @Override
+ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+ assertThat(future).isCompletedWithValueMatching(entries -> entries.size() == 2);
+ try {
+ // the cursor reads the returned entries and releases them
+ future.getNow(null).forEach(Entry::release);
+ // the entries' expected reads are done, so both cached entries can be evicted
+ for (long entryId = 0; entryId <= 1; entryId++) {
+ ReferenceCountedEntry cached = rangeEntryCache.getEntries().get(PositionFactory.create(1, entryId));
+ try {
+ assertThat(cached).as("entry %d", entryId).isNotNull();
+ assertThat(cached.hasExpectedReads()).as("entry %d", entryId).isFalse();
+ } finally {
+ if (cached != null) {
+ cached.release();
+ }
+ }
+ }
+ } finally {
+ ledgerEntries.forEach(LedgerEntryImpl::close);
+ rangeEntryCache.clear();
+ }
+ }
+
+ @Test
+ public void testTheLatestAdditionsExpectedReadCountOverridesTheCachedEntrys() {
+ EntryImpl first = EntryImpl.create(1, 20, "first".getBytes(StandardCharsets.UTF_8), 2);
+ EntryImpl second = EntryImpl.create(1, 20, "second".getBytes(StandardCharsets.UTF_8), 4);
+ EntryImpl third = EntryImpl.create(1, 20, "third".getBytes(StandardCharsets.UTF_8), 0);
+ ReferenceCountedEntry cached = null;
+ try {
+ assertThat(rangeEntryCache.insert(first)).isTrue();
+ cached = rangeEntryCache.getEntries().get(PositionFactory.create(1, 20));
+ // a later addition at the same position, as when two inserts race, isn't cached, but overrides the count
+ assertThat(rangeEntryCache.insert(second)).isFalse();
+ assertThat(new String(cached.getData(), StandardCharsets.UTF_8)).isEqualTo("first");
+ assertThat(cached.getReadCountHandler()).isSameAs(second.getReadCountHandler());
+ assertThat(cached.getReadCountHandler().getExpectedReadCount()).isEqualTo(4);
+ // also when it has no expected reads
+ assertThat(rangeEntryCache.insert(third)).isFalse();
+ assertThat(cached.hasExpectedReads()).isFalse();
+ // and the discarded duplicates' releases didn't count as reads of the additions' counts
+ assertThat(second.getReadCountHandler().getExpectedReadCount()).isEqualTo(4);
+ } finally {
+ if (cached != null) {
+ cached.release();
+ }
+ first.release();
+ second.release();
+ third.release();
+ rangeEntryCache.clear();
+ }
+ }
+
+ @Test
+ public void testBatchInsertSkipsCachedEntriesBeforeCopyingThem() {
+ // records the entries that the cache prepares for an insert: it computes their length before copying them
+ List<Long> prepared = new ArrayList<>();
+ rangeEntryCache = new RangeEntryCacheImpl(mockEntryCacheManager, mockManagedLedger, false,
+ mockRangeCacheRemovalQueue, (ml, entry) -> {
+ prepared.add(entry.getEntryId());
+ return entry.getLength();
+ }, pendingReadsManager);
+ EntryImpl cachedSource = EntryImpl.create(1, 11, "cached".getBytes(StandardCharsets.UTF_8));
+ assertThat(rangeEntryCache.insert(cachedSource)).isTrue();
+ ReferenceCountedEntry cached = rangeEntryCache.getEntries().get(PositionFactory.create(1, 11));
+ // the cached entry has no expected reads; the read's entries are expected to be read by 3 cursors
+ assertThat(cached.getReadCountHandler()).isNull();
+ List<Entry> read = new ArrayList<>();
+ for (long entryId = 10; entryId <= 12; entryId++) {
+ read.add(EntryImpl.create(1, entryId, ("read-" + entryId).getBytes(StandardCharsets.UTF_8), 3));
+ }
+ List<Entry> laterRead = List.of(EntryImpl.create(1, 11, "later".getBytes(StandardCharsets.UTF_8), 5));
+ try {
+ int sourceRefCnt = read.get(1).getDataBuffer().refCnt();
+ // a read from storage that overlaps the cached entry inserts the others, copying them
+ assertThat(rangeEntryCache.insert(read, true)).isEqualTo(2);
+ // the cached entry is immutable, so it's kept and the read's entry at its position isn't prepared or
+ // copied, but the cached entry takes the read's expected read count
+ assertThat(prepared).containsExactly(11L, 10L, 12L);
+ assertThat(cached.getReadCountHandler().getExpectedReadCount()).isEqualTo(3);
+ // and a later read's
+ assertThat(rangeEntryCache.insert(laterRead, true)).isZero();
+ assertThat(cached.getReadCountHandler().getExpectedReadCount()).isEqualTo(5);
+ ReferenceCountedEntry after = rangeEntryCache.getEntries().get(PositionFactory.create(1, 11));
+ try {
+ assertThat(after).isSameAs(cached);
+ assertThat(new String(after.getData(), StandardCharsets.UTF_8)).isEqualTo("cached");
+ } finally {
+ after.release();
+ }
+ assertThat(read.get(1).getDataBuffer().refCnt()).isEqualTo(sourceRefCnt);
+ assertThat(rangeEntryCache.getEntries().getNumberOfEntries()).isEqualTo(3);
+ } finally {
+ cached.release();
+ cachedSource.release();
+ read.forEach(Entry::release);
+ laterRead.forEach(Entry::release);
+ rangeEntryCache.clear();
+ }
+ }
+
@Test
public void testInsertDefersMetadataUntilFirstReadAndSharesIt() {
managedLedgerConfig.setPulsarMessageEntries(true);
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
index 563d49fa9044..34bdba03c6f5 100644
--- 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
@@ -64,9 +64,11 @@ public class CacheEvictionCycleBenchmark {
@Benchmark
public long insertAndEvict() {
+ // the batch's entries are consecutive, so they're inserted in order, as a read from storage inserts them
+ RangeCache.Inserter inserter = cache.newInserter();
for (EntryImpl entry : entries) {
entry.retain();
- if (!cache.put(entry.getPosition(), entry)) {
+ if (!inserter.put(entry.getPosition(), entry, entry.getLength())) {
entry.release();
throw new IllegalStateException("Duplicate cache entry");
}
diff --git a/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheCatchUpInsertBenchmark.java b/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheCatchUpInsertBenchmark.java
new file mode 100644
index 000000000000..de04a3815fe8
--- /dev/null
+++ b/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheCatchUpInsertBenchmark.java
@@ -0,0 +1,115 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.impl.EntryImpl;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Group;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+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 the managed ledger's tail inserts while a catch-up read inserts the entries that it read from storage into
+ * the same cache: the tail thread adds entries to the current ledger, and the catch-up thread inserts runs of 100
+ * consecutive entries at random positions of an earlier ledger and removes them again, as the eviction would.
+ */
+@State(Scope.Group)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 3, time = 2)
+@Measurement(iterations = 5, time = 2)
+@Fork(2)
+public class RangeCacheCatchUpInsertBenchmark {
+ private static final int CACHED_TAIL_ENTRIES = 10_000;
+ private static final int TRIM_BATCH = 100;
+ private static final int CATCH_UP_RUN = 100;
+ private static final long CATCH_UP_LEDGER = 1;
+ private static final long TAIL_LEDGER = 2;
+ private static final int CATCH_UP_LEDGER_ENTRIES = 50_000;
+
+ private final ByteBuf payload = Unpooled.wrappedBuffer(new byte[0]);
+ private RangeCacheRemovalQueue removalQueue;
+ private RangeCache cache;
+ private RangeCache.Inserter tailInserter;
+ private long tailEntryId;
+
+ @Setup(Level.Iteration)
+ public void setup() {
+ removalQueue = new RangeCacheRemovalQueue(0, false);
+ cache = new RangeCache(removalQueue);
+ tailInserter = cache.newInserter();
+ tailEntryId = 0;
+ for (int i = 0; i < CACHED_TAIL_ENTRIES; i++) {
+ insert(tailInserter, TAIL_LEDGER, tailEntryId++);
+ }
+ }
+
+ @Benchmark
+ @Group("catchUp")
+ public void tailInsert() {
+ insert(tailInserter, TAIL_LEDGER, tailEntryId++);
+ if (tailEntryId % TRIM_BATCH == 0) {
+ cache.removeRange(PositionFactory.create(TAIL_LEDGER, 0),
+ PositionFactory.create(TAIL_LEDGER, tailEntryId - CACHED_TAIL_ENTRIES), false);
+ // the eviction drops the removed entries' wrappers from the removal queue, so that they're recycled
+ removalQueue.evictLEntriesBeforeTimestamp(Long.MIN_VALUE);
+ }
+ }
+
+ @Benchmark
+ @Group("catchUp")
+ public void catchUpInsertRun() {
+ long first = ThreadLocalRandom.current().nextInt(CATCH_UP_LEDGER_ENTRIES - CATCH_UP_RUN);
+ // a read from storage inserts its entries with an inserter of its own
+ RangeCache.Inserter inserter = cache.newInserter();
+ for (int i = 0; i < CATCH_UP_RUN; i++) {
+ insert(inserter, CATCH_UP_LEDGER, first + i);
+ }
+ cache.removeRange(PositionFactory.create(CATCH_UP_LEDGER, first),
+ PositionFactory.create(CATCH_UP_LEDGER, first + CATCH_UP_RUN), false);
+ }
+
+ private void insert(RangeCache.Inserter inserter, long ledgerId, long entryId) {
+ EntryImpl entry = EntryImpl.create(ledgerId, entryId, payload);
+ if (!inserter.put(entry.getPosition(), entry, entry.getLength())) {
+ entry.release();
+ }
+ }
+
+ @TearDown(Level.Iteration)
+ public void tearDown() {
+ cache.clear();
+ // drops the removed entries' wrappers, so that they're recycled
+ removalQueue.evictLEntriesBeforeTimestamp(Long.MAX_VALUE);
+ }
+}
diff --git a/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheInsertBenchmark.java b/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheInsertBenchmark.java
new file mode 100644
index 000000000000..67ec9c2148e8
--- /dev/null
+++ b/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheInsertBenchmark.java
@@ -0,0 +1,109 @@
+/*
+ * 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 io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.bookkeeper.mledger.impl.EntryImpl;
+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 how a managed ledger's thread caches the entries that it adds: each insert goes to the tail of the cache,
+ * the cursors' progress trims its head in batches, the eviction recycles the removed entries' wrappers, and the
+ * ledger rolls over after a number of entries.
+ */
+@State(Scope.Thread)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 3, time = 2)
+@Measurement(iterations = 5, time = 2)
+@Fork(2)
+public class RangeCacheInsertBenchmark {
+ private static final int TRIM_BATCH = 100;
+ private static final int ENTRIES_PER_LEDGER = 50_000;
+
+ @Param({"1000", "100000"})
+ public int cachedEntries;
+
+ private final ByteBuf payload = Unpooled.wrappedBuffer(new byte[0]);
+ private RangeCacheRemovalQueue removalQueue;
+ private RangeCache cache;
+ private RangeCache.Inserter inserter;
+ private long ledgerId;
+ private long entryId;
+ private long sequence;
+
+ @Setup
+ public void setup() {
+ removalQueue = new RangeCacheRemovalQueue(0, false);
+ cache = new RangeCache(removalQueue);
+ // the managed ledger inserts its added entries with one inserter
+ inserter = cache.newInserter();
+ for (int i = 0; i < cachedEntries; i++) {
+ insert();
+ }
+ }
+
+ @Benchmark
+ public boolean insertAtTail() {
+ boolean inserted = insert();
+ if (sequence % TRIM_BATCH == 0) {
+ // the cursors moved past the oldest entries, as invalidateEntries does
+ long first = sequence - cachedEntries;
+ cache.removeRange(PositionFactory.create(-1, 0),
+ PositionFactory.create(first / ENTRIES_PER_LEDGER, first % ENTRIES_PER_LEDGER), false);
+ // the eviction drops the removed entries' wrappers from the removal queue, so that they're recycled;
+ // the cached entries are newer than the timestamp, which stops it at the first of them
+ removalQueue.evictLEntriesBeforeTimestamp(Long.MIN_VALUE);
+ }
+ return inserted;
+ }
+
+ private boolean insert() {
+ EntryImpl entry = EntryImpl.create(ledgerId, entryId, payload);
+ boolean inserted = inserter.put(entry.getPosition(), entry, entry.getLength());
+ if (!inserted) {
+ entry.release();
+ }
+ sequence++;
+ if (++entryId == ENTRIES_PER_LEDGER) {
+ ledgerId++;
+ entryId = 0;
+ }
+ return inserted;
+ }
+
+ @TearDown
+ public void tearDown() {
+ cache.clear();
+ }
+}