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.client.file.cache.cuckoofilter;
  
  import alluxio.Constants;
  import alluxio.client.quota.CacheScope;
  import alluxio.collections.BitSet;
  import alluxio.collections.BuiltinBitSet;
  
  import com.google.common.base.Preconditions;
  import com.google.common.hash.Funnel;
  import com.google.common.hash.HashFunction;
  import com.google.common.hash.Hashing;
  
  import java.io.Serializable;
  import java.util.LinkedList;
  import java.util.Queue;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.concurrent.atomic.AtomicLong;
  
  /**
   * A concurrent cuckoo filter with three customized field: clock, size and scope.
   *
   * @param <T> the type of item
   */
  public class ConcurrentClockCuckooFilter<T> implements ClockCuckooFilter<T>, Serializable {
    public static final double DEFAULT_FPP = 0.01;
    // The default load factor is from "Cuckoo Filter: Practically Better Than Bloom" by Fan et al.
    public static final double DEFAULT_LOAD_FACTOR = 0.955;
    public static final int TAGS_PER_BUCKET = 4;
  
    private static final long serialVersionUID = 1L;
  
    // zero tag means non-existed
    private static final int NON_EXISTENT_TAG = 0;
  
    private static final int DEFAULT_NUM_LOCKS = 4096;
  
    // the maximum number of entries in a cuckoo path from "Algorithmic Improvements for Fast
    // Concurrent Cuckoo Hashing" by Li et al.
    private static final int MAX_BFS_PATH_LEN = 5;
  
    // aging configurations
    // we do not want to block user operations for too long,
    // so hard limited the aging number of each operation.
    private static final int MAX_AGING_PER_OPERATION = 500;
    private static final int AGING_STEP_SIZE = 5;
  
    private final AtomicLong mNumItems = new AtomicLong(0);
    private final AtomicLong mTotalBytes = new AtomicLong(0);
    private final AtomicLong mOperationCount = new AtomicLong(0);
    private final AtomicLong mAgingCount = new AtomicLong(0);
  
    private final AtomicInteger[] mScopeToNumber;
    private final AtomicLong[] mScopeToSize;
    // cuckoo filter's bucket number should be a power of 2
    private final int mNumBuckets;
    private final int mBitsPerTag;
    private final int mBitsPerClock;
    private final int mBitsPerSize;
    private final int mBitsPerScope;
  
    private final int mMaxSize;
    private final int mMaxAge;
    private final Funnel<? super T> mFunnel;
    private final HashFunction mHashFunction;
    private final ScopeEncoder mScopeEncoder;
    private final SegmentedLock mLocks;
    private final int[] mSegmentedAgingPointers;
    // if count-based sliding window, windowSize is the number of operations;
    // if time-based sliding window, windowSize is the milliseconds in a period.
    private final SlidingWindowType mSlidingWindowType;
    private final long mWindowSize;
    private final long mStartTime = System.currentTimeMillis();
    private final CuckooTable mTable;
    private final CuckooTable mClockTable;
    private final CuckooTable mSizeTable;
    private final CuckooTable mScopeTable;
    // size encoder  config
    private boolean mOpenSizeEncoder = false;
    private int mPrefixBits;
    private int mSuffixBits;
    private SizeEncoder mSizeEncoder = null;
  
    /**
     * The constructor of concurrent clock cuckoo filter.
     *
     * @param table the table to store tag (fingerprint)
     * @param clockTable the table to store clock
     * @param sizeTable the table to store size
     * @param scopeTable the table to store scope
     * @param slidingWindowType the type of sliding window
     * @param windowSize the size of sliding window
     * @param funnel the funnel of T's that the constructed cuckoo filter will use
     * @param hasher the hash function the constructed cuckoo filter will use
     */
    private ConcurrentClockCuckooFilter(CuckooTable table, CuckooTable clockTable,
        CuckooTable sizeTable, CuckooTable scopeTable, SlidingWindowType slidingWindowType,
        long windowSize, Funnel<? super T> funnel, HashFunction hasher) {
      mTable = table;
      mNumBuckets = table.getNumBuckets();
      mBitsPerTag = table.getBitsPerTag();
      mClockTable = clockTable;
      mBitsPerClock = clockTable.getBitsPerTag();
      mSizeTable = sizeTable;
      mBitsPerSize = sizeTable.getBitsPerTag();
      mScopeTable = scopeTable;
      mBitsPerScope = scopeTable.getBitsPerTag();
      Preconditions.checkArgument(mBitsPerSize > 0 && mBitsPerSize < Integer.SIZE - 1,
              "check the value of bitsPerSize");
      Preconditions.checkArgument(mBitsPerClock > 0 && mBitsPerClock < Integer.SIZE - 1,
              "check the value of bitsPerClock");
      Preconditions.checkArgument(mBitsPerScope > 0 && mBitsPerScope < Integer.SIZE - 1,
              "check the value of bitsPerScope");
      Preconditions.checkArgument(mBitsPerTag > 0 && mBitsPerTag < Integer.SIZE - 1,
              "check the value of bitsPerTag");
      mMaxSize = (1 << mBitsPerSize);
      mMaxAge = (1 << mBitsPerClock) - 1;
      mSlidingWindowType = slidingWindowType;
      mWindowSize = windowSize;
      mFunnel = funnel;
      mHashFunction = hasher;
      mLocks = new SegmentedLock(Math.min(DEFAULT_NUM_LOCKS, mNumBuckets >> 1), mNumBuckets);
      // init scope statistics
      // note that the GLOBAL scope is the default scope and is always encoded to zero
      mScopeEncoder = new ScopeEncoder(mBitsPerScope);
      mScopeEncoder.encode(CacheScope.GLOBAL);
      int maxNumScopes = (1 << mBitsPerScope);
      mScopeToNumber = new AtomicInteger[maxNumScopes];
      mScopeToSize = new AtomicLong[maxNumScopes];
      for (int i = 0; i < maxNumScopes; i++) {
        mScopeToNumber[i] = new AtomicInteger(0);
        mScopeToSize[i] = new AtomicLong(0);
      }
      // init aging pointers for each lock
      mSegmentedAgingPointers = new int[mLocks.getNumLocks()];
    }
  
    /**
     * The constructor of concurrent clock cuckoo filter with size encoder.
     *
     * @param table the table to store tag (fingerprint)
     * @param clockTable the table to store clock
     * @param sizeTable the table to store size
     * @param scopeTable the table to store scope
     * @param sizeEncoder the size encoder to encode size
     * @param slidingWindowType the type of sliding window
     * @param windowSize the size of sliding window
     * @param funnel the funnel of T's that the constructed cuckoo filter will use
     * @param hasher the hash function the constructed cuckoo filter will use
     */
    private ConcurrentClockCuckooFilter(CuckooTable table, CuckooTable clockTable,
        CuckooTable sizeTable, CuckooTable scopeTable, SizeEncoder sizeEncoder,
        SlidingWindowType slidingWindowType, long windowSize, Funnel<? super T> funnel,
        HashFunction hasher) {
      mTable = table;
      mNumBuckets = table.getNumBuckets();
      mBitsPerTag = table.getBitsPerTag();
      mClockTable = clockTable;
      mBitsPerClock = clockTable.getBitsPerTag();
      mSizeTable = sizeTable;
      mBitsPerSize = sizeTable.getBitsPerTag();
      mScopeTable = scopeTable;
      mBitsPerScope = scopeTable.getBitsPerTag();
      mOpenSizeEncoder = true;
      mSizeEncoder = sizeEncoder;
      Preconditions.checkArgument(mBitsPerSize > 0 && mBitsPerSize < Integer.SIZE - 1,
          "check the value of bitsPerSize");
      Preconditions.checkArgument(mBitsPerClock > 0 && mBitsPerClock < Integer.SIZE - 1,
          "check the value of bitsPerClock");
      Preconditions.checkArgument(mBitsPerScope > 0 && mBitsPerScope < Integer.SIZE - 1,
          "check the value of bitsPerScope");
      Preconditions.checkArgument(mBitsPerTag > 0 && mBitsPerTag < Integer.SIZE - 1,
          "check the value of bitsPerTag");
      mMaxSize = (1 << mBitsPerSize);
      mMaxAge = (1 << mBitsPerClock) - 1;
      mSlidingWindowType = slidingWindowType;
      mWindowSize = windowSize;
      mFunnel = funnel;
      mHashFunction = hasher;
      mLocks = new SegmentedLock(Math.min(DEFAULT_NUM_LOCKS, mNumBuckets >> 1), mNumBuckets);
      // init scope statistics
      // note that the GLOBAL scope is the default scope and is always encoded to zero
      mScopeEncoder = new ScopeEncoder(mBitsPerScope);
      mScopeEncoder.encode(CacheScope.GLOBAL);
      int maxNumScopes = (1 << mBitsPerScope);
      mScopeToNumber = new AtomicInteger[maxNumScopes];
      mScopeToSize = new AtomicLong[maxNumScopes];
      for (int i = 0; i < maxNumScopes; i++) {
        mScopeToNumber[i] = new AtomicInteger(0);
        mScopeToSize[i] = new AtomicLong(0);
      }
      // init aging pointers for each lock
      mSegmentedAgingPointers = new int[mLocks.getNumLocks()];
    }
  
    /**
     * Create a concurrent cuckoo filter with specified parameters.
     *
     * @param funnel the funnel of T's that the constructed cuckoo filter will use
     * @param expectedInsertions the number of expected insertions to the constructed {@code
     *      ConcurrentClockCuckooFilter}; must be positive; must be a power of 2
     * @param bitsPerClock the number of bits the clock field has
     * @param bitsPerSize the number of bits the size field has
     * @param bitsPerScope the number of bits the scope field has
     * @param slidingWindowType the type of sliding window
     * @param windowSize the size of the sliding window
     * @param fpp the desired false positive probability (must be positive and less than 1.0)
     * @param loadFactor the load factor of cuckoo filter (must be positive and less than 1.0)
     * @param hasher the hash function to be used
     * @param <T> the type of item
     * @return a {@code ConcurrentClockCuckooFilter}
     */
    public static <T> ConcurrentClockCuckooFilter<T> create(Funnel<? super T> funnel,
        long expectedInsertions, int bitsPerClock, int bitsPerSize, int bitsPerScope,
        SlidingWindowType slidingWindowType, long windowSize, double fpp, double loadFactor,
        HashFunction hasher) {
      int bitsPerTag = CuckooUtils.optimalBitsPerTag(fpp, loadFactor);
      long numBuckets = CuckooUtils.optimalBuckets(expectedInsertions, loadFactor, TAGS_PER_BUCKET);
      long numBits = numBuckets * TAGS_PER_BUCKET * bitsPerTag;
      BitSet bits = new BuiltinBitSet((int) numBits);
      CuckooTable table = new SimpleCuckooTable(bits, (int) numBuckets, TAGS_PER_BUCKET, bitsPerTag);
  
      BitSet clockBits = new BuiltinBitSet((int) (numBuckets * TAGS_PER_BUCKET * bitsPerClock));
      CuckooTable clockTable =
          new SimpleCuckooTable(clockBits, (int) numBuckets, TAGS_PER_BUCKET, bitsPerClock);
  
      BitSet sizeBits = new BuiltinBitSet((int) (numBuckets * TAGS_PER_BUCKET * bitsPerSize));
      CuckooTable sizeTable =
          new SimpleCuckooTable(sizeBits, (int) numBuckets, TAGS_PER_BUCKET, bitsPerSize);
  
      BitSet scopeBits = new BuiltinBitSet((int) (numBuckets * TAGS_PER_BUCKET * bitsPerScope));
      CuckooTable scopeTable =
          new SimpleCuckooTable(scopeBits, (int) numBuckets, TAGS_PER_BUCKET, bitsPerScope);
      return new ConcurrentClockCuckooFilter<>(table, clockTable, sizeTable, scopeTable,
          slidingWindowType, windowSize, funnel, hasher);
    }
  
    /**
     * Create a concurrent cuckoo filter with size encoder.
     *
     * @param funnel the funnel of T's that the constructed cuckoo filter will use
     * @param expectedInsertions the number of expected insertions to the constructed {@code
     *      ConcurrentClockCuckooFilter}; must be positive; must be a power of 2
     * @param bitsPerClock the number of bits the clock field has
     * @param bitsPerScope the number of bits the scope field has
     * @param prefixBits the prefix bits number of the size field
     * @param suffixBits the suffix bits number of the size field
     * @param slidingWindowType the type of sliding window
     * @param windowSize the size of the sliding window
     * @param fpp the desired false positive probability (must be positive and less than 1.0)
     * @param loadFactor the load factor of cuckoo filter (must be positive and less than 1.0)
     * @param hasher the hash function to be used
     * @param <T> the type of item
     * @return a {@code ConcurrentClockCuckooFilter}
     */
    public static <T> ConcurrentClockCuckooFilter<T> create(Funnel<? super T> funnel,
          long expectedInsertions, int bitsPerClock, int bitsPerScope,
          int prefixBits, int suffixBits, SlidingWindowType slidingWindowType,
          long windowSize, double fpp, double loadFactor, HashFunction hasher) {
      int bitsPerTag = CuckooUtils.optimalBitsPerTag(fpp, loadFactor);
      long numBuckets = CuckooUtils.optimalBuckets(expectedInsertions, loadFactor, TAGS_PER_BUCKET);
      long numBits = numBuckets * TAGS_PER_BUCKET * bitsPerTag;
      BitSet bits = new BuiltinBitSet((int) numBits);
      CuckooTable table = new SimpleCuckooTable(bits, (int) numBuckets, TAGS_PER_BUCKET, bitsPerTag);
  
      BitSet clockBits = new BuiltinBitSet((int) (numBuckets * TAGS_PER_BUCKET * bitsPerClock));
      CuckooTable clockTable =
          new SimpleCuckooTable(clockBits, (int) numBuckets, TAGS_PER_BUCKET, bitsPerClock);
  
      BitSet sizeBits = new BuiltinBitSet((int) (numBuckets * TAGS_PER_BUCKET * prefixBits));
      CuckooTable sizeTable =
          new SimpleCuckooTable(sizeBits, (int) numBuckets, TAGS_PER_BUCKET, prefixBits);
  
      BitSet scopeBits = new BuiltinBitSet((int) (numBuckets * TAGS_PER_BUCKET * bitsPerScope));
      CuckooTable scopeTable =
          new SimpleCuckooTable(scopeBits, (int) numBuckets, TAGS_PER_BUCKET, bitsPerScope);
      SizeEncoder sizeEncoder = new SizeEncoder(prefixBits + suffixBits, suffixBits);
      return new ConcurrentClockCuckooFilter<>(table, clockTable, sizeTable, scopeTable, sizeEncoder,
          slidingWindowType, windowSize, funnel, hasher);
    }
  
    /**
     * Create a concurrent cuckoo filter with specified parameters.
     *
     * @param funnel the funnel of T's that the constructed {@code BloomFilter} will use
     * @param expectedInsertions the number of expected insertions to the constructed {@code
     *      ConcurrentClockCuckooFilter}; must be positive; must be a power of 2
     * @param bitsPerClock the number of bits the clock field has
     * @param bitsPerSize the number of bits the size field has
     * @param bitsPerScope the number of bits the scope field has
     * @param slidingWindowType the type of sliding window
     * @param windowSize the size of the sliding window
     * @param fpp the desired false positive probability (must be positive and less than 1.0)
     * @param loadFactor the load factor of cuckoo filter (must be positive and less than 1.0)
     * @param <T> the type of item
     * @return a {@code ConcurrentClockCuckooFilter}
     */
    public static <T> ConcurrentClockCuckooFilter<T> create(Funnel<? super T> funnel,
        long expectedInsertions, int bitsPerClock, int bitsPerSize, int bitsPerScope,
        SlidingWindowType slidingWindowType, long windowSize, double fpp, double loadFactor) {
      return create(funnel, expectedInsertions, bitsPerClock, bitsPerSize, bitsPerScope,
          slidingWindowType, windowSize, fpp, loadFactor, Hashing.murmur3_128());
    }
  
    /**
     * Create a concurrent cuckoo filter with specified parameters.
     *
     * @param funnel the funnel of T's that the constructed {@code BloomFilter} will use
     * @param expectedInsertions the number of expected insertions to the constructed {@code
     *      ConcurrentClockCuckooFilter}; must be positive; must be a power of 2
     * @param bitsPerClock the number of bits the clock field has
     * @param bitsPerSize the number of bits the size field has
     * @param bitsPerScope the number of bits the scope field has
     * @param slidingWindowType the type of sliding window
     * @param windowSize the size of the sliding window
     * @param fpp the desired false positive probability (must be positive and less than 1.0)
     * @param <T> the type of item
     * @return a {@code ConcurrentClockCuckooFilter}
     */
    public static <T> ConcurrentClockCuckooFilter<T> create(Funnel<? super T> funnel,
        long expectedInsertions, int bitsPerClock, int bitsPerSize, int bitsPerScope,
        SlidingWindowType slidingWindowType, long windowSize, double fpp) {
      return create(funnel, expectedInsertions, bitsPerClock, bitsPerSize, bitsPerScope,
          slidingWindowType, windowSize, fpp, DEFAULT_LOAD_FACTOR);
    }
  
    /**
     * Create a concurrent cuckoo filter with specified parameters.
     *
     * @param funnel the funnel of T's that the constructed {@code BloomFilter} will use
     * @param expectedInsertions the number of expected insertions to the constructed {@code
     *      ConcurrentClockCuckooFilter}; must be positive; must be a power of 2
     * @param bitsPerClock the number of bits the clock field has
     * @param bitsPerSize the number of bits the size field has
     * @param bitsPerScope the number of bits the scope field has
     * @param slidingWindowType the type of sliding window
     * @param windowSize the size of the sliding window
     * @param <T> the type of item
     * @return a {@code ConcurrentClockCuckooFilter}
     */
    public static <T> ConcurrentClockCuckooFilter<T> create(Funnel<? super T> funnel,
        long expectedInsertions, int bitsPerClock, int bitsPerSize, int bitsPerScope,
        SlidingWindowType slidingWindowType, long windowSize) {
      return create(funnel, expectedInsertions, bitsPerClock, bitsPerSize, bitsPerScope,
          slidingWindowType, windowSize, DEFAULT_FPP);
    }
  
    /**
     * Create a concurrent cuckoo filter with specified parameters.
     *
     * @param funnel the funnel of T's that the constructed {@code BloomFilter} will use
     * @param expectedInsertions the number of expected insertions to the constructed {@code
     *      ConcurrentClockCuckooFilter}; must be positive; must be a power of 2
     * @param bitsPerClock the number of bits the clock field has
     * @param bitsPerSize the number of bits the size field has
     * @param bitsPerScope the number of bits the scope field has
     * @param <T> the type of item
     * @return a {@code ConcurrentClockCuckooFilter}
     */
    public static <T> ConcurrentClockCuckooFilter<T> create(Funnel<? super T> funnel,
        long expectedInsertions, int bitsPerClock, int bitsPerSize, int bitsPerScope) {
      Preconditions.checkNotNull(funnel);
      Preconditions.checkArgument(expectedInsertions > 0);
      Preconditions.checkArgument(bitsPerClock > 0);
      Preconditions.checkArgument(bitsPerSize > 0);
      Preconditions.checkArgument(bitsPerScope > 0);
      return create(funnel, expectedInsertions, bitsPerClock, bitsPerSize, bitsPerScope,
          SlidingWindowType.NONE, -1, DEFAULT_FPP);
    }
  
    @Override
    public boolean put(T item, int size, CacheScope scopeInfo) {
      // NOTE: zero size is not allowed in our clock filter, because we use zero size as
      // a special case to indicate an size overflow (size > mMaxSize), and all the
      // overflowed size will be revised to mMaxSize in method encodeSize()
      if (size <= 0) {
        return false;
      }
      long hv = hashValue(item);
      int tag = tagHash(hv);
      int b1 = indexHash(hv);
      int b2 = altIndex(b1, tag);
      size = encodeSize(size);
      // Generally, we will hold write locks in two places:
      // 1) put/delete;
      // 2) cuckooPathSearch & cuckooPathMove.
      // But We only execute opportunistic aging in put/delete.
      // This is because we expect cuckoo path search & move to be as fast as possible,
      // or it may be more possible to fail.
      writeLockAndOpportunisticAging(b1, b2);
      TagPosition pos = cuckooInsertLoop(b1, b2, tag);
      if (pos.getStatus() == CuckooStatus.OK) {
        // update statistics
        int scope = encodeScope(scopeInfo);
        mNumItems.incrementAndGet();
        mTotalBytes.addAndGet(size);
        updateScopeStatistics(scope, 1, size);
        // b1 and b2 should be insertable for item, which means:
        // 1. b1 or b2 have at least one empty slot (this is guaranteed until we unlock two buckets);
        // 2. b1 and b2 do not contain duplicated tag.
        mTable.writeTag(pos.getBucketIndex(), pos.getSlotIndex(), tag);
        mClockTable.writeTag(pos.getBucketIndex(), pos.getSlotIndex(), mMaxAge);
        mScopeTable.writeTag(pos.getBucketIndex(), pos.getSlotIndex(), scope);
        // encode the size to group index
        if (mOpenSizeEncoder) {
          mSizeEncoder.add(size);
          size = mSizeEncoder.encode(size);
        }
        mSizeTable.writeTag(pos.getBucketIndex(), pos.getSlotIndex(), size);
  
        mLocks.unlockWrite(b1, b2);
        return true;
      }
      mLocks.unlockWrite(b1, b2);
      return false;
    }
  
    @Override
    public boolean mightContainAndResetClock(T item) {
      return mightContainAndOptionalResetClock(item, true);
    }
  
    @Override
    public boolean mightContain(T item) {
      return mightContainAndOptionalResetClock(item, false);
    }
  
    /**
     * @param item the item to be checked
     * @param shouldReset the flag to indicate whether to reset clock field
     * @return true if item is in cuckoo filter; false otherwise
     */
    private boolean mightContainAndOptionalResetClock(T item, boolean shouldReset) {
      long hv = hashValue(item);
      int tag = tagHash(hv);
      int b1 = indexHash(hv);
      int b2 = altIndex(b1, tag);
      mLocks.readLock(b1, b2);
      TagPosition pos = mTable.findTag(b1, b2, tag);
      boolean found = pos.getStatus() == CuckooStatus.OK;
      if (found && shouldReset) {
        // set C to MAX
        mClockTable.writeTag(pos.getBucketIndex(), pos.getSlotIndex(), mMaxAge);
      }
      mLocks.unlockRead(b1, b2);
      return found;
    }
  
    @Override
    public boolean delete(T item) {
      long hv = hashValue(item);
      int tag = tagHash(hv);
      int b1 = indexHash(hv);
      int b2 = altIndex(b1, tag);
      writeLockAndOpportunisticAging(b1, b2);
      TagPosition pos = mTable.deleteTag(b1, tag);
      if (pos.getStatus() != CuckooStatus.OK) {
        pos = mTable.deleteTag(b2, tag);
      }
      if (pos.getStatus() == CuckooStatus.OK) {
        mNumItems.decrementAndGet();
        int scope = mScopeTable.readTag(pos.getBucketIndex(), pos.getSlotIndex());
        int size = mSizeTable.readTag(pos.getBucketIndex(), pos.getSlotIndex());
        size = decodeSize(size);
        if (mOpenSizeEncoder) {
          size = mSizeEncoder.dec(size);
        }
        updateScopeStatistics(scope, -1, -size);
        // Clear Clock
        mClockTable.writeTag(pos.getBucketIndex(), pos.getSlotIndex(), NON_EXISTENT_TAG);
        mLocks.unlockWrite(b1, b2);
        return true;
      }
      mLocks.unlockWrite(b1, b2);
      return false;
    }
  
    @Override
    public void aging() {
      int numSegments = mLocks.getNumLocks();
      int bucketsPerSegment = mLocks.getNumBucketsPerSegment();
      for (int i = 0; i < numSegments; i++) {
        // TODO(iluoeli): avoid acquire locks here since it may be blocked
        //  for a long time if this segment is contended by multiple users.
        mLocks.writeLockSegment(i);
        if (mSegmentedAgingPointers[i] < bucketsPerSegment) {
          agingSegment(i, bucketsPerSegment);
        }
        mSegmentedAgingPointers[i] = 0;
        mLocks.unlockWriteSegment(i);
      }
    }
  
    /**
     * Get the item's clock value (age).
     *
     * @param item the item to be queried
     * @return the clock value
     */
    public int getAge(T item) {
      long hv = hashValue(item);
      int tag = tagHash(hv);
      int b1 = indexHash(hv);
      int b2 = altIndex(b1, tag);
      mLocks.readLock(b1, b2);
      TagPosition pos = mTable.findTag(b1, b2, tag);
      if (pos.getStatus() == CuckooStatus.OK) {
        int clock = mClockTable.readTag(pos.getBucketIndex(), pos.getSlotIndex());
        mLocks.unlockRead(b1, b2);
        return clock;
      }
      mLocks.unlockRead(b1, b2);
      return 0;
    }
  
    /**
     * @return the summary of this cuckoo filter
     */
    public String getSummary() {
      return "numBuckets: " + getNumBuckets() + "\ntagsPerBucket: " + getTagsPerBucket()
          + "\nbitsPerTag: " + getBitsPerTag() + "\nbitsPerClock: " + getBitsPerClock()
          + "\nbitsPerSize: " + mBitsPerSize + "\nbitsPerScope: " + mBitsPerScope + "\nSizeInMB: "
          + (getNumBuckets() * getTagsPerBucket() * getBitsPerTag() / 8.0 / Constants.MB
              + getNumBuckets() * getTagsPerBucket() * getBitsPerClock() / 8.0 / Constants.MB
              + getNumBuckets() * getTagsPerBucket() * mBitsPerSize / 8.0 / Constants.MB
              + getNumBuckets() * getTagsPerBucket() * mBitsPerScope / 8.0 / Constants.MB);
    }
  
    @Override
    public double expectedFpp() {
      // equation from "Cuckoo Filter: Simplification and Analysis" by David Eppstein (Theorem 5.1)
      return 2 * mNumItems.doubleValue()
          / (mNumBuckets * TAGS_PER_BUCKET * ((1L << mBitsPerTag) - 1));
    }
  
    @Override
    public long approximateElementCount() {
      return mNumItems.intValue();
    }
  
    @Override
    public long approximateElementCount(CacheScope scopeInfo) {
      return mScopeToNumber[encodeScope(scopeInfo)].get();
    }
  
    @Override
    public long approximateElementSize() {
      return mTotalBytes.get();
    }
  
    @Override
    public long approximateElementSize(CacheScope scopeInfo) {
      return mScopeToSize[encodeScope(scopeInfo)].get();
    }
  
    /**
     * By calling this method, cuckoo filter is informed of the number of entries have passed.
     *
     * @param count the number of operations have passed
     */
    public void increaseOperationCount(int count) {
      mOperationCount.addAndGet(count);
    }
  
    /**
     * @return the number of buckets this cuckoo filter has
     */
    public int getNumBuckets() {
      return mTable.getNumBuckets();
    }
  
    /**
     * @return the number of slots per bucket has
     */
    public int getTagsPerBucket() {
      return mTable.getNumTagsPerBuckets();
    }
  
    /**
     * @return the number of bits per slot has
     */
    public int getBitsPerTag() {
      return mTable.getBitsPerTag();
    }
  
    /**
     * @return the number of bits per slot's clock field has
     */
    public int getBitsPerClock() {
      return mClockTable.getBitsPerTag();
    }
  
    /**
     * @return the number of bits per slot's size field has
     */
    public int getBitsPerSize() {
      return mSizeTable.getBitsPerTag();
    }
  
    /**
     * @return the number of bits per slot's scope field has
     */
    public int getBitsPerScope() {
      return mScopeTable.getBitsPerTag();
    }
  
    /**
     * @param item the object to be hashed
     * @return the hash code of this item
     */
    private long hashValue(T item) {
      return mHashFunction.newHasher().putObject(item, mFunnel).hash().asLong();
    }
  
    /**
     * Compute the index from a hash value.
     *
     * @param hv the hash value used for computing
     * @return the bucket computed on given hash value
     */
    private int indexHash(long hv) {
      return CuckooUtils.indexHash((int) (hv >> 32), mNumBuckets);
    }
  
    /**
     * Compute the tag from a hash value.
     *
     * @param hv the hash value used for computing
     * @return the tag (fingerprint) computed on given hash value
     */
    private int tagHash(long hv) {
      return CuckooUtils.tagHash((int) hv, mBitsPerTag);
    }
  
    /**
     * Compute the alternative index from index and tag.
     *
     * @param index the bucket for computing
     * @param tag the tag (fingerprint) for computing
     * @return the alternative bucket
     */
    private int altIndex(int index, int tag) {
      return CuckooUtils.altIndex(index, tag, mNumBuckets);
    }
  
    /**
     * Encode a scope information into a integer type (storage type). We only store table-level or
     * higher level scope information, so file-level (partition-level) scope will not be stored.
     *
     * @param scopeInfo the scope to be encoded
     * @return the encoded number of scope
     */
    private int encodeScope(CacheScope scopeInfo) {
      if (scopeInfo.level() == CacheScope.Level.PARTITION) {
        return mScopeEncoder.encode(scopeInfo.parent());
      }
      return mScopeEncoder.encode(scopeInfo);
    }
  
    /**
     * Encode the original size to internal storage types, used for overflow handling etc.
     *
     * @param size the size to be encoded
     * @return the storage type of the encoded size
     */
    private int encodeSize(int size) {
      return Math.min(mMaxSize, size);
    }
  
    /**
     * Decode the storage type of size to original integer.
     *
     * @param size the storage type of size to be decoded
     * @return the decoded size of the storage type
     */
    private int decodeSize(int size) {
      if (size == 0) {
        size = mMaxSize;
      }
      return size;
    }
  
    /**
     * A thread-safe method to update scope statistics.
     *
     * @param scope the scope be be updated
     * @param number the number of items this scope have changed
     * @param size the size of this scope have changed
     */
    private void updateScopeStatistics(int scope, int number, int size) {
      mScopeToNumber[scope].addAndGet(number);
      mScopeToSize[scope].addAndGet(size);
    }
  
    /**
     * Try find an empty slot for the item. Assume already held the lock of buckets i1 and i2.
     *
     * @param b1 the first bucket
     * @param b2 the second bucket
     * @param tag the tag (fingerprint) of this tem
     * @return a valid tag position pointing to the empty slot; otherwise an invalid tat position
     *         indicating failure
     */
    private TagPosition cuckooInsertLoop(int b1, int b2, int tag) {
      TagPosition pos = cuckooInsert(b1, b2, tag);
      if (pos.getStatus() == CuckooStatus.OK) {
        return pos;
      }
      return new TagPosition(-1, -1, CuckooStatus.FAILURE);
    }
  
    /**
     * Try find an empty slot for the item. Assume already held the lock of buckets i1 and i2.
     *
     * @param b1 the first bucket
     * @param b2 the second bucket
     * @param tag the tag (fingerprint)
     * @return a valid tag position pointing to the empty slot; otherwise an invalid tat position
     *         indicating failure
     */
    private TagPosition cuckooInsert(int b1, int b2, int tag) {
      // try find b1 and b2 firstly
      TagPosition pos1 = tryFindInsertBucket(b1, tag);
      if (pos1.getStatus() == CuckooStatus.FAILURE_KEY_DUPLICATED) {
        return pos1;
      }
      TagPosition pos2 = tryFindInsertBucket(b2, tag);
      if (pos2.getStatus() == CuckooStatus.FAILURE_KEY_DUPLICATED) {
        return pos2;
      }
      if (pos1.getStatus() == CuckooStatus.OK) {
        return pos1;
      }
      if (pos2.getStatus() == CuckooStatus.OK) {
        return pos2;
      }
      // then BFS search from b1 and b2
      TagPosition pos = runCuckoo(b1, b2);
      if (pos.getStatus() == CuckooStatus.OK) {
        // avoid another duplicated key is inserted during runCuckoo.
        if (mTable.findTag(b1, b2, tag).getStatus() == CuckooStatus.OK) {
          pos.setStatus(CuckooStatus.FAILURE_KEY_DUPLICATED);
        }
      }
      return pos;
    }
  
    /**
     * Try find an empty slot for the item. Assume already held the lock of buckets i1 and i2.
     *
     * @param b1 the first bucket
     * @param b2 the second bucket
     * @return a valid tag position pointing to the empty slot; otherwise an invalid tat position
     *         indicating failure
     */
    private TagPosition runCuckoo(int b1, int b2) {
      mLocks.unlockWrite(b1, b2);
      TagPosition pos = new TagPosition(-1, -1, CuckooStatus.FAILURE);
      int maxPathLen = MAX_BFS_PATH_LEN;
      CuckooRecord[] cuckooPath = new CuckooRecord[maxPathLen];
      for (int i = 0; i < maxPathLen; i++) {
        cuckooPath[i] = new CuckooRecord();
      }
      boolean done = false;
      while (!done) {
        int depth = cuckooPathSearch(b1, b2, cuckooPath);
        if (depth < 0) {
          break;
        }
        if (cuckooPathMove(b1, b2, cuckooPath, depth)) {
          pos.setBucketAndSlot(cuckooPath[0].mBucketIndex, cuckooPath[0].mSlotIndex);
          pos.setStatus(CuckooStatus.OK);
          done = true;
        }
      }
      if (!done) {
        // NOTE: since we assume holding the locks of two buckets before calling this method,
        // we keep this assumptions after return.
        mLocks.writeLock(b1, b2);
      }
      return pos;
    }
  
    /**
     * Search for an empty slot from two initial buckets b1 and b2, and move items backwards along the
     * path.
     *
     * @param b1 the first bucket
     * @param b2 the second bucket
     * @param cuckooPath the information of searched path
     * @return the depth the path
     */
    private int cuckooPathSearch(int b1, int b2, CuckooRecord[] cuckooPath) {
      // 1. search a path
      BFSEntry x = slotBFSSearch(b1, b2);
      if (x.mDepth == -1) {
        return -1;
      }
      // 2. re-construct path from x
      for (int i = x.mDepth; i >= 0; i--) {
        cuckooPath[i].mSlotIndex = x.mPathcode % TAGS_PER_BUCKET;
        x.mPathcode /= TAGS_PER_BUCKET;
      }
      if (x.mPathcode == 0) {
        cuckooPath[0].mBucketIndex = b1;
      } else {
        cuckooPath[0].mBucketIndex = b2;
      }
      // 3. restore the tag (fingerprint) of the starting point
      mLocks.writeLock(cuckooPath[0].mBucketIndex);
      cuckooPath[0].mTag = mTable.readTag(cuckooPath[0].mBucketIndex, cuckooPath[0].mSlotIndex);
      if (cuckooPath[0].mTag == NON_EXISTENT_TAG) {
        mLocks.unlockWrite(cuckooPath[0].mBucketIndex);
        return 0;
      }
      mLocks.unlockWrite(cuckooPath[0].mBucketIndex);
      // 4. restore the tag (fingerprint) of other points along the cuckoo path
      for (int i = 1; i <= x.mDepth; i++) {
        CuckooRecord curr = cuckooPath[i];
        CuckooRecord prev = cuckooPath[i - 1];
        curr.mBucketIndex = altIndex(prev.mBucketIndex, prev.mTag);
        mLocks.writeLock(curr.mBucketIndex);
        curr.mTag = mTable.readTag(curr.mBucketIndex, curr.mSlotIndex);
        if (curr.mTag == NON_EXISTENT_TAG) {
          mLocks.unlockWrite(curr.mBucketIndex);
          return i;
        }
        mLocks.unlockWrite(curr.mBucketIndex);
      }
      return x.mDepth;
    }
  
    /**
     * Search an empty slot from two initial buckets.
     *
     * @param b1 the first bucket
     * @param b2 the second bucket
     * @return the last entry of searched path
     */
    private BFSEntry slotBFSSearch(int b1, int b2) {
      Queue<BFSEntry> queue = new LinkedList<>();
      queue.offer(new BFSEntry(b1, 0, 0));
      queue.offer(new BFSEntry(b2, 1, 0));
      while (!queue.isEmpty()) {
        BFSEntry x = queue.poll();
        mLocks.writeLock(x.mBucketIndex);
        // pick a random slot to start on
        int startingSlot = x.mPathcode % TAGS_PER_BUCKET;
        for (int i = 0; i < TAGS_PER_BUCKET; i++) {
          int slot = (startingSlot + i) % TAGS_PER_BUCKET;
          int tag = mTable.readTag(x.mBucketIndex, slot);
          if (tag == NON_EXISTENT_TAG) {
            x.mPathcode = x.mPathcode * TAGS_PER_BUCKET + slot;
            mLocks.unlockWrite(x.mBucketIndex);
            return x;
          }
          if (x.mDepth < MAX_BFS_PATH_LEN - 1) {
            queue.offer(new BFSEntry(altIndex(x.mBucketIndex, tag),
                x.mPathcode * TAGS_PER_BUCKET + slot, x.mDepth + 1));
          }
        }
        mLocks.unlockWrite(x.mBucketIndex);
      }
      return new BFSEntry(0, 0, -1);
    }
  
    /**
     * Move items backward along the cuckoo path.
     *
     * @param b1 the first bucket
     * @param b2 the second bucket
     * @param cuckooPath the path to move along
     * @param depth the depth of the path
     * @return true if successfully moved items along the path; false otherwise
     */
    private boolean cuckooPathMove(int b1, int b2, CuckooRecord[] cuckooPath, int depth) {
      if (depth == 0) {
        mLocks.writeLock(b1, b2);
        if (mTable.readTag(cuckooPath[0].mBucketIndex, cuckooPath[0].mSlotIndex) == 0) {
          mLocks.unlockWrite(b1, b2);
          return true;
        } else {
          mLocks.unlockWrite(b1, b2);
          return false;
        }
      }
  
      while (depth > 0) {
        CuckooRecord from = cuckooPath[depth - 1];
        CuckooRecord to = cuckooPath[depth];
        if (depth == 1) {
          // NOTE: We must hold the locks of b1 and b2.
          // Or their slots may be preempted by another key if we released locks.
          mLocks.writeLock(b1, b2, to.mBucketIndex);
        } else {
          mLocks.writeLock(from.mBucketIndex, to.mBucketIndex);
        }
        int fromTag = mTable.readTag(from.mBucketIndex, from.mSlotIndex);
        // if `to` is nonempty, or `from` is not occupied by original tag,
        // in both cases, abort this insertion.
        if (mTable.readTag(to.mBucketIndex, to.mSlotIndex) != NON_EXISTENT_TAG
            || fromTag != from.mTag) {
          if (depth == 1) {
            // NOTE: We must hold the locks of b1 and b2.
            // Or their slots may be preempted by another key if we released locks.
            mLocks.unlockWrite(b1, b2, to.mBucketIndex);
          } else {
            mLocks.unlockWrite(from.mBucketIndex, to.mBucketIndex);
          }
          return false;
        }
        // in concurrent scenarios,
        // an item may be aged twice.
        // Because that item move past the pointer,
        // we fix this by following code.
        int clock = mClockTable.readTag(from.mBucketIndex, from.mSlotIndex);
        int fromSegIndex = mLocks.getSegmentIndex(from.mBucketIndex);
        int toSegIndex = mLocks.getSegmentIndex(to.mBucketIndex);
        if (from.mBucketIndex
                % mLocks.getNumBucketsPerSegment() < mSegmentedAgingPointers[fromSegIndex]
                && to.mBucketIndex
                % mLocks.getNumBucketsPerSegment() >= mSegmentedAgingPointers[toSegIndex]) {
          clock = Math.min(mMaxAge, clock + 1);
        } else if (from.mBucketIndex
                % mLocks.getNumBucketsPerSegment() >= mSegmentedAgingPointers[fromSegIndex]
                && to.mBucketIndex
                % mLocks.getNumBucketsPerSegment() < mSegmentedAgingPointers[toSegIndex]) {
          clock = Math.max(0, clock - 1);
        }
        mTable.writeTag(to.mBucketIndex, to.mSlotIndex, fromTag);
        mClockTable.writeTag(to.mBucketIndex, to.mSlotIndex, clock);
        mScopeTable.writeTag(to.mBucketIndex, to.mSlotIndex,
            mScopeTable.readTag(from.mBucketIndex, from.mSlotIndex));
        mSizeTable.writeTag(to.mBucketIndex, to.mSlotIndex,
            mSizeTable.readTag(from.mBucketIndex, from.mSlotIndex));
        mTable.writeTag(from.mBucketIndex, from.mSlotIndex, 0);
        if (depth == 1) {
          // if to.mBucketIndex share the same lock with one of b1 and b2, we should not release its
          // lock
          int seg1 = mLocks.getSegmentIndex(b1);
          int seg2 = mLocks.getSegmentIndex(b2);
          int seg3 = mLocks.getSegmentIndex(to.mBucketIndex);
          if (seg3 != seg1 && seg3 != seg2) {
            mLocks.unlockWrite(to.mBucketIndex);
          }
        } else {
          mLocks.unlockWrite(from.mBucketIndex, to.mBucketIndex);
        }
        depth--;
      }
      return true;
    }
  
    /**
     * Find the `tag` in bucket `i`.
     *
     * @param i the bucket index
     * @param tag the tag (fingerprint)
     * @return true if no duplicated key is found, and `pos.slot` points to an empty slot (if pos.tag
     *         != -1); otherwise return false, and store the position of duplicated key in `pos.slot`.
     */
    private TagPosition tryFindInsertBucket(int i, int tag) {
      TagPosition pos = new TagPosition(i, -1, CuckooStatus.FAILURE_TABLE_FULL);
      for (int slotIndex = 0; slotIndex < TAGS_PER_BUCKET; slotIndex++) {
        int t = mTable.readTag(i, slotIndex);
        if (t != NON_EXISTENT_TAG) {
          if (t == tag) {
            pos.setSlotIndex(slotIndex);
            pos.setStatus(CuckooStatus.FAILURE_KEY_DUPLICATED);
            return pos;
          }
        } else {
          pos.setSlotIndex(slotIndex);
          pos.setStatus(CuckooStatus.OK);
        }
      }
      return pos;
    }
  
    /**
     * Lock two buckets and try opportunistic aging. Since we hold write locks, we can assure that
     * there are no other threads aging the same segment.
     *
     * @param b1 the first bucket
     * @param b2 the second bucket
     */
    private void writeLockAndOpportunisticAging(int b1, int b2) {
      mLocks.writeLock(b1, b2);
      opportunisticAgingSegment(mLocks.getSegmentIndex(b1));
      opportunisticAgingSegment(mLocks.getSegmentIndex(b2));
    }
  
    /**
     * Try opportunistic aging ith segment. Assume holding the lock of this segment.
     *
     * @param i the index of the segment to be aged
     */
    private void opportunisticAgingSegment(int i) {
      int bucketsToAge = computeAgingNumber();
      agingSegment(i, Math.min(bucketsToAge, MAX_AGING_PER_OPERATION));
    }
  
    /**
     * @Return the number of buckets should be aged.
     */
    private int computeAgingNumber() {
      int bucketsToAge;
      if (mSlidingWindowType == SlidingWindowType.NONE || mWindowSize < 0) {
        bucketsToAge = 0;
      } else if (mSlidingWindowType == SlidingWindowType.COUNT_BASED) {
        bucketsToAge =
            (int) (mNumBuckets * (mOperationCount.doubleValue() / (mWindowSize >> mBitsPerClock))
                - mAgingCount.get());
      } else {
        long elapsedTime = (System.currentTimeMillis() - mStartTime);
        bucketsToAge = Math.min(mNumBuckets,
            (int) (mNumBuckets * (elapsedTime / (double) (mWindowSize >> mBitsPerClock))
                - mAgingCount.get()));
      }
      return bucketsToAge;
    }
  
    /**
     * Aging the ith segment at most `maxAgingNumber` buckets. Assume holding the lock of this
     * segment.
     *
     * @param i the index of the segment to be aged
     * @param maxAgingNumber the maximum number of buckets to be aged
     */
    private int agingSegment(int i, int maxAgingNumber) {
      int bucketsPerSegment = mLocks.getNumBucketsPerSegment();
      int startPos = mLocks.getSegmentStartPos(i);
      int numAgedBuckets = 0;
      int numCleaned = 0;
      while (numAgedBuckets < maxAgingNumber) {
        int remainingBuckets = bucketsPerSegment - mSegmentedAgingPointers[i];
        if (remainingBuckets == 0) {
          break;
        }
        // age `AGING_STEP_SIZE` buckets and re-check window border
        int bucketsToAge = Math.min(AGING_STEP_SIZE, remainingBuckets);
        // advance agingCount to inform other threads before real aging
        int from = startPos + mSegmentedAgingPointers[i];
        mAgingCount.addAndGet(bucketsToAge);
        mSegmentedAgingPointers[i] += bucketsToAge;
        numCleaned += agingRange(from, from + bucketsToAge);
        numAgedBuckets += bucketsToAge;
      }
      return numCleaned;
    }
  
    /**
     * Aging buckets in range [from, to]. Assume holding the locks of this range.
     *
     * @param from the start bucket of the range to be aged
     * @param to the end bucket of the range to be aged
     * @return the number of cleaned buckets
     */
    private int agingRange(int from, int to) {
      int numCleaned = 0;
      for (int i = from; i < to; i++) {
        numCleaned += agingBucket(i);
      }
      return numCleaned;
    }
  
    /**
     * @param b the bucket to be aged
     * @return the number of cleaned slots
     */
    private int agingBucket(int b) {
      int numCleaned = 0;
      for (int slotIndex = 0; slotIndex < TAGS_PER_BUCKET; slotIndex++) {
        int tag = mTable.readTag(b, slotIndex);
        if (tag == NON_EXISTENT_TAG) {
          continue;
        }
        int oldClock = mClockTable.readTag(b, slotIndex);
        if (oldClock > 0) {
          mClockTable.writeTag(b, slotIndex, oldClock - 1);
        } else {
          // evict stale item
          numCleaned++;
          mTable.writeTag(b, slotIndex, NON_EXISTENT_TAG);
          mNumItems.decrementAndGet();
          int scope = mScopeTable.readTag(b, slotIndex);
          int size = mSizeTable.readTag(b, slotIndex);
          size = decodeSize(size);
          if (mOpenSizeEncoder) {
            size = mSizeEncoder.dec(size);
          }
          updateScopeStatistics(scope, -1, -size);
          mTotalBytes.addAndGet(-size);
        }
      }
      return numCleaned;
    }
  
    /**
     * A basic entry that records information of a path node during cuckoo BFS search.
     */
    static final class BFSEntry {
      public int mBucketIndex;
      public int mPathcode; // encode slot position of ancestors and it own nodes
      public int mDepth;
  
      /**
       * @param bucketIndex the bucket of entry
       * @param pathcode the encoded slot position of ancestors and it own
       * @param depth the depth of the entry in path
       */
      BFSEntry(int bucketIndex, int pathcode, int depth) {
        mBucketIndex = bucketIndex;
        mPathcode = pathcode;
        mDepth = depth;
      }
    }
  
    /**
     * This class represents a detailed cuckoo record, include its position and stored value.
     */
    static final class CuckooRecord {
      public int mBucketIndex;
      public int mSlotIndex;
      public int mTag;
  
      CuckooRecord() {
        this(-1, -1, 0);
      }
  
      /**
       * @param bucketIndex the bucket of this record
       * @param slotIndex the slot of this record
       * @param tag the tag (fingerprint) of this record
       */
      CuckooRecord(int bucketIndex, int slotIndex, int tag) {
        mBucketIndex = bucketIndex;
        mSlotIndex = slotIndex;
        mTag = tag;
      }
    }
  }
  