Repository: netty/netty
Issue: WriteTimeoutHandler missed cleanup
Labels: defect

It seems like the `WriteTimeoutHandler` does not always cancel the timeout task it schedules in all the same cases that the and IdleStateHandler does.  For example the `handlerRemoved` method is not overridden in `WrtieTimeoutHandler`.  There may be other cases where the timeout is not canceled when it should be and further investigation is needed.


Issue comments:
@Scottmitch is this still valid ?

@normanmaurer - I think so. I have not done anything to fix it.

Fixed


Pull request review comments:
can you explain why we need an array?

check `if (!timeout[0].isDone()) {..}` before adding the object?

I wonder if we need to keep a collection of timeouts. Because the time is fixed, each new event will alway fire later (in time) than the previous event. Can we just keep a reference to the "last" event, and cancel it when necessary (scheduling new event, cleanup, etc..)?

sounds good.

I guess the array is used @windie tries to access the future that is returned by `schedule(...)` in the `OneTimeTask` itself. Consider just using create a `OneTimeTask` subclass which will contain the `Future` in a field. Something like this:

``` java
private class TimeoutTask extends OneTimeTask {
    Future<?> future;

    @Override
    public void run() {
        ...
        if (future != null) {
            timeouts.remove(future); 
        }
    }
}

....
TimeoutTask task = new TimeoutTask();

task.future = ctx.executor().schedule(task, timeoutNanos, TimeUnit.NANOSECONDS);
...
```

+1

Addressed and rebased

Addressed and rebased

need to check if timeout != null a well

So it's guaranteed that ChannelPromise's listener will be called before the next `write` event?

task.timeout is assigned before this line

consider renaming it to scheduledFuture.

rename to promise

static

doh! I need more coffee... you are right.

it needs to access the outer `timeouts`

no it's not... but it is guaranteed it is called once it is written or the write failed. I just wonder if we can somehow remove the overhead of the Set in terms of memory + remove/add.

> So it's guaranteed that ChannelPromise's listener will be called before the next write event?

Just want to confirm this one before removing the collection of timeouts

what we also could do is make the WriteTimeoutTask basically a doubly-linked list and so have a ref to `prev` and `next` task. This way adding and removing would be O(1) for our use-case and we not need any extra data structure. Not sure if this is better or not though

> no it's not... but it is guaranteed it is called once it is written or the write failed. I just wonder if we can somehow remove the overhead of the Set in terms of memory + remove/add.

So if just storing a reference to the last `timeout`, it may override the previous one. Then the previous one won't be cleaned in `handlerRemoved`. Seems we need a collection to store active timeouts at least.

@windie yeah it seems so :( What you think about my "doubly-linked-list" idea ?

> @windie yeah it seems so :( What you think about my "doubly-linked-list" idea ?

Sounds good. Let me try it.

nit: remove final

makes sense

Can we also have WriteTimeoutTask implement ChannelFutureListener and so safe one object allocation ?


Patch that resolved the issue:
diff --git a/handler/src/main/java/io/netty/handler/timeout/WriteTimeoutHandler.java b/handler/src/main/java/io/netty/handler/timeout/WriteTimeoutHandler.java
index 4e87c7f17c0..d47760a857e 100644
--- a/handler/src/main/java/io/netty/handler/timeout/WriteTimeoutHandler.java
+++ b/handler/src/main/java/io/netty/handler/timeout/WriteTimeoutHandler.java
@@ -30,12 +30,10 @@
 import java.util.concurrent.TimeUnit;
 
 /**
- * Raises a {@link WriteTimeoutException} when no data was written within a
- * certain period of time.
+ * Raises a {@link WriteTimeoutException} when a write operation cannot finish in a certain period of time.
  *
  * <pre>
- * // The connection is closed when there is no outbound traffic
- * // for 30 seconds.
+ * // The connection is closed when a write operation cannot finish in 30 seconds.
  *
  * public class MyChannelInitializer extends {@link ChannelInitializer}&lt;{@link Channel}&gt; {
  *     public void initChannel({@link Channel} channel) {
@@ -70,6 +68,11 @@ public class WriteTimeoutHandler extends ChannelOutboundHandlerAdapter {
 
     private final long timeoutNanos;
 
+    /**
+     * A doubly-linked list to track all WriteTimeoutTasks
+     */
+    private WriteTimeoutTask lastTask;
+
     private boolean closed;
 
     /**
@@ -111,31 +114,62 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
         ctx.write(msg, promise);
     }
 
-    private void scheduleTimeout(final ChannelHandlerContext ctx, final ChannelPromise future) {
+    @Override
+    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+        WriteTimeoutTask task = lastTask;
+        lastTask = null;
+        while (task != null) {
+            task.scheduledFuture.cancel(false);
+            WriteTimeoutTask prev = task.prev;
+            task.prev = null;
+            task.next = null;
+            task = prev;
+        }
+    }
+
+    private void scheduleTimeout(final ChannelHandlerContext ctx, final ChannelPromise promise) {
         // Schedule a timeout.
-        final ScheduledFuture<?> sf = ctx.executor().schedule(new OneTimeTask() {
-            @Override
-            public void run() {
-                // Was not written yet so issue a write timeout
-                // The future itself will be failed with a ClosedChannelException once the close() was issued
-                // See https://github.com/netty/netty/issues/2159
-                if (!future.isDone()) {
-                    try {
-                        writeTimedOut(ctx);
-                    } catch (Throwable t) {
-                        ctx.fireExceptionCaught(t);
-                    }
-                }
-            }
-        }, timeoutNanos, TimeUnit.NANOSECONDS);
+        final WriteTimeoutTask task = new WriteTimeoutTask(ctx, promise);
+        task.scheduledFuture = ctx.executor().schedule(task, timeoutNanos, TimeUnit.NANOSECONDS);
+
+        if (!task.scheduledFuture.isDone()) {
+            addWriteTimeoutTask(task);
+
+            // Cancel the scheduled timeout if the flush promise is complete.
+            promise.addListener(task);
+        }
+    }
+
+    private void addWriteTimeoutTask(WriteTimeoutTask task) {
+        if (lastTask == null) {
+            lastTask = task;
+        } else {
+            lastTask.next = task;
+            task.prev = lastTask;
+            lastTask = task;
+        }
+    }
 
-        // Cancel the scheduled timeout if the flush future is complete.
-        future.addListener(new ChannelFutureListener() {
-            @Override
-            public void operationComplete(ChannelFuture future) throws Exception {
-                sf.cancel(false);
+    private void removeWriteTimeoutTask(WriteTimeoutTask task) {
+        if (task == lastTask) {
+            // task is the tail of list
+            assert task.next == null;
+            lastTask = lastTask.prev;
+            if (lastTask != null) {
+                lastTask.next = null;
             }
-        });
+        } else if (task.prev == null && task.next == null) {
+            // Since task is not lastTask, then it has been removed or not been added.
+            return;
+        } else if (task.prev == null) {
+            // task is the head of list and the list has at least 2 nodes
+            task.next.prev = null;
+        } else {
+            task.prev.next = task.next;
+            task.next.prev = task.prev;
+        }
+        task.prev = null;
+        task.next = null;
     }
 
     /**
@@ -148,4 +182,43 @@ protected void writeTimedOut(ChannelHandlerContext ctx) throws Exception {
             closed = true;
         }
     }
+
+    private final class WriteTimeoutTask extends OneTimeTask implements ChannelFutureListener {
+
+        private final ChannelHandlerContext ctx;
+        private final ChannelPromise promise;
+
+        // WriteTimeoutTask is also a node of a doubly-linked list
+        WriteTimeoutTask prev;
+        WriteTimeoutTask next;
+
+        ScheduledFuture<?> scheduledFuture;
+
+        WriteTimeoutTask(ChannelHandlerContext ctx, ChannelPromise promise) {
+            this.ctx = ctx;
+            this.promise = promise;
+        }
+
+        @Override
+        public void run() {
+            // Was not written yet so issue a write timeout
+            // The promise itself will be failed with a ClosedChannelException once the close() was issued
+            // See https://github.com/netty/netty/issues/2159
+            if (!promise.isDone()) {
+                try {
+                    writeTimedOut(ctx);
+                } catch (Throwable t) {
+                    ctx.fireExceptionCaught(t);
+                }
+            }
+            removeWriteTimeoutTask(this);
+        }
+
+        @Override
+        public void operationComplete(ChannelFuture future) throws Exception {
+            // scheduledFuture has already be set when reaching here
+            scheduledFuture.cancel(false);
+            removeWriteTimeoutTask(this);
+        }
+    }
 }


Test patch: