diff --git hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/AutoCloseableRWLock.java hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/AutoCloseableRWLock.java new file mode 100644 index 00000000000..d08d5f0d587 --- /dev/null +++ hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/AutoCloseableRWLock.java @@ -0,0 +1,185 @@ +/** + * 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.hadoop.util; + +import com.google.common.annotations.VisibleForTesting; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +/** + * This is a wrap class of a ReentrantLock readWrite lock. Extending AutoCloseable + * interface such that the users can use a try-with-resource syntax. + */ +public class AutoCloseableRWLock implements AutoCloseable { + private final ReentrantReadWriteLock lock; + + /** + * Creates an instance of {@code AutoCloseableRWLock}, initializes + * the underlying lock instance with a new {@code ReentrantReadWriteLock}. + */ + public AutoCloseableRWLock() { + this(new ReentrantReadWriteLock()); + } + + /** + * Wrap provided Lock instance. + * @param lock Lock instance to wrap in AutoCloseable API. + */ + public AutoCloseableRWLock(ReentrantReadWriteLock lock) { + this.lock = lock; + } + + /** + * A wrapper method that makes a call to {@code readlock()} of the underlying + * {@code ReentrantReadWriteLock} object. + * + * Acquire the lock it is not held by another thread, then sets + * lock held count to one, then returns immediately. + * + * If the current thread already holds the lock, increase the lock + * help count by one and returns immediately. + * + * If the lock is held by another thread, the current thread is + * suspended until the lock has been acquired by current thread. + * + * @return The {@code AutoCloseableRWLock} object itself. This is to + * support try-with-resource syntax. + */ + public AutoCloseableRWLock acquireRead() { + this.lock.readLock().lock(); + return this; + } + + /** + * A wrapper method that makes a call to {@code writelock()} of the underlying + * {@code ReentrantReadWriteLock} object. + * + * Acquire the lock it is not held by another thread, then sets + * lock held count to one, then returns immediately. + * + * If the current thread already holds the lock, increase the lock + * help count by one and returns immediately. + * + * If the lock is held by another thread, the current thread is + * suspended until the lock has been acquired by current thread. + * + * @return The {@code AutoCloseableRWLock} object itself. This is to + * support try-with-resource syntax. + */ + public AutoCloseableRWLock acquireWrite() { + this.lock.writeLock().lock(); + return this; + } + + /** + * A wrapper method that makes a call to {@code tryLock()} of + * the underlying {@code ReadLock} object. + * + * If the lock is not held by another thread, acquires the lock, set the + * hold count to one and returns {@code true}. + * + * If the current thread already holds the lock, the increment the hold + * count by one and returns {@code true}. + * + * If the lock is held by another thread then the method returns + * immediately with {@code false}. + * + * @return {@code true} if the lock was free and was acquired by the + * current thread, or the lock was already held by the current + * thread; and {@code false} otherwise. + */ + public boolean tryRLock() { + return this.lock.readLock().tryLock(); + } + + /** + * A wrapper method that makes a call to {@code tryLock()} of + * the underlying {@code WriteLock} object. + * + * If the lock is not held by another thread, acquires the lock, set the + * hold count to one and returns {@code true}. + * + * If the current thread already holds the lock, the increment the hold + * count by one and returns {@code true}. + * + * If the lock is held by another thread then the method returns + * immediately with {@code false}. + * + * @return {@code true} if the lock was free and was acquired by the + * current thread, or the lock was already held by the current + * thread; and {@code false} otherwise. + */ + public boolean tryWLock() { + return this.lock.writeLock().tryLock(); + } + + /** + * A wrapper method that makes a call to {@code unlock()} of the + * underlying {@code ReentrantReadWriteLock} object. + * + * Attempts to release the lock. + * + * If the current thread holds the lock, decrements the hold + * count. If the hold count reaches zero, the lock is released. + * + * If the current thread does not hold the lock, then + * {@link IllegalMonitorStateException} is thrown. + */ + public void release() { + // firs twe should release read locks to support ReentrantDowngrades + if (this.lock.getReadLockCount() > 0) { + this.lock.readLock().unlock(); + return; + } + if (this.lock.isWriteLocked()) { + this.lock.writeLock().unlock(); + } + } + + /** + * Attempts to release the lock by making a call to {@code release()}. + * + * This is to implement {@code close()} method from {@code AutoCloseable} + * interface. This allows users to user a try-with-resource syntax, where + * the lock can be automatically released. + */ + @Override + public void close() { + release(); + } + + /** + * A wrapper method that makes a call to {@code isLocked()} of + * the underlying {@code ReentrantLock} object. + * + * Queries if this lock is held by any thread. This method is + * designed for use in monitoring of the system state, + * not for synchronization control. + * + * @return {@code true} if any thread holds this lock and + * {@code false} otherwise + */ + @VisibleForTesting + public boolean isLocked() { + return (this.lock.isWriteLocked() || this.lock.getReadLockCount() > 0); + } + + public boolean isExcLockedByCurrentThread() { + return this.lock.isWriteLockedByCurrentThread(); + } +} diff --git hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestAutoCloseableRWLock.java hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestAutoCloseableRWLock.java new file mode 100644 index 00000000000..62f941a414a --- /dev/null +++ hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestAutoCloseableRWLock.java @@ -0,0 +1,176 @@ +package org.apache.hadoop.util; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; + +public class TestAutoCloseableRWLock { + /** + * Test the basic lock and unlock operation. + */ + @Test + public void testLockAcquireRelease() { + AutoCloseableRWLock lock = new AutoCloseableRWLock(); + AutoCloseableRWLock newlock = lock.acquireWrite(); + // Ensure acquire the same lock object. + assertEquals(newlock, lock); + // Ensure it locked now. + assertTrue(lock.isLocked()); + lock.release(); + newlock = lock.acquireRead(); + // Ensure acquire the same lock object. + assertEquals(newlock, lock); + lock.release(); + // Ensure it is unlocked now. + assertFalse(lock.isLocked()); + } + + @Test + public void testDowngradeLock() throws Exception { + AutoCloseableRWLock lock = new AutoCloseableRWLock(); + AutoCloseableRWLock newlock = lock.acquireWrite(); + // Ensure acquire the same lock object. + assertEquals(newlock, lock); + // Ensure it locked now. + assertTrue(lock.isLocked()); + // can we downgrade before releasing lock? + boolean tryRead = newlock.tryRLock(); + assertTrue(tryRead); + // try to read from another thread should fail + Thread competingWriteThreadA = new Thread() { + @Override + public void run() { + assertTrue(lock.isLocked()); + assertFalse(lock.tryWLock()); + assertFalse(lock.tryRLock()); + } + }; + competingWriteThreadA.start(); + competingWriteThreadA.join(); + lock.release(); + // Ensure it is still locked now because of writeLocked. + assertTrue(lock.isLocked()); + // try to read from another thread should fail + Thread competingWriteThread = new Thread() { + @Override + public void run() { + assertTrue(lock.isLocked()); + assertFalse(lock.tryWLock()); + assertFalse(lock.tryRLock()); + } + }; + competingWriteThread.start(); + competingWriteThread.join(); + assertTrue(lock.isLocked()); + lock.release(); + assertFalse(lock.isLocked()); + } + + /** + * Test when write lock is acquired, no other thread can + * lock it. + * + * @throws Exception + */ + @Test + public void testWriteMultipleThread() throws Exception { + AutoCloseableRWLock lock = new AutoCloseableRWLock(); + lock.acquireWrite(); + assertTrue(lock.isLocked()); + Thread competingThread = new Thread() { + @Override + public void run() { + assertTrue(lock.isLocked()); + assertFalse(lock.tryWLock()); + assertFalse(lock.tryRLock()); + } + }; + competingThread.start(); + competingThread.join(); + assertTrue(lock.isLocked()); + lock.release(); + assertFalse(lock.isLocked()); + } + + /** + * Test when read lock is acquired, other thread can + * read lock it. + * + * @throws Exception + */ + @Test + public void testReadMultipleThread() throws Exception { + AutoCloseableRWLock lock = new AutoCloseableRWLock(); + lock.acquireRead(); + assertTrue(lock.isLocked()); + Thread competingThread = new Thread() { + @Override + public void run() { + assertTrue(lock.isLocked()); + assertFalse(lock.tryWLock()); + assertTrue(lock.tryRLock()); + lock.release(); + } + }; + competingThread.start(); + competingThread.join(); + assertTrue(lock.isLocked()); + lock.release(); + assertFalse(lock.isLocked()); + } + + /** + * Test the correctness under try-with-resource syntax. + * + * @throws Exception + */ + @Test + public void testWriteTryWithResourceSyntax() throws Exception { + AutoCloseableRWLock lock = new AutoCloseableRWLock(); + try(AutoCloseableRWLock localLock = lock.acquireWrite()) { + assertEquals(localLock, lock); + assertTrue(lock.isLocked()); + Thread competingThread = new Thread() { + @Override + public void run() { + assertTrue(lock.isLocked()); + assertFalse(lock.tryRLock()); + assertFalse(lock.tryWLock()); + } + }; + competingThread.start(); + competingThread.join(); + assertTrue(localLock.isLocked()); + } + assertFalse(lock.isLocked()); + } + + /** + * Test the correctness under try-with-resource syntax. + * + * @throws Exception + */ + @Test + public void testReadTryWithResourceSyntax() throws Exception { + AutoCloseableRWLock lock = new AutoCloseableRWLock(); + try(AutoCloseableRWLock localLock = lock.acquireRead()) { + assertEquals(localLock, lock); + assertTrue(lock.isLocked()); + Thread competingThread = new Thread() { + @Override + public void run() { + assertTrue(lock.isLocked()); + assertTrue(lock.tryRLock()); + assertFalse(lock.tryWLock()); + lock.release(); + } + }; + competingThread.start(); + competingThread.join(); + assertTrue(localLock.isLocked()); + } + assertFalse(lock.isLocked()); + } +}