From badb9e11dd02cd8f7b51c6f61c2a583507e93b91 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 20 Aug 2013 09:34:55 -0400 Subject: [PATCH] INT-3112 Fix OOM in SimplePool The SimplePool maintains an 'allocated' set, for the sole reason of preventing a "foreign" (non-managed) object being returned. When a pool item is detected as stale, it is removed from the pool but remains in the 'allocated' set. Add a test to verify the allocated size is reduced when a stale item is popped from the pool. Add a FileTransferringMessageHandler test (where the problem was discovered). Fix a test in TCP to expect a close(). Polishing Do not allow returning null items to just release a permit - it cannot remove the item from allocated. Clients must return the stale item to the pool so it can be refreshed on the next get. Only used by the TCP caching CF when returning a connection when the factory is not running; but should not be allowed. Also, protect against double release - not currently an issue with existing users of SimplePool, but should be protected against. Could cause the permit count to exceed the pool size. Add inUse Set to the pool so we can detect attempts to release an item that has already been released. --- .../integration/util/SimplePool.java | 39 ++++++++++----- .../integration/util/SimplePoolTests.java | 22 ++++++++- .../FileTransferringMessageHandlerTests.java | 48 ++++++++++++++++++- .../CachingClientConnectionFactory.java | 23 +++++---- .../CachingClientConnectionFactoryTests.java | 25 +++++++--- 5 files changed, 124 insertions(+), 33 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java index 3ce434d996..5db999334f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java @@ -53,6 +53,8 @@ public class SimplePool implements Pool { private final Set allocated = Collections.synchronizedSet(new HashSet()); + private final Set inUse = Collections.synchronizedSet(new HashSet()); + private final PoolItemCallback callback; /** @@ -124,7 +126,7 @@ public class SimplePool implements Pool { } public int getActiveCount() { - return this.getAllocatedCount() - this.getIdleCount(); + return this.inUse.size(); } public int getAllocatedCount() { @@ -187,32 +189,42 @@ public class SimplePool implements Pool { if (logger.isDebugEnabled()) { logger.debug("Received a stale item, will attempt to get a new one."); } + doRemoveItem(item); item = doGetItem(); } + this.inUse.add(item); return item; } /** - * Returns an item to the pool. Item may be null, in which case a subsequent getItem() - * will return a new instance. + * Returns an item to the pool. */ public synchronized void releaseItem(T item) { - Assert.isTrue(item == null || this.allocated.contains(item), + Assert.notNull(item, "Item cannot be null"); + Assert.isTrue(this.allocated.contains(item), "You can only release items that were obtained from the pool"); - if (this.poolSize.get() > targetPoolSize.get()) { - poolSize.decrementAndGet(); - if (item != null) { - doRemoveItem(item); + if (this.inUse.contains(item)) { + if (this.poolSize.get() > targetPoolSize.get()) { + poolSize.decrementAndGet(); + if (item != null) { + doRemoveItem(item); + } + } + else { + if (logger.isDebugEnabled()){ + logger.debug("Releasing " + item + " back to the pool"); + } + if (item != null) { + this.available.add(item); + this.inUse.remove(item); + } + permits.release(); } } else { if (logger.isDebugEnabled()){ - logger.debug("Releasing " + item + " back to the pool"); + logger.debug("Ignoring release of " + item + " back to the pool - not in use"); } - if (item != null) { - available.add(item); - } - permits.release(); } } @@ -225,6 +237,7 @@ public class SimplePool implements Pool { private void doRemoveItem(T item) { this.allocated.remove(item); + this.inUse.remove(item); this.callback.removedFromPool(item); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java b/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java index 0b2569de99..fc7a5cfb13 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -23,10 +23,13 @@ import static org.junit.Assert.fail; import java.util.HashSet; import java.util.Set; +import java.util.concurrent.Semaphore; import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Test; + import org.springframework.integration.MessagingException; +import org.springframework.integration.test.util.TestUtils; /** * @author Gary Russell @@ -51,6 +54,7 @@ public class SimplePoolTests { s3 = pool.getItem(); assertNotSame(s1, s3); assertFalse(strings.remove(s1)); + assertEquals(2, pool.getAllocatedCount()); } @Test @@ -128,6 +132,22 @@ public class SimplePoolTests { pool.releaseItem("Hello, world!"); } + @Test + public void testDoubleReturn() { + final Set strings = new HashSet(); + final AtomicBoolean stale = new AtomicBoolean(); + SimplePool pool = stringPool(2, strings, stale); + Semaphore permits = TestUtils.getPropertyValue(pool, "permits", Semaphore.class); + assertEquals(2, permits.availablePermits()); + String s1 = pool.getItem(); + assertEquals(1, permits.availablePermits()); + pool.releaseItem(s1); + assertEquals(2, permits.availablePermits()); + pool.releaseItem(s1); + assertEquals(2, permits.availablePermits()); + } + + private SimplePool stringPool(int size, final Set strings, final AtomicBoolean stale) { SimplePool pool = new SimplePool(size, new SimplePool.PoolItemCallback() { diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandlerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandlerTests.java index 466fddd5c9..ee2299a06a 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandlerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandlerTests.java @@ -16,29 +16,39 @@ package org.springframework.integration.file.remote.handler; -import static junit.framework.Assert.assertFalse; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyString; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.io.IOException; import java.io.InputStream; +import java.util.Set; import java.util.concurrent.atomic.AtomicReference; import org.junit.Test; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; + import org.springframework.expression.ExpressionParser; import org.springframework.expression.common.LiteralExpression; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.Message; +import org.springframework.integration.file.remote.session.CachingSessionFactory; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.util.SimplePool; /** * @author Oleg Zhurakousky @@ -149,4 +159,40 @@ public class FileTransferringMessageHandlerTests { verify(session, times(0)).rename(Mockito.anyString(), Mockito.anyString()); } + @SuppressWarnings("unchecked") + @Test + public void testServerException() throws Exception{ + SessionFactory sf = mock(SessionFactory.class); + CachingSessionFactory csf = new CachingSessionFactory(sf, 2); + FileTransferringMessageHandler handler = new FileTransferringMessageHandler(csf); + Session session1 = newSession(); + Session session2 = newSession(); + Session session3 = newSession(); + when(sf.getSession()).thenReturn(session1, session2, session3); + handler.setRemoteDirectoryExpression(new LiteralExpression("foo")); + handler.afterPropertiesSet(); + for (int i = 0; i < 3; i++) { + try { + handler.handleMessage(new GenericMessage("hello")); + } + catch (Exception e) { + assertEquals("test", e.getCause().getCause().getMessage()); + } + } + verify(session1, times(1)).write(Mockito.any(InputStream.class), Mockito.anyString()); + verify(session2, times(1)).write(Mockito.any(InputStream.class), Mockito.anyString()); + verify(session3, times(1)).write(Mockito.any(InputStream.class), Mockito.anyString()); + SimplePool pool = TestUtils.getPropertyValue(csf, "pool", SimplePool.class); + assertEquals(1, pool.getAllocatedCount()); + assertEquals(1, pool.getIdleCount()); + assertSame(session3, TestUtils.getPropertyValue(pool, "allocated", Set.class).iterator().next()); + } + + private Session newSession() throws IOException { + @SuppressWarnings("unchecked") + Session session = mock(Session.class); + doThrow(new IOException("test")).when(session).write(any(InputStream.class), anyString()); + when(session.isOpen()).thenReturn(false); + return session; + } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java index 466b568995..2c4d01328a 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java @@ -109,23 +109,22 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact @Override public synchronized void close() { - /** - * If the delegate is stopped, actually close - * the connection. - */ - if (!isRunning()) { - if (logger.isDebugEnabled()){ - logger.debug("Factory not running - closing " + this.getConnectionId()); - } - pool.releaseItem(null); // just open up a permit - super.close(); - } - else if(this.released) { + if(this.released) { if (logger.isDebugEnabled()) { logger.debug("Connection " + this.getConnectionId() + " has already been released"); } } else { + /** + * If the delegate is stopped, actually close the connection, but still release + * it to the pool, it will be discarded/renewed the next time it is retrieved. + */ + if (!isRunning()) { + if (logger.isDebugEnabled()){ + logger.debug("Factory not running - closing " + this.getConnectionId()); + } + super.close(); + } pool.releaseItem(this.getTheConnection()); this.released = true; } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java index b065cb17ee..d490843fda 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java @@ -35,6 +35,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.Message; import org.springframework.integration.MessagingException; @@ -121,6 +122,13 @@ public class CachingClientConnectionFactoryTests { when(factory.isRunning()).thenReturn(true); TcpConnection mockConn1 = makeMockConnection("conn1"); TcpConnection mockConn2 = makeMockConnection("conn2"); + doAnswer(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + return null; + } + }).when(mockConn1).close(); when(factory.getConnection()).thenReturn(mockConn1) .thenReturn(mockConn2).thenReturn(mockConn1) .thenReturn(mockConn2); @@ -194,6 +202,8 @@ public class CachingClientConnectionFactoryTests { conn2.close(); verify(mockConn1).close(); verify(mockConn2).close(); + when(mockConn1.isOpen()).thenReturn(false); + when(mockConn2.isOpen()).thenReturn(false); when(factory.isRunning()).thenReturn(true); TcpConnection conn3 = cachingFactory.getConnection(); assertNotSame(TestUtils.getPropertyValue(conn1, "theConnection"), @@ -206,8 +216,11 @@ public class CachingClientConnectionFactoryTests { public void testEnlargePool() throws Exception { AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class); when(factory.isRunning()).thenReturn(true); - TcpConnection mockConn = makeMockConnection("conn"); - when(factory.getConnection()).thenReturn(mockConn); + TcpConnection mockConn1 = makeMockConnection("conn1"); + TcpConnection mockConn2 = makeMockConnection("conn2"); + TcpConnection mockConn3 = makeMockConnection("conn3"); + TcpConnection mockConn4 = makeMockConnection("conn4"); + when(factory.getConnection()).thenReturn(mockConn1, mockConn2, mockConn3, mockConn4); CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(factory, 2); cachingFactory.start(); TcpConnection conn1 = cachingFactory.getConnection(); @@ -232,10 +245,10 @@ public class CachingClientConnectionFactoryTests { public void testReducePool() throws Exception { AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class); when(factory.isRunning()).thenReturn(true); - TcpConnection mockConn1 = makeMockConnection("conn", true); - TcpConnection mockConn2 = makeMockConnection("conn", true); - TcpConnection mockConn3 = makeMockConnection("conn", true); - TcpConnection mockConn4 = makeMockConnection("conn", true); + TcpConnection mockConn1 = makeMockConnection("conn1", true); + TcpConnection mockConn2 = makeMockConnection("conn2", true); + TcpConnection mockConn3 = makeMockConnection("conn3", true); + TcpConnection mockConn4 = makeMockConnection("conn4", true); when(factory.getConnection()).thenReturn(mockConn1) .thenReturn(mockConn2).thenReturn(mockConn3) .thenReturn(mockConn4);