Remove Channel from map on timeout(#359)

Motivation:

When a Channel was closed due a timeout we did miss to remove it from the internal map.

Modifications:

Remove channel on timeout

Result:

No more leak in the map

Co-authored-by: 吴锦程 <wujincheng@wujinchengdeMacBook-Pro.local>
Co-authored-by: Norman Maurer <norman_maurer@apple.com>

Author: 3 people

codec-classes-quic/src/main/java/io/netty/incubator/codec/quic/QuicheQuicChannel.java

@@ -60,6 +60,7 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+import java.util.function.Consumer;
 import java.util.function.Function;
 
 /**
@@ -112,7 +113,7 @@ public void operationComplete(ChannelFuture future) {
     private final ChannelHandler streamHandler;
     private final Map.Entry<ChannelOption<?>, Object>[] streamOptionsArray;
     private final Map.Entry<AttributeKey<?>, Object>[] streamAttrsArray;
-    private final TimeoutHandler timeoutHandler = new TimeoutHandler();
+    private final TimeoutHandler timeoutHandler;
 
     private boolean inFireChannelReadCompleteQueue;
     private boolean fireChannelReadCompletePending;
@@ -156,6 +157,14 @@ private QuicheQuicChannel(Channel parent, boolean server, ByteBuffer key,
                       InetSocketAddress remote, boolean supportsDatagram, ChannelHandler streamHandler,
                               Map.Entry<ChannelOption<?>, Object>[] streamOptionsArray,
                               Map.Entry<AttributeKey<?>, Object>[] streamAttrsArray) {
+        this(parent, server, key, remote, supportsDatagram, streamHandler, streamOptionsArray, streamAttrsArray, null);
+    }
+
+    private QuicheQuicChannel(Channel parent, boolean server, ByteBuffer key,
+                              InetSocketAddress remote, boolean supportsDatagram, ChannelHandler streamHandler,
+                              Map.Entry<ChannelOption<?>, Object>[] streamOptionsArray,
+                              Map.Entry<AttributeKey<?>, Object>[] streamAttrsArray,
+                              Consumer<QuicheQuicChannel> timeoutTask) {
         super(parent);
         config = new QuicheQuicChannelConfig(this);
         this.server = server;
@@ -169,6 +178,7 @@ private QuicheQuicChannel(Channel parent, boolean server, ByteBuffer key,
         this.streamHandler = streamHandler;
         this.streamOptionsArray = streamOptionsArray;
         this.streamAttrsArray = streamAttrsArray;
+        timeoutHandler = new TimeoutHandler(timeoutTask);
     }
 
     static QuicheQuicChannel forClient(Channel parent, InetSocketAddress remote, ChannelHandler streamHandler,
@@ -186,6 +196,15 @@ static QuicheQuicChannel forServer(Channel parent, ByteBuffer key, InetSocketAdd
                 streamHandler, streamOptionsArray, streamAttrsArray);
     }
 
+    static QuicheQuicChannel forServer(Channel parent, ByteBuffer key, InetSocketAddress remote,
+                                       boolean supportsDatagram, ChannelHandler streamHandler,
+                                       Map.Entry<ChannelOption<?>, Object>[] streamOptionsArray,
+                                       Map.Entry<AttributeKey<?>, Object>[] streamAttrsArray,
+                                       Consumer<QuicheQuicChannel> timeoutTask) {
+        return new QuicheQuicChannel(parent, true, key, remote, supportsDatagram,
+                streamHandler, streamOptionsArray, streamAttrsArray, timeoutTask);
+    }
+
     @Override
     public boolean isTimedOut() {
         return timedOut;
@@ -1469,6 +1488,11 @@ private static final class QuicheQuicChannelAddress extends SocketAddress {
 
     private final class TimeoutHandler implements Runnable {
         private ScheduledFuture<?> timeoutFuture;
+        private final Consumer<QuicheQuicChannel> timeoutTask;
+
+        TimeoutHandler(Consumer<QuicheQuicChannel> timeoutTask) {
+            this.timeoutTask = timeoutTask;
+        }
 
         @Override
         public void run() {
@@ -1480,6 +1504,9 @@ public void run() {
 
                 if (Quiche.quiche_conn_is_closed(connAddr)) {
                     forceClose();
+                    if (timeoutTask != null){
+                        timeoutTask.accept(QuicheQuicChannel.this);
+                    }
                 } else {
                     // We need to call connectionSend when a timeout was triggered.
                     // See https://docs.rs/quiche/0.6.0/quiche/struct.Connection.html#method.send.

codec-classes-quic/src/main/java/io/netty/incubator/codec/quic/QuicheQuicCodec.java

@@ -71,6 +71,10 @@ protected void putChannel(QuicheQuicChannel channel) {
         connections.put(channel.key(), channel);
     }
 
+    protected void removeChannel(QuicheQuicChannel channel) {
+        connections.remove(channel.key());
+    }
+
     @Override
     public void handlerAdded(ChannelHandlerContext ctx) {
         sockaddrMemory = allocateNativeOrder(Quiche.SIZEOF_SOCKADDR_STORAGE);

codec-classes-quic/src/main/java/io/netty/incubator/codec/quic/QuicheQuicServerCodec.java

@@ -197,10 +197,10 @@ private QuicheQuicChannel handleServer(ChannelHandlerContext ctx, InetSocketAddr
             dcid.getBytes(dcid.readerIndex(), bytes);
             key = ByteBuffer.wrap(bytes);
         }
-
         QuicheQuicChannel channel = QuicheQuicChannel.forServer(
                 ctx.channel(), key, sender, config.isDatagramSupported(),
-                streamHandler, streamOptionsArray, streamAttrsArray);
+                streamHandler, streamOptionsArray, streamAttrsArray, this::removeChannel);
+
         Quic.setupChannel(channel, optionsArray, attrsArray, handler, LOGGER);
         QuicSslEngine engine = sslEngineProvider.apply(channel);
         if (!(engine instanceof QuicheQuicSslEngine)) {