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

INT-3112 Fix OOM in SimplePool

Browse Source 
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.
This commit is contained in:
Gary Russell
2013-08-20 09:34:55 -04:00
parent bb71fd8a37
commit e73a5dc69f
5 changed files with 124 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java
View File

@@ -54,6 +54,8 @@ public class SimplePool<T> implements Pool<T> {
private final Set<T> allocated = Collections.synchronizedSet(new HashSet<T>());
private final Set<T> inUse = Collections.synchronizedSet(new HashSet<T>());
private final PoolItemCallback<T> callback;
/**
@@ -127,7 +129,7 @@ public class SimplePool<T> implements Pool<T> {
}
public int getActiveCount() {
return this.getAllocatedCount() - this.getIdleCount();
return this.inUse.size();
}
public int getAllocatedCount() {
@@ -192,32 +194,42 @@ public class SimplePool<T> implements Pool<T> {
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();
}
}
@@ -230,6 +242,7 @@ public class SimplePool<T> implements Pool<T> {
private void doRemoveItem(T item) {
this.allocated.remove(item);
this.inUse.remove(item);
this.callback.removedFromPool(item);
}

22
spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java
View File

@@ -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<String> strings = new HashSet<String>();
final AtomicBoolean stale = new AtomicBoolean();
SimplePool<String> 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<String> stringPool(int size, final Set<String> strings,
final AtomicBoolean stale) {
SimplePool<String> pool = new SimplePool<String>(size, new SimplePool.PoolItemCallback<String>() {

48
spring-integration-file/src/test/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandlerTests.java
View File

@@ -16,29 +16,39 @@
package org.springframework.integration.file.remote.handler;
import static org.junit.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
@@ -150,4 +160,40 @@ public class FileTransferringMessageHandlerTests {
verify(session, times(0)).rename(Mockito.anyString(), Mockito.anyString());
}
@SuppressWarnings("unchecked")
@Test
public <F> void testServerException() throws Exception{
SessionFactory<F> sf = mock(SessionFactory.class);
CachingSessionFactory<F> csf = new CachingSessionFactory<F>(sf, 2);
FileTransferringMessageHandler<F> handler = new FileTransferringMessageHandler<F>(csf);
Session<F> session1 = newSession();
Session<F> session2 = newSession();
Session<F> 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<String>("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 <F> Session<F> newSession() throws IOException {
@SuppressWarnings("unchecked")
Session<F> session = mock(Session.class);
doThrow(new IOException("test")).when(session).write(any(InputStream.class), anyString());
when(session.isOpen()).thenReturn(false);
return session;
}
}

23
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java
View File

@@ -103,23 +103,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;
}

25
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java
View File

@@ -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;
@@ -124,6 +125,13 @@ public class CachingClientConnectionFactoryTests {
when(factory.isRunning()).thenReturn(true);
TcpConnectionSupport mockConn1 = makeMockConnection("conn1");
TcpConnectionSupport mockConn2 = makeMockConnection("conn2");
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(mockConn1).close();
when(factory.getConnection()).thenReturn(mockConn1)
.thenReturn(mockConn2).thenReturn(mockConn1)
.thenReturn(mockConn2);
@@ -197,6 +205,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"),
@@ -209,8 +219,11 @@ public class CachingClientConnectionFactoryTests {
public void testEnlargePool() throws Exception {
AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class);
when(factory.isRunning()).thenReturn(true);
TcpConnectionSupport mockConn = makeMockConnection("conn");
when(factory.getConnection()).thenReturn(mockConn);
TcpConnectionSupport mockConn1 = makeMockConnection("conn1");
TcpConnectionSupport mockConn2 = makeMockConnection("conn2");
TcpConnectionSupport mockConn3 = makeMockConnection("conn3");
TcpConnectionSupport mockConn4 = makeMockConnection("conn4");
when(factory.getConnection()).thenReturn(mockConn1, mockConn2, mockConn3, mockConn4);
CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(factory, 2);
cachingFactory.start();
TcpConnection conn1 = cachingFactory.getConnection();
@@ -235,10 +248,10 @@ public class CachingClientConnectionFactoryTests {
public void testReducePool() throws Exception {
AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class);
when(factory.isRunning()).thenReturn(true);
TcpConnectionSupport mockConn1 = makeMockConnection("conn", true);
TcpConnectionSupport mockConn2 = makeMockConnection("conn", true);
TcpConnectionSupport mockConn3 = makeMockConnection("conn", true);
TcpConnectionSupport mockConn4 = makeMockConnection("conn", true);
TcpConnectionSupport mockConn1 = makeMockConnection("conn1", true);
TcpConnectionSupport mockConn2 = makeMockConnection("conn2", true);
TcpConnectionSupport mockConn3 = makeMockConnection("conn3", true);
TcpConnectionSupport mockConn4 = makeMockConnection("conn4", true);
when(factory.getConnection()).thenReturn(mockConn1)
.thenReturn(mockConn2).thenReturn(mockConn3)
.thenReturn(mockConn4);