Skip to content

Commit fb43b86

Browse files
refactor: replace ScheduledExecutorTimer with in-process HashedWheelTimer
Drops the JDK ScheduledExecutorService backing in favor of a stripped-down in-process hashed-wheel implementation: single tick thread, 512x10ms wheel (~5s rotation), MPSC pending queue, O(1) insert, O(1) cancel via state CAS with lazy reaping on the bucket's next sweep. Same algorithm and tick geometry as the shaded-netty HashedWheelTimer we used pre-phase-7, but with no Netty dependency. BigtableTimer interface and the per-call Executor contract from c29e are preserved unchanged; only the impl swaps. Benchmark (jetstream100, c100, n=3 medians): HashedWheelTimer: 111,557 ops/s p99 1,527us p9999 94ms ScheduledExecutorTimer: 106,868 ops/s p99 1,592us p9999 174ms Wheel recovers the ~4% throughput regression introduced in c29e and roughly halves the p9999 deep tail.
1 parent 7d54285 commit fb43b86

8 files changed

Lines changed: 284 additions & 130 deletions

File tree

‎java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/api/Client.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@
3434
import com.google.cloud.bigtable.data.v2.internal.csm.NoopMetrics;
3535
import com.google.cloud.bigtable.data.v2.internal.csm.attributes.ClientInfo;
3636
import com.google.cloud.bigtable.data.v2.internal.session.BigtableTimer;
37-
import com.google.cloud.bigtable.data.v2.internal.session.ScheduledExecutorTimer;
37+
import com.google.cloud.bigtable.data.v2.internal.session.HashedWheelTimer;
3838
import com.google.cloud.bigtable.data.v2.internal.session.SessionPool;
3939
import com.google.cloud.bigtable.data.v2.internal.util.ClientConfigurationManager;
4040
import com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -205,7 +205,7 @@ public Client(
205205
this.configManager = configManager;
206206
this.backgroundExecutor = bgExecutor;
207207
this.userCallbackExecutor = userCallbackExecutor;
208-
this.sessionTimer = new ScheduledExecutorTimer("bigtable-session-timer");
208+
this.sessionTimer = new HashedWheelTimer("bigtable-session-timer");
209209

210210
defaultCallOptions = CallOptions.DEFAULT;
211211

Lines changed: 268 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,268 @@
1+
/*
2+
* Copyright 2026 Google LLC
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+
* https://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 com.google.cloud.bigtable.data.v2.internal.session;
17+
18+
import com.google.common.util.concurrent.ThreadFactoryBuilder;
19+
import java.util.HashSet;
20+
import java.util.Queue;
21+
import java.util.Set;
22+
import java.util.concurrent.ConcurrentHashMap;
23+
import java.util.concurrent.ConcurrentLinkedQueue;
24+
import java.util.concurrent.Executor;
25+
import java.util.concurrent.TimeUnit;
26+
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
27+
import java.util.logging.Level;
28+
import java.util.logging.Logger;
29+
30+
/**
31+
* {@link BigtableTimer} backed by a single-thread hashed wheel — same algorithm as Netty's
32+
* {@code HashedWheelTimer} but with no Netty dependency. Tick = {@value #TICK_DURATION_MS} ms,
33+
* {@value #TICKS_PER_WHEEL} buckets, so one full rotation is ~5 s.
34+
*
35+
* <p>Insert is O(1): the new {@link Timeout} is enqueued onto an MPSC pending queue and bucketed
36+
* by the tick thread on the next tick. Cancel is O(1): a single state CAS marks the entry
37+
* cancelled, and the worker unlinks it the next time the bucket is swept (so a cancelled entry
38+
* holds memory for at most one wheel rotation).
39+
*
40+
* <p>The worker thread only advances the wheel and hops fires to the caller-supplied
41+
* {@link Executor}; user code never runs on the tick thread.
42+
*/
43+
public final class HashedWheelTimer implements BigtableTimer {
44+
private static final Logger LOG = Logger.getLogger(HashedWheelTimer.class.getName());
45+
46+
static final long TICK_DURATION_MS = 10;
47+
static final int TICKS_PER_WHEEL = 512; // must be a power of two
48+
private static final long TICK_DURATION_NANOS = TimeUnit.MILLISECONDS.toNanos(TICK_DURATION_MS);
49+
private static final int MASK = TICKS_PER_WHEEL - 1;
50+
// Bound transfers per tick so a sustained insert flood can't starve bucket processing.
51+
private static final int MAX_PENDING_TRANSFERS_PER_TICK = 100_000;
52+
53+
private final HashedWheelBucket[] wheel = new HashedWheelBucket[TICKS_PER_WHEEL];
54+
private final Queue<HashedWheelTimeout> pendingTimeouts = new ConcurrentLinkedQueue<>();
55+
private final Set<Runnable> stopHooks = ConcurrentHashMap.newKeySet();
56+
57+
private final Thread worker;
58+
private final long startNanos;
59+
60+
private volatile boolean stopped = false;
61+
62+
public HashedWheelTimer(String name) {
63+
for (int i = 0; i < TICKS_PER_WHEEL; i++) {
64+
wheel[i] = new HashedWheelBucket();
65+
}
66+
this.startNanos = System.nanoTime();
67+
this.worker =
68+
new ThreadFactoryBuilder()
69+
.setNameFormat(name + "-%d")
70+
.setDaemon(true)
71+
.build()
72+
.newThread(this::workerLoop);
73+
this.worker.start();
74+
}
75+
76+
@Override
77+
public Timeout newTimeout(Runnable task, Executor executor, long delay, TimeUnit unit) {
78+
if (stopped) {
79+
throw new IllegalStateException("timer stopped");
80+
}
81+
long delayNanos = Math.max(0L, unit.toNanos(delay));
82+
long deadlineNanos = (System.nanoTime() - startNanos) + delayNanos;
83+
HashedWheelTimeout timeout = new HashedWheelTimeout(task, executor, deadlineNanos);
84+
pendingTimeouts.add(timeout);
85+
return timeout;
86+
}
87+
88+
@Override
89+
public Registration onStop(Runnable hook) {
90+
if (stopped) {
91+
throw new IllegalStateException("timer stopped");
92+
}
93+
stopHooks.add(hook);
94+
return () -> stopHooks.remove(hook);
95+
}
96+
97+
@Override
98+
public void stop() {
99+
if (stopped) {
100+
return;
101+
}
102+
stopped = true;
103+
Set<Runnable> hooks = new HashSet<>(stopHooks);
104+
stopHooks.clear();
105+
for (Runnable hook : hooks) {
106+
try {
107+
hook.run();
108+
} catch (Throwable t) {
109+
LOG.log(Level.WARNING, "stop hook threw; continuing", t);
110+
}
111+
}
112+
worker.interrupt();
113+
}
114+
115+
private void workerLoop() {
116+
long tick = 0;
117+
while (!stopped) {
118+
if (waitForNextTick(tick) < 0) {
119+
return;
120+
}
121+
transferPending(tick);
122+
wheel[(int) (tick & MASK)].expireTimeouts();
123+
tick++;
124+
}
125+
}
126+
127+
// Returns elapsed nanos since startNanos at wake, or -1 if the timer was stopped.
128+
private long waitForNextTick(long tick) {
129+
long targetNanos = (tick + 1) * TICK_DURATION_NANOS;
130+
while (true) {
131+
long elapsed = System.nanoTime() - startNanos;
132+
long sleepNanos = targetNanos - elapsed;
133+
if (sleepNanos <= 0) {
134+
return elapsed;
135+
}
136+
if (stopped) {
137+
return -1;
138+
}
139+
long sleepMs = (sleepNanos + 999_999L) / 1_000_000L;
140+
try {
141+
Thread.sleep(sleepMs);
142+
} catch (InterruptedException e) {
143+
if (stopped) {
144+
return -1;
145+
}
146+
// Spurious interrupt; recompute and keep sleeping.
147+
}
148+
}
149+
}
150+
151+
private void transferPending(long tick) {
152+
for (int i = 0; i < MAX_PENDING_TRANSFERS_PER_TICK; i++) {
153+
HashedWheelTimeout t = pendingTimeouts.poll();
154+
if (t == null) {
155+
return;
156+
}
157+
if (t.isCancelled()) {
158+
continue;
159+
}
160+
long calculated = t.deadlineNanos / TICK_DURATION_NANOS;
161+
t.remainingRounds = Math.max(0L, (calculated - tick) / TICKS_PER_WHEEL);
162+
// If the deadline is already past, place the entry in the current bucket so it fires now.
163+
long ticksTarget = Math.max(calculated, tick);
164+
wheel[(int) (ticksTarget & MASK)].add(t);
165+
}
166+
}
167+
168+
// Bucket: singly-traversed doubly-linked list, mutated only on the worker thread.
169+
private static final class HashedWheelBucket {
170+
private HashedWheelTimeout head;
171+
private HashedWheelTimeout tail;
172+
173+
void add(HashedWheelTimeout t) {
174+
t.bucket = this;
175+
if (head == null) {
176+
head = tail = t;
177+
} else {
178+
tail.next = t;
179+
t.prev = tail;
180+
tail = t;
181+
}
182+
}
183+
184+
void expireTimeouts() {
185+
HashedWheelTimeout t = head;
186+
while (t != null) {
187+
HashedWheelTimeout next = t.next;
188+
if (t.isCancelled()) {
189+
unlink(t);
190+
} else if (t.remainingRounds <= 0) {
191+
unlink(t);
192+
t.expire();
193+
} else {
194+
t.remainingRounds--;
195+
}
196+
t = next;
197+
}
198+
}
199+
200+
private void unlink(HashedWheelTimeout t) {
201+
HashedWheelTimeout p = t.prev;
202+
HashedWheelTimeout n = t.next;
203+
if (p != null) {
204+
p.next = n;
205+
}
206+
if (n != null) {
207+
n.prev = p;
208+
}
209+
if (t == head) {
210+
head = n;
211+
}
212+
if (t == tail) {
213+
tail = p;
214+
}
215+
t.prev = null;
216+
t.next = null;
217+
t.bucket = null;
218+
}
219+
}
220+
221+
private static final class HashedWheelTimeout implements Timeout {
222+
private static final int STATE_INIT = 0;
223+
private static final int STATE_CANCELLED = 1;
224+
private static final int STATE_EXPIRED = 2;
225+
226+
private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATER =
227+
AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state");
228+
229+
private final Runnable task;
230+
private final Executor executor;
231+
final long deadlineNanos;
232+
233+
volatile int state = STATE_INIT;
234+
// Worker-thread only:
235+
long remainingRounds;
236+
HashedWheelTimeout next;
237+
HashedWheelTimeout prev;
238+
HashedWheelBucket bucket;
239+
240+
HashedWheelTimeout(Runnable task, Executor executor, long deadlineNanos) {
241+
this.task = task;
242+
this.executor = executor;
243+
this.deadlineNanos = deadlineNanos;
244+
}
245+
246+
@Override
247+
public boolean cancel() {
248+
return STATE_UPDATER.compareAndSet(this, STATE_INIT, STATE_CANCELLED);
249+
}
250+
251+
@Override
252+
public boolean isCancelled() {
253+
return state == STATE_CANCELLED;
254+
}
255+
256+
void expire() {
257+
if (!STATE_UPDATER.compareAndSet(this, STATE_INIT, STATE_EXPIRED)) {
258+
// Lost a race with cancel() between transferPending and this sweep.
259+
return;
260+
}
261+
try {
262+
executor.execute(task);
263+
} catch (Throwable t) {
264+
LOG.log(Level.WARNING, "executor.execute threw; dropping task", t);
265+
}
266+
}
267+
}
268+
}

0 commit comments

Comments
 (0)