diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java index 05446b7ef1c3e..29250727f7bfe 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java @@ -2754,22 +2754,38 @@ private void updateLastMarkDeleteEntryToLatest(final Position newPosition, * Given a list of entries, filter out the entries that have already been individually deleted. * * @param entries - * a list of entries + * a non-empty list of entries ordered by position; read paths normally return entries from one ledger * @return a list of entries not containing deleted messages */ List filterReadEntries(List entries) { lock.readLock().lock(); try { - Range entriesRange = Range.closed(entries.get(0).getPosition(), - entries.get(entries.size() - 1).getPosition()); + Entry firstEntry = entries.get(0); + Entry lastEntry = entries.get(entries.size() - 1); + long firstLedgerId = firstEntry.getLedgerId(); + long firstEntryId = firstEntry.getEntryId(); + long lastLedgerId = lastEntry.getLedgerId(); + long lastEntryId = lastEntry.getEntryId(); log.debug() - .attr("entriesRange", entriesRange) + .attr("firstLedgerId", firstLedgerId) + .attr("firstEntryId", firstEntryId) + .attr("lastLedgerId", lastLedgerId) + .attr("lastEntryId", lastEntryId) .attr("deletedMessages", individualDeletedMessages) .log("Filtering entries"); - Range span = individualDeletedMessages.isEmpty() ? null : individualDeletedMessages.span(); - if (span == null || !entriesRange.isConnected(span)) { + // Read batches are ordered and normally belong to one ledger. For an unexpected cross-ledger or + // descending batch, conservatively retain per-entry filtering. + boolean containsDeletedMessages = firstLedgerId != lastLedgerId + || firstEntryId > lastEntryId + || individualDeletedMessages.containsAny(firstLedgerId, firstEntryId, lastEntryId); + if (!containsDeletedMessages) { // There are no individually deleted messages in this entry list, no need to perform filtering - log.debug().attr("entriesRange", entriesRange).log("No filtering needed for entries"); + log.debug() + .attr("firstLedgerId", firstLedgerId) + .attr("firstEntryId", firstEntryId) + .attr("lastLedgerId", lastLedgerId) + .attr("lastEntryId", lastEntryId) + .log("No filtering needed for entries"); return entries; } else { // Remove from the entry list all the entries that were already marked for deletion diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionRangeSet.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionRangeSet.java index 01e581eb444a1..6d7900cbcd2fc 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionRangeSet.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionRangeSet.java @@ -136,6 +136,25 @@ public boolean contains(long ledgerId, long entryId) { return false; } + /** + * Returns {@code true} if any entry in the closed range {@code [lowerEntryId, upperEntryId]} of + * {@code ledgerId} is present in this set. Both bounds are inclusive (matching + * {@link #cardinality}), unlike the open lower bound of {@link #addOpenClosed}. A negative + * {@code lowerEntryId} is treated as {@code 0}. Returns {@code false} for an empty range or an + * absent ledger. + */ + boolean containsAny(long ledgerId, long lowerEntryId, long upperEntryId) { + if (rangeBitmapMap.isEmpty() || lowerEntryId > upperEntryId) { + return false; + } + LongBitmap bitmap = rangeBitmapMap.get(ledgerId); + if (bitmap == null) { + return false; + } + long nextPresentEntryId = bitmap.nextPresentValue(Math.max(0, lowerEntryId)); + return nextPresentEntryId != -1 && nextPresentEntryId <= upperEntryId; + } + @Override public Range rangeContaining(long ledgerId, long entryId) { LongBitmap rangeBitmap = rangeBitmapMap.get(ledgerId); diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java index a6a0dc0411b03..3d3b09299f1cc 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java @@ -5487,6 +5487,134 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { ledger.close(); } + @Test + public void testFilterReadEntriesSkipsFilteringForGapBetweenIndividualAcks() throws Exception { + @Cleanup + ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open( + "testFilterReadEntriesSkipsFilteringForGapBetweenIndividualAcks"); + @Cleanup + ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c"); + + List positions = new ArrayList<>(); + for (int i = 0; i <= 100; i++) { + positions.add(ledger.addEntry(new byte[]{1})); + } + cursor.delete(Set.of(positions.get(1), positions.get(100))); + + List entries = new ArrayList<>(); + for (int i = 40; i < 50; i++) { + Position position = positions.get(i); + Entry entry = mock(Entry.class); + when(entry.getPosition()).thenReturn(position); + when(entry.getLedgerId()).thenReturn(position.getLedgerId()); + when(entry.getEntryId()).thenReturn(position.getEntryId()); + entries.add(entry); + } + + Range entriesRange = Range.closed(entries.get(0).getPosition(), + entries.get(entries.size() - 1).getPosition()); + assertThat(entriesRange.isConnected(cursor.getIndividuallyDeletedMessagesSet().span())).isTrue(); + for (Entry entry : entries) { + assertThat(cursor.getIndividuallyDeletedMessagesSet() + .contains(entry.getLedgerId(), entry.getEntryId())).isFalse(); + } + + List filteredEntries = cursor.filterReadEntries(entries); + + assertThat(filteredEntries).isSameAs(entries); + for (Entry entry : entries) { + verify(entry, never()).release(); + } + } + + @Test + public void testFilterReadEntriesFallsBackForCrossLedgerBatch() throws Exception { + @Cleanup + ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open( + "testFilterReadEntriesFallsBackForCrossLedgerBatch"); + @Cleanup + ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c"); + + // Production read paths currently return ordered batches from a single ledger, so this cross-ledger batch + // is not expected in normal operation. If one is ever passed in, the conservative fallback must inspect + // every entry, remove and release the acknowledged entry, and retain the unacknowledged entries. + Position firstPosition = PositionFactory.create(1, 0); + Position deletedPosition = PositionFactory.create(2, 0); + Position lastPosition = PositionFactory.create(2, 1); + cursor.lock.writeLock().lock(); + try { + cursor.getIndividuallyDeletedMessagesSet().addOpenClosed( + deletedPosition.getLedgerId(), deletedPosition.getEntryId() - 1, + deletedPosition.getLedgerId(), deletedPosition.getEntryId()); + } finally { + cursor.lock.writeLock().unlock(); + } + + Entry firstEntry = mock(Entry.class); + when(firstEntry.getPosition()).thenReturn(firstPosition); + when(firstEntry.getLedgerId()).thenReturn(firstPosition.getLedgerId()); + when(firstEntry.getEntryId()).thenReturn(firstPosition.getEntryId()); + Entry deletedEntry = mock(Entry.class); + when(deletedEntry.getPosition()).thenReturn(deletedPosition); + when(deletedEntry.getLedgerId()).thenReturn(deletedPosition.getLedgerId()); + when(deletedEntry.getEntryId()).thenReturn(deletedPosition.getEntryId()); + Entry lastEntry = mock(Entry.class); + when(lastEntry.getPosition()).thenReturn(lastPosition); + when(lastEntry.getLedgerId()).thenReturn(lastPosition.getLedgerId()); + when(lastEntry.getEntryId()).thenReturn(lastPosition.getEntryId()); + + List filteredEntries = cursor.filterReadEntries(List.of(firstEntry, deletedEntry, lastEntry)); + + assertThat(filteredEntries).containsExactly(firstEntry, lastEntry); + verify(firstEntry, never()).release(); + verify(deletedEntry).release(); + verify(lastEntry, never()).release(); + } + + @Test + public void testFilterReadEntriesFallsBackForDescendingBatch() throws Exception { + @Cleanup + ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open( + "testFilterReadEntriesFallsBackForDescendingBatch"); + @Cleanup + ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c"); + + // Production read paths currently return ordered batches, so this descending batch is not expected in + // normal operation. If one is ever passed in, the conservative fallback must inspect every entry, remove + // and release the acknowledged entry, and retain the unacknowledged entries. + Position firstPosition = PositionFactory.create(1, 10); + Position deletedPosition = PositionFactory.create(1, 5); + Position lastPosition = PositionFactory.create(1, 1); + cursor.lock.writeLock().lock(); + try { + cursor.getIndividuallyDeletedMessagesSet().addOpenClosed( + deletedPosition.getLedgerId(), deletedPosition.getEntryId() - 1, + deletedPosition.getLedgerId(), deletedPosition.getEntryId()); + } finally { + cursor.lock.writeLock().unlock(); + } + + Entry firstEntry = mock(Entry.class); + when(firstEntry.getPosition()).thenReturn(firstPosition); + when(firstEntry.getLedgerId()).thenReturn(firstPosition.getLedgerId()); + when(firstEntry.getEntryId()).thenReturn(firstPosition.getEntryId()); + Entry deletedEntry = mock(Entry.class); + when(deletedEntry.getPosition()).thenReturn(deletedPosition); + when(deletedEntry.getLedgerId()).thenReturn(deletedPosition.getLedgerId()); + when(deletedEntry.getEntryId()).thenReturn(deletedPosition.getEntryId()); + Entry lastEntry = mock(Entry.class); + when(lastEntry.getPosition()).thenReturn(lastPosition); + when(lastEntry.getLedgerId()).thenReturn(lastPosition.getLedgerId()); + when(lastEntry.getEntryId()).thenReturn(lastPosition.getEntryId()); + + List filteredEntries = cursor.filterReadEntries(List.of(firstEntry, deletedEntry, lastEntry)); + + assertThat(filteredEntries).containsExactly(firstEntry, lastEntry); + verify(firstEntry, never()).release(); + verify(deletedEntry).release(); + verify(lastEntry, never()).release(); + } + @Test public void testReadEntriesWithSkipDeletedEntries() throws Exception { @Cleanup diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/PositionRangeSetTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/PositionRangeSetTest.java index cccb49678e682..dc9636f3b60ca 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/PositionRangeSetTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/PositionRangeSetTest.java @@ -29,6 +29,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.Random; import java.util.Set; import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.PositionFactory; @@ -235,6 +236,114 @@ public void testSpanWithGuava() { assertEquals(set.span(), Range.openClosed(pos(0, 97), pos(4, 20))); } + @Test + public void testContainsAny() { + PositionRangeSet set = newSet(); + set.addOpenClosed(1, 9, 1, 10); + set.addOpenClosed(1, 999, 1, 1000); + set.addOpenClosed(3, 19, 3, 20); + + assertFalse(set.containsAny(1, 0, 9)); + assertTrue(set.containsAny(1, 0, 10)); + assertFalse(set.containsAny(1, 11, 999)); + assertTrue(set.containsAny(1, 11, 1000)); + assertFalse(set.containsAny(1, 1001, 2000)); + assertFalse(set.containsAny(2, 0, 1000)); + assertTrue(set.containsAny(3, 20, 20)); + assertFalse(set.containsAny(3, 21, 20)); + } + + @Test + public void testContainsAnyBoundaries() { + PositionRangeSet set = newSet(); + long maxEntryId = Integer.MAX_VALUE; + + assertFalse(set.containsAny(1, 0, 0)); + + set.addOpenClosed(1, -1, 1, 0); + set.addOpenClosed(1, 9, 1, 10); + set.addOpenClosed(1, maxEntryId - 1, 1, maxEntryId); + + assertTrue(set.containsAny(1, 0, 0)); + assertTrue(set.containsAny(1, -1, 0)); + assertFalse(set.containsAny(1, -5, -1)); + assertFalse(set.containsAny(1, 1, 9)); + assertTrue(set.containsAny(1, 9, 10)); + assertTrue(set.containsAny(1, 10, 11)); + assertFalse(set.containsAny(1, 11, maxEntryId - 1)); + assertTrue(set.containsAny(1, maxEntryId - 1, maxEntryId)); + assertTrue(set.containsAny(1, maxEntryId, maxEntryId)); + assertFalse(set.containsAny(1, maxEntryId + 1, maxEntryId + 1)); + } + + @Test + public void testContainsAnyRandomizedAgainstBooleanOracle() { + int ledgerCount = 16; + int entriesPerLedger = 4096; + int seedsPerDensity = 32; + int queriesPerSeed = 500; + for (int density : new int[]{0, 1, 10, 50, 90, 100}) { + for (int seedIndex = 0; seedIndex < seedsPerDensity; seedIndex++) { + long seed = 0x5EEDL + density * 1_000_003L + seedIndex * 104_729L; + Random random = new Random(seed); + boolean[][] deletedEntries = new boolean[ledgerCount][entriesPerLedger]; + PositionRangeSet set = newSet(); + + for (int ledgerId = 0; ledgerId < ledgerCount; ledgerId++) { + boolean ledgerHasDeletedEntries = random.nextInt(4) != 0; + for (int entryId = 0; entryId < entriesPerLedger; entryId++) { + deletedEntries[ledgerId][entryId] = ledgerHasDeletedEntries + && random.nextInt(100) < density; + } + addDeletedRanges(set, deletedEntries[ledgerId], ledgerId); + } + + for (int query = 0; query < queriesPerSeed; query++) { + int ledgerId = random.nextInt(10) == 0 + ? ledgerCount + random.nextInt(2) : random.nextInt(ledgerCount); + int firstEntryId = random.nextInt(entriesPerLedger); + int batchSize = 1 + random.nextInt(100); + int lastEntryId = Math.min(entriesPerLedger - 1, firstEntryId + batchSize - 1); + + long expectedNextEntryId = nextDeletedEntryId(deletedEntries, ledgerId, firstEntryId); + boolean expected = expectedNextEntryId != -1 && expectedNextEntryId <= lastEntryId; + boolean actual = set.containsAny(ledgerId, firstEntryId, lastEntryId); + assertEquals(actual, expected, + "seed=" + seed + ", query=" + query + ", range=" + + ledgerId + ":" + firstEntryId + ".." + ledgerId + ":" + lastEntryId); + } + } + } + } + + private static void addDeletedRanges(PositionRangeSet set, boolean[] deletedEntries, long ledgerId) { + int entryId = 0; + while (entryId < deletedEntries.length) { + while (entryId < deletedEntries.length && !deletedEntries[entryId]) { + entryId++; + } + int firstDeletedEntryId = entryId; + while (entryId < deletedEntries.length && deletedEntries[entryId]) { + entryId++; + } + if (firstDeletedEntryId < entryId) { + set.addOpenClosed(ledgerId, firstDeletedEntryId - 1L, ledgerId, entryId - 1L); + } + } + } + + private static long nextDeletedEntryId(boolean[][] deletedEntries, int ledgerId, int firstEntryId) { + if (ledgerId < 0 || ledgerId >= deletedEntries.length) { + return -1; + } + for (int entryId = firstEntryId; entryId < deletedEntries[ledgerId].length; entryId++) { + if (deletedEntries[ledgerId][entryId]) { + return entryId; + } + } + return -1; + } + @Test public void testFirstRange() { PositionRangeSet set = newSet(); diff --git a/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/IndividualAckReadFilterBenchmark.java b/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/IndividualAckReadFilterBenchmark.java new file mode 100644 index 0000000000000..b0073c144ae73 --- /dev/null +++ b/microbench/src/main/java/org/apache/bookkeeper/mledger/impl/IndividualAckReadFilterBenchmark.java @@ -0,0 +1,178 @@ +/* + * 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; + +import com.google.common.collect.Collections2; +import com.google.common.collect.Lists; +import com.google.common.collect.Range; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.apache.bookkeeper.mledger.Position; +import org.apache.bookkeeper.mledger.PositionFactory; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +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.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; + +/** + * Compares the old span-based and new bitmap-based individual-ack checks on the managed-ledger read path. + * A returned BookKeeper batch is a contiguous range from one ledger, although one cursor read can issue + * several such reads while moving across ledgers. + * + *

The {@code *Decision} benchmarks isolate the fast-path lookup. Their inputs are prepared in + * {@link Setup}, so they exclude surrounding read-path work such as retrieving the batch endpoints and + * logging; the reported decision time is not end-to-end fast-path latency. The {@code *ThenFilter} + * benchmarks include the existing per-entry membership checks and intermediate list allocation whenever + * the lookup reports a possible individual ack. Run with {@code -prof gc} to compare allocation rates. + * + *

Run with: + * 

{@code
+ * ./gradlew :microbench:shadowJar
+ * java -jar microbench/build/libs/microbench-*-benchmarks.jar \
+ *   IndividualAckReadFilterBenchmark -prof gc
+ * }
+ */ +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@State(Scope.Benchmark) +@Warmup(time = 1, iterations = 3, timeUnit = TimeUnit.SECONDS) +@Measurement(time = 1, iterations = 5, timeUnit = TimeUnit.SECONDS) +@Fork(2) +public class IndividualAckReadFilterBenchmark { + + private static final long TARGET_LEDGER_ID = 100; + private static final long ENTRIES_PER_LEDGER = 100_000; + + @Param({"1", "10", "100", "1000"}) + public int batchSize; + + @Param({"SPARSE_GAP", "HIT_FIRST", "HIT_LAST", "HIT_EVERY_TEN", "HIT_ALL", + "OUTSIDE_SPAN", "MISSING_LEDGER", "EMPTY_SET"}) + public String scenario; + + private PositionRangeSet individuallyDeletedMessages; + private List entries; + private Range entriesRange; + private long firstEntryId; + private long lastEntryId; + + @Setup(Level.Trial) + public void setup() { + individuallyDeletedMessages = new PositionRangeSet(PositionFactory::create, false); + firstEntryId = ENTRIES_PER_LEDGER / 2; + lastEntryId = firstEntryId + batchSize - 1; + + switch (scenario) { + case "SPARSE_GAP" -> { + addDeletedEntry(TARGET_LEDGER_ID, 0); + addDeletedEntry(TARGET_LEDGER_ID, ENTRIES_PER_LEDGER - 1); + } + case "HIT_FIRST" -> addDeletedEntry(TARGET_LEDGER_ID, firstEntryId); + case "HIT_LAST" -> addDeletedEntry(TARGET_LEDGER_ID, lastEntryId); + case "HIT_EVERY_TEN" -> { + for (long entryId = firstEntryId; entryId <= lastEntryId; entryId += 10) { + addDeletedEntry(TARGET_LEDGER_ID, entryId); + } + } + case "HIT_ALL" -> individuallyDeletedMessages.addOpenClosed( + TARGET_LEDGER_ID, firstEntryId - 1, TARGET_LEDGER_ID, lastEntryId); + case "OUTSIDE_SPAN" -> addDeletedEntry(TARGET_LEDGER_ID, 0); + case "MISSING_LEDGER" -> { + addDeletedEntry(TARGET_LEDGER_ID - 1, ENTRIES_PER_LEDGER - 1); + addDeletedEntry(TARGET_LEDGER_ID + 1, 0); + } + case "EMPTY_SET" -> { + // No individually deleted entries. + } + default -> throw new IllegalArgumentException("Unknown scenario: " + scenario); + } + + entries = new ArrayList<>(batchSize); + for (long entryId = firstEntryId; entryId <= lastEntryId; entryId++) { + entries.add(PositionFactory.create(TARGET_LEDGER_ID, entryId)); + } + entriesRange = Range.closed(entries.get(0), entries.get(entries.size() - 1)); + + List oldResult = filterWithSpan(); + List newResult = filterWithContainsAny(); + if (!oldResult.equals(newResult)) { + throw new IllegalStateException( + "Old and new filters disagree for " + scenario + ", batchSize=" + batchSize); + } + } + + @Benchmark + @Threads(1) + public boolean oldSpanDecision() { + Range span = individuallyDeletedMessages.isEmpty() ? null : individuallyDeletedMessages.span(); + return span != null && entriesRange.isConnected(span); + } + + @Benchmark + @Threads(1) + public boolean newContainsAnyDecision() { + return individuallyDeletedMessages.containsAny(TARGET_LEDGER_ID, firstEntryId, lastEntryId); + } + + @Benchmark + @Threads(1) + public List oldSpanThenFilter() { + return filterWithSpan(); + } + + @Benchmark + @Threads(1) + public List newContainsAnyThenFilter() { + return filterWithContainsAny(); + } + + private List filterWithSpan() { + Range span = individuallyDeletedMessages.isEmpty() ? null : individuallyDeletedMessages.span(); + if (span == null || !entriesRange.isConnected(span)) { + return entries; + } + return filterIndividuallyDeletedEntries(); + } + + private List filterWithContainsAny() { + if (!individuallyDeletedMessages.containsAny(TARGET_LEDGER_ID, firstEntryId, lastEntryId)) { + return entries; + } + return filterIndividuallyDeletedEntries(); + } + + private List filterIndividuallyDeletedEntries() { + return Lists.newArrayList(Collections2.filter(entries, position -> + !individuallyDeletedMessages.contains(position.getLedgerId(), position.getEntryId()))); + } + + private void addDeletedEntry(long ledgerId, long entryId) { + individuallyDeletedMessages.addOpenClosed(ledgerId, entryId - 1, ledgerId, entryId); + } +}