alluxio-master-2.x/core/server/master/src/main/java/alluxio/master/file/mdsync/SyncProcessContext.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.master.file.mdsync;

import alluxio.AlluxioURI;
import alluxio.collections.ConcurrentHashSet;
import alluxio.file.options.DescendantType;
import alluxio.grpc.FileSystemMasterCommonPOptions;
import alluxio.master.file.BlockDeletionContext;
import alluxio.master.file.FileSystemJournalEntryMerger;
import alluxio.master.file.RpcContext;
import alluxio.master.file.contexts.OperationContext;
import alluxio.master.file.meta.UfsAbsentPathCache;
import alluxio.master.journal.FileSystemMergeJournalContext;
import alluxio.master.journal.MetadataSyncMergeJournalContext;

import com.google.common.base.Preconditions;

import java.io.Closeable;
import java.io.IOException;
import java.util.Set;

/**
 * The context for the metadata sync processing.
 */
public class SyncProcessContext implements Closeable {
  private final DescendantType mDescendantType;
  private final MetadataSyncRpcContext mRpcContext;
  private final RpcContext mBaseRpcContext;
  private final boolean mAllowConcurrentModification;
  private final FileSystemMasterCommonPOptions mCommonOptions;
  private final Set<AlluxioURI> mDirectoriesToUpdateAbsentCache = new ConcurrentHashSet<>();
  private final TaskInfo mTaskInfo;
  private final LoadResult mLoadResult;

  /**
   * Creates a metadata sync context.
   *
   * @param loadResult    the load UFS result
   * @param baseRpcContext the base rpc context
   * @param rpcContext    the metadata sync rpc context
   * @param commonOptions the common options for TTL configurations
   */
  private SyncProcessContext(
      LoadResult loadResult, RpcContext baseRpcContext, MetadataSyncRpcContext rpcContext,
      FileSystemMasterCommonPOptions commonOptions,
      boolean allowConcurrentModification
  ) {
    mDescendantType = loadResult.getLoadRequest().getDescendantType();
    mRpcContext = rpcContext;
    mBaseRpcContext = baseRpcContext;
    mCommonOptions = commonOptions;
    mAllowConcurrentModification = allowConcurrentModification;
    mTaskInfo = loadResult.getTaskInfo();
    mLoadResult = loadResult;
  }

  /**
   * @return the descendant type of the sync
   * NONE -> only syncs the inode itself
   * ONE -> syncs the inode and its direct children
   * ALL -> recursively syncs a directory
   */
  public DescendantType getDescendantType() {
    return mDescendantType;
  }

  /**
   * During the sync, the inodes might be updated by other requests concurrently, that makes
   * the sync operation stale. If the concurrent modification is allowed, these inodes will be
   * skipped, otherwise the sync will fail.
   *
   * @return true, if the concurrent modification is allowed. Otherwise, false
   */
  public boolean isConcurrentModificationAllowed() {
    return mAllowConcurrentModification;
  }

  /**
   * @return if the sync is a recursive sync
   */
  public boolean isRecursive() {
    return mDescendantType == DescendantType.ALL;
  }

  /**
   * @return the rpc context
   */
  public MetadataSyncRpcContext getRpcContext() {
    return mRpcContext;
  }

  /**
   * @return the metadata sync journal context
   */
  public MetadataSyncMergeJournalContext getMetadataSyncJournalContext() {
    return mRpcContext.getJournalContext();
  }

  /**
   * @return the common options
   */
  public FileSystemMasterCommonPOptions getCommonOptions() {
    return mCommonOptions;
  }

  /**
   * adds directories which are supposed to update is children loaded flag when the sync is done.
   *
   * @param path the path
   */
  public void addDirectoriesToUpdateIsChildrenLoaded(AlluxioURI path) {
    mTaskInfo.addPathToUpdateDirectChildrenLoaded(path);
  }

  /**
   * adds directories which exists and needs to update the absent cache later.
   * @param path the path
   */
  public void addDirectoriesToUpdateAbsentCache(AlluxioURI path) {
    mDirectoriesToUpdateAbsentCache.add(path);
  }

  /**
   * Updates the absent cache and set directories existing.
   * @param ufsAbsentPathCache the absent cache
   */
  public void updateAbsentCache(UfsAbsentPathCache ufsAbsentPathCache) {
    for (AlluxioURI uri: mDirectoriesToUpdateAbsentCache) {
      ufsAbsentPathCache.processExisting(uri);
    }
  }

  /**
   * reports the completion of a successful sync operation.
   *
   * @param operation the operation
   */
  public void reportSyncOperationSuccess(SyncOperation operation) {
    reportSyncOperationSuccess(operation, 1);
  }

  /**
   * reports the completion of a successful sync operation.
   *
   * @param operation the operation
   * @param count     the number of successes
   */
  public void reportSyncOperationSuccess(SyncOperation operation, long count) {
    operation.getCounter().inc(count);
    mTaskInfo.getStats().reportSyncOperationSuccess(operation, count);
  }

  /**
   * Reports a fail reason leading to the sync failure.
   *
   * @param reason the reason
   * @param t the throwable
   */
  public void reportSyncFailReason(SyncFailReason reason, Throwable t) {
    mTaskInfo.getStats().reportSyncFailReason(mLoadResult.getLoadRequest(), mLoadResult, reason, t);
  }

  /**
   * @return the task info
   */
  public TaskInfo getTaskInfo() {
    return mTaskInfo;
  }

  @Override
  public void close() throws IOException {
    mRpcContext.close();
    mBaseRpcContext.close();
  }

  static class MetadataSyncRpcContext extends RpcContext {
    public MetadataSyncRpcContext(
        BlockDeletionContext blockDeleter, MetadataSyncMergeJournalContext journalContext,
        OperationContext operationContext) {
      super(blockDeleter, journalContext, operationContext);
    }

    @Override
    public MetadataSyncMergeJournalContext getJournalContext() {
      return (MetadataSyncMergeJournalContext) super.getJournalContext();
    }
  }

  /**
   * Creates a builder.
   */
  public static class Builder {
    private LoadResult mLoadResult;
    private MetadataSyncRpcContext mRpcContext;
    private RpcContext mBaseRpcContext;
    private FileSystemMasterCommonPOptions mCommonOptions = DefaultSyncProcess.NO_TTL_OPTION;
    private boolean mAllowConcurrentModification = true;

    /**
     * Creates a builder.
     *
     * @param rpcContext the rpc context
     * @param loadResult the load UFS result
     * @return a new builder
     */
    public static Builder builder(RpcContext rpcContext, LoadResult loadResult) {
      Preconditions.checkState(
          !(rpcContext.getJournalContext() instanceof FileSystemMergeJournalContext));
      Builder builder = new Builder();
      builder.mLoadResult = loadResult;
      /*
       * Wrap the journal context with a MetadataSyncMergeJournalContext, which behaves
       * differently in:
       *  1. the journals are merged and stayed in the context until it gets flushed
       *  2. when close() or flush() are called, the journal does not trigger a hard flush
       *  that commits the journals, instead, it only adds the journals to the async journal writer.
       *  During the metadata sync process, we are creating/updating many files, but we don't want
       *  to hard flush journals on every inode updates.
       */
      builder.mBaseRpcContext = rpcContext;
      builder.mRpcContext = new MetadataSyncRpcContext(rpcContext.getBlockDeletionContext(),
          new MetadataSyncMergeJournalContext(rpcContext.getJournalContext(),
              new FileSystemJournalEntryMerger()), rpcContext.getOperationContext());
      return builder;
    }

    /**
     * @param rpcContext the rpc context
     * @return builder
     */
    public Builder setRpcContext(MetadataSyncRpcContext rpcContext) {
      mRpcContext = rpcContext;
      return this;
    }

    /**
     * @param commonOptions the common option
     * @return builder
     */
    public Builder setCommonOptions(FileSystemMasterCommonPOptions commonOptions) {
      mCommonOptions = commonOptions;
      return this;
    }

    /**
     * @param allowModification the current modification is allowed
     * @return the builder
     */
    public Builder setAllowModification(boolean allowModification) {
      mAllowConcurrentModification = allowModification;
      return this;
    }

    /**
     * @return the built metadata sync context
     */
    public SyncProcessContext build() {
      return new SyncProcessContext(
          mLoadResult, mBaseRpcContext, mRpcContext, mCommonOptions,
          mAllowConcurrentModification);
    }
  }
}
