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.AlluxioURI;
  import alluxio.collections.Pair;
  import alluxio.conf.Configuration;
  import alluxio.conf.PropertyKey;
  import alluxio.exception.InvalidPathException;
  import alluxio.master.file.meta.options.MountInfo;
  import alluxio.metrics.MetricKey;
  import alluxio.metrics.MetricsSystem;
  import alluxio.resource.CloseableResource;
  import alluxio.underfs.UnderFileSystem;
  import alluxio.util.ThreadFactoryUtils;
  import alluxio.util.io.PathUtils;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.cache.Cache;
  import com.google.common.cache.CacheBuilder;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.IOException;
  import java.time.Clock;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.List;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.ThreadPoolExecutor;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReadWriteLock;
  import java.util.concurrent.locks.ReentrantReadWriteLock;
  import javax.annotation.concurrent.ThreadSafe;
  
  /**
   * This is a {@link UfsAbsentPathCache} which implements asynchronous addition and removal to the
   * cache, since the processing of the path may be slow.
   */
  @ThreadSafe
  public class AsyncUfsAbsentPathCache implements UfsAbsentPathCache {
    private static final Logger LOG = LoggerFactory.getLogger(AsyncUfsAbsentPathCache.class);
    /** Number of seconds to keep threads alive. */
    private static final int THREAD_KEEP_ALIVE_SECONDS = 60;
    /** Number of paths to cache. */
    private static final int MAX_PATHS =
        Configuration.getInt(PropertyKey.MASTER_UFS_PATH_CACHE_CAPACITY);
  
    /** The mount table. */
    private final MountTable mMountTable;
    /** Paths currently being processed. This is used to prevent duplicate processing. */
    private final ConcurrentHashMap<String, PathLock> mCurrentPaths;
    /** Cache of paths which are absent in the ufs, maps an alluxio path to a Pair
     *  which is the sync time and the mount id.
     */
    private final Cache<String, Pair<Long, Long>> mCache;
    /** A thread pool for the async tasks. */
    private final ThreadPoolExecutor mPool;
  
    private final Clock mClock;
  
    /**
     * Creates a new instance of {@link AsyncUfsAbsentPathCache}.
     *
     * @param mountTable the mount table
     * @param numThreads the maximum number of threads for the async thread pool
     * @param clock the clock to use to compute the sync times
     */
    public AsyncUfsAbsentPathCache(MountTable mountTable, int numThreads, Clock clock) {
      mMountTable = mountTable;
      mClock = clock;
      mCurrentPaths = new ConcurrentHashMap<>(8, 0.95f, 8);
      mCache = CacheBuilder.newBuilder().maximumSize(MAX_PATHS).concurrencyLevel(Configuration.getInt(
          PropertyKey.MASTER_UFS_PATH_CACHE_THREADS)).recordStats().build();
      /* Number of threads for the async pool. */
  
      mPool = new ThreadPoolExecutor(numThreads, numThreads, THREAD_KEEP_ALIVE_SECONDS,
          TimeUnit.SECONDS, new LinkedBlockingQueue<>(),
          ThreadFactoryUtils.build("UFS-Absent-Path-Cache-%d", true));
      mPool.allowCoreThreadTimeOut(true);
      long timeout = getCachedGaugeTimeoutMillis();
      MetricsSystem.registerCachedGaugeIfAbsent(MetricKey.MASTER_ABSENT_CACHE_SIZE.getName(),
          mCache::size, timeout, TimeUnit.MILLISECONDS);
      MetricsSystem.registerCachedGaugeIfAbsent(MetricKey.MASTER_ABSENT_CACHE_MISSES.getName(),
          () -> mCache.stats().missCount(), timeout, TimeUnit.MILLISECONDS);
      MetricsSystem.registerCachedGaugeIfAbsent(MetricKey.MASTER_ABSENT_CACHE_HITS.getName(),
          () -> mCache.stats().hitCount(), timeout, TimeUnit.MILLISECONDS);
      MetricsSystem.registerCachedGaugeIfAbsent(
          MetricKey.MASTER_ABSENT_PATH_CACHE_QUEUE_SIZE.getName(),
          () -> mPool.getQueue().size(), timeout, TimeUnit.MILLISECONDS);
    }
  
    @VisibleForTesting
    protected long getCachedGaugeTimeoutMillis() {
      return 2000;
    }
  
    @Override
    public void processAsync(AlluxioURI path, List<Inode> prefixInodes) {
      mPool.submit(() -> processPathSync(path, prefixInodes));
    }
  
    @Override
    public void addSinglePath(AlluxioURI path) {
      MountInfo mountInfo = getMountInfo(path);
      if (mountInfo == null) {
        return;
      }
      addCacheEntry(path.getPath(), mountInfo);
    }
  
    @Override
    public void processExisting(AlluxioURI path) {
      MountInfo mountInfo = getMountInfo(path);
      if (mountInfo == null) {
        return;
      }
      // This is called when we create a persisted path in Alluxio. The path components need to be
      // invalidated so the cache does not incorrectly think a path is absent.
      // As an optimization, this method avoids holding locks, to prevent waiting on UFS. However,
      // since the locks are not being used in this code path, there could be a race between this
      // invalidating thread, and a processing of the path from the thread pool. To avoid the race,
      // this invalidating thread must set the intention to invalidate before invalidating.
      for (AlluxioURI alluxioUri : getNestedPaths(path, mountInfo.getAlluxioUri().getDepth())) {
        PathLock pathLock = mCurrentPaths.get(alluxioUri.getPath());
        if (pathLock != null) {
          pathLock.setInvalidate();
        }
        removeCacheEntry(alluxioUri.getPath());
      }
    }
  
    @Override
    public boolean isAbsentSince(AlluxioURI path, long absentSince) {
      MountInfo mountInfo = getMountInfo(path);
      if (mountInfo == null) {
        return false;
      }
      AlluxioURI mountBaseUri = mountInfo.getAlluxioUri();
  
      while (path != null && !path.equals(mountBaseUri)) {
        Pair<Long, Long> cacheResult = mCache.getIfPresent(path.getPath());
  
        if (cacheResult != null && cacheResult.getFirst() != null
            && cacheResult.getSecond() != null
            && cacheResult.getFirst() >= absentSince
            && cacheResult.getSecond() == mountInfo.getMountId()) {
          return true;
        }
        path = path.getParent();
      }
      // Reached the root, without finding anything in the cache.
      return false;
    }
  
    /**
     * Processes and checks the existence of the corresponding ufs path for the given Alluxio path.
     *
     * @param alluxioUri the Alluxio path to process
     * @param mountInfo the associated {@link MountInfo} for the Alluxio path
     * @return if true, further traversal of the descendant paths should continue
     */
    private boolean processSinglePath(AlluxioURI alluxioUri, MountInfo mountInfo) {
      PathLock pathLock = new PathLock();
      Lock writeLock = pathLock.writeLock();
      Lock readLock = null;
      try {
        // Write lock this path, to only enable a single task per path
        writeLock.lock();
        PathLock existingLock = mCurrentPaths.putIfAbsent(alluxioUri.getPath(), pathLock);
        if (existingLock != null) {
          // Another thread already locked this path and is processing it. Wait for the other
          // thread to finish, by locking the existing read lock.
          writeLock.unlock();
          writeLock = null;
          readLock = existingLock.readLock();
          readLock.lock();
  
          if (mCache.getIfPresent(alluxioUri.getPath()) != null) {
            // This path is already in the cache (is absent). Further traversal is unnecessary.
            return false;
          }
        } else {
          // This thread has the exclusive lock for this path.
  
          // Resolve this Alluxio uri. It should match the original mount id.
          MountTable.Resolution resolution = mMountTable.resolve(alluxioUri);
          if (resolution.getMountId() != mountInfo.getMountId()) {
            // This mount point has changed. Further traversal is unnecessary.
            return false;
          }
  
          boolean existsInUfs;
          try (CloseableResource<UnderFileSystem> ufsResource = resolution.acquireUfsResource()) {
            UnderFileSystem ufs = ufsResource.get();
            existsInUfs = ufs.exists(resolution.getUri().toString());
          }
          if (existsInUfs) {
            // This ufs path exists. Remove the cache entry.
            removeCacheEntry(alluxioUri.getPath());
          } else {
            // This is the first ufs path which does not exist. Add it to the cache.
            addCacheEntry(alluxioUri.getPath(), mountInfo);
  
            if (pathLock.isInvalidate()) {
              // This path was marked to be invalidated, meaning this UFS path was just created,
              // and now exists. Invalidate the entry.
              // This check is necessary to avoid the race with the invalidating thread.
              removeCacheEntry(alluxioUri.getPath());
            } else {
              // Further traversal is unnecessary.
              return false;
            }
          }
        }
      } catch (InvalidPathException | IOException e) {
        LOG.warn("Processing path failed: " + alluxioUri, e);
        return false;
      } finally {
        // Unlock the path
        if (readLock != null) {
          readLock.unlock();
        }
        if (writeLock != null) {
          mCurrentPaths.remove(alluxioUri.getPath(), pathLock);
          writeLock.unlock();
        }
      }
      return true;
    }
  
    /**
     * @param alluxioUri the Alluxio path to get the mount info for
     * @return the {@link MountInfo} of the given Alluxio path, or null if it doesn't exist
     */
    private MountInfo getMountInfo(AlluxioURI alluxioUri) {
      try {
        MountTable.Resolution resolution = mMountTable.resolve(alluxioUri);
        return mMountTable.getMountInfo(resolution.getMountId());
      } catch (Exception e) {
        // Catch Exception in case the mount point doesn't exist currently.
        LOG.warn("Failed to get mount info for path {}. message: {}", alluxioUri, e.toString());
        return null;
      }
    }
  
    /**
     * Returns a sequence of Alluxio paths for a specified path, starting from the path component at
     * a specific index, to the specified path.
     *
     * @param alluxioUri the Alluxio path to get the nested paths for
     * @param startComponentIndex the index to the starting path component,
     *        root directory has index 0
     * @return a list of nested paths from the starting component to the given path
     */
    private List<AlluxioURI> getNestedPaths(AlluxioURI alluxioUri, int startComponentIndex) {
      try {
        String[] fullComponents = PathUtils.getPathComponents(alluxioUri.getPath());
        String[] baseComponents = Arrays.copyOfRange(fullComponents, 0, startComponentIndex);
        AlluxioURI uri = new AlluxioURI(
            PathUtils.concatPath(AlluxioURI.SEPARATOR, baseComponents));
        List<AlluxioURI> components = new ArrayList<>(fullComponents.length - startComponentIndex);
        for (int i = startComponentIndex; i < fullComponents.length; i++) {
          uri = uri.joinUnsafe(fullComponents[i]);
          components.add(uri);
        }
        return components;
      } catch (InvalidPathException e) {
        return Collections.emptyList();
      }
    }
  
    /**
     * This represents a lock for a path component.
     */
    private static final class PathLock {
      private final ReadWriteLock mRwLock;
      private volatile boolean mInvalidate;
  
      private PathLock() {
        mRwLock = new ReentrantReadWriteLock();
        mInvalidate = false;
      }
  
      /**
       * @return the write lock
       */
      private Lock writeLock() {
        return mRwLock.writeLock();
      }
  
      /**
       * @return the read lock
       */
      private Lock readLock() {
        return mRwLock.readLock();
      }
  
      /**
       * Sets the intention to invalidate this path.
       */
      private void setInvalidate() {
        mInvalidate = true;
      }
  
      /**
       * @return true if the path was marked to be invalidated
       */
      private boolean isInvalidate() {
        return mInvalidate;
      }
    }
  
    private void addCacheEntry(String path, MountInfo mountInfo) {
      LOG.debug("Add cacheEntry={}", path);
      mCache.put(path, new Pair<>(mClock.millis(), mountInfo.getMountId()));
    }
  
    private void removeCacheEntry(String path) {
      LOG.debug("Remove cacheEntry={}", path);
      mCache.invalidate(path);
    }
  
    /**
     * Processes a path synchronously.
     *
     * @param path the path to add
     * @param prefixInodes the existing inodes for the path prefix
     */
    @VisibleForTesting
    void processPathSync(AlluxioURI path, List<Inode> prefixInodes) {
      MountInfo mountInfo = getMountInfo(path);
      if (mountInfo == null) {
        return;
      }
  
      // baseIndex should be the index of the first non-persisted inode under the mount point.
      int baseIndex = mountInfo.getAlluxioUri().getDepth();
      while (baseIndex < prefixInodes.size()) {
        if (prefixInodes.get(baseIndex).isPersisted()) {
          baseIndex++;
        } else {
          break;
        }
      }
  
      for (AlluxioURI alluxioUri : getNestedPaths(path, baseIndex)) {
        if (!processSinglePath(alluxioUri, mountInfo)) {
          break;
        }
      }
    }
  }
  