content: /*
   * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
   * (the "License"). You may not use this work except in compliance with the License, which is
   * available at www.apache.org/licenses/LICENSE-2.0
   *
   * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
   * either express or implied, as more fully set forth in the License.
   *
   * See the NOTICE file distributed with this work for information regarding copyright ownership.
   */
  
  package alluxio.master.file.meta;
  
  import alluxio.collections.LockPool;
  import alluxio.concurrent.LockMode;
  import alluxio.conf.Configuration;
  import alluxio.conf.PropertyKey;
  import alluxio.metrics.MetricKey;
  import alluxio.metrics.MetricsSystem;
  import alluxio.resource.LockResource;
  import alluxio.resource.RWLockResource;
  import alluxio.util.interfaces.Scoped;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.cache.CacheBuilder;
  import com.google.common.cache.CacheLoader;
  import com.google.common.cache.LoadingCache;
  import com.google.common.util.concurrent.Striped;
  
  import java.io.Closeable;
  import java.io.IOException;
  import java.util.Map.Entry;
  import java.util.Optional;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantReadWriteLock;
  
  /**
   * Class for managing inode locking. We manage locks centrally instead of embedded in the inode
   * tree. This allows us to create locks only as needed, and garbage collect locks that aren't in
   * use. As a result, we save memory when the inode tree contains many millions of files.
   */
  public class InodeLockManager implements Closeable {
    /**
     * Pool for supplying inode locks. To lock an inode, its inode id must be searched in this
     * pool to get the appropriate read lock.
     *
     * We use weak values so that when nothing holds a reference to
     * a lock, the garbage collector can remove the lock's entry from the pool.
     */
    private final LockPool<Long> mInodeLocks =
        new LockPool<>((key) -> new ReentrantReadWriteLock(),
            Configuration.getInt(PropertyKey.MASTER_LOCK_POOL_INITSIZE),
            Configuration.getInt(PropertyKey.MASTER_LOCK_POOL_LOW_WATERMARK),
            Configuration.getInt(PropertyKey.MASTER_LOCK_POOL_HIGH_WATERMARK),
            Configuration.getInt(PropertyKey.MASTER_LOCK_POOL_CONCURRENCY_LEVEL));
    /**
     * Cache for supplying edge locks, similar to mInodeLocks.
     */
    private final LockPool<Edge> mEdgeLocks =
        new LockPool<>((key) -> new ReentrantReadWriteLock(),
            Configuration.getInt(PropertyKey.MASTER_LOCK_POOL_INITSIZE),
            Configuration.getInt(PropertyKey.MASTER_LOCK_POOL_LOW_WATERMARK),
            Configuration.getInt(PropertyKey.MASTER_LOCK_POOL_HIGH_WATERMARK),
            Configuration.getInt(PropertyKey.MASTER_LOCK_POOL_CONCURRENCY_LEVEL));
  
    /**
     * Locks for guarding changes to last modified time and size on read-locked parent inodes.
     *
     * When renaming, creating, or deleting, we update the last modified time, last access time
     * and size of the parent inode while holding only a read lock. In the presence of concurrent
     * operations, this could cause the last modified time to decrease, or lead to incorrect
     * directory sizes. To avoid this, we guard the parent inode read-modify-write with this lock.
     * To avoid deadlock, a thread should never acquire more than one of these locks at the same time,
     * and no other locks should be taken while holding one of these locks.
     */
    private final Striped<Lock> mParentUpdateLocks = Striped.lock(1_000);
  
    /**
     * Cache for supplying inode persistence locks. Before a thread can persist an inode, it must
     * acquire the persisting lock for the inode. The cache maps inode ids to AtomicBooleans used to
     * provide mutual exclusion for inode persisting threads.
     */
    private final LoadingCache<Long, AtomicBoolean> mPersistingLocks =
        CacheBuilder.newBuilder()
            .weakValues()
            .initialCapacity(1_000)
            .concurrencyLevel(100)
            .build(new CacheLoader<Long, AtomicBoolean>() {
              @Override
              public AtomicBoolean load(Long key) {
                return new AtomicBoolean();
              }
            });
  
    /**
     * Creates a new instance of {@link InodeLockManager}.
     */
    public InodeLockManager() {
      MetricsSystem.registerGaugeIfAbsent(
          MetricKey.MASTER_INODE_LOCK_POOL_SIZE.getName(),
          () -> mInodeLocks.size());
      MetricsSystem.registerGaugeIfAbsent(
          MetricKey.MASTER_EDGE_LOCK_POOL_SIZE.getName(),
          () -> mEdgeLocks.size());
    }
  
    @VisibleForTesting
    boolean inodeReadLockedByCurrentThread(long inodeId) {
      return mInodeLocks.getRawReadWriteLock(inodeId).getReadHoldCount() > 0;
    }
  
    @VisibleForTesting
    boolean inodeWriteLockedByCurrentThread(long inodeId) {
      return mInodeLocks.getRawReadWriteLock(inodeId).getWriteHoldCount() > 0;
    }
  
    @VisibleForTesting
    boolean edgeReadLockedByCurrentThread(Edge edge) {
      return mEdgeLocks.getRawReadWriteLock(edge).getReadHoldCount() > 0;
    }
  
    @VisibleForTesting
    boolean edgeWriteLockedByCurrentThread(Edge edge) {
      return mEdgeLocks.getRawReadWriteLock(edge).getWriteHoldCount() > 0;
    }
  
    /**
     * Asserts that all locks have been released, throwing an exception if any locks are still taken.
     */
    @VisibleForTesting
    public void assertAllLocksReleased() {
      assertAllLocksReleased(mEdgeLocks);
      assertAllLocksReleased(mInodeLocks);
    }
  
    private <T> void assertAllLocksReleased(LockPool<T> pool) {
      for (Entry<T, ReentrantReadWriteLock> entry : pool.getEntryMap().entrySet()) {
        ReentrantReadWriteLock lock = entry.getValue();
        if (lock.isWriteLocked()) {
          throw new RuntimeException(
              String.format("Found a write-locked lock for %s", entry.getKey()));
        }
        if (lock.getReadLockCount() > 0) {
          throw new RuntimeException(
              String.format("Found a read-locked lock for %s", entry.getKey()));
        }
      }
    }
  
    /**
     * Acquires an inode lock.
     *
     * @param inode the inode to lock
     * @param mode the mode to lock in
     * @param useTryLock whether to acquire with {@link Lock#tryLock()} or {@link Lock#lock()}. This
     *                   method differs from {@link #tryLockInode(Long, LockMode)} because it will
     *                   block until the inode has been successfully locked.
     * @return a lock resource which must be closed to release the lock
     * @see #tryLockInode(Long, LockMode)
     */
    public RWLockResource lockInode(InodeView inode, LockMode mode, boolean useTryLock) {
      return mInodeLocks.get(inode.getId(), mode, useTryLock);
    }
  
    /**
     * Acquires an inode lock using {@link Lock#lock()}.
     *
     * @param inodeId the inode id of the inode to lock
     * @param mode the mode to lock in
     * @return a lock resource which must be closed to release the lock
     */
    public RWLockResource lockInode(Long inodeId, LockMode mode) {
      return mInodeLocks.get(inodeId, mode, false);
    }
  
    /**
     * Attempts to acquire an inode lock.
     *
     * @param inodeId the inode id to try locking
     * @param mode the mode to lock in
     * @return either an empty optional, or a lock resource which must be closed to release the lock
     */
    public Optional<RWLockResource> tryLockInode(Long inodeId, LockMode mode) {
      return mInodeLocks.tryGet(inodeId, mode);
    }
  
    /**
     * Acquires an edge lock.
     *
     * @param edge the edge to lock
     * @param mode the mode to lock in
     * @param useTryLock whether to acquire with {@link Lock#tryLock()} or {@link Lock#lock()}. This
     *                   method differs from {@link #tryLockEdge(Edge, LockMode)} because it will
     *                   block until the edge has been successfully locked.
     * @return a lock resource which must be closed to release the lock
     * @see #tryLockEdge(Edge, LockMode)
     */
    public RWLockResource lockEdge(Edge edge, LockMode mode, boolean useTryLock) {
      return mEdgeLocks.get(edge, mode, useTryLock);
    }
  
    /**
     * Attempts to acquire an edge lock.
     *
     * @param edge the edge to try locking
     * @param mode the mode to lock in
     * @return either an empty optional, or a lock resource which must be closed to release the lock
     */
    public Optional<RWLockResource> tryLockEdge(Edge edge, LockMode mode) {
      return mEdgeLocks.tryGet(edge, mode);
    }
  
    /**
     * Tries to acquire a lock for persisting the specified inode id.
     *
     * @param inodeId the inode to acquire the lock for
     * @return an optional wrapping a closure for releasing the lock on success, or Optional.empty if
     *         the lock is already taken
     */
    public Optional<Scoped> tryAcquirePersistingLock(long inodeId) {
      AtomicBoolean lock = mPersistingLocks.getUnchecked(inodeId);
      if (lock.compareAndSet(false, true)) {
        return Optional.of(() -> lock.set(false));
      }
      return Optional.empty();
    }
  
    /**
     * Acquires the lock for modifying an inode's last modified time or size. As a pre-requisite, the
     * current thread should already hold a read lock on the inode.
     *
     * @param inodeId the id of the inode to lock
     * @return a lock resource which must be closed to release the lock
     */
    public LockResource lockUpdate(long inodeId) {
      return new LockResource(mParentUpdateLocks.get(inodeId));
    }
  
    @Override
    public void close() throws IOException {
      mInodeLocks.close();
      mEdgeLocks.close();
    }
  }
  