content: /*
   * 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.metastore.caching;
  
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertFalse;
  import static org.junit.Assert.assertTrue;
  import static org.mockito.ArgumentMatchers.any;
  import static org.mockito.ArgumentMatchers.anyLong;
  import static org.mockito.ArgumentMatchers.anyString;
  import static org.mockito.Mockito.atMost;
  import static org.mockito.Mockito.spy;
  import static org.mockito.Mockito.times;
  import static org.mockito.Mockito.verify;
  
  import alluxio.ConfigurationRule;
  import alluxio.conf.Configuration;
  import alluxio.conf.PropertyKey;
  import alluxio.master.file.contexts.CreateDirectoryContext;
  import alluxio.master.file.contexts.CreateFileContext;
  import alluxio.master.file.meta.Inode;
  import alluxio.master.file.meta.InodeLockManager;
  import alluxio.master.file.meta.MutableInodeDirectory;
  import alluxio.master.file.meta.MutableInodeFile;
  import alluxio.master.journal.checkpoint.CheckpointInputStream;
  import alluxio.master.metastore.InodeStore;
  import alluxio.master.metastore.ReadOption;
  import alluxio.master.metastore.heap.HeapInodeStore;
  import alluxio.resource.CloseableIterator;
  
  import com.google.common.collect.ImmutableMap;
  import io.netty.util.ResourceLeakDetector;
  import org.junit.After;
  import org.junit.Before;
  import org.junit.Rule;
  import org.junit.Test;
  import org.mockito.Mockito;
  
  import java.io.ByteArrayInputStream;
  import java.io.ByteArrayOutputStream;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.List;
  import java.util.Optional;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.ThreadLocalRandom;
  import java.util.concurrent.atomic.AtomicInteger;
  
  public class CachingInodeStoreMockedBackingStoreTest {
    private static final int CACHE_SIZE = 20;
    private static final long TEST_INODE_ID = 5;
    private static final MutableInodeDirectory TEST_INODE_DIR =
        MutableInodeDirectory.create(TEST_INODE_ID, 0, "name", CreateDirectoryContext.defaults());
  
    private InodeStore mBackingStore;
    private CachingInodeStore mStore;
  
    @Rule
    public ConfigurationRule mConf = new ConfigurationRule(
        ImmutableMap.of(PropertyKey.MASTER_METASTORE_INODE_CACHE_MAX_SIZE, CACHE_SIZE,
            PropertyKey.MASTER_METASTORE_INODE_CACHE_EVICT_BATCH_SIZE, 5,
            PropertyKey.LEAK_DETECTOR_LEVEL, ResourceLeakDetector.Level.PARANOID,
            PropertyKey.LEAK_DETECTOR_EXIT_ON_LEAK, true),
        Configuration.modifiableGlobal());
  
    @Before
    public void before() {
      mBackingStore = spy(new HeapInodeStore());
      mStore = new CachingInodeStore(mBackingStore, new InodeLockManager());
      mStore.writeNewInode(TEST_INODE_DIR);
    }
  
    @After
    public void after() {
      mStore.close();
    }
  
    @Test
    public void cacheGetMutable() {
      for (int i = 0; i < 10; i++) {
        assertEquals(TEST_INODE_DIR, mStore.getMutable(TEST_INODE_ID).get());
      }
  
      verifyNoBackingStoreReads();
    }
  
    @Test
    public void cacheGet() {
      Inode testInode = Inode.wrap(TEST_INODE_DIR);
      for (int i = 0; i < CACHE_SIZE / 2; i++) {
        assertEquals(testInode, mStore.get(TEST_INODE_ID).get());
      }
  
      verifyNoBackingStoreReads();
    }
  
    @Test
    public void cacheGetMany() {
      for (long inodeId = 1; inodeId < CACHE_SIZE * 2; inodeId++) {
        createInodeDir(inodeId, 0);
      }
      for (int i = 0; i < 30; i++) {
        for (long inodeId = 1; inodeId < CACHE_SIZE * 2; inodeId++) {
          assertTrue(mStore.get(inodeId).isPresent());
        }
      }
      // The workload is read-only, so we shouldn't need to write each inode to the backing store more
      // than once.
      verify(mBackingStore, atMost((int) CACHE_SIZE * 2 + 1)).writeInode(any());
    }
  
    @Test
    public void removeInodeStaysRemoved() {
      mStore.remove(TEST_INODE_DIR);
  
      assertEquals(Optional.empty(), mStore.get(TEST_INODE_DIR.getId()));
    }
  
    @Test
    public void reflectWrite() {
      MutableInodeDirectory updated = MutableInodeDirectory.create(TEST_INODE_ID, 10, "newName",
          CreateDirectoryContext.defaults());
      mStore.writeInode(updated);
  
      assertEquals("newName", mStore.get(TEST_INODE_ID).get().getName());
    }
  
    @Test
    public void cacheGetChild() {
      MutableInodeFile child =
          MutableInodeFile.create(0, TEST_INODE_ID, "child", 0, CreateFileContext.defaults());
      mStore.writeInode(child);
      mStore.addChild(TEST_INODE_ID, child);
      for (int i = 0; i < 10; i++) {
        assertTrue(mStore.getChild(TEST_INODE_DIR, "child").isPresent());
      }
  
      verifyNoBackingStoreReads();
    }
  
    @Test
    public void listChildrenOverCacheSize() {
      for (long inodeId = 10; inodeId < 10 + CACHE_SIZE * 2; inodeId++) {
        MutableInodeDirectory dir = createInodeDir(inodeId, 0);
        mStore.addChild(0, dir);
      }
      assertEquals(CACHE_SIZE * 2, CloseableIterator.size(mStore.getChildren(0L)));
    }
  
    @Test
    public void cacheGetChildMany() {
      for (long inodeId = 1; inodeId < CACHE_SIZE * 2; inodeId++) {
        MutableInodeFile child =
            MutableInodeFile.create(0, 0, "child" + inodeId, 0, CreateFileContext.defaults());
        mStore.writeInode(child);
        mStore.addChild(0, child);
      }
      for (int i = 0; i < 1000; i++) {
        for (long inodeId = 1; inodeId < CACHE_SIZE * 2; inodeId++) {
          assertTrue(mStore.getChild(0L, "child" + inodeId).isPresent());
        }
      }
  
      // The workload is read-only, so we shouldn't need to write each edge to the backing store more
      // than once.
      verify(mBackingStore, atMost((int) CACHE_SIZE * 2)).addChild(anyLong(), any(), anyLong());
    }
  
    @Test
    public void cacheGetChildrenInodeLookups() {
      List<Inode> children = new ArrayList<>();
      for (int id = 100; id < 110; id++) {
        MutableInodeFile child =
            MutableInodeFile.create(id, TEST_INODE_ID, "child" + id, 0, CreateFileContext.defaults());
        children.add(Inode.wrap(child));
        mStore.writeNewInode(child);
        mStore.addChild(TEST_INODE_ID, child);
      }
      assertEquals(10, CloseableIterator.size(mStore.getChildren(TEST_INODE_DIR)));
  
      verifyNoBackingStoreReads();
    }
  
    @Test
    public void eviction() {
      for (int id = 100; id < 100 + CACHE_SIZE * 2; id++) {
        MutableInodeFile child =
            MutableInodeFile.create(id, TEST_INODE_ID, "child" + id, 0, CreateFileContext.defaults());
        mStore.writeNewInode(child);
        mStore.addChild(TEST_INODE_ID, child);
      }
      for (int id = 100; id < 100 + CACHE_SIZE * 2; id++) {
        assertTrue(mStore.getChild(TEST_INODE_DIR, "child" + id).isPresent());
      }
      verify(mBackingStore, Mockito.atLeastOnce()).getMutable(anyLong(), any(ReadOption.class));
    }
  
    @Test
    public void edgeIndexTest() throws Exception {
      // Run many concurrent operations, then check that the edge cache's indices are accurate.
      long endTimeMs = System.currentTimeMillis() + 200;
      ThreadLocalRandom random = ThreadLocalRandom.current();
      List<MutableInodeDirectory> dirs = new ArrayList<>();
      for (int i = 1; i < 5; i++) {
        MutableInodeDirectory dir = createInodeDir(i, 0);
        dirs.add(dir);
        mStore.addChild(TEST_INODE_ID, dir);
      }
  
      AtomicInteger operations = new AtomicInteger(0);
      ExecutorService executor = Executors.newFixedThreadPool(10);
      int numThreads = 10;
      executor.invokeAll(Collections.nCopies(numThreads, () -> {
        while (operations.get() < 10_000 || System.currentTimeMillis() < endTimeMs) {
          // Sometimes add, sometimes delete.
          if (random.nextBoolean()) {
            MutableInodeDirectory dir =
                createInodeDir(random.nextLong(10, 15), random.nextLong(1, 5));
            mStore.addChild(dir.getParentId(), dir);
          } else {
            mStore.removeChild(dirs.get(random.nextInt(dirs.size())).getId(),
                Long.toString(random.nextLong(10, 15)));
          }
          operations.incrementAndGet();
          assertTrue(mStore.mEdgeCache.mMap.size() <= CACHE_SIZE + numThreads);
        }
        return null;
      }));
      alluxio.util.CommonUtils.waitFor("eviction thread to finish",
          () -> mStore.mEdgeCache.mEvictionThread.mIsSleeping);
      mStore.mEdgeCache.verifyIndices();
    }
  
    @Test
    public void listingCacheManyDirsEviction() throws Exception {
      for (int i = 1; i < CACHE_SIZE * 3; i++) {
        createInodeDir(i, TEST_INODE_ID);
      }
      assertFalse(mStore.mListingCache.getCachedChildIds(TEST_INODE_ID).isPresent());
    }
  
    @Test
    public void listingCacheBigDirEviction() throws Exception {
      MutableInodeDirectory bigDir = createInodeDir(1, 0);
      long dirSize = CACHE_SIZE;
      for (int i = 10; i < 10 + dirSize; i++) {
        mStore.addChild(bigDir.getId(), createInodeDir(i, bigDir.getId()));
      }
      // Cache the large directory
      assertEquals(dirSize, CloseableIterator.size(mStore.getChildIds(bigDir.getId())));
      // Perform another operation to trigger eviction
      mStore.addChild(bigDir.getId(), createInodeDir(10000, bigDir.getId()));
      assertFalse(mStore.mListingCache.getCachedChildIds(TEST_INODE_ID).isPresent());
    }
  
    @Test(timeout = 10000)
    public void listingCacheAddRemoveEdges() throws Exception {
      // Perform operations including adding and removing many files within a directory. This test has
      // rooted out some bugs related to cache weight tracking.
      MutableInodeDirectory bigDir = createInodeDir(1, 0);
      mStore.writeNewInode(bigDir);
      for (int i = 1000; i < 1000 + CACHE_SIZE; i++) {
        MutableInodeDirectory subDir = createInodeDir(i, bigDir.getId());
        mStore.addChild(bigDir.getId(), subDir);
        mStore.removeChild(bigDir.getId(), subDir.getName());
      }
  
      List<MutableInodeDirectory> inodes = new ArrayList<>();
      for (int i = 10; i < 10 + (CACHE_SIZE / 2); i++) {
        MutableInodeDirectory otherDir = createInodeDir(i, 0);
        inodes.add(otherDir);
        mStore.writeNewInode(otherDir);
      }
      for (MutableInodeDirectory inode : inodes) {
        for (int i = 0; i < 10; i++) {
          assertEquals(0, CloseableIterator.size(mStore.getChildIds(inode.getId())));
        }
        verify(mBackingStore, times(0)).getChildIds(inode.getId());
      }
    }
  
    @Test
    public void flushToBackingStore() throws Exception {
      for (long inodeId = 10; inodeId < 10 + CACHE_SIZE / 2; inodeId++) {
        MutableInodeDirectory dir = createInodeDir(inodeId, 0);
        mStore.addChild(0, dir);
      }
      assertEquals(0, CloseableIterator.size(mBackingStore.getChildren(0L)));
      mStore.mEdgeCache.flush();
      mStore.mInodeCache.flush();
      assertEquals(CACHE_SIZE / 2, CloseableIterator.size(mBackingStore.getChildren(0L)));
    }
  
    private MutableInodeDirectory createInodeDir(long id, long parentId) {
      MutableInodeDirectory dir = MutableInodeDirectory.create(id, parentId, Long.toString(id),
          CreateDirectoryContext.defaults());
      mStore.writeNewInode(dir);
      return dir;
    }
  
    @Test
    public void backupRestore() throws Exception {
      MutableInodeDirectory child = createInodeDir(10, 0);
      mStore.writeNewInode(child);
      mStore.addChild(0, child);
      mStore
          .writeInode(MutableInodeDirectory.create(1, 1, "blah", CreateDirectoryContext.defaults()));
  
      ByteArrayOutputStream baos = new ByteArrayOutputStream();
      mStore.writeToCheckpoint(baos);
      mStore.restoreFromCheckpoint(
          new CheckpointInputStream(new ByteArrayInputStream(baos.toByteArray())));
      assertEquals(child.getName(), mStore.get(child.getId()).get().getName());
      assertEquals(child.getId(), mStore.getChild(0L, child.getName()).get().getId());
    }
  
    @Test
    public void skipCache() throws Exception {
      assertEquals(1, mStore.mInodeCache.getCacheMap().size());
      mStore.mInodeCache.flush();
      mStore.mInodeCache.clear();
      assertEquals(0, mStore.mInodeCache.getCacheMap().size());
      assertEquals(Inode.wrap(TEST_INODE_DIR),
          mStore.get(TEST_INODE_ID, ReadOption.newBuilder().setSkipCache(true).build()).get());
      assertEquals(0, mStore.mInodeCache.getCacheMap().size());
    }
  
    private void verifyNoBackingStoreReads() {
      verify(mBackingStore, times(0)).getChild(anyLong(), anyString());
      verify(mBackingStore, times(0)).getChildId(anyLong(), anyString());
      verify(mBackingStore, times(0)).getChildren(anyLong());
      verify(mBackingStore, times(0)).getChildIds(anyLong());
      verify(mBackingStore, times(0)).get(anyLong());
      verify(mBackingStore, times(0)).getMutable(anyLong());
    }
  }
  