From 303a5aff45e38249c00603a3aebad1b04489f648 Mon Sep 17 00:00:00 2001 From: GuoHao Date: Tue, 13 Apr 2021 15:45:36 +0800 Subject: [PATCH] Fix mem leak caused by duplicate reconnect task (#7537) --- .../apache/dubbo/remoting/api/Connection.java | 56 +++++++++++++++---- .../dubbo/remoting/api/ConnectionHandler.java | 16 +++++- .../remoting/api/ConnectionListener.java | 48 ---------------- 3 files changed, 59 insertions(+), 61 deletions(-) delete mode 100644 dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionListener.java diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Connection.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Connection.java index 24833aab80..5b7d5d3211 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Connection.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Connection.java @@ -29,9 +29,11 @@ import io.netty.bootstrap.Bootstrap; import io.netty.buffer.PooledByteBufAllocator; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.ChannelPipeline; +import io.netty.channel.EventLoop; import io.netty.channel.socket.SocketChannel; import io.netty.util.AbstractReferenceCounted; import io.netty.util.AttributeKey; @@ -42,6 +44,7 @@ import io.netty.util.concurrent.GlobalEventExecutor; import io.netty.util.concurrent.Promise; import java.net.InetSocketAddress; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -61,6 +64,8 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou private final AtomicBoolean closed = new AtomicBoolean(false); private final AtomicReference channel = new AtomicReference<>(); private final ChannelFuture initPromise; + private final Bootstrap bootstrap; + private final ConnectionListener connectionListener = new ConnectionListener(); public Connection(URL url) { url = ExecutorUtil.setThreadName(url, "DubboClientHandler"); @@ -70,6 +75,7 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou this.connectTimeout = Math.max(3000, url.getPositiveParameter(Constants.CONNECT_TIMEOUT_KEY, Constants.DEFAULT_CONNECT_TIMEOUT)); this.closeFuture = new DefaultPromise<>(GlobalEventExecutor.INSTANCE); this.remote = getConnectAddress(); + this.bootstrap = create(); this.initPromise = connect(); } @@ -81,11 +87,7 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou return closeFuture; } - - public ChannelFuture connect() { - if (isClosed()) { - return null; - } + private Bootstrap create() { final Bootstrap bootstrap = new Bootstrap(); bootstrap.group(NettyEventLoopFactory.NIO_EVENT_LOOP_GROUP) .option(ChannelOption.SO_KEEPALIVE, true) @@ -112,8 +114,19 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou // TODO support Socks5 } }); + return bootstrap; + } + + public ChannelFuture connect() { + if (isClosed()) { + if (logger.isInfoEnabled()) { + logger.info(String.format("%s aborted to reconnect cause connection closed. ", Connection.this)); + } + return null; + } + final ChannelFuture promise = bootstrap.connect(); - promise.addListener(new ConnectionListener(this)); + promise.addListener(this.connectionListener); return promise; } @@ -123,13 +136,13 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou @Override public String toString() { - return "(Ref=" + ReferenceCountUtil.refCnt(this) + ",local=" + (getChannel() == null ? null : getChannel().localAddress()) + ",remote=" + getRemote(); + return super.toString() + " (Ref=" + ReferenceCountUtil.refCnt(this) + ",local=" + (getChannel() == null ? null : getChannel().localAddress()) + ",remote=" + getRemote(); } public void onGoaway(Channel channel) { if (this.channel.compareAndSet(channel, null)) { if (logger.isInfoEnabled()) { - logger.info(String.format("Connection:%s goaway", this)); + logger.info(String.format("%s goaway", this)); } } } @@ -138,7 +151,7 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou this.channel.set(channel); channel.attr(CONNECTION).set(this); if (logger.isInfoEnabled()) { - logger.info(String.format("Connection:%s connected ", this)); + logger.info(String.format("%s connected ", this)); } } @@ -175,6 +188,9 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou } public void close() { + if (logger.isInfoEnabled()) { + logger.info(String.format("Connection:%s freed ", this)); + } final Channel current = this.channel.get(); if (current != null) { current.close(); @@ -200,8 +216,26 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou return url; } - private int getConnectTimeout() { - return connectTimeout; + class ConnectionListener implements ChannelFutureListener { + + @Override + public void operationComplete(ChannelFuture future) { + if (future.isSuccess()) { + return; + } + final Connection conn = Connection.this; + if (conn.isClosed() || conn.refCnt() == 0) { + if (logger.isInfoEnabled()) { + logger.info(String.format("%s aborted to reconnect. %s", conn, future.cause().getMessage())); + } + return; + } + if (logger.isInfoEnabled()) { + logger.info(String.format("%s is reconnecting, attempt=%d cause=%s", conn, 0, future.cause().getMessage())); + } + final EventLoop loop = future.channel().eventLoop(); + loop.schedule((Runnable) conn::connect, 1L, TimeUnit.SECONDS); + } } } diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionHandler.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionHandler.java index 47dca2489e..bdb27b7839 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionHandler.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionHandler.java @@ -24,6 +24,7 @@ import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.EventLoop; +import io.netty.util.Attribute; import io.netty.util.AttributeKey; import java.util.concurrent.TimeUnit; @@ -39,10 +40,18 @@ public class ConnectionHandler extends ChannelInboundHandlerAdapter { } public void onGoAway(Channel channel) { - channel.attr(GO_AWAY_KEY).set(true); + final Attribute attr = channel.attr(GO_AWAY_KEY); + if (Boolean.TRUE.equals(attr.get())) { + return; + } + + attr.set(true); if (connection != null) { connection.onGoaway(channel); } + if (log.isInfoEnabled()) { + log.info(String.format("Channel %s go away ,schedule reconnect", channel)); + } reconnect(channel); } @@ -65,8 +74,11 @@ public class ConnectionHandler extends ChannelInboundHandlerAdapter { @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { - reconnect(ctx.channel()); super.channelInactive(ctx); + final Attribute goawayAttr = ctx.channel().attr(GO_AWAY_KEY); + if (!Boolean.TRUE.equals(goawayAttr.get())) { + reconnect(ctx.channel()); + } } private void reconnect(Channel channel) { diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionListener.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionListener.java deleted file mode 100644 index a379ce00bf..0000000000 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/ConnectionListener.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - * 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.api; - -import org.apache.dubbo.common.logger.Logger; -import org.apache.dubbo.common.logger.LoggerFactory; - -import io.netty.channel.ChannelFuture; -import io.netty.channel.ChannelFutureListener; -import io.netty.channel.EventLoop; - -import java.util.concurrent.TimeUnit; - -public class ConnectionListener implements ChannelFutureListener { - private static final Logger log = LoggerFactory.getLogger(ConnectionListener.class); - - private final Connection connection; - - public ConnectionListener(Connection connection) { - this.connection = connection; - } - - @Override - public void operationComplete(ChannelFuture channelFuture) { - if (!channelFuture.isSuccess()) { - if (log.isInfoEnabled()) { - log.info(String.format("Connection %s is reconnecting, attempt=%d", connection, 0)); - } - final EventLoop loop = channelFuture.channel().eventLoop(); - loop.schedule((Runnable) connection::connect, 1L, TimeUnit.SECONDS); - } - } -} \ No newline at end of file