27925388821

Committed 22 Jun 2026 02:14AM UTC coverage: 92.403% (-1.2%) from 93.615%

Build # 27925388821

Build Type

push

github

Committed by

Hyshmily

Commit Message

fix:fix known bugs and promote performance

Signed-off-by: Hyshmily <cxm8607@outlook.com>

Coverage Stats

1143 of 1289 branches covered (88.67%)

Branch coverage included in aggregate %.

184 of 236 new or added lines in 19 files covered. (77.97%)

6 existing lines in 3 files now uncovered.

3321 of 3542 relevant lines covered (93.76%)

4.22 hits per line

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

93.51

/common/src/main/java/io/github/hyshmily/hotkey/hotkeydetector/doublebuffer/BufferedCounter.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.doublebuffer;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.Consumer;
import javax.security.auth.Destroyable;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;

/**
 * Double-buffered counter that aggregates high-frequency single-key increments
 * and flushes them in batch to a downstream consumer.
 *
 * <p><b>Design:</b> Two {@link CounterBuffer} instances (active and standby)
 * are maintained as {@code AtomicReference<CounterBuffer>}. The active buffer
 * accepts incoming {@link #count(String, long)} calls via a lock-free
 * {@code ConcurrentHashMap}. A scheduled flusher periodically swaps and drains
 * both buffers into the downstream consumer (see {@link #flushStandby()}).
 *
 * <p><b>Eager swap:</b> When the active buffer exceeds 80 % of
 * {@link #MAX_BUFFER_SIZE}, the buffers are swapped eagerly to prevent any
 * single buffer from growing unbounded under a traffic spike. The hot path
 * (the {@code count} call) remains lock-free — it only does an atomic
 * {@code getAndSet} on the active reference.
 *
 * <p><b>Lifecycle:</b> Implements {@link InitializingBean} to start the
 * periodic flush scheduler, and {@link Destroyable} to perform a final drain
 * on shutdown. The scheduler is either self-created (and owned) or externally
 * provided (shared), controlling whether {@link #destroy()} shuts it down.
 *
 * <p>Thread-safe. All public methods can be called concurrently from
 * multiple threads.
 */
@Slf4j
public class BufferedCounter implements InitializingBean, Destroyable {

  /** Maximum number of distinct keys held in one buffer before forced swap ({@value}). */
  private static final int MAX_BUFFER_SIZE = 10_000;

  /** Fixed flush interval in milliseconds ({@value}). */
  private static final long FLUSH_INTERVAL_MS = 500;

  private final AtomicReference<CounterBuffer> active;

  private final AtomicReference<CounterBuffer> standbyRef;

  private final Consumer<Map<String, Long>> batchConsumer;

  private final ScheduledExecutorService scheduler;

  private final boolean ownsScheduler;

  /**
   * Creates a buffered counter that flushes aggregated counts to the given consumer.
   * Creates its own single-thread scheduler (backward-compatible).
   *
   * @param batchConsumer callback receiving the aggregated key-count map on each flush
   */
  public BufferedCounter(Consumer<Map<String, Long>> batchConsumer) {
    this(
      batchConsumer,
      Executors.newSingleThreadScheduledExecutor(r -> new Thread(r, "buffered-counter-flusher")),
      true
    );
  }

  /**
   * Creates a buffered counter with an externally provided scheduler.
   *
   * @param batchConsumer callback receiving the aggregated key-count map on each flush
   * @param scheduler     the shared scheduler (not shut down on destroy)
   */
  public BufferedCounter(Consumer<Map<String, Long>> batchConsumer, ScheduledExecutorService scheduler) {
    this(batchConsumer, scheduler, false);
  }

  private BufferedCounter(
    Consumer<Map<String, Long>> batchConsumer,
    ScheduledExecutorService scheduler,
    boolean ownsScheduler
  ) {
    this.batchConsumer = batchConsumer;
    this.active = new AtomicReference<>(new CounterBuffer());
    this.standbyRef = new AtomicReference<>(new CounterBuffer());
    this.scheduler = scheduler;
    this.ownsScheduler = ownsScheduler;
  }

  /**
   * Record one or more accesses for the given key into the active buffer.
   *
   * <p>This is the hot path — it performs a lock-free update on the active
   * {@link CounterBuffer}. If the active buffer exceeds 80 % capacity
   * after this increment, an eager swap is triggered to keep the buffer
   * from overflowing before the next scheduled flush.
   *
   * @param key   the accessed key (must not be {@code null})
   * @param delta the number of accesses to record (must be positive)
   */
  public void count(String key, long delta) {
    CounterBuffer buffer = active.get();
    buffer.add(key, delta);

    if (buffer.size() >= MAX_BUFFER_SIZE * 0.8) {
      trySwitch();
    }
  }

  private void trySwitch() {
    CounterBuffer newBuffer = new CounterBuffer();
    CounterBuffer oldBuffer = active.getAndSet(newBuffer);
    if (oldBuffer != null) {
      standbyRef.set(oldBuffer);
    }
  }

  private void flushStandby() {
    try {
      // Swap active and standby independently to avoid races between
      // count()->trySwitch() and the scheduled flush.
      CounterBuffer flushedActive = active.getAndSet(new CounterBuffer());
      CounterBuffer flushedStandby = standbyRef.getAndSet(new CounterBuffer());

      drainBuffer(flushedActive);
      drainBuffer(flushedStandby);
      log.trace("BufferedCounter flush tick: active={}, standby={}", flushedActive.size(), flushedStandby.size());
    } catch (Exception e) {
      log.error("Scheduled flushStandby failed", e);
    }
  }

  private void drainBuffer(CounterBuffer buf) {
    if (!buf.isEmpty()) {
      Map<String, Long> snapshot = buf.drain();
      if (!snapshot.isEmpty()) {
        batchConsumer.accept(snapshot);
      }
    }
  }

  /**
   * Start the periodic flush scheduler.  Called by the Spring container
   * after all bean properties have been set.
   */
  @Override
  public void afterPropertiesSet() {
    try {
      scheduler.scheduleAtFixedRate(this::flushStandby, FLUSH_INTERVAL_MS, FLUSH_INTERVAL_MS, TimeUnit.MILLISECONDS);
    } catch (Exception e) {
      log.error(
        "Failed to start BufferedCounter flush scheduler; buffered counts will not " +
          "be flushed to HeavyKeeper. Hot-key detection may be impaired.",
        e
      );
    }
  }

  /**
   * Perform a final drain of any remaining buffered counts and shut down
   * the scheduler only if owned (self-created). For a shared scheduler
   * the task is simply cancelled.
   *
   * <p>Called by the Spring container during context close.
   */
  @Override
  public void destroy() {
    if (ownsScheduler) {
      scheduler.shutdown();
    }
    // Swap any remaining active data into standby and flush both
    trySwitch();
    flushStandby();
  }

  private static class CounterBuffer {

    private final ConcurrentHashMap<String, LongAdder> counters = new ConcurrentHashMap<>();
    private final LongAdder totalSize = new LongAdder();

    /**
     * Record one or more accesses for the given key in this buffer.
     *
     * @param key   the accessed key
     * @param delta the number of accesses to record
     */
    void add(String key, long delta) {
      counters.computeIfAbsent(key, k -> new LongAdder()).add(delta);
      totalSize.add(delta);
    }

    /**
     * Return the number of distinct keys held in this buffer.
     *
     * @return the number of distinct keys
     */
    int size() {
      return counters.size();
    }

    /**
     * Return whether this buffer holds no entries.
     *
     * @return {@code true} if the buffer is empty
     */
    boolean isEmpty() {
      return size() == 0;
    }

    /**
     * Atomically drain all counters and return a snapshot of the accumulated
     * counts. After this call the buffer is empty and ready for reuse.
     *
     * @return a map of keys to their accumulated counts, never {@code null}
     */
    Map<String, Long> drain() {
      Map<String, LongAdder> oldCounters = counters;

      Map<String, Long> result = new HashMap<>(oldCounters.size());
      oldCounters.forEach((key, adder) -> {
        long val = adder.sumThenReset();
        if (val > 0) {
          result.put(key, val);
        }
      });
      return result;
    }
  }
}

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.doublebuffer;
17
18	import java.util.HashMap;
19	import java.util.Map;
20	import java.util.concurrent.ConcurrentHashMap;
21	import java.util.concurrent.Executors;
22	import java.util.concurrent.ScheduledExecutorService;
23	import java.util.concurrent.TimeUnit;
24	import java.util.concurrent.atomic.AtomicReference;
25	import java.util.concurrent.atomic.LongAdder;
26	import java.util.function.Consumer;
27	import javax.security.auth.Destroyable;
28	import lombok.extern.slf4j.Slf4j;
29	import org.springframework.beans.factory.InitializingBean;
30
31	/**
32	* Double-buffered counter that aggregates high-frequency single-key increments
33	* and flushes them in batch to a downstream consumer.
34	*
35	* <p><b>Design:</b> Two {@link CounterBuffer} instances (active and standby)
36	* are maintained as {@code AtomicReference<CounterBuffer>}. The active buffer
37	* accepts incoming {@link #count(String, long)} calls via a lock-free
38	* {@code ConcurrentHashMap}. A scheduled flusher periodically swaps and drains
39	* both buffers into the downstream consumer (see {@link #flushStandby()}).
40	*
41	* <p><b>Eager swap:</b> When the active buffer exceeds 80 % of
42	* {@link #MAX_BUFFER_SIZE}, the buffers are swapped eagerly to prevent any
43	* single buffer from growing unbounded under a traffic spike. The hot path
44	* (the {@code count} call) remains lock-free — it only does an atomic
45	* {@code getAndSet} on the active reference.
46	*
47	* <p><b>Lifecycle:</b> Implements {@link InitializingBean} to start the
48	* periodic flush scheduler, and {@link Destroyable} to perform a final drain
49	* on shutdown. The scheduler is either self-created (and owned) or externally
50	* provided (shared), controlling whether {@link #destroy()} shuts it down.
51	*
52	* <p>Thread-safe. All public methods can be called concurrently from
53	* multiple threads.
54	*/
55	@Slf4j	4✔
56	public class BufferedCounter implements InitializingBean, Destroyable {
57
58	/** Maximum number of distinct keys held in one buffer before forced swap ({@value}). */
59	private static final int MAX_BUFFER_SIZE = 10_000;
60
61	/** Fixed flush interval in milliseconds ({@value}). */
62	private static final long FLUSH_INTERVAL_MS = 500;
63
64	private final AtomicReference<CounterBuffer> active;
65
66	private final AtomicReference<CounterBuffer> standbyRef;
67
68	private final Consumer<Map<String, Long>> batchConsumer;
69
70	private final ScheduledExecutorService scheduler;
71
72	private final boolean ownsScheduler;
73
74	/**
75	* Creates a buffered counter that flushes aggregated counts to the given consumer.
76	* Creates its own single-thread scheduler (backward-compatible).
77	*
78	* @param batchConsumer callback receiving the aggregated key-count map on each flush
79	*/
80	public BufferedCounter(Consumer<Map<String, Long>> batchConsumer) {
81	this(	4✔
82	batchConsumer,
83	Executors.newSingleThreadScheduledExecutor(r -> new Thread(r, "buffered-counter-flusher")),	8✔
84	true
85	);
86	}	1✔
87
88	/**
89	* Creates a buffered counter with an externally provided scheduler.
90	*
91	* @param batchConsumer callback receiving the aggregated key-count map on each flush
92	* @param scheduler the shared scheduler (not shut down on destroy)
93	*/
94	public BufferedCounter(Consumer<Map<String, Long>> batchConsumer, ScheduledExecutorService scheduler) {
95	this(batchConsumer, scheduler, false);	5✔
96	}	1✔
97
98	private BufferedCounter(
99	Consumer<Map<String, Long>> batchConsumer,
100	ScheduledExecutorService scheduler,
101	boolean ownsScheduler
102	) {	2✔
103	this.batchConsumer = batchConsumer;	3✔
104	this.active = new AtomicReference<>(new CounterBuffer());	8✔
105	this.standbyRef = new AtomicReference<>(new CounterBuffer());	8✔
106	this.scheduler = scheduler;	3✔
107	this.ownsScheduler = ownsScheduler;	3✔
108	}	1✔
109
110	/**
111	* Record one or more accesses for the given key into the active buffer.
112	*
113	* <p>This is the hot path — it performs a lock-free update on the active
114	* {@link CounterBuffer}. If the active buffer exceeds 80 % capacity
115	* after this increment, an eager swap is triggered to keep the buffer
116	* from overflowing before the next scheduled flush.
117	*
118	* @param key the accessed key (must not be {@code null})
119	* @param delta the number of accesses to record (must be positive)
120	*/
121	public void count(String key, long delta) {
122	CounterBuffer buffer = active.get();	5✔
123	buffer.add(key, delta);	4✔
124
125	if (buffer.size() >= MAX_BUFFER_SIZE * 0.8) {	6!
126	trySwitch();	×
127	}
128	}	1✔
129
130	private void trySwitch() {
131	CounterBuffer newBuffer = new CounterBuffer();	4✔
132	CounterBuffer oldBuffer = active.getAndSet(newBuffer);	6✔
133	if (oldBuffer != null) {	2!
134	standbyRef.set(oldBuffer);	4✔
135	}
136	}	1✔
137
138	private void flushStandby() {
139	try {
140	// Swap active and standby independently to avoid races between
141	// count()->trySwitch() and the scheduled flush.
142	CounterBuffer flushedActive = active.getAndSet(new CounterBuffer());	8✔
143	CounterBuffer flushedStandby = standbyRef.getAndSet(new CounterBuffer());	8✔
144
145	drainBuffer(flushedActive);	3✔
146	drainBuffer(flushedStandby);	3✔
147	log.trace("BufferedCounter flush tick: active={}, standby={}", flushedActive.size(), flushedStandby.size());	9✔
148	} catch (Exception e) {	1✔
149	log.error("Scheduled flushStandby failed", e);	4✔
150	}	1✔
151	}	1✔
152
153	private void drainBuffer(CounterBuffer buf) {
154	if (!buf.isEmpty()) {	3✔
155	Map<String, Long> snapshot = buf.drain();	3✔
156	if (!snapshot.isEmpty()) {	3✔
157	batchConsumer.accept(snapshot);	4✔
158	}
159	}
160	}	1✔
161
162	/**
163	* Start the periodic flush scheduler. Called by the Spring container
164	* after all bean properties have been set.
165	*/
166	@Override
167	public void afterPropertiesSet() {
168	try {
169	scheduler.scheduleAtFixedRate(this::flushStandby, FLUSH_INTERVAL_MS, FLUSH_INTERVAL_MS, TimeUnit.MILLISECONDS);	9✔
NEW 170	} catch (Exception e) {	×
NEW 171	log.error(	×
172	"Failed to start BufferedCounter flush scheduler; buffered counts will not " +
173	"be flushed to HeavyKeeper. Hot-key detection may be impaired.",
174	e
175	);
176	}	1✔
177	}	1✔
178
179	/**
180	* Perform a final drain of any remaining buffered counts and shut down
181	* the scheduler only if owned (self-created). For a shared scheduler
182	* the task is simply cancelled.
183	*
184	* <p>Called by the Spring container during context close.
185	*/
186	@Override
187	public void destroy() {
188	if (ownsScheduler) {	3✔
189	scheduler.shutdown();	3✔
190	}
191	// Swap any remaining active data into standby and flush both
192	trySwitch();	2✔
193	flushStandby();	2✔
194	}	1✔
195
196	private static class CounterBuffer {	2✔
197
198	private final ConcurrentHashMap<String, LongAdder> counters = new ConcurrentHashMap<>();	5✔
199	private final LongAdder totalSize = new LongAdder();	6✔
200
201	/**
202	* Record one or more accesses for the given key in this buffer.
203	*
204	* @param key the accessed key
205	* @param delta the number of accesses to record
206	*/
207	void add(String key, long delta) {
208	counters.computeIfAbsent(key, k -> new LongAdder()).add(delta);	12✔
209	totalSize.add(delta);	4✔
210	}	1✔
211
212	/**
213	* Return the number of distinct keys held in this buffer.
214	*
215	* @return the number of distinct keys
216	*/
217	int size() {
218	return counters.size();	4✔
219	}
220
221	/**
222	* Return whether this buffer holds no entries.
223	*
224	* @return {@code true} if the buffer is empty
225	*/
226	boolean isEmpty() {
227	return size() == 0;	7✔
228	}
229
230	/**
231	* Atomically drain all counters and return a snapshot of the accumulated
232	* counts. After this call the buffer is empty and ready for reuse.
233	*
234	* @return a map of keys to their accumulated counts, never {@code null}
235	*/
236	Map<String, Long> drain() {
237	Map<String, LongAdder> oldCounters = counters;	3✔
238
239	Map<String, Long> result = new HashMap<>(oldCounters.size());	6✔
240	oldCounters.forEach((key, adder) -> {	4✔
241	long val = adder.sumThenReset();	3✔
242	if (val > 0) {	4✔
243	result.put(key, val);	6✔
244	}
245	});	1✔
246	return result;	2✔
247	}
248	}
249	}

Hyshmily / hotkey / 27925388821

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