28160818793

Committed 25 Jun 2026 09:33AM UTC coverage: 92.647% (-0.08%) from 92.727%

Build # 28160818793

Build Type

push

github

Committed by

Hyshmily

Commit Message

fix: add common JAR to publish workflow

Coverage Stats

1198 of 1351 branches covered (88.68%)

Branch coverage included in aggregate %.

3426 of 3640 relevant lines covered (94.12%)

4.25 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.91

/common/src/main/java/io/github/hyshmily/hotkey/hotkeydetector/heavykeeper/HeavyKeeper.java

/*
 * Copyright 2026 Hyshmily. 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 io.github.hyshmily.hotkey.hotkeydetector.heavykeeper;

import com.google.common.hash.Hashing;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.LongAdder;
import java.util.stream.Collectors;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

/**
 * HeavyKeeper — a Count-Min Sketch variant for approximate Top‑K tracking
 * of frequently accessed keys.
 *
 * <p><b>Algorithm overview:</b> Uses a 2D count array ({@code depth × width})
 * with per-slot fingerprint verification and probabilistic decay to estimate
 * the most frequent keys using bounded memory. Each key is hashed into one
 * bucket per row; if the stored fingerprint matches, the counter is
 * incremented; otherwise a probabilistic decay (sampled from a Binomial
 * distribution) determines whether the existing counter survives or is
 * replaced. This design excels at filtering out low-frequency items while
 * preserving high-frequency key rankings with low error rates.
 *
 * <p><b>Concurrency model:</b> Thread-safe with two lock tiers:
 * <ol>
 *   <li>Fine-grained striped synchronization ({@link #LOCK_STRIPES} stripes)
 *       on individual sketch buckets for low-contention sketch updates.</li>
 *   <li>A single coarse-grained lock on the sorted TopK heap
 *       ({@code sortedTopK}) for heap mutations and reads.</li>
 * </ol>
 * The separation ensures that sketch writes (the hot path) rarely contend
 * with heap operations (which happen only when a key crosses a threshold).
 *
 * <p><b>Decay:</b> {@link #fading()} halves all counters periodically to
 * age out historical data, keeping the TopK set reflective of recent access
 * patterns.
 */
@Slf4j
public class HeavyKeeper implements TopK {

  /** Pre-computed decay probability lookup table size ({@value}). */
  private static final int LOOKUP_TABLE_SIZE = 256;
  /** Number of lock stripes for fine-grained concurrency ({@value}). Must be a power of two. */
  private static final int LOCK_STRIPES = 256;

  /**
   * -- GETTER --
   * Maximum number of hot keys tracked.
   */
  @Getter
  private final int k;

  /**
   * -- GETTER --
   * Width of the Count-Min Sketch (columns per row).
   */
  @Getter
  private final int width;

  /**
   * -- GETTER --
   * Depth of the Count-Min Sketch (rows / hash functions).
   */
  @Getter
  private final int depth;

  /** Pre-computed decay probabilities: {@code decay^i} for {@code i in [0, LOOKUP_TABLE_SIZE)}. */
  private final double[] lookupTable;
  /** Per-slot fingerprint values for collision verification in the Count-Min Sketch. */
  private final long[] fingerprints;
  /** Per-slot frequency counters for the Count-Min Sketch. */
  private final long[] counts;
  /** Striped lock objects for fine-grained concurrency on sketch slot updates. */
  private final Object[] lockStripes;
  /** Bitmask for mapping bucket index to lock stripe (stripe count must be power of two). */
  private final int lockMask;
  /** Sorted map of current TopK entries, ordered by count ascending then key lexicographically.
   * Boolean.TRUE is a structural placeholder.  A TreeSet would not safely remove a Node
   * whose count had changed because the Comparator includes the count; TreeMap.remove()
   * uses the comparator (not identity — Node lacks equals/hashCode), but since Node. Count
   * is final the comparator result for a given Node instance is stable and removal works. */
  private final TreeMap<Node, Boolean> sortedTopK;
  /** Reverse index from key to its {@link Node} in the sorted map, for O(1) lookups. */
  private final Map<String, Node> heapIndex;
  /** Bounded blocking queue receiving expelled (evicted) key-count items for downstream consumption. */
  private final BlockingQueue<Item> expelledQueue;
  /** Running total of all tracked data streams since startup or last {@link #fading()}. */
  private final LongAdder total;

  /**
   * -- GETTER --
   * Minimum count threshold before a key can enter the TopK set.
   */
  @Getter
  private final int minCount;

  /**
   * Construct a HeavyKeeper instance.
   *
   * @param k        maximum number of hot keys to track
   * @param width    width of the Count-Min Sketch (number of columns per row)
   * @param depth    depth of the Count-Min Sketch (number of rows / hash functions)
   * @param decay    probabilistic decay factor (0.0–1.0); higher values preserve counts longer
   * @param minCount minimum count threshold before a key can enter the TopK set
   */
  public HeavyKeeper(int k, int width, int depth, double decay, int minCount) {
    this(k, width, depth, decay, minCount, 50_000);
  }

  /**
   * Construct a HeavyKeeper instance with a custom expelled-queue capacity.
   *
   * @param k                     maximum number of hot keys to track
   * @param width                 width of the Count-Min Sketch (number of columns per row)
   * @param depth                 depth of the Count-Min Sketch (number of rows / hash functions)
   * @param decay                 probabilistic decay factor (0.0–1.0); higher values preserve counts longer
   * @param minCount              minimum count threshold before a key can enter the TopK set
   * @param expelledQueueCapacity capacity of the bounded blocking queue for expelled items
   */
  public HeavyKeeper(int k, int width, int depth, double decay, int minCount, int expelledQueueCapacity) {
    if (k <= 0) {
      throw new IllegalArgumentException("TopK must be greater than 0, but got: " + k);
    }
    this.k = k;
    this.width = width;
    this.depth = depth;
    this.minCount = minCount;

    this.lookupTable = new double[LOOKUP_TABLE_SIZE];
    for (int i = 0; i < LOOKUP_TABLE_SIZE; i++) {
      lookupTable[i] = Math.pow(decay, i);
    }

    int totalSlots = depth * width;
    this.fingerprints = new long[totalSlots];
    this.counts = new long[totalSlots];
    this.lockStripes = new Object[LOCK_STRIPES];
    for (int i = 0; i < LOCK_STRIPES; i++) {
      lockStripes[i] = new Object();
    }
    this.lockMask = LOCK_STRIPES - 1;

    this.sortedTopK = new TreeMap<>(Comparator.comparingLong((Node a) -> a.count).thenComparing(a -> a.key));
    this.heapIndex = new ConcurrentHashMap<>();
    this.expelledQueue = new ArrayBlockingQueue<>(expelledQueueCapacity);
    this.total = new LongAdder();
  }

  /**
   * Record access to the given key.
   *
   * <p>The returned {@link AddResult} carries an {@code expelledKey} when the key
   * entered the TopK set by displacing a previous member (non-null expelledKey and
   * {@code isHot == true}).  When the key was already in the TopK set or failed to
   * meet the minimum count threshold, {@code expelledKey} is {@code null}.
   *
   * @param key       the cache key being accessed
   * @param increment the frequency increment (typically 1)
   * @return an {@link AddResult} with expelled key (or null), hot status, and the input key
   */
  @Override
  @SuppressWarnings("null")
  public AddResult addDirect(String key, int increment) {
    /* Compute fingerprint with Guava Murmur3_32 (fixed seed ensures same key -> same fingerprint) */
    long itemFingerprint = Hashing.murmur3_32_fixed().hashString(key, StandardCharsets.UTF_8).padToLong() & 0xFFFFFFFFL;
    long maxCount = 0L;

    for (int i = 0; i < depth; i++) {
      int hash = (int) (itemFingerprint ^ (i * 0x9e3779b97f4a7c15L));
      int bucketIndex = Math.floorMod(hash, width);
      int index = i * width + bucketIndex;
      Object lock = lockStripes[index & lockMask];

      synchronized (lock) {
        if (counts[index] == 0) {
          fingerprints[index] = itemFingerprint;
          counts[index] = increment;

          maxCount = Math.max(maxCount, increment);
        } else if (fingerprints[index] == itemFingerprint) {
          counts[index] += increment;

          maxCount = Math.max(maxCount, counts[index]);
        } else {
          // HeavyKeeper probabilistic decay: sample the decay count from a
          // Binomial(increment, decayProb) distribution in O(1) instead of
          // looping increment times.  This keeps lock hold time bounded even
          // when the Worker-side batch increment is large.
          ThreadLocalRandom rng = ThreadLocalRandom.current();
          double decayProb = (counts[index] < LOOKUP_TABLE_SIZE)
            ? lookupTable[(int) counts[index]]
            : lookupTable[LOOKUP_TABLE_SIZE - 1];

          int decays = sampleBinomial(increment, decayProb, rng);
          if (decays >= counts[index]) {
            fingerprints[index] = itemFingerprint;
            counts[index] = increment;
          } else {
            counts[index] -= decays;
          }
          maxCount = Math.max(maxCount, counts[index]);
        }
      }
    }

    total.add(increment);

    if (maxCount < minCount) {
      return AddResult.cold();
    }

    synchronized (sortedTopK) {
      Node existing = heapIndex.remove(key);
      if (existing != null) {
        sortedTopK.remove(existing);
      }

      boolean isHot = false;
      String expelled = null;

      if (sortedTopK.size() < k || maxCount >= sortedTopK.firstKey().count) {
        Node newNode = new Node(key, maxCount);
        if (sortedTopK.size() >= k) {
          Node removed = sortedTopK.pollFirstEntry().getKey();
          expelled = removed.key;
          heapIndex.remove(removed.key);
          if (!expelledQueue.offer(new Item(expelled, removed.count))) {
            log.warn("Expelled queue full, dropping key: {}", expelled);
          }
        }
        sortedTopK.put(newNode, Boolean.TRUE);
        heapIndex.put(key, newNode);
        isHot = true;
      }
      return new AddResult(expelled, isHot, key);
    }
  }

  /**
   * Record accesses for multiple keys.
   *
   * <p>Updates sketch counters for all keys, then updates the TopK heap with
   * keys whose estimated count meets the minimum threshold.  Returns results
   * only for keys that entered the TopK set (possibly displacing others).
   *
   * @param keyCounts map of keys to their access counts
   * @return list of {@link AddResult} for keys that entered the TopK set
   */
  @Override
  public List<AddResult> addDirect(Map<String, Long> keyCounts) {
    Map<String, Long> maxCounts = new HashMap<>(keyCounts.size());
    for (var entry : keyCounts.entrySet()) {
      long max = addToSketch(entry.getKey(), entry.getValue());
      maxCounts.put(entry.getKey(), max);
    }

    List<AddResult> results = new ArrayList<>();

    synchronized (sortedTopK) {
      for (var entry : keyCounts.entrySet()) {
        String key = entry.getKey();
        long maxCount = maxCounts.get(key);
        if (maxCount < minCount) {
          continue;
        }

        Node existing = heapIndex.remove(key);
        if (existing != null) {
          sortedTopK.remove(existing);
        }

        boolean shouldInsert = sortedTopK.size() < k || maxCount >= sortedTopK.firstKey().count;
        if (shouldInsert) {
          Node newNode = new Node(key, maxCount);
          String expelledKey = null;
          if (sortedTopK.size() >= k) {
            Node removed = sortedTopK.pollFirstEntry().getKey();
            expelledKey = removed.key;
            heapIndex.remove(removed.key);
            if (!expelledQueue.offer(new Item(removed.key, removed.count))) {
              log.warn("Expelled queue full, dropping key: {}", removed.key);
            }
          }
          sortedTopK.put(newNode, Boolean.TRUE);
          heapIndex.put(key, newNode);
          results.add(new AddResult(expelledKey, true, key));
        }
      }
    }
    return results;
  }

  @SuppressWarnings("null")
  private long addToSketch(String key, long increment) {
    long itemFingerprint = Hashing.murmur3_32_fixed().hashString(key, StandardCharsets.UTF_8).padToLong() & 0xFFFFFFFFL;
    long maxCount = 0;

    for (int i = 0; i < depth; i++) {
      int hash = (int) (itemFingerprint ^ (i * 0x9e3779b97f4a7c15L));
      int bucketIndex = Math.floorMod(hash, width);
      int index = i * width + bucketIndex;
      Object lock = lockStripes[index & lockMask];

      synchronized (lock) {
        if (counts[index] == 0) {
          fingerprints[index] = itemFingerprint;
          counts[index] = increment;

          maxCount = Math.max(maxCount, increment);
        } else if (fingerprints[index] == itemFingerprint) {
          counts[index] += increment;

          maxCount = Math.max(maxCount, counts[index]);
        } else {
          ThreadLocalRandom rng = ThreadLocalRandom.current();
          double decayProb = (counts[index] < LOOKUP_TABLE_SIZE)
            ? lookupTable[(int) counts[index]]
            : lookupTable[LOOKUP_TABLE_SIZE - 1];
          int decays = sampleBinomial((int) Math.min(increment, Integer.MAX_VALUE), decayProb, rng);

          if (decays >= counts[index]) {
            fingerprints[index] = itemFingerprint;
            counts[index] = increment;
          } else {
            counts[index] -= decays;
          }
          maxCount = Math.max(maxCount, counts[index]);
        }
      }
    }
    total.add(increment);
    return maxCount;
  }

  /**
   * Return all keys currently in the TopK set, sorted by estimated count
   * descending (highest frequency first).
   *
   * <p>The returned list is a point-in-time snapshot: it is safe to iterate
   * after the lock is released but may be stale immediately.
   *
   * @return a point-in-time list of {@link Item} entries, ordered from highest
   *         to lowest estimated count; never {@code null}
   */
  @Override
  public List<Item> list() {
    synchronized (sortedTopK) {
      List<Item> result = new ArrayList<>(sortedTopK.size());
      for (Node node : sortedTopK.descendingKeySet()) {
        result.add(new Item(node.key, node.count));
      }
      return result;
    }
  }

  /**
   * Check whether a key is currently in the TopK set.
   *
   * @param key the cache key to test
   * @return {@code true} if the key is in the TopK set, {@code false} otherwise
   */
  @Override
  public boolean contains(String key) {
    return heapIndex.containsKey(key);
  }

  /**
   * Return the blocking queue holding items that have been evicted from the TopK set.
   * Consumers should drain this queue periodically for asynchronous processing.
   *
   * @return a blocking queue of evicted items
   */
  @Override
  public BlockingQueue<Item> expelled() {
    return expelledQueue;
  }

  /**
   * Halve all frequency counters in the sketch and the sorted TopK heap.
   *
   * <p>Entries whose count drops to zero after halving are removed from
   * the heap entirely. The running total is also halved. This periodic
   * decay prevents the sketch from saturating with stale historical data
   * and ensures that the TopK ranking reflects recent access patterns
   * rather than cumulative lifetime counts.
   *
   * <p>This operation is invoked automatically by a scheduler at a
   * configured interval (typically 30 seconds). Calling it concurrently
   * with {@link #addDirect} is safe — sketch counters are halved under
   * per-stripe locks and the heap is rebuilt atomically under the shared
   * heap lock.
   */
  @Override
  public void fading() {
    // Halve all sketch counters under per-stripe locks to prevent concurrent
    // addDirect() from observing torn long values (JLS 17.7).
    // Lock overhead is negligible — called once per decay interval (~30 s).
    for (int i = 0; i < counts.length; i++) {
      synchronized (lockStripes[i & lockMask]) {
        counts[i] >>= 1;
      }
    }

    // Rebuild the TopK set: halve all counts and discard entries that drop to 0.
    synchronized (sortedTopK) {
      TreeMap<Node, Boolean> newMap = new TreeMap<>(sortedTopK.comparator());
      heapIndex.clear();
      for (Node node : sortedTopK.keySet()) {
        long half = node.count >> 1;
        if (half > 0) {
          Node newNode = new Node(node.key, half);
          newMap.put(newNode, Boolean.TRUE);
          heapIndex.put(node.key, newNode);
        }
      }
      sortedTopK.clear();
      sortedTopK.putAll(newMap);

      long half = total.sumThenReset() >> 1;
      if (half > 0) {
        total.add(half);
      }
    }
  }

  /**
   * Return the total number of data streams tracked since startup or last reset.
   *
   * @return total access count
   */
  @Override
  public long total() {
    return total.sum();
  }

  /**
   * Return the top {@code n} hot keys, ordered by estimated count descending.
   *
   * <p>If fewer than {@code n} keys are currently tracked, the returned list
   * contains all available keys. The result is a point-in-time snapshot.
   *
   * @param n maximum number of keys to return (must be non-negative)
   * @return list of at most {@code n} {@link Item} entries, never {@code null};
   *         empty if no keys are tracked or {@code n == 0}
   */
  @Override
  public List<Item> listTopN(int n) {
    synchronized (sortedTopK) {
      return sortedTopK
        .descendingKeySet()
        .stream()
        .limit(n)
        .map(node -> new Item(node.key, node.count))
        .collect(Collectors.toList());
    }
  }

  /** A key-count pair used as an entry in the sorted TopK tree. */
  @AllArgsConstructor
  private static class Node {

    /** The cache key. */
    final String key;
    /** Its current estimated count. */
    final long count;
  }

  /**
   * Sample from a Binomial(n, p) distribution in O(1) expected time.
   *
   * <p>Uses direct simulation for small n (≤10) and normal approximation
   * for larger n when np(1-p) > 9.  Falls back to direct simulation for
   * moderate n.
   */
  private static int sampleBinomial(int n, double p, ThreadLocalRandom rng) {
    if (n <= 0) {
      return 0;
    }
    if (p >= 1.0) {
      return n;
    }
    if (p <= 0.0) {
      return 0;
    }

    double q = 1.0 - p;

    if (n <= 10) {
      int k = 0;
      for (int i = 0; i < n; i++) {
        if (rng.nextDouble() < p) {
          k++;
        }
      }
      return k;
    }

    double np = n * p;
    double npq = np * q;

    if (npq > 9.0) {
      int k = (int) Math.round(np + Math.sqrt(npq) * rng.nextGaussian());
      return Math.max(0, Math.min(n, k));
    }

    int k = 0;
    for (int i = 0; i < n; i++) {
      if (rng.nextDouble() < p) {
        k++;
      }
    }
    return k;
  }
}

1	/*
2	* Copyright 2026 Hyshmily. All Rights Reserved.
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16	package io.github.hyshmily.hotkey.hotkeydetector.heavykeeper;
17
18	import com.google.common.hash.Hashing;
19	import java.nio.charset.StandardCharsets;
20	import java.util.*;
21	import java.util.concurrent.ArrayBlockingQueue;
22	import java.util.concurrent.BlockingQueue;
23	import java.util.concurrent.ConcurrentHashMap;
24	import java.util.concurrent.ThreadLocalRandom;
25	import java.util.concurrent.atomic.LongAdder;
26	import java.util.stream.Collectors;
27	import lombok.AllArgsConstructor;
28	import lombok.Getter;
29	import lombok.extern.slf4j.Slf4j;
30
31	/**
32	* HeavyKeeper — a Count-Min Sketch variant for approximate Top‑K tracking
33	* of frequently accessed keys.
34	*
35	* <p><b>Algorithm overview:</b> Uses a 2D count array ({@code depth × width})
36	* with per-slot fingerprint verification and probabilistic decay to estimate
37	* the most frequent keys using bounded memory. Each key is hashed into one
38	* bucket per row; if the stored fingerprint matches, the counter is
39	* incremented; otherwise a probabilistic decay (sampled from a Binomial
40	* distribution) determines whether the existing counter survives or is
41	* replaced. This design excels at filtering out low-frequency items while
42	* preserving high-frequency key rankings with low error rates.
43	*
44	* <p><b>Concurrency model:</b> Thread-safe with two lock tiers:
45	* <ol>
46	* <li>Fine-grained striped synchronization ({@link #LOCK_STRIPES} stripes)
47	* on individual sketch buckets for low-contention sketch updates.</li>
48	* <li>A single coarse-grained lock on the sorted TopK heap
49	* ({@code sortedTopK}) for heap mutations and reads.</li>
50	* </ol>
51	* The separation ensures that sketch writes (the hot path) rarely contend
52	* with heap operations (which happen only when a key crosses a threshold).
53	*
54	* <p><b>Decay:</b> {@link #fading()} halves all counters periodically to
55	* age out historical data, keeping the TopK set reflective of recent access
56	* patterns.
57	*/
58	@Slf4j	4✔
59	public class HeavyKeeper implements TopK {
60
61	/** Pre-computed decay probability lookup table size ({@value}). */
62	private static final int LOOKUP_TABLE_SIZE = 256;
63	/** Number of lock stripes for fine-grained concurrency ({@value}). Must be a power of two. */
64	private static final int LOCK_STRIPES = 256;
65
66	/**
67	* -- GETTER --
68	* Maximum number of hot keys tracked.
69	*/
70	@Getter
71	private final int k;
72
73	/**
74	* -- GETTER --
75	* Width of the Count-Min Sketch (columns per row).
76	*/
77	@Getter
78	private final int width;
79
80	/**
81	* -- GETTER --
82	* Depth of the Count-Min Sketch (rows / hash functions).
83	*/
84	@Getter
85	private final int depth;
86
87	/** Pre-computed decay probabilities: {@code decay^i} for {@code i in [0, LOOKUP_TABLE_SIZE)}. */
88	private final double[] lookupTable;
89	/** Per-slot fingerprint values for collision verification in the Count-Min Sketch. */
90	private final long[] fingerprints;
91	/** Per-slot frequency counters for the Count-Min Sketch. */
92	private final long[] counts;
93	/** Striped lock objects for fine-grained concurrency on sketch slot updates. */
94	private final Object[] lockStripes;
95	/** Bitmask for mapping bucket index to lock stripe (stripe count must be power of two). */
96	private final int lockMask;
97	/** Sorted map of current TopK entries, ordered by count ascending then key lexicographically.
98	* Boolean.TRUE is a structural placeholder. A TreeSet would not safely remove a Node
99	* whose count had changed because the Comparator includes the count; TreeMap.remove()
100	* uses the comparator (not identity — Node lacks equals/hashCode), but since Node. Count
101	* is final the comparator result for a given Node instance is stable and removal works. */
102	private final TreeMap<Node, Boolean> sortedTopK;
103	/** Reverse index from key to its {@link Node} in the sorted map, for O(1) lookups. */
104	private final Map<String, Node> heapIndex;
105	/** Bounded blocking queue receiving expelled (evicted) key-count items for downstream consumption. */
106	private final BlockingQueue<Item> expelledQueue;
107	/** Running total of all tracked data streams since startup or last {@link #fading()}. */
108	private final LongAdder total;
109
110	/**
111	* -- GETTER --
112	* Minimum count threshold before a key can enter the TopK set.
113	*/
114	@Getter
115	private final int minCount;
116
117	/**
118	* Construct a HeavyKeeper instance.
119	*
120	* @param k maximum number of hot keys to track
121	* @param width width of the Count-Min Sketch (number of columns per row)
122	* @param depth depth of the Count-Min Sketch (number of rows / hash functions)
123	* @param decay probabilistic decay factor (0.0–1.0); higher values preserve counts longer
124	* @param minCount minimum count threshold before a key can enter the TopK set
125	*/
126	public HeavyKeeper(int k, int width, int depth, double decay, int minCount) {
127	this(k, width, depth, decay, minCount, 50_000);	8✔
128	}	1✔
129
130	/**
131	* Construct a HeavyKeeper instance with a custom expelled-queue capacity.
132	*
133	* @param k maximum number of hot keys to track
134	* @param width width of the Count-Min Sketch (number of columns per row)
135	* @param depth depth of the Count-Min Sketch (number of rows / hash functions)
136	* @param decay probabilistic decay factor (0.0–1.0); higher values preserve counts longer
137	* @param minCount minimum count threshold before a key can enter the TopK set
138	* @param expelledQueueCapacity capacity of the bounded blocking queue for expelled items
139	*/
140	public HeavyKeeper(int k, int width, int depth, double decay, int minCount, int expelledQueueCapacity) {	2✔
141	if (k <= 0) {	2✔
142	throw new IllegalArgumentException("TopK must be greater than 0, but got: " + k);	6✔
143	}
144	this.k = k;	3✔
145	this.width = width;	3✔
146	this.depth = depth;	3✔
147	this.minCount = minCount;	3✔
148
149	this.lookupTable = new double[LOOKUP_TABLE_SIZE];	4✔
150	for (int i = 0; i < LOOKUP_TABLE_SIZE; i++) {	7✔
151	lookupTable[i] = Math.pow(decay, i);	8✔
152	}
153
154	int totalSlots = depth * width;	4✔
155	this.fingerprints = new long[totalSlots];	4✔
156	this.counts = new long[totalSlots];	4✔
157	this.lockStripes = new Object[LOCK_STRIPES];	4✔
158	for (int i = 0; i < LOCK_STRIPES; i++) {	7✔
159	lockStripes[i] = new Object();	7✔
160	}
161	this.lockMask = LOCK_STRIPES - 1;	3✔
162
163	this.sortedTopK = new TreeMap<>(Comparator.comparingLong((Node a) -> a.count).thenComparing(a -> a.key));	15✔
164	this.heapIndex = new ConcurrentHashMap<>();	5✔
165	this.expelledQueue = new ArrayBlockingQueue<>(expelledQueueCapacity);	6✔
166	this.total = new LongAdder();	5✔
167	}	1✔
168
169	/**
170	* Record access to the given key.
171	*
172	* <p>The returned {@link AddResult} carries an {@code expelledKey} when the key
173	* entered the TopK set by displacing a previous member (non-null expelledKey and
174	* {@code isHot == true}). When the key was already in the TopK set or failed to
175	* meet the minimum count threshold, {@code expelledKey} is {@code null}.
176	*
177	* @param key the cache key being accessed
178	* @param increment the frequency increment (typically 1)
179	* @return an {@link AddResult} with expelled key (or null), hot status, and the input key
180	*/
181	@Override
182	@SuppressWarnings("null")
183	public AddResult addDirect(String key, int increment) {
184	/* Compute fingerprint with Guava Murmur3_32 (fixed seed ensures same key -> same fingerprint) */
185	long itemFingerprint = Hashing.murmur3_32_fixed().hashString(key, StandardCharsets.UTF_8).padToLong() & 0xFFFFFFFFL;	8✔
186	long maxCount = 0L;	2✔
187
188	for (int i = 0; i < depth; i++) {	8✔
189	int hash = (int) (itemFingerprint ^ (i * 0x9e3779b97f4a7c15L));	8✔
190	int bucketIndex = Math.floorMod(hash, width);	5✔
191	int index = i * width + bucketIndex;	7✔
192	Object lock = lockStripes[index & lockMask];	8✔
193
194	synchronized (lock) {	4✔
195	if (counts[index] == 0) {	7✔
196	fingerprints[index] = itemFingerprint;	5✔
197	counts[index] = increment;	6✔
198
199	maxCount = Math.max(maxCount, increment);	6✔
200	} else if (fingerprints[index] == itemFingerprint) {	7✔
201	counts[index] += increment;	9✔
202
203	maxCount = Math.max(maxCount, counts[index]);	8✔
204	} else {
205	// HeavyKeeper probabilistic decay: sample the decay count from a
206	// Binomial(increment, decayProb) distribution in O(1) instead of
207	// looping increment times. This keeps lock hold time bounded even
208	// when the Worker-side batch increment is large.
209	ThreadLocalRandom rng = ThreadLocalRandom.current();	2✔
210	double decayProb = (counts[index] < LOOKUP_TABLE_SIZE)	7✔
211	? lookupTable[(int) counts[index]]	9✔
212	: lookupTable[LOOKUP_TABLE_SIZE - 1];	5✔
213
214	int decays = sampleBinomial(increment, decayProb, rng);	5✔
215	if (decays >= counts[index]) {	8✔
216	fingerprints[index] = itemFingerprint;	5✔
217	counts[index] = increment;	7✔
218	} else {
219	counts[index] -= decays;	9✔
220	}
221	maxCount = Math.max(maxCount, counts[index]);	7✔
222	}
223	}	3✔
224	}
225
226	total.add(increment);	5✔
227
228	if (maxCount < minCount) {	6✔
229	return AddResult.cold();	2✔
230	}
231
232	synchronized (sortedTopK) {	5✔
233	Node existing = heapIndex.remove(key);	6✔
234	if (existing != null) {	2✔
235	sortedTopK.remove(existing);	5✔
236	}
237
238	boolean isHot = false;	2✔
239	String expelled = null;	2✔
240
241	if (sortedTopK.size() < k \|\| maxCount >= sortedTopK.firstKey().count) {	14✔
242	Node newNode = new Node(key, maxCount);	6✔
243	if (sortedTopK.size() >= k) {	6✔
244	Node removed = sortedTopK.pollFirstEntry().getKey();	6✔
245	expelled = removed.key;	3✔
246	heapIndex.remove(removed.key);	6✔
247	if (!expelledQueue.offer(new Item(expelled, removed.count))) {	10✔
248	log.warn("Expelled queue full, dropping key: {}", expelled);	4✔
249	}
250	}
251	sortedTopK.put(newNode, Boolean.TRUE);	6✔
252	heapIndex.put(key, newNode);	6✔
253	isHot = true;	2✔
254	}
255	return new AddResult(expelled, isHot, key);	9✔
256	}
257	}
258
259	/**
260	* Record accesses for multiple keys.
261	*
262	* <p>Updates sketch counters for all keys, then updates the TopK heap with
263	* keys whose estimated count meets the minimum threshold. Returns results
264	* only for keys that entered the TopK set (possibly displacing others).
265	*
266	* @param keyCounts map of keys to their access counts
267	* @return list of {@link AddResult} for keys that entered the TopK set
268	*/
269	@Override
270	public List<AddResult> addDirect(Map<String, Long> keyCounts) {
271	Map<String, Long> maxCounts = new HashMap<>(keyCounts.size());	6✔
272	for (var entry : keyCounts.entrySet()) {	11✔
273	long max = addToSketch(entry.getKey(), entry.getValue());	10✔
274	maxCounts.put(entry.getKey(), max);	8✔
275	}	1✔
276
277	List<AddResult> results = new ArrayList<>();	4✔
278
279	synchronized (sortedTopK) {	5✔
280	for (var entry : keyCounts.entrySet()) {	11✔
281	String key = entry.getKey();	4✔
282	long maxCount = maxCounts.get(key);	6✔
283	if (maxCount < minCount) {	6✔
284	continue;	1✔
285	}
286
287	Node existing = heapIndex.remove(key);	6✔
288	if (existing != null) {	2✔
289	sortedTopK.remove(existing);	5✔
290	}
291
292	boolean shouldInsert = sortedTopK.size() < k \|\| maxCount >= sortedTopK.firstKey().count;	18✔
293	if (shouldInsert) {	2✔
294	Node newNode = new Node(key, maxCount);	6✔
295	String expelledKey = null;	2✔
296	if (sortedTopK.size() >= k) {	6✔
297	Node removed = sortedTopK.pollFirstEntry().getKey();	6✔
298	expelledKey = removed.key;	3✔
299	heapIndex.remove(removed.key);	6✔
300	if (!expelledQueue.offer(new Item(removed.key, removed.count))) {	11!
301	log.warn("Expelled queue full, dropping key: {}", removed.key);	5✔
302	}
303	}
304	sortedTopK.put(newNode, Boolean.TRUE);	6✔
305	heapIndex.put(key, newNode);	6✔
306	results.add(new AddResult(expelledKey, true, key));	9✔
307	}
308	}	1✔
309	}	3✔
310	return results;	2✔
311	}
312
313	@SuppressWarnings("null")
314	private long addToSketch(String key, long increment) {
315	long itemFingerprint = Hashing.murmur3_32_fixed().hashString(key, StandardCharsets.UTF_8).padToLong() & 0xFFFFFFFFL;	8✔
316	long maxCount = 0;	2✔
317
318	for (int i = 0; i < depth; i++) {	8✔
319	int hash = (int) (itemFingerprint ^ (i * 0x9e3779b97f4a7c15L));	8✔
320	int bucketIndex = Math.floorMod(hash, width);	5✔
321	int index = i * width + bucketIndex;	7✔
322	Object lock = lockStripes[index & lockMask];	8✔
323
324	synchronized (lock) {	4✔
325	if (counts[index] == 0) {	7✔
326	fingerprints[index] = itemFingerprint;	5✔
327	counts[index] = increment;	5✔
328
329	maxCount = Math.max(maxCount, increment);	5✔
330	} else if (fingerprints[index] == itemFingerprint) {	7✔
331	counts[index] += increment;	8✔
332
333	maxCount = Math.max(maxCount, counts[index]);	8✔
334	} else {
335	ThreadLocalRandom rng = ThreadLocalRandom.current();	2✔
336	double decayProb = (counts[index] < LOOKUP_TABLE_SIZE)	7✔
337	? lookupTable[(int) counts[index]]	9✔
338	: lookupTable[LOOKUP_TABLE_SIZE - 1];	5✔
339	int decays = sampleBinomial((int) Math.min(increment, Integer.MAX_VALUE), decayProb, rng);	8✔
340
341	if (decays >= counts[index]) {	8✔
342	fingerprints[index] = itemFingerprint;	5✔
343	counts[index] = increment;	6✔
344	} else {
345	counts[index] -= decays;	9✔
346	}
347	maxCount = Math.max(maxCount, counts[index]);	7✔
348	}
349	}	3✔
350	}
351	total.add(increment);	4✔
352	return maxCount;	2✔
353	}
354
355	/**
356	* Return all keys currently in the TopK set, sorted by estimated count
357	* descending (highest frequency first).
358	*
359	* <p>The returned list is a point-in-time snapshot: it is safe to iterate
360	* after the lock is released but may be stale immediately.
361	*
362	* @return a point-in-time list of {@link Item} entries, ordered from highest
363	* to lowest estimated count; never {@code null}
364	*/
365	@Override
366	public List<Item> list() {
367	synchronized (sortedTopK) {	5✔
368	List<Item> result = new ArrayList<>(sortedTopK.size());	7✔
369	for (Node node : sortedTopK.descendingKeySet()) {	12✔
370	result.add(new Item(node.key, node.count));	10✔
371	}	1✔
372	return result;	4✔
373	}
374	}
375
376	/**
377	* Check whether a key is currently in the TopK set.
378	*
379	* @param key the cache key to test
380	* @return {@code true} if the key is in the TopK set, {@code false} otherwise
381	*/
382	@Override
383	public boolean contains(String key) {
384	return heapIndex.containsKey(key);	5✔
385	}
386
387	/**
388	* Return the blocking queue holding items that have been evicted from the TopK set.
389	* Consumers should drain this queue periodically for asynchronous processing.
390	*
391	* @return a blocking queue of evicted items
392	*/
393	@Override
394	public BlockingQueue<Item> expelled() {
395	return expelledQueue;	3✔
396	}
397
398	/**
399	* Halve all frequency counters in the sketch and the sorted TopK heap.
400	*
401	* <p>Entries whose count drops to zero after halving are removed from
402	* the heap entirely. The running total is also halved. This periodic
403	* decay prevents the sketch from saturating with stale historical data
404	* and ensures that the TopK ranking reflects recent access patterns
405	* rather than cumulative lifetime counts.
406	*
407	* <p>This operation is invoked automatically by a scheduler at a
408	* configured interval (typically 30 seconds). Calling it concurrently
409	* with {@link #addDirect} is safe — sketch counters are halved under
410	* per-stripe locks and the heap is rebuilt atomically under the shared
411	* heap lock.
412	*/
413	@Override
414	public void fading() {
415	// Halve all sketch counters under per-stripe locks to prevent concurrent
416	// addDirect() from observing torn long values (JLS 17.7).
417	// Lock overhead is negligible — called once per decay interval (~30 s).
418	for (int i = 0; i < counts.length; i++) {	9✔
419	synchronized (lockStripes[i & lockMask]) {	10✔
420	counts[i] >>= 1;	8✔
421	}	3✔
422	}
423
424	// Rebuild the TopK set: halve all counts and discard entries that drop to 0.
425	synchronized (sortedTopK) {	5✔
426	TreeMap<Node, Boolean> newMap = new TreeMap<>(sortedTopK.comparator());	7✔
427	heapIndex.clear();	3✔
428	for (Node node : sortedTopK.keySet()) {	12✔
429	long half = node.count >> 1;	5✔
430	if (half > 0) {	4✔
431	Node newNode = new Node(node.key, half);	7✔
432	newMap.put(newNode, Boolean.TRUE);	5✔
433	heapIndex.put(node.key, newNode);	7✔
434	}
435	}	1✔
436	sortedTopK.clear();	3✔
437	sortedTopK.putAll(newMap);	4✔
438
439	long half = total.sumThenReset() >> 1;	6✔
440	if (half > 0) {	4✔
441	total.add(half);	4✔
442	}
443	}	3✔
444	}	1✔
445
446	/**
447	* Return the total number of data streams tracked since startup or last reset.
448	*
449	* @return total access count
450	*/
451	@Override
452	public long total() {
453	return total.sum();	4✔
454	}
455
456	/**
457	* Return the top {@code n} hot keys, ordered by estimated count descending.
458	*
459	* <p>If fewer than {@code n} keys are currently tracked, the returned list
460	* contains all available keys. The result is a point-in-time snapshot.
461	*
462	* @param n maximum number of keys to return (must be non-negative)
463	* @return list of at most {@code n} {@link Item} entries, never {@code null};
464	* empty if no keys are tracked or {@code n == 0}
465	*/
466	@Override
467	public List<Item> listTopN(int n) {
468	synchronized (sortedTopK) {	5✔
469	return sortedTopK	3✔
470	.descendingKeySet()	1✔
471	.stream()	3✔
472	.limit(n)	2✔
473	.map(node -> new Item(node.key, node.count))	9✔
474	.collect(Collectors.toList());	5✔
475	}
476	}
477
478	/** A key-count pair used as an entry in the sorted TopK tree. */
479	@AllArgsConstructor
480	private static class Node {
481
482	/** The cache key. */
483	final String key;
484	/** Its current estimated count. */
485	final long count;
486	}
487
488	/**
489	* Sample from a Binomial(n, p) distribution in O(1) expected time.
490	*
491	* <p>Uses direct simulation for small n (≤10) and normal approximation
492	* for larger n when np(1-p) > 9. Falls back to direct simulation for
493	* moderate n.
494	*/
495	private static int sampleBinomial(int n, double p, ThreadLocalRandom rng) {
496	if (n <= 0) {	2✔
497	return 0;	2✔
498	}
499	if (p >= 1.0) {	4✔
500	return n;	2✔
501	}
502	if (p <= 0.0) {	4✔
503	return 0;	2✔
504	}
505
506	double q = 1.0 - p;	4✔
507
508	if (n <= 10) {	3✔
509	int k = 0;	2✔
510	for (int i = 0; i < n; i++) {	7✔
511	if (rng.nextDouble() < p) {	5!
512	k++;	×
513	}
514	}
515	return k;	2✔
516	}
517
518	double np = n * p;	5✔
519	double npq = np * q;	4✔
520
521	if (npq > 9.0) {	4✔
522	int k = (int) Math.round(np + Math.sqrt(npq) * rng.nextGaussian());	10✔
523	return Math.max(0, Math.min(n, k));	6✔
524	}
525
526	int k = 0;	2✔
527	for (int i = 0; i < n; i++) {	7✔
528	if (rng.nextDouble() < p) {	5✔
529	k++;	1✔
530	}
531	}
532	return k;	2✔
533	}
534	}

Hyshmily / hotkey / 28160818793

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous