alluxio-master-2.x/core/server/worker/src/test/java/alluxio/worker/page/PagedBlockStoreDirTest.java
/*
 * 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.worker.page;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;

import alluxio.Constants;
import alluxio.client.file.cache.PageId;
import alluxio.client.file.cache.PageInfo;
import alluxio.client.file.cache.evictor.CacheEvictorOptions;
import alluxio.client.file.cache.evictor.FIFOCacheEvictor;
import alluxio.client.file.cache.store.PageStoreDir;
import alluxio.client.file.cache.store.PageStoreOptions;
import alluxio.client.file.cache.store.PageStoreType;
import alluxio.conf.Configuration;
import alluxio.conf.InstancedConfiguration;
import alluxio.conf.PropertyKey;
import alluxio.worker.block.BlockStoreLocation;

import com.google.common.collect.ImmutableList;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;

import java.nio.file.Path;

public class PagedBlockStoreDirTest {
  private static final int DIR_INDEX = 0;
  private static final long BLOCK_SIZE = Constants.MB;
  private PagedBlockStoreDir mDir;
  private Path mDirPath;

  @Rule
  public TemporaryFolder mTempFolder = new TemporaryFolder();

  @Before
  public void setup() throws Exception {
    mDirPath = mTempFolder.newFolder().toPath();
    InstancedConfiguration conf = Configuration.modifiableGlobal();
    conf.set(PropertyKey.WORKER_PAGE_STORE_DIRS, ImmutableList.of(mDirPath));
    conf.set(PropertyKey.WORKER_PAGE_STORE_SIZES, ImmutableList.of(Constants.MB));
    conf.set(PropertyKey.WORKER_PAGE_STORE_TYPE, PageStoreType.LOCAL);
    PageStoreDir pageStoreDir =
        PageStoreDir.createPageStoreDir(
            new CacheEvictorOptions().setEvictorClass(FIFOCacheEvictor.class),
            PageStoreOptions.createForWorkerPageStore(conf).get(DIR_INDEX));
    mDir = new PagedBlockStoreDir(pageStoreDir, DIR_INDEX);
  }

  private static BlockPageId blockPageId(String blockId, long pageIndex) {
    return new BlockPageId(blockId, pageIndex, BLOCK_SIZE);
  }

  @Test
  public void rootPath() {
    assertEquals(mDirPath.resolve(PageStoreType.LOCAL.name()), mDir.getRootPath());
  }

  @Test
  public void dirIndex() {
    assertEquals(DIR_INDEX, mDir.getDirIndex());
  }

  @Test
  public void blockLocation() {
    BlockStoreLocation location = mDir.getLocation();
    assertEquals(PagedBlockStoreMeta.DEFAULT_TIER, location.tierAlias());
    assertEquals(PagedBlockStoreMeta.DEFAULT_MEDIUM, location.mediumType());
    assertEquals(DIR_INDEX, location.dir());
  }

  @Test
  public void numBlocks() {
    assertEquals(0, mDir.getNumBlocks());
    mDir.putPage(new PageInfo(blockPageId("0", 0), 0, mDir));
    assertEquals(1, mDir.getNumBlocks());
    mDir.putPage(new PageInfo(blockPageId("0", 1), 0, mDir));
    assertEquals(1, mDir.getNumBlocks());
    mDir.putPage(new PageInfo(blockPageId("1", 0), 0, mDir));
    assertEquals(2, mDir.getNumBlocks());
  }

  @Test
  public void cachedBytes() {
    assertEquals(0, mDir.getCachedBytes());
    mDir.putPage(new PageInfo(blockPageId("0", 0), Constants.KB, mDir));
    assertEquals(Constants.KB, mDir.getCachedBytes());
    assertEquals(Constants.KB, mDir.getBlockCachedBytes(0));
    mDir.putPage(new PageInfo(blockPageId("0", 1), Constants.KB, mDir));
    assertEquals(2 * Constants.KB, mDir.getCachedBytes());
    assertEquals(2 * Constants.KB, mDir.getBlockCachedBytes(0));
    mDir.putPage(new PageInfo(blockPageId("1", 0), Constants.KB, mDir));
    assertEquals(3 * Constants.KB, mDir.getCachedBytes());
    assertEquals(Constants.KB, mDir.getBlockCachedBytes(1));
  }

  @Test
  public void putPage() throws Exception {
    long blockId = 0;
    PageId pageId = blockPageId(String.valueOf(blockId), 0);
    PageInfo pageInfo = new PageInfo(pageId, Constants.KB, mDir);
    mDir.putPage(pageInfo);
    assertEquals(1, mDir.getBlockCachedPages(blockId));
    assertTrue(mDir.hasFile(pageId.getFileId()));

    // duplicate adds are ignored
    mDir.putPage(pageInfo);
    assertEquals(1, mDir.getBlockCachedPages(blockId));
  }

  @Test
  public void deletePage() throws Exception {
    long blockId = 0;
    PageId pageId = blockPageId(String.valueOf(blockId), 0);
    PageInfo pageInfo = new PageInfo(pageId, Constants.KB, mDir);
    mDir.putPage(pageInfo);
    assertEquals(1, mDir.getBlockCachedPages(blockId));
    assertEquals(Constants.KB, mDir.getBlockCachedBytes(blockId));

    long cachedBytes = mDir.deletePage(pageInfo);
    assertEquals(0, mDir.getBlockCachedPages(blockId));
    assertEquals(0, mDir.getBlockCachedBytes(blockId));
    assertEquals(mDir.getCachedBytes(), cachedBytes);

    // deleting a non-existing page is no-op
    long cachedBytes2 = mDir.deletePage(pageInfo);
    assertEquals(0, mDir.getBlockCachedPages(blockId));
    assertEquals(0, mDir.getBlockCachedBytes(blockId));
    assertEquals(cachedBytes, cachedBytes2);
  }

  @Test
  public void evictor() throws Exception {
    BlockPageEvictor evictor = mDir.getEvictor();

    long blockId = 0;
    PageId pageId0 = blockPageId(String.valueOf(blockId), 0);
    PageInfo pageInfo0 = new PageInfo(pageId0, Constants.KB, mDir);
    mDir.putPage(pageInfo0);
    assertEquals(pageId0, evictor.evict());

    PageId pageId1 = blockPageId(String.valueOf(blockId), 1);
    PageInfo pageInfo1 = new PageInfo(pageId1, Constants.KB, mDir);
    mDir.putPage(pageInfo1);
    mDir.deletePage(pageInfo0);
    assertEquals(pageId1, evictor.evict());
  }

  @Test
  public void addTempPage() throws Exception {
    long blockId = 0;
    PageId pageId0 = blockPageId(String.valueOf(blockId), 0);
    final String fileId = pageId0.getFileId();
    PageInfo pageInfo0 = new PageInfo(pageId0, Constants.KB, mDir);
    mDir.putTempPage(pageInfo0);
    mDir.getPageStore().putTemporary(pageId0, new byte[Constants.KB]);
    assertEquals(0, mDir.getBlockCachedPages(blockId));
    assertFalse(mDir.hasFile(fileId));
    assertTrue(mDir.hasTempFile(fileId));

    mDir.commit(fileId);
    assertEquals(1, mDir.getBlockCachedPages(blockId));
    assertTrue(mDir.hasFile(fileId));
    assertFalse(mDir.hasTempFile(fileId));
  }

  @Test
  public void abortTempPage() throws Exception {
    long blockId = 0;
    PageId pageId0 = blockPageId(String.valueOf(blockId), 0);
    final String fileId = pageId0.getFileId();
    PageInfo pageInfo0 = new PageInfo(pageId0, Constants.KB, mDir);
    mDir.putTempPage(pageInfo0);
    mDir.getPageStore().putTemporary(pageId0, new byte[Constants.KB]);
    assertEquals(0, mDir.getBlockCachedPages(blockId));
    assertTrue(mDir.hasTempFile(fileId));
    assertFalse(mDir.hasFile(fileId));

    mDir.abort(fileId);
    assertEquals(0, mDir.getBlockCachedPages(blockId));
    assertFalse(mDir.hasTempFile(fileId));
    assertFalse(mDir.hasFile(fileId));
  }
}
