abdullah alamoudi has submitted this change and it was merged. Change subject: [NO ISSUE][TX] Fix DatasetLock for Multiple Index Builds ......................................................................
[NO ISSUE][TX] Fix DatasetLock for Multiple Index Builds - user model changes: no - storage format changes: no - interface changes: no details: - The mechanism used for allowing multiple concurrent index builds does not work if the first index build finishes before other index builds. It relied on a write lock obtained by the first index builder and released by the last index builder. This is not allowed when using ReentrantReadWriteLock and will lead to an IllegalMonitorStateException since the last thread to exit did not hold the lock. - Potential Deadlock between modify and exclusive modify can happen when an exclusive modify attempts to upgrade the lock to write lock while the modify lock waits for the exclusive lock to release. This has been fixed too. - Test cases were added. Change-Id: I3bea3ff2075d952ab13402b0c445c464b431c0f5 Reviewed-on: https://asterix-gerrit.ics.uci.edu/2144 Integration-Tests: Jenkins <[email protected]> Tested-by: Jenkins <[email protected]> Contrib: Jenkins <[email protected]> Reviewed-by: Michael Blow <[email protected]> --- M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/RecoveryTask.java M asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/DatasetLock.java A asterixdb/asterix-metadata/src/test/java/org/apache/asterix/metadata/lock/MetadataLockManagerTest.java 3 files changed, 384 insertions(+), 39 deletions(-) Approvals: Jenkins: Verified; No violations found; Verified Michael Blow: Looks good to me, approved Objections: Anon. E. Moose #1000171: Violations found Jenkins: Violations found diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/RecoveryTask.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/RecoveryTask.java index 29b38ba..0321ae4 100644 --- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/RecoveryTask.java +++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/RecoveryTask.java @@ -134,8 +134,9 @@ lockManager.acquireActiveEntityWriteLock(metadataProvider.getLocks(), listener.getEntityId().getDataverse() + '.' + listener.getEntityId().getEntityName()); for (Dataset dataset : listener.getDatasets()) { - MetadataLockUtil.modifyDatasetBegin(lockManager, metadataProvider.getLocks(), - dataset.getDataverseName(), DatasetUtil.getFullyQualifiedName(dataset)); + lockManager.acquireDataverseReadLock(metadataProvider.getLocks(), dataset.getDataverseName()); + lockManager.acquireDatasetExclusiveModificationLock(metadataProvider.getLocks(), + DatasetUtil.getFullyQualifiedName(dataset)); } synchronized (listener) { try { diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/DatasetLock.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/DatasetLock.java index 31f2089..bc302c2 100644 --- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/DatasetLock.java +++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/lock/DatasetLock.java @@ -24,23 +24,31 @@ import org.apache.asterix.common.exceptions.ErrorCode; import org.apache.asterix.common.exceptions.MetadataException; import org.apache.asterix.common.metadata.IMetadataLock; +import org.apache.asterix.common.utils.InterruptUtil; import org.apache.commons.lang3.mutable.MutableInt; import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException; public class DatasetLock implements IMetadataLock { private final String key; + // The lock private final ReentrantReadWriteLock lock; - private final ReentrantReadWriteLock dsReadLock; - private final ReentrantReadWriteLock dsModifyLock; + // Used for lock upgrade operation + private final ReentrantReadWriteLock upgradeLock; + // Used for exclusive modification + private final ReentrantReadWriteLock modifyLock; + // The two counters below are used to ensure mutual exclusivity between index builds and modifications + // order of entry indexBuildCounter -> indexModifyCounter private final MutableInt indexBuildCounter; + private final MutableInt dsModifyCounter; public DatasetLock(String key) { this.key = key; lock = new ReentrantReadWriteLock(true); - dsReadLock = new ReentrantReadWriteLock(true); - dsModifyLock = new ReentrantReadWriteLock(true); + upgradeLock = new ReentrantReadWriteLock(true); + modifyLock = new ReentrantReadWriteLock(true); indexBuildCounter = new MutableInt(0); + dsModifyCounter = new MutableInt(0); } private void readLock() { @@ -71,63 +79,97 @@ lock.writeLock().unlock(); } - private void readReadLock() { - dsReadLock.readLock().lock(); + private void upgradeReadLock() { + upgradeLock.readLock().lock(); } private void modifyReadLock() { // insert - dsModifyLock.readLock().lock(); + modifyLock.readLock().lock(); + incrementModifyCounter(); + } + + private void incrementModifyCounter() { + InterruptUtil.doUninterruptibly(() -> { + synchronized (indexBuildCounter) { + while (indexBuildCounter.getValue() > 0) { + indexBuildCounter.wait(); + } + synchronized (dsModifyCounter) { + dsModifyCounter.increment(); + } + } + }); + } + + private void decrementModifyCounter() { + synchronized (indexBuildCounter) { + synchronized (dsModifyCounter) { + if (dsModifyCounter.decrementAndGet() == 0) { + indexBuildCounter.notifyAll(); + } + } + } } private void modifyReadUnlock() { // insert - dsModifyLock.readLock().unlock(); + decrementModifyCounter(); + modifyLock.readLock().unlock(); } - private void readReadUnlock() { - dsReadLock.readLock().unlock(); + private void upgradeReadUnlock() { + upgradeLock.readLock().unlock(); } - private void readWriteUnlock() { - dsReadLock.writeLock().unlock(); + private void upgradeWriteUnlock() { + upgradeLock.writeLock().unlock(); } - private void modifySharedWriteLock() { + private void buildIndexLock() { // Build index statement synchronized (indexBuildCounter) { if (indexBuildCounter.getValue() > 0) { - indexBuildCounter.setValue(indexBuildCounter.getValue() + 1); + indexBuildCounter.increment(); } else { - dsModifyLock.writeLock().lock(); - indexBuildCounter.setValue(1); + InterruptUtil.doUninterruptibly(() -> { + while (true) { + synchronized (dsModifyCounter) { + if (dsModifyCounter.getValue() == 0) { + indexBuildCounter.increment(); + return; + } + } + indexBuildCounter.wait(); + } + }); } } } - private void modifySharedWriteUnlock() { + private void buildIndexUnlock() { // Build index statement synchronized (indexBuildCounter) { - if (indexBuildCounter.getValue() == 1) { - dsModifyLock.writeLock().unlock(); + if (indexBuildCounter.decrementAndGet() == 0) { + indexBuildCounter.notifyAll(); } - indexBuildCounter.setValue(indexBuildCounter.getValue() - 1); } } - private void modifyExclusiveWriteLock() { - dsModifyLock.writeLock().lock(); + private void modifyWriteLock() { + modifyLock.writeLock().lock(); + incrementModifyCounter(); } private void modifyExclusiveWriteUnlock() { - dsModifyLock.writeLock().unlock(); + decrementModifyCounter(); + modifyLock.writeLock().unlock(); } @Override public void upgrade(IMetadataLock.Mode from, IMetadataLock.Mode to) throws AlgebricksException { if (from == IMetadataLock.Mode.EXCLUSIVE_MODIFY && to == IMetadataLock.Mode.UPGRADED_WRITE) { - dsReadLock.readLock().unlock(); - dsReadLock.writeLock().lock(); + upgradeLock.writeLock().lock(); } else { throw new MetadataException(ErrorCode.ILLEGAL_LOCK_UPGRADE_OPERATION, from, to); } @@ -136,8 +178,7 @@ @Override public void downgrade(IMetadataLock.Mode from, IMetadataLock.Mode to) throws AlgebricksException { if (from == IMetadataLock.Mode.UPGRADED_WRITE && to == IMetadataLock.Mode.EXCLUSIVE_MODIFY) { - dsReadLock.writeLock().unlock(); - dsReadLock.readLock().lock(); + upgradeLock.writeLock().unlock(); } else { throw new MetadataException(ErrorCode.ILLEGAL_LOCK_DOWNGRADE_OPERATION, from, to); } @@ -148,24 +189,22 @@ switch (mode) { case INDEX_BUILD: readLock(); - modifySharedWriteLock(); + buildIndexLock(); break; case MODIFY: readLock(); - readReadLock(); modifyReadLock(); break; case EXCLUSIVE_MODIFY: readLock(); - readReadLock(); - modifyExclusiveWriteLock(); + modifyWriteLock(); break; case WRITE: writeLock(); break; case READ: readLock(); - readReadLock(); + upgradeReadLock(); break; default: throw new IllegalStateException("locking mode " + mode + " is not supported"); @@ -176,28 +215,26 @@ public void unlock(IMetadataLock.Mode mode) { switch (mode) { case INDEX_BUILD: - modifySharedWriteUnlock(); + buildIndexUnlock(); readUnlock(); break; case MODIFY: modifyReadUnlock(); - readReadUnlock(); readUnlock(); break; case EXCLUSIVE_MODIFY: modifyExclusiveWriteUnlock(); - readReadUnlock(); readUnlock(); break; case WRITE: writeUnlock(); break; case READ: - readReadUnlock(); + upgradeReadUnlock(); readUnlock(); break; case UPGRADED_WRITE: - readWriteUnlock(); + upgradeWriteUnlock(); modifyExclusiveWriteUnlock(); readUnlock(); break; diff --git a/asterixdb/asterix-metadata/src/test/java/org/apache/asterix/metadata/lock/MetadataLockManagerTest.java b/asterixdb/asterix-metadata/src/test/java/org/apache/asterix/metadata/lock/MetadataLockManagerTest.java new file mode 100644 index 0000000..4382860 --- /dev/null +++ b/asterixdb/asterix-metadata/src/test/java/org/apache/asterix/metadata/lock/MetadataLockManagerTest.java @@ -0,0 +1,307 @@ +/* + * 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.asterix.metadata.lock; + +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.Semaphore; + +import org.apache.asterix.common.metadata.LockList; +import org.apache.hyracks.api.util.SingleThreadEventProcessor; +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +@RunWith(Parameterized.class) +public class MetadataLockManagerTest { + + static final int REPREAT_TEST_COUNT = 3; + + @Parameterized.Parameters + public static List<Object[]> data() { + return Arrays.asList(new Object[REPREAT_TEST_COUNT][0]); + } + + private static class Request { + private enum Statement { + INDEX, + MODIFY, + EXCLUSIVE_MODIFY, + EXCLUSIVE_MODIFY_UPGRADE_DOWNGRADE, + EXCLUSIVE_MODIFY_UPGRADE, + } + + private final Statement statement; + private final String dataset; + private boolean done; + private int step = 0; + + public Request(Statement statement, String dataset) { + this.statement = statement; + this.dataset = dataset; + done = false; + } + + Statement statement() { + return statement; + } + + String dataset() { + return dataset; + } + + synchronized void complete() { + done = true; + notifyAll(); + } + + synchronized void await() throws InterruptedException { + while (!done) { + wait(); + } + } + + synchronized void step() { + step++; + notifyAll(); + } + + synchronized int getSteps() { + return step; + } + + synchronized void await(int step) throws InterruptedException { + while (this.step < step) { + wait(); + } + } + } + + public class User extends SingleThreadEventProcessor<Request> { + + private MetadataLockManager lockManager; + private Semaphore step = new Semaphore(0); + private final LockList locks = new LockList(); + + public User(String username, MetadataLockManager lockManager) { + super(username); + this.lockManager = lockManager; + } + + public void step() { + step.release(); + } + + @Override + protected void handle(Request req) throws Exception { + try { + step.acquire(); + switch (req.statement()) { + case INDEX: + lockManager.acquireDatasetCreateIndexLock(locks, req.dataset()); + break; + case MODIFY: + lockManager.acquireDatasetModifyLock(locks, req.dataset()); + break; + case EXCLUSIVE_MODIFY: + lockManager.acquireDatasetExclusiveModificationLock(locks, req.dataset()); + break; + case EXCLUSIVE_MODIFY_UPGRADE: + lockManager.acquireDatasetExclusiveModificationLock(locks, req.dataset()); + req.step(); + step.acquire(); + lockManager.upgradeDatasetLockToWrite(locks, req.dataset()); + break; + case EXCLUSIVE_MODIFY_UPGRADE_DOWNGRADE: + lockManager.acquireDatasetExclusiveModificationLock(locks, req.dataset()); + req.step(); + step.acquire(); + lockManager.upgradeDatasetLockToWrite(locks, req.dataset()); + req.step(); + step.acquire(); + lockManager.downgradeDatasetLockToExclusiveModify(locks, req.dataset()); + break; + default: + break; + } + req.step(); + step.acquire(); + } finally { + locks.reset(); + req.step(); + req.complete(); + } + } + + } + + @Test + public void testDatasetLockMultipleIndexBuildsSingleModifier() throws Exception { + MetadataLockManager lockManager = new MetadataLockManager(); + String dataset = "Dataset"; + User till = new User("till", lockManager); + Request tReq = new Request(Request.Statement.INDEX, dataset); + User dmitry = new User("dmitry", lockManager); + Request dReq = new Request(Request.Statement.INDEX, dataset); + User mike = new User("mike", lockManager); + Request mReq = new Request(Request.Statement.MODIFY, dataset); + // Till builds an index + till.add(tReq); + // Dmitry builds an index + dmitry.add(dReq); + // Mike modifies + mike.add(mReq); + // Till starts + till.step(); + // Ensure lock acquired + tReq.await(1); + // Dmitry starts + dmitry.step(); + // Ensure lock acquired + dReq.await(1); + // Mike starts and is allowed to go all the way + mike.step(); + mike.step(); + // Ensure that Mike still could not acquire locks + Assert.assertEquals(0, mReq.getSteps()); + // Till finishes first + till.step(); + // Ensure the request has been completed and lock has been released + tReq.await(); + // Ensure that Mike still could not acquire locks + Assert.assertEquals(0, mReq.getSteps()); + // Dmitry finishes second + dmitry.step(); + // Ensure the request has been completed and lock has been released + dReq.await(); + // Ensure that Mike could proceed and request has been completed + mReq.await(); + // Stop users + till.stop(); + dmitry.stop(); + mike.stop(); + } + + @Test + public void testDatasetLockMultipleModifiersSingleIndexBuilder() throws Exception { + MetadataLockManager lockManager = new MetadataLockManager(); + String dataset = "Dataset"; + User till = new User("till", lockManager); + Request tReq = new Request(Request.Statement.MODIFY, dataset); + User dmitry = new User("dmitry", lockManager); + Request dReq = new Request(Request.Statement.MODIFY, dataset); + User mike = new User("mike", lockManager); + Request mReq = new Request(Request.Statement.INDEX, dataset); + // Till modifies + till.add(tReq); + // Dmitry modifies + dmitry.add(dReq); + // Mike builds an index + mike.add(mReq); + // Till starts + till.step(); + // Ensure lock acquired + tReq.await(1); + // Dmitry starts + dmitry.step(); + // Ensure lock acquired + dReq.await(1); + // Mike starts and is allowed to go all the way + mike.step(); + mike.step(); + // Ensure that Mike still could not acquire locks + Assert.assertEquals(0, mReq.getSteps()); + // Till finishes first + till.step(); + // Ensure the request has been completed and lock has been released + tReq.await(); + // Ensure that Mike still could not acquire locks + Assert.assertEquals(0, mReq.getSteps()); + // Dmitry finishes second + dmitry.step(); + // Ensure the request has been completed and lock has been released + dReq.await(); + // Ensure that Mike could proceed and request has been completed + mReq.await(); + // Stop users + till.stop(); + dmitry.stop(); + mike.stop(); + } + + @Test + public void testDatasetLockMultipleModifiersSingleExclusiveModifier() throws Exception { + MetadataLockManager lockManager = new MetadataLockManager(); + String dataset = "Dataset"; + User till = new User("till", lockManager); + Request tReq = new Request(Request.Statement.MODIFY, dataset); + User dmitry = new User("dmitry", lockManager); + Request dReq = new Request(Request.Statement.MODIFY, dataset); + User mike = new User("mike", lockManager); + Request mReq = new Request(Request.Statement.EXCLUSIVE_MODIFY, dataset); + // Till starts + till.add(tReq); + till.step(); + // Ensure lock is acquired + tReq.await(1); + // Mike starts + mike.add(mReq); + mike.step(); + // Sleep for 1s for now as there is no way to find out user has submitted the exclusive lock request + Thread.sleep(1000); + // Ensure that Mike didn't get the lock + Assert.assertEquals(0, mReq.getSteps()); + // Dmitry starts + dmitry.add(dReq); + dmitry.step(); + // Ensure that Dmitry didn't get the lock + Assert.assertEquals(0, dReq.getSteps()); + // Till proceeds + till.step(); + // Ensure the request has been completed and lock has been released + tReq.await(); + // Ensure that Mike got the lock + mReq.await(1); + // Till submits another request + tReq = new Request(Request.Statement.MODIFY, dataset); + till.add(tReq); + till.step(); + // Ensure that Till didn't get the lock + Assert.assertEquals(0, tReq.getSteps()); + // Ensure that Dmitry didn't get the lock + Assert.assertEquals(0, dReq.getSteps()); + // Mike completes + mike.step(); + mReq.await(); + // Ensure that both Till and Dmitry got the lock + tReq.await(1); + dReq.await(1); + till.step(); + dmitry.step(); + // Ensure that both Till and Dmitry complete + tReq.await(); + dReq.await(); + // Stop users + till.stop(); + dmitry.stop(); + mike.stop(); + } + +} -- To view, visit https://asterix-gerrit.ics.uci.edu/2144 To unsubscribe, visit https://asterix-gerrit.ics.uci.edu/settings Gerrit-MessageType: merged Gerrit-Change-Id: I3bea3ff2075d952ab13402b0c445c464b431c0f5 Gerrit-PatchSet: 10 Gerrit-Project: asterixdb Gerrit-Branch: release-0.9.3-pre-rc Gerrit-Owner: abdullah alamoudi <[email protected]> Gerrit-Reviewer: Anon. E. Moose #1000171 Gerrit-Reviewer: Jenkins <[email protected]> Gerrit-Reviewer: Michael Blow <[email protected]> Gerrit-Reviewer: Murtadha Hubail <[email protected]> Gerrit-Reviewer: Till Westmann <[email protected]> Gerrit-Reviewer: abdullah alamoudi <[email protected]>
