/*
 * 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.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

import alluxio.Constants;
import alluxio.client.file.CacheContext;
import alluxio.conf.Configuration;
import alluxio.conf.InstancedConfiguration;
import alluxio.conf.PropertyKey;
import alluxio.file.ReadTargetBuffer;
import alluxio.network.protocol.databuffer.DataFileChannel;
import alluxio.util.io.BufferUtils;

import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Optional;
import java.util.function.Supplier;

/**
 * Tests for the {@link LocalCacheManager} class.
 */
@RunWith(Parameterized.class)
public final class CacheManagerWithShadowCacheTest {
  private static final int PAGE_SIZE_BYTES = Constants.KB;
  private static final PageId PAGE_ID1 = new PageId("0L", 0L);
  private static final PageId PAGE_ID2 = new PageId("1L", 1L);
  private static final byte[] PAGE1 = BufferUtils.getIncreasingByteArray(PAGE_SIZE_BYTES);
  private static final byte[] PAGE2 = BufferUtils.getIncreasingByteArray(255, PAGE_SIZE_BYTES);
  private final byte[] mBuf = new byte[PAGE_SIZE_BYTES];
  private CacheManagerWithShadowCache mCacheManager;
  private final InstancedConfiguration mConf = Configuration.copyGlobal();

  private final ShadowCacheType mShadowCacheType;
  private final int mAgeBits;
  private final int mMaxAge;

  /**
   * @return a list of types of shadow cache and bits of age field
   */
  @Parameterized.Parameters
  public static Collection<Object[]> data() {
    return Arrays.asList(new Object[][] {
      {ShadowCacheType.MULTIPLE_BLOOM_FILTER, 2},
      {ShadowCacheType.CLOCK_CUCKOO_FILTER, 2},
      {ShadowCacheType.CLOCK_CUCKOO_FILTER, 4}
    });
  }

  /**
   * @param shadowCacheType the type of shadow cache
   * @param ageBits the number of bits of age the shadow cache can represent
   */
  public CacheManagerWithShadowCacheTest(ShadowCacheType shadowCacheType, int ageBits) {
    mShadowCacheType = shadowCacheType;
    mAgeBits = ageBits;
    mMaxAge = (1 << ageBits);
  }

  @Before
  public void before() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SHADOW_TYPE, mShadowCacheType);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SHADOW_WINDOW, "20s");
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SHADOW_MEMORY_OVERHEAD, "1MB");
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SHADOW_BLOOMFILTER_NUM, mMaxAge);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SHADOW_CUCKOO_CLOCK_BITS, mAgeBits);
    mCacheManager = new CacheManagerWithShadowCache(new KVCacheManager(), mConf);
    mCacheManager.stopUpdate();
  }

  @Test
  public void putOne() throws Exception {
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
    mCacheManager.updateWorkingSetSize();
    assertEquals(mCacheManager.getShadowCachePages(), 1);
    assertEquals(mCacheManager.getShadowCacheBytes(), PAGE1.length);
  }

  @Test
  public void putTwo() throws Exception {
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    assertTrue(mCacheManager.put(PAGE_ID2, PAGE2));
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
    assertEquals(PAGE2.length, mCacheManager.get(PAGE_ID2, PAGE2.length, mBuf, 0));
    assertArrayEquals(PAGE2, mBuf);
    mCacheManager.updateWorkingSetSize();
    assertEquals(mCacheManager.getShadowCachePages(), 2);
    assertEquals(mCacheManager.getShadowCacheBytes(), PAGE1.length + PAGE2.length);
  }

  @Test
  public void putExist() throws Exception {
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE2));
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
    mCacheManager.updateWorkingSetSize();
    assertEquals(mCacheManager.getShadowCachePages(), 1);
    assertEquals(mCacheManager.getShadowCacheBytes(), PAGE1.length);
  }

  @Test
  public void bloomFilterExpire() throws Exception {
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
    for (int i = 0; i <= mMaxAge; i++) {
      mCacheManager.aging();
    }
    mCacheManager.updateWorkingSetSize();
    assertEquals(mCacheManager.getShadowCachePages(), 0);
    assertEquals(mCacheManager.getShadowCacheBytes(), 0);
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
  }

  @Test
  public void filterExpireHalf() throws Exception {
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
    for (int i = 0; i < mMaxAge / 2; i++) {
      mCacheManager.aging();
    }
    assertTrue(mCacheManager.put(PAGE_ID2, PAGE2));
    assertEquals(PAGE2.length, mCacheManager.get(PAGE_ID2, PAGE2.length, mBuf, 0));
    assertArrayEquals(PAGE2, mBuf);
    for (int i = 0; i <= mMaxAge / 2; i++) {
      mCacheManager.aging();
    }
    mCacheManager.updateWorkingSetSize();
    assertEquals(mCacheManager.getShadowCachePages(), 1);
    assertEquals(mCacheManager.getShadowCacheBytes(), PAGE2.length);
  }

  @Test
  public void delete() throws Exception {
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
    assertTrue(mCacheManager.delete(PAGE_ID1));
    assertEquals(0, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    mCacheManager.updateWorkingSetSize();
    mCacheManager.updateWorkingSetSize();
    mCacheManager.updateWorkingSetSize();
    assertEquals(1, mCacheManager.getShadowCachePages());
    assertEquals(mCacheManager.getShadowCacheBytes(), PAGE1.length);
  }

  @Test
  public void getExistInWindow() throws Exception {
    mCacheManager.put(PAGE_ID1, PAGE1);
    assertEquals(PAGE_SIZE_BYTES, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
    assertEquals(mCacheManager.getShadowCacheBytes(), PAGE1.length);
  }

  @Test
  public void getExistInRollingWindow() throws Exception {
    mCacheManager.put(PAGE_ID1, PAGE1);
    for (int i = 0; i <= mMaxAge; i++) {
      mCacheManager.aging();
    }
    mCacheManager.put(PAGE_ID2, PAGE1);
    // PAGE_ID1 is evicted, only PAGE_ID2 in the shadow cache
    assertEquals(mCacheManager.getShadowCacheBytes(), PAGE2.length);
    // PAGE_ID1 is not in the shadow cache but still in the normal cache
    assertEquals(PAGE_SIZE_BYTES, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    // PAGE_ID1 is added to the shadow cache again by 'get'
    assertEquals(mCacheManager.getShadowCacheBytes(), PAGE1.length + PAGE2.length);
  }

  @Test
  public void getNotExist() throws Exception {
    assertEquals(0, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
  }

  @Test
  public void hit() throws Exception {
    mCacheManager.put(PAGE_ID1, PAGE1);
    mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0);
    assertEquals(1, mCacheManager.getShadowCachePageRead());
    assertEquals(1, mCacheManager.getShadowCachePageHit());
    assertEquals(PAGE1.length, mCacheManager.getShadowCacheByteRead());
    assertEquals(PAGE1.length, mCacheManager.getShadowCacheByteHit());
  }

  @Test
  public void hitAfterMiss() throws Exception {
    mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0);
    mCacheManager.put(PAGE_ID1, PAGE1);
    mCacheManager.delete(PAGE_ID1);
    mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0);
    assertEquals(2, mCacheManager.getShadowCachePageRead());
    assertEquals(1, mCacheManager.getShadowCachePageHit());
    assertEquals(PAGE1.length * 2, mCacheManager.getShadowCacheByteRead());
    assertEquals(PAGE1.length, mCacheManager.getShadowCacheByteHit());
  }

  private class KVCacheManager implements CacheManager {
    private final HashMap<PageId, byte[]> mCache = new HashMap<>();

    @Override
    public boolean put(PageId pageId, ByteBuffer page, CacheContext cacheContext) {
      if (!mCache.containsKey(pageId)) {
        mCache.put(pageId, page.array());
      }
      return true;
    }

    @Override
    public int get(PageId pageId, int pageOffset, int bytesToRead, ReadTargetBuffer buffer,
        CacheContext cacheContext) {
      if (!mCache.containsKey(pageId)) {
        return 0;
      }
      byte[] page = mCache.get(pageId);
      if (bytesToRead >= 0) {
        System.arraycopy(page, pageOffset + 0, buffer.byteArray(), (int) buffer.offset(),
            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) {
      if (mCache.containsKey(pageId)) {
        mCache.remove(pageId);
        return true;
      }
      return false;
    }

    @Override
    public State state() {
      return null;
    }

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

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

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

    @Override
    public Optional<CacheUsage> getUsage() {
      return Optional.empty();
    }

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

    @Override
    public void close() throws Exception {}
  }
}
