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 static alluxio.util.io.PathUtils.cleanPath;
  
  import alluxio.AlluxioURI;
  import alluxio.collections.Pair;
  import alluxio.conf.Configuration;
  import alluxio.conf.PropertyKey;
  import alluxio.exception.InvalidPathException;
  import alluxio.file.options.DescendantType;
  import alluxio.resource.LockResource;
  import alluxio.util.io.PathUtils;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.base.Preconditions;
  import com.google.common.cache.Cache;
  import com.google.common.cache.CacheBuilder;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.time.Clock;
  import java.util.Optional;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantLock;
  import java.util.function.BiConsumer;
  import javax.annotation.Nullable;
  import javax.annotation.concurrent.ThreadSafe;
  
  /**
   * A cache of paths that have been synchronized or have been marked as not being synchronized with
   * the UFS. Paths are mapped to time values that indicate their last synchronization or
   * invalidation.
   *
   * Synchronization times are updated only at the root path for which a sync was performed by
   * {@link alluxio.master.file.InodeSyncStream}.
   * Each path contains the following values concerning synchronization:
   * - recursiveSyncTime: the last time a sync was successful on this path
   * - directChildSyncTime: the last time a sync was successful on the path which (at least)
   * included itself and its direct children
   * - recursiveSyncTime: the last time a sync was successful on this path which
   * included itself and all its children
   * Each path additionally contains the following values considering invalidation:
   * - invalidationTime: the last time this exact path was invalidated
   * - directChildInvalidation: the last time a direct child of this path was invalidated
   * - recursiveChildInvalidation: the last time a non-direct child of this path was invalidated
   * - isFile: when checking if a file needs to be synced, any descendant type from the most
   * recent sync will be valid
   *
   * Whenever an invalidation is received the path, and its parents up to the root have their
   * appropriate invalidation times updated. Validation times are updated on the path
   * when {@link #notifySyncedPath} is called on the root sync path after a successful sync.
   * An invalidation is received either when a client calls
   * {@link alluxio.master.file.DefaultFileSystemMaster#needsSync} to notify that a path
   * needs synchronization, or when a file is updated by an external Alluxio cluster, and
   * cross cluster sync is enabled.
   *
   * Checking if a path needs to be synchronized involves checking the appropriate validation
   * and invalidation times at each path component up to the root.
   */
  @ThreadSafe
  public class UfsSyncPathCache {
    private static final Logger LOG = LoggerFactory.getLogger(UfsSyncPathCache.class);
  
    public final Cache<String, SyncState> mItems;
    private final Clock mClock;
    // we keep the root as a separate value as we never want to evict it,
    // and instead we just update it under a lock
    private final SyncState mRoot = new SyncState(false);
    private final Lock mRootLock = new ReentrantLock();
  
    /**
     * Creates a new instance of {@link UfsSyncPathCache}.
     * @param clock the clock to use to compute sync times
     */
    public UfsSyncPathCache(Clock clock) {
      this(clock, null);
    }
  
    @VisibleForTesting
    UfsSyncPathCache(Clock clock, @Nullable BiConsumer<String, SyncState> onRemoval) {
      mClock = Preconditions.checkNotNull(clock);
      mItems = CacheBuilder.newBuilder().concurrencyLevel(
              Configuration.getInt(PropertyKey.MASTER_UFS_PATH_CACHE_THREADS))
          .removalListener(
              (removal) -> {
                if (removal.wasEvicted() && removal.getKey() != null && removal.getValue() != null) {
                  if (onRemoval != null) {
                    onRemoval.accept((String) removal.getKey(), (SyncState) removal.getValue());
                  }
                  // Due to issue https://github.com/Alluxio/alluxio/issues/18641,
                  // we decide to stop updating invalidation time for paths in order to avoid
                  // paths got evicted recursively. Stopping calling this onCacheEviction
                  // function will result in stopping updating invalidation time for all paths.
                  // Since no one is using this invalidation time feature, we decide to disable
                  // updating invalidation time.
                }
              })
          .maximumSize(Configuration.getInt(
              PropertyKey.MASTER_UFS_PATH_CACHE_CAPACITY))
          .build();
    }
  
    /**
     * Called when an element is evicted from the cache.
     * @param path the path
     * @param state the state
     */
    void onCacheEviction(String path, SyncState state) {
      try {
        // On eviction, we must mark our parent as needing a sync with our invalidation time.
        // Note that if the parent has a more recent sync time than this updated invalidation
        // time the parent will still not need a sync
        notifyInvalidationInternal(PathUtils.getParentCleaned(path), state.mInvalidationTime);
      } catch (InvalidPathException e) {
        throw new RuntimeException("Should not have an invalid path in the cache", e);
      }
    }
  
    /**
     * Called when starting a sync.
     * @return the time at the start of the sync
     */
    public long recordStartSync() {
      return mClock.millis();
    }
  
    /**
     * Get sync times for a given path if they exist in the cache.
     * @param path the path to check
     * @return a pair of sync times, where element 0 is the direct sync time
     * and element 1 is the recursive sync time
     */
    public Optional<Pair<Long, Long>> getSyncTimesForPath(AlluxioURI path) {
      SyncState state = path.getPath().equals(AlluxioURI.SEPARATOR)
          ? mRoot : mItems.getIfPresent(path.getPath());
      return Optional.ofNullable(state).map(syncState ->
          new Pair<>(syncState.mSyncTime, syncState.mRecursiveSyncTime));
    }
  
    /**
     * Check if sync should happen.
     * A path is checked starting from the full path, all the way up to the root.
     * At each path the largest validation time and invalidation time is computed depending
     * on its descendant type. After each path component is checked, if the last invalidation
     * time is more recent than the last validation time then a sync is needed. Otherwise,
     * a sync is needed based on the difference between
     * the current time and the last sync time and the interval.
     * @param path the path to check
     * @param intervalMs the frequency in ms that the sync should happen
     * @param descendantType the descendant type of the operation being performed
     * @return a {@link SyncCheck} object indicating if a sync should be performed
     */
    public SyncCheck shouldSyncPath(AlluxioURI path, long intervalMs, DescendantType descendantType)
        throws InvalidPathException {
      int parentLevel = 0;
      String currPath = cleanPath(path.getPath());
  
      long lastSyncTime = 0;
      long lastInvalidationTime = 0;
      SyncState syncState;
      boolean isFile = false;
      while (true) {
        syncState = currPath.equals(AlluxioURI.SEPARATOR) ? mRoot : mItems.getIfPresent(currPath);
        if (syncState != null) {
          // we always check if the current path has been invalidated
          // This invalidation time will never be updated.
          lastInvalidationTime = Math.max(lastInvalidationTime,
              syncState.mInvalidationTime);
          switch (parentLevel) {
            case 0: // the base path
              isFile = syncState.mIsFile;
              // if the last sync on this path was a file, then we can use
              // the normal validation time, as it has no children
              lastSyncTime = isFile ? Math.max(lastSyncTime, syncState.mSyncTime)
                  : lastSyncTime;
              switch (descendantType) {
                case NONE:
                  // we are syncing no children, so we use our validation time
                  lastSyncTime = Math.max(lastSyncTime, syncState.mSyncTime);
                  break;
                case ONE:
                  lastSyncTime = Math.max(lastSyncTime, syncState.mDirectChildrenSyncTime);
                  // since we are syncing the children, we must check if a child was invalidated
                  // This invalidation time will never be updated.
                  lastInvalidationTime = Math.max(lastInvalidationTime,
                      syncState.mDirectChildrenInvalidation);
                  break;
                case ALL:
                  // if the last sync on this path was a file, then we can use
                  // the normal validation time, as it has no children
                  lastSyncTime = Math.max(lastSyncTime, syncState.mRecursiveSyncTime);
                  // since we are syncing recursively, we must check if any recursive
                  // child was invalidated
                  // This invalidation time will never be updated.
                  lastInvalidationTime = Math.max(lastInvalidationTime,
                      syncState.mRecursiveChildrenInvalidation);
                  break;
                default:
                  throw new RuntimeException("Unexpected descendant type " + descendantType);
              }
              break;
            case 1: // at the parent path
              lastSyncTime = Math.max(lastSyncTime,
                  // if the child is a file we can use the direct sync time
                  isFile ? syncState.mDirectChildrenSyncTime :
                      // if we are only syncing the child we can use the direct sync time
                      descendantType != DescendantType.NONE ? syncState.mRecursiveSyncTime
                          : syncState.mDirectChildrenSyncTime);
              break;
            default: // at a higher ancestor path
              lastSyncTime = Math.max(lastSyncTime, syncState.mRecursiveSyncTime);
          }
        }
        if (currPath.equals(AlluxioURI.SEPARATOR)) {
          break;
        }
        currPath = PathUtils.getParentCleaned(currPath);
        parentLevel++;
      }
      return computeSyncResult(path, lastSyncTime, lastInvalidationTime, intervalMs);
    }
  
    private SyncCheck computeSyncResult(AlluxioURI path, long lastValidationTime,
        long lastInvalidationTime, long intervalMs) {
      long currentTime = mClock.millis();
      SyncCheck result;
      if (intervalMs == 0) {
        // Always sync.
        result = SyncCheck.shouldSyncWithTime(lastValidationTime);
      } else if (lastInvalidationTime == 0 && intervalMs < 0) {
        // if no invalidation, and a negative interval then do not sync
        result = SyncCheck.shouldNotSyncWithTime(lastValidationTime);
      } else if (lastInvalidationTime >= lastValidationTime) {
        // if a recent invalidation then we always sync
        result = SyncCheck.shouldSyncWithTime(lastValidationTime);
      } else if (intervalMs < 0) {
        // do not sync with a negative interval
        result = SyncCheck.shouldNotSyncWithTime(lastValidationTime);
      } else if ((currentTime - lastValidationTime) >= intervalMs) {
        // syncing is needed based on an interval
        result = SyncCheck.shouldSyncWithTime(lastValidationTime);
      } else {
        // syncing is not needed based on an interval
        result = SyncCheck.shouldNotSyncWithTime(lastValidationTime);
      }
      LOG.debug("Result of should sync path {}: {}, invalidation time: {}, validation time: {},"
              + " clock time: {}",
          result, path, lastInvalidationTime, lastValidationTime, currentTime);
      return result;
    }
  
    /**
     * Notify that a path has been invalidated.
     * @param path the path
     */
    @VisibleForTesting
    public void notifyInvalidation(AlluxioURI path) throws InvalidPathException {
      String currPath = cleanPath(path.getPath());
      long time = mClock.millis();
      notifyInvalidationInternal(currPath, time);
    }
  
    private void notifyInvalidationInternal(String currPath, long time)
        throws InvalidPathException {
      LOG.debug("Set sync invalidation for path {} at time {}", currPath, time);
      int parentLevel = 0;
  
      if (currPath.equals(AlluxioURI.SEPARATOR)) {
        try (LockResource ignored = new LockResource(mRootLock)) {
          mRoot.setInvalidationTime(time);
        }
      } else {
        mItems.asMap().compute(currPath, (key, state) -> {
          if (state == null) {
            state = new SyncState(false);
          }
          // set the invalidation time for the path
          state.setInvalidationTime(time);
          return state;
        });
      }
      while (!currPath.equals(AlluxioURI.SEPARATOR)) {
        currPath = PathUtils.getParentCleaned(currPath);
        parentLevel++;
        if (currPath.equals(AlluxioURI.SEPARATOR)) {
          try (LockResource ignored = new LockResource(mRootLock)) {
            updateParentInvalidation(mRoot, time, parentLevel);
          }
        } else {
          final int finalParentLevel = parentLevel;
          mItems.asMap().compute(currPath, (key, state) -> {
            if (state == null) {
              state = new SyncState(false);
            }
            updateParentInvalidation(state, time, finalParentLevel);
            return state;
          });
        }
      }
    }
  
    private void updateParentInvalidation(SyncState state, long time, long parentLevel) {
      state.setIsFile(false);
      if (parentLevel == 1) {
        // the parent has had a direct child invalidated
        state.setDirectChildInvalidation(time);
      }
      // the parent has had a recursive child invalidated
      state.setRecursiveChildInvalidation(time);
    }
  
    /**
     * Notify sync happened.
     * @param path the synced path
     * @param descendantType the descendant type for the performed operation
     * @param startTime the time at which the sync was started
     * @param syncTime the time to set the sync success to, if null then the current
     *                 clock time is used
     * @param isFile true if the synced path is a file
     * @return the sync state
     */
    public SyncState notifySyncedPath(
        AlluxioURI path, DescendantType descendantType, long startTime, @Nullable Long syncTime,
        boolean isFile) {
      long time = syncTime == null ? startTime :
          Math.min(startTime, syncTime);
      if (path.getPath().equals(AlluxioURI.SEPARATOR)) {
        try (LockResource ignored = new LockResource(mRootLock)) {
          Preconditions.checkState(!isFile);
          updateSyncState(mRoot, time, startTime, false, descendantType);
          return mRoot;
        }
      } else {
        return mItems.asMap().compute(path.getPath(), (key, state) -> {
          if (state == null) {
            state = new SyncState(isFile);
          }
          updateSyncState(state, time, startTime, isFile, descendantType);
          return state;
        });
      }
    }
  
    private void updateSyncState(
        SyncState state, long time, long startTime, boolean isFile, DescendantType descendantType) {
      state.setIsFile(isFile);
      state.setValidationTime(time, descendantType);
      // we can reset the invalidation times, as long as we have not received
      // an invalidation since the sync was started
      if (descendantType == DescendantType.ALL && state.mInvalidationTime < startTime) {
        state.mInvalidationTime = 0;
      }
      if (descendantType == DescendantType.ALL && state.mRecursiveChildrenInvalidation < startTime) {
        state.mRecursiveChildrenInvalidation = 0;
      }
      if ((descendantType == DescendantType.ALL || descendantType == DescendantType.ONE)
          && state.mDirectChildrenInvalidation < startTime) {
        state.mDirectChildrenInvalidation = 0;
      }
    }
  }
  