HBASE-27064 CME in TestRegionNormalizerWorkQueue (#4468)
Signed-off-by: Viraj Jasani <vjasani@apache.org>
This commit is contained in:
parent
4ce85aa39f
commit
88d9c4bef4
|
@ -25,7 +25,7 @@ import java.util.Queue;
|
|||
import java.util.Set;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.locks.Condition;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
import java.util.concurrent.locks.ReentrantReadWriteLock;
|
||||
import org.apache.yetus.audience.InterfaceAudience;
|
||||
|
||||
/**
|
||||
|
@ -39,8 +39,6 @@ import org.apache.yetus.audience.InterfaceAudience;
|
|||
* {@link BlockingQueue}.</li>
|
||||
* <li>Allows a producer to insert an item at the head of the queue, if desired.</li>
|
||||
* </ul>
|
||||
* Assumes low-frequency and low-parallelism concurrent access, so protects state using a simplistic
|
||||
* synchronization strategy.
|
||||
*/
|
||||
@InterfaceAudience.Private
|
||||
class RegionNormalizerWorkQueue<E> {
|
||||
|
@ -48,53 +46,15 @@ class RegionNormalizerWorkQueue<E> {
|
|||
/** Underlying storage structure that gives us the Set behavior and FIFO retrieval policy. */
|
||||
private LinkedHashSet<E> delegate;
|
||||
|
||||
// the locking structure used here follows the example found in LinkedBlockingQueue. The
|
||||
// difference is that our locks guard access to `delegate` rather than the head node.
|
||||
|
||||
/** Lock held by take, poll, etc */
|
||||
private final ReentrantLock takeLock;
|
||||
|
||||
/** Lock for puts and takes **/
|
||||
private final ReentrantReadWriteLock lock;
|
||||
/** Wait queue for waiting takes */
|
||||
private final Condition notEmpty;
|
||||
|
||||
/** Lock held by put, offer, etc */
|
||||
private final ReentrantLock putLock;
|
||||
|
||||
RegionNormalizerWorkQueue() {
|
||||
delegate = new LinkedHashSet<>();
|
||||
takeLock = new ReentrantLock();
|
||||
notEmpty = takeLock.newCondition();
|
||||
putLock = new ReentrantLock();
|
||||
}
|
||||
|
||||
/**
|
||||
* Signals a waiting take. Called only from put/offer (which do not otherwise ordinarily lock
|
||||
* takeLock.)
|
||||
*/
|
||||
private void signalNotEmpty() {
|
||||
final ReentrantLock takeLock = this.takeLock;
|
||||
takeLock.lock();
|
||||
try {
|
||||
notEmpty.signal();
|
||||
} finally {
|
||||
takeLock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Locks to prevent both puts and takes.
|
||||
*/
|
||||
private void fullyLock() {
|
||||
putLock.lock();
|
||||
takeLock.lock();
|
||||
}
|
||||
|
||||
/**
|
||||
* Unlocks to allow both puts and takes.
|
||||
*/
|
||||
private void fullyUnlock() {
|
||||
takeLock.unlock();
|
||||
putLock.unlock();
|
||||
lock = new ReentrantReadWriteLock();
|
||||
notEmpty = lock.writeLock().newCondition();
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -105,16 +65,14 @@ class RegionNormalizerWorkQueue<E> {
|
|||
if (e == null) {
|
||||
throw new NullPointerException();
|
||||
}
|
||||
|
||||
putLock.lock();
|
||||
lock.writeLock().lock();
|
||||
try {
|
||||
delegate.add(e);
|
||||
if (!delegate.isEmpty()) {
|
||||
notEmpty.signal();
|
||||
}
|
||||
} finally {
|
||||
putLock.unlock();
|
||||
}
|
||||
|
||||
if (!delegate.isEmpty()) {
|
||||
signalNotEmpty();
|
||||
lock.writeLock().unlock();
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -138,16 +96,14 @@ class RegionNormalizerWorkQueue<E> {
|
|||
if (c == null) {
|
||||
throw new NullPointerException();
|
||||
}
|
||||
|
||||
putLock.lock();
|
||||
lock.writeLock().lock();
|
||||
try {
|
||||
delegate.addAll(c);
|
||||
if (!delegate.isEmpty()) {
|
||||
notEmpty.signal();
|
||||
}
|
||||
} finally {
|
||||
putLock.unlock();
|
||||
}
|
||||
|
||||
if (!delegate.isEmpty()) {
|
||||
signalNotEmpty();
|
||||
lock.writeLock().unlock();
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -159,19 +115,17 @@ class RegionNormalizerWorkQueue<E> {
|
|||
if (c == null) {
|
||||
throw new NullPointerException();
|
||||
}
|
||||
|
||||
fullyLock();
|
||||
lock.writeLock().lock();
|
||||
try {
|
||||
final LinkedHashSet<E> copy = new LinkedHashSet<>(c.size() + delegate.size());
|
||||
copy.addAll(c);
|
||||
copy.addAll(delegate);
|
||||
delegate = copy;
|
||||
if (!delegate.isEmpty()) {
|
||||
notEmpty.signal();
|
||||
}
|
||||
} finally {
|
||||
fullyUnlock();
|
||||
}
|
||||
|
||||
if (!delegate.isEmpty()) {
|
||||
signalNotEmpty();
|
||||
lock.writeLock().unlock();
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -183,10 +137,13 @@ class RegionNormalizerWorkQueue<E> {
|
|||
*/
|
||||
public E take() throws InterruptedException {
|
||||
E x;
|
||||
takeLock.lockInterruptibly();
|
||||
// Take a write lock. If the delegate's queue is empty we need it to await(), which will
|
||||
// drop the lock, then reacquire it; or if the queue is not empty we will use an iterator
|
||||
// to mutate the head.
|
||||
lock.writeLock().lockInterruptibly();
|
||||
try {
|
||||
while (delegate.isEmpty()) {
|
||||
notEmpty.await();
|
||||
notEmpty.await(); // await drops the lock, then reacquires it
|
||||
}
|
||||
final Iterator<E> iter = delegate.iterator();
|
||||
x = iter.next();
|
||||
|
@ -195,7 +152,7 @@ class RegionNormalizerWorkQueue<E> {
|
|||
notEmpty.signal();
|
||||
}
|
||||
} finally {
|
||||
takeLock.unlock();
|
||||
lock.writeLock().unlock();
|
||||
}
|
||||
return x;
|
||||
}
|
||||
|
@ -205,11 +162,11 @@ class RegionNormalizerWorkQueue<E> {
|
|||
* returns.
|
||||
*/
|
||||
public void clear() {
|
||||
putLock.lock();
|
||||
lock.writeLock().lock();
|
||||
try {
|
||||
delegate.clear();
|
||||
} finally {
|
||||
putLock.unlock();
|
||||
lock.writeLock().unlock();
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -218,21 +175,21 @@ class RegionNormalizerWorkQueue<E> {
|
|||
* @return the number of elements in this queue
|
||||
*/
|
||||
public int size() {
|
||||
takeLock.lock();
|
||||
lock.readLock().lock();
|
||||
try {
|
||||
return delegate.size();
|
||||
} finally {
|
||||
takeLock.unlock();
|
||||
lock.readLock().unlock();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
takeLock.lock();
|
||||
lock.readLock().lock();
|
||||
try {
|
||||
return delegate.toString();
|
||||
} finally {
|
||||
takeLock.unlock();
|
||||
lock.readLock().unlock();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue