import java.util.concurrent.locks.Condition; //导入依赖的package包/类
NIOConnection(final short p_ownNodeId, final short p_destination, final int p_bufferSize, final int p_flowControlWindowSize,
        final float p_flowControlWindowThreshold, final IncomingBufferQueue p_incomingBufferQueue, final MessageHeaderPool p_messageHeaderPool,
        final MessageDirectory p_messageDirectory, final RequestMap p_requestMap, final MessageHandlers p_messageHandlers, final BufferPool p_bufferPool,
        final AbstractExporterPool p_exporterPool, final NIOSelector p_nioSelector, final NodeMap p_nodeMap, final ReentrantLock p_lock,
        final Condition p_cond) {
    super(p_ownNodeId);

    NIOFlowControl flowControl = new NIOFlowControl(p_destination, p_flowControlWindowSize, p_flowControlWindowThreshold, p_nioSelector, this);
    NIOOutgoingRingBuffer outgoingBuffer = new NIOOutgoingRingBuffer(p_bufferSize, p_exporterPool);
    NIOPipeIn pipeIn =
            new NIOPipeIn(p_ownNodeId, p_destination, p_messageHeaderPool, flowControl, p_messageDirectory, p_requestMap, p_messageHandlers, p_bufferPool,
                    p_incomingBufferQueue, this);
    NIOPipeOut pipeOut = new NIOPipeOut(p_ownNodeId, p_destination, p_bufferSize, flowControl, outgoingBuffer, p_nioSelector, p_nodeMap, this);

    setPipes(pipeIn, pipeOut);

    m_nioSelector = p_nioSelector;

    m_connectionCondLock = p_lock;
    m_connectionCond = p_cond;
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
public void testSignalAll(boolean fair, final AwaitMethod awaitMethod) {
    final PublicReentrantLock lock = new PublicReentrantLock(fair);
    final Condition c = lock.newCondition();
    final CountDownLatch pleaseSignal = new CountDownLatch(2);
    class Awaiter extends CheckedRunnable {
        public void realRun() throws InterruptedException {
            lock.lock();
            pleaseSignal.countDown();
            await(c, awaitMethod);
            lock.unlock();
        }
    }

    Thread t1 = newStartedThread(new Awaiter());
    Thread t2 = newStartedThread(new Awaiter());

    await(pleaseSignal);
    lock.lock();
    assertHasWaiters(lock, c, t1, t2);
    c.signalAll();
    assertHasNoWaiters(lock, c);
    lock.unlock();
    awaitTermination(t1);
    awaitTermination(t2);
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
@Test
public void exceptionIsThrownAndEventIsCleanedUpIfAwaitTimesOut() throws TimeoutException, InterruptedException {
    SynchronizationEvent synchronizationEvent = mock(SynchronizationEvent.class);
    when(synchronizationEvent.isSatisfied()).thenReturn(false);

    Condition condition = mock(Condition.class);
    // condition#await returns false if time was exceeded
    when(condition.await(timeout, TimeUnit.SECONDS)).thenReturn(false);
    Lock lock = mock(Lock.class);
    when(lock.newCondition()).thenReturn(condition);

    SignalizationBasedEventSynchronizer signalizationBasedEventSynchronizer = new SignalizationBasedEventSynchronizer(timeout, lock);

    expectedException.expect(TimeoutException.class);
    signalizationBasedEventSynchronizer.awaitUntil(synchronizationEvent);
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
@Test
public void ifIllegalStateExceptionIsThrownIsIsLocked() throws TimeoutException, InterruptedException {
    SynchronizationEvent synchronizationEvent = mock(SynchronizationEvent.class);
    when(synchronizationEvent.isSatisfied()).thenReturn(false);

    Condition condition = mock(Condition.class);
    doAnswer(invocationOnMock -> {
        Thread.sleep(2000);
        return false;
    }).when(condition).await(timeout, TimeUnit.SECONDS);
    Lock lock = mock(Lock.class);
    when(lock.newCondition()).thenReturn(condition);
    doNothing().doThrow(new IllegalMonitorStateException()).when(lock).lock();

    SignalizationBasedEventSynchronizer signalizationBasedEventSynchronizer = new SignalizationBasedEventSynchronizer(timeout, lock);

    Thread signalizingThread = new Thread(new SignalizingRunnable(signalizationBasedEventSynchronizer, synchronizationEvent));

    expectedException.expect(TimeoutException.class);
    signalizingThread.start();
    signalizationBasedEventSynchronizer.awaitUntil(synchronizationEvent);
    signalizingThread.join();
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
public UnlockAll lockAll() {
	iLock.lock();
	try {
		iLog.debug("Locking all ...");
		while (iAllLocked != null)
			iAllLocked.awaitUninterruptibly();
		iAllLocked = iLock.newCondition();
		while (!iIndividualLocks.isEmpty()) {
			Condition otherCondition = iIndividualLocks.values().iterator().next();
			otherCondition.awaitUninterruptibly();
		}
		iLog.debug("Locked: all");
		return new UnlockAll();
	} finally {
		iLock.unlock();
	}
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
public Unlock lock(Collection<Long> ids) {
	iLock.lock();
	try {
		if (ids == null || ids.isEmpty()) return new Unlock(ids);
		iLog.debug("Locking " + ids + " ...");
		Condition otherCondition = null;
		while ((otherCondition = hasLock(ids)) != null)
			otherCondition.awaitUninterruptibly();
		Condition myCondition = iLock.newCondition();
		for (Long id: ids)
			iIndividualLocks.put(id, myCondition);
		iLog.debug("Locked: " + ids);
		return new Unlock(ids);
	} finally {
		iLock.unlock();
	}
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
public EventQueue() {
    for (int i = 0; i < NUM_PRIORITIES; i++) {
        queues[i] = new Queue();
    }
    /*
     * NOTE: if you ever have to start the associated event dispatch
     * thread at this point, be aware of the following problem:
     * If this EventQueue instance is created in
     * SunToolkit.createNewAppContext() the started dispatch thread
     * may call AppContext.getAppContext() before createNewAppContext()
     * completes thus causing mess in thread group to appcontext mapping.
     */

    appContext = AppContext.getAppContext();
    pushPopLock = (Lock)appContext.get(AppContext.EVENT_QUEUE_LOCK_KEY);
    pushPopCond = (Condition)appContext.get(AppContext.EVENT_QUEUE_COND_KEY);
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
/**
 * Constructor for AppContext.  This method is <i>not</i> public,
 * nor should it ever be used as such.  The proper way to construct
 * an AppContext is through the use of SunToolkit.createNewAppContext.
 * A ThreadGroup is created for the new AppContext, a Thread is
 * created within that ThreadGroup, and that Thread calls
 * SunToolkit.createNewAppContext before calling anything else.
 * That creates both the new AppContext and its EventQueue.
 *
 * @param   threadGroup     The ThreadGroup for the new AppContext
 * @see     sun.awt.SunToolkit
 * @since   1.2
 */
AppContext(ThreadGroup threadGroup) {
    numAppContexts.incrementAndGet();

    this.threadGroup = threadGroup;
    threadGroup2appContext.put(threadGroup, this);

    this.contextClassLoader =
         AccessController.doPrivileged(new PrivilegedAction<ClassLoader>() {
                public ClassLoader run() {
                    return Thread.currentThread().getContextClassLoader();
                }
            });

    // Initialize push/pop lock and its condition to be used by all the
    // EventQueues within this AppContext
    Lock eventQueuePushPopLock = new ReentrantLock();
    put(EVENT_QUEUE_LOCK_KEY, eventQueuePushPopLock);
    Condition eventQueuePushPopCond = eventQueuePushPopLock.newCondition();
    put(EVENT_QUEUE_COND_KEY, eventQueuePushPopCond);
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
void test(ScheduledThreadPoolExecutor p) throws Throwable {
    Runnable dummy = new Runnable() { public void run() {
        throw new AssertionError("shouldn't get here"); }};
    BlockingQueue q = p.getQueue();
    ReentrantLock lock = getField(q, "lock");
    Condition available = getField(q, "available");

    equal(0, p.getPoolSize());
    equal(0, p.getLargestPoolSize());
    equal(0L, p.getTaskCount());
    equal(0L, p.getCompletedTaskCount());
    p.schedule(dummy, 1L, HOURS);
    // Ensure one pool thread actually waits in timed queue poll
    awaitHasWaiters(lock, available, LONG_DELAY_MS);
    equal(1, p.getPoolSize());
    equal(1, p.getLargestPoolSize());
    equal(1L, p.getTaskCount());
    equal(0L, p.getCompletedTaskCount());
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
public void testAwait(boolean fair) {
    final PublicReentrantLock lock = new PublicReentrantLock(fair);
    final Condition c = lock.newCondition();
    final CountDownLatch locked = new CountDownLatch(1);
    Thread t = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            lock.lock();
            locked.countDown();
            c.await();
            lock.unlock();
        }});

    await(locked);
    lock.lock();
    assertHasWaiters(lock, c, t);
    c.signal();
    assertHasNoWaiters(lock, c);
    assertTrue(t.isAlive());
    lock.unlock();
    awaitTermination(t);
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
/**
 * Awaits condition "indefinitely" using the specified AwaitMethod.
 */
void await(Condition 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 timeoutNanos = MILLISECONDS.toNanos(timeoutMillis);
        long nanosRemaining = c.awaitNanos(timeoutNanos);
        assertTrue(nanosRemaining > timeoutNanos / 2);
        assertTrue(nanosRemaining <= timeoutNanos);
        break;
    case awaitUntil:
        assertTrue(c.awaitUntil(delayedDate(timeoutMillis)));
        break;
    default:
        throw new AssertionError();
    }
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
public void testGetWaitQueueLength(boolean fair) {
    final PublicReentrantReadWriteLock lock =
        new PublicReentrantReadWriteLock(fair);
    final Condition c = lock.writeLock().newCondition();
    final CountDownLatch locked = new CountDownLatch(1);
    Thread t = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            lock.writeLock().lock();
            assertEquals(0, lock.getWaitQueueLength(c));
            locked.countDown();
            c.await();
            lock.writeLock().unlock();
        }});

    await(locked);
    lock.writeLock().lock();
    assertHasWaiters(lock, c, t);
    assertEquals(1, lock.getWaitQueueLength(c));
    c.signal();
    assertHasNoWaiters(lock, c);
    assertEquals(0, lock.getWaitQueueLength(c));
    lock.writeLock().unlock();
    awaitTermination(t);
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
public void testAwait(boolean fair) {
    final PublicReentrantReadWriteLock lock =
        new PublicReentrantReadWriteLock(fair);
    final Condition c = lock.writeLock().newCondition();
    final CountDownLatch locked = new CountDownLatch(1);
    Thread t = newStartedThread(new CheckedRunnable() {
        public void realRun() throws InterruptedException {
            lock.writeLock().lock();
            locked.countDown();
            c.await();
            lock.writeLock().unlock();
        }});

    await(locked);
    lock.writeLock().lock();
    assertHasWaiters(lock, c, t);
    c.signal();
    assertHasNoWaiters(lock, c);
    assertTrue(t.isAlive());
    lock.writeLock().unlock();
    awaitTermination(t);
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
private static void signal(Lock lock, Condition condition, Runnable runnable) {
    boolean interrupted = Thread.interrupted();
    lock.lock();
    try {
        runnable.run();
        condition.signal();
    } finally {
        lock.unlock();
        if (interrupted) {
            Thread.currentThread().interrupt();
        }
    }
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
/**
 * Creates a new instance, which will begin at the specified epoch and efficiently accommodate waiting for epochs up
 * to (current epoch) + (horizon).
 *
 * @param horizon how far in the future an epoch can be waited for without sacrificing efficiency.
 * @param epoch the initial epoch; must be at least -1
 */
public ImminentEpochEvent(int horizon, long epoch) {
  if (epoch < -1) {
    throw new IndexOutOfBoundsException("Initial epoch cannot be less than -1");
  }

  _locks = new ReentrantLock[horizon];
  _conditions = new Condition[horizon];
  _epoch = new AtomicLong(epoch);

  for (int i = 0; i < _locks.length; i++) {
    _locks[i] = new ReentrantLock();
    _conditions[i] = _locks[i].newCondition();
  }
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
public void testHasWaitersIAE(boolean fair) {
    final ReentrantLock lock = new ReentrantLock(fair);
    final Condition c = lock.newCondition();
    final ReentrantLock lock2 = new ReentrantLock(fair);
    try {
        lock2.hasWaiters(c);
        shouldThrow();
    } catch (IllegalArgumentException success) {}
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
public void testAwaitNanos_Timeout(boolean fair) {
    final ReentrantLock lock = new ReentrantLock(fair);
    final Condition c = lock.newCondition();
    final long timeoutMillis = timeoutMillis();
    final long timeoutNanos = MILLISECONDS.toNanos(timeoutMillis);
    lock.lock();
    final long startTime = System.nanoTime();
    try {
        long nanosRemaining = c.awaitNanos(timeoutNanos);
        assertTrue(nanosRemaining <= 0);
    } catch (InterruptedException fail) { threadUnexpectedException(fail); }
    assertTrue(millisElapsedSince(startTime) >= timeoutMillis);
    lock.unlock();
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
public void testAwaitUntil_Timeout(boolean fair) {
    final ReentrantLock lock = new ReentrantLock(fair);
    final Condition c = lock.newCondition();
    lock.lock();
    // We shouldn't assume that nanoTime and currentTimeMillis
    // use the same time source, so don't use nanoTime here.
    final java.util.Date delayedDate = delayedDate(timeoutMillis());
    try {
        assertFalse(c.awaitUntil(delayedDate));
    } catch (InterruptedException fail) { threadUnexpectedException(fail); }
    assertTrue(new java.util.Date().getTime() >= delayedDate.getTime());
    lock.unlock();
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
private void startThrift() throws Exception {
  final Lock startLock = new ReentrantLock();
  final Condition startCondition = startLock.newCondition();
  final AtomicBoolean startedServing = new AtomicBoolean();
  try (ServerSocket socket = new ServerSocket(0)) {
    thriftPort = socket.getLocalPort();
  }
  conf.setVar(ConfVars.METASTOREURIS, getThriftConnectionUri());
  final HiveConf hiveConf = new HiveConf(conf, HiveMetaStoreClient.class);
  thriftServer.execute(new Runnable() {
    @Override
    public void run() {
      try {
        HadoopThriftAuthBridge bridge = new HadoopThriftAuthBridge23();
        HiveMetaStore.startMetaStore(thriftPort, bridge, hiveConf, startLock, startCondition, startedServing);
      } catch (Throwable e) {
        LOG.error("Unable to start a Thrift server for Hive Metastore", e);
      }
    }
  });
  int i = 0;
  while (i++ < 3) {
    startLock.lock();
    try {
      if (startCondition.await(1, TimeUnit.MINUTES)) {
        break;
      }
    } finally {
      startLock.unlock();
    }
    if (i == 3) {
      throw new RuntimeException("Maximum number of tries reached whilst waiting for Thrift server to be ready");
    }
  }
}
 

import java.util.concurrent.locks.Condition; //导入依赖的package包/类
static boolean hasWaiters(ReentrantLock lock, Condition condition) {
    lock.lock();
    try {
        return lock.hasWaiters(condition);
    } finally {
        lock.unlock();
    }
}