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

import static org.junit.Assert.assertEquals;

import alluxio.AlluxioURI;
import alluxio.collections.Pair;
import alluxio.concurrent.jsr.CompletableFuture;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.AccessControlException;
import alluxio.exception.InvalidPathException;
import alluxio.file.options.DescendantType;
import alluxio.grpc.FileSystemMasterCommonPOptions;
import alluxio.master.file.meta.InodeTree;
import alluxio.master.file.meta.LockingScheme;
import alluxio.underfs.UnderFileSystem;

import com.google.common.math.IntMath;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;

import java.util.concurrent.ExecutionException;
import java.util.function.Supplier;

@RunWith(PowerMockRunner.class)
@PrepareForTest({UnderFileSystem.Factory.class})
public class FileSystemMasterSyncMetadataConcurrentTest
    extends FileSystemMasterSyncMetadataTestBase {
  private final int mNumDirsPerLevel = 2;
  private final int mNumLevels = 3;
  private final int mNumExpectedInodes = IntMath.pow(mNumDirsPerLevel, mNumLevels + 1) - 1;

  @Override
  public void before() throws Exception {
    super.before();
    Configuration.set(PropertyKey.MASTER_METADATA_CONCURRENT_SYNC_DEDUP, true);

    createUfsHierarchy(0, mNumLevels, "", mNumDirsPerLevel);

    mUfs.mIsSlow = true;
    mUfs.mSlowTimeMs = 500;
    // verify the files don't exist in alluxio
    assertEquals(1, mFileSystemMaster.getInodeTree().getInodeCount());
  }

  @Test
  public void loadMetadataForTheSameDirectory() throws Exception {
    InodeSyncStream iss1 = makeInodeSyncStream("/", false, true, -1);
    InodeSyncStream iss2 = makeInodeSyncStream("/", false, true, -1);
    assertTheSecondSyncSkipped(syncConcurrent(iss1, iss2));
    // Only load 1 level metadata
    assertEquals(1 + mNumDirsPerLevel, mFileSystemMaster.getInodeTree().getInodeCount());

    iss1 = makeInodeSyncStream("/0_0", true, true, -1);
    iss2 = makeInodeSyncStream("/0_1", true, true, -1);
    assertSyncHappenTwice(syncConcurrent(iss1, iss2));
    // Only load 1 level metadata
    assertEquals(mNumExpectedInodes, mFileSystemMaster.getInodeTree().getInodeCount());
  }

  @Test
  public void loadMetadataForDirectoryAndItsSubDirectory()
      throws Exception {
    InodeSyncStream iss1 = makeInodeSyncStream("/", true, true, -1);
    InodeSyncStream iss2 = makeInodeSyncStream("/0_1", false, true, -1);
    assertTheSecondSyncSkipped(syncConcurrent(iss1, iss2));
    assertEquals(mNumExpectedInodes, mFileSystemMaster.getInodeTree().getInodeCount());
  }

  @Test
  public void syncTheSameDirectory() throws Exception {
    InodeSyncStream iss1 = makeInodeSyncStream("/", true, false, 0);
    InodeSyncStream iss2 = makeInodeSyncStream("/", true, false, 0);
    assertTheSecondSyncSkipped(syncConcurrent(iss1, iss2));
    assertEquals(mNumExpectedInodes, mFileSystemMaster.getInodeTree().getInodeCount());
    Supplier<InodeSyncStream> createSync = () -> makeInodeSyncStream("/", true, false, 0);
    assertSyncHappenTwice(syncSequential(createSync, createSync));

    iss1 = makeInodeSyncStream("/0_1", true, false, 0);
    iss2 = makeInodeSyncStream("/0_1", false, false, 0);
    assertTheSecondSyncSkipped(syncConcurrent(iss1, iss2));

    iss1 = makeInodeSyncStream("/0_1", true, false, -1);
    iss2 = makeInodeSyncStream("/0_1", false, false, -1);
    assertSyncNotHappen(syncConcurrent(iss1, iss2));
  }

  @Test
  public void syncDirectoryAndItsSubdirectory() throws Exception {
    InodeSyncStream iss1 = makeInodeSyncStream("/", true, false, 0);
    InodeSyncStream iss2 = makeInodeSyncStream("/0_1", true, false, 0);
    assertTheSecondSyncSkipped(syncConcurrent(iss1, iss2));
    assertEquals(mNumExpectedInodes, mFileSystemMaster.getInodeTree().getInodeCount());
    assertSyncHappenTwice(syncSequential(
        () -> makeInodeSyncStream("/", true, false, 0),
        () -> makeInodeSyncStream("/0_1", true, false, 0)));

    iss1 = makeInodeSyncStream("/", false, false, 0);
    iss2 = makeInodeSyncStream("/0_1", true, false, 0);
    assertSyncHappenTwice(syncConcurrent(iss1, iss2));

    iss1 = makeInodeSyncStream("/0_1", true, false, 0);
    iss2 = makeInodeSyncStream("/", true, false, 0);
    assertSyncHappenTwice(syncConcurrent(iss1, iss2));

    iss1 = makeInodeSyncStream("/0_1", true, false, -1);
    iss2 = makeInodeSyncStream("/", true, false, -1);
    assertSyncNotHappen(syncConcurrent(iss1, iss2));
  }

  @Test
  public void syncDifferentDirectories() throws Exception {
    InodeSyncStream iss1 = makeInodeSyncStream("/0_0", true, true, 0);
    InodeSyncStream iss2 = makeInodeSyncStream("/0_1", true, true, 0);
    assertSyncHappenTwice(syncConcurrent(iss1, iss2));

    iss1 = makeInodeSyncStream("/0_0", true, false, 0);
    iss2 = makeInodeSyncStream("/0_1", true, false, 0);
    assertSyncHappenTwice(syncConcurrent(iss1, iss2));
    assertEquals(mNumExpectedInodes, mFileSystemMaster.getInodeTree().getInodeCount());
  }

  /**
   * To test if the metadata sync cancellation will result in deadlock.
   */
  @Test
  public void syncTheSameDirectoryButTheSecondCallCancelled() throws Exception {
    InodeSyncStream iss1 = makeInodeSyncStream("/", true, false, 0);
    InodeSyncStream iss2 = makeInodeSyncStream("/", true, false, 0);
    CompletableFuture<InodeSyncStream.SyncStatus> f1 = CompletableFuture.supplyAsync(() -> {
      try {
        return iss1.sync();
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    });
    Thread.sleep(10);
    CompletableFuture<InodeSyncStream.SyncStatus> f2 = CompletableFuture.supplyAsync(() -> {
      try {
        return iss2.sync();
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    });
    Thread.sleep(100);
    f2.cancel(true);
    f1.get();
    InodeSyncStream iss3 = makeInodeSyncStream("/", true, false, 0);
    assertEquals(InodeSyncStream.SyncStatus.OK, iss3.sync());
  }

  @Test
  public void syncWhenShouldSyncIsSetTrue() throws Exception {
    Supplier<InodeSyncStream> inodeSyncStreamSupplier =  () -> new InodeSyncStream(
        new LockingScheme(
            new AlluxioURI("/"), InodeTree.LockPattern.READ, true),
        mFileSystemMaster, mFileSystemMaster.getSyncPathCache(),
        RpcContext.NOOP, DescendantType.ALL, FileSystemMasterCommonPOptions.getDefaultInstance(),
        false,
        false,
        false);

    InodeSyncStream iss1 = inodeSyncStreamSupplier.get();
    InodeSyncStream iss2 = inodeSyncStreamSupplier.get();
    assertSyncHappenTwice(syncConcurrent(iss1, iss2));
    assertSyncHappenTwice(syncSequential(inodeSyncStreamSupplier, inodeSyncStreamSupplier));
  }

  private void assertTheSecondSyncSkipped(
      Pair<InodeSyncStream.SyncStatus, InodeSyncStream.SyncStatus> results) {
    assertEquals(InodeSyncStream.SyncStatus.OK, results.getFirst());
    assertEquals(InodeSyncStream.SyncStatus.NOT_NEEDED, results.getSecond());
  }

  private void assertSyncHappenTwice(
      Pair<InodeSyncStream.SyncStatus, InodeSyncStream.SyncStatus> results) {
    assertEquals(InodeSyncStream.SyncStatus.OK, results.getFirst());
    assertEquals(InodeSyncStream.SyncStatus.OK, results.getSecond());
  }

  private void assertSyncNotHappen(
      Pair<InodeSyncStream.SyncStatus, InodeSyncStream.SyncStatus> results) {
    assertEquals(InodeSyncStream.SyncStatus.NOT_NEEDED, results.getFirst());
    assertEquals(InodeSyncStream.SyncStatus.NOT_NEEDED, results.getSecond());
  }

  private InodeSyncStream makeInodeSyncStream(
      String path, boolean isRecursive, boolean forceSync, long syncInterval) {
    FileSystemMasterCommonPOptions options = FileSystemMasterCommonPOptions.newBuilder()
        .setSyncIntervalMs(syncInterval)
        .build();
    DescendantType descendantType = isRecursive ? DescendantType.ALL : DescendantType.ONE;
    try {
      LockingScheme syncScheme =
          new LockingScheme(new AlluxioURI(path), InodeTree.LockPattern.READ, options,
              mFileSystemMaster.getSyncPathCache(), descendantType); // shouldSync
      return
          new InodeSyncStream(syncScheme, mFileSystemMaster, mFileSystemMaster.getSyncPathCache(),
          RpcContext.NOOP, descendantType, options,
          forceSync,
          forceSync,
          forceSync);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  private Pair<InodeSyncStream.SyncStatus, InodeSyncStream.SyncStatus> syncConcurrent(
      InodeSyncStream iss1, InodeSyncStream iss2)
      throws ExecutionException, InterruptedException {
    Thread.sleep(10);
    CompletableFuture<InodeSyncStream.SyncStatus> f1 = CompletableFuture.supplyAsync(() -> {
      try {
        return iss1.sync();
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    });
    Thread.sleep(100);
    CompletableFuture<InodeSyncStream.SyncStatus> f2 = CompletableFuture.supplyAsync(() -> {
      try {
        return iss2.sync();
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    });
    return new Pair<>(f1.get(), f2.get());
  }

  private Pair<InodeSyncStream.SyncStatus, InodeSyncStream.SyncStatus> syncSequential(
      Supplier<InodeSyncStream> iss1, Supplier<InodeSyncStream> iss2)
      throws AccessControlException, InvalidPathException, InterruptedException {
    Thread.sleep(10);
    InodeSyncStream.SyncStatus result1 = iss1.get().sync();
    Thread.sleep(10);
    InodeSyncStream.SyncStatus result2 = iss2.get().sync();
    return new Pair<>(result1, result2);
  }
}

