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

import static org.junit.Assert.fail;

import alluxio.AlluxioURI;
import alluxio.Constants;
import alluxio.PositionReader;
import alluxio.client.file.CacheContext;
import alluxio.client.file.FileInStream;
import alluxio.client.file.FileOutStream;
import alluxio.client.file.FileSystem;
import alluxio.client.file.ListStatusPartialResult;
import alluxio.client.file.MockFileInStream;
import alluxio.client.file.URIStatus;
import alluxio.client.file.cache.context.CachePerThreadContext;
import alluxio.conf.AlluxioConfiguration;
import alluxio.conf.Configuration;
import alluxio.conf.InstancedConfiguration;
import alluxio.conf.PropertyKey;
import alluxio.exception.AlluxioException;
import alluxio.exception.DirectoryNotEmptyException;
import alluxio.exception.FileAlreadyExistsException;
import alluxio.exception.FileDoesNotExistException;
import alluxio.exception.FileIncompleteException;
import alluxio.exception.InvalidPathException;
import alluxio.exception.OpenDirectoryException;
import alluxio.file.ReadTargetBuffer;
import alluxio.grpc.CheckAccessPOptions;
import alluxio.grpc.CreateDirectoryPOptions;
import alluxio.grpc.CreateFilePOptions;
import alluxio.grpc.DeletePOptions;
import alluxio.grpc.ExistsPOptions;
import alluxio.grpc.FreePOptions;
import alluxio.grpc.GetStatusPOptions;
import alluxio.grpc.JobProgressReportFormat;
import alluxio.grpc.ListStatusPOptions;
import alluxio.grpc.ListStatusPartialPOptions;
import alluxio.grpc.MountPOptions;
import alluxio.grpc.OpenFilePOptions;
import alluxio.grpc.RenamePOptions;
import alluxio.grpc.ScheduleAsyncPersistencePOptions;
import alluxio.grpc.SetAclAction;
import alluxio.grpc.SetAclPOptions;
import alluxio.grpc.SetAttributePOptions;
import alluxio.grpc.UnmountPOptions;
import alluxio.job.JobDescription;
import alluxio.job.JobRequest;
import alluxio.metrics.MetricKey;
import alluxio.metrics.MetricsSystem;
import alluxio.network.protocol.databuffer.DataFileChannel;
import alluxio.security.authorization.AclEntry;
import alluxio.util.io.BufferUtils;
import alluxio.util.io.PathUtils;
import alluxio.wire.BlockLocationInfo;
import alluxio.wire.FileInfo;
import alluxio.wire.MountPointInfo;

import com.google.common.base.Stopwatch;
import com.google.common.base.Ticker;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import com.google.common.io.ByteStreams;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameter;
import org.junit.runners.Parameterized.Parameters;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * Unit tests for {@link LocalCacheFileInStream}.
 */
@RunWith(Parameterized.class)
public class LocalCacheFileInStreamTest {
  private static InstancedConfiguration sConf = Configuration.modifiableGlobal();

  @Parameters(name = "{index}: page_size({0}), in_stream_buffer_size({1})")
  public static Collection<Object[]> data() {
    return Arrays.asList(new Object[][] {
        {Constants.MB, 0}, {Constants.MB, 8 * Constants.KB}
    });
  }

  @Parameter(0)
  public int mPageSize;

  @Parameter(1)
  public int mBufferSize;

  @Before
  public void before() {
    MetricsSystem.clearAllMetrics();
    sConf.set(PropertyKey.USER_CLIENT_CACHE_PAGE_SIZE, mPageSize);
    sConf.set(PropertyKey.USER_CLIENT_CACHE_IN_STREAM_BUFFER_SIZE, mBufferSize);
    CachePerThreadContext.get().setCacheEnabled(true);
  }

  @Test
  public void readFullPage() throws Exception {
    int fileSize = mPageSize;
    int bufferSize = fileSize;
    int pages = 1;
    verifyReadFullFile(fileSize, bufferSize, pages);
  }

  @Test
  public void readFullPageThroughReadByteBufferMethod() throws Exception {
    int fileSize = mPageSize;
    int bufferSize = fileSize;
    int pages = 1;
    verifyReadFullFileThroughReadByteBufferMethod(fileSize, bufferSize, pages);
  }

  @Test
  public void readSmallPage() throws Exception {
    int fileSize = mPageSize / 5;
    int bufferSize = fileSize;
    int pages = 1;
    verifyReadFullFile(fileSize, bufferSize, pages);
  }

  @Test
  public void readSmallPageThroughReadByteBufferMethod() throws Exception {
    int fileSize = mPageSize / 5;
    int bufferSize = fileSize;
    int pages = 1;
    verifyReadFullFileThroughReadByteBufferMethod(fileSize, bufferSize, pages);
  }

  @Test
  public void readEmptyFileThroughReadByteBuffer() throws Exception {
    int fileSize = 0;
    byte[] fileData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    LocalCacheFileInStream stream = setupWithSingleFile(fileData, manager);

    byte[] readData = new byte[fileSize];
    ByteBuffer buffer = ByteBuffer.wrap(readData);
    int totalBytesRead = stream.read(buffer, 0, fileSize + 1);
    Assert.assertEquals(-1, totalBytesRead);
  }

  @Test
  public void readThroughReadByteBuffer() throws Exception {
    int fileSize = 10;
    byte[] fileData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    LocalCacheFileInStream stream = setupWithSingleFile(fileData, manager);

    byte[] readData = new byte[fileSize];
    ByteBuffer buffer = ByteBuffer.wrap(readData);
    int totalBytesRead = stream.read(buffer, 0, fileSize + 1);
    Assert.assertEquals(fileSize, totalBytesRead);
  }

  @Test
  public void readPartialPage() throws Exception {
    int fileSize = mPageSize;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    LocalCacheFileInStream stream = setupWithSingleFile(testData, manager);

    int partialReadSize = fileSize / 5;
    int offset = fileSize / 5;

    // cache miss
    byte[] cacheMiss = new byte[partialReadSize];
    stream.seek(offset);
    Assert.assertEquals(partialReadSize, stream.read(cacheMiss));
    Assert.assertArrayEquals(
        Arrays.copyOfRange(testData, offset, offset + partialReadSize), cacheMiss);
    Assert.assertEquals(0, manager.mPagesServed);
    Assert.assertEquals(1, manager.mPagesCached);

    // cache hit
    byte[] cacheHit = new byte[partialReadSize];
    stream.seek(offset);
    Assert.assertEquals(partialReadSize, stream.read(cacheHit));
    Assert.assertArrayEquals(
        Arrays.copyOfRange(testData, offset, offset + partialReadSize), cacheHit);
    Assert.assertEquals(1, manager.mPagesServed);
  }

  @Test
  public void readPartialPageThroughReadByteBufferMethod() throws Exception {
    int fileSize = mPageSize;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    LocalCacheFileInStream stream = setupWithSingleFile(testData, manager);

    int partialReadSize = fileSize / 5;
    int offset = fileSize / 5;

    // cache miss
    ByteBuffer cacheMissBuffer = ByteBuffer.wrap(new byte[partialReadSize]);
    stream.seek(offset);
    Assert.assertEquals(partialReadSize, stream.read(cacheMissBuffer));
    Assert.assertArrayEquals(
        Arrays.copyOfRange(testData, offset, offset + partialReadSize),
        cacheMissBuffer.array());
    Assert.assertEquals(0, manager.mPagesServed);
    Assert.assertEquals(1, manager.mPagesCached);

    // cache hit
    ByteBuffer cacheHitBuffer = ByteBuffer.wrap(new byte[partialReadSize]);
    stream.seek(offset);
    Assert.assertEquals(partialReadSize, stream.read(cacheHitBuffer));
    Assert.assertArrayEquals(Arrays.copyOfRange(testData, offset, offset + partialReadSize),
        cacheHitBuffer.array());
    Assert.assertEquals(1, manager.mPagesServed);
  }

  @Test
  public void readMultiPage() throws Exception {
    int pages = 2;
    int fileSize = mPageSize + 10;
    int bufferSize = fileSize;
    verifyReadFullFile(fileSize, bufferSize, pages);
  }

  @Test
  public void readMultiPageThroughReadByteBufferMethod() throws Exception {
    int pages = 2;
    int fileSize = mPageSize + 10;
    int bufferSize = fileSize;
    verifyReadFullFileThroughReadByteBufferMethod(fileSize, bufferSize, pages);
  }

  @Test
  public void readMultiPageMixed() throws Exception {
    int pages = 10;
    int fileSize = mPageSize * pages;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    LocalCacheFileInStream stream = setupWithSingleFile(testData, manager);

    // populate cache
    int pagesCached = 0;
    for (int i = 0; i < pages; i++) {
      stream.seek(mPageSize * i);
      if (ThreadLocalRandom.current().nextBoolean()) {
        Assert.assertEquals(testData[(i * mPageSize)], stream.read());
        pagesCached++;
      }
    }

    Assert.assertEquals(0, manager.mPagesServed);
    Assert.assertEquals(pagesCached, manager.mPagesCached);

    // sequential read
    stream.seek(0);
    byte[] fullRead = new byte[fileSize];
    Assert.assertEquals(fileSize, stream.read(fullRead));
    Assert.assertArrayEquals(testData, fullRead);
    Assert.assertEquals(pagesCached, manager.mPagesServed);
  }

  @Test
  public void readMultiPageMixedThroughReadByteBufferMethod() throws Exception {
    int pages = 10;
    int fileSize = mPageSize * pages;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    LocalCacheFileInStream stream = setupWithSingleFile(testData, manager);

    // populate cache
    int pagesCached = 0;
    for (int i = 0; i < pages; i++) {
      stream.seek(mPageSize * i);
      if (ThreadLocalRandom.current().nextBoolean()) {
        Assert.assertEquals(testData[(i * mPageSize)], stream.read());
        pagesCached++;
      }
    }

    Assert.assertEquals(0, manager.mPagesServed);
    Assert.assertEquals(pagesCached, manager.mPagesCached);

    // sequential read
    stream.seek(0);
    ByteBuffer fullReadBuf = ByteBuffer.wrap(new byte[fileSize]);
    Assert.assertEquals(fileSize, stream.read(fullReadBuf));
    Assert.assertArrayEquals(testData, fullReadBuf.array());
    Assert.assertEquals(pagesCached, manager.mPagesServed);
  }

  @Test
  public void readOversizedBuffer() throws Exception {
    int pages = 1;
    int fileSize = mPageSize;
    int bufferSize = fileSize * 2;
    verifyReadFullFile(fileSize, bufferSize, pages);
  }

  @Test
  public void readOversizedBufferThroughReadByteBufferMethod() throws Exception {
    int pages = 1;
    int fileSize = mPageSize;
    int bufferSize = fileSize * 2;
    verifyReadFullFileThroughReadByteBufferMethod(fileSize, bufferSize, pages);
  }

  @Test
  public void readSmallPageOversizedBuffer() throws Exception {
    int pages = 1;
    int fileSize = mPageSize / 3;
    int bufferSize = fileSize * 2;
    verifyReadFullFile(fileSize, bufferSize, pages);
  }

  @Test
  public void readSmallPageOversizedBufferThroughReadByteBufferMethod() throws Exception {
    int pages = 1;
    int fileSize = mPageSize / 3;
    int bufferSize = fileSize * 2;
    verifyReadFullFileThroughReadByteBufferMethod(fileSize, bufferSize, pages);
  }

  @Test
  public void positionedReadPartialPage() throws Exception {
    int fileSize = mPageSize;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    LocalCacheFileInStream stream = setupWithSingleFile(testData, manager);

    int partialReadSize = fileSize / 5;
    int offset = fileSize / 5;

    // cache miss
    byte[] cacheMiss = new byte[partialReadSize];
    Assert.assertEquals(partialReadSize,
        stream.positionedRead(offset, cacheMiss, 0, cacheMiss.length));
    Assert.assertArrayEquals(
        Arrays.copyOfRange(testData, offset, offset + partialReadSize), cacheMiss);
    Assert.assertEquals(0, manager.mPagesServed);
    Assert.assertEquals(1, manager.mPagesCached);
    Assert.assertEquals(1,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_EXTERNAL_REQUESTS.getName()).getCount());
    Assert.assertEquals(0,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_HIT_REQUESTS.getName()).getCount());

    // cache hit
    byte[] cacheHit = new byte[partialReadSize];
    Assert.assertEquals(partialReadSize,
        stream.positionedRead(offset, cacheHit, 0, cacheHit.length));
    Assert.assertArrayEquals(
        Arrays.copyOfRange(testData, offset, offset + partialReadSize), cacheHit);
    Assert.assertEquals(1, manager.mPagesServed);
    Assert.assertEquals(1,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_EXTERNAL_REQUESTS.getName()).getCount());
    Assert.assertEquals(1,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_HIT_REQUESTS.getName()).getCount());
  }

  @Test
  public void positionReadOversizedBuffer() throws Exception {
    int fileSize = mPageSize;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    LocalCacheFileInStream stream = setupWithSingleFile(testData, manager);

    // cache miss
    byte[] cacheMiss = new byte[fileSize * 2];
    Assert.assertEquals(fileSize - 1,
        stream.positionedRead(1, cacheMiss, 2, fileSize * 2));
    Assert.assertArrayEquals(
        Arrays.copyOfRange(testData, 1, fileSize - 1),
        Arrays.copyOfRange(cacheMiss, 2, fileSize));
    Assert.assertEquals(0, manager.mPagesServed);
    Assert.assertEquals(1, manager.mPagesCached);

    // cache hit
    byte[] cacheHit = new byte[fileSize * 2];
    Assert.assertEquals(fileSize - 1,
        stream.positionedRead(1, cacheHit, 2, fileSize * 2));
    Assert.assertArrayEquals(
        Arrays.copyOfRange(testData, 1, fileSize - 1),
        Arrays.copyOfRange(cacheHit, 2, fileSize));
    Assert.assertEquals(1, manager.mPagesServed);
  }

  @Test
  public void readPagesMetrics() throws Exception {
    int fileSize = mPageSize * 5;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    LocalCacheFileInStream stream = setupWithSingleFile(testData, manager);

    // cache miss
    int readSize = fileSize - 1;
    byte[] cacheMiss = new byte[readSize];
    stream.read(cacheMiss);
    Assert.assertEquals(0,
        MetricsSystem.meter(MetricKey.CLIENT_CACHE_BYTES_READ_CACHE.getName()).getCount());
    Assert.assertEquals(readSize, MetricsSystem.meter(
        MetricKey.CLIENT_CACHE_BYTES_REQUESTED_EXTERNAL.getName()).getCount());
    Assert.assertEquals(fileSize,
        MetricsSystem.meter(MetricKey.CLIENT_CACHE_BYTES_READ_EXTERNAL.getName()).getCount());
    Assert.assertEquals(5,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_EXTERNAL_REQUESTS.getName()).getCount());
    Assert.assertEquals(0,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_HIT_REQUESTS.getName()).getCount());

    // cache hit
    stream.read();
    Assert.assertEquals(readSize, MetricsSystem.meter(
        MetricKey.CLIENT_CACHE_BYTES_REQUESTED_EXTERNAL.getName()).getCount());
    Assert.assertEquals(fileSize,
        MetricsSystem.meter(MetricKey.CLIENT_CACHE_BYTES_READ_EXTERNAL.getName()).getCount());
    Assert.assertEquals(5,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_EXTERNAL_REQUESTS.getName()).getCount());
    Assert.assertEquals(1,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_HIT_REQUESTS.getName()).getCount());
  }

  @Test
  public void externalStoreMultiRead() throws Exception {
    int fileSize = mPageSize;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    Map<AlluxioURI, byte[]> files = new HashMap<>();
    AlluxioURI testFilename = new AlluxioURI("/test");
    files.put(testFilename, testData);

    ByteArrayFileSystem fs = new MultiReadByteArrayFileSystem(files);

    LocalCacheFileInStream stream = new LocalCacheFileInStream(fs.getStatus(testFilename),
        (status) -> fs.openFile(status, OpenFilePOptions.getDefaultInstance()), manager, sConf,
        Optional.empty());

    // cache miss
    byte[] cacheMiss = new byte[fileSize];
    Assert.assertEquals(fileSize, stream.read(cacheMiss));
    Assert.assertArrayEquals(testData, cacheMiss);
    Assert.assertEquals(0, manager.mPagesServed);
    Assert.assertEquals(1, manager.mPagesCached);

    // cache hit
    stream.seek(0);
    byte[] cacheHit = new byte[fileSize];
    Assert.assertEquals(fileSize, stream.read(cacheHit));
    Assert.assertArrayEquals(testData, cacheHit);
    Assert.assertEquals(1, manager.mPagesServed);
  }

  @Test
  public void externalStoreMultiReadThroughReadByteBufferMethod() throws Exception {
    int fileSize = mPageSize;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    Map<AlluxioURI, byte[]> files = new HashMap<>();
    AlluxioURI testFilename = new AlluxioURI("/test");
    files.put(testFilename, testData);

    ByteArrayFileSystem fs = new MultiReadByteArrayFileSystem(files);

    LocalCacheFileInStream stream = new LocalCacheFileInStream(fs.getStatus(testFilename),
        (status) -> fs.openFile(status, OpenFilePOptions.getDefaultInstance()), manager, sConf,
        Optional.empty());

    // cache miss
    ByteBuffer cacheMissBuf = ByteBuffer.wrap(new byte[fileSize]);
    Assert.assertEquals(fileSize, stream.read(cacheMissBuf));
    Assert.assertArrayEquals(testData, cacheMissBuf.array());
    Assert.assertEquals(0, manager.mPagesServed);
    Assert.assertEquals(1, manager.mPagesCached);

    // cache hit
    stream.seek(0);
    ByteBuffer cacheHitBuf = ByteBuffer.wrap(new byte[fileSize]);
    Assert.assertEquals(fileSize, stream.read(cacheHitBuf));
    Assert.assertArrayEquals(testData, cacheHitBuf.array());
    Assert.assertEquals(1, manager.mPagesServed);
  }

  @Test
  public void readMultipleFiles() throws Exception {
    Random random = new Random();
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    Map<String, byte[]> files = IntStream.range(0, 10)
        .mapToObj(i -> BufferUtils.getIncreasingByteArray(i, random.nextInt(100)))
        .collect(Collectors.toMap(d -> PathUtils.uniqPath(), Function.identity()));
    Map<AlluxioURI, LocalCacheFileInStream> streams = setupWithMultipleFiles(files, manager);
    for (AlluxioURI path : streams.keySet()) {
      try (InputStream stream = streams.get(path)) {
        Assert.assertArrayEquals(files.get(path.toString()), ByteStreams.toByteArray(stream));
      }
    }
  }

  @Test
  public void cacheMetricCacheHitReadTime() throws Exception {
    byte[] testData = BufferUtils.getIncreasingByteArray(mPageSize);
    AlluxioURI testFileName = new AlluxioURI("/test");
    Map<AlluxioURI, byte[]> files = ImmutableMap.of(testFileName, testData);
    StepTicker timeSource = new StepTicker();
    TimedMockByteArrayCacheManager manager = new TimedMockByteArrayCacheManager(timeSource);
    Map<String, Long> recordedMetrics = new HashMap<>();
    TimedByteArrayFileSystem fs = new TimedByteArrayFileSystem(
        files,
        (name, value) -> recordedMetrics.compute(name, (k, v) -> v == null ? value : v + value),
        timeSource
    );
    LocalCacheFileInStream stream =
        new LocalCacheFileInStream(fs.getStatus(testFileName),
            (status) -> fs.openFile(status, OpenFilePOptions.getDefaultInstance()), manager,
            sConf, Optional.empty()) {
          @Override
          protected Stopwatch createUnstartedStopwatch() {
            return Stopwatch.createUnstarted(timeSource);
          }
        };

    Assert.assertArrayEquals(testData, ByteStreams.toByteArray(stream));
    long timeReadExternal = recordedMetrics.get(
        MetricKey.CLIENT_CACHE_PAGE_READ_EXTERNAL_TIME_NS.getMetricName());
    Assert.assertEquals(timeSource.get(StepTicker.Type.CACHE_MISS), timeReadExternal);
  }

  @Test
  public void testUnbuffer() throws Exception {
    int fileSize = mPageSize;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    LocalCacheFileInStream stream = setupWithSingleFile(testData, manager);

    int partialReadSize = fileSize / 5;
    int offset = fileSize / 5;

    byte[] cacheMiss = new byte[partialReadSize];
    stream.unbuffer();
    stream.seek(offset);
    stream.unbuffer();
    Assert.assertEquals(partialReadSize, stream.read(cacheMiss));
    stream.unbuffer();
    Assert.assertArrayEquals(
        Arrays.copyOfRange(testData, offset, offset + partialReadSize), cacheMiss);
    Assert.assertEquals(0, manager.mPagesServed);
    Assert.assertEquals(1, manager.mPagesCached);

    byte[] cacheHit = new byte[partialReadSize];
    stream.unbuffer();
    stream.seek(offset);
    stream.unbuffer();
    Assert.assertEquals(partialReadSize, stream.read(cacheHit));
    stream.unbuffer();
    Assert.assertArrayEquals(
        Arrays.copyOfRange(testData, offset, offset + partialReadSize), cacheHit);
    Assert.assertEquals(1, manager.mPagesServed);
  }

  @Test
  public void testPageDataFileCorrupted() throws Exception
  {
    int pages = 10;
    int fileSize = mPageSize * pages;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    //by default local cache fallback is not enabled, the read should fail for any error
    LocalCacheFileInStream streamWithOutFallback = setupWithSingleFile(testData, manager);

    sConf.set(PropertyKey.USER_CLIENT_CACHE_FALLBACK_ENABLED, true);
    LocalCacheFileInStream streamWithFallback = setupWithSingleFile(testData, manager);
    Assert.assertEquals(100, streamWithFallback.positionedRead(0, new byte[10], 100, 100));
    Assert.assertEquals(1,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_POSITION_READ_FALLBACK.getName()).getCount());
  }

  @Test
  public void testPositionReadFallBack() throws Exception
  {
    int pages = 10;
    int fileSize = mPageSize * pages;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    sConf.set(PropertyKey.USER_CLIENT_CACHE_FALLBACK_ENABLED, false);
    //by default local cache fallback is not enabled, the read should fail for any error
    LocalCacheFileInStream streamWithOutFallback = setupWithSingleFile(testData, manager);
    try {
      streamWithOutFallback.positionedRead(0, new byte[10], 100, 100);
      fail("Expect position read fail here.");
    } catch (ArrayIndexOutOfBoundsException e) {
      //expected exception
    }
    sConf.set(PropertyKey.USER_CLIENT_CACHE_FALLBACK_ENABLED, true);
    LocalCacheFileInStream streamWithFallback = setupWithSingleFile(testData, manager);
    Assert.assertEquals(100, streamWithFallback.positionedRead(0, new byte[10], 100, 100));
    Assert.assertEquals(1,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_POSITION_READ_FALLBACK.getName()).getCount());
  }

  @Test
  public void testPositionReadWithPerThreadContextSameThread() throws Exception
  {
    int fileSize = mPageSize * 5;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    LocalCacheFileInStream stream = setupWithSingleFile(testData, manager);

    // read last page with cache enabled (cache miss)
    CachePerThreadContext.get().setCacheEnabled(true);
    stream.positionedRead(fileSize - mPageSize, new byte[mPageSize], 0, mPageSize);
    Assert.assertEquals(0,
        MetricsSystem.meter(MetricKey.CLIENT_CACHE_BYTES_READ_CACHE.getName()).getCount());
    Assert.assertEquals(1,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_EXTERNAL_REQUESTS.getName()).getCount());
    Assert.assertEquals(0,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_HIT_REQUESTS.getName()).getCount());
    // read last page with cache enabled (cache hit)
    stream.positionedRead(fileSize - mPageSize, new byte[mPageSize], 0, mPageSize);
    Assert.assertEquals(1,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_EXTERNAL_REQUESTS.getName()).getCount());
    Assert.assertEquals(1,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_HIT_REQUESTS.getName()).getCount());

    MetricsSystem.clearAllMetrics();
    // read first page with cache disabled (cache miss)
    CachePerThreadContext.get().setCacheEnabled(false);
    stream.positionedRead(0, new byte[mPageSize], 0, mPageSize);
    Assert.assertEquals(0,
        MetricsSystem.meter(MetricKey.CLIENT_CACHE_BYTES_READ_CACHE.getName()).getCount());
    Assert.assertEquals(1,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_EXTERNAL_REQUESTS.getName()).getCount());
    Assert.assertEquals(0,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_HIT_REQUESTS.getName()).getCount());
    // read last page again with cache enabled (cache miss)
    stream.positionedRead(0, new byte[mPageSize], 0, mPageSize);
    Assert.assertEquals(2,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_EXTERNAL_REQUESTS.getName()).getCount());
    Assert.assertEquals(0,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_HIT_REQUESTS.getName()).getCount());
  }

  private LocalCacheFileInStream setupWithSingleFile(byte[] data, CacheManager manager)
      throws Exception {
    Map<AlluxioURI, byte[]> files = new HashMap<>();
    AlluxioURI testFilename = new AlluxioURI("/test");
    files.put(testFilename, data);

    ByteArrayFileSystem fs = new ByteArrayFileSystem(files);

    return new LocalCacheFileInStream(fs.getStatus(testFilename),
        (status) -> fs.openFile(status, OpenFilePOptions.getDefaultInstance()), manager, sConf,
        Optional.empty());
  }

  private Map<AlluxioURI, LocalCacheFileInStream> setupWithMultipleFiles(Map<String, byte[]> files,
      CacheManager manager) {
    Map<AlluxioURI, byte[]> fileMap = files.entrySet().stream()
        .collect(Collectors.toMap(entry -> new AlluxioURI(entry.getKey()), Map.Entry::getValue));
    final ByteArrayFileSystem fs = new ByteArrayFileSystem(fileMap);

    Map<AlluxioURI, LocalCacheFileInStream> ret = new HashMap<>();
    fileMap.entrySet().forEach(entry -> {
      try {
        ret.put(entry.getKey(),
            new LocalCacheFileInStream(fs.getStatus(entry.getKey()),
                (status) -> fs.openFile(status, OpenFilePOptions.getDefaultInstance()), manager,
                sConf, Optional.empty()));
      } catch (Exception e) {
        // skip
      }
    });
    return ret;
  }

  private URIStatus generateURIStatus(String path, long len) {
    FileInfo info = new FileInfo();
    info.setFileId(path.hashCode());
    info.setPath(path);
    info.setUfsPath(path);
    info.setLength(len);
    return new URIStatus(info);
  }

  private void verifyReadFullFile(int fileSize, int bufferSize, int pages) throws Exception {
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    LocalCacheFileInStream stream = setupWithSingleFile(testData, manager);

    // cache miss
    byte[] cacheMiss = new byte[bufferSize];
    Assert.assertEquals(fileSize, stream.read(cacheMiss));
    Assert.assertArrayEquals(testData, Arrays.copyOfRange(cacheMiss, 0, fileSize));
    Assert.assertEquals(0, manager.mPagesServed);
    Assert.assertEquals(pages, manager.mPagesCached);

    // cache hit
    stream.seek(0);
    byte[] cacheHit = new byte[bufferSize];
    Assert.assertEquals(fileSize, stream.read(cacheHit));
    Assert.assertArrayEquals(testData, Arrays.copyOfRange(cacheHit, 0, fileSize));
    Assert.assertEquals(pages, manager.mPagesServed);
  }

  private void verifyReadFullFileThroughReadByteBufferMethod(int fileSize, int bufferSize,
      int pages) throws Exception {
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    LocalCacheFileInStream stream = setupWithSingleFile(testData, manager);

    // cache miss
    byte[] cacheMiss = new byte[bufferSize];
    ByteBuffer cacheMissBuffer = ByteBuffer.wrap(cacheMiss);
    Assert.assertEquals(fileSize, stream.read(cacheMissBuffer));
    Assert.assertArrayEquals(testData, Arrays.copyOfRange(cacheMiss, 0, fileSize));
    Assert.assertEquals(0, manager.mPagesServed);
    Assert.assertEquals(pages, manager.mPagesCached);

    // cache hit
    stream.seek(0);
    byte[] cacheHit = new byte[bufferSize];
    ByteBuffer cacheHitBuffer = ByteBuffer.wrap(cacheHit);
    Assert.assertEquals(fileSize, stream.read(cacheHitBuffer));
    Assert.assertArrayEquals(testData, Arrays.copyOfRange(cacheHit, 0, fileSize));
    Assert.assertEquals(pages, manager.mPagesServed);
  }

  /**
   * Implementation of cache manager that stores cached data in byte arrays in memory.
   */
  private class ByteArrayCacheManager implements CacheManager {
    private final Map<PageId, byte[]> mPages;

    /** Metrics for test validation. */
    long mPagesServed = 0;
    long mPagesCached = 0;

    ByteArrayCacheManager() {
      mPages = new HashMap<>();
    }

    @Override
    public boolean put(PageId pageId, ByteBuffer page, CacheContext cacheContext) {
      mPages.put(pageId, page.array());
      mPagesCached++;
      return true;
    }

    @Override
    public int get(PageId pageId, int pageOffset, int bytesToRead, ReadTargetBuffer target,
        CacheContext cacheContext) {
      if (!mPages.containsKey(pageId)) {
        return 0;
      }
      mPagesServed++;
      target.writeBytes(mPages.get(pageId), pageOffset, bytesToRead);
      return bytesToRead;
    }

    @Override
    public void commitFile(String fileId) {
      throw new UnsupportedOperationException("commitFile method is unsupported. ");
    }

    @Override
    public int getAndLoad(PageId pageId, int pageOffset, int bytesToRead,
        ReadTargetBuffer buffer, CacheContext cacheContext,
        Supplier<byte[]> externalDataSupplier) {
      int bytesRead = get(pageId, pageOffset,
          bytesToRead, buffer, cacheContext);
      if (bytesRead > 0) {
        return bytesRead;
      }
      byte[] page = externalDataSupplier.get();
      if (page.length == 0) {
        return 0;
      }
      buffer.writeBytes(page, pageOffset, bytesToRead);
      put(pageId, page, cacheContext);
      return bytesToRead;
    }

    @Override
    public boolean delete(PageId pageId) {
      return mPages.remove(pageId) != null;
    }

    @Override
    public State state() {
      return State.READ_WRITE;
    }

    @Override
    public boolean append(PageId pageId, int appendAt, byte[] page, CacheContext cacheContext) {
      return false;
    }

    @Override
    public void close() throws Exception {
      // no-op
    }

    @Override
    public void deleteFile(String fileId) {
      // no-op
    }

    @Override
    public void deleteTempFile(String fileId) {
      // no-op
    }

    @Override
    public Optional<CacheUsage> getUsage() {
      return Optional.of(new Usage());
    }

    @Override
    public Optional<DataFileChannel> getDataFileChannel(PageId pageId, int pageOffset,
        int bytesToRead, CacheContext cacheContext) {
      return Optional.empty();
    }

    class Usage implements CacheUsage {
      @Override
      public Optional<CacheUsage> partitionedBy(PartitionDescriptor<?> partition) {
        return Optional.empty();
      }

      @Override
      public long used() {
        return mPages.values().stream().mapToInt(page -> page.length).sum();
      }

      @Override
      public long available() {
        return Integer.MAX_VALUE;
      }

      @Override
      public long capacity() {
        return Integer.MAX_VALUE;
      }
    }
  }

  /**
   * An implementation of file system which is a store of filenames to byte arrays. Only
   * {@link FileSystem#openFile(AlluxioURI)} and its variants are supported.
   */
  private class ByteArrayFileSystem implements FileSystem {
    private final Map<AlluxioURI, byte[]> mFiles;

    ByteArrayFileSystem(Map<AlluxioURI, byte[]> files) {
      mFiles = files;
    }

    @Override
    public boolean isClosed() {
      throw new UnsupportedOperationException();
    }

    @Override
    public void checkAccess(AlluxioURI path, CheckAccessPOptions options)
        throws InvalidPathException, IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public void createDirectory(AlluxioURI path, CreateDirectoryPOptions options)
        throws FileAlreadyExistsException, InvalidPathException, IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public FileOutStream createFile(AlluxioURI path, CreateFilePOptions options)
        throws FileAlreadyExistsException, InvalidPathException, IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public void delete(AlluxioURI path, DeletePOptions options)
        throws DirectoryNotEmptyException, FileDoesNotExistException, IOException,
        AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public boolean exists(AlluxioURI path, ExistsPOptions options)
        throws InvalidPathException, IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public void free(AlluxioURI path, FreePOptions options)
        throws FileDoesNotExistException, IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public List<BlockLocationInfo> getBlockLocations(AlluxioURI path)
        throws FileDoesNotExistException, IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public List<BlockLocationInfo> getBlockLocations(URIStatus status)
        throws FileDoesNotExistException, IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public AlluxioConfiguration getConf() {
      return sConf;
    }

    @Override
    public URIStatus getStatus(AlluxioURI path, GetStatusPOptions options)
        throws FileDoesNotExistException, IOException, AlluxioException {
      if (mFiles.containsKey(path)) {
        return generateURIStatus(path.getPath(), mFiles.get(path).length);
      } else {
        throw new FileDoesNotExistException(path);
      }
    }

    @Override
    public void iterateStatus(AlluxioURI path, ListStatusPOptions options,
        Consumer<? super URIStatus> action)
        throws FileDoesNotExistException, IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public List<URIStatus> listStatus(AlluxioURI path, ListStatusPOptions options)
        throws FileDoesNotExistException, IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public ListStatusPartialResult listStatusPartial(
        AlluxioURI path, ListStatusPartialPOptions options) {
      throw new UnsupportedOperationException();
    }

    @Override
    public void loadMetadata(AlluxioURI path, ListStatusPOptions options)
        throws FileDoesNotExistException, IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public void mount(AlluxioURI alluxioPath, AlluxioURI ufsPath, MountPOptions options)
        throws IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public void updateMount(AlluxioURI alluxioPath, MountPOptions options)
        throws IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public Map<String, MountPointInfo> getMountTable(boolean checkUfs)
        throws IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public FileInStream openFile(AlluxioURI path, OpenFilePOptions options)
        throws FileDoesNotExistException, OpenDirectoryException, FileIncompleteException,
        IOException, AlluxioException {
      if (mFiles.containsKey(path)) {
        return new MockFileInStream(mFiles.get(path));
      } else {
        throw new FileDoesNotExistException(path);
      }
    }

    @Override
    public FileInStream openFile(URIStatus status, OpenFilePOptions options)
        throws FileDoesNotExistException, OpenDirectoryException, FileIncompleteException,
        IOException, AlluxioException {
      AlluxioURI path = new AlluxioURI(status.getPath());
      if (mFiles.containsKey(path)) {
        return new MockFileInStream(mFiles.get(path));
      } else {
        throw new FileDoesNotExistException(path);
      }
    }

    @Override
    public PositionReader openPositionRead(AlluxioURI path, OpenFilePOptions options) {
      throw new UnsupportedOperationException();
    }

    @Override
    public PositionReader openPositionRead(URIStatus status, OpenFilePOptions options) {
      throw new UnsupportedOperationException();
    }

    @Override
    public void persist(AlluxioURI path, ScheduleAsyncPersistencePOptions options)
        throws FileDoesNotExistException, IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public void rename(AlluxioURI src, AlluxioURI dst, RenamePOptions options)
        throws FileDoesNotExistException, IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public AlluxioURI reverseResolve(AlluxioURI ufsUri)
        throws IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public void setAcl(AlluxioURI path, SetAclAction action, List<AclEntry> entries,
        SetAclPOptions options) throws FileDoesNotExistException, IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public void setAttribute(AlluxioURI path, SetAttributePOptions options)
        throws FileDoesNotExistException, IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public void unmount(AlluxioURI path, UnmountPOptions options)
        throws IOException, AlluxioException {
      throw new UnsupportedOperationException();
    }

    @Override
    public void needsSync(AlluxioURI path) {
      throw new UnsupportedOperationException();
    }

    @Override
    public Optional<String> submitJob(JobRequest jobRequest) {
      throw new UnsupportedOperationException();
    }

    @Override
    public boolean stopJob(JobDescription jobDescription) {
      throw new UnsupportedOperationException();
    }

    @Override
    public String getJobProgress(JobDescription jobDescription,
        JobProgressReportFormat format, boolean verbose) {
      throw new UnsupportedOperationException();
    }

    @Override
    public void close() throws IOException {
      throw new UnsupportedOperationException();
    }
  }

  private static class StepTicker extends Ticker {
    private long mNow = 0;
    private final Map<Type, Long> mTimeMap = Maps.newEnumMap(Type.class);

    enum Type {
      CACHE_HIT, CACHE_MISS;
    }

    public long getNow() {
      return mNow;
    }

    public long get(Type type) {
      return mTimeMap.getOrDefault(type, 0L);
    }

    public void advance(Type type) {
      mNow += 1;
      mTimeMap.compute(type, (k, v) -> v == null ? 1 : v + 1);
    }

    @Override
    public long read() {
      return mNow;
    }
  }

  private class MockedCacheContext extends CacheContext {

    private final BiConsumer<String, Long> mCounter;

    public MockedCacheContext(BiConsumer<String, Long> counter) {
      mCounter = counter;
    }

    @Override
    public void incrementCounter(String name, StatsUnit unit, long value) {
      mCounter.accept(name, value);
    }
  }

  private class TimedByteArrayFileSystem extends ByteArrayFileSystem {
    private final StepTicker mTicker;
    private final BiConsumer<String, Long> mCounter;

    TimedByteArrayFileSystem(Map<AlluxioURI, byte[]> files,
        BiConsumer<String, Long> counter, StepTicker ticker) {
      super(files);
      mTicker = ticker;
      mCounter = counter;
    }

    @Override
    public URIStatus getStatus(AlluxioURI path, GetStatusPOptions options)
        throws FileDoesNotExistException, IOException, AlluxioException {
      URIStatus status = super.getStatus(path, options);
      URIStatus withCacheContextStatus = new URIStatus(status.getFileInfo(),
          new MockedCacheContext(mCounter));
      return withCacheContextStatus;
    }

    @Override
    public FileInStream openFile(AlluxioURI path, OpenFilePOptions options)
        throws FileDoesNotExistException, OpenDirectoryException, FileIncompleteException,
        IOException, AlluxioException {
      FileInStream delegateToStream = super.openFile(path, options);
      return new TimedMockFileInStream(delegateToStream, mTicker);
    }

    @Override
    public FileInStream openFile(URIStatus status, OpenFilePOptions options)
        throws FileDoesNotExistException, OpenDirectoryException, FileIncompleteException,
        IOException, AlluxioException {
      FileInStream delegateToStream = super.openFile(status, options);
      return new TimedMockFileInStream(delegateToStream, mTicker);
    }
  }

  private class TimedMockByteArrayCacheManager extends ByteArrayCacheManager {
    private final StepTicker mTicker;

    public TimedMockByteArrayCacheManager(StepTicker ticker) {
      mTicker = ticker;
    }

    @Override
    public int get(PageId pageId, int pageOffset, int bytesToRead, ReadTargetBuffer target,
        CacheContext cacheContext) {
      int read = super.get(pageId, pageOffset, bytesToRead, target, cacheContext);
      if (read > 0) {
        mTicker.advance(StepTicker.Type.CACHE_HIT);
      }
      // when cache misses, time elapsed during `get` should not count towards external read metric
      return read;
    }
  }

  private static class TimedMockFileInStream extends FileInStream {
    private final FileInStream mDelegate;
    private final StepTicker mTicker;

    public TimedMockFileInStream(FileInStream delegate, StepTicker ticker) {
      mDelegate = delegate;
      mTicker = ticker;
    }

    @Override
    public int read(ByteBuffer byteBuffer, int off, int len) throws IOException {
      mTicker.advance(StepTicker.Type.CACHE_MISS);
      return mDelegate.read(byteBuffer, off, len);
    }

    @Override
    public long getPos() throws IOException {
      return mDelegate.getPos();
    }

    @Override
    public long remaining() {
      return mDelegate.remaining();
    }

    @Override
    public void seek(long pos) throws IOException {
      mDelegate.seek(pos);
    }

    @Override
    public int positionedRead(long position, byte[] buffer, int offset, int length)
        throws IOException {
      mTicker.advance(StepTicker.Type.CACHE_MISS);
      return mDelegate.positionedRead(position, buffer, offset, length);
    }
  }

  private class MultiReadByteArrayFileSystem extends ByteArrayFileSystem {
    MultiReadByteArrayFileSystem(Map<AlluxioURI, byte[]> files) {
      super(files);
    }

    @Override
    public FileInStream openFile(AlluxioURI path, OpenFilePOptions options)
        throws FileDoesNotExistException, OpenDirectoryException, FileIncompleteException,
        IOException, AlluxioException {
      return new MultiReadFileInStream(super.openFile(path, options));
    }

    @Override
    public FileInStream openFile(URIStatus status, OpenFilePOptions options)
        throws FileDoesNotExistException, OpenDirectoryException, FileIncompleteException,
        IOException, AlluxioException {
      return new MultiReadFileInStream(super.openFile(status, options));
    }
  }

  /**
   * Mock implementation of {@link FileInStream} which delegates to a {@link ByteArrayInputStream}.
   * This implementation may not serve the full read in a single call.
   */
  private class MultiReadFileInStream extends FileInStream {
    private final FileInStream mIn;

    /**
     * Creates an FileInStream that may not serve read calls in a single call.
     *
     * @param in the backing FileInStream
     */
    public MultiReadFileInStream(FileInStream in) {
      mIn = in;
    }

    @Override
    public int read() throws IOException {
      return mIn.read();
    }

    @Override
    public int read(byte[] b) throws IOException {
      return read(b, 0, b.length);
    }

    @Override
    public int read(byte[] b, int off, int len) throws IOException {
      int toRead = len > 1 ? ThreadLocalRandom.current().nextInt(1, len) : len;
      return mIn.read(b, off, toRead);
    }

    @Override
    public int read(ByteBuffer buf) throws IOException {
      return mIn.read(buf);
    }

    @Override
    public long getPos() throws IOException {
      return mIn.getPos();
    }

    @Override
    public long remaining() {
      return mIn.remaining();
    }

    @Override
    public void seek(long pos) throws IOException {
      mIn.seek(pos);
    }

    @Override
    public int positionedRead(long position, byte[] buffer, int offset, int length)
        throws IOException {
      return mIn.positionedRead(position, buffer, offset, length);
    }
  }
}
