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 alluxio.master.metastore.heap.HeapInodeStore.sortedMapToIterator;
  import static java.util.stream.Collectors.toSet;
  
  import alluxio.collections.TwoKeyConcurrentSortedMap;
  import alluxio.concurrent.LockMode;
  import alluxio.conf.AlluxioConfiguration;
  import alluxio.conf.Configuration;
  import alluxio.conf.PropertyKey;
  import alluxio.exception.runtime.AlluxioRuntimeException;
  import alluxio.grpc.ErrorType;
  import alluxio.master.file.meta.Edge;
  import alluxio.master.file.meta.EdgeEntry;
  import alluxio.master.file.meta.Inode;
  import alluxio.master.file.meta.InodeDirectoryView;
  import alluxio.master.file.meta.InodeLockManager;
  import alluxio.master.file.meta.MutableInode;
  import alluxio.master.file.meta.MutableInodeDirectory;
  import alluxio.master.file.meta.MutableInodeFile;
  import alluxio.master.journal.checkpoint.CheckpointInputStream;
  import alluxio.master.journal.checkpoint.CheckpointName;
  import alluxio.master.metastore.InodeStore;
  import alluxio.master.metastore.ReadOption;
  import alluxio.master.metastore.heap.HeapInodeStore;
  import alluxio.metrics.MetricKey;
  import alluxio.metrics.MetricsSystem;
  import alluxio.resource.CloseableIterator;
  import alluxio.resource.LockResource;
  import alluxio.resource.RWLockResource;
  import alluxio.util.ConfigurationUtils;
  import alluxio.util.ObjectSizeCalculator;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.base.Preconditions;
  import com.google.common.base.Stopwatch;
  import com.google.common.base.Throwables;
  import com.google.common.collect.ImmutableSet;
  import com.google.common.collect.Sets;
  import com.google.common.io.Closer;
  import io.grpc.Status;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.Closeable;
  import java.io.File;
  import java.io.IOException;
  import java.io.OutputStream;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Map.Entry;
  import java.util.Optional;
  import java.util.Set;
  import java.util.SortedMap;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ConcurrentSkipListMap;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.concurrent.atomic.AtomicLong;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantLock;
  import javax.annotation.Nullable;
  import javax.annotation.concurrent.ThreadSafe;
  
  /**
   * An inode store which caches inode tree metadata and delegates to another inode store for cache
   * misses.
   * <p>
   * We call the other inode store the backing store. Backing store operations are much slower than
   * on-heap operations, so we aim to serve as many requests as possible from the cache.
   * <p>
   * When the inode tree fits completely within the cache, we never interact with the backing store,
   * so performance should be similar to {@link HeapInodeStore}. Once the cache reaches the high
   * watermark, we begin evicting metadata to the backing store, and performance becomes dependent on
   * cache hit rate and backing store performance.
   *
   * <h1>Implementation</h1>
   * <p>
   * The CachingInodeStore uses 3 caches: an inode cache, and edge cache, and a listing cache. The
   * inode cache stores inode metadata such as inode names, permissions, and sizes. The edge cache
   * stores edge metadata, i.e. which inodes are children of which other inodes. The listing cache
   * caches the results of calling getChildren.
   * <p>
   * See the javadoc for {@link InodeCache}, {@link EdgeCache}, and {@link ListingCache} for details
   * about their inner workings.
   */
  @ThreadSafe
  public final class CachingInodeStore implements InodeStore, Closeable {
    private static final Logger LOG = LoggerFactory.getLogger(CachingInodeStore.class);
  
    private final InodeStore mBackingStore;
    private final InodeLockManager mLockManager;
  
    // Cache recently-accessed inodes.
    @VisibleForTesting
    final InodeCache mInodeCache;
  
    // Cache recently-accessed inode tree edges.
    @VisibleForTesting
    final EdgeCache mEdgeCache;
  
    @VisibleForTesting
    final ListingCache mListingCache;
  
    // Starts true, but becomes permanently false if we ever need to spill metadata to the backing
    // store. When true, we can optimize lookups for non-existent inodes because we don't need to
    // check the backing store. We can also optimize getChildren by skipping the range query on the
    // backing store.
    private volatile boolean mBackingStoreEmpty;
  
    /**
     * @param backingStore the backing inode store
     * @param lockManager  inode lock manager
     */
    public CachingInodeStore(InodeStore backingStore, InodeLockManager lockManager) {
      mBackingStore = backingStore;
      mLockManager = lockManager;
      AlluxioConfiguration conf = Configuration.global();
      int maxSize = conf.getInt(PropertyKey.MASTER_METASTORE_INODE_CACHE_MAX_SIZE);
      Preconditions.checkState(maxSize > 0,
          "Maximum cache size %s must be positive, but is set to %s",
          PropertyKey.MASTER_METASTORE_INODE_CACHE_MAX_SIZE.getName(), maxSize);
      float highWaterMarkRatio = ConfigurationUtils.checkRatio(conf,
          PropertyKey.MASTER_METASTORE_INODE_CACHE_HIGH_WATER_MARK_RATIO);
      int highWaterMark = Math.round(maxSize * highWaterMarkRatio);
      float lowWaterMarkRatio = ConfigurationUtils.checkRatio(conf,
          PropertyKey.MASTER_METASTORE_INODE_CACHE_LOW_WATER_MARK_RATIO);
      Preconditions.checkState(lowWaterMarkRatio <= highWaterMarkRatio,
          "low water mark ratio (%s=%s) must not exceed high water mark ratio (%s=%s)",
          PropertyKey.MASTER_METASTORE_INODE_CACHE_LOW_WATER_MARK_RATIO.getName(), lowWaterMarkRatio,
          PropertyKey.MASTER_METASTORE_INODE_CACHE_HIGH_WATER_MARK_RATIO, highWaterMarkRatio);
      int lowWaterMark = Math.round(maxSize * lowWaterMarkRatio);
  
      mBackingStoreEmpty = true;
      CacheConfiguration cacheConf = CacheConfiguration.newBuilder().setMaxSize(maxSize)
          .setHighWaterMark(highWaterMark).setLowWaterMark(lowWaterMark)
          .setEvictBatchSize(conf.getInt(PropertyKey.MASTER_METASTORE_INODE_CACHE_EVICT_BATCH_SIZE))
          .build();
      mInodeCache = new InodeCache(cacheConf);
      mEdgeCache = new EdgeCache(cacheConf);
      mListingCache = new ListingCache(cacheConf);
      if (conf.getBoolean(PropertyKey.MASTER_METRICS_HEAP_ENABLED)) {
        MetricsSystem.registerCachedGaugeIfAbsent(MetricKey.MASTER_INODE_HEAP_SIZE.getName(),
            () -> {
              try {
                return ObjectSizeCalculator.getObjectSize(mInodeCache.mMap,
                    ImmutableSet.of(Long.class, MutableInodeFile.class, MutableInodeDirectory.class))
                    + ObjectSizeCalculator.getObjectSize(mEdgeCache.mMap,
                    ImmutableSet.of(Long.class, Edge.class))
                    + ObjectSizeCalculator.getObjectSize(mListingCache.mMap,
                    ImmutableSet.of(Long.class, ListingCache.ListingCacheEntry.class));
              } catch (NullPointerException e) {
                LOG.info(Throwables.getStackTraceAsString(e));
                throw e;
              }
            });
      }
    }
  
    @Override
    public Optional<MutableInode<?>> getMutable(long id, ReadOption option) {
      return mInodeCache.get(id, option);
    }
  
    @Override
    public void remove(Long inodeId) {
      mInodeCache.remove(inodeId);
    }
  
    @Override
    public void writeInode(MutableInode<?> inode) {
      mInodeCache.put(inode.getId(), inode);
    }
  
    @Override
    public void writeNewInode(MutableInode<?> inode) {
      if (inode.isDirectory()) {
        mListingCache.addEmptyDirectory(inode.getId());
      }
      mInodeCache.putNewEntry(inode.getId(), inode);
    }
  
    @Override
    public void clear() {
      mInodeCache.clear();
      mEdgeCache.clear();
      mBackingStore.clear();
    }
  
    @Override
    public void addChild(long parentId, String childName, Long childId) {
      mEdgeCache.put(new Edge(parentId, childName), childId);
    }
  
    @Override
    public void removeChild(long parentId, String name) {
      mEdgeCache.remove(new Edge(parentId, name));
    }
  
    @Override
    public CloseableIterator<Long> getChildIds(Long inodeId, ReadOption option) {
      return mListingCache.getChildIds(inodeId, option);
    }
  
    @Override
    public Optional<Long> getChildId(Long inodeId, String name, ReadOption option) {
      return mEdgeCache.get(new Edge(inodeId, name), option);
    }
  
    @Override
    public Optional<Inode> getChild(Long inodeId, String name, ReadOption option) {
      return mEdgeCache.get(new Edge(inodeId, name), option).flatMap(this::get);
    }
  
    @Override
    public boolean hasChildren(InodeDirectoryView inode, ReadOption option) {
      Optional<Collection<Long>> cached = mListingCache.getCachedChildIds(inode.getId());
      if (cached.isPresent()) {
        return !cached.get().isEmpty();
      }
      return !mListingCache.getDataFromBackingStore(inode.getId(), option).isEmpty()
          || mBackingStore.hasChildren(inode);
    }
  
    @VisibleForTesting
    @Override
    public Set<EdgeEntry> allEdges() {
      return mEdgeCache.allEdges();
    }
  
    @VisibleForTesting
    @Override
    public Set<MutableInode<?>> allInodes() {
      return mInodeCache.allInodes();
    }
  
    @Override
    public void close() {
      Closer closer = Closer.create();
      // Close the backing store last so that cache eviction threads don't hit errors.
      closer.register(mBackingStore);
      closer.register(mInodeCache);
      closer.register(mEdgeCache);
      try {
        closer.close();
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
    }
  
    @Override
    public CheckpointName getCheckpointName() {
      return CheckpointName.CACHING_INODE_STORE;
    }
  
    @Override
    public CompletableFuture<Void> writeToCheckpoint(File directory,
                                                     ExecutorService executorService) {
      return CompletableFuture.runAsync(() -> {
        LOG.info("Flushing inodes to backing store");
        try {
          mInodeCache.flush();
          mEdgeCache.flush();
        } catch (InterruptedException e) {
          throw new AlluxioRuntimeException(Status.INTERNAL,
              String.format("Failed to restore snapshot %s", getCheckpointName()),
              null, ErrorType.Internal, false);
        }
        LOG.info("Finished flushing inodes to backing store");
        mBackingStore.writeToCheckpoint(directory, executorService).join();
      }, executorService);
    }
  
    @Override
    public void writeToCheckpoint(OutputStream output) throws IOException, InterruptedException {
      LOG.info("Flushing inodes to backing store");
      mInodeCache.flush();
      mEdgeCache.flush();
      LOG.info("Finished flushing inodes to backing store");
      mBackingStore.writeToCheckpoint(output);
    }
  
    @Override
    public CompletableFuture<Void> restoreFromCheckpoint(File directory,
                                                         ExecutorService executorService) {
      return CompletableFuture.runAsync(() -> {
        mInodeCache.clear();
        mEdgeCache.clear();
        mListingCache.clear();
        mBackingStore.restoreFromCheckpoint(directory, executorService).join();
        mBackingStoreEmpty = false;
      }, executorService);
    }
  
    @Override
    public void restoreFromCheckpoint(CheckpointInputStream input) throws IOException {
      mInodeCache.clear();
      mEdgeCache.clear();
      mListingCache.clear();
      mBackingStore.restoreFromCheckpoint(input);
      mBackingStoreEmpty = false;
    }
  
    /**
     * Cache for inode metadata.
     * <p>
     * The cache supports high concurrency across different inode ids, but requires external
     * synchronization for operations on the same inode id. All inodes modifications must hold at
     * least an mLockManager read lock on the modified inode. This allows the cache to flush inodes
     * asynchronously by acquiring a write lock before serializing the inode.
     */
    @VisibleForTesting
    class InodeCache extends Cache<Long, MutableInode<?>> {
      public InodeCache(CacheConfiguration conf) {
        super(conf, "inode-cache", MetricKey.MASTER_INODE_CACHE_EVICTIONS,
            MetricKey.MASTER_INODE_CACHE_HITS, MetricKey.MASTER_INODE_CACHE_LOAD_TIMES,
            MetricKey.MASTER_INODE_CACHE_MISSES, MetricKey.MASTER_INODE_CACHE_SIZE);
      }
  
      @Override
      protected void onPut(
          Long id, @Nullable MutableInode<?> existingInode, MutableInode<?> inode, boolean newEntry) {
        if (newEntry && existingInode != null && inode != null
            && !existingInode.getName().equals(inode.getName())) {
          LOG.error(
              "[InodeTreeCorruption] trying writing the inode name {} id {}, parent id {}, "
                  + "but a different inode name {} id {} parent id {} already exists. "
                  + "Your journal files are probably corrupted!",
              inode.getName(), inode.getId(), inode.getParentId(),
              existingInode.getName(), existingInode.getId(), existingInode.getParentId());
          if (LOG.isDebugEnabled()) {
            LOG.debug("[InodeTreeCorruption] Existing inode: {}, new written inode: {}",
                getInodePathString(existingInode), getInodePathString(inode));
          }
        }
      }
  
      @Override
      protected Optional<MutableInode<?>> load(Long id) {
        if (mBackingStoreEmpty) {
          return Optional.empty();
        }
        return mBackingStore.getMutable(id, ReadOption.defaults());
      }
  
      @Override
      protected void writeToBackingStore(Long key, MutableInode<?> value) {
        mBackingStoreEmpty = false;
        mBackingStore.writeInode(value);
      }
  
      @Override
      protected void removeFromBackingStore(Long key) {
        if (!mBackingStoreEmpty) {
          mBackingStore.remove(key);
        }
      }
  
      @Override
      protected void flushEntries(List<Entry> entries) {
        mBackingStoreEmpty = false;
        boolean useBatch = entries.size() > 0 && mBackingStore.supportsBatchWrite();
        try (WriteBatch batch = useBatch ? mBackingStore.createWriteBatch() : null) {
          for (Entry entry : entries) {
            Long inodeId = entry.mKey;
            Optional<RWLockResource> lockOpt = mLockManager.tryLockInode(inodeId, LockMode.WRITE);
            if (!lockOpt.isPresent()) {
              continue;
            }
            try (LockResource lr = lockOpt.get()) {
              if (entry.mValue == null) {
                if (useBatch) {
                  batch.removeInode(inodeId);
                } else {
                  mBackingStore.remove(inodeId);
                }
              } else {
                if (useBatch) {
                  batch.writeInode(entry.mValue);
                } else {
                  mBackingStore.writeInode(entry.mValue);
                }
              }
              entry.mDirty = false;
            }
          }
          if (useBatch) {
            batch.commit();
          }
        }
      }
  
      private Set<MutableInode<?>> allInodes() {
        Set<MutableInode<?>> cached = mInodeCache.getCacheMap().values().stream()
            .filter(entry -> entry.mValue != null).map(entry -> entry.mValue).collect(toSet());
        Set<Long> unflushedRemoves = mInodeCache.getCacheMap().values().stream()
            .filter(entry -> entry.mValue == null).map(entry -> entry.mKey).collect(toSet());
        Set<MutableInode<?>> flushed = mBackingStore.allInodes().stream()
            .filter(inode -> !unflushedRemoves.contains(inode.getId())).collect(toSet());
        return Sets.union(cached, flushed);
      }
    }
  
    /**
     * Cache for edge metadata.
     * <p>
     * The edge cache is responsible for managing the mapping from (parentId, childName) to childId.
     * This works similarly to the inode cache.
     * <p>
     * To support getChildIds, the edge cache maintains two indexes: - mIdToChildMap indexes the cache
     * by parent id for fast lookups. - mUnflushedDeletes indexes the cache's "removal" entries by
     * parent id. Removal entries exist for edges which have been removed from the cache, but not yet
     * removed from the backing store.
     */
    @VisibleForTesting
    class EdgeCache extends Cache<Edge, Long> {
      // Indexes non-removed cache entries by parent id. The inner map is from child name to child id
      @VisibleForTesting
      TwoKeyConcurrentSortedMap<Long, String, Long, SortedMap<String, Long>>
          mIdToChildMap = new TwoKeyConcurrentSortedMap<>(ConcurrentSkipListMap::new);
      // Indexes removed cache entries by parent id. The inner set contains the names of deleted
      // children.
      @VisibleForTesting
      Map<Long, Set<String>> mUnflushedDeletes = new ConcurrentHashMap<>();
  
      public EdgeCache(CacheConfiguration conf) {
        super(conf, "edge-cache", MetricKey.MASTER_EDGE_CACHE_EVICTIONS,
            MetricKey.MASTER_EDGE_CACHE_HITS, MetricKey.MASTER_EDGE_CACHE_LOAD_TIMES,
            MetricKey.MASTER_EDGE_CACHE_MISSES, MetricKey.MASTER_EDGE_CACHE_SIZE);
      }
  
      /**
       * Gets the child ids for an inode. This searches the on-heap cache as well as the backing
       * store.
       * <p>
       * Consistency guarantees
       * <p>
       * 1. getChildIds will return all children that existed before getChildIds was invoked. If a
       * child is concurrently removed during the call to getChildIds, it is undefined whether it gets
       * found. 2. getChildIds will never return a child that was removed before getChildIds was
       * invoked. If a child is concurrently added during the call to getChildIds, it is undefined
       * whether it gets found.
       *
       * @param inodeId the inode to get the children for
       * @param option  the read options
       * @return the children
       */
      public SortedMap<String, Long> getChildIds(Long inodeId, ReadOption option) {
        if (mBackingStoreEmpty) {
          return mIdToChildMap.getOrDefault(inodeId, Collections.emptySortedMap());
        }
        // This implementation must be careful because edges can be asynchronously evicted from the
        // cache to the backing store. To account for this, we read from the cache before consulting
        // the backing store.
        SortedMap<String, Long> childIds = new ConcurrentSkipListMap<>(mIdToChildMap.getOrDefault(
            inodeId, Collections.emptySortedMap()));
        // Copy the list of unflushed deletes before reading the backing store to prevent racing async
        // deletion.
        Set<String> unflushedDeletes =
            new HashSet<>(mUnflushedDeletes.getOrDefault(inodeId, Collections.emptySet()));
        // Cannot use mBackingStore.getChildren because it only returns inodes cached in the backing
        // store, causing us to lose inodes stored only in the cache.
        try (CloseableIterator<Long> childIter = mBackingStore.getChildIds(inodeId)) {
          childIter.forEachRemaining(childId -> {
            CachingInodeStore.this.get(childId, option).map(inode -> {
              if (!unflushedDeletes.contains(inode.getName())) {
                childIds.put(inode.getName(), inode.getId());
              }
              return null;
            });
          });
        }
        return childIds;
      }
  
      @Override
      protected Optional<Long> load(Edge edge) {
        if (mBackingStoreEmpty) {
          return Optional.empty();
        }
        return mBackingStore.getChildId(edge.getId(), edge.getName());
      }
  
      @Override
      protected void writeToBackingStore(Edge key, Long value) {
        mBackingStoreEmpty = false;
        mBackingStore.addChild(key.getId(), key.getName(), value);
      }
  
      @Override
      protected void removeFromBackingStore(Edge key) {
        if (!mBackingStoreEmpty) {
          mBackingStore.removeChild(key.getId(), key.getName());
        }
      }
  
      @Override
      protected void flushEntries(List<Entry> entries) {
        mBackingStoreEmpty = false;
        boolean useBatch = entries.size() > 0 && mBackingStore.supportsBatchWrite();
        try (WriteBatch batch = useBatch ? mBackingStore.createWriteBatch() : null) {
          for (Entry entry : entries) {
            Edge edge = entry.mKey;
            Optional<RWLockResource> lockOpt = mLockManager.tryLockEdge(edge, LockMode.WRITE);
            if (!lockOpt.isPresent()) {
              continue;
            }
            try (LockResource lr = lockOpt.get()) {
              Long value = entry.mValue;
              if (value == null) {
                if (useBatch) {
                  batch.removeChild(edge.getId(), edge.getName());
                } else {
                  mBackingStore.removeChild(edge.getId(), edge.getName());
                }
              } else {
                if (useBatch) {
                  batch.addChild(edge.getId(), edge.getName(), value);
                } else {
                  mBackingStore.addChild(edge.getId(), edge.getName(), value);
                }
              }
              entry.mDirty = false;
            }
          }
          if (useBatch) {
            batch.commit();
          }
        }
      }
  
      @Override
      protected void onCacheUpdate(Edge edge, Long childId) {
        if (childId == null) {
          mIdToChildMap.removeInnerValue(edge.getId(), edge.getName());
          addToUnflushedDeletes(edge.getId(), edge.getName());
        } else {
          mIdToChildMap.addInnerValue(edge.getId(), edge.getName(), childId);
          removeFromUnflushedDeletes(edge.getId(), edge.getName());
        }
      }
  
      @Override
      protected void onCacheRemove(Edge edge) {
        mIdToChildMap.removeInnerValue(edge.getId(), edge.getName());
        removeFromUnflushedDeletes(edge.getId(), edge.getName());
      }
  
      @Override
      protected void onPut(Edge edge, Long ignored, Long childId, boolean newEntry) {
        mListingCache.addEdge(edge, childId);
      }
  
      @Override
      protected void onRemove(Edge edge) {
        mListingCache.removeEdge(edge);
      }
  
      private void addToUnflushedDeletes(long parentId, String childName) {
        mUnflushedDeletes.compute(parentId, (key, children) -> {
          if (children == null) {
            children = ConcurrentHashMap.newKeySet(4);
          }
          children.add(childName);
          return children;
        });
      }
  
      private void removeFromUnflushedDeletes(Long parentId, String childName) {
        mUnflushedDeletes.computeIfPresent(parentId, (key, children) -> {
          children.remove(childName);
          if (children.isEmpty()) {
            return null;
          }
          return children;
        });
      }
  
      private Set<EdgeEntry> allEdges() {
        Set<EdgeEntry> cacheEntries =
            mIdToChildMap.flattenEntries((a, b, c) -> new EdgeEntry(a, b, c));
        Set<EdgeEntry> backingStoreEntries =
            mBackingStore.allEdges().stream()
                .filter(edge -> !mUnflushedDeletes
                    .getOrDefault(edge.getParentId(), Collections.emptySet())
                    .contains(edge.getChildName()))
                .collect(toSet());
        return Sets.union(cacheEntries, backingStoreEntries);
      }
  
      @VisibleForTesting
      void verifyIndices() {
        mMap.forEachValue(1, entry -> {
          if (entry.mValue == null) {
            if (!mUnflushedDeletes.get(entry.mKey.getId()).contains(entry.mKey.getName())) {
              throw new IllegalStateException(
                  "Missing entry " + entry.mKey + " in unflushed deletes index");
            }
          } else {
            if (!mIdToChildMap.get(entry.mKey.getId()).get(entry.mKey.getName())
                .equals(entry.mValue)) {
              throw new IllegalStateException(String
                  .format("Missing entry %s=%s from id to child map", entry.mKey, entry.mValue));
            }
          }
        });
        mIdToChildMap.flattenEntries((parentId, childName, childId) -> {
          if (!mMap.get(new Edge(parentId, childName)).mValue.equals(childId)) {
            throw new IllegalStateException(String.format(
                "Entry %s->%s=%s exists in the index but not the map", parentId, childName, childId));
          }
          return null;
        });
        mUnflushedDeletes.forEach((id, names) -> {
          for (String name : names) {
            if (mMap.get(new Edge(id, name)).mValue != null) {
              throw new IllegalStateException(String.format(
                  "Entry %s->%s exists in the unflushed index but not in the map", id, name));
            }
          }
        });
      }
    }
  
    /**
     * Cache for caching the results of listing directory children.
     * <p>
     * Unlike the inode and edge caches, this cache is not a source of truth. It just does its best to
     * pre-compute getChildIds results so that they can be served quickly.
     * <p>
     * The listing cache contains a mapping from parent id to a complete set of its children. Listings
     * are always complete (containing all children in both the edge cache and the backing store). To
     * maintain our listings, the edge cache calls {@link ListingCache#addEdge(Edge, Long)} and
     * {@link ListingCache#removeEdge(Edge)} whenever edges are added or removed. When new directories
     * are created, the inode cache calls {@link ListingCache#addEmptyDirectory(long)}. This means
     * that with enough cache space, all list requests can be served without going to the backing
     * store.
     * <p>
     * getChildIds may be called concurrently with operations that add or remove child ids from the
     * directory in question. To account for this, we use a mModified field in ListingCacheEntry to
     * record whether any modifications are made to the directory while we are listing it. If any
     * modifications are detected, we cannot safely cache, so we skip caching.
     * <p>
     * The listing cache tracks its size by weight. The weight for each entry is one plus the size of
     * the listing. Once the weight reaches the high water mark, the first thread to acquire the
     * eviction lock will evict down to the low watermark before computing and caching its result.
     */
    @VisibleForTesting
    class ListingCache {
      private final int mMaxSize;
      private final int mHighWaterMark;
      private final int mLowWaterMark;
      private final AtomicLong mWeight = new AtomicLong(0);
      private final Lock mEvictionLock = new ReentrantLock();
  
      StatsCounter mStatsCounter;
  
      private final Map<Long, ListingCacheEntry> mMap = new ConcurrentHashMap<>();
      private Iterator<Map.Entry<Long, ListingCacheEntry>> mEvictionHead = mMap.entrySet().iterator();
  
      private ListingCache(CacheConfiguration conf) {
        mMaxSize = conf.getMaxSize();
        mHighWaterMark = conf.getHighWaterMark();
        mLowWaterMark = conf.getLowWaterMark();
  
        mStatsCounter = new StatsCounter(
            MetricKey.MASTER_LISTING_CACHE_EVICTIONS,
            MetricKey.MASTER_LISTING_CACHE_HITS,
            MetricKey.MASTER_LISTING_CACHE_LOAD_TIMES,
            MetricKey.MASTER_LISTING_CACHE_MISSES);
        MetricsSystem.registerGaugeIfAbsent(MetricKey.MASTER_LISTING_CACHE_SIZE.getName(),
            mWeight::get);
      }
  
      /**
       * Notifies the cache of a newly created empty directory.
       * <p>
       * This way, we can have a cache hit on the first time the directory is listed.
       *
       * @param inodeId the inode id of the directory
       */
      public void addEmptyDirectory(long inodeId) {
        evictIfNecessary();
        mMap.computeIfAbsent(inodeId, x -> {
          mWeight.incrementAndGet();
          ListingCacheEntry entry = new ListingCacheEntry();
          entry.mChildren = new ConcurrentSkipListMap<>();
          return entry;
        });
      }
  
      /**
       * Updates the cache for an added edge.
       *
       * @param edge    the edge to add
       * @param childId the child of the edge
       */
      public void addEdge(Edge edge, Long childId) {
        evictIfNecessary();
        mMap.computeIfPresent(edge.getId(), (key, entry) -> {
          entry.mModified = true;
          entry.addChild(edge.getName(), childId);
          return entry;
        });
      }
  
      /**
       * Updates the cache for a removed edge.
       *
       * @param edge the removed edge
       */
      public void removeEdge(Edge edge) {
        mMap.computeIfPresent(edge.getId(), (key, entry) -> {
          entry.mModified = true;
          entry.removeChild(edge.getName());
          return entry;
        });
      }
  
      /**
       * Returns the children of the given inode, if the child list is cached. This method only
       * consults the cache, and never looks in the backing store.
       *
       * @param inodeId the directory to list
       * @return the children of the inode, or empty if the child list isn't cached
       */
      public Optional<Collection<Long>> getCachedChildIds(Long inodeId) {
        ListingCacheEntry entry = mMap.get(inodeId);
        if (entry != null && entry.mChildren != null) {
          mStatsCounter.recordHit();
          entry.mReferenced = true;
          return Optional.of(entry.mChildren.values());
        }
        mStatsCounter.recordMiss();
        return Optional.empty();
      }
  
      /**
       * Gets all children of an inode, falling back on the edge cache if the listing isn't cached.
       *
       * @param inodeId the inode directory id
       * @param option  the read options
       * @return the ids of all children of the directory
       */
      public CloseableIterator<Long> getChildIds(Long inodeId, ReadOption option) {
        evictIfNecessary();
        AtomicBoolean createdNewEntry = new AtomicBoolean(false);
        ListingCacheEntry entry = mMap.compute(inodeId, (key, value) -> {
          if (value == null) {
            mStatsCounter.recordMiss();
            if (mWeight.get() >= mMaxSize) {
              return null;
            }
            createdNewEntry.set(true);
            return new ListingCacheEntry();
          }
          mStatsCounter.recordHit();
          value.mReferenced = true;
          return value;
        });
        SortedMap<String, Long> childMap;
        if (entry != null && entry.mChildren != null) {
          childMap = entry.mChildren;
        } else if (entry == null || !createdNewEntry.get() || option.shouldSkipCache()) {
          // Skip caching if the cache is full or someone else is already caching.
          childMap = getDataFromBackingStore(inodeId, option);
        } else {
          childMap = loadChildren(inodeId, entry, option);
        }
        return CloseableIterator.noopCloseable(sortedMapToIterator(childMap, option));
      }
  
      public void clear() {
        mMap.clear();
        mWeight.set(0);
        mEvictionHead = mMap.entrySet().iterator();
      }
  
      private SortedMap<String, Long> loadChildren(Long inodeId, ListingCacheEntry entry,
          ReadOption option) {
        evictIfNecessary();
        entry.mModified = false;
        SortedMap<String, Long> listing = getDataFromBackingStore(inodeId, option);
        mMap.computeIfPresent(inodeId, (key, value) -> {
          // Perform the update inside computeIfPresent to prevent concurrent modification to the
          // cache entry.
          if (!entry.mModified) {
            entry.mChildren = new ConcurrentSkipListMap<>(listing);
            mWeight.addAndGet(weight(entry));
            return entry;
          }
          return null;
        });
        return listing;
      }
  
      private void evictIfNecessary() {
        if (mWeight.get() <= mHighWaterMark) {
          return;
        }
        if (mEvictionLock.tryLock()) {
          try {
            evict();
          } finally {
            mEvictionLock.unlock();
          }
        }
      }
  
      private void evict() {
        long startTime = System.currentTimeMillis();
        long evictTarget = mWeight.get() - mLowWaterMark;
        AtomicInteger evicted = new AtomicInteger(0);
        while (evicted.get() < evictTarget) {
          if (!mEvictionHead.hasNext()) {
            mEvictionHead = mMap.entrySet().iterator();
          }
          if (!mEvictionHead.hasNext()) {
            break; // cache is empty.
          }
          Entry<Long, ListingCacheEntry> candidate = mEvictionHead.next();
          if (candidate.getValue().mReferenced) {
            candidate.getValue().mReferenced = false;
            continue;
          }
          mMap.compute(candidate.getKey(), (key, entry) -> {
            if (entry != null && entry.mChildren != null) {
              mWeight.addAndGet(-weight(entry));
              evicted.addAndGet(weight(entry));
              return null;
            }
            return entry;
          });
        }
        LOG.debug("Evicted weight={} from listing cache down to weight={} in {}ms", evicted.get(),
            mWeight.get(), System.currentTimeMillis() - startTime);
      }
  
      private int weight(ListingCacheEntry entry) {
        Preconditions.checkNotNull(entry);
        Preconditions.checkNotNull(entry.mChildren);
        // Add 1 to take the key into account.
        return entry.mChildren.size() + 1;
      }
  
      private SortedMap<String, Long> getDataFromBackingStore(Long inodeId, ReadOption option) {
        final Stopwatch stopwatch = Stopwatch.createStarted();
        final SortedMap<String, Long> result = mEdgeCache.getChildIds(inodeId, option);
        mStatsCounter.recordLoad(stopwatch.elapsed(TimeUnit.NANOSECONDS));
        return result;
      }
  
      private class ListingCacheEntry {
        private volatile boolean mModified = false;
        private volatile boolean mReferenced = true;
        // null indicates that we are in the process of loading the children.
        @Nullable
        private volatile SortedMap<String, Long> mChildren = null;
  
        public void addChild(String name, Long id) {
          if (mChildren != null && mChildren.put(name, id) == null) {
            mWeight.incrementAndGet();
          }
        }
  
        public void removeChild(String name) {
          if (mChildren != null && mChildren.remove(name) != null) {
            mWeight.decrementAndGet();
          }
        }
      }
    }
  }
  