diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index 3ac7ffa90c..ac505433bc 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -34,9 +34,12 @@ import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Set; +import java.util.concurrent.BlockingQueue; 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 org.springframework.context.ApplicationEventPublisher; @@ -61,6 +64,10 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport protected static final int DEFAULT_REPLY_TIMEOUT = 10000; + private static final int DEFAULT_NIO_HARVEST_INTERVAL = 2000; + + private static final int DEFAULT_READ_DELAY = 100; + private volatile String host; private volatile int port; @@ -117,7 +124,9 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport private volatile ApplicationEventPublisher applicationEventPublisher; - private static final int DEFAULT_NIO_HARVEST_INTERVAL = 2000; + private final BlockingQueue delayedReads = new LinkedBlockingQueue(); + + private volatile long readDelay = DEFAULT_READ_DELAY; public AbstractConnectionFactory(int port) { this.port = port; @@ -419,6 +428,25 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport this.nioHarvestInterval = nioHarvestInterval; } + protected BlockingQueue getDelayedReads() { + return delayedReads; + } + + protected long getReadDelay() { + return readDelay; + } + + // TODO: Expose on the namespace in 4.1 ? + /** + * The delay (in milliseconds) before retrying a read after the previous attempt + * failed due to insufficient threads. Default 100. + * @param readDelay the readDelay to set. + */ + public void setReadDelay(long readDelay) { + Assert.isTrue(readDelay > 0, "'readDelay' must be positive"); + this.readDelay = readDelay; + } + @Override protected void onInit() throws Exception { super.onInit(); @@ -533,7 +561,8 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport */ protected void processNioSelections(int selectionCount, final Selector selector, ServerSocketChannel server, Map connections) throws IOException { - long now = System.currentTimeMillis(); + final long now = System.currentTimeMillis(); + rescheduleDelayedReads(selector, now); if (this.soTimeout > 0 || now >= this.nextCheckForClosedNioConnections || selectionCount == 0) { @@ -596,31 +625,43 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport final TcpNioConnection connection; connection = (TcpNioConnection) key.attachment(); connection.setLastRead(System.currentTimeMillis()); - this.taskExecutor.execute(new Runnable() { - @Override - public void run() { - try { - connection.readPacket(); - } - catch (Exception e) { - if (connection.isOpen()) { - logger.error("Exception on read " + - connection.getConnectionId() + " " + - e.getMessage()); - connection.close(); + try { + this.taskExecutor.execute(new Runnable() { + @Override + public void run() { + boolean delayed = false; + try { + connection.readPacket(); } - else { - logger.debug("Connection closed"); + catch (RejectedExecutionException e) { + delayRead(selector, now, key); + delayed = true; } - } - if (key.channel().isOpen()) { - key.interestOps(SelectionKey.OP_READ); - selector.wakeup(); - } - else { - connection.sendExceptionToListener(new EOFException("Connection is closed")); - } - }}); + catch (Exception e) { + if (connection.isOpen()) { + logger.error("Exception on read " + + connection.getConnectionId() + " " + + e.getMessage()); + connection.close(); + } + else { + logger.debug("Connection closed"); + } + } + if (!delayed) { + if (key.channel().isOpen()) { + key.interestOps(SelectionKey.OP_READ); + selector.wakeup(); + } + else { + connection.sendExceptionToListener(new EOFException("Connection is closed")); + } + } + }}); + } + catch (RejectedExecutionException e) { + delayRead(selector, now, key); + } } else if (key.isAcceptable()) { try { @@ -632,17 +673,73 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport else { logger.error("Unexpected key: " + key); } - } catch (CancelledKeyException e) { + } + catch (CancelledKeyException e) { if (logger.isDebugEnabled()) { logger.debug("Selection key " + key + " cancelled"); } - } catch (Exception e) { + } + catch (Exception e) { logger.error("Exception on selection key " + key, e); } } } } + protected void delayRead(Selector selector, long now, final SelectionKey key) { + TcpNioConnection connection = (TcpNioConnection) key.attachment(); + if (!this.delayedReads.add(new PendingIO(now, key))) { // should never happen - unbounded queue + logger.error("Failed to delay read; closing " + connection.getConnectionId()); + connection.close(); + } + else { + if (logger.isDebugEnabled()) { + logger.debug("No threads available, delaying read for " + connection.getConnectionId()); + } + // wake the selector in case it is currently blocked, and waiting for longer than readDelay + selector.wakeup(); + } + } + + /** + * If any reads were delayed due to insufficient threads, reschedule them if + * the readDelay has passed. + * @param selector the selector to wake if necessary. + * @param now the current time. + */ + private void rescheduleDelayedReads(Selector selector, long now) { + boolean wakeSelector = false; + try { + while (this.delayedReads.size() > 0) { + if (this.delayedReads.peek().failedAt + this.readDelay < now) { + PendingIO pendingRead = this.delayedReads.take(); + if (pendingRead.key.channel().isOpen()) { + pendingRead.key.interestOps(SelectionKey.OP_READ); + wakeSelector = true; + if (logger.isDebugEnabled()) { + logger.debug("Rescheduling delayed read for " + ((TcpNioConnection) pendingRead.key.attachment()).getConnectionId()); + } + } + else { + ((TcpNioConnection) pendingRead.key.attachment()).sendExceptionToListener(new EOFException("Connection is closed")); + } + } + else { + // remaining delayed reads have not expired yet. + break; + } + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + finally { + if (wakeSelector) { + selector.wakeup(); + } + } + } + /** * @param selector The selector. * @param server The server socket channel. @@ -781,4 +878,18 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport return closed; } } + + private class PendingIO { + + private final long failedAt; + + private final SelectionKey key; + + private PendingIO(long failedAt, SelectionKey key) { + this.failedAt = failedAt; + this.key = key; + } + + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java index 73ad684f42..43b76e4a39 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java @@ -149,7 +149,11 @@ public class TcpNioClientConnectionFactory extends int soTimeout = this.getSoTimeout(); int selectionCount = 0; try { - selectionCount = selector.select(soTimeout < 0 ? 0 : soTimeout); + long timeout = soTimeout < 0 ? 0 : soTimeout; + if (getDelayedReads().size() > 0 && (timeout == 0 || getReadDelay() < timeout)) { + timeout = getReadDelay(); + } + selectionCount = selector.select(timeout); } catch (CancelledKeyException cke) { if (logger.isDebugEnabled()) { 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 8b4700972d..a42a03ea29 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 @@ -45,6 +45,7 @@ import org.springframework.util.Assert; * A TcpConnection that uses and underlying {@link SocketChannel}. * * @author Gary Russell + * @author John Anderson * @since 2.0 * */ @@ -208,8 +209,8 @@ public class TcpNioConnection extends TcpConnectionSupport { 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"); + logger.info(getConnectionId() + " Insufficient threads in the assembler fixed thread pool; consider " + + "increasing this task executor pool size; data avail: " + this.channelInputStream.available()); } } } @@ -252,27 +253,27 @@ public class TcpNioConnection extends TcpConnectionSupport { // timing was such that we were the last assembler and // a new one wasn't run try { - if (this.isOpen() && dataAvailable()) { + if (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 { + + } + 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..."); + logger.trace(this.getConnectionId() + " Nio message assembler exiting... avail: " + this.channelInputStream.available()); } } } @@ -362,27 +363,30 @@ public class TcpNioConnection extends TcpConnectionSupport { this.taskExecutor = new CompositeExecutor(executor, executor); } // If there is no assembler running, start one - 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 (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); + 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"); + } + this.sendToPipe(rawBuffer); + } + catch (RejectedExecutionException e) { + throw e; } catch (Exception e) { this.publishConnectionExceptionEvent(e); @@ -402,7 +406,7 @@ public class TcpNioConnection extends TcpConnectionSupport { rawBuffer.clear(); } - private boolean checkForAssembler() { + private void checkForAssembler() { synchronized(this.executionControl) { if (this.executionControl.incrementAndGet() <= 1) { // only execute run() if we don't already have one running @@ -419,13 +423,13 @@ public class TcpNioConnection extends TcpConnectionSupport { logger.info("Insufficient threads in the assembler fixed thread pool; consider increasing " + "this task executor pool size"); } - return false; + throw e; } - } else { + } + else { this.executionControl.decrementAndGet(); } } - return true; } /** @@ -444,6 +448,9 @@ public class TcpNioConnection extends TcpConnectionSupport { } this.closeConnection(true); } + catch (RejectedExecutionException e) { + throw e; + } catch (Exception e) { logger.error("Exception on Read " + this.getConnectionId() + " " + diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java index b9033c79ae..4ead2410ef 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java @@ -128,7 +128,14 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto int soTimeout = this.getSoTimeout(); int selectionCount = 0; try { - selectionCount = selector.select(soTimeout < 0 ? 0 : soTimeout); + long timeout = soTimeout < 0 ? 0 : soTimeout; + if (getDelayedReads().size() > 0 && (timeout == 0 || getReadDelay() < timeout)) { + timeout = getReadDelay(); + } + if (logger.isTraceEnabled()) { + logger.trace("Delayed reads:" + getDelayedReads().size() + " timeout " + timeout); + } + selectionCount = selector.select(timeout); this.processNioSelections(selectionCount, selector, server, this.channelMap); } catch (CancelledKeyException cke) { 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 390eb73a96..e704eb2dc8 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 @@ -73,7 +73,6 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.converter.MapMessageConverter; import org.springframework.integration.test.util.SocketUtils; import org.springframework.integration.test.util.TestUtils; -import org.springframework.integration.util.CallerBlocksPolicy; import org.springframework.integration.util.CompositeExecutor; import org.springframework.messaging.Message; import org.springframework.messaging.support.ErrorMessage; @@ -85,6 +84,7 @@ import org.springframework.util.ReflectionUtils.FieldFilter; /** * @author Gary Russell + * @author John Anderson * @since 2.0 * */ @@ -608,7 +608,7 @@ public class TcpNioConnectionTests { @Test public void testAllMessagesDelivered() throws Exception { - final int numberOfSockets = 25; + final int numberOfSockets = 100; final int port = SocketUtils.findAvailableServerSocket(); TcpNioServerConnectionFactory factory = new TcpNioServerConnectionFactory(port); factory.setApplicationEventPublisher(mock(ApplicationEventPublisher.class)); @@ -616,7 +616,7 @@ public class TcpNioConnectionTests { CompositeExecutor compositeExec = compositeExecutor(); factory.setTaskExecutor(compositeExec); - final CountDownLatch latch = new CountDownLatch(numberOfSockets); + final CountDownLatch latch = new CountDownLatch(numberOfSockets * 4); factory.registerListener(new TcpListener() { @Override @@ -629,7 +629,7 @@ public class TcpNioConnectionTests { }); factory.start(); - + Socket[] sockets = new Socket[numberOfSockets]; for (int i = 0; i < numberOfSockets; i++) { Socket socket = null; @@ -651,11 +651,28 @@ public class TcpNioConnectionTests { } Thread.sleep(100); for (int i = 0; i < numberOfSockets; i++) { - sockets[i].getOutputStream().write(("...foo2\r\n").getBytes()); + sockets[i].getOutputStream().write(("...foo2\r\nbar1 and...").getBytes()); + sockets[i].getOutputStream().flush(); + } + for (int i = 0; i < numberOfSockets; i++) { + sockets[i].getOutputStream().write(("...bar2\r\n").getBytes()); + sockets[i].getOutputStream().flush(); + } + for (int i = 0; i < numberOfSockets; i++) { + sockets[i].getOutputStream().write("foo3 and...".getBytes()); + sockets[i].getOutputStream().flush(); + } + Thread.sleep(100); + for (int i = 0; i < numberOfSockets; i++) { + sockets[i].getOutputStream().write(("...foo4\r\nbar3 and...").getBytes()); + sockets[i].getOutputStream().flush(); + } + for (int i = 0; i < numberOfSockets; i++) { + sockets[i].getOutputStream().write(("...bar4\r\n").getBytes()); sockets[i].close(); } - - assertTrue(latch.await(10, TimeUnit.SECONDS)); + + assertTrue("latch is still " + latch.getCount(), latch.await(60, TimeUnit.SECONDS)); factory.stop(); } @@ -666,7 +683,7 @@ public class TcpNioConnectionTests { ioExec.setMaxPoolSize(4); ioExec.setQueueCapacity(0); ioExec.setThreadNamePrefix("io-"); - ioExec.setRejectedExecutionHandler(new CallerBlocksPolicy(5000)); + ioExec.setRejectedExecutionHandler(new AbortPolicy()); ioExec.initialize(); ThreadPoolTaskExecutor assemblerExec = new ThreadPoolTaskExecutor(); assemblerExec.setCorePoolSize(2); diff --git a/src/reference/docbook/ip.xml b/src/reference/docbook/ip.xml index 273681e79f..cdf7a55808 100644 --- a/src/reference/docbook/ip.xml +++ b/src/reference/docbook/ip.xml @@ -1001,6 +1001,9 @@ the CallerRunsPolicy (CALLER_RUNS when using the <task/> namespace) and the queue capacity is small. + + The following does not apply if you are not using a fixed thread pool. + With NIO connections there are 3 distinct task types; the IO Selector processing is performed on one dedicated thread - detecting events, accepting new connections, @@ -1028,30 +1031,34 @@ We must avoid the selector (or reader) threads performing the - assembly task to avoid this deadlock. + assembly task to avoid this deadlock. It is desirable to use seperate + pools for the IO and assembly operations. - Two classes are provided by the framework to avoid this problem. The - CompositeExecutor allows the configuration + The framework providers a + CompositeExecutor, which allows the configuration of two distinct executors; one for performing IO operations, and - one for message assembly. The CallerBlocksPolicy - (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 + one for message assembly. In this environment, an IO thread can never become an assembler thread, and the deadlock cannot occur. - Example configuration of the composite executor is shown below. The - maxPoolSize (or queueCapacity) - of the assembler executor should be slightly - larger than those on the IO executor. + + + In addition, the task executors should be configured to use a + AbortPolicy (ABORT when using <task>). + When an IO cannot be completed, it is deferred for a short time and + retried continually until it can be completed and an assembler + allocated. + + + Example configuration of the composite executor is shown below. + + + + + + +]]> + @@ -1070,9 +1085,7 @@ private CompositeExecutor compositeExecutor() { - - - +