INT-3433 Delay Failed IOs

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

If a read fails due to insufficient threads, delay the read
for (default 100ms) - do not re-enable OP_READ until that
time has elapsed. Avoids spinning the CPU.

INT-3433 More Polishing

If the assembler couldn't execute a new assembler after assembling
the current message (when it detected there is more data), in the
finally block it would "continue" only if the socket was still
open.

The test case closes the socket after sending 4 messages so when
this condition occurred, the assembler failed to continue and
data was left in the buffer.

Remove the `isOpen()` check in the finally block and always continue
if there's not another assembler running and there's data available.

INT-3433 More Polishing

We can still get starvation if the selector is in a long
wait (in select()) when a read is delayed.

Whenever a read is delayed, wake the selector so its next
select will use the readDelay timeout.

INT-3433 Reference Docs

Also remove Thread.yield().

Revert redundant boolean return from `TcpNioConnection#checkForAssembler()`

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<PendingIO> delayedReads = new LinkedBlockingQueue<AbstractConnectionFactory.PendingIO>();
+
+	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<PendingIO> 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<SocketChannel, TcpNioConnection> 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;
+		}
+
+	}
+
 }

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()) {

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() + " " +

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) {

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);

src/reference/docbook/ip.xml

@@ -1001,6 +1001,9 @@
       the <classname>CallerRunsPolicy</classname> (<code>CALLER_RUNS</code> when
       using the <code>&lt;task/&gt;</code> namespace) and the queue capacity is small.
      </para>
+     <para>
+      The following does not apply if you are not using a fixed thread pool.
+     </para>
      <para>
       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 @@
      </para>
      <para>
       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.
      </para>
      <para>
-      Two classes are provided by the framework to avoid this problem. The
-      <classname>CompositeExecutor</classname> allows the configuration
+      The framework providers a
+      <classname>CompositeExecutor</classname>, which 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 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
-      <code>maxPoolSize</code> (or <code>queueCapacity</code>)
-      of the assembler executor should be slightly
-      larger than those on the IO executor.
+     </para>
+     <para>
+      In addition, the task executors should be configured to use a
+      <classname>AbortPolicy</classname> (ABORT when using <code>&lt;task&gt;</code>).
+      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.
+     </para>
+     <para>
+      Example configuration of the composite executor is shown below.
      </para>
      <programlisting language="java"><![CDATA[@Bean
 private CompositeExecutor compositeExecutor() {
     ThreadPoolTaskExecutor ioExec = new ThreadPoolTaskExecutor();
     ioExec.setCorePoolSize(4);
-    ioExec.setMaxPoolSize(8);
+    ioExec.setMaxPoolSize(10);
     ioExec.setQueueCapacity(0);
     ioExec.setThreadNamePrefix("io-");
-    ioExec.setRejectedExecutionHandler(new CallerRunsPolicy());
+    ioExec.setRejectedExecutionHandler(new AbortPolicy());
     ioExec.initialize();
     ThreadPoolTaskExecutor assemblerExec = new ThreadPoolTaskExecutor();
     assemblerExec.setCorePoolSize(4);
@@ -1062,6 +1069,14 @@ private CompositeExecutor compositeExecutor() {
     assemblerExec.initialize();
     return new CompositeExecutor(ioExec, assemblerExec);
 }]]></programlisting>
+<programlisting language="xml"><![CDATA[<bean id="myTaskExecutor" class="org.springframework.integration.util.CompositeExecutor">
+    <constructor-arg ref="io"/>
+    <constructor-arg ref="assembler"/>
+</bean>
+
+<task:executor id="io" pool-size="4-10" queue-capacity="0" rejection-policy="ABORT" />
+<task:executor id="assembler" pool-size="4-10" queue-capacity="0" rejection-policy="ABORT" />]]></programlisting>
+
 <programlisting language="xml"><![CDATA[<bean id="myTaskExecutor" class="org.springframework.integration.util.CompositeExecutor">
     <constructor-arg>
         <bean class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
@@ -1070,9 +1085,7 @@ private CompositeExecutor compositeExecutor() {
             <property name="maxPoolSize" value="8" />
             <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>