Skip to content

Commit

Permalink
Fix the problem where HashedWheelTimer puts a timeout into an incorre…
Browse files Browse the repository at this point in the history 
…ct place

- the stopIndex of a timeout is calculated based on the start time of the worker thread and the current tick count for greater accuracy
  • Loading branch information
@trustin
trustin committed Oct 7, 2013
1 parent f4edb2f commit 3c7d458
Showing 1 changed file with 54 additions and 64 deletions.
118 changes: 54 additions & 64 deletions common/src/main/java/io/netty/util/HashedWheelTimer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
Expand Down Expand Up @@ -95,7 +96,9 @@ public class HashedWheelTimer implements Timer {
final Set<HashedWheelTimeout>[] wheel;
final int mask;
final ReadWriteLock lock = new ReentrantReadWriteLock();
volatile int wheelCursor;
final CountDownLatch startTimeInitialized = new CountDownLatch(1);
volatile long startTime;
volatile long tick;

/**
* Creates a new timer with the default thread factory
Expand Down Expand Up @@ -259,6 +262,15 @@ public void start() {
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.
}
}
}

@Override
Expand Down Expand Up @@ -310,7 +322,7 @@ public Set<Timeout> stop() {

@Override
public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
final long currentTime = System.nanoTime();
start();

if (task == null) {
throw new NullPointerException("task");
Expand All @@ -319,68 +331,50 @@ public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
throw new NullPointerException("unit");
}

start();

long delayInNanos = unit.toNanos(delay);
HashedWheelTimeout timeout = new HashedWheelTimeout(task, currentTime + delayInNanos);
scheduleTimeout(timeout, delayInNanos);
return timeout;
}

void scheduleTimeout(HashedWheelTimeout timeout, long delay) {
// Prepare the required parameters to schedule the timeout object.
long relativeIndex = (delay + tickDuration - 1) / tickDuration;
// if the previous line had an overflow going on, then we’ll just schedule this timeout
// one tick early; that shouldn’t matter since we’re talking 270 years here
if (relativeIndex < 0) {
relativeIndex = delay / tickDuration;
}
if (relativeIndex == 0) {
relativeIndex = 1;
}
if ((relativeIndex & mask) == 0) {
relativeIndex--;
}
final long remainingRounds = relativeIndex / wheel.length;
long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;

// Add the timeout to the wheel.
HashedWheelTimeout timeout;
lock.readLock().lock();
try {
timeout = new HashedWheelTimeout(task, deadline);
if (workerState.get() == WORKER_STATE_SHUTDOWN) {
throw new IllegalStateException("Cannot enqueue after shutdown");
}
final int stopIndex = (int) (wheelCursor + relativeIndex & mask);
timeout.stopIndex = stopIndex;
timeout.remainingRounds = remainingRounds;
wheel[stopIndex].add(timeout);
wheel[timeout.stopIndex].add(timeout);
} finally {
lock.readLock().unlock();
}

return timeout;
}

private final class Worker implements Runnable {

private long startTime;
private long tick;

Worker() {
}

@Override
public void run() {
List<HashedWheelTimeout> expiredTimeouts =
new ArrayList<HashedWheelTimeout>();

// Initialize the startTime.
startTime = System.nanoTime();
tick = 1;
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();

List<HashedWheelTimeout> expiredTimeouts = new ArrayList<HashedWheelTimeout>();

while (workerState.get() == WORKER_STATE_STARTED) {
do {
final long deadline = waitForNextTick();
if (deadline > 0) {
fetchExpiredTimeouts(expiredTimeouts, deadline);
notifyExpiredTimeouts(expiredTimeouts);
}
}
} while (workerState.get() == WORKER_STATE_STARTED);
}

private void fetchExpiredTimeouts(
Expand All @@ -392,9 +386,12 @@ private void fetchExpiredTimeouts(
// an exclusive lock.
lock.writeLock().lock();
try {
int newWheelCursor = wheelCursor = wheelCursor + 1 & mask;
fetchExpiredTimeouts(expiredTimeouts, wheel[newWheelCursor].iterator(), deadline);
fetchExpiredTimeouts(expiredTimeouts, wheel[(int) (tick & mask)].iterator(), deadline);
} finally {
// Note that the tick is updated only while the writer lock is held,
// so that newTimeout() and consequently new HashedWheelTimeout() never see an old value
// while the reader lock is held.
tick ++;
lock.writeLock().unlock();
}
}
Expand All @@ -403,34 +400,21 @@ private void fetchExpiredTimeouts(
List<HashedWheelTimeout> expiredTimeouts,
Iterator<HashedWheelTimeout> i, long deadline) {

List<HashedWheelTimeout> slipped = null;
while (i.hasNext()) {
HashedWheelTimeout timeout = i.next();
if (timeout.remainingRounds <= 0) {
i.remove();
if (timeout.deadline <= deadline) {
expiredTimeouts.add(timeout);
} else {
// Handle the case where the timeout is put into a wrong
// place, usually one tick earlier. For now, just add
// it to a temporary list - we will reschedule it in a
// separate loop.
if (slipped == null) {
slipped = new ArrayList<HashedWheelTimeout>();
}
slipped.add(timeout);
// The timeout was placed into a wrong slot. This should never happen.
throw new Error(String.format(
"timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
}
} else {
timeout.remainingRounds --;
}
}

// Reschedule the slipped timeouts.
if (slipped != null) {
for (HashedWheelTimeout timeout: slipped) {
scheduleTimeout(timeout, timeout.deadline - deadline);
}
}
}

private void notifyExpiredTimeouts(
Expand All @@ -451,14 +435,13 @@ private void notifyExpiredTimeouts(
* current time otherwise (with Long.MIN_VALUE changed by +1)
*/
private long waitForNextTick() {
long deadline = startTime + tickDuration * tick;
long deadline = tickDuration * (tick + 1);

for (;;) {
final long currentTime = System.nanoTime();
final long currentTime = System.nanoTime() - startTime;
long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;

if (sleepTimeMs <= 0) {
tick += 1;
if (currentTime == Long.MIN_VALUE) {
return -Long.MAX_VALUE;
} else {
Expand Down Expand Up @@ -494,13 +477,17 @@ private final class HashedWheelTimeout implements Timeout {

private final TimerTask task;
final long deadline;
volatile int stopIndex;
final int stopIndex;
volatile long remainingRounds;
private final AtomicInteger state = new AtomicInteger(ST_INIT);

HashedWheelTimeout(TimerTask task, long deadline) {
this.task = task;
this.deadline = deadline;

final long ticks = Math.max(deadline / tickDuration, tick); // Ensure we don't schedule for past.
stopIndex = (int) (ticks & mask);
remainingRounds = ticks / wheel.length;
}

@Override
Expand Down Expand Up @@ -550,7 +537,7 @@ public void expire() {
@Override
public String toString() {
final long currentTime = System.nanoTime();
long remaining = deadline - currentTime;
long remaining = deadline - currentTime + startTime;

StringBuilder buf = new StringBuilder(192);
buf.append(getClass().getSimpleName());
Expand All @@ -559,18 +546,21 @@ public String toString() {
buf.append("deadline: ");
if (remaining > 0) {
buf.append(remaining);
buf.append(" ms later, ");
buf.append(" ns later");
} else if (remaining < 0) {
buf.append(-remaining);
buf.append(" ms ago, ");
buf.append(" ns ago");
} else {
buf.append("now, ");
buf.append("now");
}

if (isCancelled()) {
buf.append(", cancelled");
}

buf.append(", task: ");
buf.append(task());

return buf.append(')').toString();
}
}
Expand Down

0 comments on commit 3c7d458

Please sign in to comment.