1 /*
2 * Copyright 2012 The Netty Project
3 *
4 * The Netty Project licenses this file to you under the Apache License,
5 * version 2.0 (the "License"); you may not use this file except in compliance
6 * with the License. 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, WITHOUT
12 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 * License for the specific language governing permissions and limitations
14 * under the License.
15 */
16 package io.netty.util;
17
18 import static io.netty.util.internal.ObjectUtil.checkPositive;
19 import static io.netty.util.internal.ObjectUtil.checkNotNull;
20
21 import io.netty.util.concurrent.ImmediateExecutor;
22 import io.netty.util.internal.MathUtil;
23 import io.netty.util.internal.PlatformDependent;
24 import io.netty.util.internal.logging.InternalLogger;
25 import io.netty.util.internal.logging.InternalLoggerFactory;
26
27 import java.util.Collections;
28 import java.util.HashSet;
29 import java.util.Queue;
30 import java.util.Set;
31 import java.util.concurrent.CountDownLatch;
32 import java.util.concurrent.Executor;
33 import java.util.concurrent.Executors;
34 import java.util.concurrent.RejectedExecutionException;
35 import java.util.concurrent.ThreadFactory;
36 import java.util.concurrent.TimeUnit;
37 import java.util.concurrent.atomic.AtomicBoolean;
38 import java.util.concurrent.atomic.AtomicInteger;
39 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
40 import java.util.concurrent.atomic.AtomicLong;
41
42 import static io.netty.util.internal.StringUtil.simpleClassName;
43
44 /**
45 * A {@link Timer} optimized for approximated I/O timeout scheduling.
46 *
47 * <h3>Tick Duration</h3>
48 *
49 * As described with 'approximated', this timer does not execute the scheduled
50 * {@link TimerTask} on time. {@link HashedWheelTimer}, on every tick, will
51 * check if there are any {@link TimerTask}s behind the schedule and execute
52 * them.
53 * <p>
54 * You can increase or decrease the accuracy of the execution timing by
55 * specifying smaller or larger tick duration in the constructor. In most
56 * network applications, I/O timeout does not need to be accurate. Therefore,
57 * the default tick duration is 100 milliseconds and you will not need to try
58 * different configurations in most cases.
59 *
60 * <h3>Ticks per Wheel (Wheel Size)</h3>
61 *
62 * {@link HashedWheelTimer} maintains a data structure called 'wheel'.
63 * To put simply, a wheel is a hash table of {@link TimerTask}s whose hash
64 * function is 'dead line of the task'. The default number of ticks per wheel
65 * (i.e. the size of the wheel) is 512. You could specify a larger value
66 * if you are going to schedule a lot of timeouts.
67 *
68 * <h3>Do not create many instances.</h3>
69 *
70 * {@link HashedWheelTimer} creates a new thread whenever it is instantiated and
71 * started. Therefore, you should make sure to create only one instance and
72 * share it across your application. One of the common mistakes, that makes
73 * your application unresponsive, is to create a new instance for every connection.
74 *
75 * <h3>Implementation Details</h3>
76 *
77 * {@link HashedWheelTimer} is based on
78 * <a href="https://cseweb.ucsd.edu/users/varghese/">George Varghese</a> and
79 * Tony Lauck's paper,
80 * <a href="https://cseweb.ucsd.edu/users/varghese/PAPERS/twheel.ps.Z">'Hashed
81 * and Hierarchical Timing Wheels: data structures to efficiently implement a
82 * timer facility'</a>. More comprehensive slides are located
83 * <a href="https://www.cse.wustl.edu/~cdgill/courses/cs6874/TimingWheels.ppt">here</a>.
84 */
85 public class HashedWheelTimer implements Timer {
86
87 static final InternalLogger logger =
88 InternalLoggerFactory.getInstance(HashedWheelTimer.class);
89
90 private static final AtomicInteger INSTANCE_COUNTER = new AtomicInteger();
91 private static final AtomicBoolean WARNED_TOO_MANY_INSTANCES = new AtomicBoolean();
92 private static final int INSTANCE_COUNT_LIMIT = 64;
93 private static final long MILLISECOND_NANOS = TimeUnit.MILLISECONDS.toNanos(1);
94 private static final ResourceLeakDetector<HashedWheelTimer> leakDetector = ResourceLeakDetectorFactory.instance()
95 .newResourceLeakDetector(HashedWheelTimer.class, 1);
96
97 private static final AtomicIntegerFieldUpdater<HashedWheelTimer> WORKER_STATE_UPDATER =
98 AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimer.class, "workerState");
99
100 private final ResourceLeakTracker<HashedWheelTimer> leak;
101 private final Worker worker = new Worker();
102 private final Thread workerThread;
103
104 public static final int WORKER_STATE_INIT = 0;
105 public static final int WORKER_STATE_STARTED = 1;
106 public static final int WORKER_STATE_SHUTDOWN = 2;
107 @SuppressWarnings({"unused", "FieldMayBeFinal"})
108 private volatile int workerState; // 0 - init, 1 - started, 2 - shut down
109
110 private final long tickDuration;
111 private final HashedWheelBucket[] wheel;
112 private final int mask;
113 private final CountDownLatch startTimeInitialized = new CountDownLatch(1);
114 private final Queue<HashedWheelTimeout> timeouts = PlatformDependent.newMpscQueue();
115 private final Queue<HashedWheelTimeout> cancelledTimeouts = PlatformDependent.newMpscQueue();
116 private final AtomicLong pendingTimeouts = new AtomicLong(0);
117 private final long maxPendingTimeouts;
118 private final Executor taskExecutor;
119
120 private volatile long startTime;
121
122 /**
123 * Creates a new timer with the default thread factory
124 * ({@link Executors#defaultThreadFactory()}), default tick duration, and
125 * default number of ticks per wheel.
126 */
127 public HashedWheelTimer() {
128 this(Executors.defaultThreadFactory());
129 }
130
131 /**
132 * Creates a new timer with the default thread factory
133 * ({@link Executors#defaultThreadFactory()}) and default number of ticks
134 * per wheel.
135 *
136 * @param tickDuration the duration between tick
137 * @param unit the time unit of the {@code tickDuration}
138 * @throws NullPointerException if {@code unit} is {@code null}
139 * @throws IllegalArgumentException if {@code tickDuration} is <= 0
140 */
141 public HashedWheelTimer(long tickDuration, TimeUnit unit) {
142 this(Executors.defaultThreadFactory(), tickDuration, unit);
143 }
144
145 /**
146 * Creates a new timer with the default thread factory
147 * ({@link Executors#defaultThreadFactory()}).
148 *
149 * @param tickDuration the duration between tick
150 * @param unit the time unit of the {@code tickDuration}
151 * @param ticksPerWheel the size of the wheel
152 * @throws NullPointerException if {@code unit} is {@code null}
153 * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0
154 */
155 public HashedWheelTimer(long tickDuration, TimeUnit unit, int ticksPerWheel) {
156 this(Executors.defaultThreadFactory(), tickDuration, unit, ticksPerWheel);
157 }
158
159 /**
160 * Creates a new timer with the default tick duration and default number of
161 * ticks per wheel.
162 *
163 * @param threadFactory a {@link ThreadFactory} that creates a
164 * background {@link Thread} which is dedicated to
165 * {@link TimerTask} execution.
166 * @throws NullPointerException if {@code threadFactory} is {@code null}
167 */
168 public HashedWheelTimer(ThreadFactory threadFactory) {
169 this(threadFactory, 100, TimeUnit.MILLISECONDS);
170 }
171
172 /**
173 * Creates a new timer with the default number of ticks per wheel.
174 *
175 * @param threadFactory a {@link ThreadFactory} that creates a
176 * background {@link Thread} which is dedicated to
177 * {@link TimerTask} execution.
178 * @param tickDuration the duration between tick
179 * @param unit the time unit of the {@code tickDuration}
180 * @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null}
181 * @throws IllegalArgumentException if {@code tickDuration} is <= 0
182 */
183 public HashedWheelTimer(
184 ThreadFactory threadFactory, long tickDuration, TimeUnit unit) {
185 this(threadFactory, tickDuration, unit, 512);
186 }
187
188 /**
189 * Creates a new timer.
190 *
191 * @param threadFactory a {@link ThreadFactory} that creates a
192 * background {@link Thread} which is dedicated to
193 * {@link TimerTask} execution.
194 * @param tickDuration the duration between tick
195 * @param unit the time unit of the {@code tickDuration}
196 * @param ticksPerWheel the size of the wheel
197 * @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null}
198 * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0
199 */
200 public HashedWheelTimer(
201 ThreadFactory threadFactory,
202 long tickDuration, TimeUnit unit, int ticksPerWheel) {
203 this(threadFactory, tickDuration, unit, ticksPerWheel, true);
204 }
205
206 /**
207 * Creates a new timer.
208 *
209 * @param threadFactory a {@link ThreadFactory} that creates a
210 * background {@link Thread} which is dedicated to
211 * {@link TimerTask} execution.
212 * @param tickDuration the duration between tick
213 * @param unit the time unit of the {@code tickDuration}
214 * @param ticksPerWheel the size of the wheel
215 * @param leakDetection {@code true} if leak detection should be enabled always,
216 * if false it will only be enabled if the worker thread is not
217 * a daemon thread.
218 * @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null}
219 * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0
220 */
221 public HashedWheelTimer(
222 ThreadFactory threadFactory,
223 long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection) {
224 this(threadFactory, tickDuration, unit, ticksPerWheel, leakDetection, -1);
225 }
226
227 /**
228 * Creates a new timer.
229 *
230 * @param threadFactory a {@link ThreadFactory} that creates a
231 * background {@link Thread} which is dedicated to
232 * {@link TimerTask} execution.
233 * @param tickDuration the duration between tick
234 * @param unit the time unit of the {@code tickDuration}
235 * @param ticksPerWheel the size of the wheel
236 * @param leakDetection {@code true} if leak detection should be enabled always,
237 * if false it will only be enabled if the worker thread is not
238 * a daemon thread.
239 * @param maxPendingTimeouts The maximum number of pending timeouts after which call to
240 * {@code newTimeout} will result in
241 * {@link java.util.concurrent.RejectedExecutionException}
242 * being thrown. No maximum pending timeouts limit is assumed if
243 * this value is 0 or negative.
244 * @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null}
245 * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0
246 */
247 public HashedWheelTimer(
248 ThreadFactory threadFactory,
249 long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection,
250 long maxPendingTimeouts) {
251 this(threadFactory, tickDuration, unit, ticksPerWheel, leakDetection,
252 maxPendingTimeouts, ImmediateExecutor.INSTANCE);
253 }
254 /**
255 * Creates a new timer.
256 *
257 * @param threadFactory a {@link ThreadFactory} that creates a
258 * background {@link Thread} which is dedicated to
259 * {@link TimerTask} execution.
260 * @param tickDuration the duration between tick
261 * @param unit the time unit of the {@code tickDuration}
262 * @param ticksPerWheel the size of the wheel
263 * @param leakDetection {@code true} if leak detection should be enabled always,
264 * if false it will only be enabled if the worker thread is not
265 * a daemon thread.
266 * @param maxPendingTimeouts The maximum number of pending timeouts after which call to
267 * {@code newTimeout} will result in
268 * {@link java.util.concurrent.RejectedExecutionException}
269 * being thrown. No maximum pending timeouts limit is assumed if
270 * this value is 0 or negative.
271 * @param taskExecutor The {@link Executor} that is used to execute the submitted {@link TimerTask}s.
272 * The caller is responsible to shutdown the {@link Executor} once it is not needed
273 * anymore.
274 * @throws NullPointerException if either of {@code threadFactory} and {@code unit} is {@code null}
275 * @throws IllegalArgumentException if either of {@code tickDuration} and {@code ticksPerWheel} is <= 0
276 */
277 public HashedWheelTimer(
278 ThreadFactory threadFactory,
279 long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection,
280 long maxPendingTimeouts, Executor taskExecutor) {
281
282 checkNotNull(threadFactory, "threadFactory");
283 checkNotNull(unit, "unit");
284 checkPositive(tickDuration, "tickDuration");
285 checkPositive(ticksPerWheel, "ticksPerWheel");
286 this.taskExecutor = checkNotNull(taskExecutor, "taskExecutor");
287
288 // Normalize ticksPerWheel to power of two and initialize the wheel.
289 wheel = createWheel(ticksPerWheel);
290 mask = wheel.length - 1;
291
292 // Convert tickDuration to nanos.
293 long duration = unit.toNanos(tickDuration);
294
295 // Prevent overflow.
296 if (duration >= Long.MAX_VALUE / wheel.length) {
297 throw new IllegalArgumentException(String.format(
298 "tickDuration: %d (expected: 0 < tickDuration in nanos < %d",
299 tickDuration, Long.MAX_VALUE / wheel.length));
300 }
301
302 if (duration < MILLISECOND_NANOS) {
303 logger.warn("Configured tickDuration {} smaller than {}, using 1ms.",
304 tickDuration, MILLISECOND_NANOS);
305 this.tickDuration = MILLISECOND_NANOS;
306 } else {
307 this.tickDuration = duration;
308 }
309
310 workerThread = threadFactory.newThread(worker);
311
312 leak = leakDetection || !workerThread.isDaemon() ? leakDetector.track(this) : null;
313
314 this.maxPendingTimeouts = maxPendingTimeouts;
315
316 if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&
317 WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {
318 reportTooManyInstances();
319 }
320 }
321
322 @Override
323 protected void finalize() throws Throwable {
324 try {
325 super.finalize();
326 } finally {
327 // This object is going to be GCed and it is assumed the ship has sailed to do a proper shutdown. If
328 // we have not yet shutdown then we want to make sure we decrement the active instance count.
329 if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
330 INSTANCE_COUNTER.decrementAndGet();
331 }
332 }
333 }
334
335 private static HashedWheelBucket[] createWheel(int ticksPerWheel) {
336 ticksPerWheel = MathUtil.findNextPositivePowerOfTwo(ticksPerWheel);
337
338 HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel];
339 for (int i = 0; i < wheel.length; i ++) {
340 wheel[i] = new HashedWheelBucket();
341 }
342 return wheel;
343 }
344
345 /**
346 * Starts the background thread explicitly. The background thread will
347 * start automatically on demand even if you did not call this method.
348 *
349 * @throws IllegalStateException if this timer has been
350 * {@linkplain #stop() stopped} already
351 */
352 public void start() {
353 switch (WORKER_STATE_UPDATER.get(this)) {
354 case WORKER_STATE_INIT:
355 if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
356 workerThread.start();
357 }
358 break;
359 case WORKER_STATE_STARTED:
360 break;
361 case WORKER_STATE_SHUTDOWN:
362 throw new IllegalStateException("cannot be started once stopped");
363 default:
364 throw new Error("Invalid WorkerState");
365 }
366
367 // Wait until the startTime is initialized by the worker.
368 while (startTime == 0) {
369 try {
370 startTimeInitialized.await();
371 } catch (InterruptedException ignore) {
372 // Ignore - it will be ready very soon.
373 }
374 }
375 }
376
377 @Override
378 public Set<Timeout> stop() {
379 if (Thread.currentThread() == workerThread) {
380 throw new IllegalStateException(
381 HashedWheelTimer.class.getSimpleName() +
382 ".stop() cannot be called from " +
383 TimerTask.class.getSimpleName());
384 }
385
386 if (!WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)) {
387 // workerState can be 0 or 2 at this moment - let it always be 2.
388 if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
389 INSTANCE_COUNTER.decrementAndGet();
390 if (leak != null) {
391 boolean closed = leak.close(this);
392 assert closed;
393 }
394 }
395
396 return Collections.emptySet();
397 }
398
399 try {
400 boolean interrupted = false;
401 while (workerThread.isAlive()) {
402 workerThread.interrupt();
403 try {
404 workerThread.join(100);
405 } catch (InterruptedException ignored) {
406 interrupted = true;
407 }
408 }
409
410 if (interrupted) {
411 Thread.currentThread().interrupt();
412 }
413 } finally {
414 INSTANCE_COUNTER.decrementAndGet();
415 if (leak != null) {
416 boolean closed = leak.close(this);
417 assert closed;
418 }
419 }
420 Set<Timeout> unprocessed = worker.unprocessedTimeouts();
421 Set<Timeout> cancelled = new HashSet<Timeout>(unprocessed.size());
422 for (Timeout timeout : unprocessed) {
423 if (timeout.cancel()) {
424 cancelled.add(timeout);
425 }
426 }
427 return cancelled;
428 }
429
430 @Override
431 public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
432 checkNotNull(task, "task");
433 checkNotNull(unit, "unit");
434
435 long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();
436
437 if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) {
438 pendingTimeouts.decrementAndGet();
439 throw new RejectedExecutionException("Number of pending timeouts ("
440 + pendingTimeoutsCount + ") is greater than or equal to maximum allowed pending "
441 + "timeouts (" + maxPendingTimeouts + ")");
442 }
443
444 try {
445 start();
446 } catch (Throwable cause) {
447 pendingTimeouts.decrementAndGet();
448 PlatformDependent.throwException(cause);
449 }
450
451 // Add the timeout to the timeout queue which will be processed on the next tick.
452 // During processing all the queued HashedWheelTimeouts will be added to the correct HashedWheelBucket.
453 long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
454
455 // Guard against overflow.
456 if (delay > 0 && deadline < 0) {
457 deadline = Long.MAX_VALUE;
458 }
459 HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
460 timeouts.add(timeout);
461
462 // stop() might have been called after start() returned, in which case the worker might have already drained
463 // the timeouts queue for the last time. If we can still cancel the timeout it was neither expired nor returned
464 // by stop(), so reject it as if start() had failed.
465 if (WORKER_STATE_UPDATER.get(this) == WORKER_STATE_SHUTDOWN &&
466 timeout.compareAndSetState(HashedWheelTimeout.ST_INIT, HashedWheelTimeout.ST_CANCELLED)) {
467 pendingTimeouts.decrementAndGet();
468 throw new IllegalStateException("cannot be started once stopped");
469 }
470 return timeout;
471 }
472
473 /**
474 * Returns the number of pending timeouts of this {@link Timer}.
475 */
476 public long pendingTimeouts() {
477 return pendingTimeouts.get();
478 }
479
480 private static void reportTooManyInstances() {
481 if (logger.isErrorEnabled()) {
482 String resourceType = simpleClassName(HashedWheelTimer.class);
483 logger.error("You are creating too many " + resourceType + " instances. " +
484 resourceType + " is a shared resource that must be reused across the JVM, " +
485 "so that only a few instances are created.");
486 }
487 }
488
489 private final class Worker implements Runnable {
490 private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();
491
492 private long tick;
493
494 @Override
495 public void run() {
496 // Initialize the startTime.
497 startTime = System.nanoTime();
498 if (startTime == 0) {
499 // We use 0 as an indicator for the uninitialized value here, so make sure it's not 0 when initialized.
500 startTime = 1;
501 }
502
503 // Notify the other threads waiting for the initialization at start().
504 startTimeInitialized.countDown();
505
506 do {
507 final long deadline = waitForNextTick();
508 if (deadline > 0) {
509 int idx = (int) (tick & mask);
510 processCancelledTasks();
511 HashedWheelBucket bucket =
512 wheel[idx];
513 transferTimeoutsToBuckets();
514 bucket.expireTimeouts(deadline);
515 tick++;
516 }
517 } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);
518
519 // Fill the unprocessedTimeouts so we can return them from stop() method.
520 for (HashedWheelBucket bucket: wheel) {
521 bucket.clearTimeouts(unprocessedTimeouts);
522 }
523 for (;;) {
524 HashedWheelTimeout timeout = timeouts.poll();
525 if (timeout == null) {
526 break;
527 }
528 if (!timeout.isCancelled()) {
529 unprocessedTimeouts.add(timeout);
530 }
531 }
532 processCancelledTasks();
533 }
534
535 private void transferTimeoutsToBuckets() {
536 // transfer only max. 100000 timeouts per tick to prevent a thread to stale the workerThread when it just
537 // adds new timeouts in a loop.
538 for (int i = 0; i < 100000; i++) {
539 HashedWheelTimeout timeout = timeouts.poll();
540 if (timeout == null) {
541 // all processed
542 break;
543 }
544 if (timeout.state() == HashedWheelTimeout.ST_CANCELLED) {
545 // Was cancelled in the meantime.
546 continue;
547 }
548
549 long calculated = timeout.deadline / tickDuration;
550 timeout.remainingRounds = (calculated - tick) / wheel.length;
551
552 final long ticks = Math.max(calculated, tick); // Ensure we don't schedule for past.
553 int stopIndex = (int) (ticks & mask);
554
555 HashedWheelBucket bucket = wheel[stopIndex];
556 bucket.addTimeout(timeout);
557 }
558 }
559
560 private void processCancelledTasks() {
561 for (;;) {
562 HashedWheelTimeout timeout = cancelledTimeouts.poll();
563 if (timeout == null) {
564 // all processed
565 break;
566 }
567 try {
568 timeout.remove();
569 } catch (Throwable t) {
570 if (logger.isWarnEnabled()) {
571 logger.warn("An exception was thrown while process a cancellation task", t);
572 }
573 }
574 }
575 }
576
577 /**
578 * calculate goal nanoTime from startTime and current tick number,
579 * then wait until that goal has been reached.
580 * @return Long.MIN_VALUE if received a shutdown request,
581 * current time otherwise (with Long.MIN_VALUE changed by +1)
582 */
583 private long waitForNextTick() {
584 long deadline = tickDuration * (tick + 1);
585
586 for (;;) {
587 final long currentTime = System.nanoTime() - startTime;
588 long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;
589
590 if (sleepTimeMs <= 0) {
591 if (currentTime == Long.MIN_VALUE) {
592 return -Long.MAX_VALUE;
593 } else {
594 return currentTime;
595 }
596 }
597
598 // Check if we run on windows, as if thats the case we will need
599 // to round the sleepTime as workaround for a bug that only affect
600 // the JVM if it runs on windows.
601 //
602 // See https://github.com/netty/netty/issues/356
603 if (PlatformDependent.isWindows()) {
604 sleepTimeMs = sleepTimeMs / 10 * 10;
605 if (sleepTimeMs == 0) {
606 sleepTimeMs = 1;
607 }
608 }
609
610 try {
611 Thread.sleep(sleepTimeMs);
612 } catch (InterruptedException ignored) {
613 if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {
614 return Long.MIN_VALUE;
615 }
616 }
617 }
618 }
619
620 public Set<Timeout> unprocessedTimeouts() {
621 return Collections.unmodifiableSet(unprocessedTimeouts);
622 }
623 }
624
625 private static final class HashedWheelTimeout implements Timeout, Runnable {
626
627 private static final int ST_INIT = 0;
628 private static final int ST_CANCELLED = 1;
629 private static final int ST_EXPIRED = 2;
630 private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATER =
631 AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state");
632
633 private final HashedWheelTimer timer;
634 private final TimerTask task;
635 private final long deadline;
636
637 @SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization" })
638 private volatile int state = ST_INIT;
639
640 // remainingRounds will be calculated and set by Worker.transferTimeoutsToBuckets() before the
641 // HashedWheelTimeout will be added to the correct HashedWheelBucket.
642 long remainingRounds;
643
644 // This will be used to chain timeouts in HashedWheelTimerBucket via a double-linked-list.
645 // As only the workerThread will act on it there is no need for synchronization / volatile.
646 HashedWheelTimeout next;
647 HashedWheelTimeout prev;
648
649 // The bucket to which the timeout was added
650 HashedWheelBucket bucket;
651
652 HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) {
653 this.timer = timer;
654 this.task = task;
655 this.deadline = deadline;
656 }
657
658 @Override
659 public Timer timer() {
660 return timer;
661 }
662
663 @Override
664 public TimerTask task() {
665 return task;
666 }
667
668 @Override
669 public boolean cancel() {
670 // only update the state it will be removed from HashedWheelBucket on next tick.
671 if (!compareAndSetState(ST_INIT, ST_CANCELLED)) {
672 return false;
673 }
674 // If a task should be canceled we put this to another queue which will be processed on each tick.
675 // So this means that we will have a GC latency of max. 1 tick duration which is good enough. This way
676 // we can make again use of our MpscLinkedQueue and so minimize the locking / overhead as much as possible.
677 timer.cancelledTimeouts.add(this);
678 return true;
679 }
680
681 void remove() {
682 HashedWheelBucket bucket = this.bucket;
683 if (bucket != null) {
684 bucket.remove(this);
685 }
686 timer.pendingTimeouts.decrementAndGet();
687 }
688
689 public boolean compareAndSetState(int expected, int state) {
690 return STATE_UPDATER.compareAndSet(this, expected, state);
691 }
692
693 public int state() {
694 return state;
695 }
696
697 @Override
698 public boolean isCancelled() {
699 return state() == ST_CANCELLED;
700 }
701
702 @Override
703 public boolean isExpired() {
704 return state() == ST_EXPIRED;
705 }
706
707 public void expire() {
708 if (!compareAndSetState(ST_INIT, ST_EXPIRED)) {
709 return;
710 }
711
712 try {
713 remove();
714 timer.taskExecutor.execute(this);
715 } catch (Throwable t) {
716 if (logger.isWarnEnabled()) {
717 logger.warn("An exception was thrown while submit " + TimerTask.class.getSimpleName()
718 + " for execution.", t);
719 }
720 }
721 }
722
723 @Override
724 public void run() {
725 try {
726 task.run(this);
727 } catch (Throwable t) {
728 if (logger.isWarnEnabled()) {
729 logger.warn("An exception was thrown by " + TimerTask.class.getSimpleName() + '.', t);
730 }
731 }
732 }
733
734 @Override
735 public String toString() {
736 final long currentTime = System.nanoTime();
737 long remaining = deadline - currentTime + timer.startTime;
738
739 StringBuilder buf = new StringBuilder(192)
740 .append(simpleClassName(this))
741 .append('(')
742 .append("deadline: ");
743 if (remaining > 0) {
744 buf.append(remaining)
745 .append(" ns later");
746 } else if (remaining < 0) {
747 buf.append(-remaining)
748 .append(" ns ago");
749 } else {
750 buf.append("now");
751 }
752
753 if (isCancelled()) {
754 buf.append(", cancelled");
755 }
756
757 return buf.append(", task: ")
758 .append(task())
759 .append(')')
760 .toString();
761 }
762 }
763
764 /**
765 * Bucket that stores HashedWheelTimeouts. These are stored in a linked-list like datastructure to allow easy
766 * removal of HashedWheelTimeouts in the middle. Also the HashedWheelTimeout act as nodes themself and so no
767 * extra object creation is needed.
768 */
769 private static final class HashedWheelBucket {
770 // Used for the linked-list datastructure
771 private HashedWheelTimeout head;
772 private HashedWheelTimeout tail;
773
774 /**
775 * Add {@link HashedWheelTimeout} to this bucket.
776 */
777 public void addTimeout(HashedWheelTimeout timeout) {
778 assert timeout.bucket == null;
779 timeout.bucket = this;
780 if (head == null) {
781 head = tail = timeout;
782 } else {
783 tail.next = timeout;
784 timeout.prev = tail;
785 tail = timeout;
786 }
787 }
788
789 /**
790 * Expire all {@link HashedWheelTimeout}s for the given {@code deadline}.
791 */
792 public void expireTimeouts(long deadline) {
793 HashedWheelTimeout timeout = head;
794
795 // process all timeouts
796 while (timeout != null) {
797 HashedWheelTimeout next = timeout.next;
798 if (timeout.remainingRounds <= 0) {
799 if (timeout.deadline <= deadline) {
800 timeout.expire();
801 } else {
802 // The timeout was placed into a wrong slot. This should never happen.
803 throw new IllegalStateException(String.format(
804 "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
805 }
806 } else if (!timeout.isCancelled()) {
807 timeout.remainingRounds --;
808 }
809 timeout = next;
810 }
811 }
812
813 public HashedWheelTimeout remove(HashedWheelTimeout timeout) {
814 HashedWheelTimeout next = timeout.next;
815 // remove timeout that was either processed or cancelled by updating the linked-list
816 if (timeout.prev != null) {
817 timeout.prev.next = next;
818 }
819 if (timeout.next != null) {
820 timeout.next.prev = timeout.prev;
821 }
822
823 if (timeout == head) {
824 // if timeout is also the tail we need to adjust the entry too
825 if (timeout == tail) {
826 tail = null;
827 head = null;
828 } else {
829 head = next;
830 }
831 } else if (timeout == tail) {
832 // if the timeout is the tail modify the tail to be the prev node.
833 tail = timeout.prev;
834 }
835 // null out prev, next and bucket to allow for GC.
836 timeout.prev = null;
837 timeout.next = null;
838 timeout.bucket = null;
839 return next;
840 }
841
842 /**
843 * Clear this bucket and return all not expired / cancelled {@link Timeout}s.
844 */
845 public void clearTimeouts(Set<Timeout> set) {
846 for (;;) {
847 HashedWheelTimeout timeout = pollTimeout();
848 if (timeout == null) {
849 return;
850 }
851 if (timeout.isExpired() || timeout.isCancelled()) {
852 continue;
853 }
854 set.add(timeout);
855 }
856 }
857
858 private HashedWheelTimeout pollTimeout() {
859 HashedWheelTimeout head = this.head;
860 if (head == null) {
861 return null;
862 }
863 HashedWheelTimeout next = head.next;
864 if (next == null) {
865 tail = this.head = null;
866 } else {
867 this.head = next;
868 next.prev = null;
869 }
870
871 // null out prev and next to allow for GC.
872 head.next = null;
873 head.prev = null;
874 head.bucket = null;
875 return head;
876 }
877 }
878 }