核心限流类FrequencySketch

限流的核心类

/*
 * Copyright 2015 Ben Manes. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.github.benmanes.caffeine.cache;

import static com.github.benmanes.caffeine.cache.Caffeine.requireArgument;

import com.google.errorprone.annotations.Var;

/**
 * A probabilistic multiset for estimating the popularity of an element within a time window. The
 * maximum frequency of an element is limited to 15 (4-bits) and an aging process periodically
 * halves the popularity of all elements.
 *
 * @author ben.manes@gmail.com (Ben Manes)
 */
final class FrequencySketch<E> {

  /*
   * This class maintains a 4-bit CountMinSketch [1] with periodic aging to provide the popularity
   * history for the TinyLfu admission policy [2]. The time and space efficiency of the sketch
   * allows it to cheaply estimate the frequency of an entry in a stream of cache access events.
   *
   * The counter matrix is represented as a single-dimensional array holding 16 counters per slot. A
   * fixed depth of four balances the accuracy and cost, resulting in a width of four times the
   * length of the array. To retain an accurate estimation, the array's length equals the maximum
   * number of entries in the cache, increased to the closest power-of-two to exploit more efficient
   * bit masking. This configuration results in a confidence of 93.75% and an error bound of
   * e / width.
   *
   * To improve hardware efficiency, an item's counters are constrained to a 64-byte block, which is
   * the size of an L1 cache line. This differs from the theoretical ideal where counters are
   * uniformly distributed to minimize collisions. In that configuration, the memory accesses are
   * not predictable and lack spatial locality, which may cause the pipeline to need to wait for
   * four memory loads. Instead, the items are uniformly distributed to blocks, and each counter is
   * uniformly selected from a distinct 16-byte segment. While the runtime memory layout may result
   * in the blocks not being cache-aligned, the L2 spatial prefetcher tries to load aligned pairs of
   * cache lines, so the typical cost is only one memory access.
   *
   * The frequency of all entries is aged periodically using a sampling window based on the maximum
   * number of entries in the cache. This is referred to as the reset operation by TinyLfu and keeps
   * the sketch fresh by dividing all counters by two and subtracting based on the number of odd
   * counters found. The O(n) cost of aging is amortized, ideal for hardware prefetching, and uses
   * inexpensive bit manipulations per array location.
   *
   * [1] An Improved Data Stream Summary: The Count-Min Sketch and its Applications
   * http://dimacs.rutgers.edu/~graham/pubs/papers/cm-full.pdf
   * [2] TinyLFU: A Highly Efficient Cache Admission Policy
   * https://dl.acm.org/citation.cfm?id=3149371
   * [3] Hash Function Prospector: Three round functions
   * https://github.com/skeeto/hash-prospector#three-round-functions
   */

  static final long RESET_MASK = 0x7777777777777777L;
  static final long ONE_MASK = 0x1111111111111111L;

  int sampleSize;
  int blockMask;
  long[] table;
  int size;

  /**
   * Creates a lazily initialized frequency sketch, requiring {@link #ensureCapacity} be called
   * when the maximum size of the cache has been determined.
   */
  @SuppressWarnings("NullAway.Init")
  public FrequencySketch() {}

  /**
   * Initializes and increases the capacity of this {@code FrequencySketch} instance, if necessary,
   * to ensure that it can accurately estimate the popularity of elements given the maximum size of
   * the cache. This operation forgets all previous counts when resizing.
   *
   * @param maximumSize the maximum size of the cache
   */
  @SuppressWarnings("Varifier")
  public void ensureCapacity(long maximumSize) {
    requireArgument(maximumSize >= 0);
    int maximum = (int) Math.min(maximumSize, Integer.MAX_VALUE >>> 1);
    if ((table != null) && (table.length >= maximum)) {
      return;
    }

    table = new long[Math.max(Caffeine.ceilingPowerOfTwo(maximum), 8)];
    sampleSize = (maximumSize == 0) ? 10 : (10 * maximum);
    blockMask = (table.length >>> 3) - 1;
    if (sampleSize <= 0) {
      sampleSize = Integer.MAX_VALUE;
    }
    size = 0;
  }

  /**
   * Returns if the sketch has not yet been initialized, requiring that {@link #ensureCapacity} is
   * called before it begins to track frequencies.
   */
  public boolean isNotInitialized() {
    return (table == null);
  }

  /**
   * Returns the estimated number of occurrences of an element, up to the maximum (15).
   *
   * @param e the element to count occurrences of
   * @return the estimated number of occurrences of the element; possibly zero but never negative
   */
  @SuppressWarnings("Varifier")
  public int frequency(E e) {
    if (isNotInitialized()) {
      return 0;
    }

    @Var int frequency = Integer.MAX_VALUE;
    int blockHash = spread(e.hashCode());
    int counterHash = rehash(blockHash);
    int block = (blockHash & blockMask) << 3;
    for (int i = 0; i < 4; i++) {
      int h = counterHash >>> (i << 3);
      int index = (h >>> 1) & 15;
      int offset = h & 1;
      int slot = block + offset + (i << 1);
      int count = (int) ((table[slot] >>> (index << 2)) & 0xfL);
      frequency = Math.min(frequency, count);
    }
    return frequency;
  }

  /**
   * Increments the popularity of the element if it does not exceed the maximum (15). The popularity
   * of all elements will be periodically down sampled when the observed events exceed a threshold.
   * This process provides a frequency aging to allow expired long term entries to fade away.
   *
   * @param e the element to add
   */
  @SuppressWarnings({"ShortCircuitBoolean", "UnnecessaryLocalVariable"})
  public void increment(E e) {
    if (isNotInitialized()) {
      return;
    }

    int blockHash = spread(e.hashCode());
    int counterHash = rehash(blockHash);
    int block = (blockHash & blockMask) << 3;

    // Loop unrolling improves throughput by 10m ops/s
    int h0 = counterHash;
    int h1 = counterHash >>> 8;
    int h2 = counterHash >>> 16;
    int h3 = counterHash >>> 24;

    int index0 = (h0 >>> 1) & 15;
    int index1 = (h1 >>> 1) & 15;
    int index2 = (h2 >>> 1) & 15;
    int index3 = (h3 >>> 1) & 15;

    int slot0 = block + (h0 & 1);
    int slot1 = block + (h1 & 1) + 2;
    int slot2 = block + (h2 & 1) + 4;
    int slot3 = block + (h3 & 1) + 6;

    boolean added =
          incrementAt(slot0, index0)
        | incrementAt(slot1, index1)
        | incrementAt(slot2, index2)
        | incrementAt(slot3, index3);

    if (added && (++size == sampleSize)) {
      reset();
    }
  }

  /** Applies a supplemental hash function to defend against a poor quality hash. */
  static int spread(@Var int x) {
    x ^= x >>> 17;
    x *= 0xed5ad4bb;
    x ^= x >>> 11;
    x *= 0xac4c1b51;
    x ^= x >>> 15;
    return x;
  }

  /** Applies another round of hashing for additional randomization. */
  static int rehash(@Var int x) {
    x *= 0x31848bab;
    x ^= x >>> 14;
    return x;
  }

  /**
   * Increments the specified counter by 1 if it is not already at the maximum value (15).
   *
   * @param i the table index (16 counters)
   * @param j the counter to increment
   * @return if incremented
   */
  boolean incrementAt(int i, int j) {
    int offset = j << 2;
    long mask = (0xfL << offset);
    if ((table[i] & mask) != mask) {
      table[i] += (1L << offset);
      return true;
    }
    return false;
  }

  /** Reduces every counter by half of its original value. */
  void reset() {
    @Var int count = 0;
    for (int i = 0; i < table.length; i++) {
      count += Long.bitCount(table[i] & ONE_MASK);
      table[i] = (table[i] >>> 1) & RESET_MASK;
    }
    size = (size - (count >>> 2)) >>> 1;
  }
}

第一个方法

/**
 * Initializes and increases the capacity of this {@code FrequencySketch} instance, if necessary,
 * to ensure that it can accurately estimate the popularity of elements given the maximum size of
 * the cache. This operation forgets all previous counts when resizing.
 *
 * @param maximumSize the maximum size of the cache
 */
@SuppressWarnings("Varifier")
public void ensureCapacity(long maximumSize) {
  requireArgument(maximumSize >= 0);
  int maximum = (int) Math.min(maximumSize, Integer.MAX_VALUE >>> 1); // 这个是最大容量是 maximumSize 最大不超过2^30 -1 大概10亿左右
  if ((table != null) && (table.length >= maximum)) {
    return;
  }

  table = new long[Math.max(Caffeine.ceilingPowerOfTwo(maximum), 8)]; // 分配table的长度，最低是8
  sampleSize = (maximumSize == 0) ? 10 : (10 * maximum); // 采样 传进来的值是0，默认为10，不为0 就10 *  maximum  决定“每添加多少条才检查一次容量”，用来平衡精度与性能。
  blockMask = (table.length >>> 3) - 1; // blockMask 提供“快速取模 8”的位运算掩码，把计数或回收任务分散到 8 个条带，减少并发冲突。
  if (sampleSize <= 0) {
    sampleSize = Integer.MAX_VALUE;
  }
  size = 0;
}

/** Returns the smallest power of two greater than or equal to {@code x}. */
static int ceilingPowerOfTwo(int x) {
  // From Hacker's Delight, Chapter 3, Harry S. Warren Jr.
  return 1 << -Integer.numberOfLeadingZeros(x - 1);
}
// 把x对齐到最近的2的次幂, x-1防止x本身已经是2的幂 Integer.numberOfLeadingZeros 这个是返回最高位1前面0的个数

@IntrinsicCandidate
public static int numberOfLeadingZeros(int i) {
        // HD, Count leading 0's
        if (i <= 0)
            return i == 0 ? 32 : 0;
        int n = 31;
        if (i >= 1 << 16) { n -= 16; i >>>= 16; }
        if (i >= 1 <<  8) { n -=  8; i >>>=  8; }
        if (i >= 1 <<  4) { n -=  4; i >>>=  4; }
        if (i >= 1 <<  2) { n -=  2; i >>>=  2; }
        return n - (i >>> 1);
    }
// @IntrinsicCandidate 告诉 HotSpot：**“这个方法可以被虚拟机替换成一条机器指令”**。在 x86 上就是 `LZCNT`（如果 CPU 支持）或 `BSR`+简单换算，几乎一个时钟周期完成，比 Java 代码更快。
// 这几个if其实就是二分法了，先算高16 再往下找 然后返回前面0的个数，可以方便

第二个方法

/**
  * Returns the estimated number of occurrences of an element, up to the maximum (15).
  *
  * @param e the element to count occurrences of
  * @return the estimated number of occurrences of the element; possibly zero but never negative
  */
 @SuppressWarnings("Varifier")
 public int frequency(E e) {
   if (isNotInitialized()) { // 缓存还没被启用就当成从未访问过
     return 0;
   }

   @Var int frequency = Integer.MAX_VALUE;
   int blockHash = spread(e.hashCode());
   int counterHash = rehash(blockHash);
   int block = (blockHash & blockMask) << 3;  // 用来把“元素哈希”映射到一条 8-byte 物理缓存行（64 bit） 上
   for (int i = 0; i < 4; i++) {  // 这段是计数器算法的核心实现，循环4次
     int h = counterHash >>> (i << 3);  // hash 是32位的 i << 3就是8 把hash 分成4部分
     int index = (h >>> 1) & 15;        //4bit 的计数值是0-15
     int offset = h & 1;                //0 或者1 决定使用long中的哪一个
     int slot = block + offset + (i << 1);// 4次循环分别落在 block + 0, block + 2, block + 4, block + 6
     int count = (int) ((table[slot] >>> (index << 2)) & 0xfL); //取出 4bit 并计数 0-15
     frequency = Math.min(frequency, count);
   }
   return frequency;
 }

 /** Applies a supplemental hash function to defend against a poor quality hash. */
 static int spread(@Var int x) { // hash扰乱函数，确保hash的随机性 这个常数确实有点吊
   x ^= x >>> 17;
   x *= 0xed5ad4bb;
   x ^= x >>> 11;
   x *= 0xac4c1b51;
   x ^= x >>> 15;
   return x;
 }

 /** Applies another round of hashing for additional randomization. */
 static int rehash(@Var int x) { // 再hash 随后 4 轮循环每次 counterHash >>> (i<<3) 取不同字节，因为经过二次扰动，4 个字节之间几乎独立，保证 4 个采样计数器不会系统性偏向同一值，TinyLFU 的“最小值”估计才更准。
   x *= 0x31848bab;
   x ^= x >>> 14;
   return x;
 }

这里讲一下block是什么。物理上，一个block = 8个long 1个long 是8个字节，在几乎所有现代主流 CPU（Intel Core, AMD Ryzen, ARM Neoverse）中，L1/L2 Cache 的基本管理单位——“缓存行”的大小固定为 64 字节（64 Bytes）。

所以用8个long，刚好是一个缓存行，一次cpu操作就可以拿到整个block的数据。可以算是针对硬件架构的特性做的优化。

/**
 * Increments the popularity of the element if it does not exceed the maximum (15). The popularity
 * of all elements will be periodically down sampled when the observed events exceed a threshold.
 * This process provides a frequency aging to allow expired long term entries to fade away.
 *
 * @param e the element to add
 */
@SuppressWarnings({"ShortCircuitBoolean", "UnnecessaryLocalVariable"})
public void increment(Object e) {  // 核心计数的方法
  if (isNotInitialized()) {
    return;
  }

  int blockHash = spread(e.hashCode()); 
  int counterHash = rehash(blockHash);      // 通过rehash 算出一个更均匀分布的hash
  int block = (blockHash & blockMask) << 3; // 计算数据落在哪一个block里，缓存行的优势所在

  // Loop unrolling improves throughput by 10m ops/s
  int h0 = counterHash;                     //这里就是计数了，对于同一个对象来说，会在四个不同的long里进行操作
  int h1 = counterHash >>> 8;
  int h2 = counterHash >>> 16;
  int h3 = counterHash >>> 24;

  int index0 = (h0 >>> 1) & 15;
  int index1 = (h1 >>> 1) & 15;
  int index2 = (h2 >>> 1) & 15;
  int index3 = (h3 >>> 1) & 15;

  int slot0 = block + (h0 & 1);
  int slot1 = block + (h1 & 1) + 2;
  int slot2 = block + (h2 & 1) + 4;
  int slot3 = block + (h3 & 1) + 6;

  boolean added =
        incrementAt(slot0, index0)
      | incrementAt(slot1, index1)
      | incrementAt(slot2, index2)
      | incrementAt(slot3, index3);

  if (added && (++size == sampleSize)) {
    reset();
  }
}

/**
 * Increments the specified counter by 1 if it is not already at the maximum value (15).
 *
 * @param i the table index (16 counters)
 * @param j the counter to increment
 * @return if incremented
 */
boolean incrementAt(int i, int j) { //这段算计数器加1，这里也有个很精妙的操作，无锁原子更新
  int offset = j << 2; // 等于j * 4
  long mask = (0xfL << offset); //16
  if ((table[i] & mask) != mask) { 不是1111 
    table[i] += (1L << offset); // 直接加1
    return true;
  }
  return false;
}