From a57a9e56e64a313f654f0c5fc0832861552e690b Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 18 Apr 2012 10:57:50 -0400 Subject: [PATCH] INT-2519 Fix Memory Leak 2.0.x Backport 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. --- .../connection/AbstractConnectionFactory.java | 47 +++++-- .../TcpNioClientConnectionFactory.java | 7 + .../tcp/connection/TcpNioConnectionTests.java | 123 +++++++++++++++++- 3 files changed, 163 insertions(+), 14 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index a6f2c4e690..79c49416e8 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -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 connections = new LinkedList(); 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 connections) throws IOException { - long now = 0; - if (this.soTimeout > 0) { - Iterator 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> 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 " + diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java index 9a1095fa24..9c375ce955 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java @@ -150,4 +150,11 @@ public class TcpNioClientConnectionFactory extends return this.active; } + /** + * @return the connections + */ + protected Map getConnections() { + return connections; + } + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java index 0431d6b494..0b6b412cd2 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java @@ -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 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 connections = new HashMap(); + 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 fields = new ArrayList(); + 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 keys = new HashSet(); + 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();