Java 类java.util.concurrent.locks.AbstractQueuedSynchronizer.ConditionObject 实例源码

/**
 * Awaits condition using the specified AwaitMethod.
 */
void await(ConditionObject c, AwaitMethod awaitMethod)
        throws InterruptedException {
    long timeoutMillis = 2 * LONG_DELAY_MS;
    switch (awaitMethod) {
    case await:
        c.await();
        break;
    case awaitTimed:
        assertTrue(c.await(timeoutMillis, MILLISECONDS));
        break;
    case awaitNanos:
        long nanosTimeout = MILLISECONDS.toNanos(timeoutMillis);
        long nanosRemaining = c.awaitNanos(nanosTimeout);
        assertTrue(nanosRemaining > 0);
        break;
    case awaitUntil:
        assertTrue(c.awaitUntil(delayedDate(timeoutMillis)));
        break;
    default:
        throw new AssertionError();
    }
}

public void testSignal(final AwaitMethod awaitMethod) {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final BooleanLatch acquired = new BooleanLatch();
    Thread t = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            sync.acquire();
            assertTrue(acquired.releaseShared(0));
            await(c, awaitMethod);
            sync.release();
        }});

    acquired.acquireShared(0);
    sync.acquire();
    assertHasWaitersLocked(sync, c, t);
    assertHasExclusiveQueuedThreads(sync, NO_THREADS);
    c.signal();
    assertHasWaitersLocked(sync, c, NO_THREADS);
    assertHasExclusiveQueuedThreads(sync, t);
    sync.release();
    awaitTermination(t);
}

/**
 * Awaits condition using the specified AwaitMethod.
 */
void await(ConditionObject c, AwaitMethod awaitMethod)
        throws InterruptedException {
    long timeoutMillis = 2 * LONG_DELAY_MS;
    switch (awaitMethod) {
    case await:
        c.await();
        break;
    case awaitTimed:
        assertTrue(c.await(timeoutMillis, MILLISECONDS));
        break;
    case awaitNanos:
        long nanosTimeout = MILLISECONDS.toNanos(timeoutMillis);
        long nanosRemaining = c.awaitNanos(nanosTimeout);
        assertTrue(nanosRemaining > 0);
        break;
    case awaitUntil:
        assertTrue(c.awaitUntil(delayedDate(timeoutMillis)));
        break;
    default:
        throw new AssertionError();
    }
}

public void testSignal(final AwaitMethod awaitMethod) {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final BooleanLatch acquired = new BooleanLatch();
    Thread t = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            sync.acquire();
            assertTrue(acquired.releaseShared(0));
            await(c, awaitMethod);
            sync.release();
        }});

    acquired.acquireShared(0);
    sync.acquire();
    assertHasWaitersLocked(sync, c, t);
    assertHasExclusiveQueuedThreads(sync, NO_THREADS);
    c.signal();
    assertHasWaitersLocked(sync, c, NO_THREADS);
    assertHasExclusiveQueuedThreads(sync, t);
    sync.release();
    awaitTermination(t);
}

/**
 * Checks that condition c has exactly the given waiter threads,
 * after acquiring mutex.
 */
void assertHasWaitersUnlocked(Mutex sync, ConditionObject c,
                             Thread... threads) {
    sync.acquire();
    assertHasWaitersLocked(sync, c, threads);
    sync.release();
}

/**
 * Checks that condition c has exactly the given waiter threads.
 */
void assertHasWaitersLocked(Mutex sync, ConditionObject c,
                            Thread... threads) {
    assertEquals(threads.length > 0, sync.hasWaiters(c));
    assertEquals(threads.length, sync.getWaitQueueLength(c));
    assertEquals(threads.length == 0, sync.getWaitingThreads(c).isEmpty());
    assertEquals(threads.length, sync.getWaitingThreads(c).size());
    assertEquals(new HashSet<Thread>(sync.getWaitingThreads(c)),
                 new HashSet<Thread>(Arrays.asList(threads)));
}

/**
 * Checks that awaiting the given condition times out (using the
 * default timeout duration).
 */
void assertAwaitTimesOut(ConditionObject c, AwaitMethod awaitMethod) {
    final long timeoutMillis = timeoutMillis();
    final long startTime;
    try {
        switch (awaitMethod) {
        case awaitTimed:
            startTime = System.nanoTime();
            assertFalse(c.await(timeoutMillis, MILLISECONDS));
            assertTrue(millisElapsedSince(startTime) >= timeoutMillis);
            break;
        case awaitNanos:
            startTime = System.nanoTime();
            long nanosTimeout = MILLISECONDS.toNanos(timeoutMillis);
            long nanosRemaining = c.awaitNanos(nanosTimeout);
            assertTrue(nanosRemaining <= 0);
            assertTrue(nanosRemaining > -MILLISECONDS.toNanos(LONG_DELAY_MS));
            assertTrue(millisElapsedSince(startTime) >= timeoutMillis);
            break;
        case awaitUntil:
            // We shouldn't assume that nanoTime and currentTimeMillis
            // use the same time source, so don't use nanoTime here.
            java.util.Date delayedDate = delayedDate(timeoutMillis);
            assertFalse(c.awaitUntil(delayedDate(timeoutMillis)));
            assertTrue(new java.util.Date().getTime() >= delayedDate.getTime());
            break;
        default:
            throw new UnsupportedOperationException();
        }
    } catch (InterruptedException ie) { threadUnexpectedException(ie); }
}

/**
 * owns is true for a condition created by sync else false
 */
public void testOwns() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final Mutex sync2 = new Mutex();
    assertTrue(sync.owns(c));
    assertFalse(sync2.owns(c));
}

/**
 * Calling await without holding sync throws IllegalMonitorStateException
 */
public void testAwait_IMSE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    for (AwaitMethod awaitMethod : AwaitMethod.values()) {
        long startTime = System.nanoTime();
        try {
            await(c, awaitMethod);
            shouldThrow();
        } catch (IllegalMonitorStateException success) {
        } catch (InterruptedException e) { threadUnexpectedException(e); }
        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS);
    }
}

/**
 * Calling signal without holding sync throws IllegalMonitorStateException
 */
public void testSignal_IMSE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    try {
        c.signal();
        shouldThrow();
    } catch (IllegalMonitorStateException success) {}
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * Calling signalAll without holding sync throws IllegalMonitorStateException
 */
public void testSignalAll_IMSE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    try {
        c.signalAll();
        shouldThrow();
    } catch (IllegalMonitorStateException success) {}
}

public void testAwait_Timeout(AwaitMethod awaitMethod) {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    sync.acquire();
    assertAwaitTimesOut(c, awaitMethod);
    sync.release();
}

/**
 * hasWaiters throws IllegalArgumentException if not owned
 */
public void testHasWaitersIAE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final Mutex sync2 = new Mutex();
    try {
        sync2.hasWaiters(c);
        shouldThrow();
    } catch (IllegalArgumentException success) {}
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * hasWaiters throws IllegalMonitorStateException if not synced
 */
public void testHasWaitersIMSE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    try {
        sync.hasWaiters(c);
        shouldThrow();
    } catch (IllegalMonitorStateException success) {}
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * getWaitQueueLength throws IllegalArgumentException if not owned
 */
public void testGetWaitQueueLengthIAE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final Mutex sync2 = new Mutex();
    try {
        sync2.getWaitQueueLength(c);
        shouldThrow();
    } catch (IllegalArgumentException success) {}
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * getWaitQueueLength throws IllegalMonitorStateException if not synced
 */
public void testGetWaitQueueLengthIMSE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    try {
        sync.getWaitQueueLength(c);
        shouldThrow();
    } catch (IllegalMonitorStateException success) {}
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * getWaitingThreads throws IllegalArgumentException if not owned
 */
public void testGetWaitingThreadsIAE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final Mutex sync2 = new Mutex();
    try {
        sync2.getWaitingThreads(c);
        shouldThrow();
    } catch (IllegalArgumentException success) {}
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * getWaitingThreads throws IllegalMonitorStateException if not synced
 */
public void testGetWaitingThreadsIMSE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    try {
        sync.getWaitingThreads(c);
        shouldThrow();
    } catch (IllegalMonitorStateException success) {}
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * hasWaiters returns true when a thread is waiting, else false
 */
public void testHasWaiters() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final BooleanLatch acquired = new BooleanLatch();
    Thread t = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            sync.acquire();
            assertHasWaitersLocked(sync, c, NO_THREADS);
            assertFalse(sync.hasWaiters(c));
            assertTrue(acquired.releaseShared(0));
            c.await();
            sync.release();
        }});

    acquired.acquireShared(0);
    sync.acquire();
    assertHasWaitersLocked(sync, c, t);
    assertHasExclusiveQueuedThreads(sync, NO_THREADS);
    assertTrue(sync.hasWaiters(c));
    c.signal();
    assertHasWaitersLocked(sync, c, NO_THREADS);
    assertHasExclusiveQueuedThreads(sync, t);
    assertFalse(sync.hasWaiters(c));
    sync.release();

    awaitTermination(t);
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * awaitUninterruptibly is uninterruptible
 */
public void testAwaitUninterruptibly() {
    final Mutex sync = new Mutex();
    final ConditionObject condition = sync.newCondition();
    final BooleanLatch pleaseInterrupt = new BooleanLatch();
    Thread t = newStartedThread(new CheckedRunnable() {
        public void realRun() {
            sync.acquire();
            assertTrue(pleaseInterrupt.releaseShared(0));
            condition.awaitUninterruptibly();
            assertTrue(Thread.interrupted());
            assertHasWaitersLocked(sync, condition, NO_THREADS);
            sync.release();
        }});

    pleaseInterrupt.acquireShared(0);
    sync.acquire();
    assertHasWaitersLocked(sync, condition, t);
    sync.release();
    t.interrupt();
    assertHasWaitersUnlocked(sync, condition, t);
    assertThreadBlocks(t, Thread.State.WAITING);
    sync.acquire();
    assertHasWaitersLocked(sync, condition, t);
    assertHasExclusiveQueuedThreads(sync, NO_THREADS);
    condition.signal();
    assertHasWaitersLocked(sync, condition, NO_THREADS);
    assertHasExclusiveQueuedThreads(sync, t);
    sync.release();
    awaitTermination(t);
}

public void testInterruptible(final AwaitMethod awaitMethod) {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final BooleanLatch pleaseInterrupt = new BooleanLatch();
    Thread t = newStartedThread(new CheckedInterruptedRunnable() {
        public void realRun() throws InterruptedException {
            sync.acquire();
            assertTrue(pleaseInterrupt.releaseShared(0));
            await(c, awaitMethod);
        }});

    pleaseInterrupt.acquireShared(0);
    t.interrupt();
    awaitTermination(t);
}

public void testSignalAll(final AwaitMethod awaitMethod) {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final BooleanLatch acquired1 = new BooleanLatch();
    final BooleanLatch acquired2 = new BooleanLatch();
    Thread t1 = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            sync.acquire();
            acquired1.releaseShared(0);
            await(c, awaitMethod);
            sync.release();
        }});

    Thread t2 = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            sync.acquire();
            acquired2.releaseShared(0);
            await(c, awaitMethod);
            sync.release();
        }});

    acquired1.acquireShared(0);
    acquired2.acquireShared(0);
    sync.acquire();
    assertHasWaitersLocked(sync, c, t1, t2);
    assertHasExclusiveQueuedThreads(sync, NO_THREADS);
    c.signalAll();
    assertHasWaitersLocked(sync, c, NO_THREADS);
    assertHasExclusiveQueuedThreads(sync, t1, t2);
    sync.release();
    awaitTermination(t1);
    awaitTermination(t2);
}

/**
 * awaitNanos/timed await with 0 wait times out immediately
 */
public void testAwait_Zero() throws InterruptedException {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    sync.acquire();
    assertTrue(c.awaitNanos(0L) <= 0);
    assertFalse(c.await(0L, NANOSECONDS));
    sync.release();
}

/**
 * awaitNanos/timed await with maximum negative wait times does not underflow
 */
public void testAwait_NegativeInfinity() throws InterruptedException {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    sync.acquire();
    assertTrue(c.awaitNanos(Long.MIN_VALUE) <= 0);
    assertFalse(c.await(Long.MIN_VALUE, NANOSECONDS));
    sync.release();
}

/**
 * Checks that condition c has exactly the given waiter threads,
 * after acquiring mutex.
 */
void assertHasWaitersUnlocked(Mutex sync, ConditionObject c,
                             Thread... threads) {
    sync.acquire();
    assertHasWaitersLocked(sync, c, threads);
    sync.release();
}

/**
 * Checks that condition c has exactly the given waiter threads.
 */
void assertHasWaitersLocked(Mutex sync, ConditionObject c,
                            Thread... threads) {
    assertEquals(threads.length > 0, sync.hasWaiters(c));
    assertEquals(threads.length, sync.getWaitQueueLength(c));
    assertEquals(threads.length == 0, sync.getWaitingThreads(c).isEmpty());
    assertEquals(threads.length, sync.getWaitingThreads(c).size());
    assertEquals(new HashSet<Thread>(sync.getWaitingThreads(c)),
                 new HashSet<Thread>(Arrays.asList(threads)));
}

/**
 * Checks that awaiting the given condition times out (using the
 * default timeout duration).
 */
void assertAwaitTimesOut(ConditionObject c, AwaitMethod awaitMethod) {
    long timeoutMillis = timeoutMillis();
    long startTime;
    try {
        switch (awaitMethod) {
        case awaitTimed:
            startTime = System.nanoTime();
            assertFalse(c.await(timeoutMillis, MILLISECONDS));
            assertTrue(millisElapsedSince(startTime) >= timeoutMillis);
            break;
        case awaitNanos:
            startTime = System.nanoTime();
            long nanosTimeout = MILLISECONDS.toNanos(timeoutMillis);
            long nanosRemaining = c.awaitNanos(nanosTimeout);
            assertTrue(nanosRemaining <= 0);
            assertTrue(nanosRemaining > -MILLISECONDS.toNanos(LONG_DELAY_MS));
            assertTrue(millisElapsedSince(startTime) >= timeoutMillis);
            break;
        case awaitUntil:
            // We shouldn't assume that nanoTime and currentTimeMillis
            // use the same time source, so don't use nanoTime here.
            java.util.Date delayedDate = delayedDate(timeoutMillis());
            assertFalse(c.awaitUntil(delayedDate(timeoutMillis)));
            assertTrue(new java.util.Date().getTime() >= delayedDate.getTime());
            break;
        default:
            throw new UnsupportedOperationException();
        }
    } catch (InterruptedException ie) { threadUnexpectedException(ie); }
}

/**
 * owns is true for a condition created by sync else false
 */
public void testOwns() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final Mutex sync2 = new Mutex();
    assertTrue(sync.owns(c));
    assertFalse(sync2.owns(c));
}

/**
 * Calling await without holding sync throws IllegalMonitorStateException
 */
public void testAwait_IMSE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    for (AwaitMethod awaitMethod : AwaitMethod.values()) {
        long startTime = System.nanoTime();
        try {
            await(c, awaitMethod);
            shouldThrow();
        } catch (IllegalMonitorStateException success) {
        } catch (InterruptedException e) { threadUnexpectedException(e); }
        assertTrue(millisElapsedSince(startTime) < LONG_DELAY_MS);
    }
}

/**
 * Calling signal without holding sync throws IllegalMonitorStateException
 */
public void testSignal_IMSE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    try {
        c.signal();
        shouldThrow();
    } catch (IllegalMonitorStateException success) {}
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * Calling signalAll without holding sync throws IllegalMonitorStateException
 */
public void testSignalAll_IMSE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    try {
        c.signalAll();
        shouldThrow();
    } catch (IllegalMonitorStateException success) {}
}

public void testAwait_Timeout(AwaitMethod awaitMethod) {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    sync.acquire();
    assertAwaitTimesOut(c, awaitMethod);
    sync.release();
}

/**
 * hasWaiters throws IllegalArgumentException if not owned
 */
public void testHasWaitersIAE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final Mutex sync2 = new Mutex();
    try {
        sync2.hasWaiters(c);
        shouldThrow();
    } catch (IllegalArgumentException success) {}
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * hasWaiters throws IllegalMonitorStateException if not synced
 */
public void testHasWaitersIMSE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    try {
        sync.hasWaiters(c);
        shouldThrow();
    } catch (IllegalMonitorStateException success) {}
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * getWaitQueueLength throws IllegalArgumentException if not owned
 */
public void testGetWaitQueueLengthIAE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final Mutex sync2 = new Mutex();
    try {
        sync2.getWaitQueueLength(c);
        shouldThrow();
    } catch (IllegalArgumentException success) {}
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * getWaitQueueLength throws IllegalMonitorStateException if not synced
 */
public void testGetWaitQueueLengthIMSE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    try {
        sync.getWaitQueueLength(c);
        shouldThrow();
    } catch (IllegalMonitorStateException success) {}
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * getWaitingThreads throws IllegalArgumentException if not owned
 */
public void testGetWaitingThreadsIAE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final Mutex sync2 = new Mutex();
    try {
        sync2.getWaitingThreads(c);
        shouldThrow();
    } catch (IllegalArgumentException success) {}
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * getWaitingThreads throws IllegalMonitorStateException if not synced
 */
public void testGetWaitingThreadsIMSE() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    try {
        sync.getWaitingThreads(c);
        shouldThrow();
    } catch (IllegalMonitorStateException success) {}
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * hasWaiters returns true when a thread is waiting, else false
 */
public void testHasWaiters() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final BooleanLatch acquired = new BooleanLatch();
    Thread t = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            sync.acquire();
            assertHasWaitersLocked(sync, c, NO_THREADS);
            assertFalse(sync.hasWaiters(c));
            assertTrue(acquired.releaseShared(0));
            c.await();
            sync.release();
        }});

    acquired.acquireShared(0);
    sync.acquire();
    assertHasWaitersLocked(sync, c, t);
    assertHasExclusiveQueuedThreads(sync, NO_THREADS);
    assertTrue(sync.hasWaiters(c));
    c.signal();
    assertHasWaitersLocked(sync, c, NO_THREADS);
    assertHasExclusiveQueuedThreads(sync, t);
    assertFalse(sync.hasWaiters(c));
    sync.release();

    awaitTermination(t);
    assertHasWaitersUnlocked(sync, c, NO_THREADS);
}

/**
 * awaitUninterruptibly is uninterruptible
 */
public void testAwaitUninterruptibly() {
    final Mutex sync = new Mutex();
    final ConditionObject c = sync.newCondition();
    final BooleanLatch pleaseInterrupt = new BooleanLatch();
    Thread t = newStartedThread(new CheckedRunnable() {
        public void realRun() {
            sync.acquire();
            assertTrue(pleaseInterrupt.releaseShared(0));
            c.awaitUninterruptibly();
            assertTrue(Thread.interrupted());
            assertHasWaitersLocked(sync, c, NO_THREADS);
            sync.release();
        }});

    pleaseInterrupt.acquireShared(0);
    sync.acquire();
    assertHasWaitersLocked(sync, c, t);
    sync.release();
    t.interrupt();
    assertHasWaitersUnlocked(sync, c, t);
    assertThreadStaysAlive(t);
    sync.acquire();
    assertHasWaitersLocked(sync, c, t);
    assertHasExclusiveQueuedThreads(sync, NO_THREADS);
    c.signal();
    assertHasWaitersLocked(sync, c, NO_THREADS);
    assertHasExclusiveQueuedThreads(sync, t);
    sync.release();
    awaitTermination(t);
}