From 5544ed2861d5cd60e953adc811c74498493d3f1d Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Fri, 22 Jul 2022 10:47:54 -0700 Subject: [PATCH 01/10] PIP-174: New managed ledger entry cache implementation --- conf/broker.conf | 6 + conf/standalone.conf | 6 + .../mledger/ManagedLedgerFactoryConfig.java | 12 + .../impl/ManagedLedgerFactoryImpl.java | 12 +- .../impl/cache/EntryCacheDisabled.java | 4 +- .../mledger/impl/cache/EntryCacheManager.java | 36 +- .../impl/cache/RangeEntryCacheImpl.java | 6 +- .../cache/RangeEntryCacheManagerImpl.java | 32 +- .../impl/cache/SharedCacheSegment.java | 35 + .../cache/SharedCacheSegmentBufferCopy.java | 99 +++ .../SharedCacheSegmentBufferRefCount.java | 91 +++ .../impl/cache/SharedEntryCacheImpl.java | 216 ++++++ .../cache/SharedEntryCacheManagerImpl.java | 218 ++++++ .../mledger/impl/EntryCacheManagerTest.java | 83 ++- .../mledger/impl/ManagedCursorTest.java | 2 +- .../mledger/impl/ManagedLedgerTest.java | 2 + .../test/MockedBookKeeperTestCase.java | 6 +- microbenchmarks/pom.xml | 117 ++++ .../cache/ManagedLedgerCacheBenchmark.java | 147 ++++ pom.xml | 15 + .../pulsar/broker/ServiceConfiguration.java | 9 + .../broker/ManagedLedgerClientFactory.java | 1 + .../nonpersistent/NonPersistentTopic.java | 6 +- .../ConcurrentLongLongPairHashMap.java | 55 ++ .../ConcurrentLongPairObjectHashMap.java | 647 ++++++++++++++++++ site2/docs/reference-configuration.md | 1 + 26 files changed, 1784 insertions(+), 80 deletions(-) create mode 100644 managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegment.java create mode 100644 managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java create mode 100644 managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferRefCount.java create mode 100644 managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java create mode 100644 managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheManagerImpl.java create mode 100644 microbenchmarks/pom.xml create mode 100644 microbenchmarks/src/main/java/org/apache/bookkeeper/mledger/impl/cache/ManagedLedgerCacheBenchmark.java create mode 100644 pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairObjectHashMap.java diff --git a/conf/broker.conf b/conf/broker.conf index c19afe981cabc..8392cd7721042 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -1008,6 +1008,12 @@ managedLedgerCacheSizeMB= # Whether we should make a copy of the entry payloads when inserting in cache managedLedgerCacheCopyEntries=false +# The class name for the implementation of ManagedLedger cache manager component. +# Options are: +# - org.apache.bookkeeper.mledger.impl.cache.SharedEntryCacheManagerImpl +# - org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheManagerImpl +managedLedgerCacheManagerImplementationClass=org.apache.bookkeeper.mledger.impl.cache.SharedEntryCacheManagerImpl + # Threshold to which bring down the cache level when eviction is triggered managedLedgerCacheEvictionWatermark=0.9 diff --git a/conf/standalone.conf b/conf/standalone.conf index b3281c0273794..fdaae830e5417 100644 --- a/conf/standalone.conf +++ b/conf/standalone.conf @@ -680,6 +680,12 @@ managedLedgerCacheSizeMB= # Whether we should make a copy of the entry payloads when inserting in cache managedLedgerCacheCopyEntries=false +# The class name for the implementation of ManagedLedger cache manager component. +# Options are: +# - org.apache.bookkeeper.mledger.impl.cache.SharedEntryCacheManagerImpl +# - org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheManagerImpl +managedLedgerCacheManagerImplementationClass=org.apache.bookkeeper.mledger.impl.cache.SharedEntryCacheManagerImpl + # Threshold to which bring down the cache level when eviction is triggered managedLedgerCacheEvictionWatermark=0.9 diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactoryConfig.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactoryConfig.java index 25fcb377e3e11..e8aeb69521d74 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactoryConfig.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerFactoryConfig.java @@ -21,6 +21,7 @@ import lombok.Data; import org.apache.bookkeeper.common.annotation.InterfaceAudience; import org.apache.bookkeeper.common.annotation.InterfaceStability; +import org.apache.bookkeeper.mledger.impl.cache.SharedEntryCacheManagerImpl; import org.apache.bookkeeper.mledger.proto.MLDataFormats; /** @@ -91,4 +92,15 @@ public class ManagedLedgerFactoryConfig { * ManagedCursorInfo compression type. If the compression type is null or invalid, don't compress data. */ private String managedCursorInfoCompressionType = MLDataFormats.CompressionType.NONE.name(); + + /** + * Class name for the implementation of {@link org.apache.bookkeeper.mledger.impl.cache.EntryCacheManager}. + * + * Options are: + * + */ + private String entryCacheManagerClassName = SharedEntryCacheManagerImpl.class.getName(); } diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java index 8e3271a03934e..80cce93e1d085 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java @@ -69,7 +69,6 @@ import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.State; import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback; import org.apache.bookkeeper.mledger.impl.cache.EntryCacheManager; -import org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheManagerImpl; import org.apache.bookkeeper.mledger.proto.MLDataFormats; import org.apache.bookkeeper.mledger.proto.MLDataFormats.LongProperty; import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo; @@ -193,7 +192,11 @@ private ManagedLedgerFactoryImpl(MetadataStoreExtended metadataStore, config.getManagedCursorInfoCompressionType()); this.config = config; this.mbean = new ManagedLedgerFactoryMBeanImpl(this); - this.entryCacheManager = new RangeEntryCacheManagerImpl(this); + + Class ecmClass = + (Class) Class.forName(config.getEntryCacheManagerClassName()); + this.entryCacheManager = ecmClass.getDeclaredConstructor(ManagedLedgerFactoryImpl.class).newInstance(this); + this.statsTask = scheduledExecutor.scheduleWithFixedDelay(catchingAndLoggingThrowables(this::refreshStats), 0, config.getStatsPeriodSeconds(), TimeUnit.SECONDS); this.flushCursorsTask = scheduledExecutor.scheduleAtFixedRate(catchingAndLoggingThrowables(this::flushCursors), @@ -592,7 +595,8 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) { })); } })); - entryCacheManager.clear(); + + entryCacheManager.close(); return FutureUtil.waitForAll(futures).thenAccept(__ -> { //wait for tasks in scheduledExecutor executed. scheduledExecutor.shutdown(); @@ -653,7 +657,7 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) { scheduledExecutor.shutdownNow(); - entryCacheManager.clear(); + entryCacheManager.close(); } @Override diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabled.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabled.java index a09b8ba27fc24..3bba6a163a013 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabled.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabled.java @@ -87,7 +87,7 @@ public void asyncReadEntry(ReadHandle lh, long firstEntry, long lastEntry, boole try { for (LedgerEntry e : ledgerEntries) { // Insert the entries at the end of the list (they will be unsorted for now) - EntryImpl entry = RangeEntryCacheManagerImpl.create(e, interceptor); + EntryImpl entry = EntryCacheManager.create(e, interceptor); entries.add(entry); totalSize += entry.getLength(); } @@ -119,7 +119,7 @@ public void asyncReadEntry(ReadHandle lh, PositionImpl position, AsyncCallbacks. Iterator iterator = ledgerEntries.iterator(); if (iterator.hasNext()) { LedgerEntry ledgerEntry = iterator.next(); - EntryImpl returnEntry = RangeEntryCacheManagerImpl.create(ledgerEntry, interceptor); + EntryImpl returnEntry = EntryCacheManager.create(ledgerEntry, interceptor); ml.getFactory().getMbean().recordCacheMiss(1, returnEntry.getLength()); ml.getMbean().addReadEntriesSample(1, returnEntry.getLength()); diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheManager.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheManager.java index 12cbb023f8691..dacb52a1d758a 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheManager.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheManager.java @@ -18,9 +18,15 @@ */ package org.apache.bookkeeper.mledger.impl.cache; +import io.netty.buffer.ByteBuf; +import org.apache.bookkeeper.client.api.LedgerEntry; +import org.apache.bookkeeper.client.impl.LedgerEntryImpl; +import org.apache.bookkeeper.mledger.Entry; +import org.apache.bookkeeper.mledger.impl.EntryImpl; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; +import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor; -public interface EntryCacheManager { +public interface EntryCacheManager extends AutoCloseable { EntryCache getEntryCache(ManagedLedgerImpl ml); void removeEntryCache(String name); @@ -36,4 +42,32 @@ public interface EntryCacheManager { void updateCacheEvictionWatermark(double cacheEvictionWatermark); double getCacheEvictionWatermark(); + + static Entry create(long ledgerId, long entryId, ByteBuf data) { + return EntryImpl.create(ledgerId, entryId, data); + } + + static EntryImpl create(LedgerEntry ledgerEntry, ManagedLedgerInterceptor interceptor) { + ManagedLedgerInterceptor.PayloadProcessorHandle processorHandle = null; + if (interceptor != null) { + ByteBuf duplicateBuffer = ledgerEntry.getEntryBuffer().retainedDuplicate(); + processorHandle = interceptor + .processPayloadBeforeEntryCache(duplicateBuffer); + if (processorHandle != null) { + ledgerEntry = LedgerEntryImpl.create(ledgerEntry.getLedgerId(), ledgerEntry.getEntryId(), + ledgerEntry.getLength(), processorHandle.getProcessedPayload()); + } else { + duplicateBuffer.release(); + } + } + EntryImpl returnEntry = EntryImpl.create(ledgerEntry); + if (processorHandle != null) { + processorHandle.release(); + ledgerEntry.close(); + } + return returnEntry; + } + + @Override + void close(); } 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 d4831b3a0fc9a..99df7c037bbd5 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 @@ -50,7 +50,7 @@ public class RangeEntryCacheImpl implements EntryCache { private final RangeEntryCacheManagerImpl manager; private final ManagedLedgerImpl ml; - private ManagedLedgerInterceptor interceptor; + private final ManagedLedgerInterceptor interceptor; private final RangeCache entries; private final boolean copyEntries; @@ -221,7 +221,7 @@ private void asyncReadEntry0(ReadHandle lh, PositionImpl position, final ReadEnt Iterator iterator = ledgerEntries.iterator(); if (iterator.hasNext()) { LedgerEntry ledgerEntry = iterator.next(); - EntryImpl returnEntry = RangeEntryCacheManagerImpl.create(ledgerEntry, interceptor); + EntryImpl returnEntry = EntryCacheManager.create(ledgerEntry, interceptor); manager.mlFactoryMBean.recordCacheMiss(1, returnEntry.getLength()); ml.getMbean().addReadEntriesSample(1, returnEntry.getLength()); @@ -306,7 +306,7 @@ private void asyncReadEntry0(ReadHandle lh, long firstEntry, long lastEntry, boo long totalSize = 0; final List entriesToReturn = Lists.newArrayListWithExpectedSize(entriesToRead); for (LedgerEntry e : ledgerEntries) { - EntryImpl entry = RangeEntryCacheManagerImpl.create(e, interceptor); + EntryImpl entry = EntryCacheManager.create(e, interceptor); entriesToReturn.add(entry); totalSize += entry.getLength(); if (shouldCacheEntry) { diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheManagerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheManagerImpl.java index 4c27781b1f010..69e99173827a1 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheManagerImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheManagerImpl.java @@ -21,19 +21,13 @@ import static org.apache.bookkeeper.mledger.util.SafeRun.safeRun; import com.google.common.collect.Lists; import com.google.common.collect.Maps; -import io.netty.buffer.ByteBuf; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; -import org.apache.bookkeeper.client.api.LedgerEntry; -import org.apache.bookkeeper.client.impl.LedgerEntryImpl; -import org.apache.bookkeeper.mledger.Entry; -import org.apache.bookkeeper.mledger.impl.EntryImpl; import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl; import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryMBeanImpl; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; -import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -168,29 +162,9 @@ public void clear() { caches.values().forEach(EntryCache::clear); } - public static Entry create(long ledgerId, long entryId, ByteBuf data) { - return EntryImpl.create(ledgerId, entryId, data); - } - - public static EntryImpl create(LedgerEntry ledgerEntry, ManagedLedgerInterceptor interceptor) { - ManagedLedgerInterceptor.PayloadProcessorHandle processorHandle = null; - if (interceptor != null) { - ByteBuf duplicateBuffer = ledgerEntry.getEntryBuffer().retainedDuplicate(); - processorHandle = interceptor - .processPayloadBeforeEntryCache(duplicateBuffer); - if (processorHandle != null) { - ledgerEntry = LedgerEntryImpl.create(ledgerEntry.getLedgerId(), ledgerEntry.getEntryId(), - ledgerEntry.getLength(), processorHandle.getProcessedPayload()); - } else { - duplicateBuffer.release(); - } - } - EntryImpl returnEntry = EntryImpl.create(ledgerEntry); - if (processorHandle != null) { - processorHandle.release(); - ledgerEntry.close(); - } - return returnEntry; + @Override + public void close() { + clear(); } private static final Logger log = LoggerFactory.getLogger(RangeEntryCacheManagerImpl.class); diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegment.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegment.java new file mode 100644 index 0000000000000..c7d4102abcb90 --- /dev/null +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegment.java @@ -0,0 +1,35 @@ +/** + * 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; + +public interface SharedCacheSegment extends AutoCloseable { + + boolean insert(long ledgerId, long entryId, ByteBuf entry); + + ByteBuf get(long ledgerId, long entryId); + + int getSize(); + + void clear(); + + @Override + void close(); +} diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java new file mode 100644 index 0000000000000..93dc0083fdd3d --- /dev/null +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java @@ -0,0 +1,99 @@ +/** + * 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 java.util.concurrent.atomic.AtomicInteger; +import org.apache.pulsar.common.allocator.PulsarByteBufAllocator; +import org.apache.pulsar.common.util.collections.ConcurrentLongLongPairHashMap; + +class SharedCacheSegmentBufferCopy implements AutoCloseable, SharedCacheSegment { + + private final ByteBuf cacheBuffer; + private final AtomicInteger currentOffset = new AtomicInteger(); + private final ConcurrentLongLongPairHashMap index; + private final int segmentSize; + + private static final int ALIGN_64_MASK = ~(64 - 1); + + SharedCacheSegmentBufferCopy(int segmentSize) { + this.segmentSize = segmentSize; + this.cacheBuffer = PulsarByteBufAllocator.DEFAULT.buffer(segmentSize, segmentSize); + this.cacheBuffer.writerIndex(segmentSize - 1); + this.index = ConcurrentLongLongPairHashMap.newBuilder() + // We are going to often clear() the map, with the expectation that it's going to get filled again + // immediately after. In these conditions it does not make sense to shrink it each time. + .autoShrink(false) + .concurrencyLevel(Runtime.getRuntime().availableProcessors() * 8) + .build(); + } + + @Override + public boolean insert(long ledgerId, long entryId, ByteBuf entry) { + int entrySize = entry.readableBytes(); + int alignedSize = align64(entrySize); + int offset = currentOffset.getAndAdd(alignedSize); + + if (offset + entrySize > segmentSize) { + // The segment is full + return false; + } else { + // Copy entry into read cache segment + cacheBuffer.setBytes(offset, entry, entry.readerIndex(), entry.readableBytes()); + long value = offset << 32 | entrySize; + index.put(ledgerId, entryId, value, 0); + return true; + } + } + + @Override + public ByteBuf get(long ledgerId, long entryId) { + long value = index.getFirstValue(ledgerId, entryId); + if (value >= 0) { + int offset = (int) (value >> 32); + int entryLen = (int) value; + + ByteBuf entry = PulsarByteBufAllocator.DEFAULT.buffer(entryLen, entryLen); + entry.writeBytes(cacheBuffer, offset, entryLen); + return entry; + } else { + return null; + } + } + + @Override + public int getSize() { + return currentOffset.get(); + } + + @Override + public void close() { + cacheBuffer.release(); + } + + private static int align64(int size) { + return (size + 64 - 1) & ALIGN_64_MASK; + } + + @Override + public void clear() { + index.clear(); + currentOffset.set(0); + } +} diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferRefCount.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferRefCount.java new file mode 100644 index 0000000000000..3b5f441915886 --- /dev/null +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferRefCount.java @@ -0,0 +1,91 @@ +/** + * 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.util.IllegalReferenceCountException; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.pulsar.common.util.collections.ConcurrentLongPairObjectHashMap; + +class SharedCacheSegmentBufferRefCount implements SharedCacheSegment { + + private final AtomicInteger currentSize = new AtomicInteger(); + private final ConcurrentLongPairObjectHashMap index; + private final int segmentSize; + + SharedCacheSegmentBufferRefCount(int segmentSize) { + this.segmentSize = segmentSize; + this.index = ConcurrentLongPairObjectHashMap.newBuilder() + // We are going to often clear() the map, with the expectation that it's going to get filled again + // immediately after. In these conditions it does not make sense to shrink it each time. + .autoShrink(false) + .concurrencyLevel(Runtime.getRuntime().availableProcessors() * 2) + .build(); + } + + @Override + public boolean insert(long ledgerId, long entryId, ByteBuf entry) { + int newSize = currentSize.addAndGet(entry.readableBytes()); + + if (newSize > segmentSize) { + // The segment is full + return false; + } else { + // Insert entry into read cache segment + ByteBuf oldValue = index.putIfAbsent(ledgerId, entryId, entry.retain()); + if (oldValue != null) { + entry.release(); + return false; + } else { + return true; + } + } + } + + @Override + public ByteBuf get(long ledgerId, long entryId) { + ByteBuf entry = index.get(ledgerId, entryId); + if (entry != null) { + try { + return entry.retain(); + } catch (IllegalReferenceCountException e) { + // Entry was removed between the get() and the retain() calls + return null; + } + } else { + return null; + } + } + + @Override + public int getSize() { + return currentSize.get(); + } + + @Override + public void close() { + clear(); + } + + @Override + public void clear() { + index.forEach((ledgerId, entryId, e) -> e.release()); + index.clear(); + } +} diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java new file mode 100644 index 0000000000000..7afa0d067018e --- /dev/null +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java @@ -0,0 +1,216 @@ +/** + * 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 static com.google.common.base.Preconditions.checkNotNull; +import static org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.createManagedLedgerException; +import com.google.common.collect.Lists; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.client.api.BKException; +import org.apache.bookkeeper.client.api.LedgerEntry; +import org.apache.bookkeeper.client.api.ReadHandle; +import org.apache.bookkeeper.mledger.AsyncCallbacks; +import org.apache.bookkeeper.mledger.Entry; +import org.apache.bookkeeper.mledger.ManagedLedgerException; +import org.apache.bookkeeper.mledger.impl.EntryImpl; +import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; +import org.apache.bookkeeper.mledger.impl.PositionImpl; +import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor; +import org.apache.commons.lang3.tuple.Pair; + +@Slf4j +class SharedEntryCacheImpl implements EntryCache { + + private final SharedEntryCacheManagerImpl entryCacheManager; + private final ManagedLedgerImpl ml; + private final ManagedLedgerInterceptor interceptor; + + SharedEntryCacheImpl(ManagedLedgerImpl ml, SharedEntryCacheManagerImpl entryCacheManager) { + this.ml = ml; + this.entryCacheManager = entryCacheManager; + this.interceptor = ml.getManagedLedgerInterceptor(); + } + + @Override + public String getName() { + return ml.getName(); + } + + @Override + public boolean insert(EntryImpl entry) { + return entryCacheManager.insert(entry); + } + + @Override + public void invalidateEntries(PositionImpl lastPosition) { + // No-Op. The cache invalidation is based only on rotating the segment buffers + } + + @Override + public void invalidateEntriesBeforeTimestamp(long timestamp) { + // No-Op. The cache invalidation is based only on rotating the segment buffers + } + + @Override + public void invalidateAllEntries(long ledgerId) { + // No-Op. The cache invalidation is based only on rotating the segment buffers + } + + @Override + public void clear() { + // No-Op. The cache invalidation is based only on rotating the segment buffers + } + + private static final Pair NO_EVICTION = Pair.of(0, 0L); + + @Override + public Pair evictEntries(long sizeToFree) { + return NO_EVICTION; + } + + @Override + public void asyncReadEntry(ReadHandle lh, long firstEntry, long lastEntry, boolean isSlowestReader, + AsyncCallbacks.ReadEntriesCallback callback, Object ctx) { + final long ledgerId = lh.getId(); + final int entriesToRead = (int) (lastEntry - firstEntry) + 1; + + if (log.isDebugEnabled()) { + log.debug("[{}] Reading entries range ledger {}: {} to {}", ml.getName(), ledgerId, firstEntry, lastEntry); + } + + List cachedEntries = new ArrayList<>(entriesToRead); + long totalCachedSize = entryCacheManager.getRange(ledgerId, firstEntry, lastEntry, cachedEntries); + + if (cachedEntries.size() == entriesToRead) { + // All entries found in cache + entryCacheManager.getFactoryMBean().recordCacheHits(entriesToRead, totalCachedSize); + if (log.isDebugEnabled()) { + log.debug("[{}] Ledger {} -- Found in cache entries: {}-{}", ml.getName(), ledgerId, firstEntry, + lastEntry); + } + + callback.readEntriesComplete(cachedEntries, ctx); + + } else { + if (!cachedEntries.isEmpty()) { + cachedEntries.forEach(entry -> entry.release()); + } + + // Read all the entries from bookkeeper + lh.readAsync(firstEntry, lastEntry).thenAcceptAsync( + ledgerEntries -> { + checkNotNull(ml.getName()); + checkNotNull(ml.getExecutor()); + + try { + // We got the entries, we need to transform them to a List<> type + long totalSize = 0; + final List entriesToReturn = Lists.newArrayListWithExpectedSize(entriesToRead); + for (LedgerEntry e : ledgerEntries) { + EntryImpl entry = EntryCacheManager.create(e, interceptor); + + entriesToReturn.add(entry); + totalSize += entry.getLength(); + } + + entryCacheManager.getFactoryMBean().recordCacheMiss(entriesToReturn.size(), totalSize); + ml.getMbean().addReadEntriesSample(entriesToReturn.size(), totalSize); + + callback.readEntriesComplete(entriesToReturn, ctx); + } finally { + ledgerEntries.close(); + } + }, ml.getExecutor().chooseThread(ml.getName())).exceptionally(exception -> { + if (exception instanceof BKException + && ((BKException) exception).getCode() == BKException.Code.TooManyRequestsException) { + callback.readEntriesFailed(createManagedLedgerException(exception), ctx); + } else { + ml.invalidateLedgerHandle(lh); + ManagedLedgerException mlException = createManagedLedgerException(exception); + callback.readEntriesFailed(mlException, ctx); + } + return null; + }); + } + } + + @Override + public void asyncReadEntry(ReadHandle lh, PositionImpl position, AsyncCallbacks.ReadEntryCallback callback, + Object ctx) { + try { + asyncReadEntry0(lh, position, callback, ctx); + } catch (Throwable t) { + log.warn("[{}] Failed to read entries for {}-{}", getName(), lh.getId(), position, t); + callback.readEntryFailed(createManagedLedgerException(t), ctx); + } + } + + private void asyncReadEntry0(ReadHandle lh, PositionImpl position, AsyncCallbacks.ReadEntryCallback callback, + Object ctx) { + if (log.isDebugEnabled()) { + log.debug("[{}] Reading entry ledger {}: {}", ml.getName(), lh.getId(), position.getEntryId()); + } + + EntryImpl cachedEntry = entryCacheManager.get(position.getLedgerId(), position.getEntryId()); + + if (cachedEntry != null) { + entryCacheManager.getFactoryMBean().recordCacheHit(cachedEntry.getLength()); + callback.readEntryComplete(cachedEntry, ctx); + } else { + lh.readAsync(position.getEntryId(), position.getEntryId()).thenAcceptAsync( + ledgerEntries -> { + try { + Iterator iterator = ledgerEntries.iterator(); + if (iterator.hasNext()) { + LedgerEntry ledgerEntry = iterator.next(); + EntryImpl returnEntry = EntryCacheManager.create(ledgerEntry, interceptor); + + entryCacheManager.getFactoryMBean().recordCacheMiss(1, returnEntry.getLength()); + ml.getMbean().addReadEntriesSample(1, returnEntry.getLength()); + callback.readEntryComplete(returnEntry, ctx); + } else { + // got an empty sequence + callback.readEntryFailed(new ManagedLedgerException("Could not read given position"), + ctx); + } + } finally { + ledgerEntries.close(); + } + }, ml.getExecutor().chooseThread(ml.getName())).exceptionally(exception -> { + ml.invalidateLedgerHandle(lh); + callback.readEntryFailed(createManagedLedgerException(exception), ctx); + return null; + }); + } + } + + @Override + public long getSize() { + return 0; + } + + @Override + public int compareTo(EntryCache o) { + // The individual topic caches cannot be compared since the cache is shared + return 0; + } +} diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheManagerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheManagerImpl.java new file mode 100644 index 0000000000000..be03b7fe01915 --- /dev/null +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheManagerImpl.java @@ -0,0 +1,218 @@ +/** + * 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 java.util.ArrayList; +import java.util.List; +import java.util.concurrent.locks.StampedLock; +import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.mledger.Entry; +import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig; +import org.apache.bookkeeper.mledger.impl.EntryImpl; +import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl; +import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryMBeanImpl; +import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; + +@Slf4j +public class SharedEntryCacheManagerImpl implements EntryCacheManager { + + private final ManagedLedgerFactoryConfig config; + private final ManagedLedgerFactoryMBeanImpl factoryMBean; + private final List segments = new ArrayList<>(); + private int currentSegmentIdx = 0; + private final int segmentSize; + private final int segmentsCount; + + private final StampedLock lock = new StampedLock(); + + private static final int DEFAULT_MAX_SEGMENT_SIZE = 1 * 1024 * 1024 * 1024; + + public SharedEntryCacheManagerImpl(ManagedLedgerFactoryImpl factory) { + this.config = factory.getConfig(); + this.factoryMBean = factory.getMbean(); + long maxCacheSize = config.getMaxCacheSize(); + if (maxCacheSize > 0) { + this.segmentsCount = Math.max(2, (int) (maxCacheSize / DEFAULT_MAX_SEGMENT_SIZE)); + this.segmentSize = (int) (maxCacheSize / segmentsCount); + + for (int i = 0; i < segmentsCount; i++) { + if (config.isCopyEntriesInCache()) { + segments.add(new SharedCacheSegmentBufferCopy(segmentSize)); + } else { + segments.add(new SharedCacheSegmentBufferRefCount(segmentSize)); + } + } + } else { + this.segmentsCount = 0; + this.segmentSize = 0; + } + } + + ManagedLedgerFactoryMBeanImpl getFactoryMBean() { + return factoryMBean; + } + + @Override + public EntryCache getEntryCache(ManagedLedgerImpl ml) { + if (getMaxSize() > 0) { + return new SharedEntryCacheImpl(ml, this); + } else { + return new EntryCacheDisabled(ml); + } + } + + @Override + public void removeEntryCache(String name) { + // no-op + } + + @Override + public long getSize() { + long totalSize = 0; + for (int i = 0; i < segmentsCount; i++) { + totalSize += segments.get(i).getSize(); + } + return totalSize; + } + + @Override + public long getMaxSize() { + return config.getMaxCacheSize(); + } + + @Override + public void clear() { + segments.forEach(SharedCacheSegment::clear); + } + + @Override + public void close() { + segments.forEach(SharedCacheSegment::close); + } + + @Override + public void updateCacheSizeAndThreshold(long maxSize) { + + } + + @Override + public void updateCacheEvictionWatermark(double cacheEvictionWatermark) { + // No-Op. We don't use the cache eviction watermark in this implementation + } + + @Override + public double getCacheEvictionWatermark() { + return config.getCacheEvictionWatermark(); + } + + boolean insert(EntryImpl entry) { + int entrySize = entry.getLength(); + + if (entrySize > segmentSize) { + log.debug("entrySize {} > segmentSize {}, skip update read cache!", entrySize, segmentSize); + return false; + } + + long stamp = lock.readLock(); + try { + SharedCacheSegment s = segments.get(currentSegmentIdx); + + if (s.insert(entry.getLedgerId(), entry.getEntryId(), entry.getDataBuffer())) { + return true; + } + } finally { + lock.unlockRead(stamp); + } + + // We could not insert in segment, we to get the write lock and roll-over to + // next segment + stamp = lock.writeLock(); + + try { + SharedCacheSegment segment = segments.get(currentSegmentIdx); + + if (segment.insert(entry.getLedgerId(), entry.getEntryId(), entry.getDataBuffer())) { + return true; + } + + // Roll to next segment + currentSegmentIdx = (currentSegmentIdx + 1) % segmentsCount; + segment = segments.get(currentSegmentIdx); + segment.clear(); + return segment.insert(entry.getLedgerId(), entry.getEntryId(), entry.getDataBuffer()); + } finally { + lock.unlockWrite(stamp); + } + } + + EntryImpl get(long ledgerId, long entryId) { + long stamp = lock.readLock(); + + try { + // We need to check all the segments, starting from the current one and looking + // backward to minimize the checks for recently inserted entries + for (int i = 0; i < segmentsCount; i++) { + int segmentIdx = (currentSegmentIdx + (segmentsCount - i)) % segmentsCount; + + ByteBuf res = segments.get(segmentIdx).get(ledgerId, entryId); + if (res != null) { + return EntryImpl.create(ledgerId, entryId, res); + } + } + } finally { + lock.unlockRead(stamp); + } + + return null; + } + + long getRange(long ledgerId, long firstEntryId, long lastEntryId, List results) { + long totalSize = 0; + long stamp = lock.readLock(); + + try { + // We need to check all the segments, starting from the current one and looking + // backward to minimize the checks for recently inserted entries + long entryId = firstEntryId; + for (int i = 0; i < segmentsCount; i++) { + int segmentIdx = (currentSegmentIdx + (segmentsCount - i)) % segmentsCount; + SharedCacheSegment s = segments.get(segmentIdx); + + for (; entryId <= lastEntryId; entryId++) { + ByteBuf res = s.get(ledgerId, entryId); + if (res != null) { + results.add(EntryImpl.create(ledgerId, entryId, res)); + totalSize += res.readableBytes(); + } else { + break; + } + } + + if (entryId == lastEntryId) { + break; + } + } + } finally { + lock.unlockRead(stamp); + } + + return totalSize; + } +} diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheManagerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheManagerTest.java index 55f58ecd11c08..efc9db8bf2295 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheManagerTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/EntryCacheManagerTest.java @@ -43,8 +43,11 @@ import org.apache.bookkeeper.mledger.impl.cache.EntryCache; import org.apache.bookkeeper.mledger.impl.cache.EntryCacheDisabled; import org.apache.bookkeeper.mledger.impl.cache.EntryCacheManager; +import org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheManagerImpl; +import org.apache.bookkeeper.mledger.impl.cache.SharedEntryCacheManagerImpl; import org.apache.bookkeeper.test.MockedBookKeeperTestCase; import org.testng.Assert; +import org.testng.annotations.DataProvider; import org.testng.annotations.Test; public class EntryCacheManagerTest extends MockedBookKeeperTestCase { @@ -52,6 +55,14 @@ public class EntryCacheManagerTest extends MockedBookKeeperTestCase { ManagedLedgerImpl ml1; ManagedLedgerImpl ml2; + @DataProvider(name = "EntryCacheManagerClass") + public static Object[][] primeNumbers() { + return new Object[][]{ + {SharedEntryCacheManagerImpl.class.getName()}, + {RangeEntryCacheManagerImpl.class.getName()}}; + } + + @Override protected void setUpTestCase() throws Exception { OrderedScheduler executor = OrderedScheduler.newSchedulerBuilder().numThreads(1).build(); @@ -68,11 +79,12 @@ protected void setUpTestCase() throws Exception { when(ml2.getName()).thenReturn("cache2"); } - @Test - public void simple() throws Exception { + @Test(dataProvider = "EntryCacheManagerClass") + public void simple(String entryCacheManagerClass) throws Exception { ManagedLedgerFactoryConfig config = new ManagedLedgerFactoryConfig(); config.setMaxCacheSize(10); config.setCacheEvictionWatermark(0.8); + config.setEntryCacheManagerClassName(entryCacheManagerClass); @Cleanup("shutdown") ManagedLedgerFactoryImpl factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc, config); @@ -84,12 +96,11 @@ public void simple() throws Exception { cache1.insert(EntryImpl.create(1, 1, new byte[4])); cache1.insert(EntryImpl.create(1, 0, new byte[3])); - assertEquals(cache1.getSize(), 7); - assertEquals(cacheManager.getSize(), 7); + assertTrue(cacheManager.getSize() > 0); factory2.getMbean().refreshStats(1, TimeUnit.SECONDS); assertEquals(factory2.getMbean().getCacheMaxSize(), 10); - assertEquals(factory2.getMbean().getCacheUsedSize(), 7); + assertTrue(factory2.getMbean().getCacheUsedSize() > 0); assertEquals(factory2.getMbean().getCacheHitsRate(), 0.0); assertEquals(factory2.getMbean().getCacheMissesRate(), 0.0); assertEquals(factory2.getMbean().getCacheHitsThroughput(), 0.0); @@ -99,8 +110,7 @@ public void simple() throws Exception { cache2.insert(EntryImpl.create(2, 1, new byte[1])); cache2.insert(EntryImpl.create(2, 2, new byte[1])); - assertEquals(cache2.getSize(), 3); - assertEquals(cacheManager.getSize(), 10); + assertTrue(cacheManager.getSize() > 0); // Next insert should trigger a cache eviction to force the size to 8 // The algorithm should evict entries from cache1 @@ -108,34 +118,30 @@ public void simple() throws Exception { // Wait for eviction to be completed in background Thread.sleep(100); - assertEquals(cacheManager.getSize(), 7); - assertEquals(cache1.getSize(), 4); - assertEquals(cache2.getSize(), 3); + assertTrue(cacheManager.getSize() > 0); cacheManager.removeEntryCache("cache1"); - assertEquals(cacheManager.getSize(), 3); - assertEquals(cache2.getSize(), 3); + assertTrue(cacheManager.getSize() > 0); // Should remove 1 entry cache2.invalidateEntries(new PositionImpl(2, 1)); - assertEquals(cacheManager.getSize(), 2); - assertEquals(cache2.getSize(), 2); + assertTrue(cacheManager.getSize() > 0); factory2.getMbean().refreshStats(1, TimeUnit.SECONDS); assertEquals(factory2.getMbean().getCacheMaxSize(), 10); - assertEquals(factory2.getMbean().getCacheUsedSize(), 2); + assertTrue(factory2.getMbean().getCacheUsedSize() > 0); assertEquals(factory2.getMbean().getCacheHitsRate(), 0.0); assertEquals(factory2.getMbean().getCacheMissesRate(), 0.0); assertEquals(factory2.getMbean().getCacheHitsThroughput(), 0.0); - assertEquals(factory2.getMbean().getNumberOfCacheEvictions(), 1); } - @Test - public void doubleInsert() throws Exception { + @Test(dataProvider = "EntryCacheManagerClass") + public void doubleInsert(String entryCacheManager) throws Exception { ManagedLedgerFactoryConfig config = new ManagedLedgerFactoryConfig(); config.setMaxCacheSize(10); config.setCacheEvictionWatermark(0.8); + config.setEntryCacheManagerClassName(entryCacheManager); @Cleanup("shutdown") ManagedLedgerFactoryImpl factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc, config); @@ -146,13 +152,11 @@ public void doubleInsert() throws Exception { assertTrue(cache1.insert(EntryImpl.create(1, 1, new byte[4]))); assertTrue(cache1.insert(EntryImpl.create(1, 0, new byte[3]))); - assertEquals(cache1.getSize(), 7); - assertEquals(cacheManager.getSize(), 7); + assertTrue(cacheManager.getSize() > 0); assertFalse(cache1.insert(EntryImpl.create(1, 0, new byte[5]))); - assertEquals(cache1.getSize(), 7); - assertEquals(cacheManager.getSize(), 7); + assertTrue(cacheManager.getSize() > 0); } @Test @@ -160,6 +164,7 @@ public void cacheSizeUpdate() throws Exception { ManagedLedgerFactoryConfig config = new ManagedLedgerFactoryConfig(); config.setMaxCacheSize(200); config.setCacheEvictionWatermark(0.8); + config.setEntryCacheManagerClassName(RangeEntryCacheManagerImpl.class.getName()); @Cleanup("shutdown") ManagedLedgerFactoryImpl factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc, config); @@ -188,11 +193,12 @@ public void cacheSizeUpdate() throws Exception { } - @Test - public void cacheDisabled() throws Exception { + @Test(dataProvider = "EntryCacheManagerClass") + public void cacheDisabled(String entryCacheManager) throws Exception { ManagedLedgerFactoryConfig config = new ManagedLedgerFactoryConfig(); config.setMaxCacheSize(0); config.setCacheEvictionWatermark(0.8); + config.setEntryCacheManagerClassName(entryCacheManager); @Cleanup("shutdown") ManagedLedgerFactoryImpl factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc, config); @@ -226,11 +232,12 @@ public void cacheDisabled() throws Exception { assertEquals(cacheManager.getSize(), 0); } - @Test - public void verifyNoCacheIfNoConsumer() throws Exception { + @Test(dataProvider = "EntryCacheManagerClass") + public void verifyNoCacheIfNoConsumer(String entryCacheManager) throws Exception { ManagedLedgerFactoryConfig config = new ManagedLedgerFactoryConfig(); config.setMaxCacheSize(7 * 10); config.setCacheEvictionWatermark(0.8); + config.setEntryCacheManagerClassName(entryCacheManager); @Cleanup("shutdown") ManagedLedgerFactoryImpl factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc, config); @@ -255,12 +262,13 @@ public void verifyNoCacheIfNoConsumer() throws Exception { assertEquals(factory2.getMbean().getNumberOfCacheEvictions(), 0); } - @Test - public void verifyHitsMisses() throws Exception { + @Test(dataProvider = "EntryCacheManagerClass") + public void verifyHitsMisses(String entryCacheManager) throws Exception { ManagedLedgerFactoryConfig config = new ManagedLedgerFactoryConfig(); - config.setMaxCacheSize(7 * 10); + config.setMaxCacheSize(150); config.setCacheEvictionWatermark(0.8); config.setCacheEvictionFrequency(1); + config.setEntryCacheManagerClassName(entryCacheManager); @Cleanup("shutdown") ManagedLedgerFactoryImpl factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc, config); @@ -276,7 +284,7 @@ public void verifyHitsMisses() throws Exception { } factory2.getMbean().refreshStats(1, TimeUnit.SECONDS); - assertEquals(factory2.getMbean().getCacheUsedSize(), 70); + assertTrue(factory2.getMbean().getCacheUsedSize() > 0); assertEquals(factory2.getMbean().getCacheHitsRate(), 0.0); assertEquals(factory2.getMbean().getCacheMissesRate(), 0.0); assertEquals(factory2.getMbean().getCacheHitsThroughput(), 0.0); @@ -287,7 +295,7 @@ public void verifyHitsMisses() throws Exception { entries.forEach(Entry::release); factory2.getMbean().refreshStats(1, TimeUnit.SECONDS); - assertEquals(factory2.getMbean().getCacheUsedSize(), 70); + assertTrue(factory2.getMbean().getCacheUsedSize() > 0); assertEquals(factory2.getMbean().getCacheHitsRate(), 10.0); assertEquals(factory2.getMbean().getCacheMissesRate(), 0.0); assertEquals(factory2.getMbean().getCacheHitsThroughput(), 70.0); @@ -296,7 +304,7 @@ public void verifyHitsMisses() throws Exception { ledger.deactivateCursor(c1); factory2.getMbean().refreshStats(1, TimeUnit.SECONDS); - assertEquals(factory2.getMbean().getCacheUsedSize(), 70); + assertTrue(factory2.getMbean().getCacheUsedSize() > 0); assertEquals(factory2.getMbean().getCacheHitsRate(), 0.0); assertEquals(factory2.getMbean().getCacheMissesRate(), 0.0); assertEquals(factory2.getMbean().getCacheHitsThroughput(), 0.0); @@ -306,7 +314,7 @@ public void verifyHitsMisses() throws Exception { assertEquals(entries.size(), 10); factory2.getMbean().refreshStats(1, TimeUnit.SECONDS); - assertEquals(factory2.getMbean().getCacheUsedSize(), 70); + assertTrue(factory2.getMbean().getCacheUsedSize() > 0); assertEquals(factory2.getMbean().getCacheHitsRate(), 10.0); assertEquals(factory2.getMbean().getCacheMissesRate(), 0.0); assertEquals(factory2.getMbean().getCacheHitsThroughput(), 70.0); @@ -318,7 +326,7 @@ public void verifyHitsMisses() throws Exception { entries.forEach(Entry::release); factory2.getMbean().refreshStats(1, TimeUnit.SECONDS); - assertEquals(factory2.getMbean().getCacheUsedSize(), 7); + assertTrue(factory2.getMbean().getCacheUsedSize() > 0); assertEquals(factory2.getMbean().getCacheHitsRate(), 0.0); assertEquals(factory2.getMbean().getCacheMissesRate(), 0.0); assertEquals(factory2.getMbean().getCacheHitsThroughput(), 0.0); @@ -331,6 +339,8 @@ public void verifyTimeBasedEviction() throws Exception { config.setMaxCacheSize(1000); config.setCacheEvictionFrequency(100); config.setCacheEvictionTimeThresholdMillis(100); + // This is only relevant for this specific implementation + config.setEntryCacheManagerClassName(RangeEntryCacheManagerImpl.class.getName()); @Cleanup("shutdown") ManagedLedgerFactoryImpl factory = new ManagedLedgerFactoryImpl(metadataStore, bkc, config); @@ -360,12 +370,13 @@ public void verifyTimeBasedEviction() throws Exception { assertEquals(cache.getSize(), 0); } - @Test(timeOut = 5000) - void entryCacheDisabledAsyncReadEntry() throws Exception { + @Test(dataProvider = "EntryCacheManagerClass") + void entryCacheDisabledAsyncReadEntry(String entryCacheManager) throws Exception { ReadHandle lh = EntryCacheTest.getLedgerHandle(); ManagedLedgerFactoryConfig config = new ManagedLedgerFactoryConfig(); config.setMaxCacheSize(0); + config.setEntryCacheManagerClassName(entryCacheManager); @Cleanup("shutdown") ManagedLedgerFactoryImpl factory = new ManagedLedgerFactoryImpl(metadataStore, bkc, config); 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 b5d677f8b04f9..f35516fac6a31 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 @@ -428,7 +428,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { cursor.rewind(); // Clear the cache to force reading from BK - ledger.entryCache.clear(); + factory.getEntryCacheManager().clear(); final CountDownLatch counter2 = new CountDownLatch(1); diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java index 291ddc645722a..0c45244363413 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java @@ -111,6 +111,7 @@ import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback; import org.apache.bookkeeper.mledger.impl.cache.EntryCache; import org.apache.bookkeeper.mledger.impl.cache.EntryCacheManager; +import org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheManagerImpl; import org.apache.bookkeeper.mledger.proto.MLDataFormats; import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo; import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo; @@ -2669,6 +2670,7 @@ public void testGetNextValidPosition() throws Exception { public void testActiveDeactiveCursorWithDiscardEntriesFromCache() throws Exception { ManagedLedgerFactoryConfig conf = new ManagedLedgerFactoryConfig(); conf.setCacheEvictionFrequency(0.1); + conf.setEntryCacheManagerClassName(RangeEntryCacheManagerImpl.class.getName()); @Cleanup("shutdown") ManagedLedgerFactory factory = new ManagedLedgerFactoryImpl(metadataStore, bkc, conf); diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/test/MockedBookKeeperTestCase.java b/managed-ledger/src/test/java/org/apache/bookkeeper/test/MockedBookKeeperTestCase.java index 0fd8902f8253e..7ba211f7e0709 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/test/MockedBookKeeperTestCase.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/test/MockedBookKeeperTestCase.java @@ -26,7 +26,9 @@ import lombok.SneakyThrows; import org.apache.bookkeeper.client.PulsarMockBookKeeper; import org.apache.bookkeeper.common.util.OrderedScheduler; +import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig; import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl; +import org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheManagerImpl; import org.apache.pulsar.metadata.api.MetadataStoreConfig; import org.apache.pulsar.metadata.api.MetadataStoreException; import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; @@ -79,7 +81,9 @@ public final void setUp(Method method) throws Exception { throw e; } - factory = new ManagedLedgerFactoryImpl(metadataStore, bkc); + ManagedLedgerFactoryConfig conf = new ManagedLedgerFactoryConfig(); + conf.setEntryCacheManagerClassName(RangeEntryCacheManagerImpl.class.getName()); + factory = new ManagedLedgerFactoryImpl(metadataStore, bkc, conf); setUpTestCase(); } diff --git a/microbenchmarks/pom.xml b/microbenchmarks/pom.xml new file mode 100644 index 0000000000000..4be062576809f --- /dev/null +++ b/microbenchmarks/pom.xml @@ -0,0 +1,117 @@ + + + + + pulsar + org.apache.pulsar + 2.11.0-SNAPSHOT + .. + + + 4.0.0 + + microbenchmarks + + + org.openjdk.jmh + jmh-core + + + org.openjdk.jmh + jmh-generator-annprocess + provided + + + ${project.groupId} + managed-ledger + ${project.version} + + + + ${project.groupId} + testmocks + ${project.version} + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + ${maven.compiler.source} + ${maven.compiler.source} + ${maven.compiler.target} + + + + org.openjdk.jmh + jmh-generator-annprocess + ${jmh.version} + + + + + + org.apache.maven.plugins + maven-shade-plugin + + + package + + shade + + + benchmarks + + + org.openjdk.jmh.Main + + + + + + + + + + + + + + + + + + + + org.apache.maven.plugins + maven-checkstyle-plugin + + true + + + + + \ No newline at end of file diff --git a/microbenchmarks/src/main/java/org/apache/bookkeeper/mledger/impl/cache/ManagedLedgerCacheBenchmark.java b/microbenchmarks/src/main/java/org/apache/bookkeeper/mledger/impl/cache/ManagedLedgerCacheBenchmark.java new file mode 100644 index 0000000000000..822c381ea4edd --- /dev/null +++ b/microbenchmarks/src/main/java/org/apache/bookkeeper/mledger/impl/cache/ManagedLedgerCacheBenchmark.java @@ -0,0 +1,147 @@ +/** + * 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 com.google.common.collect.ImmutableMap; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.PooledByteBufAllocator; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.bookkeeper.client.PulsarMockBookKeeper; +import org.apache.bookkeeper.common.util.OrderedExecutor; +import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig; +import org.apache.bookkeeper.mledger.impl.EntryImpl; +import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl; +import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; +import org.apache.pulsar.metadata.api.MetadataStoreConfig; +import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; +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.TearDown; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; + +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.MICROSECONDS) +@Threads(16) +@Fork(1) +@Warmup(iterations = 1, time = 10, timeUnit = TimeUnit.SECONDS) +@Measurement(iterations = 3, time = 10, timeUnit = TimeUnit.SECONDS) +public class ManagedLedgerCacheBenchmark { + + private static Map cacheManagers = ImmutableMap.of( + "RangeEntryCacheManager", RangeEntryCacheManagerImpl.class.getName(), + "SharedEntryCacheManager", SharedEntryCacheManagerImpl.class.getName()); + + public enum CopyMode { + Copy, + RefCount, + } + + @State(Scope.Benchmark) + public static class TestState { + @Param({ +// "RangeEntryCacheManager", + "SharedEntryCacheManager", + }) + private String entryCacheManagerName; + + @Param({ + "Copy", + "RefCount", + }) + private CopyMode copyMode; + + @Param({ +// "100", +// "1024", + "65536", + }) + private int entrySize; + + private OrderedExecutor executor; + private MetadataStoreExtended metadataStore; + private ManagedLedgerFactoryImpl mlf; + private EntryCache entryCache; + + private ByteBuf buffer; + + @Setup(Level.Trial) + public void setup() throws Exception { + executor = OrderedExecutor.newBuilder().build(); + metadataStore = MetadataStoreExtended.create("memory:local", MetadataStoreConfig.builder().build()); + + ManagedLedgerFactoryConfig mlfc = new ManagedLedgerFactoryConfig(); + mlfc.setEntryCacheManagerClassName(cacheManagers.get(entryCacheManagerName)); + mlfc.setCopyEntriesInCache(copyMode == CopyMode.Copy); + mlfc.setMaxCacheSize(1 * 1024 * 1024 * 1024); + PulsarMockBookKeeper bkc = new PulsarMockBookKeeper(executor); + mlf = new ManagedLedgerFactoryImpl(metadataStore, bkc, mlfc); + + ManagedLedgerImpl ml = (ManagedLedgerImpl) mlf.open("test-managed-ledger"); + + entryCache = mlf.getEntryCacheManager().getEntryCache(ml); + + buffer = PooledByteBufAllocator.DEFAULT.directBuffer(); + buffer.writeBytes(new byte[entrySize]); + } + + @TearDown(Level.Trial) + public void tearDown() throws Exception { + mlf.shutdown(); + metadataStore.close(); + + System.out.println("REF-COUNT: " + buffer.refCnt()); + buffer.release(); + executor.shutdownNow(); + } + } + + private static final AtomicLong ledgerIdSeq = new AtomicLong(); + + @State(Scope.Thread) + public static class ThreadState { + private long ledgerId; + private long entryId; + + @Setup(Level.Iteration) + public void setup() throws Exception { + ledgerId = ledgerIdSeq.incrementAndGet(); + entryId = 0; + } + } + + @Benchmark + public void insertIntoCache(TestState s, ThreadState ts) { + EntryImpl entry = EntryImpl.create(ts.ledgerId, ts.entryId, s.buffer.duplicate()); + s.entryCache.insert(entry); + ts.entryId++; + entry.release(); + } +} diff --git a/pom.xml b/pom.xml index 34e302ffc4893..1c5202a105537 100644 --- a/pom.xml +++ b/pom.xml @@ -231,6 +231,7 @@ flexible messaging model and an intuitive client API. 1.17.2 2.2 + 1.35 3.2.13 @@ -1290,6 +1291,18 @@ flexible messaging model and an intuitive client API. netty-reactive-streams ${netty-reactive-streams.version} + + + + org.openjdk.jmh + jmh-core + ${jmh.version} + + + org.openjdk.jmh + jmh-generator-annprocess + ${jmh.version} + @@ -2117,6 +2130,7 @@ flexible messaging model and an intuitive client API. distribution docker tests + microbenchmarks @@ -2145,6 +2159,7 @@ flexible messaging model and an intuitive client API. pulsar-broker-auth-sasl pulsar-client-auth-sasl pulsar-config-validation + microbenchmarks pulsar-transaction diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index ad4188d288275..05d25526bc692 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -1786,6 +1786,15 @@ public class ServiceConfiguration implements PulsarConfiguration { @FieldContext(category = CATEGORY_STORAGE_ML, doc = "Whether we should make a copy of the entry payloads when " + "inserting in cache") private boolean managedLedgerCacheCopyEntries = false; + + @FieldContext(category = CATEGORY_STORAGE_ML, + doc = "The class name for the implementation of ManagedLedger cache manager component.\n" + + "Options are:\n" + + " - org.apache.bookkeeper.mledger.impl.cache.SharedEntryCacheManagerImpl\n" + + " - org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheManagerImpl") + private String managedLedgerCacheManagerImplementationClass = + "org.apache.bookkeeper.mledger.impl.cache.SharedEntryCacheManagerImpl"; + @FieldContext( category = CATEGORY_STORAGE_ML, dynamic = true, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java index bb7cb6ffd8d7e..c4f8ad4407218 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java @@ -63,6 +63,7 @@ public void initialize(ServiceConfiguration conf, MetadataStoreExtended metadata managedLedgerFactoryConfig.setCacheEvictionTimeThresholdMillis( conf.getManagedLedgerCacheEvictionTimeThresholdMillis()); managedLedgerFactoryConfig.setCopyEntriesInCache(conf.isManagedLedgerCacheCopyEntries()); + managedLedgerFactoryConfig.setEntryCacheManagerClassName(conf.getManagedLedgerStorageClassName()); managedLedgerFactoryConfig.setPrometheusStatsLatencyRolloverSeconds( conf.getManagedLedgerPrometheusStatsLatencyRolloverSeconds()); managedLedgerFactoryConfig.setTraceTaskExecution(conf.isManagedLedgerTraceTaskExecution()); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java index 7473fdaf78632..2fbbeeccbca3f 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java @@ -19,7 +19,6 @@ package org.apache.pulsar.broker.service.nonpersistent; import static com.google.common.base.Preconditions.checkArgument; -import static org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheManagerImpl.create; import static org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType; import static org.apache.pulsar.common.protocol.Commands.DEFAULT_CONSUMER_EPOCH; import com.carrotsearch.hppc.ObjectObjectHashMap; @@ -40,6 +39,7 @@ import java.util.concurrent.atomic.AtomicLongFieldUpdater; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.Position; +import org.apache.bookkeeper.mledger.impl.cache.EntryCacheManager; import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.resources.NamespaceResources; @@ -183,7 +183,7 @@ public void publishMessage(ByteBuf data, PublishContext callback) { subscriptions.forEach((name, subscription) -> { ByteBuf duplicateBuffer = data.retainedDuplicate(); - Entry entry = create(0L, 0L, duplicateBuffer); + Entry entry = EntryCacheManager.create(0L, 0L, duplicateBuffer); // entry internally retains data so, duplicateBuffer should be release here duplicateBuffer.release(); if (subscription.getDispatcher() != null) { @@ -198,7 +198,7 @@ public void publishMessage(ByteBuf data, PublishContext callback) { if (!replicators.isEmpty()) { replicators.forEach((name, replicator) -> { ByteBuf duplicateBuffer = data.retainedDuplicate(); - Entry entry = create(0L, 0L, duplicateBuffer); + Entry entry = EntryCacheManager.create(0L, 0L, duplicateBuffer); // entry internally retains data so, duplicateBuffer should be release here duplicateBuffer.release(); replicator.sendMessage(entry); diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java index eac7268ba672d..d7ef217bd2bae 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongLongPairHashMap.java @@ -204,6 +204,12 @@ public LongPair get(long key1, long key2) { return getSection(h).get(key1, key2, (int) h); } + public long getFirstValue(long key1, long key2) { + checkBiggerEqualZero(key1); + long h = hash(key1, key2); + return getSection(h).getFirstValue(key1, key2, (int) h); + } + public boolean containsKey(long key1, long key2) { return get(key1, key2) != null; } @@ -370,6 +376,55 @@ LongPair get(long key1, long key2, int keyHash) { } } + long getFirstValue(long key1, long key2, int keyHash) { + long stamp = tryOptimisticRead(); + boolean acquiredLock = false; + int bucket = signSafeMod(keyHash, capacity); + + try { + while (true) { + // First try optimistic locking + long storedKey1 = table[bucket]; + long storedKey2 = table[bucket + 1]; + long storedValue1 = table[bucket + 2]; + + if (!acquiredLock && validate(stamp)) { + // The values we have read are consistent + if (key1 == storedKey1 && key2 == storedKey2) { + return storedValue1; + } else if (storedKey1 == EmptyKey) { + // Not found + return ValueNotFound; + } + } else { + // Fallback to acquiring read lock + if (!acquiredLock) { + stamp = readLock(); + acquiredLock = true; + + bucket = signSafeMod(keyHash, capacity); + storedKey1 = table[bucket]; + storedKey2 = table[bucket + 1]; + storedValue1 = table[bucket + 2]; + } + + if (key1 == storedKey1 && key2 == storedKey2) { + return storedValue1; + } else if (storedKey1 == EmptyKey) { + // Not found + return ValueNotFound; + } + } + + bucket = (bucket + 4) & (table.length - 1); + } + } finally { + if (acquiredLock) { + unlockRead(stamp); + } + } + } + boolean put(long key1, long key2, long value1, long value2, int keyHash, boolean onlyIfAbsent) { long stamp = writeLock(); int bucket = signSafeMod(keyHash, capacity); diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairObjectHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairObjectHashMap.java new file mode 100644 index 0000000000000..d3ccb61e3f40e --- /dev/null +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairObjectHashMap.java @@ -0,0 +1,647 @@ +/** + * 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.pulsar.common.util.collections; + +import static com.google.common.base.Preconditions.checkArgument; +import static java.util.Objects.requireNonNull; +import com.google.common.collect.Lists; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; +import java.util.concurrent.locks.StampedLock; +import org.apache.commons.lang3.tuple.Pair; + +/** + * Map from long to an Object. + * + *

Provides similar methods as a {@code ConcurrentMap>,Object>} with 2 differences: + *

    + *
  1. No boxing/unboxing from (long,long) -> Object + *
  2. Open hash map with linear probing, no node allocations to store the values + *
+ * + * @param + */ +@SuppressWarnings("unchecked") +public class ConcurrentLongPairObjectHashMap { + + private static final Object EmptyValue = null; + private static final Object DeletedValue = new Object(); + + private static final int DefaultExpectedItems = 256; + private static final int DefaultConcurrencyLevel = 16; + + private static final float DefaultMapFillFactor = 0.66f; + private static final float DefaultMapIdleFactor = 0.15f; + + private static final float DefaultExpandFactor = 2; + private static final float DefaultShrinkFactor = 2; + + private static final boolean DefaultAutoShrink = false; + + public interface LongLongFunction { + R apply(long key1, long key2); + } + + public static Builder newBuilder() { + return new Builder<>(); + } + + /** + * Builder of ConcurrentLongHashMap. + */ + public static class Builder { + int expectedItems = DefaultExpectedItems; + int concurrencyLevel = DefaultConcurrencyLevel; + float mapFillFactor = DefaultMapFillFactor; + float mapIdleFactor = DefaultMapIdleFactor; + float expandFactor = DefaultExpandFactor; + float shrinkFactor = DefaultShrinkFactor; + boolean autoShrink = DefaultAutoShrink; + + public Builder expectedItems(int expectedItems) { + this.expectedItems = expectedItems; + return this; + } + + public Builder concurrencyLevel(int concurrencyLevel) { + this.concurrencyLevel = concurrencyLevel; + return this; + } + + public Builder mapFillFactor(float mapFillFactor) { + this.mapFillFactor = mapFillFactor; + return this; + } + + public Builder mapIdleFactor(float mapIdleFactor) { + this.mapIdleFactor = mapIdleFactor; + return this; + } + + public Builder expandFactor(float expandFactor) { + this.expandFactor = expandFactor; + return this; + } + + public Builder shrinkFactor(float shrinkFactor) { + this.shrinkFactor = shrinkFactor; + return this; + } + + public Builder autoShrink(boolean autoShrink) { + this.autoShrink = autoShrink; + return this; + } + + public ConcurrentLongPairObjectHashMap build() { + return new ConcurrentLongPairObjectHashMap<>(expectedItems, concurrencyLevel, + mapFillFactor, mapIdleFactor, autoShrink, expandFactor, shrinkFactor); + } + } + + private final Section[] sections; + + @Deprecated + public ConcurrentLongPairObjectHashMap() { + this(DefaultExpectedItems); + } + + @Deprecated + public ConcurrentLongPairObjectHashMap(int expectedItems) { + this(expectedItems, DefaultConcurrencyLevel); + } + + @Deprecated + public ConcurrentLongPairObjectHashMap(int expectedItems, int concurrencyLevel) { + this(expectedItems, concurrencyLevel, DefaultMapFillFactor, DefaultMapIdleFactor, + DefaultAutoShrink, DefaultExpandFactor, DefaultShrinkFactor); + } + + public ConcurrentLongPairObjectHashMap(int expectedItems, int concurrencyLevel, + float mapFillFactor, float mapIdleFactor, + boolean autoShrink, float expandFactor, float shrinkFactor) { + checkArgument(expectedItems > 0); + checkArgument(concurrencyLevel > 0); + checkArgument(expectedItems >= concurrencyLevel); + checkArgument(mapFillFactor > 0 && mapFillFactor < 1); + checkArgument(mapIdleFactor > 0 && mapIdleFactor < 1); + checkArgument(mapFillFactor > mapIdleFactor); + checkArgument(expandFactor > 1); + checkArgument(shrinkFactor > 1); + + int numSections = concurrencyLevel; + int perSectionExpectedItems = expectedItems / numSections; + int perSectionCapacity = (int) (perSectionExpectedItems / mapFillFactor); + this.sections = (Section[]) new Section[numSections]; + + for (int i = 0; i < numSections; i++) { + sections[i] = new Section<>(perSectionCapacity, mapFillFactor, mapIdleFactor, + autoShrink, expandFactor, shrinkFactor); + } + } + + public long size() { + long size = 0; + for (Section s : sections) { + size += s.size; + } + return size; + } + + long getUsedBucketCount() { + long usedBucketCount = 0; + for (Section s : sections) { + usedBucketCount += s.usedBuckets; + } + return usedBucketCount; + } + + public long capacity() { + long capacity = 0; + for (Section s : sections) { + capacity += s.capacity; + } + return capacity; + } + + public boolean isEmpty() { + for (Section s : sections) { + if (s.size != 0) { + return false; + } + } + + return true; + } + + public V get(long key1, long key2) { + long h = hash(key1, key2); + return getSection(h).get(key1, key2, (int) h); + } + + public boolean containsKey(long key1, long key2) { + return get(key1, key2) != null; + } + + public V put(long key1, long key2, V value) { + requireNonNull(value); + long h = hash(key1, key2); + return getSection(h).put(key1, key2, value, (int) h, false, null); + } + + public V putIfAbsent(long key1, long key2, V value) { + requireNonNull(value); + long h = hash(key1, key2); + return getSection(h).put(key1, key2, value, (int) h, true, null); + } + + public V computeIfAbsent(long key1, long key2, LongLongFunction provider) { + requireNonNull(provider); + long h = hash(key1, key2); + return getSection(h).put(key1, key2, null, (int) h, true, provider); + } + + public V remove(long key1, long key2) { + long h = hash(key1, key2); + return getSection(h).remove(key1, key2, null, (int) h); + } + + public boolean remove(long key1, long key2, Object value) { + requireNonNull(value); + long h = hash(key1, key2); + return getSection(h).remove(key1, key2, value, (int) h) != null; + } + + private Section getSection(long hash) { + // Use 32 msb out of long to get the section + final int sectionIdx = (int) (hash >>> 32) & (sections.length - 1); + return sections[sectionIdx]; + } + + public void clear() { + for (int i = 0; i < sections.length; i++) { + sections[i].clear(); + } + } + + public void forEach(EntryProcessor processor) { + for (int i = 0; i < sections.length; i++) { + sections[i].forEach(processor); + } + } + + /** + * @return a new list of all keys (makes a copy) + */ + public List> keys() { + List> keys = Lists.newArrayListWithExpectedSize((int) size()); + forEach((key1, key2, value) -> keys.add(Pair.of(key1, key2))); + return keys; + } + + public List values() { + List values = Lists.newArrayListWithExpectedSize((int) size()); + forEach((key1, key2, value) -> values.add(value)); + return values; + } + + /** + * Processor for one key-value entry, where the key is {@code long}. + * + * @param type of the value. + */ + public interface EntryProcessor { + void accept(long key1, long key2, V value); + } + + // A section is a portion of the hash map that is covered by a single + @SuppressWarnings("serial") + private static final class Section extends StampedLock { + private volatile long[] keys1; + private volatile long[] keys2; + private volatile V[] values; + + private volatile int capacity; + private final int initCapacity; + private static final AtomicIntegerFieldUpdater
SIZE_UPDATER = + AtomicIntegerFieldUpdater.newUpdater(Section.class, "size"); + + private volatile int size; + private int usedBuckets; + private int resizeThresholdUp; + private int resizeThresholdBelow; + private final float mapFillFactor; + private final float mapIdleFactor; + private final float expandFactor; + private final float shrinkFactor; + private final boolean autoShrink; + + Section(int capacity, float mapFillFactor, float mapIdleFactor, boolean autoShrink, + float expandFactor, float shrinkFactor) { + this.capacity = alignToPowerOfTwo(capacity); + this.initCapacity = this.capacity; + this.keys1 = new long[this.capacity]; + this.keys2 = new long[this.capacity]; + this.values = (V[]) new Object[this.capacity]; + this.size = 0; + this.usedBuckets = 0; + this.autoShrink = autoShrink; + this.mapFillFactor = mapFillFactor; + this.mapIdleFactor = mapIdleFactor; + this.expandFactor = expandFactor; + this.shrinkFactor = shrinkFactor; + this.resizeThresholdUp = (int) (this.capacity * mapFillFactor); + this.resizeThresholdBelow = (int) (this.capacity * mapIdleFactor); + } + + V get(long key1, long key2, int keyHash) { + int bucket = keyHash; + + long stamp = tryOptimisticRead(); + boolean acquiredLock = false; + + try { + while (true) { + int capacity = this.capacity; + bucket = signSafeMod(bucket, capacity); + + // First try optimistic locking + long storedKey1 = keys1[bucket]; + long storedKey2 = keys2[bucket]; + V storedValue = values[bucket]; + + if (!acquiredLock && validate(stamp)) { + // The values we have read are consistent + if (storedKey1 == key1 && storedKey2 == key2) { + return storedValue != DeletedValue ? storedValue : null; + } else if (storedValue == EmptyValue) { + // Not found + return null; + } + } else { + // Fallback to acquiring read lock + if (!acquiredLock) { + stamp = readLock(); + acquiredLock = true; + storedKey1 = keys1[bucket]; + storedKey2 = keys2[bucket]; + storedValue = values[bucket]; + } + + if (capacity != this.capacity) { + // There has been a rehashing. We need to restart the search + bucket = keyHash; + continue; + } + + if (storedKey1 == key1 && storedKey2 == key2) { + return storedValue != DeletedValue ? storedValue : null; + } else if (storedValue == EmptyValue) { + // Not found + return null; + } + } + + ++bucket; + } + } finally { + if (acquiredLock) { + unlockRead(stamp); + } + } + } + + V put(long key1, long key2, V value, int keyHash, boolean onlyIfAbsent, LongLongFunction valueProvider) { + int bucket = keyHash; + + long stamp = writeLock(); + int capacity = this.capacity; + + // Remember where we find the first available spot + int firstDeletedKey = -1; + + try { + while (true) { + bucket = signSafeMod(bucket, capacity); + + long storedKey1 = keys1[bucket]; + long storedKey2 = keys2[bucket]; + V storedValue = values[bucket]; + + if (storedKey1 == key1 && storedKey2 == key2) { + if (storedValue == EmptyValue) { + values[bucket] = value != null ? value : valueProvider.apply(key1, key2); + SIZE_UPDATER.incrementAndGet(this); + ++usedBuckets; + return valueProvider != null ? values[bucket] : null; + } else if (storedValue == DeletedValue) { + values[bucket] = value != null ? value : valueProvider.apply(key1, key2); + SIZE_UPDATER.incrementAndGet(this); + return valueProvider != null ? values[bucket] : null; + } else if (!onlyIfAbsent) { + // Over written an old value for same key + values[bucket] = value; + return storedValue; + } else { + return storedValue; + } + } else if (storedValue == EmptyValue) { + // Found an empty bucket. This means the key is not in the map. If we've already seen a deleted + // key, we should write at that position + if (firstDeletedKey != -1) { + bucket = firstDeletedKey; + } else { + ++usedBuckets; + } + + keys1[bucket] = key1; + keys2[bucket] = key2; + values[bucket] = value != null ? value : valueProvider.apply(key1, key2); + SIZE_UPDATER.incrementAndGet(this); + return valueProvider != null ? values[bucket] : null; + } else if (storedValue == DeletedValue) { + // The bucket contained a different deleted key + if (firstDeletedKey == -1) { + firstDeletedKey = bucket; + } + } + + ++bucket; + } + } finally { + if (usedBuckets > resizeThresholdUp) { + try { + int newCapacity = alignToPowerOfTwo((int) (capacity * expandFactor)); + rehash(newCapacity); + } finally { + unlockWrite(stamp); + } + } else { + unlockWrite(stamp); + } + } + } + + private V remove(long key1, long key2, Object value, int keyHash) { + int bucket = keyHash; + long stamp = writeLock(); + + try { + while (true) { + int capacity = this.capacity; + bucket = signSafeMod(bucket, capacity); + + long storedKey1 = keys1[bucket]; + long storedKey2 = keys2[bucket]; + V storedValue = values[bucket]; + if (storedKey1 == key1 && storedKey2 == key2) { + if (value == null || value.equals(storedValue)) { + if (storedValue == EmptyValue || storedValue == DeletedValue) { + return null; + } + + SIZE_UPDATER.decrementAndGet(this); + V nextValueInArray = values[signSafeMod(bucket + 1, capacity)]; + if (nextValueInArray == EmptyValue) { + values[bucket] = (V) EmptyValue; + --usedBuckets; + + // Cleanup all the buckets that were in `DeletedValue` state, + // so that we can reduce unnecessary expansions + int lastBucket = signSafeMod(bucket - 1, capacity); + while (values[lastBucket] == DeletedValue) { + values[lastBucket] = (V) EmptyValue; + --usedBuckets; + + lastBucket = signSafeMod(lastBucket - 1, capacity); + } + } else { + values[bucket] = (V) DeletedValue; + } + + return storedValue; + } else { + return null; + } + } else if (storedValue == EmptyValue) { + // Key wasn't found + return null; + } + + ++bucket; + } + + } finally { + if (autoShrink && size < resizeThresholdBelow) { + try { + int newCapacity = alignToPowerOfTwo((int) (capacity / shrinkFactor)); + int newResizeThresholdUp = (int) (newCapacity * mapFillFactor); + if (newCapacity < capacity && newResizeThresholdUp > size) { + // shrink the hashmap + rehash(newCapacity); + } + } finally { + unlockWrite(stamp); + } + } else { + unlockWrite(stamp); + } + } + } + + void clear() { + long stamp = writeLock(); + + try { + Arrays.fill(keys1, 0); + Arrays.fill(keys2, 0); + Arrays.fill(values, EmptyValue); + this.size = 0; + this.usedBuckets = 0; + if (autoShrink) { + rehash(initCapacity); + } + } finally { + unlockWrite(stamp); + } + } + + public void forEach(EntryProcessor processor) { + long stamp = tryOptimisticRead(); + + // We need to make sure that we read these 3 variables in a consistent way + int capacity = this.capacity; + long[] keys1 = this.keys1; + long[] keys2 = this.keys2; + V[] values = this.values; + + // Validate no rehashing + if (!validate(stamp)) { + // Fallback to read lock + stamp = readLock(); + + capacity = this.capacity; + keys1 = this.keys1; + keys2 = this.keys2; + values = this.values; + unlockRead(stamp); + } + + // Go through all the buckets for this section. We try to renew the stamp only after a validation + // error, otherwise we keep going with the same. + for (int bucket = 0; bucket < capacity; bucket++) { + if (stamp == 0) { + stamp = tryOptimisticRead(); + } + + long storedKey1 = keys1[bucket]; + long storedKey2 = keys2[bucket]; + V storedValue = values[bucket]; + + if (!validate(stamp)) { + // Fallback to acquiring read lock + stamp = readLock(); + + try { + storedKey1 = keys1[bucket]; + storedKey2 = keys2[bucket]; + storedValue = values[bucket]; + } finally { + unlockRead(stamp); + } + + stamp = 0; + } + + if (storedValue != DeletedValue && storedValue != EmptyValue) { + processor.accept(storedKey1, storedKey2, storedValue); + } + } + } + + private void rehash(int newCapacity) { + // Expand the hashmap + long[] newKeys1 = new long[newCapacity]; + long[] newKeys2 = new long[newCapacity]; + V[] newValues = (V[]) new Object[newCapacity]; + + // Re-hash table + for (int i = 0; i < keys1.length; i++) { + long storedKey1 = keys1[i]; + long storedKey2 = keys2[i]; + V storedValue = values[i]; + if (storedValue != EmptyValue && storedValue != DeletedValue) { + insertKeyValueNoLock(newKeys1, newKeys2, newValues, storedKey1, storedKey2, storedValue); + } + } + + keys1 = newKeys1; + keys2 = newKeys2; + values = newValues; + capacity = newCapacity; + usedBuckets = size; + resizeThresholdUp = (int) (capacity * mapFillFactor); + resizeThresholdBelow = (int) (capacity * mapIdleFactor); + } + + private static void insertKeyValueNoLock(long[] keys1, long[] keys2, V[] values, long key1, long key2, + V value) { + int bucket = (int) hash(key1, key2); + + while (true) { + bucket = signSafeMod(bucket, keys1.length); + + V storedValue = values[bucket]; + + if (storedValue == EmptyValue) { + // The bucket is empty, so we can use it + keys1[bucket] = key1; + keys2[bucket] = key2; + values[bucket] = value; + return; + } + + ++bucket; + } + } + } + + private static final long HashMixer = 0xc6a4a7935bd1e995L; + private static final int R = 47; + + static final long hash(long key1, long key2) { + long hash = key1 * HashMixer; + hash ^= hash >>> R; + hash *= HashMixer; + hash += 31 + (key2 * HashMixer); + hash ^= hash >>> R; + hash *= HashMixer; + return hash; + } + + + static final int signSafeMod(long n, int max) { + return (int) n & (max - 1); + } + + private static int alignToPowerOfTwo(int n) { + return (int) Math.pow(2, 32 - Integer.numberOfLeadingZeros(n - 1)); + } +} diff --git a/site2/docs/reference-configuration.md b/site2/docs/reference-configuration.md index e42ace8acf226..9fc578024da6a 100644 --- a/site2/docs/reference-configuration.md +++ b/site2/docs/reference-configuration.md @@ -297,6 +297,7 @@ brokerServiceCompactionThresholdInBytes|If the estimated backlog size is greater |managedLedgerDefaultAckQuorum| Number of guaranteed copies (acks to wait before write is complete) |2| |managedLedgerCacheSizeMB| Amount of memory to use for caching data payload in managed ledger. This memory is allocated from JVM direct memory and it’s shared across all the topics running in the same broker. By default, uses 1/5th of available direct memory || |managedLedgerCacheCopyEntries| Whether we should make a copy of the entry payloads when inserting in cache| false| +|managedLedgerCacheManagerImplementationClass| The class name for the implementation of ManagedLedger cache manager component. Options are: org.apache.bookkeeper.mledger.impl.cache.SharedEntryCacheManagerImpl, org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheManagerImpl| org.apache.bookkeeper.mledger.impl.cache.SharedEntryCacheManagerImpl| |managedLedgerCacheEvictionWatermark| Threshold to which bring down the cache level when eviction is triggered |0.9| |managedLedgerCacheEvictionFrequency| Configure the cache eviction frequency for the managed ledger cache (evictions/sec) | 100.0 | |managedLedgerCacheEvictionTimeThresholdMillis| All entries that have stayed in cache for more than the configured time, will be evicted | 1000 | From 519cb0141f7d29091db86fc5726994a784ca73c0 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Wed, 3 Aug 2022 12:25:17 +0200 Subject: [PATCH 02/10] Fixed test --- .../cache/SharedCacheSegmentBufferCopy.java | 2 +- .../ConcurrentLongPairObjectHashMap.java | 18 ++++++++++++++++++ .../test/MockedBookKeeperTestCase.java | 2 ++ 3 files changed, 21 insertions(+), 1 deletion(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java index 93dc0083fdd3d..1b677851c5ad5 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java @@ -56,7 +56,7 @@ public boolean insert(long ledgerId, long entryId, ByteBuf entry) { } else { // Copy entry into read cache segment cacheBuffer.setBytes(offset, entry, entry.readerIndex(), entry.readableBytes()); - long value = offset << 32 | entrySize; + long value = ((long)offset << 32) | entrySize; index.put(ledgerId, entryId, value, 0); return true; } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairObjectHashMap.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairObjectHashMap.java index d3ccb61e3f40e..b3e48b7c69c9b 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairObjectHashMap.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentLongPairObjectHashMap.java @@ -25,6 +25,7 @@ import java.util.List; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.locks.StampedLock; +import org.apache.commons.lang3.mutable.MutableBoolean; import org.apache.commons.lang3.tuple.Pair; /** @@ -644,4 +645,21 @@ static final int signSafeMod(long n, int max) { private static int alignToPowerOfTwo(int n) { return (int) Math.pow(2, 32 - Integer.numberOfLeadingZeros(n - 1)); } + + @Override + public String toString() { + StringBuilder sb = new StringBuilder(); + sb.append('{'); + MutableBoolean isFirst = new MutableBoolean(true); + forEach((k1, k2, v) -> { + if (!isFirst.getValue()) { + sb.append(", "); + } + isFirst.setFalse(); + + sb.append(k1).append(':').append(k2).append('=').append(v); + }); + sb.append('}'); + return sb.toString(); + } } diff --git a/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/test/MockedBookKeeperTestCase.java b/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/test/MockedBookKeeperTestCase.java index 3c50a82daacb4..821d97f7a49be 100644 --- a/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/test/MockedBookKeeperTestCase.java +++ b/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/test/MockedBookKeeperTestCase.java @@ -27,6 +27,7 @@ import org.apache.bookkeeper.conf.ClientConfiguration; import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig; import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl; +import org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheManagerImpl; import org.apache.pulsar.metadata.api.MetadataStoreConfig; import org.apache.pulsar.metadata.api.MetadataStoreException; import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; @@ -81,6 +82,7 @@ public void setUp(Method method) throws Exception { } ManagedLedgerFactoryConfig conf = new ManagedLedgerFactoryConfig(); + conf.setCopyEntriesInCache(true); factory = new ManagedLedgerFactoryImpl(metadataStore, bkc, conf); } From 2273fc64fb078cd0ccf1b34fde0f17730ca3992f Mon Sep 17 00:00:00 2001 From: Jiwe Guo Date: Wed, 3 Aug 2022 20:26:33 +0800 Subject: [PATCH 03/10] fix checkstyle. --- .../mledger/impl/cache/SharedCacheSegmentBufferCopy.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java index 1b677851c5ad5..58c798b27dd38 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java @@ -56,7 +56,7 @@ public boolean insert(long ledgerId, long entryId, ByteBuf entry) { } else { // Copy entry into read cache segment cacheBuffer.setBytes(offset, entry, entry.readerIndex(), entry.readableBytes()); - long value = ((long)offset << 32) | entrySize; + long value = ((long) offset << 32) | entrySize; index.put(ledgerId, entryId, value, 0); return true; } From c2a27e9001cba04cc386f2012e1a467049902386 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Wed, 3 Aug 2022 16:00:53 +0200 Subject: [PATCH 04/10] Removed unused import --- .../transaction/coordinator/test/MockedBookKeeperTestCase.java | 1 - 1 file changed, 1 deletion(-) diff --git a/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/test/MockedBookKeeperTestCase.java b/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/test/MockedBookKeeperTestCase.java index 821d97f7a49be..de6a5672edf0a 100644 --- a/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/test/MockedBookKeeperTestCase.java +++ b/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/test/MockedBookKeeperTestCase.java @@ -27,7 +27,6 @@ import org.apache.bookkeeper.conf.ClientConfiguration; import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig; import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl; -import org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheManagerImpl; import org.apache.pulsar.metadata.api.MetadataStoreConfig; import org.apache.pulsar.metadata.api.MetadataStoreException; import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; From 46d4547815786339643baccf6c12f612e99281d5 Mon Sep 17 00:00:00 2001 From: mattison chao Date: Thu, 4 Aug 2022 09:58:13 +0800 Subject: [PATCH 05/10] Fix wrong configuration --- .../org/apache/pulsar/broker/ManagedLedgerClientFactory.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java index ae040714187d6..a219e7b62c35a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java @@ -63,7 +63,7 @@ public void initialize(ServiceConfiguration conf, MetadataStoreExtended metadata managedLedgerFactoryConfig.setCacheEvictionTimeThresholdMillis( conf.getManagedLedgerCacheEvictionTimeThresholdMillis()); managedLedgerFactoryConfig.setCopyEntriesInCache(conf.isManagedLedgerCacheCopyEntries()); - managedLedgerFactoryConfig.setEntryCacheManagerClassName(conf.getManagedLedgerStorageClassName()); + managedLedgerFactoryConfig.setEntryCacheManagerClassName(conf.getManagedLedgerCacheManagerImplementationClass()); managedLedgerFactoryConfig.setPrometheusStatsLatencyRolloverSeconds( conf.getManagedLedgerPrometheusStatsLatencyRolloverSeconds()); managedLedgerFactoryConfig.setTraceTaskExecution(conf.isManagedLedgerTraceTaskExecution()); From f6d0bd8952c316bd3a5741a3a5bf00b5bb202ccf Mon Sep 17 00:00:00 2001 From: mattison chao Date: Thu, 4 Aug 2022 10:10:54 +0800 Subject: [PATCH 06/10] Fix checkstyle --- .../org/apache/pulsar/broker/ManagedLedgerClientFactory.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java index a219e7b62c35a..9d5f9849b635d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java @@ -63,7 +63,8 @@ public void initialize(ServiceConfiguration conf, MetadataStoreExtended metadata managedLedgerFactoryConfig.setCacheEvictionTimeThresholdMillis( conf.getManagedLedgerCacheEvictionTimeThresholdMillis()); managedLedgerFactoryConfig.setCopyEntriesInCache(conf.isManagedLedgerCacheCopyEntries()); - managedLedgerFactoryConfig.setEntryCacheManagerClassName(conf.getManagedLedgerCacheManagerImplementationClass()); + managedLedgerFactoryConfig.setEntryCacheManagerClassName( + conf.getManagedLedgerCacheManagerImplementationClass()); managedLedgerFactoryConfig.setPrometheusStatsLatencyRolloverSeconds( conf.getManagedLedgerPrometheusStatsLatencyRolloverSeconds()); managedLedgerFactoryConfig.setTraceTaskExecution(conf.isManagedLedgerTraceTaskExecution()); From 8a5e1c068760867dc93cc8f023e8dd4d539e913c Mon Sep 17 00:00:00 2001 From: mattison chao Date: Thu, 4 Aug 2022 14:08:38 +0800 Subject: [PATCH 07/10] Fix cache entry --- .../bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java index 7afa0d067018e..9b08481bc7954 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java @@ -173,8 +173,10 @@ private void asyncReadEntry0(ReadHandle lh, PositionImpl position, AsyncCallback EntryImpl cachedEntry = entryCacheManager.get(position.getLedgerId(), position.getEntryId()); if (cachedEntry != null) { - entryCacheManager.getFactoryMBean().recordCacheHit(cachedEntry.getLength()); - callback.readEntryComplete(cachedEntry, ctx); + EntryImpl entry = EntryImpl.create(cachedEntry); + cachedEntry.release(); + entryCacheManager.getFactoryMBean().recordCacheHit(entry.getLength()); + callback.readEntryComplete(entry, ctx); } else { lh.readAsync(position.getEntryId(), position.getEntryId()).thenAcceptAsync( ledgerEntries -> { From 2cc107d81a178d32480fc3d20923884e268c012e Mon Sep 17 00:00:00 2001 From: mattison chao Date: Thu, 4 Aug 2022 14:58:09 +0800 Subject: [PATCH 08/10] Fix read entires --- .../mledger/impl/cache/SharedEntryCacheImpl.java | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java index 9b08481bc7954..5db97f5ec17af 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java @@ -101,14 +101,19 @@ public void asyncReadEntry(ReadHandle lh, long firstEntry, long lastEntry, boole long totalCachedSize = entryCacheManager.getRange(ledgerId, firstEntry, lastEntry, cachedEntries); if (cachedEntries.size() == entriesToRead) { + final List entriesToReturn = Lists.newArrayListWithExpectedSize(entriesToRead); // All entries found in cache - entryCacheManager.getFactoryMBean().recordCacheHits(entriesToRead, totalCachedSize); + for (Entry entry : cachedEntries) { + entriesToReturn.add(EntryImpl.create((EntryImpl) entry)); + entry.release(); + } + // All entries found in cache + entryCacheManager.getFactoryMBean().recordCacheHits(entriesToReturn.size(), totalCachedSize); if (log.isDebugEnabled()) { log.debug("[{}] Ledger {} -- Found in cache entries: {}-{}", ml.getName(), ledgerId, firstEntry, lastEntry); } - - callback.readEntriesComplete(cachedEntries, ctx); + callback.readEntriesComplete(entriesToReturn, ctx); } else { if (!cachedEntries.isEmpty()) { From 870d15a0e7363d647515bbbb9cdbd0828a994adb Mon Sep 17 00:00:00 2001 From: Jiwe Guo Date: Tue, 16 Aug 2022 21:44:07 +0800 Subject: [PATCH 09/10] fix comment. --- .../mledger/impl/cache/RangeEntryCacheManagerImpl.java | 2 -- .../mledger/impl/cache/SharedCacheSegmentBufferCopy.java | 1 + .../mledger/impl/cache/SharedCacheSegmentBufferRefCount.java | 3 +++ 3 files changed, 4 insertions(+), 2 deletions(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheManagerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheManagerImpl.java index b257ec329f13d..c575bb165fc47 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheManagerImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheManagerImpl.java @@ -20,8 +20,6 @@ import static org.apache.bookkeeper.mledger.util.SafeRun.safeRun; import com.google.common.collect.Lists; -import com.google.common.collect.Maps; -import io.netty.buffer.ByteBuf; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java index 58c798b27dd38..102f62ec9aefc 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferCopy.java @@ -52,6 +52,7 @@ public boolean insert(long ledgerId, long entryId, ByteBuf entry) { if (offset + entrySize > segmentSize) { // The segment is full + currentOffset.getAndAdd(-alignedSize); return false; } else { // Copy entry into read cache segment diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferRefCount.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferRefCount.java index 3b5f441915886..6f71cc59e0dcd 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferRefCount.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedCacheSegmentBufferRefCount.java @@ -45,11 +45,13 @@ public boolean insert(long ledgerId, long entryId, ByteBuf entry) { if (newSize > segmentSize) { // The segment is full + currentSize.getAndAdd(-entry.readableBytes()); return false; } else { // Insert entry into read cache segment ByteBuf oldValue = index.putIfAbsent(ledgerId, entryId, entry.retain()); if (oldValue != null) { + currentSize.getAndAdd(-entry.readableBytes()); entry.release(); return false; } else { @@ -87,5 +89,6 @@ public void close() { public void clear() { index.forEach((ledgerId, entryId, e) -> e.release()); index.clear(); + currentSize.set(0); } } From f6855e4e0b7d5a264ce01bdd64cb73cf88940b1e Mon Sep 17 00:00:00 2001 From: Jiwe Guo Date: Wed, 17 Aug 2022 20:50:33 +0800 Subject: [PATCH 10/10] fix modernizer plugin check. --- .../bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java index 5db97f5ec17af..f820d6ecb2c27 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/SharedEntryCacheImpl.java @@ -18,7 +18,7 @@ */ package org.apache.bookkeeper.mledger.impl.cache; -import static com.google.common.base.Preconditions.checkNotNull; +import static java.util.Objects.requireNonNull; import static org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.createManagedLedgerException; import com.google.common.collect.Lists; import java.util.ArrayList; @@ -123,8 +123,8 @@ public void asyncReadEntry(ReadHandle lh, long firstEntry, long lastEntry, boole // Read all the entries from bookkeeper lh.readAsync(firstEntry, lastEntry).thenAcceptAsync( ledgerEntries -> { - checkNotNull(ml.getName()); - checkNotNull(ml.getExecutor()); + requireNonNull(ml.getName()); + requireNonNull(ml.getExecutor()); try { // We got the entries, we need to transform them to a List<> type