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

INT-3433 NIO Thread Starvation with Fixed Pool

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3433

INT-3433 added the test-case

Conflicts:
	spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
This commit is contained in:
JohnA2
2014-06-10 23:36:32 +04:00
committed by Artem Bilan
parent 973e677fc2
commit eafbe9b6e3
3 changed files with 183 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

205
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

@@ -26,11 +26,11 @@ import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -189,67 +189,97 @@ public class TcpNioConnection extends TcpConnectionSupport {
if (logger.isTraceEnabled()) {
logger.trace(this.getConnectionId() + " Nio message assembler running...");
}
try {
if (this.getListener() == null && !this.isSingleUse()) {
logger.debug("TcpListener exiting - no listener and not single use");
return;
}
boolean moreDataAvailable = true;
while(moreDataAvailable) {
try {
if (dataAvailable()) {
Message<?> message = convert();
if (this.getListener() == null && !this.isSingleUse()) {
logger.debug("TcpListener exiting - no listener and not single use");
return;
}
try {
if (dataAvailable()) {
// there is more data in the pipe; run another assembler
// to assemble the next message, while we send ours
this.executionControl.incrementAndGet();
this.taskExecutor.execute(this);
Message<?> message = convert();
if (dataAvailable()) {
// there is more data in the pipe; run another assembler
// to assemble the next message, while we send ours
this.executionControl.incrementAndGet();
try {
this.taskExecutor.execute2(this);
}
catch (RejectedExecutionException e) {
this.executionControl.decrementAndGet();
if (logger.isInfoEnabled()) {
logger.info("Insufficient threads in the assembler fixed thread pool; consider " +
"increasing this task executor pool size");
}
}
}
this.executionControl.decrementAndGet();
if (message != null) {
sendToChannel(message);
}
}
this.executionControl.decrementAndGet();
if (message != null) {
sendToChannel(message);
else {
this.executionControl.decrementAndGet();
}
}
else {
this.executionControl.decrementAndGet();
}
}
catch (Exception e) {
if (logger.isTraceEnabled()) {
logger.error("Read exception " +
this.getConnectionId(), e);
}
else if (!this.isNoReadErrorOnClose()) {
logger.error("Read exception " +
this.getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + e.getCause() + ":" + e.getMessage());
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Read exception " +
catch (Exception e) {
if (logger.isTraceEnabled()) {
logger.error("Read exception " +
this.getConnectionId(), e);
}
else if (!this.isNoReadErrorOnClose()) {
logger.error("Read exception " +
this.getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + e.getCause() + ":" + e.getMessage());
}
}
this.closeConnection(true);
this.sendExceptionToListener(e);
return;
}
}
finally {
if (logger.isTraceEnabled()) {
logger.trace(this.getConnectionId() + " Nio message assembler exiting...");
}
// Final check in case new data came in and the
// timing was such that we were the last assembler and
// a new one wasn't run
try {
if (this.isOpen() && dataAvailable()) {
checkForAssembler();
else {
if (logger.isDebugEnabled()) {
logger.debug("Read exception " +
this.getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + e.getCause() + ":" + e.getMessage());
}
}
this.closeConnection(true);
this.sendExceptionToListener(e);
return;
}
}
catch (IOException e) {
logger.error("Exception when checking for assembler", e);
finally {
moreDataAvailable = false;
// Final check in case new data came in and the
// timing was such that we were the last assembler and
// a new one wasn't run
try {
if (this.isOpen() && dataAvailable()) {
synchronized(this.executionControl) {
if (this.executionControl.incrementAndGet() <= 1) {
// only continue if we don't already have another assembler running
this.executionControl.set(1);
moreDataAvailable = true;
} else {
this.executionControl.decrementAndGet();
}
}
}
if (moreDataAvailable) {
Thread.yield();
if (logger.isTraceEnabled()) {
logger.trace(this.getConnectionId() + " Nio message assembler continuing...");
}
}
else {
if (logger.isTraceEnabled()) {
logger.trace(this.getConnectionId() + " Nio message assembler exiting...");
}
}
}
catch (IOException e) {
logger.error("Exception when checking for assembler", e);
}
}
}
}
@@ -333,47 +363,25 @@ public class TcpNioConnection extends TcpConnectionSupport {
this.taskExecutor = new CompositeExecutor(executor, executor);
}
// If there is no assembler running, start one
checkForAssembler();
if (logger.isTraceEnabled()) {
logger.trace("Before read:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
}
int len = this.socketChannel.read(this.rawBuffer);
if (len < 0) {
this.writingToPipe = false;
this.closeConnection(true);
}
if (logger.isTraceEnabled()) {
logger.trace("After read:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
}
this.rawBuffer.flip();
if (logger.isTraceEnabled()) {
logger.trace("After flip:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
}
if (logger.isDebugEnabled()) {
logger.debug("Read " + rawBuffer.limit() + " into raw buffer");
}
final CountDownLatch latch = new CountDownLatch(1);
/*
* If there are insufficient threads, either to run the
* write to the pipe, or to assemble the data, we need
* avoid a deadlock (block on the write to the pipe).
* Hence the count down latch.
*/
this.taskExecutor.execute2(new Runnable() {
public void run() {
try {
TcpNioConnection.this.sendToPipe(rawBuffer);
latch.countDown();
}
catch (Exception e) {
logger.error(getConnectionId() + " Failed to write to pipe", e);
}
if (checkForAssembler()) {
if (logger.isTraceEnabled()) {
logger.trace("Before read:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit()); }
int len = this.socketChannel.read(this.rawBuffer);
if (len < 0) {
this.writingToPipe = false;
this.closeConnection(true);
}
});
if (!latch.await(this.pipeTimeout , TimeUnit.MILLISECONDS)) {
this.close();
throw new MessagingException("Timed out writing to ChannelInputStream, probably due to insufficient threads in " +
"a fixed thread pool; consider increasing this task executor pool size");
if (logger.isTraceEnabled()) {
logger.trace("After read:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
}
this.rawBuffer.flip();
if (logger.isTraceEnabled()) {
logger.trace("After flip:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
}
if (logger.isDebugEnabled()) {
logger.debug("Read " + rawBuffer.limit() + " into raw buffer");
}
this.sendToPipe(rawBuffer);
}
}
catch (Exception e) {
@@ -394,7 +402,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
rawBuffer.clear();
}
private void checkForAssembler() {
private boolean checkForAssembler() {
synchronized(this.executionControl) {
if (this.executionControl.incrementAndGet() <= 1) {
// only execute run() if we don't already have one running
@@ -402,11 +410,22 @@ public class TcpNioConnection extends TcpConnectionSupport {
if (logger.isDebugEnabled()) {
logger.debug(this.getConnectionId() + " Running an assembler");
}
this.taskExecutor.execute2(this);
try {
this.taskExecutor.execute2(this);
}
catch (RejectedExecutionException e) {
this.executionControl.decrementAndGet();
if (logger.isInfoEnabled()) {
logger.info("Insufficient threads in the assembler fixed thread pool; consider increasing " +
"this task executor pool size");
}
return false;
}
} else {
this.executionControl.decrementAndGet();
}
}
return true;
}
/**

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -45,6 +45,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor.AbortPolicy;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
@@ -329,9 +330,7 @@ public class TcpNioConnectionTests {
fail("Expected exception, got " + o);
}
catch (ExecutionException e) {
assertEquals("Timed out writing to ChannelInputStream, probably due to insufficient threads in " +
"a fixed thread pool; consider increasing this task executor pool size", e.getCause()
.getMessage());
assertEquals("Timed out waiting for buffer space", e.getCause().getMessage());
}
}
@@ -601,18 +600,74 @@ public class TcpNioConnectionTests {
factory.stop();
}
@Test
public void testAllMessagesDelivered() throws Exception {
final int numberOfSockets = 25;
final int port = SocketUtils.findAvailableServerSocket();
TcpNioServerConnectionFactory factory = new TcpNioServerConnectionFactory(port);
factory.setApplicationEventPublisher(mock(ApplicationEventPublisher.class));
CompositeExecutor compositeExec = compositeExecutor();
factory.setTaskExecutor(compositeExec);
final CountDownLatch latch = new CountDownLatch(numberOfSockets);
factory.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
if (!(message instanceof ErrorMessage)) {
latch.countDown();
}
return false;
}
});
factory.start();
Socket[] sockets = new Socket[numberOfSockets];
for (int i = 0; i < numberOfSockets; i++) {
Socket socket = null;
int n = 0;
while (n++ < 100) {
try {
socket = SocketFactory.getDefault().createSocket("localhost", port);
break;
}
catch (ConnectException e) {}
Thread.sleep(100);
}
assertTrue("Could not open socket to localhost:" + port, n < 100);
sockets[i] = socket;
}
for (int i = 0; i < numberOfSockets; i++) {
sockets[i].getOutputStream().write("foo1 and...".getBytes());
sockets[i].getOutputStream().flush();
}
Thread.sleep(100);
for (int i = 0; i < numberOfSockets; i++) {
sockets[i].getOutputStream().write(("...foo2\r\n").getBytes());
sockets[i].close();
}
assertTrue(latch.await(10, TimeUnit.SECONDS));
factory.stop();
}
private CompositeExecutor compositeExecutor() {
ThreadPoolTaskExecutor ioExec = new ThreadPoolTaskExecutor();
ioExec.setCorePoolSize(2);
ioExec.setQueueCapacity(10);
ioExec.setMaxPoolSize(4);
ioExec.setQueueCapacity(0);
ioExec.setThreadNamePrefix("io-");
ioExec.setRejectedExecutionHandler(new CallerBlocksPolicy(10000));
ioExec.setRejectedExecutionHandler(new CallerBlocksPolicy(5000));
ioExec.initialize();
ThreadPoolTaskExecutor assemblerExec = new ThreadPoolTaskExecutor();
assemblerExec.setCorePoolSize(2);
assemblerExec.setQueueCapacity(10);
assemblerExec.setMaxPoolSize(5);
assemblerExec.setQueueCapacity(0);
assemblerExec.setThreadNamePrefix("assembler-");
assemblerExec.setRejectedExecutionHandler(new CallerBlocksPolicy(10000));
assemblerExec.setRejectedExecutionHandler(new AbortPolicy());
assemblerExec.initialize();
return new CompositeExecutor(ioExec, assemblerExec);
}

18
src/reference/docbook/ip.xml
View File

@@ -1027,7 +1027,7 @@
<classname>CompositeExecutor</classname> allows the configuration
of two distinct executors; one for performing IO operations, and
one for message assembly. The <classname>CallerBlocksPolicy</classname>
(which should be configured for both task executors) will suspend
(which should be configured for the first task executors) will suspend
the IO operation until an assembler thread is available (or a timeout
occurs). In this environment, an IO thread can never
become an assembler thread, and the deadlock cannot occur.
@@ -1041,16 +1041,16 @@ private CompositeExecutor compositeExecutor() {
ThreadPoolTaskExecutor ioExec = new ThreadPoolTaskExecutor();
ioExec.setCorePoolSize(4);
ioExec.setMaxPoolSize(8);
ioExec.setQueueCapacity(10);
ioExec.setQueueCapacity(0);
ioExec.setThreadNamePrefix("io-");
ioExec.setRejectedExecutionHandler(new CallerRunsPolicy());
ioExec.initialize();
ThreadPoolTaskExecutor assemblerExec = new ThreadPoolTaskExecutor();
assemblerExec.setCorePoolSize(2);
assemblerExec.setCorePoolSize(4);
assemblerExec.setMaxPoolSize(10);
assemblerExec.setQueueCapacity(12);
assemblerExec.setQueueCapacity(0);
assemblerExec.setThreadNamePrefix("assembler-");
assemblerExec.setRejectedExecutionHandler(new CallerBlocksPolicy(10000));
assemblerExec.setRejectedExecutionHandler(new AbortPolicy());
assemblerExec.initialize();
return new CompositeExecutor(ioExec, assemblerExec);
}]]></programlisting>
@@ -1060,7 +1060,7 @@ private CompositeExecutor compositeExecutor() {
<property name="threadNamePrefix" value="io-" />
<property name="corePoolSize" value="4" />
<property name="maxPoolSize" value="8" />
<property name="queueCapacity" value="10" />
<property name="queueCapacity" value="0" />
<property name="rejectedExecutionHandler">
<bean class="org.springframework.integration.util.CallerBlocksPolicy">
<constructor-arg value="10000" />
@@ -1073,11 +1073,9 @@ private CompositeExecutor compositeExecutor() {
<property name="threadNamePrefix" value="assembler-" />
<property name="corePoolSize" value="4" />
<property name="maxPoolSize" value="10" />
<property name="queueCapacity" value="10" />
<property name="queueCapacity" value="0" />
<property name="rejectedExecutionHandler">
<bean class="org.springframework.integration.util.CallerBlocksPolicy">
<constructor-arg value="10000" />
</bean>
<bean class="java.util.concurrent.ThreadPoolExecutor.AbortPolicy" />
</property>
</bean>
</constructor-arg>