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    *   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 org.jboss.netty.util;
17  
18  import org.jboss.netty.channel.ChannelPipelineFactory;
19  import org.jboss.netty.logging.InternalLogger;
20  import org.jboss.netty.logging.InternalLoggerFactory;
21  import org.jboss.netty.util.internal.DetectionUtil;
22  import org.jboss.netty.util.internal.SharedResourceMisuseDetector;
23  
24  import java.util.Collections;
25  import java.util.HashSet;
26  import java.util.Queue;
27  import java.util.Set;
28  import java.util.concurrent.ConcurrentLinkedQueue;
29  import java.util.concurrent.CountDownLatch;
30  import java.util.concurrent.Executors;
31  import java.util.concurrent.ThreadFactory;
32  import java.util.concurrent.TimeUnit;
33  import java.util.concurrent.atomic.AtomicInteger;
34  import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
35  
36  /**
37   * A {@link Timer} optimized for approximated I/O timeout scheduling.
38   *
39   * <h3>Tick Duration</h3>
40   *
41   * As described with 'approximated', this timer does not execute the scheduled
42   * {@link TimerTask} on time.  {@link HashedWheelTimer}, on every tick, will
43   * check if there are any {@link TimerTask}s behind the schedule and execute
44   * them.
45   * <p>
46   * You can increase or decrease the accuracy of the execution timing by
47   * specifying smaller or larger tick duration in the constructor.  In most
48   * network applications, I/O timeout does not need to be accurate.  Therefore,
49   * the default tick duration is 100 milliseconds and you will not need to try
50   * different configurations in most cases.
51   *
52   * <h3>Ticks per Wheel (Wheel Size)</h3>
53   *
54   * {@link HashedWheelTimer} maintains a data structure called 'wheel'.
55   * To put simply, a wheel is a hash table of {@link TimerTask}s whose hash
56   * function is 'dead line of the task'.  The default number of ticks per wheel
57   * (i.e. the size of the wheel) is 512.  You could specify a larger value
58   * if you are going to schedule a lot of timeouts.
59   *
60   * <h3>Do not create many instances.</h3>
61   *
62   * {@link HashedWheelTimer} creates a new thread whenever it is instantiated and
63   * started.  Therefore, you should make sure to create only one instance and
64   * share it across your application.  One of the common mistakes, that makes
65   * your application unresponsive, is to create a new instance in
66   * {@link ChannelPipelineFactory}, which results in the creation of a new thread
67   * 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      private static final AtomicInteger id = new AtomicInteger();
84  
85      private static final SharedResourceMisuseDetector misuseDetector =
86          new SharedResourceMisuseDetector(HashedWheelTimer.class);
87  
88      private static final AtomicIntegerFieldUpdater<HashedWheelTimer> WORKER_STATE_UPDATER =
89              AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimer.class, "workerState");
90  
91      private final Worker worker = new Worker();
92      private final Thread workerThread;
93  
94      public static final int WORKER_STATE_INIT = 0;
95      public static final int WORKER_STATE_STARTED = 1;
96      public static final int WORKER_STATE_SHUTDOWN = 2;
97      @SuppressWarnings({ "unused", "FieldMayBeFinal", "RedundantFieldInitialization" })
98      private volatile int workerState = WORKER_STATE_INIT; // 0 - init, 1 - started, 2 - shut down
99  
100     private final long tickDuration;
101     private final HashedWheelBucket[] wheel;
102     private final int mask;
103     private final CountDownLatch startTimeInitialized = new CountDownLatch(1);
104     private final Queue<HashedWheelTimeout> timeouts = new ConcurrentLinkedQueue<HashedWheelTimeout>();
105     private volatile long startTime;
106 
107     /**
108      * Creates a new timer with the default thread factory
109      * ({@link Executors#defaultThreadFactory()}), default tick duration, and
110      * default number of ticks per wheel.
111      */
112     public HashedWheelTimer() {
113         this(Executors.defaultThreadFactory());
114     }
115 
116     /**
117      * Creates a new timer with the default thread factory
118      * ({@link Executors#defaultThreadFactory()}) and default number of ticks
119      * per wheel.
120      *
121      * @param tickDuration   the duration between tick
122      * @param unit           the time unit of the {@code tickDuration}
123      */
124     public HashedWheelTimer(long tickDuration, TimeUnit unit) {
125         this(Executors.defaultThreadFactory(), tickDuration, unit);
126     }
127 
128     /**
129      * Creates a new timer with the default thread factory
130      * ({@link Executors#defaultThreadFactory()}).
131      *
132      * @param tickDuration   the duration between tick
133      * @param unit           the time unit of the {@code tickDuration}
134      * @param ticksPerWheel  the size of the wheel
135      */
136     public HashedWheelTimer(long tickDuration, TimeUnit unit, int ticksPerWheel) {
137         this(Executors.defaultThreadFactory(), tickDuration, unit, ticksPerWheel);
138     }
139 
140     /**
141      * Creates a new timer with the default tick duration and default number of
142      * ticks per wheel.
143      *
144      * @param threadFactory  a {@link ThreadFactory} that creates a
145      *                       background {@link Thread} which is dedicated to
146      *                       {@link TimerTask} execution.
147      */
148     public HashedWheelTimer(ThreadFactory threadFactory) {
149         this(threadFactory, 100, TimeUnit.MILLISECONDS);
150     }
151 
152     /**
153      * Creates a new timer with the default number of 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      * @param tickDuration   the duration between tick
159      * @param unit           the time unit of the {@code tickDuration}
160      */
161     public HashedWheelTimer(
162             ThreadFactory threadFactory, long tickDuration, TimeUnit unit) {
163         this(threadFactory, tickDuration, unit, 512);
164     }
165 
166     /**
167      * Creates a new timer.
168      *
169      * @param threadFactory  a {@link ThreadFactory} that creates a
170      *                       background {@link Thread} which is dedicated to
171      *                       {@link TimerTask} execution.
172      * @param tickDuration   the duration between tick
173      * @param unit           the time unit of the {@code tickDuration}
174      * @param ticksPerWheel  the size of the wheel
175      */
176     public HashedWheelTimer(
177             ThreadFactory threadFactory,
178             long tickDuration, TimeUnit unit, int ticksPerWheel) {
179         this(threadFactory, null, tickDuration, unit, ticksPerWheel);
180     }
181 
182     /**
183      * Creates a new timer.
184      *
185      * @param threadFactory  a {@link ThreadFactory} that creates a
186      *                       background {@link Thread} which is dedicated to
187      *                       {@link TimerTask} execution.
188      * @param determiner     thread name determiner to control thread name.
189      * @param tickDuration   the duration between tick
190      * @param unit           the time unit of the {@code tickDuration}
191      * @param ticksPerWheel  the size of the wheel
192      */
193     public HashedWheelTimer(
194             ThreadFactory threadFactory,
195             ThreadNameDeterminer determiner,
196             long tickDuration, TimeUnit unit, int ticksPerWheel) {
197 
198         if (threadFactory == null) {
199             throw new NullPointerException("threadFactory");
200         }
201         if (unit == null) {
202             throw new NullPointerException("unit");
203         }
204         if (tickDuration <= 0) {
205             throw new IllegalArgumentException(
206                     "tickDuration must be greater than 0: " + tickDuration);
207         }
208         if (ticksPerWheel <= 0) {
209             throw new IllegalArgumentException(
210                     "ticksPerWheel must be greater than 0: " + ticksPerWheel);
211         }
212 
213         // Normalize ticksPerWheel to power of two and initialize the wheel.
214         wheel = createWheel(ticksPerWheel);
215         mask = wheel.length - 1;
216 
217         // Convert tickDuration to nanos.
218         this.tickDuration = unit.toNanos(tickDuration);
219 
220         // Prevent overflow.
221         if (this.tickDuration >= Long.MAX_VALUE / wheel.length) {
222             throw new IllegalArgumentException(String.format(
223                     "tickDuration: %d (expected: 0 < tickDuration in nanos < %d",
224                     tickDuration, Long.MAX_VALUE / wheel.length));
225         }
226 
227         workerThread = threadFactory.newThread(new ThreadRenamingRunnable(
228                         worker, "Hashed wheel timer #" + id.incrementAndGet(),
229                         determiner));
230 
231         // Misuse check
232         misuseDetector.increase();
233     }
234 
235     @SuppressWarnings("unchecked")
236     private static HashedWheelBucket[] createWheel(int ticksPerWheel) {
237         if (ticksPerWheel <= 0) {
238             throw new IllegalArgumentException(
239                     "ticksPerWheel must be greater than 0: " + ticksPerWheel);
240         }
241         if (ticksPerWheel > 1073741824) {
242             throw new IllegalArgumentException(
243                     "ticksPerWheel may not be greater than 2^30: " + ticksPerWheel);
244         }
245 
246         ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);
247         HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel];
248         for (int i = 0; i < wheel.length; i ++) {
249             wheel[i] = new HashedWheelBucket();
250         }
251         return wheel;
252     }
253 
254     private static int normalizeTicksPerWheel(int ticksPerWheel) {
255         int normalizedTicksPerWheel = 1;
256         while (normalizedTicksPerWheel < ticksPerWheel) {
257             normalizedTicksPerWheel <<= 1;
258         }
259         return normalizedTicksPerWheel;
260     }
261 
262     /**
263      * Starts the background thread explicitly.  The background thread will
264      * start automatically on demand even if you did not call this method.
265      *
266      * @throws IllegalStateException if this timer has been
267      *                               {@linkplain #stop() stopped} already
268      */
269     public void start() {
270         switch (WORKER_STATE_UPDATER.get(this)) {
271             case WORKER_STATE_INIT:
272                 if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
273                     workerThread.start();
274                 }
275                 break;
276             case WORKER_STATE_STARTED:
277                 break;
278             case WORKER_STATE_SHUTDOWN:
279                 throw new IllegalStateException("cannot be started once stopped");
280             default:
281                 throw new Error("Invalid WorkerState");
282         }
283 
284         // Wait until the startTime is initialized by the worker.
285         while (startTime == 0) {
286             try {
287                 startTimeInitialized.await();
288             } catch (InterruptedException ignore) {
289                 // Ignore - it will be ready very soon.
290             }
291         }
292     }
293 
294     public Set<Timeout> stop() {
295         if (Thread.currentThread() == workerThread) {
296             throw new IllegalStateException(
297                     HashedWheelTimer.class.getSimpleName() +
298                             ".stop() cannot be called from " +
299                             TimerTask.class.getSimpleName());
300         }
301 
302         if (!WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)) {
303             // workerState can be 0 or 2 at this moment - let it always be 2.
304             WORKER_STATE_UPDATER.set(this, WORKER_STATE_SHUTDOWN);
305 
306             misuseDetector.decrease();
307 
308             return Collections.emptySet();
309         }
310 
311         boolean interrupted = false;
312         while (workerThread.isAlive()) {
313             workerThread.interrupt();
314             try {
315                 workerThread.join(100);
316             } catch (InterruptedException e) {
317                 interrupted = true;
318             }
319         }
320 
321         if (interrupted) {
322             Thread.currentThread().interrupt();
323         }
324 
325         misuseDetector.decrease();
326 
327         return worker.unprocessedTimeouts();
328     }
329 
330     public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
331         if (task == null) {
332             throw new NullPointerException("task");
333         }
334         if (unit == null) {
335             throw new NullPointerException("unit");
336         }
337         start();
338 
339         // Add the timeout to the timeout queue which will be processed on the next tick.
340         // During processing all the queued HashedWheelTimeouts will be added to the correct HashedWheelBucket.
341         long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
342         HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
343         timeouts.add(timeout);
344         return timeout;
345     }
346 
347     private final class Worker implements Runnable {
348         private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();
349 
350         private long tick;
351 
352         public void run() {
353             // Initialize the startTime.
354             startTime = System.nanoTime();
355             if (startTime == 0) {
356                 // We use 0 as an indicator for the uninitialized value here, so make sure it's not 0 when initialized.
357                 startTime = 1;
358             }
359 
360             // Notify the other threads waiting for the initialization at start().
361             startTimeInitialized.countDown();
362 
363             do {
364                 final long deadline = waitForNextTick();
365                 if (deadline > 0) {
366                     transferTimeoutsToBuckets();
367                     HashedWheelBucket bucket =
368                             wheel[(int) (tick & mask)];
369                     bucket.expireTimeouts(deadline);
370                     tick++;
371                 }
372             } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);
373 
374             // Fill the unprocessedTimeouts so we can return them from stop() method.
375             for (HashedWheelBucket bucket: wheel) {
376                 bucket.clearTimeouts(unprocessedTimeouts);
377             }
378             for (;;) {
379                 HashedWheelTimeout timeout = timeouts.poll();
380                 if (timeout == null) {
381                     break;
382                 }
383                 unprocessedTimeouts.add(timeout);
384             }
385         }
386 
387         private void transferTimeoutsToBuckets() {
388             // transfer only max. 100000 timeouts per tick to prevent a thread to stale the workerThread when it just
389             // adds new timeouts in a loop.
390             for (int i = 0; i < 100000; i++) {
391                 HashedWheelTimeout timeout = timeouts.poll();
392                 if (timeout == null) {
393                     // all processed
394                     break;
395                 }
396                 if (timeout.state() == HashedWheelTimeout.ST_CANCELLED
397                         || !timeout.compareAndSetState(HashedWheelTimeout.ST_INIT, HashedWheelTimeout.ST_IN_BUCKET)) {
398                     // Was cancelled in the meantime. So just remove it and continue with next HashedWheelTimeout
399                     // in the queue
400                     timeout.remove();
401                     continue;
402                 }
403                 long calculated = timeout.deadline / tickDuration;
404                 long remainingRounds = (calculated - tick) / wheel.length;
405                 timeout.remainingRounds = remainingRounds;
406 
407                 final long ticks = Math.max(calculated, tick); // Ensure we don't schedule for past.
408                 int stopIndex = (int) (ticks & mask);
409 
410                 HashedWheelBucket bucket = wheel[stopIndex];
411                 bucket.addTimeout(timeout);
412             }
413         }
414         /**
415          * calculate goal nanoTime from startTime and current tick number,
416          * then wait until that goal has been reached.
417          * @return Long.MIN_VALUE if received a shutdown request,
418          * current time otherwise (with Long.MIN_VALUE changed by +1)
419          */
420         private long waitForNextTick() {
421             long deadline = tickDuration * (tick + 1);
422 
423             for (;;) {
424                 final long currentTime = System.nanoTime() - startTime;
425                 long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;
426 
427                 if (sleepTimeMs <= 0) {
428                     if (currentTime == Long.MIN_VALUE) {
429                         return -Long.MAX_VALUE;
430                     } else {
431                         return currentTime;
432                     }
433                 }
434 
435                 // Check if we run on windows, as if thats the case we will need
436                 // to round the sleepTime as workaround for a bug that only affect
437                 // the JVM if it runs on windows.
438                 //
439                 // See https://github.com/netty/netty/issues/356
440                 if (DetectionUtil.isWindows()) {
441                     sleepTimeMs = sleepTimeMs / 10 * 10;
442                 }
443 
444                 try {
445                     Thread.sleep(sleepTimeMs);
446                 } catch (InterruptedException e) {
447                     if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {
448                         return Long.MIN_VALUE;
449                     }
450                 }
451             }
452         }
453 
454         public Set<Timeout> unprocessedTimeouts() {
455             return Collections.unmodifiableSet(unprocessedTimeouts);
456         }
457     }
458 
459     private static final class HashedWheelTimeout implements Timeout {
460 
461         private static final int ST_INIT = 0;
462         private static final int ST_IN_BUCKET = 1;
463         private static final int ST_CANCELLED = 2;
464         private static final int ST_EXPIRED = 3;
465         private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATER =
466                 AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state");
467 
468         private final HashedWheelTimer timer;
469         private final TimerTask task;
470         private final long deadline;
471 
472         @SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization" })
473         private volatile int state = ST_INIT;
474 
475         // remainingRounds will be calculated and set by Worker.transferTimeoutsToBuckets() before the
476         // HashedWheelTimeout will be added to the correct HashedWheelBucket.
477         long remainingRounds;
478 
479         // This will be used to chain timeouts in HashedWheelTimerBucket via a double-linked-list.
480         // As only the workerThread will act on it there is no need for synchronization / volatile.
481         HashedWheelTimeout next;
482         HashedWheelTimeout prev;
483 
484         // The bucket to which the timeout was added
485         HashedWheelBucket bucket;
486 
487         HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) {
488             this.timer = timer;
489             this.task = task;
490             this.deadline = deadline;
491         }
492 
493         public Timer getTimer() {
494             return timer;
495         }
496 
497         public TimerTask getTask() {
498             return task;
499         }
500 
501         public void cancel() {
502             int state = state();
503             if (state >= ST_CANCELLED) {
504                 // fail fast if the task was cancelled or expired before.
505                 return;
506             }
507             if (state != ST_IN_BUCKET && compareAndSetState(ST_INIT, ST_CANCELLED)) {
508                 // Was cancelled before the HashedWheelTimeout was added to its HashedWheelBucket.
509                 // In this case we can just return here as it will be discarded by the WorkerThread when handling
510                 // the adding of HashedWheelTimeout to the HashedWheelBuckets.
511                 return;
512             }
513             // only update the state it will be removed from HashedWheelBucket on next tick.
514             if (!compareAndSetState(ST_IN_BUCKET, ST_CANCELLED)) {
515                 return;
516             }
517             // Add the HashedWheelTimeout back to the timeouts queue so it will be picked up on the next tick
518             // and remove this HashedTimeTask from the HashedWheelBucket. After this is done it is ready to get
519             // GC'ed once the user has no reference to it anymore.
520             timer.timeouts.add(this);
521         }
522 
523         public void remove() {
524             if (bucket != null) {
525                 bucket.remove(this);
526             }
527         }
528 
529         public boolean compareAndSetState(int expected, int state) {
530             return STATE_UPDATER.compareAndSet(this, expected, state);
531         }
532 
533         public int state() {
534             return state;
535         }
536 
537         public boolean isCancelled() {
538             return state == ST_CANCELLED;
539         }
540 
541         public boolean isExpired() {
542             return state > ST_IN_BUCKET;
543         }
544 
545         public HashedWheelTimeout value() {
546             return this;
547         }
548 
549         public void expire() {
550             if (!compareAndSetState(ST_IN_BUCKET, ST_EXPIRED)) {
551                 assert state() != ST_INIT;
552                 return;
553             }
554 
555             try {
556                 task.run(this);
557             } catch (Throwable t) {
558                 if (logger.isWarnEnabled()) {
559                     logger.warn("An exception was thrown by " + TimerTask.class.getSimpleName() + '.', t);
560                 }
561             }
562         }
563 
564         @Override
565         public String toString() {
566             final long currentTime = System.nanoTime();
567             long remaining = deadline - currentTime + timer.startTime;
568 
569             StringBuilder buf = new StringBuilder(192);
570             buf.append(getClass().getSimpleName());
571             buf.append('(');
572 
573             buf.append("deadline: ");
574             if (remaining > 0) {
575                 buf.append(remaining);
576                 buf.append(" ns later");
577             } else if (remaining < 0) {
578                 buf.append(-remaining);
579                 buf.append(" ns ago");
580             } else {
581                 buf.append("now");
582             }
583 
584             if (isCancelled()) {
585                 buf.append(", cancelled");
586             }
587 
588             buf.append(", task: ");
589             buf.append(getTask());
590 
591             return buf.append(')').toString();
592         }
593     }
594 
595     /**
596      * Bucket that stores HashedWheelTimeouts. These are stored in a linked-list like datastructure to allow easy
597      * removal of HashedWheelTimeouts in the middle. Also the HashedWheelTimeout act as nodes themself and so no
598      * extra object creation is needed.
599      */
600     private static final class HashedWheelBucket {
601 
602         // Used for the linked-list datastructure
603         private HashedWheelTimeout head;
604         private HashedWheelTimeout tail;
605 
606         /**
607          * Add {@link HashedWheelTimeout} to this bucket.
608          */
609         public void addTimeout(HashedWheelTimeout timeout) {
610             assert timeout.bucket == null;
611             timeout.bucket = this;
612             if (head == null) {
613                 head = tail = timeout;
614             } else {
615                 tail.next = timeout;
616                 timeout.prev = tail;
617                 tail = timeout;
618             }
619         }
620 
621         /**
622          * Expire all {@link HashedWheelTimeout}s for the given {@code deadline}.
623          */
624         public void expireTimeouts(long deadline) {
625             HashedWheelTimeout timeout = head;
626 
627             // process all timeouts
628             while (timeout != null) {
629                 boolean remove = false;
630                 if (timeout.remainingRounds <= 0) {
631                     if (timeout.deadline <= deadline) {
632                         timeout.expire();
633                     } else {
634                         // The timeout was placed into a wrong slot. This should never happen.
635                         throw new IllegalStateException(String.format(
636                                 "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
637                     }
638                     remove = true;
639                 } else if (timeout.isCancelled()) {
640                     remove = true;
641                 } else {
642                     timeout.remainingRounds --;
643                 }
644                 // store reference to next as we may null out timeout.next in the remove block.
645                 HashedWheelTimeout next = timeout.next;
646                 if (remove) {
647                     remove(timeout);
648                 }
649                 timeout = next;
650             }
651         }
652 
653         public void remove(HashedWheelTimeout timeout) {
654             HashedWheelTimeout next = timeout.next;
655             // remove timeout that was either processed or cancelled by updating the linked-list
656             if (timeout.prev != null) {
657                 timeout.prev.next = next;
658             }
659             if (timeout.next != null) {
660                 timeout.next.prev = timeout.prev;
661             }
662 
663             if (timeout == head) {
664                 // if timeout is also the tail we need to adjust the entry too
665                 if (timeout == tail) {
666                     tail = null;
667                     head = null;
668                 } else {
669                     head = next;
670                 }
671             } else if (timeout == tail) {
672                 // if the timeout is the tail modify the tail to be the prev node.
673                 tail = timeout.prev;
674             }
675             // null out prev, next and bucket to allow for GC.
676             timeout.prev = null;
677             timeout.next = null;
678             timeout.bucket = null;
679          }
680 
681         /**
682          * Clear this bucket and return all not expired / cancelled {@link Timeout}s.
683          */
684         public void clearTimeouts(Set<Timeout> set) {
685             for (;;) {
686                 HashedWheelTimeout timeout = pollTimeout();
687                 if (timeout == null) {
688                     return;
689                 }
690                 if (timeout.isExpired() || timeout.isCancelled()) {
691                     continue;
692                 }
693                 set.add(timeout);
694             }
695         }
696 
697         private HashedWheelTimeout pollTimeout() {
698             HashedWheelTimeout head = this.head;
699             if (head == null) {
700                 return null;
701             }
702             HashedWheelTimeout next = head.next;
703             if (next == null) {
704                 tail = this.head =  null;
705             } else {
706                 this.head = next;
707                 next.prev = null;
708             }
709 
710             // null out prev and next to allow for GC.
711             head.next = null;
712             head.prev = null;
713             return head;
714         }
715     }
716 }