diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
index 8c40fb0aff..fbc396d56b 100644
--- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
+++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
@@ -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;
}
/**
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 fc444866dc..36a91fb73c 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-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);
}
diff --git a/src/reference/docbook/ip.xml b/src/reference/docbook/ip.xml
index 7302ba6323..dd1c944d74 100644
--- a/src/reference/docbook/ip.xml
+++ b/src/reference/docbook/ip.xml
@@ -1027,7 +1027,7 @@
CompositeExecutor allows the configuration
of two distinct executors; one for performing IO operations, and
one for message assembly. The CallerBlocksPolicy
- (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);
}]]>
@@ -1060,7 +1060,7 @@ private CompositeExecutor compositeExecutor() {
-
+
@@ -1073,11 +1073,9 @@ private CompositeExecutor compositeExecutor() {
-
+
-
-
-
+