Fixed issue: NETTY-360 Changing Channel.interestOps sometimes does not work under heavy write traffic in NIO transport

* Made sure all setRawInterestOps() calls are protected by interestOpsLock
Fixed a race condition in the HexDumpProxy example

src/main/java/org/jboss/netty/channel/socket/nio/NioDatagramWorker.java

@@ -604,24 +604,18 @@ class NioDatagramWorker implements Runnable {
             close(key);
             return;
         }
-        int interestOps;
-        boolean changed = false;
 
         // interestOps can change at any time and at any thread.
         // Acquire a lock to avoid possible race condition.
         synchronized (channel.interestOpsLock) {
-            interestOps = channel.getRawInterestOps();
+            int interestOps = channel.getRawInterestOps();
             if ((interestOps & SelectionKey.OP_WRITE) == 0) {
                 interestOps |= SelectionKey.OP_WRITE;
                 key.interestOps(interestOps);
-                changed = true;
-            }
-        }
-
-        if (changed) {
                 channel.setRawInterestOpsNow(interestOps);
             }
         }
+    }
 
     private void clearOpWrite(NioDatagramChannel channel) {
         Selector selector = this.selector;
@@ -633,24 +627,18 @@ class NioDatagramWorker implements Runnable {
             close(key);
             return;
         }
-        int interestOps;
-        boolean changed = false;
 
         // interestOps can change at any time and at any thread.
         // Acquire a lock to avoid possible race condition.
         synchronized (channel.interestOpsLock) {
-            interestOps = channel.getRawInterestOps();
+            int interestOps = channel.getRawInterestOps();
             if ((interestOps & SelectionKey.OP_WRITE) != 0) {
                 interestOps &= ~SelectionKey.OP_WRITE;
                 key.interestOps(interestOps);
-                changed = true;
-            }
-        }
-
-        if (changed) {
                 channel.setRawInterestOpsNow(interestOps);
             }
         }
+    }
 
     static void disconnect(NioDatagramChannel channel, ChannelFuture future) {
         boolean connected = channel.isConnected();
@@ -811,11 +799,13 @@ class NioDatagramWorker implements Runnable {
                 default:
                     throw new Error();
                 }
+                if (changed) {
+                    channel.setRawInterestOpsNow(interestOps);
+                }
             }
 
             future.setSuccess();
             if (changed) {
-                channel.setRawInterestOpsNow(interestOps);
                 fireChannelInterestChanged(channel);
             }
         } catch (final CancelledKeyException e) {

src/main/java/org/jboss/netty/channel/socket/nio/NioWorker.java

@@ -536,24 +536,18 @@ class NioWorker implements Runnable {
             close(key);
             return;
         }
-        int interestOps;
-        boolean changed = false;
 
         // interestOps can change at any time and at any thread.
         // Acquire a lock to avoid possible race condition.
         synchronized (channel.interestOpsLock) {
-            interestOps = channel.getRawInterestOps();
+            int interestOps = channel.getRawInterestOps();
             if ((interestOps & SelectionKey.OP_WRITE) == 0) {
                 interestOps |= SelectionKey.OP_WRITE;
                 key.interestOps(interestOps);
-                changed = true;
-            }
-        }
-
-        if (changed) {
                 channel.setRawInterestOpsNow(interestOps);
             }
         }
+    }
 
     private void clearOpWrite(NioSocketChannel channel) {
         Selector selector = this.selector;
@@ -565,24 +559,18 @@ class NioWorker implements Runnable {
             close(key);
             return;
         }
-        int interestOps;
-        boolean changed = false;
 
         // interestOps can change at any time and at any thread.
         // Acquire a lock to avoid possible race condition.
         synchronized (channel.interestOpsLock) {
-            interestOps = channel.getRawInterestOps();
+            int interestOps = channel.getRawInterestOps();
             if ((interestOps & SelectionKey.OP_WRITE) != 0) {
                 interestOps &= ~SelectionKey.OP_WRITE;
                 key.interestOps(interestOps);
-                changed = true;
-            }
-        }
-
-        if (changed) {
                 channel.setRawInterestOpsNow(interestOps);
             }
         }
+    }
 
     void close(NioSocketChannel channel, ChannelFuture future) {
         boolean connected = channel.isConnected();
@@ -719,11 +707,14 @@ class NioWorker implements Runnable {
                 default:
                     throw new Error();
                 }
+
+                if (changed) {
+                    channel.setRawInterestOpsNow(interestOps);
+                }
             }
 
             future.setSuccess();
             if (changed) {
-                channel.setRawInterestOpsNow(interestOps);
                 fireChannelInterestChanged(channel);
             }
         } catch (CancelledKeyException e) {

src/main/java/org/jboss/netty/example/proxy/HexDumpProxyInboundHandler.java

@@ -41,6 +41,11 @@ public class HexDumpProxyInboundHandler extends SimpleChannelUpstreamHandler {
     private final String remoteHost;
     private final int remotePort;
 
+    // This lock guards against the race condition that overrides the
+    // OP_READ flag incorrectly.
+    // See the related discussion: http://markmail.org/message/x7jc6mqx6ripynqf
+    final Object trafficLock = new Object();
+
     private volatile Channel outboundChannel;
 
     public HexDumpProxyInboundHandler(
@@ -78,10 +83,11 @@ public class HexDumpProxyInboundHandler extends SimpleChannelUpstreamHandler {
     }
 
     @Override
-    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
+    public void messageReceived(ChannelHandlerContext ctx, final MessageEvent e)
             throws Exception {
         ChannelBuffer msg = (ChannelBuffer) e.getMessage();
-        System.out.println(">>> " + ChannelBuffers.hexDump(msg));
+        //System.out.println(">>> " + ChannelBuffers.hexDump(msg));
+        synchronized (trafficLock) {
             outboundChannel.write(msg);
             // If outboundChannel is saturated, do not read until notified in
             // OutboundHandler.channelInterestChanged().
@@ -89,16 +95,19 @@ public class HexDumpProxyInboundHandler extends SimpleChannelUpstreamHandler {
                 e.getChannel().setReadable(false);
             }
         }
+    }
 
     @Override
     public void channelInterestChanged(ChannelHandlerContext ctx,
             ChannelStateEvent e) throws Exception {
         // If inboundChannel is not saturated anymore, continue accepting
         // the incoming traffic from the outboundChannel.
+        synchronized (trafficLock) {
             if (e.getChannel().isWritable()) {
                 outboundChannel.setReadable(true);
             }
         }
+    }
 
     @Override
     public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e)
@@ -115,7 +124,7 @@ public class HexDumpProxyInboundHandler extends SimpleChannelUpstreamHandler {
         closeOnFlush(e.getChannel());
     }
 
-    private static class OutboundHandler extends SimpleChannelUpstreamHandler {
+    private class OutboundHandler extends SimpleChannelUpstreamHandler {
 
         private final Channel inboundChannel;
 
@@ -124,10 +133,11 @@ public class HexDumpProxyInboundHandler extends SimpleChannelUpstreamHandler {
         }
 
         @Override
-        public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
+        public void messageReceived(ChannelHandlerContext ctx, final MessageEvent e)
                 throws Exception {
             ChannelBuffer msg = (ChannelBuffer) e.getMessage();
-            System.out.println("<<< " + ChannelBuffers.hexDump(msg));
+            //System.out.println("<<< " + ChannelBuffers.hexDump(msg));
+            synchronized (trafficLock) {
                 inboundChannel.write(msg);
                 // If inboundChannel is saturated, do not read until notified in
                 // HexDumpProxyInboundHandler.channelInterestChanged().
@@ -135,16 +145,19 @@ public class HexDumpProxyInboundHandler extends SimpleChannelUpstreamHandler {
                     e.getChannel().setReadable(false);
                 }
             }
+        }
 
         @Override
         public void channelInterestChanged(ChannelHandlerContext ctx,
                 ChannelStateEvent e) throws Exception {
             // If outboundChannel is not saturated anymore, continue accepting
             // the incoming traffic from the inboundChannel.
+            synchronized (trafficLock) {
                 if (e.getChannel().isWritable()) {
                     inboundChannel.setReadable(true);
                 }
             }
+        }
 
         @Override
         public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e)