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 1efd1f2f9f..2f3634a453 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; @@ -113,7 +120,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; @@ -409,8 +418,26 @@ 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 ? /** - * Closes the factory. + * 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; + } + + /** Closes the factory. * @deprecated As of 3.0; use {@link #stop()}. */ @Deprecated @@ -521,7 +548,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) { @@ -584,31 +612,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 { @@ -620,21 +660,78 @@ 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(); + } + } + /** - * @param selector - * @param now - * @throws IOException + * 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. + * @param now The current time. + * @throws IOException Any IOException. */ protected void doAccept(final Selector selector, ServerSocketChannel server, long now) throws IOException { throw new UnsupportedOperationException("Nio server factory must override this method"); @@ -768,4 +865,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 877d4998e2..10204ac813 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 @@ -148,7 +148,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 fbc396d56b..dc59a647d9 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 @@ -46,6 +46,7 @@ import org.springframework.util.Assert; * A TcpConnection that uses and underlying {@link SocketChannel}. * * @author Gary Russell + * @author John Anderson * @since 2.0 * */ @@ -209,8 +210,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()); } } } @@ -253,27 +254,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()); } } } @@ -363,26 +364,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 +407,7 @@ if (checkForAssembler()) { 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 +424,13 @@ if (checkForAssembler()) { 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 +449,9 @@ if (checkForAssembler()) { } 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 53f2921e33..782df53890 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 @@ -130,7 +130,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 36a91fb73c..d4fc53692e 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 @@ -69,7 +69,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.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.util.ReflectionUtils; @@ -79,6 +78,7 @@ import org.springframework.util.ReflectionUtils.FieldFilter; /** * @author Gary Russell + * @author John Anderson * @since 2.0 * */ @@ -602,7 +602,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)); @@ -610,7 +610,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 @@ -623,7 +623,7 @@ public class TcpNioConnectionTests { }); factory.start(); - + Socket[] sockets = new Socket[numberOfSockets]; for (int i = 0; i < numberOfSockets; i++) { Socket socket = null; @@ -645,11 +645,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(); } @@ -660,7 +677,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 dd1c944d74..d46e4c5e17 100644 --- a/src/reference/docbook/ip.xml +++ b/src/reference/docbook/ip.xml @@ -993,6 +993,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, @@ -1020,30 +1023,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. + + + + + + +]]> + @@ -1062,9 +1077,7 @@ private CompositeExecutor compositeExecutor() { - - - +