/*
 * 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.CacheContext.StatsUnit.BYTE;

import alluxio.client.file.CacheContext;
import alluxio.client.quota.CacheScope;
import alluxio.conf.AlluxioConfiguration;
import alluxio.exception.PageNotFoundException;
import alluxio.file.ReadTargetBuffer;
import alluxio.metrics.MetricKey;
import alluxio.metrics.MetricsSystem;
import alluxio.network.protocol.databuffer.DataFileChannel;

import com.codahale.metrics.Counter;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.hash.Funnel;
import com.google.common.hash.PrimitiveSink;

import java.nio.ByteBuffer;
import java.util.Optional;
import java.util.function.Supplier;
import javax.annotation.Nonnull;

/**
 * A wrapper class of CacheManager with shadow cache.
 */
public class CacheManagerWithShadowCache implements CacheManager {
  private final CacheManager mCacheManager;
  private final ShadowCacheManager mShadowCacheManager;
  private long mShadowCachePages = 0;
  private long mShadowCacheBytes = 0;

  /**
   * @param cacheManager the real cache manager
   * @param conf the alluxio configuration
   */
  public CacheManagerWithShadowCache(CacheManager cacheManager, AlluxioConfiguration conf) {
    mCacheManager = cacheManager;
    mShadowCacheManager = ShadowCacheManager.create(conf);
  }

  @Override
  public void commitFile(String fileId) {
    mCacheManager.commitFile(fileId);
  }

  @Override
  public boolean put(PageId pageId, ByteBuffer page, CacheContext cacheContext) {
    updateShadowCache(pageId, page.remaining(), cacheContext);
    return mCacheManager.put(pageId, page, cacheContext);
  }

  @Override
  public int get(PageId pageId, int pageOffset, int bytesToRead, ReadTargetBuffer target,
      CacheContext cacheContext) {
    getOrUpdateShadowCache(pageId, bytesToRead, cacheContext);
    return mCacheManager.get(pageId, pageOffset, bytesToRead, target, cacheContext);
  }

  @Override
  public int getAndLoad(PageId pageId, int pageOffset, int bytesToRead,
      ReadTargetBuffer buffer, CacheContext cacheContext, Supplier<byte[]> externalDataSupplier) {
    getOrUpdateShadowCache(pageId, bytesToRead, cacheContext);
    return mCacheManager.getAndLoad(pageId, pageOffset, bytesToRead,
        buffer, cacheContext, externalDataSupplier);
  }

  private void getOrUpdateShadowCache(PageId pageId, int bytesToRead, CacheContext cacheContext) {
    int nread = mShadowCacheManager.get(pageId, bytesToRead, getCacheScope(cacheContext));
    if (nread > 0) {
      Metrics.SHADOW_CACHE_PAGES_HIT.inc();
      Metrics.SHADOW_CACHE_BYTES_HIT.inc(nread);
    } else {
      updateShadowCache(pageId, bytesToRead, cacheContext);
    }
    Metrics.SHADOW_CACHE_PAGES_READ.inc();
    Metrics.SHADOW_CACHE_BYTES_READ.inc(bytesToRead);
  }

  /**
   * @param pageId page identifier
   * @param pageLength length of this page
   * @param cacheContext cache related context
   */
  private void updateShadowCache(PageId pageId, int pageLength, CacheContext cacheContext) {
    if (mShadowCacheManager.put(pageId, pageLength, getCacheScope(cacheContext))) {
      updateFalsePositiveRatio();
      updateWorkingSetSize();
      if (cacheContext != null) {
        cacheContext.incrementCounter(MetricKey.CLIENT_CACHE_SHADOW_CACHE_BYTES.getName(), BYTE,
            pageLength);
      }
    }
  }

  /**
   * @param cacheContext cache related context
   * @return the cache scope
   */
  private CacheScope getCacheScope(CacheContext cacheContext) {
    return (cacheContext == null) ? CacheScope.GLOBAL : cacheContext.getCacheScope();
  }

  /**
   * Update the false positive ratio statistics.
   */
  private void updateFalsePositiveRatio() {
    int falsePositiveRatio = (int) mShadowCacheManager.getFalsePositiveRatio() * 100;
    long oldFalsePositiveRatio = Metrics.SHADOW_CACHE_FALSE_POSITIVE_RATIO.getCount();
    Metrics.SHADOW_CACHE_FALSE_POSITIVE_RATIO.inc(falsePositiveRatio - oldFalsePositiveRatio);
  }

  /**
   * Update working set size in number of pages and bytes.
   */
  @VisibleForTesting
  public void updateWorkingSetSize() {
    mShadowCacheManager.updateWorkingSetSize();
    long oldPages = Metrics.SHADOW_CACHE_PAGES.getCount();
    mShadowCachePages = mShadowCacheManager.getShadowCachePages();
    Metrics.SHADOW_CACHE_PAGES.inc(mShadowCachePages - oldPages);
    long oldBytes = Metrics.SHADOW_CACHE_BYTES.getCount();
    mShadowCacheBytes = mShadowCacheManager.getShadowCacheBytes();
    Metrics.SHADOW_CACHE_BYTES.inc(mShadowCacheBytes - oldBytes);
  }

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

  /**
   * Stop to switch bloom filters and update working set size.
   */
  @VisibleForTesting
  public void stopUpdate() {
    mShadowCacheManager.stopUpdate();
  }

  @Override
  public State state() {
    return mCacheManager.state();
  }

  @Override
  public boolean append(PageId pageId, int appendAt, byte[] page, CacheContext cacheContext) {
    return mCacheManager.append(pageId, appendAt, page, cacheContext);
  }

  @Override
  public void close() throws Exception {
    mCacheManager.close();
  }

  @Override
  public void deleteFile(String fileId) {
    mCacheManager.deleteFile(fileId);
  }

  @Override
  public void deleteTempFile(String fileId) {
    mCacheManager.deleteTempFile(fileId);
  }

  @Override
  public Optional<CacheUsage> getUsage() {
    return mCacheManager.getUsage();
  }

  @Override
  public Optional<DataFileChannel> getDataFileChannel(PageId pageId, int pageOffset,
      int bytesToRead, CacheContext cacheContext) throws PageNotFoundException {
    return mCacheManager.getDataFileChannel(pageId, pageOffset, bytesToRead, cacheContext);
  }

  /**
   * Decrease each item's clock and clean stale items.
   */
  public void aging() {
    mShadowCacheManager.aging();
  }

  /**
   * @return ShadowCachePages
   */
  public long getShadowCachePages() {
    return mShadowCachePages;
  }

  /**
   * @return ShadowCacheBytes
   */
  public long getShadowCacheBytes() {
    return mShadowCacheBytes;
  }

  /**
   * @return ShadowCacheBytes
   */
  public long getShadowCachePageRead() {
    return mShadowCacheManager.getShadowCachePageRead();
  }

  /**
   * @return ShadowCacheBytes
   */
  public long getShadowCachePageHit() {
    return mShadowCacheManager.getShadowCachePageHit();
  }

  /**
   * @return ShadowCacheBytes
   */
  public long getShadowCacheByteRead() {
    return mShadowCacheManager.getShadowCacheByteRead();
  }

  /**
   * @return ShadowCacheBytes
   */
  public long getShadowCacheByteHit() {
    return mShadowCacheManager.getShadowCacheByteHit();
  }

  /**
   * Funnel for PageId.
   */
  public enum PageIdFunnel implements Funnel<PageId> {
    FUNNEL;

    /**
     * @param from source
     * @param into destination
     */
    @Override
    public void funnel(@Nonnull PageId from, PrimitiveSink into) {
      into.putUnencodedChars(from.getFileId()).putLong(from.getPageIndex());
    }
  }

  private static final class Metrics {
    // Note that only counter can be added here.
    // Both meter and timer need to be used inline
    // because new meter and timer will be created after {@link MetricsSystem.resetAllMetrics()}
    private static final Counter SHADOW_CACHE_BYTES_READ =
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_SHADOW_CACHE_BYTES_READ.getName());
    private static final Counter SHADOW_CACHE_BYTES_HIT =
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_SHADOW_CACHE_BYTES_HIT.getName());
    private static final Counter SHADOW_CACHE_PAGES_READ =
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_SHADOW_CACHE_PAGES_READ.getName());
    private static final Counter SHADOW_CACHE_PAGES_HIT =
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_SHADOW_CACHE_PAGES_HIT.getName());
    private static final Counter SHADOW_CACHE_PAGES =
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_SHADOW_CACHE_PAGES.getName());
    private static final Counter SHADOW_CACHE_BYTES =
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_SHADOW_CACHE_BYTES.getName());
    private static final Counter SHADOW_CACHE_FALSE_POSITIVE_RATIO =
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_SHADOW_CACHE_FALSE_POSITIVE_RATIO.getName());
  }
}
