From 830c460c0a2db7ad6d5caed9a0b7e2890952da77 Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Fri, 28 Apr 2023 17:14:01 +0800 Subject: [PATCH] Fix port unification channel leak (#12212) --- .../transport/netty4/NettyChannelHandler.java | 77 +++++++++++++++++++ .../netty4/NettyPortUnificationServer.java | 7 +- .../NettyPortUnificationServerHandler.java | 19 +---- 3 files changed, 83 insertions(+), 20 deletions(-) create mode 100644 dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyChannelHandler.java diff --git a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyChannelHandler.java b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyChannelHandler.java new file mode 100644 index 0000000000..cfedb14371 --- /dev/null +++ b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyChannelHandler.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.remoting.transport.netty4; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.logger.Logger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.common.utils.NetUtils; +import org.apache.dubbo.remoting.Channel; +import org.apache.dubbo.remoting.ChannelHandler; + +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; + +import java.net.InetSocketAddress; +import java.util.Map; + +public class NettyChannelHandler extends ChannelInboundHandlerAdapter { + private static final Logger logger = LoggerFactory.getLogger(NettyChannelHandler.class); + + private final Map dubboChannels; + + private final URL url; + private final ChannelHandler handler; + + public NettyChannelHandler(Map dubboChannels, URL url, ChannelHandler handler) { + this.dubboChannels = dubboChannels; + this.url = url; + this.handler = handler; + } + + @Override + public void channelActive(ChannelHandlerContext ctx) throws Exception { + super.channelActive(ctx); + NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler); + if (channel != null) { + dubboChannels.put(NetUtils.toAddressString((InetSocketAddress) ctx.channel().remoteAddress()), channel); + handler.connected(channel); + + if (logger.isInfoEnabled()) { + logger.info("The connection of " + channel.getRemoteAddress() + " -> " + channel.getLocalAddress() + " is established."); + } + } + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + super.channelInactive(ctx); + NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler); + try { + dubboChannels.remove(NetUtils.toAddressString((InetSocketAddress) ctx.channel().remoteAddress())); + if (channel != null) { + handler.disconnected(channel); + if (logger.isInfoEnabled()) { + logger.info("The connection of " + channel.getRemoteAddress() + " -> " + channel.getLocalAddress() + " is disconnected."); + } + } + } finally { + NettyChannel.removeChannel(ctx.channel()); + } + } + +} diff --git a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServer.java b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServer.java index 239367320a..42d8d89895 100644 --- a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServer.java +++ b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServer.java @@ -122,10 +122,11 @@ public class NettyPortUnificationServer extends AbstractPortUnificationServer { protected void initChannel(SocketChannel ch) throws Exception { // Do not add idle state handler here, because it should be added in the protocol handler. final ChannelPipeline p = ch.pipeline(); - final NettyPortUnificationServerHandler puHandler; - puHandler = new NettyPortUnificationServerHandler(getUrl(), true, getProtocols(), - NettyPortUnificationServer.this, NettyPortUnificationServer.this.dubboChannels, + NettyChannelHandler nettyChannelHandler = new NettyChannelHandler(dubboChannels, getUrl(), NettyPortUnificationServer.this); + NettyPortUnificationServerHandler puHandler = new NettyPortUnificationServerHandler(getUrl(), true, getProtocols(), + NettyPortUnificationServer.this, getSupportedUrls(), getSupportedHandlers()); + p.addLast("channel-handler", nettyChannelHandler); p.addLast("negotiation-protocol", puHandler); } }); diff --git a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServerHandler.java b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServerHandler.java index 5d02ea4e16..cb7e672c67 100644 --- a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServerHandler.java +++ b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServerHandler.java @@ -22,8 +22,6 @@ import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.ssl.CertManager; import org.apache.dubbo.common.ssl.ProviderCert; -import org.apache.dubbo.common.utils.NetUtils; -import org.apache.dubbo.remoting.Channel; import org.apache.dubbo.remoting.ChannelHandler; import org.apache.dubbo.remoting.api.ProtocolDetector; import org.apache.dubbo.remoting.api.WireProtocol; @@ -39,7 +37,6 @@ import io.netty.handler.ssl.SslHandler; import io.netty.handler.ssl.SslHandshakeCompletionEvent; import javax.net.ssl.SSLSession; -import java.net.InetSocketAddress; import java.util.List; import java.util.Map; import java.util.Set; @@ -54,19 +51,17 @@ public class NettyPortUnificationServerHandler extends ByteToMessageDecoder { private final ChannelHandler handler; private final boolean detectSsl; private final List protocols; - private final Map dubboChannels; private final Map urlMapper; private final Map handlerMapper; public NettyPortUnificationServerHandler(URL url, boolean detectSsl, List protocols, ChannelHandler handler, - Map dubboChannels, Map urlMapper, Map handlerMapper) { + Map urlMapper, Map handlerMapper) { this.url = url; this.protocols = protocols; this.detectSsl = detectSsl; this.handler = handler; - this.dubboChannels = dubboChannels; this.urlMapper = urlMapper; this.handlerMapper = handlerMapper; } @@ -76,16 +71,6 @@ public class NettyPortUnificationServerHandler extends ByteToMessageDecoder { LOGGER.error(INTERNAL_ERROR, "unknown error in remoting module", "", "Unexpected exception from downstream before protocol detected.", cause); } - @Override - public void channelActive(ChannelHandlerContext ctx) throws Exception { - super.channelActive(ctx); - NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler); - if (channel != null) { - // this is needed by some test cases - dubboChannels.put(NetUtils.toAddressString((InetSocketAddress) ctx.channel().remoteAddress()), channel); - } - } - @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof SslHandshakeCompletionEvent) { @@ -162,7 +147,7 @@ public class NettyPortUnificationServerHandler extends ByteToMessageDecoder { p.addLast("ssl", sslContext.newHandler(ctx.alloc())); p.addLast("unificationA", new NettyPortUnificationServerHandler(url, false, protocols, - handler, dubboChannels, urlMapper, handlerMapper)); + handler, urlMapper, handlerMapper)); p.remove(this); }