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/08/15 07:36:35 UTC
[GitHub] sijie closed pull request #1600: Provide consistent locking
mechanism in EntryLogManagerForEntryLogPerLedger
sijie closed pull request #1600: Provide consistent locking mechanism in EntryLogManagerForEntryLogPerLedger
URL: https://github.com/apache/bookkeeper/pull/1600
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/EntryLogManagerForEntryLogPerLedger.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java
index 41e66d9fe0..96f38c5945 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogManagerForEntryLogPerLedger.java
@@ -49,6 +49,7 @@
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -87,12 +88,16 @@ BufferedLogChannel getLogChannel() {
}
}
- static class EntryLogAndLockTuple {
+ class EntryLogAndLockTuple {
private final Lock ledgerLock;
private BufferedLogChannelWithDirInfo entryLogWithDirInfo;
- private EntryLogAndLockTuple() {
- ledgerLock = new ReentrantLock();
+ private EntryLogAndLockTuple(long ledgerId) {
+ int lockIndex = Long.hashCode(ledgerId) % lockArrayPool.length();
+ if (lockArrayPool.get(lockIndex) == null) {
+ lockArrayPool.compareAndSet(lockIndex, null, new ReentrantLock());
+ }
+ ledgerLock = lockArrayPool.get(lockIndex);
}
private Lock getLedgerLock() {
@@ -212,6 +217,7 @@ void doCounterMapCleanup() {
}
}
+ private final AtomicReferenceArray<Lock> lockArrayPool;
private final LoadingCache<Long, EntryLogAndLockTuple> ledgerIdEntryLogMap;
/*
* every time active logChannel is accessed from ledgerIdEntryLogMap
@@ -244,10 +250,11 @@ void doCounterMapCleanup() {
this.entryLogPerLedgerCounterLimitsMultFactor = conf.getEntryLogPerLedgerCounterLimitsMultFactor();
ledgerDirsManager.addLedgerDirsListener(getLedgerDirsListener());
+ this.lockArrayPool = new AtomicReferenceArray<Lock>(maximumNumberOfActiveEntryLogs * 2);
this.entryLogAndLockTupleCacheLoader = new CacheLoader<Long, EntryLogAndLockTuple>() {
@Override
public EntryLogAndLockTuple load(Long key) throws Exception {
- return new EntryLogAndLockTuple();
+ return new EntryLogAndLockTuple(key);
}
};
/*
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CreateNewLogTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CreateNewLogTest.java
index 3cdfd9fffe..f5d4edce96 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CreateNewLogTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CreateNewLogTest.java
@@ -34,10 +34,12 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
import java.util.stream.IntStream;
import org.apache.bookkeeper.bookie.EntryLogManagerForEntryLogPerLedger.BufferedLogChannelWithDirInfo;
import org.apache.bookkeeper.bookie.EntryLogger.BufferedLogChannel;
+import org.apache.bookkeeper.bookie.LedgerDirsManager.NoWritableLedgerDirException;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
import org.apache.bookkeeper.stats.Counter;
@@ -769,4 +771,119 @@ private static void createNewLogs(EntryLogManagerForEntryLogPerLedger entrylogMa
entrylogManager.createNewLog(ledgerId);
}
}
+
+ @Test
+ public void testLockConsistency() throws Exception {
+ ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
+
+ conf.setLedgerDirNames(ledgerDirs);
+ conf.setEntryLogFilePreAllocationEnabled(false);
+ conf.setEntryLogPerLedgerEnabled(true);
+ conf.setMaximumNumberOfActiveEntryLogs(5);
+
+ CountDownLatch latch = new CountDownLatch(1);
+ AtomicInteger count = new AtomicInteger(0);
+
+ /*
+ * Inject wait operation in 'getWritableLedgerDirsForNewLog' method of
+ * ledgerDirsManager. getWritableLedgerDirsForNewLog will be called when
+ * entryLogManager.createNewLog is called.
+ */
+ LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
+ new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold())) {
+ /*
+ * getWritableLedgerDirsForNewLog is called for the first time, it
+ * will await on 'latch' latch before calling super
+ * getWritableLedgerDirsForNewLog.
+ */
+ @Override
+ public List<File> getWritableLedgerDirsForNewLog() throws NoWritableLedgerDirException {
+ if (count.incrementAndGet() == 1) {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ LOG.error("Got InterruptedException while awaiting for latch countdown", e);
+ }
+ }
+ return super.getWritableLedgerDirsForNewLog();
+ }
+ };
+
+ EntryLogger el = new EntryLogger(conf, ledgerDirsManager);
+ EntryLogManagerForEntryLogPerLedger entryLogManager = (EntryLogManagerForEntryLogPerLedger) el
+ .getEntryLogManager();
+
+ long firstLedgerId = 100L;
+ AtomicBoolean newLogCreated = new AtomicBoolean(false);
+
+ Assert.assertFalse("EntryLogManager cacheMap should not contain entry for firstLedgerId",
+ entryLogManager.getCacheAsMap().containsKey(firstLedgerId));
+ Assert.assertEquals("Value of the count should be 0", 0, count.get());
+ /*
+ * In a new thread, create newlog for 'firstLedgerId' and then set
+ * 'newLogCreated' to true. Since this is the first createNewLog call,
+ * it is going to be blocked untill latch is countdowned to 0.
+ */
+ new Thread() {
+ @Override
+ public void run() {
+ try {
+ entryLogManager.createNewLog(firstLedgerId);
+ newLogCreated.set(true);
+ } catch (IOException e) {
+ LOG.error("Got IOException while creating new log", e);
+ }
+ }
+ }.start();
+
+ /*
+ * Wait until entry for 'firstLedgerId' is created in cacheMap. It will
+ * be created because in the other thread createNewLog is called.
+ */
+ while (!entryLogManager.getCacheAsMap().containsKey(firstLedgerId)) {
+ Thread.sleep(200);
+ }
+ Lock firstLedgersLock = entryLogManager.getLock(firstLedgerId);
+
+ /*
+ * since 'latch' is not counteddown, newlog should not be created even
+ * after waitign for 2 secs.
+ */
+ Thread.sleep(2000);
+ Assert.assertFalse("New log shouldn't have created", newLogCreated.get());
+
+ /*
+ * create MaximumNumberOfActiveEntryLogs of entrylogs and do cache
+ * cleanup, so that the earliest entry from cache will be removed.
+ */
+ for (int i = 1; i <= conf.getMaximumNumberOfActiveEntryLogs(); i++) {
+ entryLogManager.createNewLog(firstLedgerId + i);
+ }
+ entryLogManager.doEntryLogMapCleanup();
+ Assert.assertFalse("Entry for that ledger shouldn't be there",
+ entryLogManager.getCacheAsMap().containsKey(firstLedgerId));
+
+ /*
+ * now countdown the latch, so that the other thread can make progress
+ * with createNewLog and since this entry is evicted from cache,
+ * entrylog of the newly created entrylog will be added to
+ * rotatedentrylogs.
+ */
+ latch.countDown();
+ while (!newLogCreated.get()) {
+ Thread.sleep(200);
+ }
+ while (entryLogManager.getRotatedLogChannels().size() < 1) {
+ Thread.sleep(200);
+ }
+
+ /*
+ * Entry for 'firstLedgerId' is removed from cache, but even in this
+ * case when we get lock for the 'firstLedgerId' it should be the same
+ * as we got earlier.
+ */
+ Lock lockForThatLedgerAfterRemoval = entryLogManager.getLock(firstLedgerId);
+ Assert.assertEquals("For a given ledger lock should be the same before and after removal", firstLedgersLock,
+ lockForThatLedgerAfterRemoval);
+ }
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/EntryLogTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/EntryLogTest.java
index 1dc535cb0e..4b52a09b13 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/EntryLogTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/EntryLogTest.java
@@ -1003,10 +1003,17 @@ private void validateLockAcquireAndRelease(int numOfLedgers, int numOfThreadsPer
latchToStart.countDown();
Thread.sleep(1000);
/*
- * since there are only "numOfLedgers" ledgers, only "numOfLedgers" threads should have been able to acquire
- * lock. After acquiring the lock there must be waiting on 'latchToWait' latch
+ * since there are only "numOfLedgers" ledgers, only < "numOfLedgers"
+ * threads should have been able to acquire lock, because multiple
+ * ledgers can end up getting same lock because their hashcode might
+ * fall in the same bucket.
+ *
+ *
+ * After acquiring the lock there must be waiting on 'latchToWait' latch
*/
- assertEquals("Number Of Threads acquired Lock", numOfLedgers, numberOfThreadsAcquiredLock.get());
+ int currentNumberOfThreadsAcquiredLock = numberOfThreadsAcquiredLock.get();
+ assertTrue("Number Of Threads acquired Lock " + currentNumberOfThreadsAcquiredLock,
+ (currentNumberOfThreadsAcquiredLock > 0) && (currentNumberOfThreadsAcquiredLock <= numOfLedgers));
latchToWait.countDown();
Thread.sleep(2000);
assertEquals("Number Of Threads acquired Lock", numOfLedgers * numOfThreadsPerLedger,
----------------------------------------------------------------
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