alluxio-master-2.x/core/server/master/src/main/java/alluxio/underfs/UfsStatusCache.java
/*
 * 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.underfs;

import alluxio.AlluxioURI;
import alluxio.collections.UnmodifiableArrayList;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.InvalidPathException;
import alluxio.master.file.DefaultFileSystemMaster;
import alluxio.master.file.RpcContext;
import alluxio.master.file.meta.MountTable;
import alluxio.master.file.meta.UfsAbsentPathCache;
import alluxio.resource.CloseableResource;
import alluxio.util.LogUtils;

import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Collection;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;

/**
 * This class is a cache from an Alluxio namespace URI ({@link AlluxioURI}, i.e. /path/to/inode) to
 * UFS statuses.
 *
 * It also allows associating a path with child inodes, so that the statuses for a specific path can
 * be searched for later.
 */
@ThreadSafe
public class UfsStatusCache {
  private static final Logger LOG = LoggerFactory.getLogger(UfsStatusCache.class);

  private final ConcurrentHashMap<AlluxioURI, UfsStatus> mStatuses;
  private final ConcurrentHashMap<AlluxioURI, Future<Collection<UfsStatus>>> mActivePrefetchJobs;
  private final ConcurrentHashMap<AlluxioURI, Collection<UfsStatus>> mChildren;
  private final UfsAbsentPathCache mAbsentCache;
  private final long mCacheValidTime;
  private final ExecutorService mPrefetchExecutor;
  private final long mUfsFetchTimeout;

  /**
   * Create a new instance of {@link UfsStatusCache}.
   *
   * @param prefetchExecutor the executor service used to prefetch statuses. If set to null, then
   *                         calls to {@link #prefetchChildren(AlluxioURI, MountTable)} will not
   *                         schedule any tasks.
   * @param absentPathCache the absent cache that ufsStatusCache should consult
   * @param cacheValidTime  the time when the absent cache entry would be considered valid
   */
  public UfsStatusCache(@Nullable ExecutorService prefetchExecutor,
      UfsAbsentPathCache absentPathCache, long cacheValidTime) {
    mStatuses = new ConcurrentHashMap<>();
    mChildren = new ConcurrentHashMap<>();
    mActivePrefetchJobs = new ConcurrentHashMap<>();
    mAbsentCache = absentPathCache;
    mCacheValidTime = cacheValidTime;
    mPrefetchExecutor = prefetchExecutor;
    mUfsFetchTimeout =
        Configuration.getMs(PropertyKey.MASTER_METADATA_SYNC_UFS_PREFETCH_TIMEOUT);
  }

  /**
   * Add a new status to the cache.
   *
   * The last component of the path in the {@link AlluxioURI} must match the result of
   * {@link UfsStatus#getName()}. This method overrides any status currently cached for the same
   * URI.
   *
   * @param path the Alluxio path to key on
   * @param status the ufs status to store
   * @return the previous status for the path if it existed, null otherwise
   * @throws IllegalArgumentException if the status name doesn't match the final URI path component
   */
  @Nullable
  public UfsStatus addStatus(AlluxioURI path, UfsStatus status) {
    if (!path.getName().equals(status.getName())) {
      throw new IllegalArgumentException(
          String.format("path name %s does not match ufs status name %s",
              path.getName(), status.getName()));
    }
    return addStatusUnchecked(path, status);
  }

  // unchecked: path and status must have the same name
  private UfsStatus addStatusUnchecked(AlluxioURI path, UfsStatus status) {
    mAbsentCache.processExisting(path);
    UfsStatus previousStatus = mStatuses.put(path, status);
    if (previousStatus == null) {
      // Update global counters for all InodeSyncStream
      DefaultFileSystemMaster.Metrics.UFS_STATUS_CACHE_SIZE_TOTAL.inc();
    }
    return previousStatus;
  }

  /**
   * Add a parent-child mapping to the status cache.
   *
   * All child statuses added via this method will be available via {@link #getStatus(AlluxioURI)}.
   *
   * @param path the directory inode path which contains the children
   * @param children the children of the {@code path}
   * @return the previous set of children if the mapping existed, null otherwise
   */
  @Nullable
  public Collection<UfsStatus> addChildren(AlluxioURI path, Collection<UfsStatus> children) {
    children.forEach(child -> {
      AlluxioURI childPath = path.joinUnsafe(child.getName());
      // childPath is derived from its UFS status, therefore has the same name
      addStatusUnchecked(childPath, child);
    });
    Collection<UfsStatus> previousStatuses = mChildren.put(path, children);
    // Update global counters for all InodeSyncStream
    int sizeChange = children.size() - (previousStatuses != null ? previousStatuses.size() : 0);
    DefaultFileSystemMaster.Metrics.UFS_STATUS_CACHE_CHILDREN_SIZE_TOTAL.inc(sizeChange);
    return previousStatuses;
  }

  /**
   * Remove a status from the cache.
   *
   * This will remove any references to child {@link UfsStatus}.
   *
   * @param path the path corresponding to the {@link UfsStatus} to remove
   * @return the removed UfsStatus
   */
  @Nullable
  public UfsStatus remove(AlluxioURI path) {
    Preconditions.checkNotNull(path, "can't remove null status cache path");
    UfsStatus removed = mStatuses.remove(path);
    if (removed != null) {
      // Update global counters for all InodeSyncStream
      DefaultFileSystemMaster.Metrics.UFS_STATUS_CACHE_SIZE_TOTAL.dec();
    }
    Collection<UfsStatus> removedChildren = mChildren.remove(path);
    if (removedChildren != null) {
      DefaultFileSystemMaster.Metrics.UFS_STATUS_CACHE_SIZE_TOTAL.dec(removedChildren.size());
      DefaultFileSystemMaster.Metrics
          .UFS_STATUS_CACHE_CHILDREN_SIZE_TOTAL.dec(removedChildren.size());
    }
    return removed;
  }

  private void checkAbsentCache(AlluxioURI path) throws FileNotFoundException {
    if (mAbsentCache.isAbsentSince(path, mCacheValidTime)) {
      throw new FileNotFoundException("UFS Status not found for path " + path.toString());
    }
  }

  /**
   * Get the UfsStatus from a given AlluxioURI.
   *
   * @param path the path the retrieve
   * @return The corresponding {@link UfsStatus} or {@code null} if there is none stored
   * @throws FileNotFoundException if the UFS does not contain the file
   */
  @Nullable
  public UfsStatus getStatus(AlluxioURI path) throws FileNotFoundException {
    checkAbsentCache(path);
    return mStatuses.get(path);
  }

  /**
   * Check if a status has been loaded in the cache for the given AlluxioURI.
   * Same as {@link UfsStatusCache#getStatus(AlluxioURI)} except does not
   * check the absent cache.
   *
   * @param path the path the retrieve
   * @return The corresponding {@link UfsStatus}
   */
  public Optional<UfsStatus> hasStatus(AlluxioURI path) {
    return Optional.ofNullable(mStatuses.get(path));
  }

  /**
   * Attempts to return a status from the cache. If it doesn't exist, reaches to the UFS for it.
   *
   * @param path the path the retrieve
   * @param mountTable the Alluxio mount table
   * @return The corresponding {@link UfsStatus} or {@code null} if there is none stored
   */
  @Nullable
  public UfsStatus fetchStatusIfAbsent(AlluxioURI path, MountTable mountTable)
      throws InvalidPathException {
    UfsStatus status;
    try {
      status = getStatus(path);
    } catch (FileNotFoundException e) {
      return null;
    }
    if (status != null) {
      return status;
    }
    MountTable.Resolution resolution = mountTable.resolve(path);
    AlluxioURI ufsUri = resolution.getUri();
    try (CloseableResource<UnderFileSystem> ufsResource = resolution.acquireUfsResource()) {
      UnderFileSystem ufs = ufsResource.get();
      UfsStatus ufsStatus = ufs.getStatus(ufsUri.toString());
      mountTable.getUfsSyncMetric(resolution.getMountId()).inc();
      if (ufsStatus == null) {
        mAbsentCache.addSinglePath(path);
        return null;
      }
      ufsStatus.setName(path.getName());
      addStatus(path, ufsStatus);
      return ufsStatus;
    } catch (FileNotFoundException e) {
      // If the ufs can not find the file, we explicitly mark it absent so we do not recheck it
      mAbsentCache.addSinglePath(path);
    } catch (IllegalArgumentException | IOException e) {
      LogUtils.warnWithException(LOG, "Failed to fetch status for {}", path, e);
    }
    return null;
  }

  /**
   * Fetches children of a given alluxio path, stores them in the cache, then returns them.
   *
   * Children can be returned in a few ways
   * 1. Children already exist in the internal index. We simply return them
   * 2. If children did not already exist in the index, then check if there was a scheduled
   * prefetch job running for this path. If so, wait for the job to finish and return the result.
   * 3. If no prefetch job, and children don't yet exist in the cache, then if the fallback
   * parameter is true, fetch them from the UFS and store them in the cache. Otherwise, simply
   * return null.
   *
   * @param rpcContext the rpcContext of the source of this call
   * @param path the Alluxio path to get the children of
   * @param mountTable the Alluxio mount table
   * @param useFallback whether or not to fall back to calling the UFS
   * @return child UFS statuses of the alluxio path, or null if no prefetch job and fallback
   *         specified as false
   * @throws InvalidPathException if the alluxio path can't be resolved to a UFS mount
   */
  @Nullable
  public Collection<UfsStatus> fetchChildrenIfAbsent(RpcContext rpcContext, AlluxioURI path,
       MountTable mountTable, boolean useFallback)
      throws InterruptedException, InvalidPathException {
    Future<Collection<UfsStatus>> prefetchJob = mActivePrefetchJobs.get(path);
    if (prefetchJob != null) {
      while (true) {
        try {
          Collection<UfsStatus> statuses = prefetchJob.get(
              mUfsFetchTimeout, TimeUnit.MILLISECONDS);
          if (statuses != null) {
            DefaultFileSystemMaster.Metrics.METADATA_SYNC_PREFETCH_PATHS.inc(statuses.size());
          }
          DefaultFileSystemMaster.Metrics.METADATA_SYNC_PREFETCH_SUCCESS.inc();
          return statuses;
        } catch (TimeoutException e) {
          if (rpcContext != null) {
            rpcContext.throwIfCancelled();
          }
          DefaultFileSystemMaster.Metrics.METADATA_SYNC_PREFETCH_RETRIES.inc();
        } catch (InterruptedException | ExecutionException e) {
          LogUtils.warnWithException(LOG,
              "Failed to get result for prefetch job on alluxio path {}", path, e);
          DefaultFileSystemMaster.Metrics.METADATA_SYNC_PREFETCH_FAIL.inc();
          if (e instanceof InterruptedException) {
            Thread.currentThread().interrupt();
            throw (InterruptedException) e;
          }
          break;
        } finally {
          mActivePrefetchJobs.remove(path);
        }
      }
    }
    Collection<UfsStatus> children = getChildren(path);
    if (children != null) {
      return children;
    }

    if (useFallback) {
      if (prefetchJob != null) {
        prefetchJob.cancel(false);
      }
      return getChildrenIfAbsent(path, mountTable);
    }
    return null;
  }

  /**
   * Fetches children of a given alluxio path stores them in the cache, then returns them.
   *
   * Will always return statuses from the UFS whether or not they exist in the cache, and whether
   * a prefetch job was scheduled or not.
   *
   * @param rpcContext the rpcContext of the source of this call
   * @param path the Alluxio path
   * @param mountTable the Alluxio mount table
   * @return child UFS statuses of the alluxio path
   * @throws InvalidPathException if the alluxio path can't be resolved to a UFS mount
   */
  @Nullable
  public Collection<UfsStatus> fetchChildrenIfAbsent(RpcContext rpcContext, AlluxioURI path,
       MountTable mountTable)
      throws InterruptedException, InvalidPathException {
    return fetchChildrenIfAbsent(rpcContext, path, mountTable, true);
  }

  /**
   * Retrieves the child UFS statuses for a given path and stores them in the cache.
   *
   * This method first checks if the children have already been retrieved, and if not, then
   * retrieves them.

   * @param path the path to get the children for
   * @param mountTable the Alluxio mount table
   * @return the child statuses that were stored in the cache, or null if the UFS couldn't list the
   *         statuses
   * @throws InvalidPathException when the table can't resolve the mount for the given URI
   */
  @Nullable
  Collection<UfsStatus> getChildrenIfAbsent(AlluxioURI path, MountTable mountTable)
      throws InvalidPathException {
    Collection<UfsStatus> children = getChildren(path);
    if (children != null) {
      return children;
    }
    if (mAbsentCache.isAbsentSince(path, mCacheValidTime)) {
      return null;
    }
    MountTable.Resolution resolution = mountTable.resolve(path);
    AlluxioURI ufsUri = resolution.getUri();
    try (CloseableResource<UnderFileSystem> ufsResource = resolution.acquireUfsResource()) {
      UnderFileSystem ufs = ufsResource.get();
      UfsStatus[] statuses = ufs.listStatus(ufsUri.toString());
      mountTable.getUfsSyncMetric(resolution.getMountId()).inc();
      if (statuses == null) {
        mAbsentCache.addSinglePath(path);
        return null;
      }
      children = new UnmodifiableArrayList<>(statuses);
      addChildren(path, children);
    } catch (IllegalArgumentException | IOException e) {
      LOG.debug("Failed to add status to cache {}", path, e);
    }
    return children;
  }

  /**
   * Get the child {@link UfsStatus}es from a given {@link AlluxioURI}.
   *
   * @param path the path the retrieve
   * @return The corresponding {@link UfsStatus} or {@code null} if there is none stored
   */
  @Nullable
  public Collection<UfsStatus> getChildren(AlluxioURI path) {
    return mChildren.get(path);
  }

  /**
   * Submit a request to asynchronously fetch the statuses corresponding to a given directory.
   *
   * Retrieve any fetched statuses by calling
   * {@link #fetchChildrenIfAbsent(RpcContext, AlluxioURI, MountTable)} with the same Alluxio path.
   *
   * If no {@link ExecutorService} was provided to this object before instantiation, this method is
   * a no-op.
   *
   * @param path the path to prefetch
   * @param mountTable the Alluxio mount table
   * @return the future corresponding to the fetch task
   */
  @Nullable
  public Future<Collection<UfsStatus>> prefetchChildren(AlluxioURI path, MountTable mountTable) {
    if (mPrefetchExecutor == null) {
      return null;
    }
    Future<Collection<UfsStatus>> prev = mActivePrefetchJobs.get(path);
    if (prev != null) {
      return prev;
    }
    try {
      Future<Collection<UfsStatus>> job =
          mPrefetchExecutor.submit(() -> getChildrenIfAbsent(path, mountTable));
      DefaultFileSystemMaster.Metrics.METADATA_SYNC_PREFETCH_OPS_COUNT.inc();
      prev = mActivePrefetchJobs.put(path, job);
      if (prev != null) {
        prev.cancel(true);
        DefaultFileSystemMaster.Metrics.METADATA_SYNC_PREFETCH_CANCEL.inc();
      }
      return job;
    } catch (RejectedExecutionException e) {
      LOG.debug("Failed to submit prefetch job for path {}", path, e);
      return null;
    }
  }

  /**
   * Interrupts and cancels any currently running prefetch jobs.
   */
  public void cancelAllPrefetch() {
    for (Future<?> f : mActivePrefetchJobs.values()) {
      f.cancel(false);
    }
    DefaultFileSystemMaster.Metrics.METADATA_SYNC_PREFETCH_CANCEL.inc(mActivePrefetchJobs.size());
    mActivePrefetchJobs.clear();
  }
}
