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 alluxio.Constants;
  import alluxio.master.metastore.ReadOption;
  import alluxio.metrics.MetricKey;
  import alluxio.metrics.MetricsSystem;
  import alluxio.util.logging.SamplingLogger;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.base.Stopwatch;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.Closeable;
  import java.lang.Thread.State;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.TimeUnit;
  import javax.annotation.Nullable;
  import javax.annotation.concurrent.ThreadSafe;
  
  /**
   * Base class for write-back caches which asynchronously evict entries to backing stores.
   *
   * The cache uses water mark based eviction. A dedicated thread waits for the cache to reach its
   * high water mark, then evicts entries until the cache size reaches the low water mark. All backing
   * store write operations are performed asynchronously in the eviction thread, unless the cache hits
   * maximum capacity. At maximum capacity, methods interact synchronously with the backing store. For
   * best performance, maximum capacity should never be reached. This requires that the eviction
   * thread can keep up cache writes.
   *
   * Cache hit reads are served without any locking. Writes and cache miss reads take locks on their
   * cache key.
   *
   * This class leverages the entry-level locks of ConcurrentHashMap to synchronize operations on the
   * same key.
   *
   * @param <K> the cache key type
   * @param <V> the cache value type
   */
  @ThreadSafe
  public abstract class Cache<K, V> implements Closeable {
    private static final Logger LOG = LoggerFactory.getLogger(Cache.class);
  
    private final int mMaxSize;
    private final int mHighWaterMark;
    private final int mLowWaterMark;
    private final int mEvictBatchSize;
    private final String mName;
    @VisibleForTesting
    final ConcurrentHashMap<K, Entry> mMap;
    // TODO(andrew): Support using multiple threads to speed up backing store writes.
    // Thread for performing eviction to the backing store.
    @VisibleForTesting
    final EvictionThread mEvictionThread;
  
    private final StatsCounter mStatsCounter;
  
    /**
     * @param conf cache configuration
     * @param name a name for the cache
     * @param evictionsKey the cache evictions metric key
     * @param hitsKey the cache hits metrics key
     * @param loadTimesKey the load times metrics key
     * @param missesKey the misses metrics key
     * @param sizeKey the size metrics key
     */
    public Cache(CacheConfiguration conf, String name, MetricKey evictionsKey, MetricKey hitsKey,
                 MetricKey loadTimesKey, MetricKey missesKey, MetricKey sizeKey) {
      mMaxSize = conf.getMaxSize();
      mHighWaterMark = conf.getHighWaterMark();
      mLowWaterMark = conf.getLowWaterMark();
      mEvictBatchSize = conf.getEvictBatchSize();
      mName = name;
      mMap = new ConcurrentHashMap<>(mMaxSize);
      mEvictionThread = new EvictionThread();
      mEvictionThread.setDaemon(true);
      // The eviction thread is started lazily when we first reach the high water mark.
      mStatsCounter = new StatsCounter(evictionsKey, hitsKey, loadTimesKey, missesKey);
  
      MetricsSystem.registerGaugeIfAbsent(sizeKey.getName(), mMap::size);
    }
  
    /**
     * Retrieves a value from the cache, loading it from the backing store if necessary.
     *
     * If the value needs to be loaded, concurrent calls to get(key) will block while waiting for the
     * first call to finish loading the value.
     *
     * If option.shouldSkipCache() is true, then the value loaded from the backing store will not be
     * cached and the eviction thread will not be woken up.
     *
     * @param key the key to get the value for
     * @param option the read options
     * @return the value, or empty if the key doesn't exist in the cache or in the backing store
     */
    public Optional<V> get(K key, ReadOption option) {
      if (option.shouldSkipCache()) {
        return getSkipCache(key);
      }
      if (cacheIsFull()) {
        wakeEvictionThreadIfNecessary();
        return getSkipCache(key);
      }
      Entry result = mMap.compute(key, (k, entry) -> {
        if (entry != null) {
          mStatsCounter.recordHit();
          entry.mReferenced = true;
          return entry;
        }
        mStatsCounter.recordMiss();
        final Stopwatch stopwatch = Stopwatch.createStarted();
        Optional<V> value = load(key);
        mStatsCounter.recordLoad(stopwatch.elapsed(TimeUnit.NANOSECONDS));
        if (value.isPresent()) {
          onCacheUpdate(key, value.get());
          Entry newEntry = new Entry(key, value.get());
          newEntry.mDirty = false;
          return newEntry;
        }
        return null;
      });
      if (result == null || result.mValue == null) {
        return Optional.empty();
      }
      wakeEvictionThreadIfNecessary();
      return Optional.of(result.mValue);
    }
  
    /**
     * @param key the key to get the value for
     * @return the result of {@link #get(Object, ReadOption)} with default option
     */
    public Optional<V> get(K key) {
      return get(key, ReadOption.defaults());
    }
  
    /**
     * Retrieves a value from the cache if already cached, otherwise, loads from the backing store
     * without caching the value. Eviction is not triggered.
     *
     * @param key the key to get the value for
     * @return the value, or empty if the key doesn't exist in the cache or in the backing store
     */
    private Optional<V> getSkipCache(K key) {
      Entry entry = mMap.get(key);
      if (entry == null) {
        mStatsCounter.recordMiss();
        final Stopwatch stopwatch = Stopwatch.createStarted();
        final Optional<V> result = load(key);
        mStatsCounter.recordLoad(stopwatch.elapsed(TimeUnit.NANOSECONDS));
        return result;
      }
      mStatsCounter.recordHit();
      return Optional.ofNullable(entry.mValue);
    }
  
    /**
     * Writes a key/value pair to the cache.
     * This method is similar to {@link #put(Object, Object)}, but with an added information that
     * the entry is new.
     *
     * @param key the key
     * @param value the value
     */
    public void putNewEntry(K key, V value) {
      putInternal(key, value, true);
    }
  
    /**
     * Writes a key/value pair to the cache.
     * If it is known that the entry is new, prefer {@link #putNewEntry(Object, Object)}.
     *
     * @param key the key
     * @param value the value
     */
    public void put(K key, V value) {
      putInternal(key, value, false);
    }
  
    private void putInternal(K key, V value, boolean isNewEntry) {
      mMap.compute(key, (k, entry) -> {
        V existingValue = entry == null ? null : entry.mValue;
        onPut(key, existingValue, value, isNewEntry);
        if (entry == null && cacheIsFull()) {
          writeToBackingStore(key, value);
          return null;
        }
        if (entry == null || entry.mValue == null) {
          onCacheUpdate(key, value);
          return new Entry(key, value);
        }
        entry.mValue = value;
        entry.mReferenced = true;
        entry.mDirty = true;
        return entry;
      });
      wakeEvictionThreadIfNecessary();
    }
  
    /**
     * Removes a key from the cache.
     *
     * The key is not immediately removed from the backing store. Instead, we set the entry's value to
     * null to indicate to the eviction thread that to evict the entry, it must first remove the key
     * from the backing store. However, if the cache is full we must synchronously write to the
     * backing store instead.
     *
     * @param key the key to remove
     */
    public void remove(K key) {
      // Set the entry so that it will be removed from the backing store when it is encountered by
      // the eviction thread.
      mMap.compute(key, (k, entry) -> {
        onRemove(key);
        if (entry == null && cacheIsFull()) {
          removeFromBackingStore(k);
          return null;
        }
        onCacheUpdate(key, null);
        if (entry == null) {
          entry = new Entry(key, null);
        } else {
          entry.mValue = null;
        }
        entry.mReferenced = false;
        entry.mDirty = true;
        return entry;
      });
      wakeEvictionThreadIfNecessary();
    }
  
    /**
     * Flushes all data to the backing store.
     */
    public void flush() throws InterruptedException {
      List<Entry> toFlush = new ArrayList<>(mEvictBatchSize);
      Iterator<Entry> it = mMap.values().iterator();
      while (it.hasNext()) {
        if (Thread.interrupted()) {
          throw new InterruptedException();
        }
        while (toFlush.size() < mEvictBatchSize && it.hasNext()) {
          Entry candidate = it.next();
          if (candidate.mDirty) {
            toFlush.add(candidate);
          }
        }
        flushEntries(toFlush);
        toFlush.clear();
      }
    }
  
    /**
     * Clears all entries from the map. This is not threadsafe, and requires external synchronization
     * to prevent concurrent modifications to the cache.
     */
    public void clear() {
      mMap.forEach((key, value) -> {
        onCacheUpdate(key, value.mValue);
        onRemove(key);
      });
      mMap.clear();
    }
  
    private boolean underHighWaterMark() {
      return mMap.size() < mHighWaterMark;
    }
  
    private boolean cacheIsFull() {
      return mMap.size() >= mMaxSize;
    }
  
    private void wakeEvictionThreadIfNecessary() {
      if (mEvictionThread.mIsSleeping && mMap.size() >= mHighWaterMark) {
        kickEvictionThread();
      }
    }
  
    private void kickEvictionThread() {
      synchronized (mEvictionThread) {
        if (mEvictionThread.getState() == State.NEW) {
          mEvictionThread.start();
        }
        mEvictionThread.notifyAll();
      }
    }
  
    @Override
    public void close() {
      mEvictionThread.interrupt();
      try {
        mEvictionThread.join(10L * Constants.SECOND_MS);
        if (mEvictionThread.isAlive()) {
          LOG.warn("Failed to stop eviction thread");
        }
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new RuntimeException(e);
      }
    }
  
    @VisibleForTesting
    class EvictionThread extends Thread {
      @VisibleForTesting
      volatile boolean mIsSleeping = true;
  
      // Populated with #fillBatch, cleared with #evictBatch. We keep it around so that we don't need
      // to keep re-allocating the list.
      private final List<Entry> mEvictionCandidates = new ArrayList<>(mEvictBatchSize);
      private final List<Entry> mDirtyEvictionCandidates = new ArrayList<>(mEvictBatchSize);
      private final Logger mCacheFullLogger = new SamplingLogger(LOG, 10L * Constants.SECOND_MS);
  
      private Iterator<Entry> mEvictionHead = Collections.emptyIterator();
  
      private EvictionThread() {
        super(mName + "-eviction-thread");
      }
  
      @Override
      public void run() {
        while (!Thread.interrupted()) {
          // Wait for the cache to get over the high water mark.
          while (underHighWaterMark()) {
            synchronized (mEvictionThread) {
              if (underHighWaterMark()) {
                try {
                  mIsSleeping = true;
                  mEvictionThread.wait();
                  mIsSleeping = false;
                } catch (InterruptedException e) {
                  return;
                }
              }
            }
          }
          if (cacheIsFull()) {
            mCacheFullLogger.warn(
                "Metastore {} cache is full. Consider increasing the cache size or lowering the "
                    + "high water mark. size:{} lowWaterMark:{} highWaterMark:{} maxSize:{}",
                mName, mMap.size(), mLowWaterMark, mHighWaterMark, mMaxSize);
          }
          evictToLowWaterMark();
        }
      }
  
      private void evictToLowWaterMark() {
        long evictionStart = System.nanoTime();
        int toEvict = mMap.size() - mLowWaterMark;
        int evictionCount = 0;
        while (evictionCount < toEvict) {
          if (!mEvictionHead.hasNext()) {
            mEvictionHead = mMap.values().iterator();
          }
          fillBatch(toEvict - evictionCount);
          evictionCount += evictBatch();
        }
        if (evictionCount > 0) {
          mStatsCounter.recordEvictions(evictionCount);
          LOG.debug("{}: Evicted {} entries in {}ms", mName, evictionCount,
              (System.nanoTime() - evictionStart) / Constants.MS_NANO);
        }
      }
  
      /**
       * Attempts to fill mEvictionCandidates with up to min(count, mEvictBatchSize) candidates for
       * eviction.
       *
       * @param count maximum number of entries to store in the batch
       */
      private void fillBatch(int count) {
        int targetSize = Math.min(count, mEvictBatchSize);
        while (mEvictionCandidates.size() < targetSize && mEvictionHead.hasNext()) {
          Entry candidate = mEvictionHead.next();
          if (candidate.mReferenced) {
            candidate.mReferenced = false;
            continue;
          }
          mEvictionCandidates.add(candidate);
          if (candidate.mDirty) {
            mDirtyEvictionCandidates.add(candidate);
          }
        }
      }
  
      /**
       * Attempts to evict all entries in mEvictionCandidates.
       *
       * @return the number of candidates actually evicted
       */
      private int evictBatch() {
        int evicted = 0;
        if (mEvictionCandidates.isEmpty()) {
          return evicted;
        }
        flushEntries(mDirtyEvictionCandidates);
        for (Entry entry : mEvictionCandidates) {
          if (evictIfClean(entry)) {
            evicted++;
          }
        }
        mEvictionCandidates.clear();
        mDirtyEvictionCandidates.clear();
        return evicted;
      }
  
      /**
       * @param entry the entry to try to evict
       * @return whether the entry was successfully evicted
       */
      private boolean evictIfClean(Entry entry) {
        return null == mMap.computeIfPresent(entry.mKey, (key, e) -> {
          if (entry.mDirty) {
            return entry; // entry must have been written since we evicted.
          }
          onCacheRemove(entry.mKey);
          return null;
        });
      }
    }
  
    @VisibleForTesting
    protected Map<K, Entry> getCacheMap() {
      return new HashMap<>(mMap);
    }
  
    //
    // Callbacks so that sub-classes can listen for cache changes. All callbacks on the same key
    // happen atomically with respect to each other and other cache operations.
    //
  
    /**
     * Callback triggered when an update is made to a key/value pair in the cache. For removals, value
     * will be null.
     *
     * @param key the updated key
     * @param value the updated value, or null if the key is being removed
     */
    protected void onCacheUpdate(K key, @Nullable V value) {}
  
    /**
     * Callback triggered when a key is removed from the cache.
     *
     * This may be used in conjunction with onCacheUpdate to keep track of all changes to the cache.
     *
     * @param key the removed key
     */
    protected void onCacheRemove(K key) {}
  
    /**
     * Callback triggered whenever a new key/value pair is added by put(key, value).
     *
     * @param key the added key
     * @param existingValue the current value if exists, otherwise null
     * @param value the added value
     * @param isNewKey a user input boolean indicates if the key is a new key or not
     */
    protected void onPut(K key, @Nullable V existingValue, V value, boolean isNewKey) {}
  
    /**
     * Callback triggered whenever a key is removed by remove(key).
     *
     * @param key the removed key
     */
    protected void onRemove(K key) {}
  
    /**
     * Loads a key from the backing store.
     *
     * @param key the key to load
     * @return the value for the key, or empty if the key doesn't exist in the backing store
     */
    protected abstract Optional<V> load(K key);
  
    /**
     * Writes a key/value pair to the backing store.
     *
     * @param key the key
     * @param value the value
     */
    protected abstract void writeToBackingStore(K key, V value);
  
    /**
     * Removes a key from the backing store.
     *
     * @param key the key
     */
    protected abstract void removeFromBackingStore(K key);
  
    /**
     * Attempts to flush the given entries to the backing store.
     *
     * The subclass is responsible for setting each candidate's mDirty field to false on success.
     *
     * @param candidates the candidate entries to flush
     */
    protected abstract void flushEntries(List<Entry> candidates);
  
    protected class Entry {
      protected K mKey;
      // null value means that the key has been removed from the cache, but still needs to be removed
      // from the backing store.
      @Nullable
      protected V mValue;
  
      // Whether the entry is out of sync with the backing store. If mDirty is true, the entry must be
      // flushed to the backing store before it can be evicted.
      protected volatile boolean mDirty = true;
  
      // Whether the entry has been recently accessed. Accesses set the bit to true, while the
      // eviction thread sets it to false. This is the same as the "referenced" bit described in the
      // CLOCK algorithm.
      private volatile boolean mReferenced = true;
  
      private Entry(K key, V value) {
        mKey = key;
        mValue = value;
      }
    }
  }
  