/*
 * 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.journal;

import alluxio.Constants;
import alluxio.underfs.UnderFileSystem;
import alluxio.util.UnderFileSystemUtils;

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

import java.io.IOException;

/**
 *
 */
public final class CheckpointManager {
  private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);

  /** The UFS where the journal is being written to. */
  private final UnderFileSystem mUfs;
  /**
   * Absolute path to the checkpoint file. This is where the latest checkpoint is stored. During
   * normal operation (when not writing a new checkpoint file), this is the only checkpoint file
   * that exists. If this file exists and there is no {@link #mTempBackupCheckpointPath}, it plus
   * all completed logs, plus the active log, should represent the full state of the master.
   */
  private final String mCheckpointPath;
  /**
   * Absolute path to the backup checkpoint file. The latest checkpoint is saved here while renaming
   * the temporary checkpoint so that we can recover in case the rename fails. If this file and
   * {@link #mCheckpointPath} both exist, {@link #mCheckpointPath} is the most up to date checkpoint
   * and {@link #mBackupCheckpointPath} should be deleted.
   */
  private final String mBackupCheckpointPath;
  /**
   * Absolute path to the temporary backup checkpoint file. This path is used as an intermediate
   * rename step when backing up {@link #mCheckpointPath} to {@link #mBackupCheckpointPath}. As long
   * as this file exists, it supercedes mCheckpointPath as the most up to date checkpoint file.
   */
  private final String mTempBackupCheckpointPath;
  /**
   * A journal writer through which this checkpoint manager can delete completed logs when the
   * checkpoint is updated.
   */
  private final JournalWriter mWriter;

  /**
   * Creates a new instance of {@link CheckpointManager}.
   *
   * @param ufs the under file system holding the journal
   * @param the directory for the journal
   */
  public CheckpointManager(UnderFileSystem ufs, String checkpointPath, JournalWriter writer) {
    mUfs = ufs;
    mCheckpointPath = checkpointPath;
    mBackupCheckpointPath = mCheckpointPath + ".backup";
    mTempBackupCheckpointPath = mBackupCheckpointPath + ".tmp";
    mWriter = writer;
  }

  /**
   * Recovers the checkpoint file in case the master crashed while updating it previously.
   *
   * The checkpointing process goes
   * <pre>
   * 1. Write mTempCheckpointPath based on all completed logs
   * 2. Rename mCheckpointPath to mTempBackupCheckpointPath
   * 3. Rename mTempBackupCheckpointPath to mBackupCheckpointPath
   * 4. Rename mTempCheckpointPath to mCheckpointPath
   * 5. Delete completed logs
   * 6. Delete mBackupCheckpointPath
   * </pre>
   */
  public void recoverCheckpoint() {
    try {
      Preconditions.checkState(
          !(mUfs.exists(mCheckpointPath) && mUfs.exists(mTempBackupCheckpointPath)
              && mUfs.exists(mBackupCheckpointPath)),
          "checkpoint, temp backup checkpoint, and backup checkpoint should never exist "
              + "simultaneously");
      if (mUfs.exists(mTempBackupCheckpointPath)) {
        // If mCheckpointPath exists, step 2 must have implemented rename as copy + delete, and
        // failed during the delete.
        UnderFileSystemUtils.deleteIfExists(mUfs, mCheckpointPath);
        mUfs.rename(mTempBackupCheckpointPath, mCheckpointPath);
      }
      if (mUfs.exists(mBackupCheckpointPath)) {
        // We must have crashed after step 3
        if (mUfs.exists(mCheckpointPath)) {
          // We crashed after step 4, so we can finish steps 5 and 6.
          mWriter.deleteCompletedLogs();
          mUfs.delete(mBackupCheckpointPath, false);
        } else {
          // We crashed before step 4, so we roll back to backup checkpoint.
          mUfs.rename(mBackupCheckpointPath, mCheckpointPath);
        }
      }
    } catch (IOException e) {
      throw Throwables.propagate(e);
    }
  }

  /**
   * @throws Exception
   *
   */
  public void updateCheckpoint(String newCheckpointPath) {
    try {
      if (mUfs.exists(mCheckpointPath)) {
        UnderFileSystemUtils.deleteIfExists(mUfs, mTempBackupCheckpointPath);
        UnderFileSystemUtils.deleteIfExists(mUfs, mBackupCheckpointPath);
        // Rename in two steps so that we never have identical mCheckpointPath and
        // mBackupCheckpointPath. This is a concern since UFS may implement rename as copy + delete.
        mUfs.rename(mCheckpointPath, mTempBackupCheckpointPath);
        mUfs.rename(mTempBackupCheckpointPath, mBackupCheckpointPath);
      }
      mUfs.rename(newCheckpointPath, mCheckpointPath);
      LOG.info("Renamed checkpoint file {} to {}", newCheckpointPath, mCheckpointPath);

      // The checkpoint already reflects the information in the completed logs.
      mWriter.deleteCompletedLogs();
      UnderFileSystemUtils.deleteIfExists(mUfs, mBackupCheckpointPath);
    } catch (IOException e) {
      throw Throwables.propagate(e);
    }
  }
}

@@ -829,16 +829,669 @@
 %0A%0A/**%0A *
+ A class which manages the checkpoint for a journal. The %7B@link #updateCheckpoint(String)%7D method%0A * will update the journal's checkpoint file to a specified checkpoint, and%0A * %7B@link recoverCheckpoint()%7D will recover from any failures that may occur during%0A * %7B@link #updateCheckpoint(String)%7D.%0A *%0A * The checkpoint updating process goes%0A * %3Cpre%3E%0A * 1. Write a new checkpoint named checkpoint.data.tmp%0A * 2. Rename checkpoint.data to checkpoint.data.backup.tmp%0A * 3. Rename checkpoint.data.backup.tmp to checkpoint.data.backup%0A * 4. Rename checkpoint.data.tmp to checkpoint.data%0A * 5. Delete completed logs%0A * 6. Delete checkpoint.data.backup%0A * %3C/pre%3E
 %0A */%0Apub
@@ -3661,381 +3661,8 @@
 ly.%0A
-   *%0A   * The checkpointing process goes%0A   * %3Cpre%3E%0A   * 1. Write mTempCheckpointPath based on all completed logs%0A   * 2. Rename mCheckpointPath to mTempBackupCheckpointPath%0A   * 3. Rename mTempBackupCheckpointPath to mBackupCheckpointPath%0A   * 4. Rename mTempCheckpointPath to mCheckpointPath%0A   * 5. Delete completed logs%0A   * 6. Delete mBackupCheckpointPath%0A   * %3C/pre%3E%0A
    *
@@ -4898,30 +4898,285 @@
   * 
-@throws Exception%0A   *
+Updates the checkpoint file to the checkpoint at the specified path. The update is done in such%0A   * a way that %7B@link #recoverCheckpoint()%7D will recover from any failures that occur during the%0A   * update.%0A   *%0A   * @param newCheckpointPath the path to the new checkpoint file
 %0A   

Update javadoc