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

import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.master.Master;
import alluxio.master.journal.JournalReader;
import alluxio.master.journal.JournalUtils;
import alluxio.master.journal.sink.JournalSink;
import alluxio.proto.journal.Journal.JournalEntry;
import alluxio.retry.ExponentialBackoffRetry;
import alluxio.thread.AutopsyThread;
import alluxio.util.CommonUtils;
import alluxio.util.ExceptionUtils;

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

import java.io.IOException;
import java.util.OptionalLong;
import java.util.Set;
import java.util.function.Supplier;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.NotThreadSafe;

/**
 * This thread continually replays the journal and applies it to the master, until the master
 * initiates the shutdown of the thread.
 *
 * It periodically creates checkpoints. When the thread is stopped while it is writing checkpoint,
 * the checkpoint being written will be cancelled.
 *
 * Once awaitTermination is called, the thread will ignore InterruptedException and shut down like
 * normal (replaying all completed journal logs and waiting for a quiet period to elapse).
 */
@NotThreadSafe
public final class UfsJournalCheckpointThread extends AutopsyThread {
  private static final Logger LOG = LoggerFactory.getLogger(UfsJournalCheckpointThread.class);

  /** The master to apply the journal entries to. */
  private final Master mMaster;
  /** The journal. */
  private final UfsJournal mJournal;
  /** Make sure no new journal logs are found for this amount of time before shutting down. */
  private final long mShutdownQuietWaitTimeMs;
  /** If not journal log is found, sleep for this amount of time and check again. */
  private final int mJournalCheckpointSleepTimeMs;
  /** Writes a new checkpoint after processing this many journal entries. */
  private final int mCheckpointPeriodEntries;
  /** Object for sycnhronizing accesses to mCheckpointing. */
  private final Object mCheckpointingLock = new Object();
  /** Whether we are currently creating a checkpoint. */
  @GuardedBy("mCheckpointingLock")
  private volatile boolean mCheckpointing = false;
  /** This becomes true when the master initiates the shutdown. */
  private volatile boolean mShutdownInitiated = false;

  /** True if this thread is no longer running. */
  private volatile boolean mStopped = false;

  /** Controls whether the thread will wait for a quiet period to elapse before terminating. */
  private volatile boolean mWaitQuietPeriod = true;

  /** The journal reader. */
  private JournalReader mJournalReader;

  /**
   * The next sequence number of the log journal entry to checkpoint.
   */
  private long mNextSequenceNumberToCheckpoint;

  /** A supplier of journal sinks for this journal. */
  private final Supplier<Set<JournalSink>> mJournalSinks;

  /** The last sequence number applied to the journal. */
  private volatile long mLastAppliedSN;

  /**
   * The state of the journal catchup.
   */
  public enum CatchupState {
    NOT_STARTED, IN_PROGRESS, DONE
  }

  private volatile CatchupState mCatchupState = CatchupState.NOT_STARTED;

  /**
   * Creates a new instance of {@link UfsJournalCheckpointThread}.
   *
   * @param master the master to apply the journal entries to
   * @param journal the journal
   * @param journalSinks a supplier for journal sinks
   */
  public UfsJournalCheckpointThread(Master master, UfsJournal journal,
      Supplier<Set<JournalSink>> journalSinks) {
    this(master, journal, 0L, journalSinks);
  }

  /**
   * Creates a new instance of {@link UfsJournalCheckpointThread}.
   *
   * @param master the master to apply the journal entries to
   * @param journal the journal
   * @param startSequence the journal start sequence
   * @param journalSinks a supplier for journal sinks
   */
  public UfsJournalCheckpointThread(Master master, UfsJournal journal, long startSequence,
      Supplier<Set<JournalSink>> journalSinks) {
    super();
    mMaster = Preconditions.checkNotNull(master, "master");
    mJournal = Preconditions.checkNotNull(journal, "journal");
    mShutdownQuietWaitTimeMs = journal.getQuietPeriodMs();
    mJournalCheckpointSleepTimeMs =
        (int) Configuration.getMs(PropertyKey.MASTER_JOURNAL_TAILER_SLEEP_TIME_MS);
    mJournalReader = new UfsJournalReader(mJournal, startSequence, false);
    mCheckpointPeriodEntries =
        Configuration.getInt(PropertyKey.MASTER_JOURNAL_CHECKPOINT_PERIOD_ENTRIES);
    mJournalSinks = journalSinks;
    setName(String.format("ufs-checkpoint-thread-%s", mMaster.getName()));
  }

  /**
   * Initiates the shutdown of this checkpointer thread, and also waits for it to finish.
   *
   * @param waitQuietPeriod whether to wait for a quiet period to pass before terminating the thread
   * @throws RuntimeException if {@link #join()} throws an InterruptedException or if
   * {@link #run()} completed exceptionally
   */
  public void awaitTermination(boolean waitQuietPeriod) {
    LOG.info("{}: Journal checkpointer shutdown has been initiated.", mMaster.getName());
    mWaitQuietPeriod = waitQuietPeriod;
    mShutdownInitiated = true;
    // Actively interrupt to cancel slow checkpoints.
    synchronized (mCheckpointingLock) {
      if (mCheckpointing) {
        interrupt();
      }
    }

    try {
      // Wait for the thread to finish.
      join();
      if (crashed()) {
        LOG.error("Journal checkpointer has crashed internally before the shutdown");
        throw new RuntimeException(getError());
      }
      LOG.info("{}: Journal checkpointer shutdown complete", mMaster.getName());
    } catch (InterruptedException e) {
      LOG.error("{}: journal checkpointer shutdown is interrupted.", mMaster.getName(), e);
      // Kills the master. This can happen in the following two scenarios:
      // 1. The user Ctrl-C the server.
      // 2. Zookeeper selects this master as standby before the master finishes the previous
      //    standby->leader transition. It is safer to crash the server because the behavior is
      //    undefined to have two journal checkpointer running concurrently.
      throw new RuntimeException(e);
    }
    mStopped = true;
  }

  /**
   * This should only be called after {@link UfsJournalCheckpointThread#awaitTermination(boolean)}.
   *
   * @return the last edit log sequence number read plus 1
   */
  public long getNextSequenceNumber() {
    Preconditions.checkState(mStopped);
    return mJournalReader.getNextSequenceNumber();
  }

  @Override
  public void run() {
    // Start a new thread which tracks the journal replay statistics
    ExponentialBackoffRetry retry = new ExponentialBackoffRetry(1000, 30000, Integer.MAX_VALUE);
    long start = System.currentTimeMillis();
    final OptionalLong finalSN = UfsJournalReader.getLastSN(mJournal);
    Thread t = new Thread(() -> {
      UfsJournalProgressLogger progressLogger =
          new UfsJournalProgressLogger(mJournal, finalSN, () -> mLastAppliedSN);
      while (!Thread.currentThread().isInterrupted() && retry.attempt()) {
        // log current stats
        progressLogger.logProgress();
      }
    });
    try {
      t.start();
      runInternal();
    } finally {
      t.interrupt();
      try {
        t.join();
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        LOG.warn("interrupted while waiting for journal stats thread to shut down.");
      }
    }
  }

  /**
   * If the journal is corrupted, this thread will crash the master process.
   * On IOExceptions this thread will retry indefinitely.
   * If an exception is thrown from this method, it suggests an uncaught exception/RuntimeException
   * or the master is failing over/shutting down.
   */
  private void runInternal() {
    // Keeps reading journal entries. If none is found, sleep for sometime. Periodically write
    // checkpoints if some conditions are met. When a shutdown signal is received, wait until
    // no new journal entries.

    LOG.info("{}: Journal checkpoint thread started.", mMaster.getName());
    // Set to true if it has waited for a quiet period. Reset if a valid journal entry is read.
    boolean quietPeriodWaited = false;
    mCatchupState = CatchupState.IN_PROGRESS;
    while (true) {
      JournalEntry entry = null;
      try {
        switch (mJournalReader.advance()) {
          case CHECKPOINT:
            LOG.debug("{}: Restoring from checkpoint", mMaster.getName());
            mMaster.restoreFromCheckpoint(mJournalReader.getCheckpoint());
            LOG.debug("{}: Finished restoring from checkpoint", mMaster.getName());
            break;
          case LOG:
            entry = mJournalReader.getEntry();
            try {
              if (mMaster.processJournalEntry(entry)) {
                JournalUtils.sinkAppend(mJournalSinks, entry);
              } else {
                JournalUtils.handleJournalReplayFailure(LOG, null,
                    "%s: Unrecognized journal entry: %s", mMaster.getName(), entry);
              }
            } catch (RuntimeException e) {
              throw e;
            } catch (Throwable t) {
              JournalUtils.handleJournalReplayFailure(LOG, t,
                  "%s: Failed to read or process journal entry %s.", mMaster.getName(), entry);
            }
            if (quietPeriodWaited) {
              LOG.info("Quiet period interrupted by new journal entry");
              quietPeriodWaited = false;
            }
            mLastAppliedSN = entry.getSequenceNumber();
            break;
          default:
            mCatchupState = CatchupState.DONE;
            break;
        }
      } catch (IOException e) {
        LOG.error("{}: Failed to read or process a journal entry.", mMaster.getName(), e);
        try {
          mJournalReader.close();
        } catch (IOException ee) {
          LOG.warn("{}: Failed to close the journal reader with error {}.", mMaster.getName(),
              ee.toString());
        }
        long nextSequenceNumber = mJournalReader.getNextSequenceNumber();

        mJournalReader = new UfsJournalReader(mJournal, nextSequenceNumber, false);
        quietPeriodWaited = false;
        continue;
      }

      // Sleep for a while if no entry is found.
      if (entry == null) {
        // Only try to checkpoint when it can keep up.
        maybeCheckpoint();
        if (mShutdownInitiated) {
          if (quietPeriodWaited || !mWaitQuietPeriod) {
            mCatchupState = CatchupState.DONE;
            LOG.info("{}: Journal checkpoint thread has been shutdown. No new logs have been found "
                + "during the quiet period.", mMaster.getName());
            if (mJournalReader != null) {
              try {
                mJournalReader.close();
              } catch (IOException e) {
                LOG.warn("{}: Failed to close the journal reader with error {}.", mMaster.getName(),
                    e.toString());
              }
            }
            return;
          }
          CommonUtils.sleepMs(LOG, mShutdownQuietWaitTimeMs);
          quietPeriodWaited = true;
        } else {
          CommonUtils.sleepMs(LOG, mJournalCheckpointSleepTimeMs);
        }
      }
      // If mShutdownInitiated then the thread is already shutting down
      if (Thread.interrupted() && !mShutdownInitiated) {
        LOG.info("{}: Checkpoint thread interrupted, shutting down", mMaster.getName());
        return;
      }
    }
  }

  /**
   * @return the state of the master process journal catchup
   */
  public CatchupState getCatchupState() {
    return mCatchupState;
  }

  /**
   * Creates a new checkpoint if necessary.
   */
  private void maybeCheckpoint() {
    if (mShutdownInitiated) {
      return;
    }
    long nextSequenceNumber = mJournalReader.getNextSequenceNumber();
    if (nextSequenceNumber - mNextSequenceNumberToCheckpoint < mCheckpointPeriodEntries) {
      return;
    }
    try {
      mNextSequenceNumberToCheckpoint = mJournal.getNextSequenceNumberToCheckpoint();
    } catch (IOException e) {
      LOG.warn("{}: Failed to get the next sequence number to checkpoint with error {}.",
          mMaster.getName(), e.toString());
      return;
    }
    if (nextSequenceNumber - mNextSequenceNumberToCheckpoint < mCheckpointPeriodEntries) {
      return;
    }

    writeCheckpoint(nextSequenceNumber);
  }

  private void writeCheckpoint(long nextSequenceNumber) {
    LOG.info("{}: Writing checkpoint [sequence number {}].", mMaster.getName(), nextSequenceNumber);
    try {
      UfsJournalCheckpointWriter journalWriter =
          mJournal.getCheckpointWriter(nextSequenceNumber);
      try {
        synchronized (mCheckpointingLock) {
          if (mShutdownInitiated) {
            // This checkpoint thread is signaled to shutdown, so any checkpoint in progress must be
            // canceled/invalidated.
            journalWriter.cancel();
            return;
          }
          mCheckpointing = true;
        }
        mMaster.writeToCheckpoint(journalWriter);
      } catch (Throwable t) {
        if (ExceptionUtils.containsInterruptedException(t)) {
          Thread.currentThread().interrupt();
        } else {
          LOG.error("{}: Failed to create checkpoint", mMaster.getName(), t);
        }
        journalWriter.cancel();
        LOG.info("{}: Cancelled checkpoint [sequence number {}].", mMaster.getName(),
            nextSequenceNumber);
        return;
      } finally {
        synchronized (mCheckpointingLock) {
          mCheckpointing = false;
        }
        // If shutdown has been initiated, we assume that the interrupt was just intended to break
        // out of writeToCheckpoint early. We complete an orderly shutdown instead of stopping the
        // thread early.
        if (Thread.interrupted() && !mShutdownInitiated) {
          LOG.warn("{}: Checkpoint was interrupted but shutdown has not be initiated",
              mMaster.getName());
          // Interrupt to break out of runInternal()
          Thread.currentThread().interrupt();
        }
        journalWriter.close();
      }
      LOG.info("{}: Finished checkpoint [sequence number {}].", mMaster.getName(),
          nextSequenceNumber);
      mNextSequenceNumberToCheckpoint = nextSequenceNumber;
    } catch (IOException e) {
      // IOException here means either failing to create UfsWriter
      // or failing to close JournalWriter
      LOG.error("{}: Failed to checkpoint.", mMaster.getName(), e);
    }
  }

  @Override
  public void onError(Throwable t) {
    // if the catchup thread terminates exceptionally, it has caught up as much as it can and
    // is done
    mCatchupState = CatchupState.DONE;
    super.onError(t);
  }
}
