Repository: netty/netty
Issue: Channel writability change notifications sometimes fail to fire

As of the changes made for #1766, sometimes channel writability notifications are lost.  In particular, I've noticed this when writing from a thread other than the channel's EventLoop.

I believe that there's a logic error in `ChannelOutboundBuffer` in the `incrementPendingOutboundBytes` and `decrementPendingOutboundBytes` methods.

In particular, each method does something like this:

``` java
if (newWriteBufferSize > highWaterMark) {
    if (WRITABLE_UPDATER.compareAndSet(this, 1, 0)) {
        if (fireEvent) {
            channel.pipeline().fireChannelWritabilityChanged();
        }
    }
}
```

`WRITABLE_UPDATER.compareAndSet(this, 1, 0)` handles updating the writability flag. However, we only notify if `fireEvent` is true.  The problem is that if the writability flag happens to change inside of a call where `fireEvent` is false, then the pipeline will never be notified about the fact that the writability changed.

I don't actually quite understand the need for the `fireEvent` flag, but assuming it's necessary, there seem to be a couple of parsimonious solutions available.
## Solution 1

Reverse the calls:

``` java
if (newWriteBufferSize > highWaterMark) {
    if (fireEvent) {
        if (WRITABLE_UPDATER.compareAndSet(this, 1, 0)) {
            channel.pipeline().fireChannelWritabilityChanged();
        }
    }
}
```

In this implementation, the result from Channel.isWritable() is always in sync with whatever notifications have been made.
## Solution 2

Use a second writable status:

``` java
if (newWriteBufferSize > highWaterMark) {
    if (WRITABLE_UPDATER.compareAndSet(this, 1, 0)) {
        if (fireEvent) {
            if (LAST_NOTIFIED_WRITABLE_UPDATER.compareAndSet(this, 1, 0)) {
                channel.pipeline().fireChannelWritabilityChanged();
            }
        }
    }
}
```

In this implementation, the result from Channel.isWritable() is always super up-to-date and the notification can lag, but you will get notified at some point.


Issue comments:
@wgallagher and assigned to you... have fun ;)

@wgallagher ping ;)

@wgallagher any news here ? I would love to cut 4.0.10.Final early next week.

@oxtoacart was fixed... please retest with latest snapshot.

@wgallagher I tried `4.1.0.Alpha1-SNAPSHOT` and this seems to be working fine for me now.  Thanks for fixing it!


Pull request review comments:

Patch that resolved the issue:
diff --git a/transport/src/main/java/io/netty/channel/ChannelOutboundBuffer.java b/transport/src/main/java/io/netty/channel/ChannelOutboundBuffer.java
index 48a8c31aa78..393aad82463 100644
--- a/transport/src/main/java/io/netty/channel/ChannelOutboundBuffer.java
+++ b/transport/src/main/java/io/netty/channel/ChannelOutboundBuffer.java
@@ -129,7 +129,7 @@ void addMessage(Object msg, ChannelPromise promise) {
 
         // increment pending bytes after adding message to the unflushed arrays.
         // See https://github.com/netty/netty/issues/1619
-        incrementPendingOutboundBytes(size, true);
+        incrementPendingOutboundBytes(size);
     }
 
     private void addCapacity() {
@@ -164,7 +164,7 @@ void addFlush() {
      * Increment the pending bytes which will be written at some point.
      * This method is thread-safe!
      */
-    void incrementPendingOutboundBytes(int size, boolean fireEvent) {
+    void incrementPendingOutboundBytes(int size) {
         // Cache the channel and check for null to make sure we not produce a NPE in case of the Channel gets
         // recycled while process this method.
         Channel channel = this.channel;
@@ -183,9 +183,7 @@ void incrementPendingOutboundBytes(int size, boolean fireEvent) {
 
         if (newWriteBufferSize > highWaterMark) {
             if (WRITABLE_UPDATER.compareAndSet(this, 1, 0)) {
-                if (fireEvent) {
-                    channel.pipeline().fireChannelWritabilityChanged();
-                }
+                channel.pipeline().fireChannelWritabilityChanged();
             }
         }
     }
@@ -194,7 +192,7 @@ void incrementPendingOutboundBytes(int size, boolean fireEvent) {
      * Decrement the pending bytes which will be written at some point.
      * This method is thread-safe!
      */
-    void decrementPendingOutboundBytes(int size, boolean fireEvent) {
+    void decrementPendingOutboundBytes(int size) {
         // Cache the channel and check for null to make sure we not produce a NPE in case of the Channel gets
         // recycled while process this method.
         Channel channel = this.channel;
@@ -213,9 +211,7 @@ void decrementPendingOutboundBytes(int size, boolean fireEvent) {
 
         if (newWriteBufferSize == 0 || newWriteBufferSize < lowWaterMark) {
             if (WRITABLE_UPDATER.compareAndSet(this, 0, 1)) {
-                if (fireEvent) {
-                    channel.pipeline().fireChannelWritabilityChanged();
-                }
+                channel.pipeline().fireChannelWritabilityChanged();
             }
         }
     }
@@ -316,7 +312,7 @@ public boolean remove() {
         safeRelease(msg);
 
         promise.trySuccess();
-        decrementPendingOutboundBytes(size, true);
+        decrementPendingOutboundBytes(size);
 
         return true;
     }
@@ -342,7 +338,7 @@ public boolean remove(Throwable cause) {
         safeRelease(msg);
 
         safeFail(promise, cause);
-        decrementPendingOutboundBytes(size, true);
+        decrementPendingOutboundBytes(size);
 
         return true;
     }
diff --git a/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java b/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java
index 38eab413988..352aa6426d8 100644
--- a/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java
+++ b/transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java
@@ -642,7 +642,7 @@ private void write(Object msg, boolean flush, ChannelPromise promise) {
                 ChannelOutboundBuffer buffer = channel.unsafe().outboundBuffer();
                 // Check for null as it may be set to null if the channel is closed already
                 if (buffer != null) {
-                    buffer.incrementPendingOutboundBytes(size, false);
+                    buffer.incrementPendingOutboundBytes(size);
                 }
             }
             executor.execute(WriteTask.newInstance(next, msg, size, flush, promise));
@@ -822,7 +822,7 @@ public void run() {
                     ChannelOutboundBuffer buffer = ctx.channel.unsafe().outboundBuffer();
                     // Check for null as it may be set to null if the channel is closed already
                     if (buffer != null) {
-                        buffer.decrementPendingOutboundBytes(size, false);
+                        buffer.decrementPendingOutboundBytes(size);
                     }
                 }
                 ctx.invokeWrite(msg, promise);


Test patch:
diff --git a/transport/src/test/java/io/netty/channel/ReentrantChannelTest.java b/transport/src/test/java/io/netty/channel/ReentrantChannelTest.java
index 67cd5fbe177..f0bf9bb308d 100644
--- a/transport/src/test/java/io/netty/channel/ReentrantChannelTest.java
+++ b/transport/src/test/java/io/netty/channel/ReentrantChannelTest.java
@@ -54,6 +54,8 @@ public void testWritabilityChanged() throws Exception {
         clientChannel.close().sync();
 
         assertLog(
+            "WRITABILITY: writable=false\n" +
+            "WRITABILITY: writable=true\n" +
             "WRITE\n" +
             "WRITABILITY: writable=false\n" +
             "FLUSH\n" +
@@ -91,6 +93,9 @@ public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exceptio
         clientChannel.close().sync();
 
         assertLog(
+            "WRITABILITY: writable=false\n" +
+            "FLUSH\n" +
+            "WRITABILITY: writable=true\n" +
             "WRITE\n" +
             "WRITABILITY: writable=false\n" +
             "FLUSH\n" +