/*
 * 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.FileInStream;
import alluxio.client.file.FileOutStream;
import alluxio.client.file.FileSystem;
import alluxio.client.file.FileSystemContext;
import alluxio.client.file.URIStatus;
import alluxio.exception.AlluxioException;
import alluxio.exception.ExceptionMessage;
import alluxio.exception.FileAlreadyExistsException;
import alluxio.exception.FileDoesNotExistException;
import alluxio.exception.InvalidPathException;
import alluxio.exception.status.InvalidArgumentException;
import alluxio.grpc.DeletePOptions;
import alluxio.grpc.SetAclAction;
import alluxio.grpc.SetAttributePOptions;
import alluxio.security.authorization.Mode;
import alluxio.util.io.PathUtils;

import com.google.common.base.Joiner;
import com.google.common.io.Closer;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.RandomStringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.PrintStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.ThreadSafe;

/**
 * Copies a file or a directory in the Alluxio filesystem.
 */
@ThreadSafe
@PublicApi
public final class CpCommand extends AbstractFileSystemCommand {
  private static final Logger LOG = LoggerFactory.getLogger(CpCommand.class);
  private static final String COPY_SUCCEED_MESSAGE = "Copied %s to %s";
  private static final String COPY_FAIL_MESSAGE = "Failed to copy %s to %s";
  private static final int COPY_FROM_LOCAL_BUFFER_SIZE_DEFAULT = 8 * Constants.MB;
  private static final int COPY_TO_LOCAL_BUFFER_SIZE_DEFAULT = 64 * Constants.MB;

  private static final Option RECURSIVE_OPTION =
      Option.builder("R").longOpt("recursive")
          .required(false)
          .hasArg(false)
          .desc("copy files in subdirectories recursively")
          .build();
  private static final Option RECURSIVE_ALIAS_OPTION =
      Option.builder("r")
          .required(false)
          .hasArg(false)
          .desc("copy files in subdirectories recursively")
          .build();
  public static final Option THREAD_OPTION =
      Option.builder()
          .longOpt("thread")
          .required(false)
          .hasArg(true)
          .numberOfArgs(1)
          .argName("threads")
          .type(Number.class)
          .desc("Number of threads used to copy files in parallel, default value is CPU cores * 2")
          .build();
  public static final Option BUFFER_SIZE_OPTION =
      Option.builder()
          .longOpt("buffersize")
          .required(false)
          .hasArg(true)
          .numberOfArgs(1)
          .argName("buffer size")
          .type(Number.class)
          .desc("Read buffer size in bytes, "
              + "default is 8MB when copying from local, "
              + "and 64MB when copying to local")
          .build();
  private static final Option PRESERVE_OPTION =
      Option.builder("p")
          .longOpt("preserve")
          .required(false)
          .desc("Preserve file permission attributes when copying files. "
              + "All ownership, permissions and ACLs will be preserved")
          .build();

  private int mCopyFromLocalBufferSize = COPY_FROM_LOCAL_BUFFER_SIZE_DEFAULT;
  private int mCopyToLocalBufferSize = COPY_TO_LOCAL_BUFFER_SIZE_DEFAULT;
  private int mThread = Runtime.getRuntime().availableProcessors() * 2;
  private boolean mPreservePermissions = false;

  /**
   * A thread pool executor for asynchronous copy.
   *
   * Copy tasks can send messages to an output stream in a thread safe way.
   */
  @ThreadSafe
  private static final class CopyThreadPoolExecutor {
    private static final class CopyException extends Exception {
      public CopyException(AlluxioURI src, AlluxioURI dst, Exception cause) {
        super(String.format(COPY_FAIL_MESSAGE, src, dst), cause);
      }
    }

    private static final Object MESSAGE_DONE = new Object();

    private final ThreadPoolExecutor mPool;
    /**
     * Message queue used by mPrinter.
     * Only supports objects of type String and Exception.
     * String messages will be printed to stdout;
     * Exception messages will be printed to stderr and collected into mExceptions.
     * Other types of messages will be ignored.
     */
    private final BlockingQueue<Object> mMessages;
    private final ConcurrentLinkedQueue<Exception> mExceptions;
    private final PrintStream mStdout;
    private final PrintStream mStderr;
    private final Thread mPrinter;
    private final FileSystem mFileSystem;
    private final AlluxioURI mPath;

    /**
     * Creates a new thread pool with the specified number of threads,
     * specify the output stream for tasks to send messages to, and
     * starts the background thread for printing messages.
     *
     * NOTE: needs to call {@link #shutdown()} to release resources.
     *
     * @param threads number of threads
     * @param stdout the stdout stream for tasks to send messages to
     * @param stderr the stderr stream for tasks to send error messages to
     * @param fileSystem the Alluxio filesystem used to delete path
     * @param path the path to delete on shutdown when it's empty, otherwise can be {@code null}
     */
    public CopyThreadPoolExecutor(int threads, PrintStream stdout, PrintStream stderr,
        FileSystem fileSystem, AlluxioURI path) {
      mPool = new ThreadPoolExecutor(threads, threads,
          1, TimeUnit.SECONDS, new ArrayBlockingQueue<>(threads * 2),
          new ThreadPoolExecutor.CallerRunsPolicy());
      mMessages = new LinkedBlockingQueue<>();
      mExceptions = new ConcurrentLinkedQueue<>();
      mStdout = stdout;
      mStderr = stderr;
      mPrinter = new Thread(() -> {
        while (!Thread.currentThread().isInterrupted()) {
          try {
            Object message = mMessages.take();
            if (message == MESSAGE_DONE) {
              break;
            }
            if (message instanceof String) {
              mStdout.println(message);
            } else if (message instanceof CopyException) {
              CopyException e = (CopyException) message;
              mStderr.println(messageAndCause(e));
            } else {
              LOG.error("Unsupported message type " + message.getClass()
                  + " in message queue of copy thread pool");
            }
          } catch (InterruptedException e) {
            break;
          }
        }
      });
      mPrinter.start();
      mFileSystem = fileSystem;
      mPath = path;
    }

    /**
     * Submits a copy task, returns immediately without waiting for completion.
     *
     * @param task the copy task
     */
    public <T> void submit(Callable<T> task) {
      mPool.submit(task);
    }

    /**
     * Sends a message to the pool to indicate that src is copied to dst,
     * the message will be displayed in the stdout stream.
     *
     * @param src the source path
     * @param dst the destination path
     * @throws InterruptedException if interrupted while waiting to send the message
     */
    public void succeed(AlluxioURI src, AlluxioURI dst) throws InterruptedException {
      mMessages.put(String.format(COPY_SUCCEED_MESSAGE, src, dst));
    }

    /**
     * Sends the exception to the pool to indicate that src fails to be copied to dst,
     * the exception will be displayed in the stderr stream.
     *
     * @param src the source path
     * @param dst the destination path
     * @param cause the cause of the failure
     * @throws InterruptedException if interrupted while waiting to send the exception
     */
    public void fail(AlluxioURI src, AlluxioURI dst, Exception cause) throws InterruptedException {
      CopyException exception = new CopyException(src, dst, cause);
      mExceptions.add(exception);
      mMessages.put(exception);
    }

    /**
     * Waits until all asynchronous copy tasks succeed or fail, then shuts down the thread pool,
     * joins the printer thread, and deletes the copy destination in case of error.
     *
     * @throws IOException summarizing all exceptions thrown in the submitted tasks and in shutdown
     */
    public void shutdown() throws IOException {
      mPool.shutdown();
      try {
        mPool.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
      } catch (InterruptedException e) {
        LOG.warn("Copy thread pool is interrupted in shutdown.", e);
        Thread.currentThread().interrupt();
        mPool.shutdownNow();
      }

      try {
        mMessages.put(MESSAGE_DONE);
        mPrinter.join();
      } catch (InterruptedException e) {
        LOG.warn("Message queue or printer in copy thread pool is interrupted in shutdown.", e);
        Thread.currentThread().interrupt();
        mPrinter.interrupt();
      }

      try {
        if (mPath != null
            && mFileSystem.exists(mPath)
            && mFileSystem.getStatus(mPath).isFolder()
            && mFileSystem.listStatus(mPath).isEmpty()) {
          mFileSystem.delete(mPath);
        }
      } catch (Exception e) {
        mExceptions.add(new IOException("Failed to delete path " + mPath, e));
      }

      if (!mExceptions.isEmpty()) {
        List<String> errors = new ArrayList<>();
        for (Exception e : mExceptions) {
          LOG.error(stacktrace(e));
          errors.add(messageAndCause(e));
        }
        throw new IOException("ERRORS:\n" + Joiner.on("\n").join(errors));
      }
    }

    private String messageAndCause(Exception e) {
      return e.getMessage() + ": " + e.getCause().getMessage();
    }

    private String stacktrace(Exception e) {
      ByteArrayOutputStream os = new ByteArrayOutputStream();
      PrintStream ps = new PrintStream(os, true);
      e.printStackTrace(ps);
      ps.close();
      return os.toString();
    }
  }

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

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

  @Override
  public void validateArgs(CommandLine cl) throws InvalidArgumentException {
    CommandUtils.checkNumOfArgsEquals(this, cl, 2);
    if (cl.hasOption(BUFFER_SIZE_OPTION.getLongOpt())) {
      try {
        int bufSize = ((Number) cl.getParsedOptionValue(BUFFER_SIZE_OPTION.getLongOpt()))
            .intValue();
        if (bufSize < 0) {
          throw new InvalidArgumentException(BUFFER_SIZE_OPTION.getLongOpt() + " must be > 0");
        }
        mCopyFromLocalBufferSize = bufSize;
        mCopyToLocalBufferSize = bufSize;
      } catch (ParseException e) {
        throw new InvalidArgumentException("Failed to parse option "
            + BUFFER_SIZE_OPTION.getLongOpt() + " into an integer", e);
      }
    }
    if (cl.hasOption(THREAD_OPTION.getLongOpt())) {
      try {
        mThread = ((Number) cl.getParsedOptionValue(THREAD_OPTION.getLongOpt())).intValue();
        if (mThread <= 0) {
          throw new InvalidArgumentException(THREAD_OPTION.getLongOpt() + " must be > 0");
        }
      } catch (ParseException e) {
        throw new InvalidArgumentException("Failed to parse option " + THREAD_OPTION.getLongOpt()
            + " into an integer", e);
      }
    }
    mPreservePermissions = cl.hasOption(PRESERVE_OPTION.getLongOpt());
  }

  @Override
  public Options getOptions() {
    return new Options().addOption(RECURSIVE_OPTION)
        .addOption(RECURSIVE_ALIAS_OPTION)
        .addOption(THREAD_OPTION)
        .addOption(PRESERVE_OPTION);
  }

  @Override
  public int run(CommandLine cl) throws AlluxioException, IOException {
    String[] args = cl.getArgs();
    AlluxioURI srcPath = new AlluxioURI(args[0]);
    AlluxioURI dstPath = new AlluxioURI(args[1]);
    if ((dstPath.getScheme() == null || isAlluxio(dstPath.getScheme()))
        && isFile(srcPath.getScheme())) {
      // If the dstPath ends with "/" but does not exist, then the dstPath should be directory
      // and must be created first
      boolean dstExist = mFileSystem.exists(dstPath);
      if (!dstExist && args[1].endsWith(AlluxioURI.SEPARATOR)) {
        throw new IOException(
                ExceptionMessage.CANNOT_COPY_TO_NOT_EXIST_DIRECTORY.getMessage(dstPath.getPath()));
      }
      if (dstExist && args[1].endsWith(AlluxioURI.SEPARATOR)
              && !mFileSystem.getStatus(dstPath).isFolder()) {
        throw new IOException(ExceptionMessage.FILE_TYPE_NOT_MATCH.getMessage(dstPath.getPath()));
      }
      List<AlluxioURI> srcPaths = new ArrayList<>();
      if (srcPath.containsWildcard()) {
        List<File> srcFiles = FileSystemShellUtils.getFiles(srcPath.getPath());
        if (srcFiles.size() == 0) {
          throw new IOException(ExceptionMessage.PATH_DOES_NOT_EXIST.getMessage(srcPath));
        }
        for (File srcFile : srcFiles) {
          srcPaths.add(
              new AlluxioURI(srcPath.getScheme(), srcPath.getAuthority(), srcFile.getPath()));
        }
      } else {
        File src = new File(srcPath.getPath());
        if (src.isDirectory()) {
          File[] files = src.listFiles();
          if (files == null) {
            throw new IOException(String.format("Failed to list files for directory %s", src));
          }
          for (File f : files) {
            srcPaths.add(new AlluxioURI(srcPath.getScheme(), srcPath.getAuthority(), f.getPath()));
          }
        } else {
          srcPaths.add(srcPath);
        }
      }
      if (srcPaths.size() == 1 && !(new File(srcPaths.get(0).getPath())).isDirectory()) {
        copyFromLocalFile(srcPaths.get(0), dstPath, dstExist);
      } else {
        CopyThreadPoolExecutor pool = new CopyThreadPoolExecutor(mThread, System.out, System.err,
            mFileSystem, mFileSystem.exists(dstPath) ? null : dstPath);
        try {
          createDstDir(dstPath);
          for (AlluxioURI src : srcPaths) {
            AlluxioURI dst = new AlluxioURI(dstPath, new AlluxioURI(src.getName()));
            asyncCopyLocalPath(pool, src, dst);
          }
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          throw new RuntimeException(e);
        } finally {
          pool.shutdown();
        }
      }
      System.out.println(String.format(COPY_SUCCEED_MESSAGE, srcPath, dstPath));
    } else if ((srcPath.getScheme() == null || isAlluxio(srcPath.getScheme()))
        && isFile(dstPath.getScheme())) {
      List<AlluxioURI> srcPaths = FileSystemShellUtils.getAlluxioURIs(mFileSystem, srcPath);
      if (srcPaths.size() == 0) {
        throw new IOException(ExceptionMessage.PATH_DOES_NOT_EXIST.getMessage(srcPath));
      }
      if (srcPath.containsWildcard()) {
        copyWildcardToLocal(srcPaths, dstPath);
      } else {
        copyToLocal(srcPath, dstPath);
      }
    } else if ((srcPath.getScheme() == null || isAlluxio(srcPath.getScheme()))
        && (dstPath.getScheme() == null || isAlluxio(dstPath.getScheme()))) {
      List<AlluxioURI> srcPaths = FileSystemShellUtils.getAlluxioURIs(mFileSystem, srcPath);
      if (srcPaths.size() == 0) {
        throw new FileDoesNotExistException(
            ExceptionMessage.PATH_DOES_NOT_EXIST.getMessage(srcPath.getPath()));
      }
      boolean recursive = cl.hasOption(RECURSIVE_OPTION.getOpt())
          || cl.hasOption(RECURSIVE_ALIAS_OPTION.getOpt());
      if (srcPath.containsWildcard()) {
        copyWildcard(srcPaths, dstPath, recursive);
      } else {
        copy(srcPath, dstPath, recursive);
      }
    } else {
      throw new InvalidPathException(
          "Schemes must be either file or alluxio, and at most one file scheme is allowed.");
    }
    return 0;
  }

  /**
   * Copies a list of files or directories specified by srcPaths to the destination specified by
   * dstPath. This method is used when the original source path contains wildcards.
   *
   * @param srcPaths a list of files or directories in the Alluxio filesystem
   * @param dstPath the destination in the Alluxio filesystem
   * @param recursive indicates whether directories should be copied recursively
   */
  private void copyWildcard(List<AlluxioURI> srcPaths, AlluxioURI dstPath, boolean recursive)
      throws AlluxioException, IOException {
    URIStatus dstStatus = null;
    try {
      dstStatus = mFileSystem.getStatus(dstPath);
    } catch (FileDoesNotExistException e) {
      // if the destination does not exist, it will be created
    }

    if (dstStatus != null && !dstStatus.isFolder()) {
      throw new InvalidPathException(ExceptionMessage.DESTINATION_CANNOT_BE_FILE.getMessage());
    }
    if (dstStatus == null) {
      mFileSystem.createDirectory(dstPath);
      System.out.println("Created directory: " + dstPath);
    }
    List<String> errorMessages = new ArrayList<>();
    for (AlluxioURI srcPath : srcPaths) {
      try {
        copy(srcPath, new AlluxioURI(dstPath.getScheme(), dstPath.getAuthority(),
            PathUtils.concatPath(dstPath.getPath(), srcPath.getName())), recursive);
      } catch (AlluxioException | IOException e) {
        errorMessages.add(e.getMessage());
      }
    }
    if (errorMessages.size() != 0) {
      throw new IOException(Joiner.on('\n').join(errorMessages));
    }
  }

  /**
   * Copies a file or a directory in the Alluxio filesystem.
   *
   * @param srcPath the source {@link AlluxioURI} (could be a file or a directory)
   * @param dstPath the {@link AlluxioURI} of the destination path in the Alluxio filesystem
   * @param recursive indicates whether directories should be copied recursively
   */
  private void copy(AlluxioURI srcPath, AlluxioURI dstPath, boolean recursive)
      throws AlluxioException, IOException {
    URIStatus srcStatus = mFileSystem.getStatus(srcPath);

    URIStatus dstStatus = null;
    try {
      dstStatus = mFileSystem.getStatus(dstPath);
    } catch (FileDoesNotExistException e) {
      // if the destination does not exist, it will be created
    }

    if (!srcStatus.isFolder()) {
      if (dstStatus != null && dstStatus.isFolder()) {
        dstPath = new AlluxioURI(PathUtils.concatPath(dstPath.getPath(), srcPath.getName()));
      }
      copyFile(srcPath, dstPath);
    } else {
      if (!recursive) {
        throw new IOException(srcPath.getPath() + " is a directory,"
            + " to copy it please use \"cp -R/-r/--recursive <src> <dst>\"");
      }

      List<URIStatus> statuses;
      statuses = mFileSystem.listStatus(srcPath);

      if (dstStatus != null) {
        if (!dstStatus.isFolder()) {
          throw new InvalidPathException(ExceptionMessage.DESTINATION_CANNOT_BE_FILE.getMessage());
        }
        // if copying a directory to an existing directory, the copied directory will become a
        // subdirectory of the destination
        if (srcStatus.isFolder()) {
          dstPath = new AlluxioURI(PathUtils.concatPath(dstPath.getPath(), srcPath.getName()));
          mFileSystem.createDirectory(dstPath);
          System.out.println("Created directory: " + dstPath);
        }
      }

      if (dstStatus == null) {
        mFileSystem.createDirectory(dstPath);
        System.out.println("Created directory: " + dstPath);
      }

      preserveAttributes(srcPath, dstPath);
      List<String> errorMessages = new ArrayList<>();
      for (URIStatus status : statuses) {
        try {
          copy(new AlluxioURI(srcPath.getScheme(), srcPath.getAuthority(), status.getPath()),
              new AlluxioURI(dstPath.getScheme(), dstPath.getAuthority(),
                  PathUtils.concatPath(dstPath.getPath(), status.getName())), recursive);
        } catch (IOException e) {
          errorMessages.add(e.getMessage());
        }
      }

      if (errorMessages.size() != 0) {
        throw new IOException(Joiner.on('\n').join(errorMessages));
      }
    }
  }

  /**
   * Copies a file in the Alluxio filesystem.
   *
   * @param srcPath the source {@link AlluxioURI} (has to be a file)
   * @param dstPath the destination path in the Alluxio filesystem
   */
  private void copyFile(AlluxioURI srcPath, AlluxioURI dstPath)
      throws AlluxioException, IOException {
    try (FileInStream is = mFileSystem.openFile(srcPath);
         FileOutStream os = mFileSystem.createFile(dstPath)) {
      try {
        IOUtils.copyLarge(is, os, new byte[8 * Constants.MB]);
      } catch (Exception e) {
        os.cancel();
        // clean up the incomplete file
        mFileSystem.delete(dstPath, DeletePOptions.newBuilder().setUnchecked(true).build());
        throw e;
      }
      System.out.println(String.format(COPY_SUCCEED_MESSAGE, srcPath, dstPath));
    }
    preserveAttributes(srcPath, dstPath);
  }

  /**
   * Preserves attributes from the source file to the target file.
   *
   * @param srcPath the source path
   * @param dstPath the destination path in the Alluxio filesystem
   */
  private void preserveAttributes(AlluxioURI srcPath, AlluxioURI dstPath)
      throws IOException, AlluxioException {
    if (mPreservePermissions) {
      URIStatus srcStatus = mFileSystem.getStatus(srcPath);
      mFileSystem.setAttribute(dstPath, SetAttributePOptions.newBuilder()
          .setOwner(srcStatus.getOwner())
          .setGroup(srcStatus.getGroup())
          .setMode(new Mode((short) srcStatus.getMode()).toProto())
          .build());
      mFileSystem.setAcl(dstPath, SetAclAction.REPLACE, srcStatus.getAcl().getEntries());
    }
  }

  /**
   * Creates a directory in the Alluxio filesystem space. It will not throw any exception if the
   * destination directory already exists.
   *
   * @param dstPath the {@link AlluxioURI} of the destination directory which will be created
   */
  private void createDstDir(AlluxioURI dstPath) throws AlluxioException, IOException {
    try {
      mFileSystem.createDirectory(dstPath);
    } catch (FileAlreadyExistsException e) {
      // it's fine if the directory already exists
    }

    URIStatus dstStatus = mFileSystem.getStatus(dstPath);
    if (!dstStatus.isFolder()) {
      throw new InvalidPathException(ExceptionMessage.DESTINATION_CANNOT_BE_FILE.getMessage());
    }
  }

  private void copyFromLocalFile(AlluxioURI srcPath, AlluxioURI dstPath)
          throws AlluxioException, IOException {
    copyFromLocalFile(srcPath, dstPath, mFileSystem.exists(dstPath));
  }

  private void copyFromLocalFile(AlluxioURI srcPath, AlluxioURI dstPath, boolean dstExist)
      throws AlluxioException, IOException {
    File src = new File(srcPath.getPath());
    if (src.isDirectory()) {
      throw new IOException("Source " + src.getAbsolutePath() + " is not a file.");
    }

    // If the dstPath is a directory, then it should be updated to be the path of the file where
    // src will be copied to.
    if (dstExist && mFileSystem.getStatus(dstPath).isFolder()) {
      dstPath = dstPath.join(src.getName());
    }

    FileOutStream os = null;
    try (Closer closer = Closer.create()) {
      os = closer.register(mFileSystem.createFile(dstPath));
      FileInputStream in = closer.register(new FileInputStream(src));
      FileChannel channel = closer.register(in.getChannel());
      ByteBuffer buf = ByteBuffer.allocate(mCopyFromLocalBufferSize);
      while (channel.read(buf) != -1) {
        buf.flip();
        os.write(buf.array(), 0, buf.limit());
      }
    } catch (Exception e) {
      // Close the out stream and delete the file, so we don't have an incomplete file lying
      // around.
      if (os != null) {
        os.cancel();
        if (mFileSystem.exists(dstPath)) {
          mFileSystem.delete(dstPath);
        }
      }
      throw e;
    }
  }

  /**
   * Asynchronously copies a file or directory specified by srcPath from the local filesystem to
   * dstPath in the Alluxio filesystem space, assuming dstPath does not exist.
   *
   * @param srcPath the {@link AlluxioURI} of the source file in the local filesystem
   * @param dstPath the {@link AlluxioURI} of the destination
   * @throws InterruptedException when failed to send messages to the pool
   */
  private void asyncCopyLocalPath(CopyThreadPoolExecutor pool, AlluxioURI srcPath,
      AlluxioURI dstPath) throws InterruptedException {
    File src = new File(srcPath.getPath());
    if (!src.isDirectory()) {
      pool.submit(() -> {
        try {
          copyFromLocalFile(srcPath, dstPath);
          pool.succeed(srcPath, dstPath);
        } catch (Exception e) {
          pool.fail(srcPath, dstPath, e);
        }
        return null;
      });
    } else {
      try {
        mFileSystem.createDirectory(dstPath);
      } catch (Exception e) {
        pool.fail(srcPath, dstPath, e);
        return;
      }
      File[] fileList = src.listFiles();
      if (fileList == null) {
        pool.fail(srcPath, dstPath,
            new IOException(String.format("Failed to list directory %s.", src)));
        return;
      }
      for (File srcFile : fileList) {
        AlluxioURI newURI = new AlluxioURI(dstPath, new AlluxioURI(srcFile.getName()));
        asyncCopyLocalPath(pool,
              new AlluxioURI(srcPath.getScheme(), srcPath.getAuthority(), srcFile.getPath()),
              newURI);
      }
    }
  }

  /**
   * Copies a list of files or directories specified by srcPaths from the Alluxio filesystem to
   * dstPath in the local filesystem. This method is used when the input path contains wildcards.
   *
   * @param srcPaths the list of files in the Alluxio filesystem
   * @param dstPath the {@link AlluxioURI} of the destination directory in the local filesystem
   */
  private void copyWildcardToLocal(List<AlluxioURI> srcPaths, AlluxioURI dstPath)
      throws AlluxioException, IOException {
    File dstFile = new File(dstPath.getPath());
    if (dstFile.exists() && !dstFile.isDirectory()) {
      throw new InvalidPathException(ExceptionMessage.DESTINATION_CANNOT_BE_FILE.getMessage());
    }
    if (!dstFile.exists()) {
      if (!dstFile.mkdirs()) {
        throw new IOException("Fail to create directory: " + dstPath);
      } else {
        System.out.println("Create directory: " + dstPath);
      }
    }
    List<String> errorMessages = new ArrayList<>();
    for (AlluxioURI srcPath : srcPaths) {
      try {
        File dstSubFile = new File(dstFile.getAbsoluteFile(), srcPath.getName());
        copyToLocal(srcPath,
            new AlluxioURI(dstPath.getScheme(), dstPath.getAuthority(), dstSubFile.getPath()));
      } catch (IOException e) {
        errorMessages.add(e.getMessage());
      }
    }
    if (errorMessages.size() != 0) {
      throw new IOException(Joiner.on('\n').join(errorMessages));
    }
  }

  /**
   * Copies a file or a directory from the Alluxio filesystem to the local filesystem.
   *
   * @param srcPath the source {@link AlluxioURI} (could be a file or a directory)
   * @param dstPath the {@link AlluxioURI} of the destination in the local filesystem
   */
  private void copyToLocal(AlluxioURI srcPath, AlluxioURI dstPath) throws AlluxioException,
      IOException {
    URIStatus srcStatus = mFileSystem.getStatus(srcPath);
    File dstFile = new File(dstPath.getPath());
    if (srcStatus.isFolder()) {
      // make a local directory
      if (!dstFile.exists()) {
        if (!dstFile.mkdirs()) {
          throw new IOException("mkdir failure for directory: " + dstPath);
        } else {
          System.out.println("Create directory: " + dstPath);
        }
      }

      List<URIStatus> statuses;
      try {
        statuses = mFileSystem.listStatus(srcPath);
      } catch (AlluxioException e) {
        throw new IOException(e.getMessage());
      }

      List<String> errorMessages = new ArrayList<>();
      for (URIStatus status : statuses) {
        try {
          File subDstFile = new File(dstFile.getAbsolutePath(), status.getName());
          copyToLocal(
              new AlluxioURI(srcPath.getScheme(), srcPath.getAuthority(), status.getPath()),
              new AlluxioURI(dstPath.getScheme(), dstPath.getAuthority(), subDstFile.getPath()));
        } catch (IOException e) {
          errorMessages.add(e.getMessage());
        }
      }

      if (errorMessages.size() != 0) {
        throw new IOException(Joiner.on('\n').join(errorMessages));
      }
    } else {
      copyFileToLocal(srcPath, dstPath);
    }
  }

  /**
   * Copies a file specified by argv from the filesystem to the local filesystem. This is the
   * utility function.
   *
   * @param srcPath The source {@link AlluxioURI} (has to be a file)
   * @param dstPath The {@link AlluxioURI} of the destination in the local filesystem
   */
  private void copyFileToLocal(AlluxioURI srcPath, AlluxioURI dstPath)
      throws AlluxioException, IOException {
    File dstFile = new File(dstPath.getPath());
    String randomSuffix =
        String.format(".%s_copyToLocal_", RandomStringUtils.randomAlphanumeric(8));
    File outputFile;
    if (dstFile.isDirectory()) {
      outputFile = new File(PathUtils.concatPath(dstFile.getAbsolutePath(), srcPath.getName()));
    } else {
      outputFile = dstFile;
    }
    File tmpDst = new File(outputFile.getPath() + randomSuffix);

    try (Closer closer = Closer.create()) {
      FileInStream is = closer.register(mFileSystem.openFile(srcPath));
      FileOutputStream out = closer.register(new FileOutputStream(tmpDst));
      byte[] buf = new byte[mCopyToLocalBufferSize];
      int t = is.read(buf);
      while (t != -1) {
        out.write(buf, 0, t);
        t = is.read(buf);
      }
      if (!tmpDst.renameTo(outputFile)) {
        throw new IOException(
            "Failed to rename " + tmpDst.getPath() + " to destination " + outputFile.getPath());
      }
      System.out.println("Copied " + srcPath + " to " + "file://" + outputFile.getPath());
    } finally {
      tmpDst.delete();
    }
  }

  @Override
  public String getUsage() {
    return "cp "
        + "[-R/-r/--recursive] "
        + "[--buffersize <bytes>] "
        + "<src> <dst>";
  }

  @Override
  public String getDescription() {
    return "Copies a file or a directory in the Alluxio filesystem or between local filesystem "
        + "and Alluxio filesystem. The -R/-r/--recursive flags are needed to copy"
        + "directories in the Alluxio filesystem. Local Path with schema \"file\".";
  }

  private static boolean isAlluxio(String scheme) {
    return Constants.SCHEME.equals(scheme);
  }

  private static boolean isFile(String scheme) {
    return "file".equals(scheme);
  }
}
