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

import alluxio.AlluxioURI;
import alluxio.Constants;
import alluxio.conf.PropertyKey;
import alluxio.client.file.FileInStream;
import alluxio.client.file.FileOutStream;
import alluxio.client.file.FileSystem;
import alluxio.client.file.FileSystemTestUtils;
import alluxio.grpc.CreateFilePOptions;
import alluxio.grpc.OpenFilePOptions;
import alluxio.grpc.ReadPType;
import alluxio.grpc.WritePType;
import alluxio.testutils.BaseIntegrationTest;
import alluxio.testutils.IntegrationTestUtils;
import alluxio.testutils.LocalAlluxioClusterResource;
import alluxio.util.io.BufferUtils;
import alluxio.util.io.PathUtils;

import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Integration tests for reading data which is only stored in Alluxio's under storage.
 */
public class UnderStorageReadIntegrationTest extends BaseIntegrationTest {
  private static final Logger LOG = LoggerFactory.getLogger(UnderStorageReadIntegrationTest.class);
  private static final int MIN_LEN = 0;
  private static final int MAX_LEN = 255;
  private static final int DELTA = 33;

  private FileSystem mFileSystem = null;
  private CreateFilePOptions mWriteUnderStore;
  private OpenFilePOptions mReadNoCache;
  private OpenFilePOptions mReadCache;

  @Rule
  public LocalAlluxioClusterResource mLocalAlluxioClusterResource =
      new LocalAlluxioClusterResource.Builder()
          .setProperty(PropertyKey.USER_BLOCK_REMOTE_READ_BUFFER_SIZE_BYTES, "100")
          .setProperty(PropertyKey.USER_UFS_BLOCK_READ_CONCURRENCY_MAX, 2).build();

  @Rule
  public ExpectedException mThrown = ExpectedException.none();

  @Before
  public final void before() throws Exception {
    mFileSystem = mLocalAlluxioClusterResource.get().getClient();
    mWriteUnderStore = CreateFilePOptions.newBuilder().setWriteType(WritePType.THROUGH)
        .setRecursive(true).build();
    mReadCache = OpenFilePOptions.newBuilder().setReadType(ReadPType.CACHE_PROMOTE).build();
    mReadNoCache = OpenFilePOptions.newBuilder().setReadType(ReadPType.NO_CACHE).build();
  }

  /**
   * Tests single byte reads from the underfs.
   */
  @Test
  public void read() throws Exception {
    String uniqPath = PathUtils.uniqPath();
    for (int k = MIN_LEN; k <= MAX_LEN; k += DELTA) {
      AlluxioURI uri = new AlluxioURI(uniqPath + "/file_" + k);
      FileSystemTestUtils.createByteFile(mFileSystem, uri, mWriteUnderStore, k);

      FileInStream is = mFileSystem.openFile(uri, mReadNoCache);
      byte[] ret = new byte[k];
      int value = is.read();
      int cnt = 0;
      while (value != -1) {
        Assert.assertTrue(value >= 0);
        Assert.assertTrue(value < 256);
        ret[cnt++] = (byte) value;
        value = is.read();
      }
      Assert.assertEquals(cnt, k);
      Assert.assertTrue(BufferUtils.equalIncreasingByteArray(k, ret));
      is.close();
      if (k == 0) {
        Assert.assertEquals(100, mFileSystem.getStatus(uri).getInAlluxioPercentage());
      } else {
        Assert.assertNotEquals(100, mFileSystem.getStatus(uri).getInAlluxioPercentage());
      }

      is = mFileSystem.openFile(uri, mReadCache);
      ret = new byte[k];
      value = is.read();
      cnt = 0;
      while (value != -1) {
        Assert.assertTrue(value >= 0);
        Assert.assertTrue(value < 256);
        ret[cnt++] = (byte) value;
        value = is.read();
      }
      Assert.assertEquals(cnt, k);
      Assert.assertTrue(BufferUtils.equalIncreasingByteArray(k, ret));
      is.close();
      Assert.assertEquals(100, mFileSystem.getStatus(uri).getInAlluxioPercentage());

      is = mFileSystem.openFile(uri, mReadCache);
      ret = new byte[k];
      value = is.read();
      cnt = 0;
      while (value != -1) {
        Assert.assertTrue(value >= 0);
        Assert.assertTrue(value < 256);
        ret[cnt++] = (byte) value;
        value = is.read();
      }
      Assert.assertEquals(cnt, k);
      Assert.assertTrue(BufferUtils.equalIncreasingByteArray(k, ret));
      is.close();
      Assert.assertEquals(100, mFileSystem.getStatus(uri).getInAlluxioPercentage());
    }
  }

  /**
   * Tests single byte reads from the underfs concurrently.
   */
  @Test
  public void concurrentUfsRead() throws Exception {
    String uniqPath = PathUtils.uniqPath();
    final AlluxioURI uri = new AlluxioURI(uniqPath + "/file_" + MAX_LEN);
    FileSystemTestUtils.createByteFile(mFileSystem, uri, mWriteUnderStore, MAX_LEN);

    ExecutorService executorService = Executors.newFixedThreadPool(100);

    final AtomicInteger count = new AtomicInteger(0);
    final Random random = new Random();
    int expectedCount = 100;
    for (int i = 0; i < expectedCount; i++) {
      final int index = i;
      executorService.submit(new Runnable() {
        @Override
        public void run() {
          try {
            // Add some randomness here to avoid opening too many connections too fast.
            Thread.sleep(random.nextInt(100));
            FileInStream is = mFileSystem.openFile(uri, mReadCache);

            // Sleep here so that we can make sure the file is not cached too fast.
            Thread.sleep(100);
            byte[] ret = new byte[MAX_LEN];
            int value = is.read();
            int cnt = 0;
            while (value != -1) {
              Assert.assertTrue(value >= 0);
              Assert.assertTrue(value < 256);
              ret[cnt++] = (byte) value;
              value = is.read();
            }
            is.close();
            Assert.assertEquals(cnt, MAX_LEN);
            Assert.assertTrue(BufferUtils.equalIncreasingByteArray(MAX_LEN, ret));
            while (mFileSystem.getStatus(uri).getInAlluxioPercentage() < 100) {
              Thread.sleep(1000);
            }
            Assert.assertEquals(100, mFileSystem.getStatus(uri).getInAlluxioPercentage());
            count.incrementAndGet();
          } catch (Throwable e) {
            LOG.error("Failed to read file {}.", index, e);
          }
        }
      });
    }
    executorService.shutdown();
    executorService.awaitTermination(Constants.MINUTE_MS * 5, TimeUnit.MILLISECONDS);
    // There can be some errors due because SASL handsake failure or opening the connections too
    // fast. If that happens, consider loosing the condition a bit.
    Assert.assertTrue(expectedCount >= count.get() - 1);
  }

  /**
   * Tests seeking through files only in the underfs.
   */
  @Test
  public void seek() throws Exception {
    String uniqPath = PathUtils.uniqPath();
    for (int k = MIN_LEN + DELTA; k <= MAX_LEN; k += DELTA) {
      AlluxioURI uri = new AlluxioURI(uniqPath + "/file_" + k);
      FileSystemTestUtils.createByteFile(mFileSystem, uri, mWriteUnderStore, k);

      FileInStream is = mFileSystem.openFile(uri, mReadCache);

      Assert.assertEquals(0, is.read());
      is.seek(k / 3);
      Assert.assertEquals(k / 3, is.read());
      is.seek(k / 2);
      Assert.assertEquals(k / 2, is.read());
      is.seek(k / 4);
      Assert.assertEquals(k / 4, is.read());
      is.close();
    }
  }

  /**
   * Tests skipping through files only in the underfs.
   */
  @Test
  public void skip() throws Exception {
    String uniqPath = PathUtils.uniqPath();
    for (int k = MIN_LEN + DELTA; k <= MAX_LEN; k += DELTA) {
      AlluxioURI uri = new AlluxioURI(uniqPath + "/file_" + k);
      FileSystemTestUtils.createByteFile(mFileSystem, uri, mWriteUnderStore, k);

      FileInStream is = mFileSystem.openFile(uri, mReadCache);
      Assert.assertEquals(k / 2, is.skip(k / 2));
      Assert.assertEquals(k / 2, is.read());
      is.close();
      IntegrationTestUtils.waitForFileCached(mFileSystem, uri, 1000);

      if (k >= 3) {
        is = mFileSystem.openFile(uri, mReadCache);
        int t = k / 3;
        Assert.assertEquals(t, is.skip(t));
        Assert.assertEquals(t, is.read());
        Assert.assertEquals(t, is.skip(t));
        Assert.assertEquals(2 * t + 1, is.read());
        is.close();
        IntegrationTestUtils.waitForFileCached(mFileSystem, uri, 1000);
      }
    }
  }

  /**
   * Tests that reading a file consisting of more than one block from the underfs works.
   */
  @Test
  public void readMultiBlockFile() throws Exception {
    String uniqPath = PathUtils.uniqPath();
    int blockSizeByte = 10;
    int numBlocks = 10;
    AlluxioURI uri = new AlluxioURI(uniqPath);
    FileOutStream os = mFileSystem.createFile(uri, mWriteUnderStore);
    for (int i = 0; i < numBlocks; i++) {
      for (int j = 0; j < blockSizeByte; j++) {
        os.write((byte) (i * blockSizeByte + j));
      }
    }
    os.close();

    FileInStream is = mFileSystem.openFile(uri, mReadCache);
    for (int i = 0; i < blockSizeByte * numBlocks; i++) {
      Assert.assertEquals((byte) i, is.read());
    }
    is.close();
    Assert.assertTrue(mFileSystem.getStatus(uri).getInAlluxioPercentage() == 100);
  }
}
