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