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.concurrent.LockMode;
  import alluxio.resource.LockResource;
  import alluxio.resource.RWLockResource;
  
  import com.google.common.base.Preconditions;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.util.ArrayList;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.concurrent.locks.Lock;
  import java.util.stream.Collectors;
  import javax.annotation.concurrent.NotThreadSafe;
  
  /**
   * A simple inode lock list.
   */
  @NotThreadSafe
  public class SimpleInodeLockList implements InodeLockList {
    private static final Logger LOG = LoggerFactory.getLogger(SimpleInodeLockList.class);
  
    /**
     * Default value for {@link #mFirstWriteLockIndex} when there is no write lock.
     */
    private static final int NO_WRITE_LOCK_INDEX = -1;
    private static final Edge ROOT_EDGE = new Edge(-1, "");
  
    private final InodeLockManager mInodeLockManager;
  
    /**
     * Inode list.
     */
    private LinkedList<Inode> mInodes;
    /**
     * If last lock in mLocks is an edge lock, then this is the edge.
     * Otherwise, null.
     */
    private Edge mLastEdge;
    /**
     * Lock list.
     * The locks always alternate between Inode lock and Edge lock.
     * The first lock can be either Inode or Edge lock.
     */
    private LinkedList<RWLockResource> mLocks;
    /**
     * The index of the first write lock entry in {@link #mLocks}.
     * If all locks are read locks, mFirstWriteLockIndex == NO_WRITE_LOCK.
     * Otherwise, all locks before this index are read locks, and all
     * locks after and including this index are write locks.
     */
    private int mFirstWriteLockIndex;
  
    /** Whether to use {@link Lock#tryLock()} or {@link Lock#lock()}. */
    private final boolean mUseTryLock;
  
    /**
     * Creates a new empty lock list.
     *
     * @param inodeLockManager manager for inode locks
     * @param useTryLock whether or not use {@link Lock#tryLock()} or {@link Lock#lock()}
     */
    public SimpleInodeLockList(InodeLockManager inodeLockManager, boolean useTryLock) {
      mInodeLockManager = inodeLockManager;
      mInodes = new LinkedList<>();
      mLocks = new LinkedList<>();
      mFirstWriteLockIndex = NO_WRITE_LOCK_INDEX;
      mUseTryLock = useTryLock;
    }
  
    @Override
    public void lockInode(Inode inode, LockMode mode) {
      mode = nextLockMode(mode);
      if (!mLocks.isEmpty()) {
        Preconditions.checkState(!endsInInode(),
            "Cannot lock inode %s for lock list %s because the lock list already ends in an inode",
            inode.getId(), this);
        checkInodeNameAndEdgeNameMatch(inode);
      }
      lockAndAddInode(inode, mode);
    }
  
    /**
     * Checks if the inode name and the edge name match.
     * @param inode the inode to check
     */
    private void checkInodeNameAndEdgeNameMatch(Inode inode) throws IllegalStateException {
      if (!inode.getName().equals(mLastEdge.getName())) {
        StringBuilder sb = new StringBuilder();
        for (InodeView currentInode : mInodes) {
          sb.append("[");
          sb.append(currentInode.toProto());
          sb.append("]->");
        }
        sb.append("[END]");
        throw new IllegalStateException(
            String.format(
                "Expected to lock inode %s but locked inode name %s, id: %s, parent_id: %s. %n",
                mLastEdge.getName(), inode.getName(), inode.getId(), inode.getParentId())
                + "Locked inode path: " + sb);
      }
    }
  
    @Override
    public void lockEdge(Inode lastInode, String childName, LockMode mode) {
      mode = nextLockMode(mode);
      long edgeParentId = lastInode.getId();
      Edge edge = new Edge(lastInode.getId(), childName);
      if (!mLocks.isEmpty()) {
        Preconditions.checkState(endsInInode(),
            "Cannot lock edge %s when lock list %s already ends in an edge", edge, this);
        Preconditions.checkState(lastInode().getId() == edgeParentId,
            "Cannot lock edge %s when the last inode id in %s is %s", edge, this, lastInode.getId());
      }
      lockAndAddEdge(edge, mode);
    }
  
    @Override
    public void lockRootEdge(LockMode mode) {
      Preconditions.checkState(mLocks.isEmpty(),
          "Cannot lock root edge when lock list %s is nonempty", this);
      lockAndAddEdge(ROOT_EDGE, mode);
    }
  
    @Override
    public void pushWriteLockedEdge(Inode inode, String childName) {
      Edge edge = new Edge(inode.getId(), childName);
      Preconditions.checkState(!endsInInode(),
          "Cannot push edge write lock to edge %s; lock list %s ends in an inode", edge, this);
      Preconditions.checkState(endsInWriteLock(),
          "Cannot push write lock; lock list %s ends in a read lock");
  
      if (endsInMultipleWriteLocks()) {
        // If the lock before the edge lock is already WRITE, we can just acquire more WRITE locks.
        lockInode(inode, LockMode.WRITE);
        lockEdge(inode, childName, LockMode.WRITE);
      } else {
        Edge lastEdge = lastEdge();
        RWLockResource lastEdgeReadLock = mInodeLockManager.lockEdge(lastEdge, LockMode.READ,
            mUseTryLock);
        RWLockResource inodeLock = mInodeLockManager.lockInode(inode, LockMode.READ, mUseTryLock);
        RWLockResource nextEdgeLock = mInodeLockManager.lockEdge(edge, LockMode.WRITE, mUseTryLock);
        removeLastLock(); // Remove edge write lock
        addEdgeLock(lastEdge, LockMode.READ, lastEdgeReadLock);
        addInodeLock(inode, LockMode.READ, inodeLock);
        addEdgeLock(edge, LockMode.WRITE, nextEdgeLock);
      }
    }
  
    @Override
    public void unlockLastInode() {
      Preconditions.checkState(endsInInode(),
          "Cannot unlock last inode when the lock list %s ends in an edge", this);
      Preconditions.checkState(!mLocks.isEmpty(),
          "Cannot unlock last inode when the lock list is empty");
      removeLastLock();
    }
  
    @Override
    public void unlockLastEdge() {
      Preconditions.checkState(!endsInInode(),
          "Cannot unlock last edge when the lock list %s ends in an inode", this);
      Preconditions.checkState(!mLocks.isEmpty(),
          "Cannot unlock last edge when the lock list is empty");
      removeLastLock();
    }
  
    @Override
    public void downgradeToReadLocks() {
      mLocks.forEach(RWLockResource::downgrade);
      mFirstWriteLockIndex = NO_WRITE_LOCK_INDEX;
    }
  
    @Override
    public void downgradeLastEdge() {
      Preconditions.checkState(!endsInInode(),
          "Cannot downgrade last edge when lock list %s ends in an inode", this);
      Preconditions.checkState(!mLocks.isEmpty(),
          "Cannot downgrade last edge when the lock list is empty");
      Preconditions.checkState(endsInWriteLock(),
          "Cannot downgrade last edge when lock list %s is not write locked", this);
  
      if (!endsInMultipleWriteLocks()) {
        Edge lastEdge = lastEdge();
        RWLockResource newLock = mInodeLockManager.lockEdge(lastEdge, LockMode.READ, mUseTryLock);
        removeLastLock();
        addEdgeLock(lastEdge, LockMode.READ, newLock);
      }
    }
  
    /**
     * If mode is read but the lock list is write locked, returns LockMode.WRITE.
     *
     * This helps us preserve the invariant that there is never a READ lock following a WRITE lock.
     *
     * @param mode a lock mode
     * @return the mode
     */
    private LockMode nextLockMode(LockMode mode) {
      return endsInWriteLock() ? LockMode.WRITE : mode;
    }
  
    private void addLock(RWLockResource lock, LockMode mode) {
      if (!endsInWriteLock() && mode == LockMode.WRITE) {
        mFirstWriteLockIndex = mLocks.size();
      }
      try {
        mLocks.add(lock);
      } catch (Error e) {
        // If adding to mLocks fails due to OOM, this lock
        // will not be tracked so we must close it manually
        lock.close();
        throw e;
      }
    }
  
    private void addInodeLock(Inode inode, LockMode mode, RWLockResource lock) {
      mInodes.add(inode);
      mLastEdge = null;
      addLock(lock, mode);
    }
  
    private void lockAndAddInode(Inode inode, LockMode mode) {
      addInodeLock(inode, mode, mInodeLockManager.lockInode(inode, mode, mUseTryLock));
    }
  
    private void addEdgeLock(Edge edge, LockMode mode, RWLockResource lock) {
      mLastEdge = edge;
      addLock(lock, mode);
    }
  
    private void lockAndAddEdge(Edge edge, LockMode mode) {
      addEdgeLock(edge, mode, mInodeLockManager.lockEdge(edge, mode, mUseTryLock));
    }
  
    /**
     * Removes and unlocks the last lock.
     */
    private void removeLastLock() {
      mLocks.removeLast().close();
      if (mFirstWriteLockIndex >= mLocks.size()) {
        mFirstWriteLockIndex = NO_WRITE_LOCK_INDEX;
      }
      if (mLastEdge != null) {
        mLastEdge = null;
      } else {
        Inode last = mInodes.removeLast();
        if (!mLocks.isEmpty()) {
          if (mInodes.isEmpty()) {
            mLastEdge = ROOT_EDGE;
          } else {
            mLastEdge = new Edge(mInodes.getLast().getId(), last.getName());
          }
        }
      }
    }
  
    @Override
    public LockMode getLockMode() {
      return endsInWriteLock() ? LockMode.WRITE : LockMode.READ;
    }
  
    @Override
    public List<Inode> getLockedInodes() {
      return new ArrayList<>(mInodes);
    }
  
    @Override
    public Inode get(int index) {
      return mInodes.get(index);
    }
  
    @Override
    public int numInodes() {
      return mInodes.size();
    }
  
    @Override
    public boolean isEmpty() {
      return mLocks.isEmpty();
    }
  
    @Override
    public InodeLockManager getInodeLockManager() {
      return mInodeLockManager;
    }
  
    @Override
    public boolean endsInInode() {
      return mLastEdge == null;
    }
  
    /**
     * @return the last inode
     */
    private Inode lastInode() {
      Preconditions.checkState(endsInInode(),
          "Cannot get last inode for lock list %s which does not end in an inode", this);
      return mInodes.getLast();
    }
  
    /**
     * @return the last edge
     */
    private Edge lastEdge() {
      Preconditions.checkState(!endsInInode(),
          "Cannot get last edge for lock list %s which does not end in an edge", this);
      return mLastEdge;
    }
  
    /**
     * @return whether this lock list ends in a write lock
     */
    private boolean endsInWriteLock() {
      return mFirstWriteLockIndex != NO_WRITE_LOCK_INDEX;
    }
  
    private boolean endsInMultipleWriteLocks() {
      return mFirstWriteLockIndex != NO_WRITE_LOCK_INDEX && mFirstWriteLockIndex < mLocks.size() - 1;
    }
  
    @Override
    public String toString() {
      String path = mInodes.stream()
          .map(Inode::getName)
          .collect(Collectors.joining("/"));
      StringBuilder sb = new StringBuilder("Path: " + path);
      if (mLastEdge != null) {
        sb.append(String.format(", Last edge -> %s", mLastEdge));
      }
      sb.append(String.format(", Index of first write lock: %d", mFirstWriteLockIndex));
      return sb.toString();
    }
  
    @Override
    public void close() {
      mInodes.clear();
      mLocks.forEach(LockResource::close);
      mLocks.clear();
    }
  }
  