Merge pull request #224 from dhakimTRX/4.0.x-stable

Handle interrupts correctly, switch to a monotonically growing clock function in BlockingCell and SingleShotLinearTimer

Author: acogoluegnes

src/main/java/com/rabbitmq/utility/BlockingCell.java

@@ -66,9 +66,9 @@ public synchronized T get(long timeout) throws InterruptedException, TimeoutExce
         if (timeout < 0)
             throw new AssertionError("Timeout cannot be less than zero");
 
-        long maxTime = System.currentTimeMillis() + timeout;
-        long now;
-        while (!_filled && (now = System.currentTimeMillis()) < maxTime) {
+        long now = System.nanoTime() / NANOS_IN_MILLI;
+        long maxTime = now + timeout;
+        while (!_filled && (now = (System.nanoTime() / NANOS_IN_MILLI)) < maxTime) {
             wait(maxTime - now);
         }
 
@@ -83,12 +83,19 @@ public synchronized T get(long timeout) throws InterruptedException, TimeoutExce
      * @return the waited-for value
      */
     public synchronized T uninterruptibleGet() {
-        while (true) {
-            try {
-                return get();
-            } catch (InterruptedException ex) {
-                // no special handling necessary
+        boolean wasInterrupted = false;
+        try {
+            while (true) {
+                try {
+                    return get();
+                } catch (InterruptedException ex) {
+                    // no special handling necessary
+                    wasInterrupted = true;
+                }
             }
+        } finally {
+            if (wasInterrupted)
+                Thread.currentThread().interrupt();
         }
     }
 
@@ -104,14 +111,20 @@ public synchronized T uninterruptibleGet() {
     public synchronized T uninterruptibleGet(int timeout) throws TimeoutException {
         long now = System.nanoTime() / NANOS_IN_MILLI;
         long runTime = now + timeout;
-
-        do {
-            try {
-                return get(runTime - now);
-            } catch (InterruptedException e) {
-                // Ignore.
-            }
-        } while ((timeout == INFINITY) || ((now = System.nanoTime() / NANOS_IN_MILLI) < runTime));
+        boolean wasInterrupted = false;
+        try {
+            do {
+                try {
+                    return get(runTime - now);
+                } catch (InterruptedException e) {
+                    // Ignore.
+                    wasInterrupted = true;
+                }
+            } while ((timeout == INFINITY) || ((now = System.nanoTime() / NANOS_IN_MILLI) < runTime));
+        } finally {
+            if (wasInterrupted)
+                Thread.currentThread().interrupt();
+        }
 
         throw new TimeoutException();
     }

src/main/java/com/rabbitmq/utility/SingleShotLinearTimer.java

@@ -72,21 +72,27 @@ public TimerThread(long timeoutMillisec) {
         public void run() {
             try {
                 long now;
-                while ((now = System.nanoTime() / NANOS_IN_MILLI) < _runTime) {
-                    if (_task == null) break;
+                boolean wasInterrupted = false;
+                try {
+                    while ((now = System.nanoTime() / NANOS_IN_MILLI) < _runTime) {
+                        if (_task == null) break;
 
-                    try {
-                        synchronized(this) {
-                            wait(_runTime - now);
+                        try {
+                            synchronized(this) {
+                                wait(_runTime - now);
+                            }
+                        } catch (InterruptedException e) {
+                            wasInterrupted = true;
                         }
-                    } catch (InterruptedException e) {
-                        Thread.currentThread().interrupt();
                     }
+                } finally {
+                    if (wasInterrupted)
+                        Thread.currentThread().interrupt();
                 }
 
                 Runnable task = _task;
                 if (task != null) {
-                    task.run();                    
+                    task.run();
                 }
 
             } finally {