alluxio-master-2.x/core/server/master/src/main/java/alluxio/master/file/mdsync/BaseTask.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.Pair;
import alluxio.exception.FileDoesNotExistException;
import alluxio.exception.InvalidPathException;
import alluxio.exception.runtime.DeadlineExceededRuntimeException;
import alluxio.exception.runtime.InternalRuntimeException;
import alluxio.exception.status.CancelledException;
import alluxio.exception.status.UnavailableException;
import alluxio.file.options.DescendantType;
import alluxio.file.options.DirectoryLoadType;
import alluxio.grpc.SyncMetadataState;
import alluxio.grpc.SyncMetadataTask;
import alluxio.master.file.DefaultFileSystemMaster;
import alluxio.master.file.meta.InodeTree;
import alluxio.master.file.meta.LockedInodePath;
import alluxio.master.journal.JournalContext;
import alluxio.resource.CloseableResource;
import alluxio.underfs.UfsClient;
import alluxio.util.CommonUtils;
import alluxio.util.ExceptionUtils;

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

import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

/**
 * This is the overall task for a sync operation.
 */
public abstract class BaseTask implements PathWaiter {
  enum State {
    RUNNING,
    SUCCEEDED,
    FAILED,
    CANCELED;

    SyncMetadataState toProto() {
      switch (this) {
        case RUNNING:
          return SyncMetadataState.RUNNING;
        case SUCCEEDED:
          return SyncMetadataState.SUCCEEDED;
        case FAILED:
          return SyncMetadataState.FAILED;
        case CANCELED:
          return SyncMetadataState.CANCELED;
        default:
          return SyncMetadataState.UNKNOWN;
      }
    }
  }

  private static final Logger LOG = LoggerFactory.getLogger(BaseTask.class);

  private final long mStartTime;
  private volatile Long mFinishTime = null;
  BaseTaskResult mIsCompleted = null;
  private final TaskInfo mTaskInfo;
  private final PathLoaderTask mPathLoadTask;
  private final boolean mRemoveOnComplete;

  /**
   * @return the task state
   */
  public synchronized State getState() {
    if (!isCompleted().isPresent()) {
      return State.RUNNING;
    }
    BaseTaskResult result = isCompleted().get();
    if (result.succeeded()) {
      return State.SUCCEEDED;
    } else if (result.getThrowable().orElse(null) instanceof CancelledException) {
      return State.CANCELED;
    } else {
      return State.FAILED;
    }
  }

  /**
   * @return true if the task is completed
   */
  public synchronized Optional<BaseTaskResult> isCompleted() {
    return Optional.ofNullable(mIsCompleted);
  }

  /**
   * @return if the task is succeeded
   */
  public synchronized boolean succeeded() {
    return mIsCompleted != null && mIsCompleted.succeeded();
  }

  @VisibleForTesting
  PathLoaderTask getPathLoadTask() {
    return mPathLoadTask;
  }

  static BaseTask create(
      TaskInfo info, long startTime,
      Function<AlluxioURI, CloseableResource<UfsClient>> clientSupplier,
      boolean removeOnComplete) {
    if (info.getLoadByDirectory() != DirectoryLoadType.SINGLE_LISTING
        && info.getDescendantType() == DescendantType.ALL) {
      return new DirectoryPathWaiter(
          info, startTime, clientSupplier, removeOnComplete);
    } else {
      return new BatchPathWaiter(
          info, startTime, clientSupplier, removeOnComplete);
    }
  }

  static BaseTask create(
      TaskInfo info, long startTime,
      Function<AlluxioURI, CloseableResource<UfsClient>> clientSupplier) {
    return create(info, startTime, clientSupplier, true);
  }

  BaseTask(
      TaskInfo info, long startTime,
      Function<AlluxioURI, CloseableResource<UfsClient>> clientSupplier, boolean removeOnComplete) {
    mTaskInfo = info;
    mStartTime = startTime;
    mPathLoadTask = new PathLoaderTask(mTaskInfo, null, clientSupplier);
    mRemoveOnComplete = removeOnComplete;
  }

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

  /**
   * @return true, if the task should be removed on completion, otherwise it will be
   * moved to a completed task cache.
   */
  boolean removeOnComplete() {
    return mRemoveOnComplete;
  }

  /**
   * @return the sync task time in ms
   */
  public synchronized long getStartTime() {
    Preconditions.checkState(mIsCompleted != null,
        "Task must be completed before accessing the start time");
    return mStartTime;
  }

  PathLoaderTask getLoadTask() {
    return mPathLoadTask;
  }

  synchronized void onComplete(
      boolean isFile, DefaultFileSystemMaster fileSystemMaster, InodeTree inodeTree) {
    if (mIsCompleted != null) {
      return;
    }
    updateDirectChildrenLoaded(fileSystemMaster, inodeTree);
    mFinishTime = CommonUtils.getCurrentMs();
    mIsCompleted = new BaseTaskResult(null);
    mTaskInfo.getMdSync().onTaskComplete(mTaskInfo.getId(), isFile);
    notifyAll();
  }

  /**
   * Blocking waits until the task completes.
   * If the task fails, the exception causing the failure is thrown.
   * If the wait times-out a {@link DeadlineExceededRuntimeException} is thrown.
   *
   * @param timeoutMs the timeout in ms, 0 for an endless wait
   */
  public synchronized void waitComplete(long timeoutMs) throws Throwable {
    Stopwatch sw = Stopwatch.createStarted();
    long waitTime = timeoutMs;
    while (mIsCompleted == null && (timeoutMs == 0 || waitTime > 0)) {
      wait(waitTime);
      if (timeoutMs != 0) {
        waitTime = waitTime - sw.elapsed(TimeUnit.MILLISECONDS);
        sw.reset();
      }
    }
    if (mIsCompleted == null) {
      throw new DeadlineExceededRuntimeException("Task still running.");
    }
    if (mIsCompleted.getThrowable().isPresent()) {
      throw mIsCompleted.getThrowable().get();
    }
  }

  synchronized void onFailed(Throwable t) {
    mFinishTime = CommonUtils.getCurrentMs();
    if (mIsCompleted != null) {
      return;
    }
    mIsCompleted = new BaseTaskResult(t);
    LOG.warn("Task {} failed with error", mTaskInfo, t);
    cancel();
    mTaskInfo.getMdSync().onTaskError(mTaskInfo.getId(), t);
  }

  synchronized long cancel() {
    mFinishTime = CommonUtils.getCurrentMs();
    if (mIsCompleted == null) {
      mIsCompleted = new BaseTaskResult(new CancelledException("Task was cancelled"));
    }
    mPathLoadTask.cancel();
    notifyAll();
    return mTaskInfo.getId();
  }

  boolean pathIsCovered(AlluxioURI path, DescendantType depth) {
    switch (mTaskInfo.getDescendantType()) {
      case NONE:
        return depth == DescendantType.NONE && mTaskInfo.getBasePath().equals(path);
      case ONE:
        return (depth != DescendantType.ALL && mTaskInfo.getBasePath().equals(path))
            || (depth == DescendantType.NONE && mTaskInfo.getBasePath().equals(path.getParent()));
      case ALL:
        try {
          return mTaskInfo.getBasePath().isAncestorOf(path);
        } catch (InvalidPathException e) {
          throw new InternalRuntimeException(e);
        }
      default:
        throw new InternalRuntimeException(String.format(
            "Unknown descendant type %s", mTaskInfo.getDescendantType()));
    }
  }

  /**
   * @return the sync duration in ms
   */
  public long getSyncDuration() {
    final Long finishTime = mFinishTime;
    if (finishTime == null) {
      return CommonUtils.getCurrentMs() - mStartTime;
    }
    return mFinishTime - mStartTime;
  }

  /**
   * @return the sync metadata task in proto
   */
  public synchronized SyncMetadataTask toProtoTask() {
    SyncMetadataTask.Builder builder = SyncMetadataTask.newBuilder();
    builder.setId(getTaskInfo().getId());
    builder.setState(getState().toProto());
    builder.setSyncDurationMs(getSyncDuration());
    Throwable t = null;
    if (mIsCompleted != null && mIsCompleted.getThrowable().isPresent()) {
      t = mIsCompleted.getThrowable().get();
    }
    if (t != null && getState() != State.CANCELED) {
      builder.setException(SyncMetadataTask.Exception.newBuilder()
          .setExceptionType(t.getClass().getTypeName())
          .setExceptionMessage(t.getMessage() == null ? "" : t.getMessage())
          .setStacktrace(ExceptionUtils.asPlainText(t)));
    }
    builder.setTaskInfoString(getTaskInfo().toString());
    Pair<Long, String> statReport = getTaskInfo().getStats().toReportString();
    builder.setSuccessOpCount(statReport.getFirst());
    builder.setTaskStatString(statReport.getSecond());
    return builder.build();
  }

  /**
   * Updates direct children loaded for directories affected by the metadata sync.
   * @param fileSystemMaster the file system master
   * @param inodeTree the inode tree
   */
  public void updateDirectChildrenLoaded(
      DefaultFileSystemMaster fileSystemMaster, InodeTree inodeTree) {
    try (JournalContext journalContext = fileSystemMaster.createJournalContext()) {
      getTaskInfo().getPathsToUpdateDirectChildrenLoaded().forEach(
          uri -> {
            try (LockedInodePath lockedInodePath =
                     inodeTree.lockInodePath(
                         uri, InodeTree.LockPattern.WRITE_INODE,
                         journalContext)) {
              if (lockedInodePath.fullPathExists() && lockedInodePath.getInode().isDirectory()
                  && !lockedInodePath.getInode().asDirectory().isDirectChildrenLoaded()) {
                inodeTree.setDirectChildrenLoaded(
                    () -> journalContext,
                    lockedInodePath.getInode().asDirectory());
              }
            } catch (FileDoesNotExistException | InvalidPathException e) {
              throw new RuntimeException(e);
            }
          });
    } catch (UnavailableException e) {
      throw new RuntimeException(e);
    }
  }
}
