make HashedWheelTimer tolerate wrap-arounds, see #2686

akka · Nov 14, 2012 · 7e590f3 · 7e590f3
1 parent 54555ce
commit 7e590f3
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 5 deletions.
diff --git a/akka-actor-tests/src/test/scala/akka/util/DurationSpec.scala b/akka-actor-tests/src/test/scala/akka/util/DurationSpec.scala
@@ -6,8 +6,27 @@ package akka.util
 import org.scalatest.WordSpec
 import org.scalatest.matchers.MustMatchers
 import duration._
+import akka.testkit.AkkaSpec
+import akka.testkit.TestLatch
+import java.util.concurrent.TimeoutException
+import akka.dispatch.Await
+import akka.testkit.TimingTest
 
-class DurationSpec extends WordSpec with MustMatchers {
+class DurationSpec extends AkkaSpec {
+
+  "A HashedWheelTimer" must {
+
+    "not mess up long timeouts" taggedAs TimingTest in {
+      val longish = Timeout(Long.MaxValue)
+      val barrier = TestLatch()
+      val job = system.scheduler.scheduleOnce(longish.duration)(barrier.countDown())
+      intercept[TimeoutException] {
+        Await.ready(barrier, 90 seconds)
+      }
+      job.cancel()
+    }
+
+  }
 
   "Duration" must {
 

diff --git a/akka-actor/src/main/java/org/jboss/netty/akka/util/HashedWheelTimer.java b/akka-actor/src/main/java/org/jboss/netty/akka/util/HashedWheelTimer.java
@@ -268,6 +268,9 @@ public Timeout newTimeout(TimerTask task, Duration delay) {
     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;
         }
@@ -313,7 +316,7 @@ public void run() {
 
             while (!shutdown()) {
                 final long deadline = waitForNextTick();
-                if (deadline > 0) {
+                if (deadline > Long.MIN_VALUE) {
                     fetchExpiredTimeouts(expiredTimeouts, deadline);
                     notifyExpiredTimeouts(expiredTimeouts);
                 }
@@ -346,7 +349,7 @@ private void fetchExpiredTimeouts(
                 HashedWheelTimeout timeout = i.next();
                 if (timeout.remainingRounds <= 0) {
                     i.remove();
-                    if (timeout.deadline <= deadline) {
+                    if (timeout.deadline - deadline <= 0) {
                         expiredTimeouts.add(timeout);
                     } else {
                         // Handle the case where the timeout is put into a wrong
@@ -382,6 +385,12 @@ private void notifyExpiredTimeouts(
             expiredTimeouts.clear();
         }
 
+        /**
+         * 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() {
             final long deadline = startTime + tickDuration * tick;
 
@@ -392,7 +401,8 @@ private long waitForNextTick() {
 
                 if (sleepTimeMs <= 0) {
                     tick += 1;
-                    return currentTime;
+                    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
@@ -408,7 +418,7 @@ private long waitForNextTick() {
                     Thread.sleep(sleepTimeMs);
                 } catch (InterruptedException e) {
                     if (shutdown()) {
-                        return -1;
+                        return Long.MIN_VALUE;
                     }
                 }
             }