/*
 * 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 alluxio.client.file.cache.cuckoofilter.ConcurrentClockCuckooFilter.DEFAULT_FPP;
import static alluxio.client.file.cache.cuckoofilter.ConcurrentClockCuckooFilter.DEFAULT_LOAD_FACTOR;
import static java.util.concurrent.TimeUnit.MILLISECONDS;

import alluxio.client.file.cache.cuckoofilter.ClockCuckooFilter;
import alluxio.client.file.cache.cuckoofilter.ConcurrentClockCuckooFilter;
import alluxio.client.file.cache.cuckoofilter.SlidingWindowType;
import alluxio.client.quota.CacheScope;
import alluxio.conf.AlluxioConfiguration;
import alluxio.conf.PropertyKey;

import com.google.common.hash.Hashing;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicLong;

/**
 * This class is a shadow cache with {@link ClockCuckooFilter} implementation.
 */
public class ClockCuckooShadowCacheManager implements ShadowCacheManager {
  private static final int BITS_PER_TAG = 8;

  private final ScheduledExecutorService mScheduler = Executors.newScheduledThreadPool(0);
  private final ClockCuckooFilter<PageId> mFilter;
  private long mShadowCacheBytes = 0;
  private long mShadowCachePages = 0;

  private final AtomicLong mShadowCachePageRead = new AtomicLong(0);
  private final AtomicLong mShadowCachePageHit = new AtomicLong(0);
  private final AtomicLong mShadowCacheByteRead = new AtomicLong(0);
  private final AtomicLong mShadowCacheByteHit = new AtomicLong(0);

  /**
   * Create a ClockCuckooShadowCacheManager.
   *
   * @param conf the alluxio configuration
   */
  public ClockCuckooShadowCacheManager(AlluxioConfiguration conf) {
    long windowMs = conf.getMs(PropertyKey.USER_CLIENT_CACHE_SHADOW_WINDOW);
    long budgetInBits = conf.getBytes(PropertyKey.USER_CLIENT_CACHE_SHADOW_MEMORY_OVERHEAD) * 8;
    int bitsPerClock = conf.getInt(PropertyKey.USER_CLIENT_CACHE_SHADOW_CUCKOO_CLOCK_BITS);
    int bitsPerSize = conf.getInt(PropertyKey.USER_CLIENT_CACHE_SHADOW_CUCKOO_SIZE_BITS);
    int bitsPerScope = conf.getInt(PropertyKey.USER_CLIENT_CACHE_SHADOW_CUCKOO_SCOPE_BITS);
    boolean isSizeEncoderEnabled =
        conf.getBoolean(PropertyKey.USER_CLIENT_CACHE_SHADOW_CUCKOO_SIZE_ENCODER_ENABLED);
    long bitsPerSlot = BITS_PER_TAG + bitsPerClock + bitsPerSize + bitsPerScope;
    long totalSlots = budgetInBits / bitsPerSlot;
    long expectedInsertions = (long) (Long.highestOneBit(totalSlots) * DEFAULT_LOAD_FACTOR);
    if (isSizeEncoderEnabled) {
      int prefixBits = conf.getInt(PropertyKey.USER_CLIENT_CACHE_SHADOW_CUCKOO_SIZE_PREFIX_BITS);
      int suffixBits = conf.getInt(PropertyKey.USER_CLIENT_CACHE_SHADOW_CUCKOO_SIZE_SUFFIX_BITS);
      mFilter = ConcurrentClockCuckooFilter.create(CacheManagerWithShadowCache.PageIdFunnel.FUNNEL,
          expectedInsertions, bitsPerClock, bitsPerScope, prefixBits, suffixBits,
          SlidingWindowType.TIME_BASED, windowMs, DEFAULT_FPP, DEFAULT_LOAD_FACTOR,
          Hashing.murmur3_128());
    } else {
      mFilter = ConcurrentClockCuckooFilter.create(CacheManagerWithShadowCache.PageIdFunnel.FUNNEL,
          expectedInsertions, bitsPerClock, bitsPerSize, bitsPerScope, SlidingWindowType.TIME_BASED,
          windowMs);
    }
    long agingPeriod = windowMs >> bitsPerClock;
    mScheduler.scheduleAtFixedRate(this::aging, agingPeriod, agingPeriod, MILLISECONDS);
  }

  @Override
  public boolean put(PageId pageId, int size, CacheScope cacheScope) {
    return updateClockCuckoo(pageId, size, cacheScope);
  }

  /**
   * Put a page into shadow cache if it is not existed.
   *
   * @param pageId page identifier
   * @param size page size
   * @param cacheScope cache scope
   * @return true if page is put successfully; false otherwise
   */
  private boolean updateClockCuckoo(PageId pageId, int size, CacheScope cacheScope) {
    boolean ok = true;
    if (!mFilter.mightContainAndResetClock(pageId)) {
      ok = mFilter.put(pageId, size, cacheScope);
      updateWorkingSetSize();
    }
    return ok;
  }

  @Override
  public int get(PageId pageId, int bytesToRead, CacheScope cacheScope) {
    mShadowCachePageRead.getAndIncrement();
    mShadowCacheByteRead.getAndAdd(bytesToRead);
    if (mFilter.mightContainAndResetClock(pageId)) {
      mShadowCachePageHit.getAndIncrement();
      mShadowCacheByteHit.getAndAdd(bytesToRead);
      return bytesToRead;
    }
    return 0;
  }

  @Override
  public boolean delete(PageId pageId) {
    return mFilter.delete(pageId);
  }

  @Override
  public void aging() {
    mFilter.aging();
  }

  @Override
  public void updateWorkingSetSize() {
    mShadowCachePages = mFilter.approximateElementCount();
    mShadowCacheBytes = mFilter.approximateElementSize();
  }

  @Override
  public void stopUpdate() {
    mScheduler.shutdown();
  }

  @Override
  public long getShadowCachePages() {
    return mShadowCachePages;
  }

  @Override
  public long getShadowCachePages(CacheScope scope) {
    return mFilter.approximateElementCount(scope);
  }

  @Override
  public long getShadowCacheBytes() {
    return mShadowCacheBytes;
  }

  @Override
  public long getShadowCacheBytes(CacheScope scope) {
    return mFilter.approximateElementSize(scope);
  }

  @Override
  public long getShadowCachePageRead() {
    return mShadowCachePageRead.get();
  }

  @Override
  public long getShadowCachePageHit() {
    return mShadowCachePageHit.get();
  }

  @Override
  public long getShadowCacheByteRead() {
    return mShadowCacheByteRead.get();
  }

  @Override
  public long getShadowCacheByteHit() {
    return mShadowCacheByteHit.get();
  }

  @Override
  public double getFalsePositiveRatio() {
    return mFilter.expectedFpp();
  }
}
