natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix memory leak when hashed wheel time re-put (#12022)

This commit is contained in:
Albumen Kevin 2023-04-11 19:00:02 +08:00 committed by GitHub
parent d4032660d5
commit 7facd5be19
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 105 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectory.java
View File

@ -16,8 +16,6 @@
*/
package org.apache.dubbo.rpc.cluster.directory;
import java.util.List;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
@ -29,6 +27,8 @@ import org.apache.dubbo.rpc.cluster.RouterChain;
import org.apache.dubbo.rpc.cluster.SingleRouterChain;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import java.util.List;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.CLUSTER_FAILED_SITE_SELECTION;
/**
@ -75,6 +75,8 @@ public class StaticDirectory<T> extends AbstractDirectory<T> {
for (Invoker<T> invoker : getValidInvokers()) {
if (invoker.isAvailable()) {
return true;
} else {
addInvalidateInvoker(invoker);
}
}
return false;

10
dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java
View File

@ -26,8 +26,8 @@ import org.apache.dubbo.rpc.cluster.router.MockInvoker;
import org.apache.dubbo.rpc.cluster.router.condition.ConditionStateRouterFactory;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.cluster.router.state.StateRouter;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@ -52,9 +52,9 @@ class StaticDirectoryTest {
List<StateRouter> routers = new ArrayList<StateRouter>();
routers.add(router);
List<Invoker<String>> originInvokers = new ArrayList<Invoker<String>>();
Invoker<String> invoker1 = new MockInvoker<String>(URL.valueOf("dubbo://10.20.3.3:20880/com.foo.BarService"));
Invoker<String> invoker2 = new MockInvoker<String>(URL.valueOf("dubbo://" + NetUtils.getLocalHost() + ":20880/com.foo.BarService"));
Invoker<String> invoker3 = new MockInvoker<String>(URL.valueOf("dubbo://" + NetUtils.getLocalHost() + ":20880/com.foo.BarService"));
Invoker<String> invoker1 = new MockInvoker<String>(URL.valueOf("dubbo://10.20.3.3:20880/com.foo.BarService"), true);
Invoker<String> invoker2 = new MockInvoker<String>(URL.valueOf("dubbo://" + NetUtils.getLocalHost() + ":20880/com.foo.BarService"), true);
Invoker<String> invoker3 = new MockInvoker<String>(URL.valueOf("dubbo://" + NetUtils.getLocalHost() + ":20880/com.foo.BarService"), true);
originInvokers.add(invoker1);
originInvokers.add(invoker2);
originInvokers.add(invoker3);
@ -65,7 +65,7 @@ class StaticDirectoryTest {
ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
StaticDirectory<String> staticDirectory = new StaticDirectory<>(filteredInvokers);
boolean isAvailable = staticDirectory.isAvailable();
Assertions.assertTrue(!isAvailable);
Assertions.assertTrue(isAvailable);
List<Invoker<String>> newInvokers = staticDirectory.list(new MockDirInvocation());
Assertions.assertTrue(newInvokers.size() > 0);
staticDirectory.destroy();

10
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java
View File

@ -287,8 +287,14 @@ public abstract class DynamicDirectory<T> extends AbstractDirectory<T> implement
if (isDestroyed() || this.forbidden) {
return false;
}
return CollectionUtils.isNotEmpty(getValidInvokers())
&& getValidInvokers().stream().anyMatch(Invoker::isAvailable);
for (Invoker<T> validInvoker : getValidInvokers()) {
if (validInvoker.isAvailable()) {
return true;
} else {
addInvalidateInvoker(validInvoker);
}
}
return false;
}
@Override

44
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/AbstractTimerTask.java
View File

@ -17,8 +17,8 @@
package org.apache.dubbo.remoting.exchange.support.header;
import org.apache.dubbo.common.timer.HashedWheelTimer;
import org.apache.dubbo.common.timer.Timeout;
import org.apache.dubbo.common.timer.Timer;
import org.apache.dubbo.common.timer.TimerTask;
import org.apache.dubbo.remoting.Channel;
@ -31,17 +31,22 @@ import java.util.concurrent.TimeUnit;
public abstract class AbstractTimerTask implements TimerTask {
private final ChannelProvider channelProvider;
private final HashedWheelTimer hashedWheelTimer;
private final Long tick;
protected volatile boolean cancel = false;
AbstractTimerTask(ChannelProvider channelProvider, Long tick) {
if (channelProvider == null || tick == null) {
private volatile Timeout timeout;
AbstractTimerTask(ChannelProvider channelProvider, HashedWheelTimer hashedWheelTimer, Long tick) {
if (channelProvider == null || hashedWheelTimer == null || tick == null) {
throw new IllegalArgumentException();
}
this.tick = tick;
this.channelProvider = channelProvider;
this.hashedWheelTimer = hashedWheelTimer;
this.tick = tick;
start();
}
static Long lastRead(Channel channel) {
@ -56,12 +61,17 @@ public abstract class AbstractTimerTask implements TimerTask {
return System.currentTimeMillis();
}
public void cancel() {
this.cancel = true;
private void start() {
this.timeout = hashedWheelTimer.newTimeout(this, tick, TimeUnit.MILLISECONDS);
}
private void reput(Timeout timeout, Long tick) {
if (timeout == null || tick == null) {
public synchronized void cancel() {
this.cancel = true;
this.timeout.cancel();
}
private synchronized void reput(Timeout timeout) {
if (timeout == null) {
throw new IllegalArgumentException();
}
@ -69,24 +79,22 @@ public abstract class AbstractTimerTask implements TimerTask {
return;
}
Timer timer = timeout.timer();
if (timer.isStop() || timeout.isCancelled()) {
if (hashedWheelTimer.isStop() || timeout.isCancelled()) {
return;
}
timer.newTimeout(timeout.task(), tick, TimeUnit.MILLISECONDS);
this.timeout = hashedWheelTimer.newTimeout(timeout.task(), tick, TimeUnit.MILLISECONDS);
}
@Override
public void run(Timeout timeout) throws Exception {
Collection<Channel> c = channelProvider.getChannels();
for (Channel channel : c) {
if (channel.isClosed()) {
continue;
public synchronized void run(Timeout timeout) throws Exception {
Collection<Channel> channels = channelProvider.getChannels();
for (Channel channel : channels) {
if (!channel.isClosed()) {
doTask(channel);
}
doTask(channel);
}
reput(timeout, tick);
reput(timeout);
}
protected abstract void doTask(Channel channel);

5
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/CloseTimerTask.java
View File

@ -19,6 +19,7 @@ package org.apache.dubbo.remoting.exchange.support.header;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.timer.HashedWheelTimer;
import org.apache.dubbo.remoting.Channel;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_RESPONSE;
@ -33,8 +34,8 @@ public class CloseTimerTask extends AbstractTimerTask {
private final int idleTimeout;
public CloseTimerTask(ChannelProvider channelProvider, Long heartbeatTimeoutTick, int idleTimeout) {
super(channelProvider, heartbeatTimeoutTick);
public CloseTimerTask(ChannelProvider channelProvider, HashedWheelTimer hashedWheelTimer, Long heartbeatTimeoutTick, int idleTimeout) {
super(channelProvider, hashedWheelTimer, heartbeatTimeoutTick);
this.idleTimeout = idleTimeout;
}

47
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeClient.java
View File

@ -19,7 +19,6 @@ package org.apache.dubbo.remoting.exchange.support.header;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.resource.GlobalResourceInitializer;
import org.apache.dubbo.common.timer.HashedWheelTimer;
import org.apache.dubbo.common.timer.Timeout;
import org.apache.dubbo.common.utils.Assert;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.remoting.ChannelHandler;
@ -51,12 +50,13 @@ public class HeaderExchangeClient implements ExchangeClient {
private final ExchangeChannel channel;
public static GlobalResourceInitializer<HashedWheelTimer> IDLE_CHECK_TIMER = new GlobalResourceInitializer<>(() ->
new HashedWheelTimer(new NamedThreadFactory("dubbo-client-idleCheck", true), 1,
new HashedWheelTimer(new NamedThreadFactory("dubbo-client-heartbeat-reconnect", true), 1,
TimeUnit.SECONDS, TICKS_PER_WHEEL),
HashedWheelTimer::stop);
private Timeout reconnectTimer;
private Timeout heartBeatTimer;
private ReconnectTimerTask reconnectTimerTask;
private HeartbeatTimerTask heartBeatTimerTask;
private final int idleTimeout;
public HeaderExchangeClient(Client client, boolean startTimer) {
Assert.notNull(client, "Client can't be null");
@ -65,8 +65,11 @@ public class HeaderExchangeClient implements ExchangeClient {
if (startTimer) {
URL url = client.getUrl();
idleTimeout = getIdleTimeout(url);
startReconnectTask(url);
startHeartBeatTask(url);
} else {
idleTimeout = 0;
}
}
@ -107,7 +110,16 @@ public class HeaderExchangeClient implements ExchangeClient {
@Override
public boolean isConnected() {
return channel.isConnected();
if (channel.isConnected()) {
if (idleTimeout <= 0) {
return true;
}
Long lastRead = (Long) channel.getAttribute(HeartbeatHandler.KEY_READ_TIMESTAMP);
Long now = System.currentTimeMillis();
return lastRead == null || now - lastRead < idleTimeout;
}
return false;
}
@Override
@ -136,7 +148,7 @@ public class HeaderExchangeClient implements ExchangeClient {
}
@Override
public void close() {
public synchronized void close() {
doClose();
channel.close();
}
@ -167,7 +179,7 @@ public class HeaderExchangeClient implements ExchangeClient {
}
@Override
public void reconnect() throws RemotingException {
public synchronized void reconnect() throws RemotingException {
client.reconnect();
}
@ -193,32 +205,27 @@ public class HeaderExchangeClient implements ExchangeClient {
private void startHeartBeatTask(URL url) {
if (!client.canHandleIdle()) {
AbstractTimerTask.ChannelProvider cp = () -> Collections.singletonList(HeaderExchangeClient.this);
int heartbeat = getHeartbeat(url);
long heartbeatTick = calculateLeastDuration(heartbeat);
HeartbeatTimerTask heartBeatTimerTask = new HeartbeatTimerTask(cp, heartbeatTick, heartbeat);
heartBeatTimer = IDLE_CHECK_TIMER.get().newTimeout(heartBeatTimerTask, heartbeatTick, TimeUnit.MILLISECONDS);
heartBeatTimerTask = new HeartbeatTimerTask(() -> Collections.singleton(this), IDLE_CHECK_TIMER.get(), heartbeatTick, heartbeat);
}
}
private void startReconnectTask(URL url) {
if (shouldReconnect(url)) {
AbstractTimerTask.ChannelProvider cp = () -> Collections.singletonList(HeaderExchangeClient.this);
int idleTimeout = getIdleTimeout(url);
long heartbeatTimeoutTick = calculateLeastDuration(idleTimeout);
ReconnectTimerTask reconnectTimerTask = new ReconnectTimerTask(cp, heartbeatTimeoutTick, idleTimeout);
reconnectTimer = IDLE_CHECK_TIMER.get().newTimeout(reconnectTimerTask, heartbeatTimeoutTick, TimeUnit.MILLISECONDS);
reconnectTimerTask = new ReconnectTimerTask(() -> Collections.singleton(this), IDLE_CHECK_TIMER.get(), heartbeatTimeoutTick, idleTimeout);
}
}
private void doClose() {
if (heartBeatTimer != null) {
heartBeatTimer.cancel();
heartBeatTimer = null;
if (heartBeatTimerTask != null) {
heartBeatTimerTask.cancel();
heartBeatTimerTask = null;
}
if (reconnectTimer != null) {
reconnectTimer.cancel();
reconnectTimer = null;
if (reconnectTimerTask != null) {
reconnectTimerTask.cancel();
reconnectTimerTask = null;
}
}

12
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeServer.java
View File

@ -22,7 +22,6 @@ import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.resource.GlobalResourceInitializer;
import org.apache.dubbo.common.timer.HashedWheelTimer;
import org.apache.dubbo.common.timer.Timeout;
import org.apache.dubbo.common.utils.Assert;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.NamedThreadFactory;
@ -44,9 +43,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
import static java.util.Collections.unmodifiableCollection;
import static org.apache.dubbo.common.constants.CommonConstants.READONLY_EVENT;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_CLOSE;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_RESPONSE;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR;
import static org.apache.dubbo.remoting.Constants.HEARTBEAT_CHECK_TICK;
import static org.apache.dubbo.remoting.Constants.LEAST_HEARTBEAT_DURATION;
import static org.apache.dubbo.remoting.Constants.TICKS_PER_WHEEL;
@ -66,7 +65,7 @@ public class HeaderExchangeServer implements ExchangeServer {
public static GlobalResourceInitializer<HashedWheelTimer> IDLE_CHECK_TIMER = new GlobalResourceInitializer<>(() -> new HashedWheelTimer(new NamedThreadFactory("dubbo-server-idleCheck", true), 1, TimeUnit.SECONDS, TICKS_PER_WHEEL), HashedWheelTimer::stop);
private Timeout closeTimer;
private CloseTimerTask closeTimer;
public HeaderExchangeServer(RemotingServer server) {
Assert.notNull(server, "server == null");
@ -141,7 +140,7 @@ public class HeaderExchangeServer implements ExchangeServer {
}
} catch (RemotingException e) {
if (closed.get() && e.getCause() instanceof ClosedChannelException) {
// ignore ClosedChannelException which means the connection has been closed.
// ignore ClosedChannelException which means the connection has been closed.
continue;
}
logger.warn(TRANSPORT_FAILED_RESPONSE, "", "", "send cannot write message error.", e);
@ -265,10 +264,7 @@ public class HeaderExchangeServer implements ExchangeServer {
AbstractTimerTask.ChannelProvider cp = () -> unmodifiableCollection(HeaderExchangeServer.this.getChannels());
int idleTimeout = getIdleTimeout(url);
long idleTimeoutTick = calculateLeastDuration(idleTimeout);
CloseTimerTask closeTimerTask = new CloseTimerTask(cp, idleTimeoutTick, idleTimeout);
// init task and start timer.
this.closeTimer = IDLE_CHECK_TIMER.get().newTimeout(closeTimerTask, idleTimeoutTick, TimeUnit.MILLISECONDS);
this.closeTimer = new CloseTimerTask(cp, IDLE_CHECK_TIMER.get(), idleTimeoutTick, idleTimeout);
}
}
}

5
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeartbeatTimerTask.java
View File

@ -20,6 +20,7 @@ package org.apache.dubbo.remoting.exchange.support.header;
import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.timer.HashedWheelTimer;
import org.apache.dubbo.remoting.Channel;
import org.apache.dubbo.remoting.exchange.Request;
@ -35,8 +36,8 @@ public class HeartbeatTimerTask extends AbstractTimerTask {
private final int heartbeat;
HeartbeatTimerTask(ChannelProvider channelProvider, Long heartbeatTick, int heartbeat) {
super(channelProvider, heartbeatTick);
HeartbeatTimerTask(ChannelProvider channelProvider, HashedWheelTimer hashedWheelTimer, Long heartbeatTick, int heartbeat) {
super(channelProvider, hashedWheelTimer, heartbeatTick);
this.heartbeat = heartbeat;
}

7
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/ReconnectTimerTask.java
View File

@ -19,11 +19,12 @@ package org.apache.dubbo.remoting.exchange.support.header;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.timer.HashedWheelTimer;
import org.apache.dubbo.remoting.Channel;
import org.apache.dubbo.remoting.Client;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_RECONNECT;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_RECONNECT;
/**
* ReconnectTimerTask
@ -34,8 +35,8 @@ public class ReconnectTimerTask extends AbstractTimerTask {
private final int idleTimeout;
public ReconnectTimerTask(ChannelProvider channelProvider, Long heartbeatTimeoutTick, int idleTimeout) {
super(channelProvider, heartbeatTimeoutTick);
public ReconnectTimerTask(ChannelProvider channelProvider, HashedWheelTimer hashedWheelTimer, Long heartbeatTimeoutTick, int idleTimeout) {
super(channelProvider, hashedWheelTimer, heartbeatTimeoutTick);
this.idleTimeout = idleTimeout;
}

8
dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/CloseTimerTaskTest.java
View File

@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.timer.HashedWheelTimer;
import org.apache.dubbo.remoting.Channel;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@ -54,7 +55,12 @@ class CloseTimerTaskTest {
};
AbstractTimerTask.ChannelProvider cp = () -> Collections.<Channel>singletonList(channel);
closeTimerTask = new CloseTimerTask(cp, tickDuration / HEARTBEAT_CHECK_TICK, (int) tickDuration);
closeTimerTask = new CloseTimerTask(cp, closeTimer, tickDuration / HEARTBEAT_CHECK_TICK, (int) tickDuration);
}
@AfterEach
public void teardown() {
closeTimerTask.cancel();
}
@Test

12
dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/HeartBeatTaskTest.java
View File

@ -19,9 +19,9 @@ package org.apache.dubbo.remoting.exchange.support.header;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.timer.HashedWheelTimer;
import org.apache.dubbo.remoting.Channel;
import org.apache.dubbo.remoting.exchange.Request;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@ -55,8 +55,12 @@ class HeartBeatTaskTest {
}
};
AbstractTimerTask.ChannelProvider cp = () -> Collections.<Channel>singletonList(channel);
heartbeatTimerTask = new HeartbeatTimerTask(cp, tickDuration / HEARTBEAT_CHECK_TICK, (int) tickDuration);
heartbeatTimerTask = new HeartbeatTimerTask(() -> Collections.singleton(channel), heartbeatTimer, tickDuration / HEARTBEAT_CHECK_TICK, (int) tickDuration);
}
@AfterEach
public void teardown() {
heartbeatTimerTask.cancel();
}
@Test
@ -67,8 +71,6 @@ class HeartBeatTaskTest {
channel.setAttribute(HeartbeatHandler.KEY_READ_TIMESTAMP, now);
channel.setAttribute(HeartbeatHandler.KEY_WRITE_TIMESTAMP, now);
heartbeatTimer.newTimeout(heartbeatTimerTask, 250, TimeUnit.MILLISECONDS);
Thread.sleep(2000L);
List<Object> objects = channel.getSentObjects();
Assertions.assertTrue(objects.size() > 0);

12
dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/exchange/support/header/ReconnectTimerTaskTest.java
View File

@ -18,8 +18,8 @@ package org.apache.dubbo.remoting.exchange.support.header;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.timer.HashedWheelTimer;
import org.apache.dubbo.remoting.Channel;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@ -56,8 +56,12 @@ class ReconnectTimerTaskTest {
}
};
AbstractTimerTask.ChannelProvider cp = () -> Collections.<Channel>singletonList(channel);
reconnectTimerTask = new ReconnectTimerTask(cp, tickDuration / HEARTBEAT_CHECK_TICK, (int) tickDuration);
reconnectTimerTask = new ReconnectTimerTask(() -> Collections.singleton(channel), reconnectTimer, tickDuration / HEARTBEAT_CHECK_TICK, (int) tickDuration);
}
@AfterEach
public void teardown() {
reconnectTimerTask.cancel();
}
@Test
@ -68,8 +72,6 @@ class ReconnectTimerTaskTest {
channel.setAttribute(HeartbeatHandler.KEY_READ_TIMESTAMP, now - 1000);
channel.setAttribute(HeartbeatHandler.KEY_WRITE_TIMESTAMP, now - 1000);
reconnectTimer.newTimeout(reconnectTimerTask, 250, TimeUnit.MILLISECONDS);
Thread.sleep(2000L);
Assertions.assertTrue(channel.getReconnectCount() > 0);
isConnected = true;