Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-2519 Fix Memory Leak 2.0.x Backport

Browse Source 
When using NIO, a Map of connections is maintained (keyed
by the SocketChannel) and used to manage registrations
for NIO events, timeouts etc.

When a connection is closed, the corresponding entry
should be removed from the map. The code to do this
is in AbstractConnectionFactory.processNioSelections().
However, if no socket timeout (soTimeout) has been set,
or it was explicitly set to 0, connections are not
removed from the map. This was because the timeout
logic and map cleanup is done in the same iteration
loop.

Now, if soTimeout is not set, the clean up loop
operates every nioHarvestInterval milliseconds
(default 2000), we don't want to run it on
every selector event.

In addition, it runs if the selectionCount is
zero - this might occur when a socket
is closed, when the selector.wakeup() is called.
This commit is contained in:
Gary Russell
2012-04-18 10:57:50 -04:00
parent 82c71852a4
commit a57a9e56e6
3 changed files with 163 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -28,6 +28,7 @@ import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
@@ -102,7 +103,13 @@ public abstract class AbstractConnectionFactory
private List<TcpConnection> connections = new LinkedList<TcpConnection>();
protected final Object lifecycleMonitor = new Object();
private volatile long nextCheckForClosedNioConnections;
private volatile int nioHarvestInterval = DEFAULT_NIO_HARVEST_INTERVAL;
private static final int DEFAULT_NIO_HARVEST_INTERVAL = 2000;
/**
* Sets socket attributes on the socket.
* @param socket The socket.
@@ -334,6 +341,18 @@ public abstract class AbstractConnectionFactory
return lookupHost;
}
/**
* How often we clean up closed NIO connections if soTimeout is 0.
* Ignored when soTimeout > 0 because the clean up
* process is run as part of the timeout handling.
* Default 2000 milliseconds.
* @param nioHarvestInterval The interval in milliseconds.
*/
public void setNioHarvestInterval(int nioHarvestInterval) {
Assert.isTrue(nioHarvestInterval > 0, "NIO Harvest interval must be > 0");
this.nioHarvestInterval = nioHarvestInterval;
}
/**
* Closes the server.
*/
@@ -435,24 +454,28 @@ public abstract class AbstractConnectionFactory
/**
*
* Times out any expired connections then, if selectionCount > 0, processes the selected keys.
*
* @param selectionCount
* @param selector
* @param connections
* Removes closed connections from the connections field, and from the connections parameter.
*
* @param selectionCount Number of IO Events, if 0 we were probably woken up by a close.
* @param selector The selector
* @param connections Map of connections
* @throws IOException
*/
protected void processNioSelections(int selectionCount, final Selector selector, ServerSocketChannel server,
Map<SocketChannel, TcpNioConnection> connections) throws IOException {
long now = 0;
if (this.soTimeout > 0) {
Iterator<SocketChannel> it = connections.keySet().iterator();
now = System.currentTimeMillis();
long now = System.currentTimeMillis();
if (this.soTimeout > 0 ||
now >= this.nextCheckForClosedNioConnections ||
selectionCount == 0) {
this.nextCheckForClosedNioConnections = now + this.nioHarvestInterval;
Iterator<Entry<SocketChannel, TcpNioConnection>> it = connections.entrySet().iterator();
while (it.hasNext()) {
SocketChannel channel = it.next();
SocketChannel channel = it.next().getKey();
if (!channel.isOpen()) {
logger.debug("Removing closed channel");
it.remove();
} else {
}
else if (soTimeout > 0) {
TcpNioConnection connection = connections.get(channel);
if (now - connection.getLastRead() > this.soTimeout) {
logger.warn("Timing out TcpNioConnection " +

7
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java
View File

@@ -150,4 +150,11 @@ public class TcpNioClientConnectionFactory extends
return this.active;
}
/**
* @return the connections
*/
protected Map<SocketChannel, TcpNioConnection> getConnections() {
return connections;
}
}

123
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,14 +16,26 @@
package org.springframework.integration.ip.tcp.connection;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Field;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -33,6 +45,10 @@ import javax.net.ServerSocketFactory;
import org.junit.Test;
import org.springframework.integration.ip.util.SocketTestUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.ReflectionUtils.FieldCallback;
import org.springframework.util.ReflectionUtils.FieldFilter;
/**
@@ -110,9 +126,112 @@ public class TcpNioConnectionTests {
} catch (Exception e) {
fail("Unexpected exception " + e);
}
}
@Test
public void testMemoryLeak() throws Exception {
final int port = SocketTestUtils.findAvailableServerSocket();
TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", port);
factory.setNioHarvestInterval(100);
factory.start();
final CountDownLatch latch = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
Socket socket = server.accept();
byte[] b = new byte[6];
readFully(socket.getInputStream(), b);
} catch (Exception e) {
e.printStackTrace();
}
}
});
assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
try {
TcpConnection connection = factory.getConnection();
Map<SocketChannel, TcpNioConnection> connections = factory.getConnections();
assertEquals(1, connections.size());
connection.close();
assertTrue(!connection.isOpen());
int n = 0;
while (connections.size() > 0) {
Thread.sleep(100);
if (n++ > 100) {
break;
}
}
assertEquals(0, connections.size());
} catch (Exception e) {
e.printStackTrace();
fail("Unexpected exception " + e);
}
factory.stop();
}
@Test
public void testCleanup() throws Exception {
TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", 0);
factory.setNioHarvestInterval(100);
Map<SocketChannel, TcpNioConnection> connections = new HashMap<SocketChannel, TcpNioConnection>();
SocketChannel chan1 = mock(SocketChannel.class);
SocketChannel chan2 = mock(SocketChannel.class);
SocketChannel chan3 = mock(SocketChannel.class);
TcpNioConnection conn1 = mock(TcpNioConnection.class);
TcpNioConnection conn2 = mock(TcpNioConnection.class);
TcpNioConnection conn3 = mock(TcpNioConnection.class);
connections.put(chan1, conn1);
connections.put(chan2, conn2);
connections.put(chan3, conn3);
final List<Field> fields = new ArrayList<Field>();
ReflectionUtils.doWithFields(SocketChannel.class, new FieldCallback() {
public void doWith(Field field) throws IllegalArgumentException,
IllegalAccessException {
field.setAccessible(true);
fields.add(field);
}
}, new FieldFilter() {
public boolean matches(Field field) {
return field.getName().equals("open");
}});
Field field = fields.get(0);
// Can't use Mockito because isOpen() is final
ReflectionUtils.setField(field, chan1, true);
ReflectionUtils.setField(field, chan2, true);
ReflectionUtils.setField(field, chan3, true);
Selector selector = mock(Selector.class);
HashSet<SelectionKey> keys = new HashSet<SelectionKey>();
when(selector.selectedKeys()).thenReturn(keys);
factory.processNioSelections(1, selector, null, connections);
assertEquals(3, connections.size()); // all open
ReflectionUtils.setField(field, chan1, false);
factory.processNioSelections(1, selector, null, connections);
assertEquals(3, connections.size()); // interval didn't pass
Thread.sleep(110);
factory.processNioSelections(1, selector, null, connections);
assertEquals(2, connections.size()); // first is closed
ReflectionUtils.setField(field, chan2, false);
factory.processNioSelections(1, selector, null, connections);
assertEquals(2, connections.size()); // interval didn't pass
Thread.sleep(110);
factory.processNioSelections(1, selector, null, connections);
assertEquals(1, connections.size()); // second is closed
ReflectionUtils.setField(field, chan3, false);
factory.processNioSelections(1, selector, null, connections);
assertEquals(1, connections.size()); // interval didn't pass
Thread.sleep(110);
factory.processNioSelections(1, selector, null, connections);
assertEquals(0, connections.size()); // third is closed
assertEquals(0, TestUtils.getPropertyValue(factory, "connections", List.class).size());
}
private void readFully(InputStream is, byte[] buff) throws IOException {
for (int i = 0; i < buff.length; i++) {
buff[i] = (byte) is.read();