2b401d4587
Motivation: At the moment the HashedWheelTimer will only remove the cancelled Timeouts once the HashedWheelBucket is processed again. Until this the instance will not be able to be GC'ed as there are still strong referenced to it even if the user not reference it by himself/herself. This can cause to waste a lot of memory even if the Timeout was cancelled before. Modification: Add a new queue which holds CancelTasks that will be processed on each tick to remove cancelled Timeouts. Because all of this is done only by the WorkerThread there is no need for synchronization and only one extra object creation is needed when cancel() is executed. For addTimeout(...) no new overhead is introduced. Result: Less memory usage for cancelled Timeouts.
717 lines
27 KiB
Java
717 lines
27 KiB
Java
/*
|
|
* Copyright 2012 The Netty Project
|
|
*
|
|
* The Netty Project licenses this file to you under the Apache License,
|
|
* version 2.0 (the "License"); you may not use this file except in compliance
|
|
* with the License. You may obtain a copy of the License at:
|
|
*
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
*
|
|
* Unless required by applicable law or agreed to in writing, software
|
|
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
* License for the specific language governing permissions and limitations
|
|
* under the License.
|
|
*/
|
|
package org.jboss.netty.util;
|
|
|
|
import org.jboss.netty.channel.ChannelPipelineFactory;
|
|
import org.jboss.netty.logging.InternalLogger;
|
|
import org.jboss.netty.logging.InternalLoggerFactory;
|
|
import org.jboss.netty.util.internal.DetectionUtil;
|
|
import org.jboss.netty.util.internal.SharedResourceMisuseDetector;
|
|
|
|
import java.util.Collections;
|
|
import java.util.HashSet;
|
|
import java.util.Queue;
|
|
import java.util.Set;
|
|
import java.util.concurrent.ConcurrentLinkedQueue;
|
|
import java.util.concurrent.CountDownLatch;
|
|
import java.util.concurrent.Executors;
|
|
import java.util.concurrent.ThreadFactory;
|
|
import java.util.concurrent.TimeUnit;
|
|
import java.util.concurrent.atomic.AtomicInteger;
|
|
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
|
|
|
|
/**
|
|
* A {@link Timer} optimized for approximated I/O timeout scheduling.
|
|
*
|
|
* <h3>Tick Duration</h3>
|
|
*
|
|
* As described with 'approximated', this timer does not execute the scheduled
|
|
* {@link TimerTask} on time. {@link HashedWheelTimer}, on every tick, will
|
|
* check if there are any {@link TimerTask}s behind the schedule and execute
|
|
* them.
|
|
* <p>
|
|
* You can increase or decrease the accuracy of the execution timing by
|
|
* specifying smaller or larger tick duration in the constructor. In most
|
|
* network applications, I/O timeout does not need to be accurate. Therefore,
|
|
* the default tick duration is 100 milliseconds and you will not need to try
|
|
* different configurations in most cases.
|
|
*
|
|
* <h3>Ticks per Wheel (Wheel Size)</h3>
|
|
*
|
|
* {@link HashedWheelTimer} maintains a data structure called 'wheel'.
|
|
* To put simply, a wheel is a hash table of {@link TimerTask}s whose hash
|
|
* function is 'dead line of the task'. The default number of ticks per wheel
|
|
* (i.e. the size of the wheel) is 512. You could specify a larger value
|
|
* if you are going to schedule a lot of timeouts.
|
|
*
|
|
* <h3>Do not create many instances.</h3>
|
|
*
|
|
* {@link HashedWheelTimer} creates a new thread whenever it is instantiated and
|
|
* started. Therefore, you should make sure to create only one instance and
|
|
* share it across your application. One of the common mistakes, that makes
|
|
* your application unresponsive, is to create a new instance in
|
|
* {@link ChannelPipelineFactory}, which results in the creation of a new thread
|
|
* for every connection.
|
|
*
|
|
* <h3>Implementation Details</h3>
|
|
*
|
|
* {@link HashedWheelTimer} is based on
|
|
* <a href="http://cseweb.ucsd.edu/users/varghese/">George Varghese</a> and
|
|
* Tony Lauck's paper,
|
|
* <a href="http://cseweb.ucsd.edu/users/varghese/PAPERS/twheel.ps.Z">'Hashed
|
|
* and Hierarchical Timing Wheels: data structures to efficiently implement a
|
|
* timer facility'</a>. More comprehensive slides are located
|
|
* <a href="http://www.cse.wustl.edu/~cdgill/courses/cs6874/TimingWheels.ppt">here</a>.
|
|
*/
|
|
public class HashedWheelTimer implements Timer {
|
|
|
|
static final InternalLogger logger =
|
|
InternalLoggerFactory.getInstance(HashedWheelTimer.class);
|
|
private static final AtomicInteger id = new AtomicInteger();
|
|
|
|
private static final SharedResourceMisuseDetector misuseDetector =
|
|
new SharedResourceMisuseDetector(HashedWheelTimer.class);
|
|
|
|
private static final AtomicIntegerFieldUpdater<HashedWheelTimer> WORKER_STATE_UPDATER =
|
|
AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimer.class, "workerState");
|
|
|
|
private final Worker worker = new Worker();
|
|
private final Thread workerThread;
|
|
|
|
public static final int WORKER_STATE_INIT = 0;
|
|
public static final int WORKER_STATE_STARTED = 1;
|
|
public static final int WORKER_STATE_SHUTDOWN = 2;
|
|
@SuppressWarnings({ "unused", "FieldMayBeFinal", "RedundantFieldInitialization" })
|
|
private volatile int workerState = WORKER_STATE_INIT; // 0 - init, 1 - started, 2 - shut down
|
|
|
|
private final long tickDuration;
|
|
private final HashedWheelBucket[] wheel;
|
|
private final int mask;
|
|
private final CountDownLatch startTimeInitialized = new CountDownLatch(1);
|
|
private final Queue<HashedWheelTimeout> timeouts = new ConcurrentLinkedQueue<HashedWheelTimeout>();
|
|
private volatile long startTime;
|
|
|
|
/**
|
|
* Creates a new timer with the default thread factory
|
|
* ({@link Executors#defaultThreadFactory()}), default tick duration, and
|
|
* default number of ticks per wheel.
|
|
*/
|
|
public HashedWheelTimer() {
|
|
this(Executors.defaultThreadFactory());
|
|
}
|
|
|
|
/**
|
|
* Creates a new timer with the default thread factory
|
|
* ({@link Executors#defaultThreadFactory()}) and default number of ticks
|
|
* per wheel.
|
|
*
|
|
* @param tickDuration the duration between tick
|
|
* @param unit the time unit of the {@code tickDuration}
|
|
*/
|
|
public HashedWheelTimer(long tickDuration, TimeUnit unit) {
|
|
this(Executors.defaultThreadFactory(), tickDuration, unit);
|
|
}
|
|
|
|
/**
|
|
* Creates a new timer with the default thread factory
|
|
* ({@link Executors#defaultThreadFactory()}).
|
|
*
|
|
* @param tickDuration the duration between tick
|
|
* @param unit the time unit of the {@code tickDuration}
|
|
* @param ticksPerWheel the size of the wheel
|
|
*/
|
|
public HashedWheelTimer(long tickDuration, TimeUnit unit, int ticksPerWheel) {
|
|
this(Executors.defaultThreadFactory(), tickDuration, unit, ticksPerWheel);
|
|
}
|
|
|
|
/**
|
|
* Creates a new timer with the default tick duration and default number of
|
|
* ticks per wheel.
|
|
*
|
|
* @param threadFactory a {@link ThreadFactory} that creates a
|
|
* background {@link Thread} which is dedicated to
|
|
* {@link TimerTask} execution.
|
|
*/
|
|
public HashedWheelTimer(ThreadFactory threadFactory) {
|
|
this(threadFactory, 100, TimeUnit.MILLISECONDS);
|
|
}
|
|
|
|
/**
|
|
* Creates a new timer with the default number of ticks per wheel.
|
|
*
|
|
* @param threadFactory a {@link ThreadFactory} that creates a
|
|
* background {@link Thread} which is dedicated to
|
|
* {@link TimerTask} execution.
|
|
* @param tickDuration the duration between tick
|
|
* @param unit the time unit of the {@code tickDuration}
|
|
*/
|
|
public HashedWheelTimer(
|
|
ThreadFactory threadFactory, long tickDuration, TimeUnit unit) {
|
|
this(threadFactory, tickDuration, unit, 512);
|
|
}
|
|
|
|
/**
|
|
* Creates a new timer.
|
|
*
|
|
* @param threadFactory a {@link ThreadFactory} that creates a
|
|
* background {@link Thread} which is dedicated to
|
|
* {@link TimerTask} execution.
|
|
* @param tickDuration the duration between tick
|
|
* @param unit the time unit of the {@code tickDuration}
|
|
* @param ticksPerWheel the size of the wheel
|
|
*/
|
|
public HashedWheelTimer(
|
|
ThreadFactory threadFactory,
|
|
long tickDuration, TimeUnit unit, int ticksPerWheel) {
|
|
this(threadFactory, null, tickDuration, unit, ticksPerWheel);
|
|
}
|
|
|
|
/**
|
|
* Creates a new timer.
|
|
*
|
|
* @param threadFactory a {@link ThreadFactory} that creates a
|
|
* background {@link Thread} which is dedicated to
|
|
* {@link TimerTask} execution.
|
|
* @param determiner thread name determiner to control thread name.
|
|
* @param tickDuration the duration between tick
|
|
* @param unit the time unit of the {@code tickDuration}
|
|
* @param ticksPerWheel the size of the wheel
|
|
*/
|
|
public HashedWheelTimer(
|
|
ThreadFactory threadFactory,
|
|
ThreadNameDeterminer determiner,
|
|
long tickDuration, TimeUnit unit, int ticksPerWheel) {
|
|
|
|
if (threadFactory == null) {
|
|
throw new NullPointerException("threadFactory");
|
|
}
|
|
if (unit == null) {
|
|
throw new NullPointerException("unit");
|
|
}
|
|
if (tickDuration <= 0) {
|
|
throw new IllegalArgumentException(
|
|
"tickDuration must be greater than 0: " + tickDuration);
|
|
}
|
|
if (ticksPerWheel <= 0) {
|
|
throw new IllegalArgumentException(
|
|
"ticksPerWheel must be greater than 0: " + ticksPerWheel);
|
|
}
|
|
|
|
// Normalize ticksPerWheel to power of two and initialize the wheel.
|
|
wheel = createWheel(ticksPerWheel);
|
|
mask = wheel.length - 1;
|
|
|
|
// Convert tickDuration to nanos.
|
|
this.tickDuration = unit.toNanos(tickDuration);
|
|
|
|
// Prevent overflow.
|
|
if (this.tickDuration >= Long.MAX_VALUE / wheel.length) {
|
|
throw new IllegalArgumentException(String.format(
|
|
"tickDuration: %d (expected: 0 < tickDuration in nanos < %d",
|
|
tickDuration, Long.MAX_VALUE / wheel.length));
|
|
}
|
|
|
|
workerThread = threadFactory.newThread(new ThreadRenamingRunnable(
|
|
worker, "Hashed wheel timer #" + id.incrementAndGet(),
|
|
determiner));
|
|
|
|
// Misuse check
|
|
misuseDetector.increase();
|
|
}
|
|
|
|
@SuppressWarnings("unchecked")
|
|
private static HashedWheelBucket[] createWheel(int ticksPerWheel) {
|
|
if (ticksPerWheel <= 0) {
|
|
throw new IllegalArgumentException(
|
|
"ticksPerWheel must be greater than 0: " + ticksPerWheel);
|
|
}
|
|
if (ticksPerWheel > 1073741824) {
|
|
throw new IllegalArgumentException(
|
|
"ticksPerWheel may not be greater than 2^30: " + ticksPerWheel);
|
|
}
|
|
|
|
ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);
|
|
HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel];
|
|
for (int i = 0; i < wheel.length; i ++) {
|
|
wheel[i] = new HashedWheelBucket();
|
|
}
|
|
return wheel;
|
|
}
|
|
|
|
private static int normalizeTicksPerWheel(int ticksPerWheel) {
|
|
int normalizedTicksPerWheel = 1;
|
|
while (normalizedTicksPerWheel < ticksPerWheel) {
|
|
normalizedTicksPerWheel <<= 1;
|
|
}
|
|
return normalizedTicksPerWheel;
|
|
}
|
|
|
|
/**
|
|
* Starts the background thread explicitly. The background thread will
|
|
* start automatically on demand even if you did not call this method.
|
|
*
|
|
* @throws IllegalStateException if this timer has been
|
|
* {@linkplain #stop() stopped} already
|
|
*/
|
|
public void start() {
|
|
switch (WORKER_STATE_UPDATER.get(this)) {
|
|
case WORKER_STATE_INIT:
|
|
if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
|
|
workerThread.start();
|
|
}
|
|
break;
|
|
case WORKER_STATE_STARTED:
|
|
break;
|
|
case WORKER_STATE_SHUTDOWN:
|
|
throw new IllegalStateException("cannot be started once stopped");
|
|
default:
|
|
throw new Error("Invalid WorkerState");
|
|
}
|
|
|
|
// Wait until the startTime is initialized by the worker.
|
|
while (startTime == 0) {
|
|
try {
|
|
startTimeInitialized.await();
|
|
} catch (InterruptedException ignore) {
|
|
// Ignore - it will be ready very soon.
|
|
}
|
|
}
|
|
}
|
|
|
|
public Set<Timeout> stop() {
|
|
if (Thread.currentThread() == workerThread) {
|
|
throw new IllegalStateException(
|
|
HashedWheelTimer.class.getSimpleName() +
|
|
".stop() cannot be called from " +
|
|
TimerTask.class.getSimpleName());
|
|
}
|
|
|
|
if (!WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)) {
|
|
// workerState can be 0 or 2 at this moment - let it always be 2.
|
|
WORKER_STATE_UPDATER.set(this, WORKER_STATE_SHUTDOWN);
|
|
|
|
misuseDetector.decrease();
|
|
|
|
return Collections.emptySet();
|
|
}
|
|
|
|
boolean interrupted = false;
|
|
while (workerThread.isAlive()) {
|
|
workerThread.interrupt();
|
|
try {
|
|
workerThread.join(100);
|
|
} catch (InterruptedException e) {
|
|
interrupted = true;
|
|
}
|
|
}
|
|
|
|
if (interrupted) {
|
|
Thread.currentThread().interrupt();
|
|
}
|
|
|
|
misuseDetector.decrease();
|
|
|
|
return worker.unprocessedTimeouts();
|
|
}
|
|
|
|
public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
|
|
if (task == null) {
|
|
throw new NullPointerException("task");
|
|
}
|
|
if (unit == null) {
|
|
throw new NullPointerException("unit");
|
|
}
|
|
start();
|
|
|
|
// Add the timeout to the timeout queue which will be processed on the next tick.
|
|
// During processing all the queued HashedWheelTimeouts will be added to the correct HashedWheelBucket.
|
|
long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
|
|
HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
|
|
timeouts.add(timeout);
|
|
return timeout;
|
|
}
|
|
|
|
private final class Worker implements Runnable {
|
|
private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();
|
|
|
|
private long tick;
|
|
|
|
public void run() {
|
|
// Initialize the startTime.
|
|
startTime = System.nanoTime();
|
|
if (startTime == 0) {
|
|
// We use 0 as an indicator for the uninitialized value here, so make sure it's not 0 when initialized.
|
|
startTime = 1;
|
|
}
|
|
|
|
// Notify the other threads waiting for the initialization at start().
|
|
startTimeInitialized.countDown();
|
|
|
|
do {
|
|
final long deadline = waitForNextTick();
|
|
if (deadline > 0) {
|
|
transferTimeoutsToBuckets();
|
|
HashedWheelBucket bucket =
|
|
wheel[(int) (tick & mask)];
|
|
bucket.expireTimeouts(deadline);
|
|
tick++;
|
|
}
|
|
} while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);
|
|
|
|
// Fill the unprocessedTimeouts so we can return them from stop() method.
|
|
for (HashedWheelBucket bucket: wheel) {
|
|
bucket.clearTimeouts(unprocessedTimeouts);
|
|
}
|
|
for (;;) {
|
|
HashedWheelTimeout timeout = timeouts.poll();
|
|
if (timeout == null) {
|
|
break;
|
|
}
|
|
unprocessedTimeouts.add(timeout);
|
|
}
|
|
}
|
|
|
|
private void transferTimeoutsToBuckets() {
|
|
// transfer only max. 100000 timeouts per tick to prevent a thread to stale the workerThread when it just
|
|
// adds new timeouts in a loop.
|
|
for (int i = 0; i < 100000; i++) {
|
|
HashedWheelTimeout timeout = timeouts.poll();
|
|
if (timeout == null) {
|
|
// all processed
|
|
break;
|
|
}
|
|
if (timeout.state() == HashedWheelTimeout.ST_CANCELLED
|
|
|| !timeout.compareAndSetState(HashedWheelTimeout.ST_INIT, HashedWheelTimeout.ST_IN_BUCKET)) {
|
|
// Was cancelled in the meantime. So just remove it and continue with next HashedWheelTimeout
|
|
// in the queue
|
|
timeout.remove();
|
|
continue;
|
|
}
|
|
long calculated = timeout.deadline / tickDuration;
|
|
long remainingRounds = (calculated - tick) / wheel.length;
|
|
timeout.remainingRounds = remainingRounds;
|
|
|
|
final long ticks = Math.max(calculated, tick); // Ensure we don't schedule for past.
|
|
int stopIndex = (int) (ticks & mask);
|
|
|
|
HashedWheelBucket bucket = wheel[stopIndex];
|
|
bucket.addTimeout(timeout);
|
|
}
|
|
}
|
|
/**
|
|
* calculate goal nanoTime from startTime and current tick number,
|
|
* then wait until that goal has been reached.
|
|
* @return Long.MIN_VALUE if received a shutdown request,
|
|
* current time otherwise (with Long.MIN_VALUE changed by +1)
|
|
*/
|
|
private long waitForNextTick() {
|
|
long deadline = tickDuration * (tick + 1);
|
|
|
|
for (;;) {
|
|
final long currentTime = System.nanoTime() - startTime;
|
|
long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;
|
|
|
|
if (sleepTimeMs <= 0) {
|
|
if (currentTime == Long.MIN_VALUE) {
|
|
return -Long.MAX_VALUE;
|
|
} else {
|
|
return currentTime;
|
|
}
|
|
}
|
|
|
|
// Check if we run on windows, as if thats the case we will need
|
|
// to round the sleepTime as workaround for a bug that only affect
|
|
// the JVM if it runs on windows.
|
|
//
|
|
// See https://github.com/netty/netty/issues/356
|
|
if (DetectionUtil.isWindows()) {
|
|
sleepTimeMs = sleepTimeMs / 10 * 10;
|
|
}
|
|
|
|
try {
|
|
Thread.sleep(sleepTimeMs);
|
|
} catch (InterruptedException e) {
|
|
if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {
|
|
return Long.MIN_VALUE;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
public Set<Timeout> unprocessedTimeouts() {
|
|
return Collections.unmodifiableSet(unprocessedTimeouts);
|
|
}
|
|
}
|
|
|
|
private static final class HashedWheelTimeout implements Timeout {
|
|
|
|
private static final int ST_INIT = 0;
|
|
private static final int ST_IN_BUCKET = 1;
|
|
private static final int ST_CANCELLED = 2;
|
|
private static final int ST_EXPIRED = 3;
|
|
private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATER =
|
|
AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state");
|
|
|
|
private final HashedWheelTimer timer;
|
|
private final TimerTask task;
|
|
private final long deadline;
|
|
|
|
@SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization" })
|
|
private volatile int state = ST_INIT;
|
|
|
|
// remainingRounds will be calculated and set by Worker.transferTimeoutsToBuckets() before the
|
|
// HashedWheelTimeout will be added to the correct HashedWheelBucket.
|
|
long remainingRounds;
|
|
|
|
// This will be used to chain timeouts in HashedWheelTimerBucket via a double-linked-list.
|
|
// As only the workerThread will act on it there is no need for synchronization / volatile.
|
|
HashedWheelTimeout next;
|
|
HashedWheelTimeout prev;
|
|
|
|
// The bucket to which the timeout was added
|
|
HashedWheelBucket bucket;
|
|
|
|
HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) {
|
|
this.timer = timer;
|
|
this.task = task;
|
|
this.deadline = deadline;
|
|
}
|
|
|
|
public Timer getTimer() {
|
|
return timer;
|
|
}
|
|
|
|
public TimerTask getTask() {
|
|
return task;
|
|
}
|
|
|
|
public void cancel() {
|
|
int state = state();
|
|
if (state >= ST_CANCELLED) {
|
|
// fail fast if the task was cancelled or expired before.
|
|
return;
|
|
}
|
|
if (state != ST_IN_BUCKET && compareAndSetState(ST_INIT, ST_CANCELLED)) {
|
|
// Was cancelled before the HashedWheelTimeout was added to its HashedWheelBucket.
|
|
// In this case we can just return here as it will be discarded by the WorkerThread when handling
|
|
// the adding of HashedWheelTimeout to the HashedWheelBuckets.
|
|
return;
|
|
}
|
|
// only update the state it will be removed from HashedWheelBucket on next tick.
|
|
if (!compareAndSetState(ST_IN_BUCKET, ST_CANCELLED)) {
|
|
return;
|
|
}
|
|
// Add the HashedWheelTimeout back to the timeouts queue so it will be picked up on the next tick
|
|
// and remove this HashedTimeTask from the HashedWheelBucket. After this is done it is ready to get
|
|
// GC'ed once the user has no reference to it anymore.
|
|
timer.timeouts.add(this);
|
|
}
|
|
|
|
public void remove() {
|
|
if (bucket != null) {
|
|
bucket.remove(this);
|
|
}
|
|
}
|
|
|
|
public boolean compareAndSetState(int expected, int state) {
|
|
return STATE_UPDATER.compareAndSet(this, expected, state);
|
|
}
|
|
|
|
public int state() {
|
|
return state;
|
|
}
|
|
|
|
public boolean isCancelled() {
|
|
return state == ST_CANCELLED;
|
|
}
|
|
|
|
public boolean isExpired() {
|
|
return state > ST_IN_BUCKET;
|
|
}
|
|
|
|
public HashedWheelTimeout value() {
|
|
return this;
|
|
}
|
|
|
|
public void expire() {
|
|
if (!compareAndSetState(ST_IN_BUCKET, ST_EXPIRED)) {
|
|
assert state() != ST_INIT;
|
|
return;
|
|
}
|
|
|
|
try {
|
|
task.run(this);
|
|
} catch (Throwable t) {
|
|
if (logger.isWarnEnabled()) {
|
|
logger.warn("An exception was thrown by " + TimerTask.class.getSimpleName() + '.', t);
|
|
}
|
|
}
|
|
}
|
|
|
|
@Override
|
|
public String toString() {
|
|
final long currentTime = System.nanoTime();
|
|
long remaining = deadline - currentTime + timer.startTime;
|
|
|
|
StringBuilder buf = new StringBuilder(192);
|
|
buf.append(getClass().getSimpleName());
|
|
buf.append('(');
|
|
|
|
buf.append("deadline: ");
|
|
if (remaining > 0) {
|
|
buf.append(remaining);
|
|
buf.append(" ns later");
|
|
} else if (remaining < 0) {
|
|
buf.append(-remaining);
|
|
buf.append(" ns ago");
|
|
} else {
|
|
buf.append("now");
|
|
}
|
|
|
|
if (isCancelled()) {
|
|
buf.append(", cancelled");
|
|
}
|
|
|
|
buf.append(", task: ");
|
|
buf.append(getTask());
|
|
|
|
return buf.append(')').toString();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Bucket that stores HashedWheelTimeouts. These are stored in a linked-list like datastructure to allow easy
|
|
* removal of HashedWheelTimeouts in the middle. Also the HashedWheelTimeout act as nodes themself and so no
|
|
* extra object creation is needed.
|
|
*/
|
|
private static final class HashedWheelBucket {
|
|
|
|
// Used for the linked-list datastructure
|
|
private HashedWheelTimeout head;
|
|
private HashedWheelTimeout tail;
|
|
|
|
/**
|
|
* Add {@link HashedWheelTimeout} to this bucket.
|
|
*/
|
|
public void addTimeout(HashedWheelTimeout timeout) {
|
|
assert timeout.bucket == null;
|
|
timeout.bucket = this;
|
|
if (head == null) {
|
|
head = tail = timeout;
|
|
} else {
|
|
tail.next = timeout;
|
|
timeout.prev = tail;
|
|
tail = timeout;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Expire all {@link HashedWheelTimeout}s for the given {@code deadline}.
|
|
*/
|
|
public void expireTimeouts(long deadline) {
|
|
HashedWheelTimeout timeout = head;
|
|
|
|
// process all timeouts
|
|
while (timeout != null) {
|
|
boolean remove = false;
|
|
if (timeout.remainingRounds <= 0) {
|
|
if (timeout.deadline <= deadline) {
|
|
timeout.expire();
|
|
} else {
|
|
// The timeout was placed into a wrong slot. This should never happen.
|
|
throw new IllegalStateException(String.format(
|
|
"timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
|
|
}
|
|
remove = true;
|
|
} else if (timeout.isCancelled()) {
|
|
remove = true;
|
|
} else {
|
|
timeout.remainingRounds --;
|
|
}
|
|
// store reference to next as we may null out timeout.next in the remove block.
|
|
HashedWheelTimeout next = timeout.next;
|
|
if (remove) {
|
|
remove(timeout);
|
|
}
|
|
timeout = next;
|
|
}
|
|
}
|
|
|
|
public void remove(HashedWheelTimeout timeout) {
|
|
HashedWheelTimeout next = timeout.next;
|
|
// remove timeout that was either processed or cancelled by updating the linked-list
|
|
if (timeout.prev != null) {
|
|
timeout.prev.next = next;
|
|
}
|
|
if (timeout.next != null) {
|
|
timeout.next.prev = timeout.prev;
|
|
}
|
|
|
|
if (timeout == head) {
|
|
// if timeout is also the tail we need to adjust the entry too
|
|
if (timeout == tail) {
|
|
tail = null;
|
|
head = null;
|
|
} else {
|
|
head = next;
|
|
}
|
|
} else if (timeout == tail) {
|
|
// if the timeout is the tail modify the tail to be the prev node.
|
|
tail = timeout.prev;
|
|
}
|
|
// null out prev, next and bucket to allow for GC.
|
|
timeout.prev = null;
|
|
timeout.next = null;
|
|
timeout.bucket = null;
|
|
}
|
|
|
|
/**
|
|
* Clear this bucket and return all not expired / cancelled {@link Timeout}s.
|
|
*/
|
|
public void clearTimeouts(Set<Timeout> set) {
|
|
for (;;) {
|
|
HashedWheelTimeout timeout = pollTimeout();
|
|
if (timeout == null) {
|
|
return;
|
|
}
|
|
if (timeout.isExpired() || timeout.isCancelled()) {
|
|
continue;
|
|
}
|
|
set.add(timeout);
|
|
}
|
|
}
|
|
|
|
private HashedWheelTimeout pollTimeout() {
|
|
HashedWheelTimeout head = this.head;
|
|
if (head == null) {
|
|
return null;
|
|
}
|
|
HashedWheelTimeout next = head.next;
|
|
if (next == null) {
|
|
tail = this.head = null;
|
|
} else {
|
|
this.head = next;
|
|
next.prev = null;
|
|
}
|
|
|
|
// null out prev and next to allow for GC.
|
|
head.next = null;
|
|
head.prev = null;
|
|
return head;
|
|
}
|
|
}
|
|
}
|