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
This commit is contained in:
JohnA2
2014-06-10 23:36:32 +04:00
committed by Artem Bilan
parent 119787f020
commit 69e20fc7db
3 changed files with 184 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

208
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;
@@ -39,7 +39,6 @@ import org.springframework.core.serializer.Serializer;
import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
import org.springframework.integration.util.CompositeExecutor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
/**
@@ -189,67 +188,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,48 +362,26 @@ 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() {
@Override
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());
}
});
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");
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");
}
this.sendToPipe(rawBuffer);
}
}
catch (Exception e) {
@@ -395,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
@@ -403,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.
@@ -51,6 +51,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;
@@ -335,9 +336,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());
}
}
@@ -607,18 +606,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);
}