View Javadoc

    1   /*
    2    * Copyright 2009 Red Hat, Inc.
    3    *
    4    * Red Hat licenses this file to you under the Apache License, version 2.0
    5    * (the "License"); you may not use this file except in compliance with the
    6    * 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 java.util.ArrayList;
    19  import java.util.Collections;
    20  import java.util.HashSet;
    21  import java.util.List;
    22  import java.util.Set;
    23  import java.util.concurrent.Executors;
    24  import java.util.concurrent.ThreadFactory;
    25  import java.util.concurrent.TimeUnit;
    26  import java.util.concurrent.atomic.AtomicBoolean;
    27  import java.util.concurrent.atomic.AtomicInteger;
    28  import java.util.concurrent.locks.ReadWriteLock;
    29  import java.util.concurrent.locks.ReentrantReadWriteLock;
    30
    31  import org.jboss.netty.channel.ChannelPipelineFactory;
    32  import org.jboss.netty.logging.InternalLogger;
    33  import org.jboss.netty.logging.InternalLoggerFactory;
    34  import org.jboss.netty.util.internal.ConcurrentIdentityHashMap;
    35  import org.jboss.netty.util.internal.ReusableIterator;
    36  import org.jboss.netty.util.internal.SharedResourceMisuseDetector;
    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 in
    68   * {@link ChannelPipelineFactory}, which results in the creation of a new thread
    69   * for every connection.
    70   *
    71   * <h3>Implementation Details</h3>
    72   *
    73   * {@link HashedWheelTimer} is based on
    74   * <a href="http://cseweb.ucsd.edu/users/varghese/">George Varghese</a> and
    75   * Tony Lauck's paper,
    76   * <a href="http://cseweb.ucsd.edu/users/varghese/PAPERS/twheel.ps.Z">'Hashed
    77   * and Hierarchical Timing Wheels: data structures to efficiently implement a
    78   * timer facility'</a>.  More comprehensive slides are located
    79   * <a href="http://www.cse.wustl.edu/~cdgill/courses/cs6874/TimingWheels.ppt">here</a>.
    80   *
    81   * @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
    82   * @author <a href="http://gleamynode.net/">Trustin Lee</a>
    83   * @version $Rev: 2297 $, $Date: 2010-06-07 10:50:02 +0900 (Mon, 07 Jun 2010) $
    84   */
    85  public class HashedWheelTimer implements Timer {
    86
    87      static final InternalLogger logger =
    88          InternalLoggerFactory.getInstance(HashedWheelTimer.class);
    89      private static final AtomicInteger id = new AtomicInteger();
    90
    91      private static final SharedResourceMisuseDetector misuseDetector =
    92          new SharedResourceMisuseDetector(HashedWheelTimer.class);
    93
    94      private final Worker worker = new Worker();
    95      final Thread workerThread;
    96      final AtomicBoolean shutdown = new AtomicBoolean();
    97
    98      private final long roundDuration;
    99      final long tickDuration;
    100     final Set<HashedWheelTimeout>[] wheel;
    101     final ReusableIterator<HashedWheelTimeout>[] iterators;
    102     final int mask;
    103     final ReadWriteLock lock = new ReentrantReadWriteLock();
    104     volatile int wheelCursor;
    105
    106     /**
    107      * Creates a new timer with the default thread factory
    108      * ({@link Executors#defaultThreadFactory()}), default tick duration, and
    109      * default number of ticks per wheel.
    110      */
    111     public HashedWheelTimer() {
    112         this(Executors.defaultThreadFactory());
    113     }
    114
    115     /**
    116      * Creates a new timer with the default thread factory
    117      * ({@link Executors#defaultThreadFactory()}) and default number of ticks
    118      * per wheel.
    119      *
    120      * @param tickDuration   the duration between tick
    121      * @param unit           the time unit of the {@code tickDuration}
    122      */
    123     public HashedWheelTimer(long tickDuration, TimeUnit unit) {
    124         this(Executors.defaultThreadFactory(), tickDuration, unit);
    125     }
    126
    127     /**
    128      * Creates a new timer with the default thread factory
    129      * ({@link Executors#defaultThreadFactory()}).
    130      *
    131      * @param tickDuration   the duration between tick
    132      * @param unit           the time unit of the {@code tickDuration}
    133      * @param ticksPerWheel  the size of the wheel
    134      */
    135     public HashedWheelTimer(long tickDuration, TimeUnit unit, int ticksPerWheel) {
    136         this(Executors.defaultThreadFactory(), tickDuration, unit, ticksPerWheel);
    137     }
    138
    139     /**
    140      * Creates a new timer with the default tick duration and default number of
    141      * ticks per wheel.
    142      *
    143      * @param threadFactory  a {@link ThreadFactory} that creates a
    144      *                       background {@link Thread} which is dedicated to
    145      *                       {@link TimerTask} execution.
    146      */
    147     public HashedWheelTimer(ThreadFactory threadFactory) {
    148         this(threadFactory, 100, TimeUnit.MILLISECONDS);
    149     }
    150
    151     /**
    152      * Creates a new timer with the default number of ticks per wheel.
    153      *
    154      * @param threadFactory  a {@link ThreadFactory} that creates a
    155      *                       background {@link Thread} which is dedicated to
    156      *                       {@link TimerTask} execution.
    157      * @param tickDuration   the duration between tick
    158      * @param unit           the time unit of the {@code tickDuration}
    159      */
    160     public HashedWheelTimer(
    161             ThreadFactory threadFactory, long tickDuration, TimeUnit unit) {
    162         this(threadFactory, tickDuration, unit, 512);
    163     }
    164
    165     /**
    166      * Creates a new timer.
    167      *
    168      * @param threadFactory  a {@link ThreadFactory} that creates a
    169      *                       background {@link Thread} which is dedicated to
    170      *                       {@link TimerTask} execution.
    171      * @param tickDuration   the duration between tick
    172      * @param unit           the time unit of the {@code tickDuration}
    173      * @param ticksPerWheel  the size of the wheel
    174      */
    175     public HashedWheelTimer(
    176             ThreadFactory threadFactory,
    177             long tickDuration, TimeUnit unit, int ticksPerWheel) {
    178
    179         if (threadFactory == null) {
    180             throw new NullPointerException("threadFactory");
    181         }
    182         if (unit == null) {
    183             throw new NullPointerException("unit");
    184         }
    185         if (tickDuration <= 0) {
    186             throw new IllegalArgumentException(
    187                     "tickDuration must be greater than 0: " + tickDuration);
    188         }
    189         if (ticksPerWheel <= 0) {
    190             throw new IllegalArgumentException(
    191                     "ticksPerWheel must be greater than 0: " + ticksPerWheel);
    192         }
    193
    194         // Normalize ticksPerWheel to power of two and initialize the wheel.
    195         wheel = createWheel(ticksPerWheel);
    196         iterators = createIterators(wheel);
    197         mask = wheel.length - 1;
    198
    199         // Convert tickDuration to milliseconds.
    200         this.tickDuration = tickDuration = unit.toMillis(tickDuration);
    201
    202         // Prevent overflow.
    203         if (tickDuration == Long.MAX_VALUE ||
    204                 tickDuration >= Long.MAX_VALUE / wheel.length) {
    205             throw new IllegalArgumentException(
    206                     "tickDuration is too long: " +
    207                     tickDuration +  ' ' + unit);
    208         }
    209
    210         roundDuration = tickDuration * wheel.length;
    211
    212         workerThread = threadFactory.newThread(new ThreadRenamingRunnable(
    213                         worker, "Hashed wheel timer #" + id.incrementAndGet()));
    214
    215         // Misuse check
    216         misuseDetector.increase();
    217     }
    218
    219     @SuppressWarnings("unchecked")
    220     private static Set<HashedWheelTimeout>[] createWheel(int ticksPerWheel) {
    221         if (ticksPerWheel <= 0) {
    222             throw new IllegalArgumentException(
    223                     "ticksPerWheel must be greater than 0: " + ticksPerWheel);
    224         }
    225         if (ticksPerWheel > 1073741824) {
    226             throw new IllegalArgumentException(
    227                     "ticksPerWheel may not be greater than 2^30: " + ticksPerWheel);
    228         }
    229
    230         ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);
    231         Set<HashedWheelTimeout>[] wheel = new Set[ticksPerWheel];
    232         for (int i = 0; i < wheel.length; i ++) {
    233             wheel[i] = new MapBackedSet<HashedWheelTimeout>(
    234                     new ConcurrentIdentityHashMap<HashedWheelTimeout, Boolean>(16, 0.95f, 4));
    235         }
    236         return wheel;
    237     }
    238
    239     @SuppressWarnings("unchecked")
    240     private static ReusableIterator<HashedWheelTimeout>[] createIterators(Set<HashedWheelTimeout>[] wheel) {
    241         ReusableIterator<HashedWheelTimeout>[] iterators = new ReusableIterator[wheel.length];
    242         for (int i = 0; i < wheel.length; i ++) {
    243             iterators[i] = (ReusableIterator<HashedWheelTimeout>) wheel[i].iterator();
    244         }
    245         return iterators;
    246     }
    247
    248     private static int normalizeTicksPerWheel(int ticksPerWheel) {
    249         int normalizedTicksPerWheel = 1;
    250         while (normalizedTicksPerWheel < ticksPerWheel) {
    251             normalizedTicksPerWheel <<= 1;
    252         }
    253         return normalizedTicksPerWheel;
    254     }
    255
    256     /**
    257      * Starts the background thread explicitly.  The background thread will
    258      * start automatically on demand even if you did not call this method.
    259      *
    260      * @throws IllegalStateException if this timer has been
    261      *                               {@linkplain #stop() stopped} already
    262      */
    263     public synchronized void start() {
    264         if (shutdown.get()) {
    265             throw new IllegalStateException("cannot be started once stopped");
    266         }
    267
    268         if (!workerThread.isAlive()) {
    269             workerThread.start();
    270         }
    271     }
    272
    273     public synchronized Set<Timeout> stop() {
    274         if (!shutdown.compareAndSet(false, true)) {
    275             return Collections.emptySet();
    276         }
    277
    278         boolean interrupted = false;
    279         while (workerThread.isAlive()) {
    280             workerThread.interrupt();
    281             try {
    282                 workerThread.join(100);
    283             } catch (InterruptedException e) {
    284                 interrupted = true;
    285             }
    286         }
    287
    288         if (interrupted) {
    289             Thread.currentThread().interrupt();
    290         }
    291
    292         misuseDetector.decrease();
    293
    294         Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();
    295         for (Set<HashedWheelTimeout> bucket: wheel) {
    296             unprocessedTimeouts.addAll(bucket);
    297             bucket.clear();
    298         }
    299
    300         return Collections.unmodifiableSet(unprocessedTimeouts);
    301     }
    302
    303     public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
    304         final long currentTime = System.currentTimeMillis();
    305
    306         if (task == null) {
    307             throw new NullPointerException("task");
    308         }
    309         if (unit == null) {
    310             throw new NullPointerException("unit");
    311         }
    312
    313         delay = unit.toMillis(delay);
    314         if (delay < tickDuration) {
    315             delay = tickDuration;
    316         }
    317
    318         if (!workerThread.isAlive()) {
    319             start();
    320         }
    321
    322         // Prepare the required parameters to create the timeout object.
    323         HashedWheelTimeout timeout;
    324         final long lastRoundDelay = delay % roundDuration;
    325         final long lastTickDelay = delay % tickDuration;
    326         final long relativeIndex =
    327             lastRoundDelay / tickDuration + (lastTickDelay != 0? 1 : 0);
    328         final long deadline = currentTime + delay;
    329
    330         final long remainingRounds =
    331             delay / roundDuration - (delay % roundDuration == 0? 1 : 0);
    332
    333         // Add the timeout to the wheel.
    334         lock.readLock().lock();
    335         try {
    336             timeout =
    337                 new HashedWheelTimeout(
    338                         task, deadline,
    339                         (int) (wheelCursor + relativeIndex & mask),
    340                         remainingRounds);
    341
    342             wheel[timeout.stopIndex].add(timeout);
    343         } finally {
    344             lock.readLock().unlock();
    345         }
    346
    347         return timeout;
    348     }
    349
    350     private final class Worker implements Runnable {
    351
    352         private long startTime;
    353         private long tick;
    354
    355         Worker() {
    356             super();
    357         }
    358
    359         public void run() {
    360             List<HashedWheelTimeout> expiredTimeouts =
    361                 new ArrayList<HashedWheelTimeout>();
    362
    363             startTime = System.currentTimeMillis();
    364             tick = 1;
    365
    366             while (!shutdown.get()) {
    367                 waitForNextTick();
    368                 fetchExpiredTimeouts(expiredTimeouts);
    369                 notifyExpiredTimeouts(expiredTimeouts);
    370             }
    371         }
    372
    373         private void fetchExpiredTimeouts(
    374                 List<HashedWheelTimeout> expiredTimeouts) {
    375
    376             // Find the expired timeouts and decrease the round counter
    377             // if necessary.  Note that we don't send the notification
    378             // immediately to make sure the listeners are called without
    379             // an exclusive lock.
    380             lock.writeLock().lock();
    381             try {
    382                 int oldBucketHead = wheelCursor;
    383                 int newBucketHead = oldBucketHead + 1 & mask;
    384                 wheelCursor = newBucketHead;
    385
    386                 ReusableIterator<HashedWheelTimeout> i = iterators[oldBucketHead];
    387                 fetchExpiredTimeouts(expiredTimeouts, i);
    388             } finally {
    389                 lock.writeLock().unlock();
    390             }
    391         }
    392
    393         private void fetchExpiredTimeouts(
    394                 List<HashedWheelTimeout> expiredTimeouts,
    395                 ReusableIterator<HashedWheelTimeout> i) {
    396
    397             long currentDeadline = System.currentTimeMillis() + tickDuration;
    398             i.rewind();
    399             while (i.hasNext()) {
    400                 HashedWheelTimeout timeout = i.next();
    401                 if (timeout.remainingRounds <= 0) {
    402                     if (timeout.deadline < currentDeadline) {
    403                         i.remove();
    404                         expiredTimeouts.add(timeout);
    405                     } else {
    406                         // A rare case where a timeout is put for the next
    407                         // round: just wait for the next round.
    408                     }
    409                 } else {
    410                     timeout.remainingRounds --;
    411                 }
    412             }
    413         }
    414
    415         private void notifyExpiredTimeouts(
    416                 List<HashedWheelTimeout> expiredTimeouts) {
    417             // Notify the expired timeouts.
    418             for (int i = expiredTimeouts.size() - 1; i >= 0; i --) {
    419                 expiredTimeouts.get(i).expire();
    420             }
    421
    422             // Clean up the temporary list.
    423             expiredTimeouts.clear();
    424         }
    425
    426         private void waitForNextTick() {
    427             for (;;) {
    428                 final long currentTime = System.currentTimeMillis();
    429                 final long sleepTime = tickDuration * tick - (currentTime - startTime);
    430
    431                 if (sleepTime <= 0) {
    432                     break;
    433                 }
    434
    435                 try {
    436                     Thread.sleep(sleepTime);
    437                 } catch (InterruptedException e) {
    438                     if (shutdown.get()) {
    439                         return;
    440                     }
    441                 }
    442             }
    443
    444             // Reset the tick if overflow is expected.
    445             if (tickDuration * tick > Long.MAX_VALUE - tickDuration) {
    446                 startTime = System.currentTimeMillis();
    447                 tick = 1;
    448             } else {
    449                 // Increase the tick if overflow is not likely to happen.
    450                 tick ++;
    451             }
    452         }
    453     }
    454
    455     private final class HashedWheelTimeout implements Timeout {
    456
    457         private final TimerTask task;
    458         final int stopIndex;
    459         final long deadline;
    460         volatile long remainingRounds;
    461         private volatile boolean cancelled;
    462
    463         HashedWheelTimeout(
    464                 TimerTask task, long deadline, int stopIndex, long remainingRounds) {
    465             this.task = task;
    466             this.deadline = deadline;
    467             this.stopIndex = stopIndex;
    468             this.remainingRounds = remainingRounds;
    469         }
    470
    471         public Timer getTimer() {
    472             return HashedWheelTimer.this;
    473         }
    474
    475         public TimerTask getTask() {
    476             return task;
    477         }
    478
    479         public void cancel() {
    480             if (isExpired()) {
    481                 return;
    482             }
    483
    484             cancelled = true;
    485
    486             // Might be called more than once, but doesn't matter.
    487             wheel[stopIndex].remove(this);
    488         }
    489
    490         public boolean isCancelled() {
    491             return cancelled;
    492         }
    493
    494         public boolean isExpired() {
    495             return cancelled || System.currentTimeMillis() > deadline;
    496         }
    497
    498         public void expire() {
    499             if (cancelled) {
    500                 return;
    501             }
    502
    503             try {
    504                 task.run(this);
    505             } catch (Throwable t) {
    506                 logger.warn(
    507                         "An exception was thrown by " +
    508                         TimerTask.class.getSimpleName() + ".", t);
    509             }
    510         }
    511
    512         @Override
    513         public String toString() {
    514             long currentTime = System.currentTimeMillis();
    515             long remaining = deadline - currentTime;
    516
    517             StringBuilder buf = new StringBuilder(192);
    518             buf.append(getClass().getSimpleName());
    519             buf.append('(');
    520
    521             buf.append("deadline: ");
    522             if (remaining > 0) {
    523                 buf.append(remaining);
    524                 buf.append(" ms later, ");
    525             } else if (remaining < 0) {
    526                 buf.append(-remaining);
    527                 buf.append(" ms ago, ");
    528             } else {
    529                 buf.append("now, ");
    530             }
    531
    532             if (isCancelled()) {
    533                 buf.append (", cancelled");
    534             }
    535
    536             return buf.append(')').toString();
    537         }
    538     }
    539 }