/*
 * 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.cli.fs.command;

import alluxio.AlluxioURI;
import alluxio.Constants;
import alluxio.annotation.PublicApi;
import alluxio.cli.CommandUtils;
import alluxio.cli.fs.FileSystemShellUtils;
import alluxio.client.file.FileSystemContext;
import alluxio.client.file.FileSystemUtils;
import alluxio.client.file.URIStatus;
import alluxio.exception.AlluxioException;
import alluxio.exception.status.InvalidArgumentException;
import alluxio.util.ThreadFactoryUtils;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

import javax.annotation.concurrent.ThreadSafe;

/**
 * Persists files or directories currently stored only in Alluxio to the UnderFileSystem.
 */
@ThreadSafe
@PublicApi
public final class PersistCommand extends AbstractFileSystemCommand {
  private static final Logger LOG = LoggerFactory.getLogger(PersistCommand.class);
  private static final int DEFAULT_PARALLELISM = 4;
  private static final Option PARALLELISM_OPTION =
      Option.builder("p")
          .longOpt("parallelism")
          .argName("# concurrent operations")
          .numberOfArgs(1)
          .desc("Number of concurrent persist operations, default: " + DEFAULT_PARALLELISM)
          .required(false)
          .build();
  private static final int DEFAULT_TIMEOUT = 20 * Constants.MINUTE_MS;
  private static final Option TIMEOUT_OPTION =
      Option.builder("t")
          .longOpt("timeout")
          .argName("timeout in milliseconds")
          .numberOfArgs(1)
          .desc("Time in milliseconds for a single file persist to time out; default:"
              + DEFAULT_TIMEOUT)
          .required(false)
          .build();
  private static final int DEFAULT_WAIT_TIME = 0;
  private static final Option WAIT_OPTION =
      Option.builder("w")
          .longOpt("wait")
          .argName("the initial persistence wait time")
          .numberOfArgs(1)
          .desc("The time to wait before persisting. default: " + DEFAULT_WAIT_TIME)
          .required(false)
          .build();

  /**
   * @param fsContext the filesystem of Alluxio
   */
  public PersistCommand(FileSystemContext fsContext) {
    super(fsContext);
  }

  @Override
  public String getCommandName() {
    return "persist";
  }

  @Override
  public Options getOptions() {
    return new Options().addOption(PARALLELISM_OPTION).addOption(TIMEOUT_OPTION)
        .addOption(WAIT_OPTION);
  }

  @Override
  public void validateArgs(CommandLine cl) throws InvalidArgumentException {
    CommandUtils.checkNumOfArgsNoLessThan(this, cl, 1);
  }

  @Override
  public String getUsage() {
    return "persist [-p|--parallelism <#>] [-t|--timeout <milliseconds>] "
        + "[-w|--wait <milliseconds>] <path> [<path> ...]";
  }

  @Override
  public String getDescription() {
    return "Persists files or directories currently stored only in Alluxio to the UnderFileSystem.";
  }

  @Override
  public int run(CommandLine cl) throws AlluxioException, IOException {
    // Parse arguments
    int parallelism = FileSystemShellUtils.getIntArg(cl, PARALLELISM_OPTION, DEFAULT_PARALLELISM);
    int timeoutMs = (int) FileSystemShellUtils.getMsArg(cl, TIMEOUT_OPTION, DEFAULT_TIMEOUT);
    long persistenceWaitTimeMs = FileSystemShellUtils.getMsArg(cl, WAIT_OPTION, DEFAULT_WAIT_TIME);

    // Validate arguments
    if ((persistenceWaitTimeMs > timeoutMs && timeoutMs != -1) || persistenceWaitTimeMs < 0) {
      System.out.println("Persistence initial wait time should be smaller than persist timeout "
          + "and bigger than zero");
      return -1;
    }

    String[] args = cl.getArgs();

    // Gather files to persist and enqueue them.
    List<AlluxioURI> candidateUris = new ArrayList<>();
    for (String path : args) {
      candidateUris.addAll(FileSystemShellUtils.getAlluxioURIs(mFileSystem, new AlluxioURI(path)));
    }
    final Queue<AlluxioURI> toPersist = new ConcurrentLinkedQueue<>();
    for (AlluxioURI uri : candidateUris) {
      queueNonPersistedRecursive(mFileSystem.getStatus(uri), toPersist);
    }
    int totalFiles = toPersist.size();
    System.out.println("Found " + totalFiles + " files to persist.");
    if (totalFiles == 0) {
      return 0;
    }

    // Launch persist tasks in parallel.
    parallelism = Math.min(totalFiles, parallelism);
    ExecutorService service =
        Executors.newFixedThreadPool(parallelism, ThreadFactoryUtils.build("persist-cli-%d", true));
    final Object progressLock = new Object();
    AtomicInteger completedFiles = new AtomicInteger(0);
    List<Future<Void>> futures = new ArrayList<>(parallelism);
    for (int i = 0; i < parallelism; i++) {
      futures.add(service.submit(new PersistCallable(toPersist, totalFiles, completedFiles,
          progressLock, persistenceWaitTimeMs, timeoutMs)));
    }

    // Await result.
    try {
      for (Future<Void> future : futures) {
        future.get();
      }
    } catch (ExecutionException e) {
      System.out.println("Fatal error: " + e);
      return -1;
    } catch (InterruptedException e) {
      System.out.println("Persist interrupted, exiting.");
      return -1;
    } finally {
      service.shutdownNow();
    }
    return 0;
  }

  private void queueNonPersistedRecursive(URIStatus status, Queue<AlluxioURI> toPersist)
      throws AlluxioException, IOException {
    AlluxioURI uri = new AlluxioURI(status.getPath());
    if (status.isFolder()) {
      List<URIStatus> statuses = mFileSystem.listStatus(uri);
      for (URIStatus s : statuses) {
        queueNonPersistedRecursive(s, toPersist);
      }
    } else if (!status.isPersisted()) {
      toPersist.add(uri);
    }
  }

  /**
   * Thread that polls a persist queue and persists files.
   */
  private class PersistCallable implements Callable<Void> {
    private final Queue<AlluxioURI> mFilesToPersist;
    private final int mTotalFiles;
    private final Object mProgressLock;
    private final AtomicInteger mCompletedFiles;
    private final long mPersistenceWaitTime;
    private final int mTimeoutMs;

    PersistCallable(Queue<AlluxioURI> toPersist, int totalFiles, AtomicInteger completedFiles,
        Object progressLock, long persistenceWaitTime, int timeoutMs) {
      mFilesToPersist = toPersist;
      mTotalFiles = totalFiles;
      mProgressLock = progressLock;
      mCompletedFiles = completedFiles;
      mPersistenceWaitTime = persistenceWaitTime;
      mTimeoutMs = timeoutMs;
    }

    @Override
    public Void call() throws Exception {
      AlluxioURI toPersist = mFilesToPersist.poll();
      while (toPersist != null) {
        try {
          FileSystemUtils.persistAndWait(mFileSystem, toPersist, mPersistenceWaitTime, mTimeoutMs);
          synchronized (mProgressLock) { // Prevents out of order progress tracking.
            String progress = "(" + mCompletedFiles.incrementAndGet() + "/" + mTotalFiles + ")";
            System.out.println(progress + " Successfully persisted file: " + toPersist);
          }
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          throw e;
        } catch (TimeoutException e) {
          String timeoutMsg =
              String.format("Timed out waiting for file to be persisted: %s", toPersist);
          System.out.println(timeoutMsg);
          LOG.error(timeoutMsg, e);
        } catch (Exception e) {
          System.out.println("Failed to persist file " + toPersist);
          LOG.error("Failed to persist file {}", toPersist, e);
        }
        toPersist = mFilesToPersist.poll();
      }
      return null;
    }
  }
}
