Fix Spinning in Reaper Thread

miniway · May 13, 2015 · 3a07b8d · 3a07b8d
1 parent c2fd3c4
commit 3a07b8d
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 7 deletions.
diff --git a/src/main/java/org/zeromq/ZBeacon.java b/src/main/java/org/zeromq/ZBeacon.java
@@ -26,6 +26,7 @@
 import java.net.SocketAddress;
 import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
 import java.nio.channels.DatagramChannel;
 import java.util.Arrays;
 
@@ -216,6 +217,9 @@ public void run()
                     size = read - buffer.remaining();
                     handleMessage(buffer, size, senderAddress);
                 }
+                catch (ClosedChannelException ioException) {
+                    break;
+                }
                 catch (IOException ioException) {
                     throw new RuntimeException(ioException);
                 }

diff --git a/src/main/java/zmq/Poller.java b/src/main/java/zmq/Poller.java
@@ -28,6 +28,7 @@
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class Poller extends PollerBase implements Runnable
 {
@@ -50,7 +51,7 @@ protected PollSet(IPollEvents handler)
     private final Map<SelectableChannel, PollSet> fdTable;
 
     //  If true, there's at least one retired event source.
-    private boolean retired;
+    private final AtomicBoolean retired = new AtomicBoolean(false);
 
     //  If true, thread is in the process of shutting down.
     private volatile boolean stopping;
@@ -68,7 +69,6 @@ public Poller()
     public Poller(String name)
     {
         this.name = name;
-        retired = false;
         stopping = false;
         stopped = false;
 
@@ -109,7 +109,7 @@ public final void addHandle(SelectableChannel fd, IPollEvents events)
     public final void removeHandle(SelectableChannel handle)
     {
         fdTable.get(handle).cancelled = true;
-        retired = true;
+        retired.set(true);
 
         //  Decrease the load metric of the thread.
         adjustLoad(-1);
@@ -160,7 +160,7 @@ private final void register(SelectableChannel handle, int ops, boolean negate)
             pollset.key.interestOps(pollset.ops);
         }
         else {
-            retired = true;
+            retired.set(true);
         }
     }
 
@@ -186,7 +186,7 @@ public void run()
             //  Execute any due timers.
             long timeout = executeTimers();
 
-            if (retired) {
+            while (retired.compareAndSet(true, false)) {
                 Iterator<Map.Entry<SelectableChannel, PollSet>> it = fdTable.entrySet().iterator();
                 while (it.hasNext()) {
                     Map.Entry<SelectableChannel, PollSet> entry = it.next();
@@ -207,7 +207,6 @@ public void run()
                         it.remove();
                     }
                 }
-                retired = false;
             }
 
             //  Wait for events.
@@ -289,6 +288,6 @@ private void rebuildSelector()
             pollSet.key = null;
         }
 
-        retired = true;
+        retired.set(true);
     }
 }
diff --git a/src/main/java/zmq/Signaler.java b/src/main/java/zmq/Signaler.java
@@ -105,6 +105,7 @@ public void send()
 
         while (true) {
             try {
+                Thread.interrupted();
                 nbytes = w.write(dummy);
             }
             catch (IOException e) {