Java 类java.util.concurrent.locks.ReentrantLock 实例源码

/**
 * Save the state of this deque to a stream (that is, serialize it).
 *
 * @param s the stream
 * @serialData The capacity (int), followed by elements (each an
 * <tt>Object</tt>) in the proper order, followed by a null
 */
private void writeObject(java.io.ObjectOutputStream s) throws java.io.IOException {
    final ReentrantLock reentrantLock = this.lock;
    reentrantLock.lock();
    try {
        // Write out capacity and any hidden stuff
        s.defaultWriteObject();
        // Write out all elements in the proper order.
        for (Node<E> p = first; p != null; p = p.next)
            s.writeObject(p.item);
        // Use trailing null as sentinel
        s.writeObject(null);
    } finally {
        reentrantLock.unlock();
    }
}

public boolean removeFirstOccurrence(Object o) {
    if (o == null) {
        return false;
    }
    ReentrantLock lock = this.lock;
    lock.lock();
    try {
        for (Node<E> p = this.first; p != null; p = p.next) {
            if (o.equals(p.item)) {
                unlink(p);
                return true;
            }
        }
        lock.unlock();
        return false;
    } finally {
        lock.unlock();
    }
}

public E take() throws InterruptedException {
    int c = -1;
    AtomicInteger count = this.count;
    ReentrantLock takeLock = this.takeLock;
    takeLock.lockInterruptibly();
    E x;
    try {
        try {
            while (count.get() == 0)
                this.notEmpty.await();
        } catch (InterruptedException ie) {
            this.notEmpty.signal();
            throw ie;
        }

        x = extract();
        c = count.getAndDecrement();
        if (c > 1)
            this.notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    return x;
}

/**
 * Returns the approximate total number of tasks that have ever been
 * scheduled for execution. Because the states of tasks and
 * threads may change dynamically during computation, the returned
 * value is only an approximation.
 *
 * @return the number of tasks
 */
public long getTaskCount() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        long n = completedTaskCount;
        for (Worker w : workers) {
            n += w.completedTasks;
            if (w.isLocked())
                ++n;
        }
        return n + workQueue.size();
    } finally {
        mainLock.unlock();
    }
}

/**
 * Create a AppStateStorage
 * @param context master context
 * @param writeDir storage file directory
 * @param fs 
 */
public AppStateStorage(AMContext context, String writeDir, FileSystem fs){
  super("app-state-writter");
  LOG.info("writeDir=" + writeDir);
  this.context = context;
  this.writeDir = writeDir;
  this.writeDirPath = new Path(writeDir);
  this.fs = fs;

  splitFilePath = new Path(writeDirPath, splitFileName);
  matrixMetaLock = new ReentrantLock();
  taskMetaLock = new ReentrantLock();
  psMetaLock = new ReentrantLock();

  writeIntervalMS = context.getConf().getInt(
      AngelConf.ANGEL_AM_WRITE_STATE_INTERVAL_MS,
      AngelConf.DEFAULT_ANGEL_AM_WRITE_STATE_INTERVAL_MS);
  this.stopped = new AtomicBoolean(false);
}

/** Returns the element with the next timestamp, waiting until it is available.
    *
    * <p>Note that because of the reordering semantics, an invocation of this method
    * on a {@linkplain #isEmpty() nonempty} queue might block nonetheless.
    *
    * @return the element with the next timestamp.
    */
public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (a[start] == null) nextObjectReady.await();
        @SuppressWarnings("unchecked")
        final E x = (E)a[start];
        a[start] = null;
        start = start + 1 & mask;
        --count;
        timeStamp++;
        newSpaceAvailable.signalAll();
        return x;
    }
    finally {
        lock.unlock();
    }
}

/**
 * Identity-based version for use in Itr.remove
 */
void removeEQ(Object o) {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        Object[] array = queue;
        for (int i = 0, n = size; i < n; i++) {
            if (o == array[i]) {
                removeAt(i);
                break;
            }
        }
    } finally {
        lock.unlock();
    }
}

public void tReentrantLock() {
    System.currentTimeMillis();
    ReentrantLock lock = new ReentrantLock();

    long t1 = System.currentTimeMillis();
    for (int i = 0; i < 10000000; i++) {
        if (lock.tryLock()) {
            try {
                // ...
            } finally {
                lock.unlock();
            }
        }
    }
    long t2 = System.currentTimeMillis();

    System.out.println("take time:" + (t2 - t1) + " ms.");
}

public AsyncTraceDispatcher(Properties properties) {
    int queueSize = Integer.parseInt(properties.getProperty(OnsTraceConstants.AsyncBufferSize, "2048"));
    // queueSize 取大于或等于 value 的 2 的 n 次方数
    queueSize = 1 << (32 - Integer.numberOfLeadingZeros(queueSize - 1));
    this.queueSize = queueSize;
    this.entries = new Object[queueSize];
    this.indexMask = queueSize - 1;
    /**
     * 默认的消费者唤醒阈值，这个值需要让消费者能较持续的有事情做， 这个值设置过小，会导致生产者频繁唤起消费者；
     * 设置过大，可能导致生产者速度过快导致队列满丢日志的问题。
     */
    this.notifyThreshold = Integer.parseInt(properties.getProperty(OnsTraceConstants.WakeUpNum, "1"));
    this.putIndex = new AtomicLong(0L);
    this.discardCount = new AtomicLong(0L);
    this.takeIndex = new AtomicLong(0L);

    this.running = new AtomicBoolean(false);

    this.lock = new ReentrantLock(false);
    this.notEmpty = lock.newCondition();
}

@Test
public void testResourceIsInUseHasAnActiveApp() throws Exception {
  FileSystem fs = mock(FileSystem.class);
  CleanerMetrics metrics = mock(CleanerMetrics.class);
  SCMStore store = mock(SCMStore.class);

  FileStatus resource = mock(FileStatus.class);
  when(resource.getPath()).thenReturn(new Path(ROOT + "/a/b/c/abc"));
  // resource is stale
  when(store.isResourceEvictable(isA(String.class), isA(FileStatus.class)))
      .thenReturn(true);
  // but still has appIds
  when(store.removeResource(isA(String.class))).thenReturn(false);

  CleanerTask task =
      createSpiedTask(fs, store, metrics, new ReentrantLock());

  // process the resource
  task.processSingleResource(resource);

  // metrics should record a processed file (but not delete)
  verify(metrics).reportAFileProcess();
  verify(metrics, never()).reportAFileDelete();
}

public boolean contains(Object o) {
    if (o == null) {
        return false;
    }
    ReentrantLock lock = this.lock;
    lock.lock();
    try {
        for (Node<E> p = this.first; p != null; p = p.next) {
            if (o.equals(p.item)) {
                return true;
            }
        }
        lock.unlock();
        return false;
    } finally {
        lock.unlock();
    }
}

/**
 * Atomically removes all of the elements from this deque. The deque will be
 * empty after this call returns.
 */
public void clear() {
    final ReentrantLock reentrantLock = this.lock;
    reentrantLock.lock();
    try {
        for (Node<E> f = first; f != null; ) {
            f.item = null;
            Node<E> n = f.next;
            f.prev = null;
            f.next = null;
            f = n;
        }
        first = last = null;
        count = 0;
        notFull.signalAll();
    } finally {
        reentrantLock.unlock();
    }
}

public AbstractConnPool(
        final ConnFactory<T, C> connFactory,
        int defaultMaxPerRoute,
        int maxTotal) {
    super();
    if (connFactory == null) {
        throw new IllegalArgumentException("Connection factory may not null");
    }
    if (defaultMaxPerRoute <= 0) {
        throw new IllegalArgumentException("Max per route value may not be negative or zero");
    }
    if (maxTotal <= 0) {
        throw new IllegalArgumentException("Max total value may not be negative or zero");
    }
    this.lock = new ReentrantLock();
    this.connFactory = connFactory;
    this.routeToPool = new HashMap<T, RouteSpecificPool<T, C, E>>();
    this.leased = new HashSet<E>();
    this.available = new LinkedList<E>();
    this.pending = new LinkedList<PoolEntryFuture<E>>();
    this.maxPerRoute = new HashMap<T, Integer>();
    this.defaultMaxPerRoute = defaultMaxPerRoute;
    this.maxTotal = maxTotal;
}

/**
 * Atomically removes all of the elements from this deque.
 * The deque will be empty after this call returns.
 */
public void clear() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        for (Node<E> f = first; f != null; ) {
            f.item = null;
            Node<E> n = f.next;
            f.prev = null;
            f.next = null;
            f = n;
        }
        first = last = null;
        count = 0;
        notFull.signalAll();
    } finally {
        lock.unlock();
    }
}

public Object[] toArray() {
    ReentrantLock lock = this.lock;
    lock.lock();
    try {
        Object[] a = new Object[this.count];
        Node<E> p = this.first;
        int k = 0;
        while (p != null) {
            int k2 = k + 1;
            a[k] = p.item;
            p = p.next;
            k = k2;
        }
        return a;
    } finally {
        lock.unlock();
    }
}

/**
 * Creates an {@code ArrayBlockingQueue} with the given (fixed)
 * capacity, the specified access policy and initially containing the
 * elements of the given collection,
 * added in traversal order of the collection's iterator.
 *
 * @param capacity the capacity of this queue
 * @param fair if {@code true} then queue accesses for threads blocked
 *        on insertion or removal, are processed in FIFO order;
 *        if {@code false} the access order is unspecified.
 * @param c the collection of elements to initially contain
 * @throws IllegalArgumentException if {@code capacity} is less than
 *         {@code c.size()}, or less than 1.
 * @throws NullPointerException if the specified collection or any
 *         of its elements are null
 */
public ArrayBlockingQueue(int capacity, boolean fair,
                          Collection<? extends E> c) {
    this(capacity, fair);

    final ReentrantLock lock = this.lock;
    lock.lock(); // Lock only for visibility, not mutual exclusion
    try {
        final Object[] items = this.items;
        int i = 0;
        try {
            for (E e : c)
                items[i++] = Objects.requireNonNull(e);
        } catch (ArrayIndexOutOfBoundsException ex) {
            throw new IllegalArgumentException();
        }
        count = i;
        putIndex = (i == capacity) ? 0 : i;
    } finally {
        lock.unlock();
    }
}

public boolean offerLast(E e, long timeout, TimeUnit unit) throws InterruptedException {
    if (e == null) {
        throw new NullPointerException();
    }
    Node<E> node = new Node(e);
    long nanos = unit.toNanos(timeout);
    ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    while (!linkLast(node)) {
        try {
            if (nanos <= 0) {
                return false;
            }
            nanos = this.notFull.awaitNanos(nanos);
        } finally {
            lock.unlock();
        }
    }
    lock.unlock();
    return true;
}

public E poll() {
    final AtomicInteger count = this.count;
    if (count.get() == 0) return null;
    E x = null;
    int c = -1;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        if (count.get() > 0) {
            x = opQueue(null);
            c = count.getAndDecrement();
            if (c > 1) notEmpty.signal();
        }
    } finally {
        takeLock.unlock();
    }
    if (c == capacity) signalNotFull();
    return x;
}

public boolean removeLastOccurrence(Object o) {
    if (o == null) return false;
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        for (Node<E> p = last; p != null; p = p.prev) {
            if (o.equals(p.item)) {
                unlink(p);
                return true;
            }
        }
        return false;
    } finally {
        lock.unlock();
    }
}

/**
 * @throws NullPointerException {@inheritDoc}
 * @throws InterruptedException {@inheritDoc}
 */
public boolean offerFirst(E e, long timeout, TimeUnit unit)
        throws InterruptedException {
    if (e == null) throw new NullPointerException();
    Node<E> node = new Node<E>(e);
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (!linkFirst(node)) {
            if (nanos <= 0)
                return false;
            nanos = notFull.awaitNanos(nanos);
        }
        return true;
    } finally {
        lock.unlock();
    }
}

/**
 * Inserts the specified element at the tail of this queue if it is
 * possible to do so immediately without exceeding the queue's capacity,
 * returning {@code true} upon success and {@code false} if this queue
 * is full.
 * When using a capacity-restricted queue, this method is generally
 * preferable to method {@link BlockingQueue#add add}, which can fail to
 * insert an element only by throwing an exception.
 *
 * @throws NullPointerException if the specified element is null
 */
public boolean offer(E e) {
    if (e == null) throw new NullPointerException();
    final AtomicInteger count = this.count;
    if (count.get() == capacity)
        return false;
    int c = -1;
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        if (count.get() < capacity) {
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        }
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        signalNotEmpty();
    return c >= 0;
}

/**
 * Enters this monitor when the guard is satisfied. Blocks indefinitely, but may be interrupted.
 *
 * @throws InterruptedException if interrupted while waiting
 */
public void enterWhen(Guard guard) throws InterruptedException {
  if (guard.monitor != this) {
    throw new IllegalMonitorStateException();
  }
  final ReentrantLock lock = this.lock;
  boolean signalBeforeWaiting = lock.isHeldByCurrentThread();
  lock.lockInterruptibly();

  boolean satisfied = false;
  try {
    if (!guard.isSatisfied()) {
      await(guard, signalBeforeWaiting);
    }
    satisfied = true;
  } finally {
    if (!satisfied) {
      leave();
    }
  }
}

/**
 * Creates an {@code ArrayBlockingQueue} with the given (fixed)
 * capacity, the specified access policy and initially containing the
 * elements of the given collection,
 * added in traversal order of the collection's iterator.
 *
 * @param capacity the capacity of this queue
 * @param fair if {@code true} then queue accesses for threads blocked
 *        on insertion or removal, are processed in FIFO order;
 *        if {@code false} the access order is unspecified.
 * @param c the collection of elements to initially contain
 * @throws IllegalArgumentException if {@code capacity} is less than
 *         {@code c.size()}, or less than 1.
 * @throws NullPointerException if the specified collection or any
 *         of its elements are null
 */
public ArrayBlockingQueue(int capacity, boolean fair,
                          Collection<? extends E> c) {
    this(capacity, fair);

    final ReentrantLock lock = this.lock;
    lock.lock(); // Lock only for visibility, not mutual exclusion
    try {
        int i = 0;
        try {
            for (E e : c) {
                checkNotNull(e);
                items[i++] = e;
            }
        } catch (ArrayIndexOutOfBoundsException ex) {
            throw new IllegalArgumentException();
        }
        count = i;
        putIndex = (i == capacity) ? 0 : i;
    } finally {
        lock.unlock();
    }
}

/**
 * Atomically removes all of the elements from this deque.
 * The deque will be empty after this call returns.
 */
@Override
public void clear() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        for (Node<E> f = first; f != null; ) {
            f.item = null;
            Node<E> n = f.next;
            f.prev = null;
            f.next = null;
            f = n;
        }
        first = last = null;
        count = 0;
        notFull.signalAll();
    } finally {
        lock.unlock();
    }
}

public void reloadMetaData(ServerConfig conf) {
    ProxyMetaManager tmpManager = tmManager;
    for (; ; ) {
        if (tmpManager.getMetaCount() > 0) {
            continue;
        }
        ReentrantLock lock = tmpManager.getMetaLock();
        lock.lock();
        try {
            if (tmpManager.getMetaCount() > 0) {
                continue;
            }
            ProxyMetaManager newManager = new ProxyMetaManager();
            newManager.initMeta(conf);
            tmManager = newManager;
            tmpManager.terminate();
            break;
        } finally {
            lock.unlock();
        }
    }
}

public boolean offer(Runnable x) {
    if (x == null)
        throw new NullPointerException();
    RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        int i = size;
        if (i >= queue.length)
            grow();
        size = i + 1;
        if (i == 0) {
            queue[0] = e;
            setIndex(e, 0);
        } else {
            siftUp(i, e);
        }
        if (queue[0] == e) {
            leader = null;
            available.signal();
        }
    } finally {
        lock.unlock();
    }
    return true;
}

public E pollFirst(long timeout, TimeUnit unit)
    throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        E x;
        while ( (x = unlinkFirst()) == null) {
            if (nanos <= 0)
                return null;
            nanos = notEmpty.awaitNanos(nanos);
        }
        return x;
    } finally {
        lock.unlock();
    }
}

/**
 * Identity-based version for use in Itr.remove.
 */
void removeEQ(Object o) {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        Object[] array = queue;
        for (int i = 0, n = size; i < n; i++) {
            if (o == array[i]) {
                removeAt(i);
                break;
            }
        }
    } finally {
        lock.unlock();
    }
}

/**
 * Returns a rethrowable exception for this task, if available.
 * To provide accurate stack traces, if the exception was not
 * thrown by the current thread, we try to create a new exception
 * of the same type as the one thrown, but with the recorded
 * exception as its cause. If there is no such constructor, we
 * instead try to use a no-arg constructor, followed by initCause,
 * to the same effect. If none of these apply, or any fail due to
 * other exceptions, we return the recorded exception, which is
 * still correct, although it may contain a misleading stack
 * trace.
 *
 * @return the exception, or null if none
 */
private Throwable getThrowableException() {
    int h = System.identityHashCode(this);
    ExceptionNode e;
    final ReentrantLock lock = exceptionTableLock;
    lock.lock();
    try {
        expungeStaleExceptions();
        ExceptionNode[] t = exceptionTable;
        e = t[h & (t.length - 1)];
        while (e != null && e.get() != this)
            e = e.next;
    } finally {
        lock.unlock();
    }
    Throwable ex;
    if (e == null || (ex = e.ex) == null)
        return null;
    if (e.thrower != Thread.currentThread().getId()) {
        try {
            Constructor<?> noArgCtor = null;
            // public ctors only
            for (Constructor<?> c : ex.getClass().getConstructors()) {
                Class<?>[] ps = c.getParameterTypes();
                if (ps.length == 0)
                    noArgCtor = c;
                else if (ps.length == 1 && ps[0] == Throwable.class)
                    return (Throwable)c.newInstance(ex);
            }
            if (noArgCtor != null) {
                Throwable wx = (Throwable)noArgCtor.newInstance();
                wx.initCause(ex);
                return wx;
            }
        } catch (Exception ignore) {
        }
    }
    return ex;
}

/**
 * If there is a security manager, makes sure caller has
 * permission to shut down threads in general (see shutdownPerm).
 * If this passes, additionally makes sure the caller is allowed
 * to interrupt each worker thread. This might not be true even if
 * first check passed, if the SecurityManager treats some threads
 * specially.
 */
private void checkShutdownAccess() {
    SecurityManager security = System.getSecurityManager();
    if (security != null) {
        security.checkPermission(shutdownPerm);
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers)
                security.checkAccess(w.thread);
        } finally {
            mainLock.unlock();
        }
    }
}

public boolean tryAdvance(Consumer<? super E> action) {
    if (action == null) throw new NullPointerException();
    final LinkedBlockingDeque<E> q = this.queue;
    final ReentrantLock lock = q.lock;
    if (!exhausted) {
        E e = null;
        lock.lock();
        try {
            if (current == null)
                current = q.first;
            while (current != null) {
                e = current.item;
                current = current.next;
                if (e != null)
                    break;
            }
        } finally {
            lock.unlock();
        }
        if (current == null)
            exhausted = true;
        if (e != null) {
            action.accept(e);
            return true;
        }
    }
    return false;
}

public E peekLast() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return (last == null) ? null : last.item;
    } finally {
        lock.unlock();
    }
}

public void sort(Comparator<? super E> c) {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        Object[] elements = getArray();
        Object[] newElements = Arrays.copyOf(elements, elements.length);
        @SuppressWarnings("unchecked") E[] es = (E[])newElements;
        Arrays.sort(es, c);
        setArray(newElements);
    } finally {
        lock.unlock();
    }
}

public RunnableScheduledFuture<?> peek() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return queue[0];
    } finally {
        lock.unlock();
    }
}

/**
 * Inserts all of the elements in the specified collection into this
 * list, starting at the specified position.  Shifts the element
 * currently at that position (if any) and any subsequent elements to
 * the right (increases their indices).  The new elements will appear
 * in this list in the order that they are returned by the
 * specified collection's iterator.
 *
 * @param index index at which to insert the first element
 *        from the specified collection
 * @param c collection containing elements to be added to this list
 * @return {@code true} if this list changed as a result of the call
 * @throws IndexOutOfBoundsException {@inheritDoc}
 * @throws NullPointerException if the specified collection is null
 * @see #add(int,Object)
 */
public boolean addAll(int index, Collection<? extends E> c) {
    Object[] cs = c.toArray();
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        Object[] elements = getArray();
        int len = elements.length;
        if (index > len || index < 0)
            throw new IndexOutOfBoundsException("Index: "+index+
                                                ", Size: "+len);
        if (cs.length == 0)
            return false;
        int numMoved = len - index;
        Object[] newElements;
        if (numMoved == 0)
            newElements = Arrays.copyOf(elements, len + cs.length);
        else {
            newElements = new Object[len + cs.length];
            System.arraycopy(elements, 0, newElements, 0, index);
            System.arraycopy(elements, index,
                             newElements, index + cs.length,
                             numMoved);
        }
        System.arraycopy(cs, 0, newElements, index, cs.length);
        setArray(newElements);
        return true;
    } finally {
        lock.unlock();
    }
}

@Override
public boolean remove(Object key, Object value)
{
  boolean removed = false;

  final ReentrantLock writeLock = this._writeLock;

  writeLock.lock();

  try
  {
    ConcurrentEntry<K,V>[] entries = _entries;

    int removeIndex = _getEntryIndex(entries, key);

    if (removeIndex >= 0)
    {
      ConcurrentEntry<K,V> entry = entries[removeIndex];

      V entryValue = entry.getValue();

      boolean valuesEqual = (entryValue != null) ? entryValue.equals(value) : (value == null);

      if (valuesEqual)
      {
        _entries = _removeEntryByIndex(entries, removeIndex);
        removed = true;
      }
    }
  }
  finally
  {
    writeLock.unlock();
  }

  return removed;
}

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        if (count == 0) {
            return null;
        }
        return extract();
    } finally {
        lock.unlock();
    }
}

public int size() {
    final ReentrantLock lock = l.lock;
    lock.lock();
    try {
        checkForComodification();
        return size;
    } finally {
        lock.unlock();
    }
}

public E pollLast() {
    ReentrantLock lock = this.lock;
    lock.lock();
    try {
        E unlinkLast = unlinkLast();
        return unlinkLast;
    } finally {
        lock.unlock();
    }
}

@Test
public void testClose() {
  ReentrantLock lock = new ReentrantLock();
  assertFalse(lock.isHeldByCurrentThread());
  AutoLock al = new AutoLock(lock);
  assertFalse(lock.isHeldByCurrentThread());
  al.autoLock();
  assertTrue(lock.isHeldByCurrentThread());
  al.close();
  assertFalse(lock.isHeldByCurrentThread());
}