You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@bookkeeper.apache.org by GitBox <gi...@apache.org> on 2018/01/08 21:18:55 UTC
[GitHub] sijie closed pull request #913: Refactor FileInfo locking and refcounting out IndexPersistenceMgr
sijie closed pull request #913: Refactor FileInfo locking and refcounting out IndexPersistenceMgr
URL: https://github.com/apache/bookkeeper/pull/913
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/FileInfo.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/FileInfo.java
index 38edbda18..296acb93c 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/FileInfo.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/FileInfo.java
@@ -33,7 +33,6 @@
import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
-import java.util.concurrent.atomic.AtomicInteger;
import org.apache.bookkeeper.common.util.Watchable;
import org.apache.bookkeeper.common.util.Watcher;
import org.slf4j.Logger;
@@ -79,7 +78,6 @@
static final long START_OF_DATA = 1024;
private long size;
- private AtomicInteger useCount = new AtomicInteger(0);
private boolean isClosed;
private long sizeSinceLastwrite;
@@ -139,6 +137,10 @@ synchronized boolean waitForLastAddConfirmedUpdate(long previousLAC,
return true;
}
+ public boolean isClosed() {
+ return isClosed;
+ }
+
public synchronized File getLf() {
return lf;
}
@@ -387,35 +389,26 @@ private int readAbsolute(ByteBuffer bb, long start, boolean bestEffort)
* if set to false, the index is not forced to create.
*/
public void close(boolean force) throws IOException {
- boolean closing = false;
- try {
- boolean changed = false;
- synchronized (this) {
- if (isClosed) {
- return;
- }
- isClosed = true;
- closing = true;
- checkOpen(force, true);
- // Any time when we force close a file, we should try to flush header.
- // otherwise, we might lose fence bit.
- if (force) {
- flushHeader();
- }
- changed = true;
- if (useCount.get() == 0 && fc != null) {
- fc.close();
- fc = null;
- }
+ boolean changed = false;
+ synchronized (this) {
+ if (isClosed) {
+ return;
}
- if (changed) {
- notifyWatchers(LastAddConfirmedUpdateNotification.FUNC, Long.MAX_VALUE);
+ isClosed = true;
+ checkOpen(force, true);
+ // Any time when we force close a file, we should try to flush header.
+ // otherwise, we might lose fence bit.
+ if (force) {
+ flushHeader();
}
- } finally {
- if (closing) {
- // recycle this watchable after the FileInfo is closed.
- recycle();
+ changed = true;
+ if (fc != null) {
+ fc.close();
}
+ fc = null;
+ }
+ if (changed) {
+ notifyWatchers(LastAddConfirmedUpdateNotification.FUNC, Long.MAX_VALUE);
}
}
@@ -502,26 +495,6 @@ public synchronized void moveToNewLocation(File newFile, long size) throws IOExc
return masterKey;
}
- public void use() {
- useCount.incrementAndGet();
- }
-
- @VisibleForTesting
- int getUseCount() {
- return useCount.get();
- }
-
- public synchronized void release() {
- int count = useCount.decrementAndGet();
- if (isClosed && (count == 0) && fc != null) {
- try {
- fc.close();
- } catch (IOException e) {
- LOG.error("Error closing file channel", e);
- }
- }
- }
-
public synchronized boolean delete() {
return lf.delete();
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/FileInfoBackingCache.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/FileInfoBackingCache.java
new file mode 100644
index 000000000..8cdadc79c
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/FileInfoBackingCache.java
@@ -0,0 +1,168 @@
+/**
+ *
+ * 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.bookie;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+class FileInfoBackingCache {
+ static final int DEAD_REF = -0xdead;
+
+ final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+ final ConcurrentHashMap<Long, CachedFileInfo> fileInfos = new ConcurrentHashMap<>();
+ final FileLoader fileLoader;
+
+ FileInfoBackingCache(FileLoader fileLoader) {
+ this.fileLoader = fileLoader;
+ }
+
+ CachedFileInfo loadFileInfo(long ledgerId, byte[] masterKey) throws IOException {
+ lock.readLock().lock();
+ try {
+ CachedFileInfo fi = fileInfos.get(ledgerId);
+ if (fi != null) {
+ // tryRetain only fails if #markDead() has been called
+ // on fi. This is only called from within the write lock,
+ // and if it is called (and succeeds) the fi will have been
+ // removed from fileInfos at the same time, so we should not
+ // have been able to get a reference to it here.
+ // The caller of loadFileInfo owns the refence, and is
+ // responsible for calling the corresponding #release().
+ assert(fi.tryRetain());
+ return fi;
+ }
+ } finally {
+ lock.readLock().unlock();
+ }
+
+ // else FileInfo not found, create it under write lock
+ lock.writeLock().lock();
+ try {
+ File backingFile = fileLoader.load(ledgerId, masterKey != null);
+ CachedFileInfo fi = new CachedFileInfo(ledgerId, backingFile, masterKey);
+ fileInfos.put(ledgerId, fi);
+
+ // see comment above for why we assert
+ assert(fi.tryRetain());
+ return fi;
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private void releaseFileInfo(long ledgerId, CachedFileInfo fileInfo) {
+ lock.writeLock().lock();
+ try {
+ if (fileInfo.markDead()) {
+ fileInfo.close(true);
+ fileInfos.remove(ledgerId, fileInfo);
+ }
+ } catch (IOException ioe) {
+ log.error("Error evicting file info({}) for ledger {} from backing cache",
+ fileInfo, ledgerId, ioe);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ void closeAllWithoutFlushing() throws IOException {
+ for (Map.Entry<Long, CachedFileInfo> entry : fileInfos.entrySet()) {
+ entry.getValue().close(false);
+ }
+ }
+
+ class CachedFileInfo extends FileInfo {
+ final long ledgerId;
+ final AtomicInteger refCount;
+
+ CachedFileInfo(long ledgerId, File lf, byte[] masterKey) throws IOException {
+ super(lf, masterKey);
+ this.ledgerId = ledgerId;
+ this.refCount = new AtomicInteger(0);
+ }
+
+ /**
+ * Mark this fileinfo as dead. We can only mark a fileinfo as
+ * dead if noone currently holds a reference to it.
+ *
+ * @return true if we marked as dead, false otherwise
+ */
+ private boolean markDead() {
+ return refCount.compareAndSet(0, DEAD_REF);
+ }
+
+ /**
+ * Attempt to retain the file info.
+ * When a client obtains a fileinfo from a container object,
+ * but that container object may release the fileinfo before
+ * the client has a chance to call retain. In this case, the
+ * file info could be released and the destroyed before we ever
+ * get a chance to use it.
+ *
+ * <p>tryRetain avoids this problem, by doing a compare-and-swap on
+ * the reference count. If the refCount is negative, it means that
+ * the fileinfo is being cleaned up, and this fileinfo object should
+ * not be used. This works in tandem with #markDead, which will only
+ * set the refCount to negative if noone currently has it retained
+ * (i.e. the refCount is 0).
+ *
+ * @return true if we managed to increment the refcount, false otherwise
+ */
+ boolean tryRetain() {
+ while (true) {
+ int count = refCount.get();
+ if (count < 0) {
+ return false;
+ } else if (refCount.compareAndSet(count, count + 1)) {
+ return true;
+ }
+ }
+ }
+
+ int getRefCount() {
+ return refCount.get();
+ }
+
+ void release() {
+ if (refCount.decrementAndGet() == 0) {
+ releaseFileInfo(ledgerId, this);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "CachedFileInfo(ledger=" + ledgerId
+ + ",refCount=" + refCount.get()
+ + ",closed=" + isClosed()
+ + ",id=" + System.identityHashCode(this) + ")";
+ }
+ }
+
+ interface FileLoader {
+ File load(long ledgerId, boolean createIfMissing) throws IOException;
+ }
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/IndexPersistenceMgr.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/IndexPersistenceMgr.java
index 8598bc4b6..0945506b4 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/IndexPersistenceMgr.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/IndexPersistenceMgr.java
@@ -38,13 +38,10 @@
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.bookkeeper.bookie.FileInfoBackingCache.CachedFileInfo;
import org.apache.bookkeeper.bookie.LedgerDirsManager.LedgerDirsListener;
import org.apache.bookkeeper.bookie.LedgerDirsManager.NoWritableLedgerDirException;
import org.apache.bookkeeper.common.util.Watcher;
@@ -80,15 +77,13 @@ public static final String getLedgerName(long ledgerId) {
}
// use two separate cache for write and read
- final Cache<Long, FileInfo> writeFileInfoCache;
- final Cache<Long, FileInfo> readFileInfoCache;
+ final Cache<Long, CachedFileInfo> writeFileInfoCache;
+ final Cache<Long, CachedFileInfo> readFileInfoCache;
+ final FileInfoBackingCache fileInfoBackingCache;
+
final int openFileLimit;
final int pageSize;
final int entriesPerPage;
- // Lock
- final ReentrantReadWriteLock fileInfoLock = new ReentrantReadWriteLock();
- // ThreadPool
- final ScheduledExecutorService evictionThreadPool = Executors.newSingleThreadScheduledExecutor();
// Manage all active ledgers in LedgerManager
// so LedgerManager has knowledge to garbage collect inactive/deleted ledgers
@@ -117,7 +112,8 @@ public IndexPersistenceMgr(int pageSize,
// build the file info cache
int concurrencyLevel = Math.max(1, Math.max(conf.getNumAddWorkerThreads(), conf.getNumReadWorkerThreads()));
- RemovalListener<Long, FileInfo> fileInfoEvictionListener = this::handleLedgerEviction;
+ fileInfoBackingCache = new FileInfoBackingCache(this::createFileInfoBackingFile);
+ RemovalListener<Long, CachedFileInfo> fileInfoEvictionListener = this::handleLedgerEviction;
writeFileInfoCache = buildCache(
concurrencyLevel,
conf.getFileInfoCacheInitialCapacity(),
@@ -158,12 +154,12 @@ public Number getSample() {
});
}
- private static Cache<Long, FileInfo> buildCache(int concurrencyLevel,
- int initialCapacity,
- int maximumSize,
- long expireAfterAccessSeconds,
- RemovalListener<Long, FileInfo> removalListener) {
- CacheBuilder<Long, FileInfo> builder = CacheBuilder.newBuilder()
+ private static Cache<Long, CachedFileInfo> buildCache(int concurrencyLevel,
+ int initialCapacity,
+ int maximumSize,
+ long expireAfterAccessSeconds,
+ RemovalListener<Long, CachedFileInfo> removalListener) {
+ CacheBuilder<Long, CachedFileInfo> builder = CacheBuilder.newBuilder()
.concurrencyLevel(concurrencyLevel)
.initialCapacity(initialCapacity)
.maximumSize(maximumSize)
@@ -174,67 +170,34 @@ public Number getSample() {
return builder.build();
}
+ private File createFileInfoBackingFile(long ledger, boolean createIfMissing) throws IOException {
+ File lf = findIndexFile(ledger);
+ if (null == lf) {
+ if (!createIfMissing) {
+ throw new Bookie.NoLedgerException(ledger);
+ }
+ // We don't have a ledger index file on disk or in cache, so create it.
+ lf = getNewLedgerIndexFile(ledger, null);
+ }
+ return lf;
+ }
+
/**
* When a ledger is evicted, we need to make sure there's no other thread
* trying to get FileInfo for that ledger at the same time when we close
* the FileInfo.
*/
- private void handleLedgerEviction(RemovalNotification<Long, FileInfo> notification) {
- FileInfo fileInfo = notification.getValue();
- Long ledgerId = notification.getKey();
+ private void handleLedgerEviction(RemovalNotification<Long, CachedFileInfo> notification) {
+ CachedFileInfo fileInfo = notification.getValue();
if (null == fileInfo || null == notification.getKey()) {
return;
}
if (notification.wasEvicted()) {
evictedLedgersCounter.inc();
- // we need to acquire the write lock in another thread,
- // otherwise there could be dead lock happening.
- evictionThreadPool.execute(() -> {
- fileInfoLock.writeLock().lock();
- try {
- // We only close the fileInfo when we evict the FileInfo from both cache
- if (!readFileInfoCache.asMap().containsKey(ledgerId)
- && !writeFileInfoCache.asMap().containsKey(ledgerId)) {
- fileInfo.close(true);
- }
- } catch (IOException e) {
- LOG.error("Exception closing file info when ledger {} is evicted from file info cache.",
- ledgerId, e);
- } finally {
- fileInfoLock.writeLock().unlock();
- }
- });
}
fileInfo.release();
}
- class FileInfoLoader implements Callable<FileInfo> {
-
- final long ledger;
- final byte[] masterKey;
-
- FileInfoLoader(long ledger, byte[] masterKey) {
- this.ledger = ledger;
- this.masterKey = masterKey;
- }
-
- @Override
- public FileInfo call() throws IOException {
- File lf = findIndexFile(ledger);
- if (null == lf) {
- if (null == masterKey) {
- throw new Bookie.NoLedgerException(ledger);
- }
- // We don't have a ledger index file on disk or in cache, so create it.
- lf = getNewLedgerIndexFile(ledger, null);
- activeLedgers.put(ledger, true);
- }
- FileInfo fi = new FileInfo(lf, masterKey);
- fi.use();
- return fi;
- }
- }
-
/**
* Get the FileInfo and increase reference count.
* When we get FileInfo from cache, we need to make sure it is synchronized
@@ -242,22 +205,23 @@ public FileInfo call() throws IOException {
* the FileInfo from cache, that FileInfo is then evicted and closed before we
* could even increase the reference counter.
*/
- FileInfo getFileInfo(final Long ledger, final byte masterKey[]) throws IOException {
+ CachedFileInfo getFileInfo(final Long ledger, final byte masterKey[]) throws IOException {
try {
- FileInfo fi;
+ CachedFileInfo fi;
pendingGetFileInfoCounter.inc();
- fileInfoLock.readLock().lock();
- if (null != masterKey) {
- fi = writeFileInfoCache.get(ledger,
- new FileInfoLoader(ledger, masterKey));
- if (null == readFileInfoCache.asMap().putIfAbsent(ledger, fi)) {
- fi.use();
+ Callable<CachedFileInfo> loader = () -> {
+ CachedFileInfo fileInfo = fileInfoBackingCache.loadFileInfo(ledger, masterKey);
+ activeLedgers.put(ledger, true);
+ return fileInfo;
+ };
+ do {
+ if (null != masterKey) {
+ fi = writeFileInfoCache.get(ledger, loader);
+ } else {
+ fi = readFileInfoCache.get(ledger, loader);
}
- } else {
- fi = readFileInfoCache.get(ledger,
- new FileInfoLoader(ledger, null));
- }
- fi.use();
+ } while (!fi.tryRetain());
+
return fi;
} catch (ExecutionException | UncheckedExecutionException ee) {
if (ee.getCause() instanceof IOException) {
@@ -267,7 +231,6 @@ FileInfo getFileInfo(final Long ledger, final byte masterKey[]) throws IOExcepti
}
} finally {
pendingGetFileInfoCounter.dec();
- fileInfoLock.readLock().unlock();
}
}
@@ -354,8 +317,7 @@ private void getActiveLedgers() throws IOException {
*/
void removeLedger(Long ledgerId) throws IOException {
// Delete the ledger's index file and close the FileInfo
- FileInfo fi = null;
- fileInfoLock.writeLock().lock();
+ CachedFileInfo fi = null;
try {
fi = getFileInfo(ledgerId, null);
// Don't force flush. There's no need since we're deleting the ledger
@@ -364,18 +326,14 @@ void removeLedger(Long ledgerId) throws IOException {
fi.close(false);
fi.delete();
} finally {
- try {
- if (fi != null) {
- // should release use count
- fi.release();
- // Remove it from the active ledger manager
- activeLedgers.remove(ledgerId);
- // Now remove it from cache
- writeFileInfoCache.invalidate(ledgerId);
- readFileInfoCache.invalidate(ledgerId);
- }
- } finally {
- fileInfoLock.writeLock().unlock();
+ if (fi != null) {
+ // should release use count
+ fi.release();
+ // Remove it from the active ledger manager
+ activeLedgers.remove(ledgerId);
+ // Now remove it from cache
+ writeFileInfoCache.invalidate(ledgerId);
+ readFileInfoCache.invalidate(ledgerId);
}
}
}
@@ -399,29 +357,13 @@ void close() throws IOException {
// Don't force create the file. We may have many dirty ledgers and file create/flush
// can be quite expensive as a result. We can use this optimization in this case
// because metadata will be recovered from the journal when we restart anyway.
- try {
- fileInfoLock.writeLock().lock();
- for (Map.Entry<Long, FileInfo> entry : writeFileInfoCache.asMap().entrySet()) {
- entry.getValue().close(false);
- }
- for (Map.Entry<Long, FileInfo> entry : readFileInfoCache.asMap().entrySet()) {
- entry.getValue().close(false);
- }
- writeFileInfoCache.invalidateAll();
- readFileInfoCache.invalidateAll();
- } finally {
- fileInfoLock.writeLock().unlock();
- }
- evictionThreadPool.shutdown();
- try {
- evictionThreadPool.awaitTermination(5, TimeUnit.SECONDS);
- } catch (InterruptedException e) {
- //ignore
- }
+ fileInfoBackingCache.closeAllWithoutFlushing();
+ writeFileInfoCache.invalidateAll();
+ readFileInfoCache.invalidateAll();
}
Long getLastAddConfirmed(long ledgerId) throws IOException {
- FileInfo fi = null;
+ CachedFileInfo fi = null;
try {
fi = getFileInfo(ledgerId, null);
return fi.getLastAddConfirmed();
@@ -435,7 +377,7 @@ Long getLastAddConfirmed(long ledgerId) throws IOException {
boolean waitForLastAddConfirmedUpdate(long ledgerId,
long previousLAC,
Watcher<LastAddConfirmedUpdateNotification> watcher) throws IOException {
- FileInfo fi = null;
+ CachedFileInfo fi = null;
try {
fi = getFileInfo(ledgerId, null);
return fi.waitForLastAddConfirmedUpdate(previousLAC, watcher);
@@ -447,7 +389,7 @@ boolean waitForLastAddConfirmedUpdate(long ledgerId,
}
long updateLastAddConfirmed(long ledgerId, long lac) throws IOException {
- FileInfo fi = null;
+ CachedFileInfo fi = null;
try {
fi = getFileInfo(ledgerId, null);
return fi.setLastAddConfirmed(lac);
@@ -459,7 +401,7 @@ long updateLastAddConfirmed(long ledgerId, long lac) throws IOException {
}
byte[] readMasterKey(long ledgerId) throws IOException, BookieException {
- FileInfo fi = null;
+ CachedFileInfo fi = null;
try {
fi = getFileInfo(ledgerId, null);
return fi.getMasterKey();
@@ -471,7 +413,7 @@ long updateLastAddConfirmed(long ledgerId, long lac) throws IOException {
}
void setMasterKey(long ledgerId, byte[] masterKey) throws IOException {
- FileInfo fi = null;
+ CachedFileInfo fi = null;
try {
fi = getFileInfo(ledgerId, masterKey);
} finally {
@@ -482,7 +424,7 @@ void setMasterKey(long ledgerId, byte[] masterKey) throws IOException {
}
boolean setFenced(long ledgerId) throws IOException {
- FileInfo fi = null;
+ CachedFileInfo fi = null;
try {
fi = getFileInfo(ledgerId, null);
return fi.setFenced();
@@ -494,7 +436,7 @@ boolean setFenced(long ledgerId) throws IOException {
}
boolean isFenced(long ledgerId) throws IOException {
- FileInfo fi = null;
+ CachedFileInfo fi = null;
try {
fi = getFileInfo(ledgerId, null);
return fi.isFenced();
@@ -506,7 +448,7 @@ boolean isFenced(long ledgerId) throws IOException {
}
void setExplicitLac(long ledgerId, ByteBuf lac) throws IOException {
- FileInfo fi = null;
+ CachedFileInfo fi = null;
try {
fi = getFileInfo(ledgerId, null);
fi.setExplicitLac(lac);
@@ -519,7 +461,7 @@ void setExplicitLac(long ledgerId, ByteBuf lac) throws IOException {
}
public ByteBuf getExplicitLac(long ledgerId) {
- FileInfo fi = null;
+ CachedFileInfo fi = null;
try {
fi = getFileInfo(ledgerId, null);
return fi.getExplicitLac();
@@ -600,7 +542,7 @@ private void moveLedgerIndexFile(Long l, FileInfo fi) throws NoWritableLedgerDir
}
void flushLedgerHeader(long ledger) throws IOException {
- FileInfo fi = null;
+ CachedFileInfo fi = null;
try {
fi = getFileInfo(ledger, null);
relocateIndexFileAndFlushHeader(ledger, fi);
@@ -616,7 +558,7 @@ void flushLedgerHeader(long ledger) throws IOException {
}
void flushLedgerEntries(long l, List<LedgerEntryPage> entries) throws IOException {
- FileInfo fi = null;
+ CachedFileInfo fi = null;
try {
Collections.sort(entries, new Comparator<LedgerEntryPage>() {
@Override
@@ -711,7 +653,7 @@ boolean updatePage(LedgerEntryPage lep) throws IOException {
if (!lep.isClean()) {
throw new IOException("Trying to update a dirty page");
}
- FileInfo fi = null;
+ CachedFileInfo fi = null;
try {
fi = getFileInfo(lep.getLedger(), null);
long pos = lep.getFirstEntryPosition();
@@ -730,7 +672,7 @@ boolean updatePage(LedgerEntryPage lep) throws IOException {
}
long getPersistEntryBeyondInMem(long ledgerId, long lastEntryInMem) throws IOException {
- FileInfo fi = null;
+ CachedFileInfo fi = null;
long lastEntry = lastEntryInMem;
try {
fi = getFileInfo(ledgerId, null);
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/IndexPersistenceMgrTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/IndexPersistenceMgrTest.java
index 974bde296..35c860429 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/IndexPersistenceMgrTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/IndexPersistenceMgrTest.java
@@ -28,6 +28,7 @@
import static org.junit.Assert.fail;
import java.io.File;
+import org.apache.bookkeeper.bookie.FileInfoBackingCache.CachedFileInfo;
import org.apache.bookkeeper.common.util.Watcher;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.stats.NullStatsLogger;
@@ -151,12 +152,12 @@ public void testGetFileInfoReadBeforeWrite() throws Exception {
assertEquals(0, indexPersistenceMgr.writeFileInfoCache.size());
assertEquals(0, indexPersistenceMgr.readFileInfoCache.size());
- FileInfo writeFileInfo = indexPersistenceMgr.getFileInfo(lid, masterKey);
- assertEquals(3, writeFileInfo.getUseCount());
+ CachedFileInfo writeFileInfo = indexPersistenceMgr.getFileInfo(lid, masterKey);
+ assertEquals(2, writeFileInfo.getRefCount());
assertEquals(1, indexPersistenceMgr.writeFileInfoCache.size());
- assertEquals(1, indexPersistenceMgr.readFileInfoCache.size());
+ assertEquals(0, indexPersistenceMgr.readFileInfoCache.size());
writeFileInfo.release();
- assertEquals(2, writeFileInfo.getUseCount());
+ assertEquals(1, writeFileInfo.getRefCount());
} finally {
if (null != indexPersistenceMgr) {
indexPersistenceMgr.close();
@@ -170,19 +171,19 @@ public void testGetFileInfoWriteBeforeRead() throws Exception {
try {
indexPersistenceMgr = createIndexPersistenceManager(1);
- FileInfo writeFileInfo = indexPersistenceMgr.getFileInfo(lid, masterKey);
- assertEquals(3, writeFileInfo.getUseCount());
+ CachedFileInfo writeFileInfo = indexPersistenceMgr.getFileInfo(lid, masterKey);
+ assertEquals(2, writeFileInfo.getRefCount());
assertEquals(1, indexPersistenceMgr.writeFileInfoCache.size());
- assertEquals(1, indexPersistenceMgr.readFileInfoCache.size());
+ assertEquals(0, indexPersistenceMgr.readFileInfoCache.size());
writeFileInfo.release();
- FileInfo readFileInfo = indexPersistenceMgr.getFileInfo(lid, null);
- assertEquals(3, readFileInfo.getUseCount());
+ CachedFileInfo readFileInfo = indexPersistenceMgr.getFileInfo(lid, null);
+ assertEquals(3, readFileInfo.getRefCount());
assertEquals(1, indexPersistenceMgr.writeFileInfoCache.size());
assertEquals(1, indexPersistenceMgr.readFileInfoCache.size());
readFileInfo.release();
- assertEquals(2, writeFileInfo.getUseCount());
- assertEquals(2, readFileInfo.getUseCount());
+ assertEquals(2, writeFileInfo.getRefCount());
+ assertEquals(2, readFileInfo.getRefCount());
} finally {
if (null != indexPersistenceMgr) {
indexPersistenceMgr.close();
@@ -196,9 +197,13 @@ public void testReadFileInfoCacheEviction() throws Exception {
try {
indexPersistenceMgr = createIndexPersistenceManager(1);
for (int i = 0; i < 3; i++) {
- FileInfo fileInfo = indexPersistenceMgr.getFileInfo(lid + i, masterKey);
+ CachedFileInfo fileInfo = indexPersistenceMgr.getFileInfo(lid + i, masterKey);
// We need to make sure index file is created, otherwise the test case can be flaky
fileInfo.checkOpen(true);
+ fileInfo.release();
+
+ // load into read cache also
+ indexPersistenceMgr.getFileInfo(lid + i, null).release();
}
indexPersistenceMgr.getFileInfo(lid, masterKey);
@@ -212,9 +217,9 @@ public void testReadFileInfoCacheEviction() throws Exception {
assertEquals(1, indexPersistenceMgr.writeFileInfoCache.size());
assertEquals(2, indexPersistenceMgr.readFileInfoCache.size());
- FileInfo fileInfo = indexPersistenceMgr.writeFileInfoCache.asMap().get(lid);
+ CachedFileInfo fileInfo = indexPersistenceMgr.writeFileInfoCache.asMap().get(lid);
assertNotNull(fileInfo);
- assertEquals(2, fileInfo.getUseCount());
+ assertEquals(2, fileInfo.getRefCount());
fileInfo = indexPersistenceMgr.writeFileInfoCache.asMap().get(lid + 1);
assertNull(fileInfo);
fileInfo = indexPersistenceMgr.writeFileInfoCache.asMap().get(lid + 2);
@@ -223,10 +228,10 @@ public void testReadFileInfoCacheEviction() throws Exception {
assertNull(fileInfo);
fileInfo = indexPersistenceMgr.readFileInfoCache.asMap().get(lid + 1);
assertNotNull(fileInfo);
- assertEquals(2, fileInfo.getUseCount());
+ assertEquals(2, fileInfo.getRefCount());
fileInfo = indexPersistenceMgr.readFileInfoCache.asMap().get(lid + 2);
assertNotNull(fileInfo);
- assertEquals(2, fileInfo.getUseCount());
+ assertEquals(2, fileInfo.getRefCount());
} finally {
if (null != indexPersistenceMgr) {
indexPersistenceMgr.close();
@@ -264,4 +269,36 @@ public void testEvictionShouldNotAffectLongPollRead() throws Exception {
}
}
}
+
+ @Test
+ public void testEvictBeforeReleaseRace() throws Exception {
+ IndexPersistenceMgr indexPersistenceMgr = null;
+ Watcher<LastAddConfirmedUpdateNotification> watcher = notification -> notification.recycle();
+ try {
+ indexPersistenceMgr = createIndexPersistenceManager(1);
+
+ indexPersistenceMgr.getFileInfo(1L, masterKey);
+ indexPersistenceMgr.getFileInfo(2L, masterKey);
+ indexPersistenceMgr.getFileInfo(3L, masterKey);
+ indexPersistenceMgr.getFileInfo(4L, masterKey);
+
+ CachedFileInfo fi = indexPersistenceMgr.getFileInfo(1L, masterKey);
+
+ // trigger eviction
+ indexPersistenceMgr.getFileInfo(2L, masterKey);
+ indexPersistenceMgr.getFileInfo(3L, null);
+ indexPersistenceMgr.getFileInfo(4L, null);
+
+ Thread.sleep(1000);
+
+ fi.setFenced();
+ fi.release();
+
+ assertTrue(indexPersistenceMgr.isFenced(1));
+ } finally {
+ if (null != indexPersistenceMgr) {
+ indexPersistenceMgr.close();
+ }
+ }
+ }
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/LedgerCacheTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/LedgerCacheTest.java
index 2fe5e75db..b14358b66 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/LedgerCacheTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/LedgerCacheTest.java
@@ -38,6 +38,7 @@
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.bookkeeper.bookie.Bookie.NoLedgerException;
+import org.apache.bookkeeper.bookie.FileInfoBackingCache.CachedFileInfo;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
import org.apache.bookkeeper.meta.LedgerManager;
@@ -272,7 +273,6 @@ public void testLedgerCacheFlushFailureOnDiskFull() throws Exception {
File ledgerDir2 = createTempDir("bkTest", ".dir");
ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
conf.setLedgerDirNames(new String[] { ledgerDir1.getAbsolutePath(), ledgerDir2.getAbsolutePath() });
- conf.setJournalDirName(ledgerDir1.toString());
Bookie bookie = new Bookie(conf);
InterleavedLedgerStorage ledgerStorage = ((InterleavedLedgerStorage) bookie.ledgerStorage);
@@ -280,12 +280,8 @@ public void testLedgerCacheFlushFailureOnDiskFull() throws Exception {
// Create ledger index file
ledgerStorage.setMasterKey(1, "key".getBytes());
- FileInfo fileInfo = ledgerCache.getIndexPersistenceManager().getFileInfo(Long.valueOf(1), null);
+ CachedFileInfo fileInfo = ledgerCache.getIndexPersistenceManager().getFileInfo(Long.valueOf(1), null);
- // Simulate the flush failure
- FileInfo newFileInfo = new FileInfo(fileInfo.getLf(), fileInfo.getMasterKey());
- ledgerCache.getIndexPersistenceManager().writeFileInfoCache.put(Long.valueOf(1), newFileInfo);
- ledgerCache.getIndexPersistenceManager().readFileInfoCache.put(Long.valueOf(1), newFileInfo);
// Add entries
ledgerStorage.addEntry(generateEntry(1, 1));
ledgerStorage.addEntry(generateEntry(1, 2));
@@ -294,13 +290,11 @@ public void testLedgerCacheFlushFailureOnDiskFull() throws Exception {
ledgerStorage.addEntry(generateEntry(1, 3));
// add the dir to failed dirs
bookie.getIndexDirsManager().addToFilledDirs(
- newFileInfo.getLf().getParentFile().getParentFile().getParentFile());
- File before = newFileInfo.getLf();
+ fileInfo.getLf().getParentFile().getParentFile().getParentFile());
+ File before = fileInfo.getLf();
// flush after disk is added as failed.
ledgerStorage.flush();
- File after = newFileInfo.getLf();
-
- assertEquals("Reference counting for the file info should be zero.", 0, newFileInfo.getUseCount());
+ File after = fileInfo.getLf();
assertFalse("After flush index file should be changed", before.equals(after));
// Verify written entries
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/TestFileInfoBackingCache.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/TestFileInfoBackingCache.java
new file mode 100644
index 000000000..489db74f3
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/TestFileInfoBackingCache.java
@@ -0,0 +1,288 @@
+/**
+ *
+ * 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.bookie;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.RemovalNotification;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.LongStream;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.bookie.FileInfoBackingCache.CachedFileInfo;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Tests for FileInfoBackingCache.
+ */
+@Slf4j
+public class TestFileInfoBackingCache {
+ final byte[] masterKey = new byte[0];
+ final File baseDir;
+ final ThreadFactory threadFactory = new ThreadFactoryBuilder()
+ .setNameFormat("backing-cache-test-%d").setDaemon(true).build();
+ ExecutorService executor;
+
+ public TestFileInfoBackingCache() throws Exception {
+ baseDir = File.createTempFile("foo", "bar");
+ }
+
+ @Before
+ public void setup() throws Exception {
+ Assert.assertTrue(baseDir.delete());
+ Assert.assertTrue(baseDir.mkdirs());
+ baseDir.deleteOnExit();
+
+ executor = Executors.newCachedThreadPool(threadFactory);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (executor != null) {
+ executor.shutdown();
+ }
+ }
+
+ @Test
+ public void basicTest() throws Exception {
+ FileInfoBackingCache cache = new FileInfoBackingCache(
+ (ledgerId, createIfNotFound) -> {
+ File f = new File(baseDir, String.valueOf(ledgerId));
+ f.deleteOnExit();
+ return f;
+ });
+ CachedFileInfo fi = cache.loadFileInfo(1, masterKey);
+ Assert.assertEquals(fi.getRefCount(), 1);
+ CachedFileInfo fi2 = cache.loadFileInfo(2, masterKey);
+ Assert.assertEquals(fi2.getRefCount(), 1);
+ CachedFileInfo fi3 = cache.loadFileInfo(1, null);
+ Assert.assertEquals(fi, fi3);
+ Assert.assertEquals(fi3.getRefCount(), 2);
+
+ // check that it expires correctly
+ fi.release();
+ fi3.release();
+
+ Assert.assertEquals(fi.getRefCount(), FileInfoBackingCache.DEAD_REF);
+ CachedFileInfo fi4 = cache.loadFileInfo(1, null);
+ Assert.assertFalse(fi4 == fi);
+ Assert.assertEquals(fi.getRefCount(), FileInfoBackingCache.DEAD_REF);
+ Assert.assertEquals(fi4.getRefCount(), 1);
+ Assert.assertEquals(fi.getLf(), fi4.getLf());
+ }
+
+ @Test(expected = IOException.class)
+ public void testNoKey() throws Exception {
+ FileInfoBackingCache cache = new FileInfoBackingCache(
+ (ledgerId, createIfNotFound) -> {
+ Assert.assertFalse(createIfNotFound);
+ throw new Bookie.NoLedgerException(ledgerId);
+ });
+ cache.loadFileInfo(1, null);
+ }
+
+ /**
+ * Of course this can't prove they don't exist, but
+ * try to shake them out none the less.
+ */
+ @Test
+ public void testForDeadlocks() throws Exception {
+ int numRunners = 20;
+ int maxLedgerId = 10;
+ AtomicBoolean done = new AtomicBoolean(false);
+
+ FileInfoBackingCache cache = new FileInfoBackingCache(
+ (ledgerId, createIfNotFound) -> {
+ File f = new File(baseDir, String.valueOf(ledgerId));
+ f.deleteOnExit();
+ return f;
+ });
+ Iterable<Future<Set<CachedFileInfo>>> futures =
+ IntStream.range(0, numRunners).mapToObj(
+ (i) -> {
+ Callable<Set<CachedFileInfo>> c = () -> {
+ Random r = new Random();
+ List<CachedFileInfo> fileInfos = new ArrayList<>();
+ Set<CachedFileInfo> allFileInfos = new HashSet<>();
+ while (!done.get()) {
+ if (r.nextBoolean() && fileInfos.size() < 5) { // take a reference
+ CachedFileInfo fi = cache.loadFileInfo(r.nextInt(maxLedgerId), masterKey);
+ Assert.assertFalse(fi.isClosed());
+ allFileInfos.add(fi);
+ fileInfos.add(fi);
+ } else { // release a reference
+ Collections.shuffle(fileInfos);
+ if (!fileInfos.isEmpty()) {
+ fileInfos.remove(0).release();
+ }
+ }
+ }
+ for (CachedFileInfo fi : fileInfos) {
+ Assert.assertFalse(fi.isClosed());
+ fi.release();
+ }
+ return allFileInfos;
+ };
+ return executor.submit(c);
+ }).collect(Collectors.toList());
+ Thread.sleep(TimeUnit.SECONDS.toMillis(10));
+ done.set(true);
+
+ // ensure all threads are finished operating on cache, before checking any
+ for (Future<Set<CachedFileInfo>> f : futures) {
+ f.get();
+ }
+
+ for (Future<Set<CachedFileInfo>> f : futures) {
+ for (CachedFileInfo fi : f.get()) {
+ Assert.assertTrue(fi.isClosed());
+ Assert.assertEquals(FileInfoBackingCache.DEAD_REF, fi.getRefCount());
+ }
+ }
+
+ // try to load all ledgers again.
+ // They should be loaded fresh (i.e. this load should be only reference)
+ for (int i = 0; i < maxLedgerId; i++) {
+ Assert.assertEquals(1, cache.loadFileInfo(i, masterKey).getRefCount());
+ }
+ }
+
+ @Test
+ public void testRefCountRace() throws Exception {
+ AtomicBoolean done = new AtomicBoolean(false);
+ FileInfoBackingCache cache = new FileInfoBackingCache(
+ (ledgerId, createIfNotFound) -> {
+ File f = new File(baseDir, String.valueOf(ledgerId));
+ f.deleteOnExit();
+ return f;
+ });
+
+ Iterable<Future<Set<CachedFileInfo>>> futures =
+ IntStream.range(0, 2).mapToObj(
+ (i) -> {
+ Callable<Set<CachedFileInfo>> c = () -> {
+ Set<CachedFileInfo> allFileInfos = new HashSet<>();
+ while (!done.get()) {
+ CachedFileInfo fi = cache.loadFileInfo(1, masterKey);
+ Assert.assertFalse(fi.isClosed());
+ allFileInfos.add(fi);
+ fi.release();
+ }
+ return allFileInfos;
+ };
+ return executor.submit(c);
+ }).collect(Collectors.toList());
+ Thread.sleep(TimeUnit.SECONDS.toMillis(10));
+ done.set(true);
+
+ // ensure all threads are finished operating on cache, before checking any
+ for (Future<Set<CachedFileInfo>> f : futures) {
+ f.get();
+ }
+
+ for (Future<Set<CachedFileInfo>> f : futures) {
+ for (CachedFileInfo fi : f.get()) {
+ Assert.assertTrue(fi.isClosed());
+ Assert.assertEquals(FileInfoBackingCache.DEAD_REF, fi.getRefCount());
+ }
+ }
+ }
+
+ private void guavaEvictionListener(RemovalNotification<Long, CachedFileInfo> notification) {
+ notification.getValue().release();
+ }
+
+ @Test
+ public void testRaceGuavaEvictAndReleaseBeforeRetain() throws Exception {
+ AtomicBoolean done = new AtomicBoolean(false);
+ FileInfoBackingCache cache = new FileInfoBackingCache(
+ (ledgerId, createIfNotFound) -> {
+ File f = new File(baseDir, String.valueOf(ledgerId));
+ f.deleteOnExit();
+ return f;
+ });
+
+ Cache<Long, CachedFileInfo> guavaCache = CacheBuilder.newBuilder()
+ .maximumSize(1)
+ .removalListener(this::guavaEvictionListener)
+ .build();
+
+ Iterable<Future<Set<CachedFileInfo>>> futures =
+ LongStream.range(0L, 2L).mapToObj(
+ (i) -> {
+ Callable<Set<CachedFileInfo>> c = () -> {
+ Set<CachedFileInfo> allFileInfos = new HashSet<>();
+ while (!done.get()) {
+ CachedFileInfo fi = null;
+
+ do {
+ fi = guavaCache.get(
+ i, () -> cache.loadFileInfo(i, masterKey));
+ allFileInfos.add(fi);
+ Thread.sleep(100);
+ } while (!fi.tryRetain());
+
+ Assert.assertFalse(fi.isClosed());
+ fi.release();
+ }
+ return allFileInfos;
+ };
+ return executor.submit(c);
+ }).collect(Collectors.toList());
+ Thread.sleep(TimeUnit.SECONDS.toMillis(10));
+ done.set(true);
+
+ // ensure all threads are finished operating on cache, before checking any
+ for (Future<Set<CachedFileInfo>> f : futures) {
+ f.get();
+ }
+ guavaCache.invalidateAll();
+
+ for (Future<Set<CachedFileInfo>> f : futures) {
+ for (CachedFileInfo fi : f.get()) {
+ Assert.assertTrue(fi.isClosed());
+ Assert.assertEquals(FileInfoBackingCache.DEAD_REF, fi.getRefCount());
+ }
+ }
+
+ }
+}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services