View Javadoc
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 &lt;= 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 &lt;= 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 &lt;= 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 &lt;= 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 &lt;= 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 &lt;= 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 &lt;= 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             // This object is going to be GCed and it is assumed the ship has sailed to do a proper shutdown. If
326             // we have not yet shutdown then we want to make sure we decrement the active instance count.
327             if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
328                 INSTANCE_COUNTER.decrementAndGet();
329             }
330         } finally {
331             super.finalize();
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         int state = WORKER_STATE_UPDATER.get(this);
354         switch (state) {
355             case WORKER_STATE_INIT:
356                 if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
357                     workerThread.start();
358                 }
359                 break;
360             case WORKER_STATE_STARTED:
361                 break;
362             case WORKER_STATE_SHUTDOWN:
363                 throw new IllegalStateException("cannot be started once stopped");
364             default:
365                 throw new Error("Invalid WorkerState: " + state);
366         }
367 
368         // Wait until the startTime is initialized by the worker.
369         while (startTime == 0) {
370             try {
371                 startTimeInitialized.await();
372             } catch (InterruptedException ignore) {
373                 // Ignore - it will be ready very soon.
374             }
375         }
376     }
377 
378     @Override
379     public Set<Timeout> stop() {
380         if (Thread.currentThread() == workerThread) {
381             throw new IllegalStateException(
382                     HashedWheelTimer.class.getSimpleName() +
383                             ".stop() cannot be called from " +
384                             TimerTask.class.getSimpleName());
385         }
386 
387         if (!WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)) {
388             // workerState can be 0 or 2 at this moment - let it always be 2.
389             if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
390                 INSTANCE_COUNTER.decrementAndGet();
391                 if (leak != null) {
392                     boolean closed = leak.close(this);
393                     assert closed;
394                 }
395             }
396 
397             return Collections.emptySet();
398         }
399 
400         try {
401             boolean interrupted = false;
402             while (workerThread.isAlive()) {
403                 workerThread.interrupt();
404                 try {
405                     workerThread.join(100);
406                 } catch (InterruptedException ignored) {
407                     interrupted = true;
408                 }
409             }
410 
411             if (interrupted) {
412                 Thread.currentThread().interrupt();
413             }
414         } finally {
415             INSTANCE_COUNTER.decrementAndGet();
416             if (leak != null) {
417                 boolean closed = leak.close(this);
418                 assert closed;
419             }
420         }
421         Set<Timeout> unprocessed = worker.unprocessedTimeouts();
422         Set<Timeout> cancelled = new HashSet<Timeout>(unprocessed.size());
423         for (Timeout timeout : unprocessed) {
424             if (timeout.cancel()) {
425                 cancelled.add(timeout);
426             }
427         }
428         return cancelled;
429     }
430 
431     @Override
432     public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
433         checkNotNull(task, "task");
434         checkNotNull(unit, "unit");
435 
436         long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();
437 
438         if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) {
439             pendingTimeouts.decrementAndGet();
440             throw new RejectedExecutionException("Number of pending timeouts ("
441                 + pendingTimeoutsCount + ") is greater than or equal to maximum allowed pending "
442                 + "timeouts (" + maxPendingTimeouts + ")");
443         }
444 
445         try {
446             start();
447         } catch (Throwable cause) {
448             pendingTimeouts.decrementAndGet();
449             throw cause;
450         }
451 
452         // Add the timeout to the timeout queue which will be processed on the next tick.
453         // During processing all the queued HashedWheelTimeouts will be added to the correct HashedWheelBucket.
454         long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
455 
456         // Guard against overflow.
457         if (delay > 0 && deadline < 0) {
458             deadline = Long.MAX_VALUE;
459         }
460         HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
461         timeouts.add(timeout);
462 
463         // stop() might have been called after start() returned, in which case the worker might have already drained
464         // the timeouts queue for the last time. If we can still cancel the timeout it was neither expired nor returned
465         // by stop(), so reject it as if start() had failed.
466         if (WORKER_STATE_UPDATER.get(this) == WORKER_STATE_SHUTDOWN &&
467                 timeout.compareAndSetState(HashedWheelTimeout.ST_INIT, HashedWheelTimeout.ST_CANCELLED)) {
468             pendingTimeouts.decrementAndGet();
469             throw new IllegalStateException("cannot be started once stopped");
470         }
471         return timeout;
472     }
473 
474     /**
475      * Returns the number of pending timeouts of this {@link Timer}.
476      */
477     public long pendingTimeouts() {
478         return pendingTimeouts.get();
479     }
480 
481     private static void reportTooManyInstances() {
482         if (logger.isErrorEnabled()) {
483             String resourceType = simpleClassName(HashedWheelTimer.class);
484             logger.error("You are creating too many " + resourceType + " instances. " +
485                     resourceType + " is a shared resource that must be reused across the JVM, " +
486                     "so that only a few instances are created.");
487         }
488     }
489 
490     private final class Worker implements Runnable {
491         private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();
492 
493         private long tick;
494 
495         @Override
496         public void run() {
497             // Initialize the startTime.
498             startTime = System.nanoTime();
499             if (startTime == 0) {
500                 // We use 0 as an indicator for the uninitialized value here, so make sure it's not 0 when initialized.
501                 startTime = 1;
502             }
503 
504             // Notify the other threads waiting for the initialization at start().
505             startTimeInitialized.countDown();
506 
507             do {
508                 final long deadline = waitForNextTick();
509                 if (deadline > 0) {
510                     int idx = (int) (tick & mask);
511                     processCancelledTasks();
512                     HashedWheelBucket bucket =
513                             wheel[idx];
514                     transferTimeoutsToBuckets();
515                     bucket.expireTimeouts(deadline);
516                     tick++;
517                 }
518             } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);
519 
520             // Fill the unprocessedTimeouts so we can return them from stop() method.
521             for (HashedWheelBucket bucket: wheel) {
522                 bucket.clearTimeouts(unprocessedTimeouts);
523             }
524             for (;;) {
525                 HashedWheelTimeout timeout = timeouts.poll();
526                 if (timeout == null) {
527                     break;
528                 }
529                 if (!timeout.isCancelled()) {
530                     unprocessedTimeouts.add(timeout);
531                 }
532             }
533             processCancelledTasks();
534         }
535 
536         private void transferTimeoutsToBuckets() {
537             // transfer only max. 100000 timeouts per tick to prevent a thread to stale the workerThread when it just
538             // adds new timeouts in a loop.
539             for (int i = 0; i < 100000; i++) {
540                 HashedWheelTimeout timeout = timeouts.poll();
541                 if (timeout == null) {
542                     // all processed
543                     break;
544                 }
545                 if (timeout.state() == HashedWheelTimeout.ST_CANCELLED) {
546                     // Was cancelled in the meantime.
547                     continue;
548                 }
549 
550                 long calculated = timeout.deadline / tickDuration;
551                 timeout.remainingRounds = (calculated - tick) / wheel.length;
552 
553                 final long ticks = Math.max(calculated, tick); // Ensure we don't schedule for past.
554                 int stopIndex = (int) (ticks & mask);
555 
556                 HashedWheelBucket bucket = wheel[stopIndex];
557                 bucket.addTimeout(timeout);
558             }
559         }
560 
561         private void processCancelledTasks() {
562             for (;;) {
563                 HashedWheelTimeout timeout = cancelledTimeouts.poll();
564                 if (timeout == null) {
565                     // all processed
566                     break;
567                 }
568                 try {
569                     timeout.removeAfterCancellation();
570                 } catch (Throwable t) {
571                     if (logger.isWarnEnabled()) {
572                         logger.warn("An exception was thrown while process a cancellation task", t);
573                     }
574                 }
575             }
576         }
577 
578         /**
579          * calculate goal nanoTime from startTime and current tick number,
580          * then wait until that goal has been reached.
581          * @return Long.MIN_VALUE if received a shutdown request,
582          * current time otherwise (with Long.MIN_VALUE changed by +1)
583          */
584         private long waitForNextTick() {
585             long deadline = tickDuration * (tick + 1);
586 
587             for (;;) {
588                 final long currentTime = System.nanoTime() - startTime;
589                 long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;
590 
591                 if (sleepTimeMs <= 0) {
592                     if (currentTime == Long.MIN_VALUE) {
593                         return -Long.MAX_VALUE;
594                     } else {
595                         return currentTime;
596                     }
597                 }
598 
599                 // Check if we run on windows, as if thats the case we will need
600                 // to round the sleepTime as workaround for a bug that only affect
601                 // the JVM if it runs on windows.
602                 //
603                 // See https://github.com/netty/netty/issues/356
604                 if (PlatformDependent.isWindows()) {
605                     sleepTimeMs = sleepTimeMs / 10 * 10;
606                     if (sleepTimeMs == 0) {
607                         sleepTimeMs = 1;
608                     }
609                 }
610 
611                 try {
612                     Thread.sleep(sleepTimeMs);
613                 } catch (InterruptedException ignored) {
614                     if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {
615                         return Long.MIN_VALUE;
616                     }
617                 }
618             }
619         }
620 
621         public Set<Timeout> unprocessedTimeouts() {
622             return Collections.unmodifiableSet(unprocessedTimeouts);
623         }
624     }
625 
626     private static final class HashedWheelTimeout implements Timeout, Runnable {
627 
628         private static final int ST_INIT = 0;
629         private static final int ST_CANCELLED = 1;
630         private static final int ST_EXPIRED = 2;
631         private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATER =
632                 AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state");
633 
634         private final HashedWheelTimer timer;
635         private final TimerTask task;
636         private final long deadline;
637 
638         @SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization" })
639         private volatile int state = ST_INIT;
640 
641         // remainingRounds will be calculated and set by Worker.transferTimeoutsToBuckets() before the
642         // HashedWheelTimeout will be added to the correct HashedWheelBucket.
643         long remainingRounds;
644 
645         // This will be used to chain timeouts in HashedWheelTimerBucket via a double-linked-list.
646         // As only the workerThread will act on it there is no need for synchronization / volatile.
647         HashedWheelTimeout next;
648         HashedWheelTimeout prev;
649 
650         // The bucket to which the timeout was added
651         HashedWheelBucket bucket;
652 
653         HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) {
654             this.timer = timer;
655             this.task = task;
656             this.deadline = deadline;
657         }
658 
659         @Override
660         public Timer timer() {
661             return timer;
662         }
663 
664         @Override
665         public TimerTask task() {
666             return task;
667         }
668 
669         @Override
670         public boolean cancel() {
671             // only update the state it will be removed from HashedWheelBucket on next tick.
672             if (!compareAndSetState(ST_INIT, ST_CANCELLED)) {
673                 return false;
674             }
675             // If a task should be canceled we put this to another queue which will be processed on each tick.
676             // So this means that we will have a GC latency of max. 1 tick duration which is good enough. This way
677             // we can make again use of our MpscLinkedQueue and so minimize the locking / overhead as much as possible.
678             timer.cancelledTimeouts.add(this);
679             return true;
680         }
681 
682         private void remove() {
683             HashedWheelBucket bucket = this.bucket;
684             if (bucket != null) {
685                 bucket.remove(this);
686             }
687             timer.pendingTimeouts.decrementAndGet();
688         }
689         void removeAfterCancellation() {
690             remove();
691             task.cancelled(this);
692         }
693 
694         public boolean compareAndSetState(int expected, int state) {
695             return STATE_UPDATER.compareAndSet(this, expected, state);
696         }
697 
698         public int state() {
699             return state;
700         }
701 
702         @Override
703         public boolean isCancelled() {
704             return state() == ST_CANCELLED;
705         }
706 
707         @Override
708         public boolean isExpired() {
709             return state() == ST_EXPIRED;
710         }
711 
712         public void expire() {
713             if (!compareAndSetState(ST_INIT, ST_EXPIRED)) {
714                 return;
715             }
716 
717             try {
718                 remove();
719                 timer.taskExecutor.execute(this);
720             } catch (Throwable t) {
721                 if (logger.isWarnEnabled()) {
722                     logger.warn("An exception was thrown while submit " + TimerTask.class.getSimpleName()
723                             + " for execution.", t);
724                 }
725             }
726         }
727 
728         @Override
729         public void run() {
730             try {
731                 task.run(this);
732             } catch (Throwable t) {
733                 if (logger.isWarnEnabled()) {
734                     logger.warn("An exception was thrown by " + TimerTask.class.getSimpleName() + '.', t);
735                 }
736             }
737         }
738 
739         @Override
740         public String toString() {
741             final long currentTime = System.nanoTime();
742             long remaining = deadline - currentTime + timer.startTime;
743 
744             StringBuilder buf = new StringBuilder(192)
745                .append(simpleClassName(this))
746                .append('(')
747                .append("deadline: ");
748             if (remaining > 0) {
749                 buf.append(remaining)
750                    .append(" ns later");
751             } else if (remaining < 0) {
752                 buf.append(-remaining)
753                    .append(" ns ago");
754             } else {
755                 buf.append("now");
756             }
757 
758             if (isCancelled()) {
759                 buf.append(", cancelled");
760             }
761 
762             return buf.append(", task: ")
763                       .append(task())
764                       .append(')')
765                       .toString();
766         }
767     }
768 
769     /**
770      * Bucket that stores HashedWheelTimeouts. These are stored in a linked-list like datastructure to allow easy
771      * removal of HashedWheelTimeouts in the middle. Also the HashedWheelTimeout act as nodes themself and so no
772      * extra object creation is needed.
773      */
774     private static final class HashedWheelBucket {
775         // Used for the linked-list datastructure
776         private HashedWheelTimeout head;
777         private HashedWheelTimeout tail;
778 
779         /**
780          * Add {@link HashedWheelTimeout} to this bucket.
781          */
782         public void addTimeout(HashedWheelTimeout timeout) {
783             assert timeout.bucket == null;
784             timeout.bucket = this;
785             if (head == null) {
786                 head = tail = timeout;
787             } else {
788                 tail.next = timeout;
789                 timeout.prev = tail;
790                 tail = timeout;
791             }
792         }
793 
794         /**
795          * Expire all {@link HashedWheelTimeout}s for the given {@code deadline}.
796          */
797         public void expireTimeouts(long deadline) {
798             HashedWheelTimeout timeout = head;
799 
800             // process all timeouts
801             while (timeout != null) {
802                 HashedWheelTimeout next = timeout.next;
803                 if (timeout.remainingRounds <= 0) {
804                     if (timeout.deadline <= deadline) {
805                         timeout.expire();
806                     } else {
807                         // The timeout was placed into a wrong slot. This should never happen.
808                         throw new IllegalStateException(String.format(
809                                 "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
810                     }
811                 } else if (!timeout.isCancelled()) {
812                     timeout.remainingRounds --;
813                 }
814                 timeout = next;
815             }
816         }
817 
818         public HashedWheelTimeout remove(HashedWheelTimeout timeout) {
819             HashedWheelTimeout prev = timeout.prev;
820             HashedWheelTimeout next = timeout.next;
821 
822             // remove timeout that was either processed or cancelled by updating the linked-list
823             if (prev != null) {
824                 prev.next = next;
825             }
826             if (next != null) {
827                 next.prev = prev;
828             }
829 
830             if (timeout == head) {
831                 head = next;
832             }
833             if (timeout == tail) {
834                 tail = prev;
835             }
836             // null out prev, next and bucket to allow for GC.
837             timeout.prev = null;
838             timeout.next = null;
839             timeout.bucket = null;
840             return next;
841         }
842 
843         /**
844          * Clear this bucket and return all not expired / cancelled {@link Timeout}s.
845          */
846         public void clearTimeouts(Set<Timeout> set) {
847             for (;;) {
848                 HashedWheelTimeout timeout = pollTimeout();
849                 if (timeout == null) {
850                     return;
851                 }
852                 if (timeout.isExpired() || timeout.isCancelled()) {
853                     continue;
854                 }
855                 set.add(timeout);
856             }
857         }
858 
859         private HashedWheelTimeout pollTimeout() {
860             HashedWheelTimeout head = this.head;
861             if (head == null) {
862                 return null;
863             }
864             HashedWheelTimeout next = head.next;
865             if (next == null) {
866                 tail = this.head =  null;
867             } else {
868                 this.head = next;
869                 next.prev = null;
870             }
871 
872             // null out prev and next to allow for GC.
873             head.next = null;
874             head.prev = null;
875             head.bucket = null;
876             return head;
877         }
878     }
879 }