INT-3410 TCP NIO Deadlock with Bound TE

JIRA: https://jira.spring.io/browse/INT-3410

When running a fixed thread pool with a bound queue,
and CALLER_RUNS execution rejection policy, it
was possible to deadlock the IO selector thread.

Add a `CompositeExecutor` to use different threads for
IO to those used for message assembly.

Add a `CallerBlocksPolicy` to block the invoking thread
(for a specified time) if the pool is exhausted.

Add documentation.

Polishing

spring-integration-core/src/main/java/org/springframework/integration/util/CallerBlocksPolicy.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.util;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * A {@link RejectedExecutionHandler} that blocks the caller until
+ * the executor has room in its queue, or a timeout occurs (in which
+ * case a {@link RejectedExecutionException} is thrown.
+ *
+ * @author Gary Russell
+ * @since 3.0.3
+ *
+ */
+public class CallerBlocksPolicy implements RejectedExecutionHandler {
+
+	private static final Log logger = LogFactory.getLog(CallerBlocksPolicy.class);
+
+	private final long maxWait;
+
+	/**
+	 * @param maxWait The maximum time to wait for a queue slot to be
+	 * available, in milliseconds.
+	 */
+	public CallerBlocksPolicy(long maxWait) {
+		this.maxWait = maxWait;
+	}
+
+	@Override
+	public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
+		if (!executor.isShutdown()) {
+			try {
+				BlockingQueue<Runnable> queue = executor.getQueue();
+				if (logger.isDebugEnabled()) {
+					logger.debug("Attempting to queue task execution for " + this.maxWait + " milliseconds");
+				}
+				if (!queue.offer(r, this.maxWait, TimeUnit.MILLISECONDS)) {
+					throw new RejectedExecutionException("Max wait time expired to queue task");
+				}
+				if (logger.isDebugEnabled()) {
+					logger.debug("Task execution queued");
+				}
+			}
+			catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+				throw new RejectedExecutionException("Interrupted", e);
+			}
+		}
+		else {
+			throw new RejectedExecutionException("Executor has been shut down");
+		}
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/util/CompositeExecutor.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.util;
+
+import java.util.concurrent.Executor;
+
+import org.springframework.util.Assert;
+
+/**
+ * An {@link Executor} that encapsulates two underlying executors. Used in cases
+ * where two distinct operation types are being used where sharing threads could
+ * adversely affect the operation of the system. For example, NIO event processing
+ * threads being used for other blocking operations (such as assembling messages).
+ * If a {@code CallerRunsPolicy} rejected execution policy is used, the NIO event thread
+ * might deadlock, and stop processing events.
+ * <p>
+ * In order to work in such an environment, the secondary executor must <b>not</b>
+ * have a {@code CallerRunsPolicy} rejected execution policy.
+ * <p>
+ * It is generally recommended to use a {@code CallerBlocksPolicy} on both
+ * executors.
+ *
+ * @author Gary Russell
+ * @since 3.0.3
+ *
+ */
+public class CompositeExecutor implements Executor {
+
+	private final Executor primaryTaskExecutor;
+
+	private final Executor secondaryTaskExecutor;
+
+	public CompositeExecutor(Executor primaryTaskExecutor, Executor secondaryTaskExecutor) {
+		Assert.notNull(primaryTaskExecutor, "'primaryTaskExecutor' cannot be null");
+		Assert.notNull(primaryTaskExecutor, "'secondaryTaskExecutor' cannot be null");
+		this.primaryTaskExecutor = primaryTaskExecutor;
+		this.secondaryTaskExecutor = secondaryTaskExecutor;
+	}
+
+	/**
+	 * Execute using the primary executor.
+	 * @param task the task to run.
+	 */
+	@Override
+	public void execute(Runnable task) {
+		this.primaryTaskExecutor.execute(task);
+	}
+
+	/**
+	 * Execute using the secondary executor.
+	 * @param task the task to run.
+	 */
+	public void execute2(Runnable task) {
+		this.secondaryTaskExecutor.execute(task);
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/util/CallerBlocksPolicyTests.java

@@ -0,0 +1,111 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.util;
+
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.Test;
+
+import org.springframework.core.task.TaskRejectedException;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+/**
+ * @author Gary Russell
+ * @since 3.0.3
+ *
+ */
+public class CallerBlocksPolicyTests {
+
+	@Test
+	public void test0() throws Exception {
+		final ThreadPoolTaskExecutor te = new ThreadPoolTaskExecutor();
+		te.setCorePoolSize(1);
+		te.setMaxPoolSize(1);
+		te.setQueueCapacity(0);
+		te.setRejectedExecutionHandler(new CallerBlocksPolicy(1000));
+		te.initialize();
+		final AtomicReference<Throwable> e = new AtomicReference<Throwable>();
+		final CountDownLatch latch = new CountDownLatch(1);
+		te.execute(new Runnable() {
+
+			@Override
+			public void run() {
+				try {
+					te.execute(this);
+				}
+				catch (TaskRejectedException tre) {
+					e.set(tre.getCause());
+				}
+				latch.countDown();
+			}
+		});
+		assertTrue(latch.await(10,  TimeUnit.SECONDS));
+		assertThat(e.get(), instanceOf(RejectedExecutionException.class));
+		assertEquals("Max wait time expired to queue task", e.get().getMessage());
+	}
+
+	@Test
+	public void test1() throws Exception {
+		final ThreadPoolTaskExecutor te = new ThreadPoolTaskExecutor();
+		te.setCorePoolSize(2);
+		te.setMaxPoolSize(2);
+		te.setQueueCapacity(1);
+		te.setRejectedExecutionHandler(new CallerBlocksPolicy(10000));
+		te.initialize();
+		final AtomicReference<Throwable> e = new AtomicReference<Throwable>();
+		final CountDownLatch latch = new CountDownLatch(3);
+		te.execute(new Runnable() {
+
+			@Override
+			public void run() {
+				try {
+					Runnable foo = new Runnable() {
+
+						@Override
+						public void run() {
+							try {
+								Thread.sleep(1000);
+							}
+							catch (InterruptedException e) {
+								Thread.currentThread().interrupt();
+								throw new RuntimeException();
+							}
+							latch.countDown();
+						}
+					};
+					te.execute(foo);
+					te.execute(foo); // this one will be queued
+					te.execute(foo); // this one will be blocked and successful later
+				}
+				catch (TaskRejectedException tre) {
+					e.set(tre.getCause());
+				}
+			}
+		});
+		assertTrue(latch.await(10,  TimeUnit.SECONDS));
+		assertNull(e.get());
+	}
+
+}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java

@@ -28,6 +28,7 @@ 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.TimeUnit;
@@ -36,6 +37,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 import org.springframework.context.ApplicationEventPublisher;
 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;
@@ -59,7 +61,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 
 	private volatile boolean usingDirectBuffers;
 
-	private volatile Executor taskExecutor;
+	private volatile CompositeExecutor taskExecutor;
 
 	private volatile ByteBuffer rawBuffer;
 
@@ -327,7 +329,8 @@ public class TcpNioConnection extends TcpConnectionSupport {
 		this.writingToPipe = true;
 		try {
 			if (this.taskExecutor == null) {
-				this.taskExecutor = Executors.newCachedThreadPool();
+				ExecutorService executor = Executors.newCachedThreadPool();
+				this.taskExecutor = new CompositeExecutor(executor, executor);
 			}
 			// If there is no assembler running, start one
 			checkForAssembler();
@@ -356,7 +359,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 			 * avoid a deadlock (block on the write to the pipe).
 			 * Hence the count down latch.
 			 */
-			this.taskExecutor.execute(new Runnable() {
+			this.taskExecutor.execute2(new Runnable() {
 				@Override
 				public void run() {
 					try {
@@ -400,7 +403,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
 				if (logger.isDebugEnabled()) {
 					logger.debug(this.getConnectionId() + " Running an assembler");
 				}
-				this.taskExecutor.execute(this);
+				this.taskExecutor.execute2(this);
 			} else {
 				this.executionControl.decrementAndGet();
 			}
@@ -443,7 +446,12 @@ public class TcpNioConnection extends TcpConnectionSupport {
 	 * @param taskExecutor the taskExecutor to set
 	 */
 	public void setTaskExecutor(Executor taskExecutor) {
-		this.taskExecutor = taskExecutor;
+		if (taskExecutor instanceof CompositeExecutor) {
+			this.taskExecutor = (CompositeExecutor) taskExecutor;
+		}
+		else {
+			this.taskExecutor = new CompositeExecutor(taskExecutor, taskExecutor);
+		}
 	}
 
 	/**

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java

@@ -16,8 +16,10 @@
 
 package org.springframework.integration.ip.tcp.connection;
 
+import static org.hamcrest.Matchers.containsString;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
@@ -30,6 +32,7 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
+import java.net.ConnectException;
 import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.SocketTimeoutException;
@@ -52,6 +55,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
 
 import org.junit.Test;
 import org.mockito.Mockito;
@@ -68,7 +72,11 @@ 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;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.util.ReflectionUtils;
 import org.springframework.util.ReflectionUtils.FieldCallback;
 import org.springframework.util.ReflectionUtils.FieldFilter;
@@ -553,6 +561,68 @@ public class TcpNioConnectionTests {
 		assertEquals("baz", inboundMessage.get().getHeaders().get("bar"));
 	}
 
+	@Test
+	public void testAssemblerUsesSecondaryExecutor() throws Exception {
+		final int port = SocketUtils.findAvailableServerSocket();
+		TcpNioServerConnectionFactory factory = new TcpNioServerConnectionFactory(port);
+		factory.setApplicationEventPublisher(mock(ApplicationEventPublisher.class));
+
+		CompositeExecutor compositeExec = compositeExecutor();
+
+		factory.setSoTimeout(1000);
+		factory.setTaskExecutor(compositeExec);
+		final AtomicReference<String> threadName = new AtomicReference<String>();
+		final CountDownLatch latch = new CountDownLatch(1);
+		factory.registerListener(new TcpListener() {
+
+			@Override
+			public boolean onMessage(Message<?> message) {
+				if (!(message instanceof ErrorMessage)) {
+					threadName.set(Thread.currentThread().getName());
+					latch.countDown();
+				}
+				return false;
+			}
+
+		});
+		factory.start();
+
+		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);
+		socket.getOutputStream().write("foo\r\n".getBytes());
+		socket.close();
+
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+		assertThat(threadName.get(), containsString("assembler"));
+
+		factory.stop();
+	}
+
+	private CompositeExecutor compositeExecutor() {
+		ThreadPoolTaskExecutor ioExec = new ThreadPoolTaskExecutor();
+		ioExec.setCorePoolSize(2);
+		ioExec.setQueueCapacity(10);
+		ioExec.setThreadNamePrefix("io-");
+		ioExec.setRejectedExecutionHandler(new CallerBlocksPolicy(10000));
+		ioExec.initialize();
+		ThreadPoolTaskExecutor assemblerExec = new ThreadPoolTaskExecutor();
+		assemblerExec.setCorePoolSize(2);
+		assemblerExec.setQueueCapacity(10);
+		assemblerExec.setThreadNamePrefix("assembler-");
+		assemblerExec.setRejectedExecutionHandler(new CallerBlocksPolicy(10000));
+		assemblerExec.initialize();
+		return new CompositeExecutor(ioExec, assemblerExec);
+	}
+
 	private void readFully(InputStream is, byte[] buff) throws IOException {
 		for (int i = 0; i < buff.length; i++) {
 			buff[i] = (byte) is.read();

src/reference/docbook/ip.xml

@@ -987,13 +987,110 @@
     detection logic has been added such that if thread starvation occurs, instead of
     deadlocking, an exception is thrown, thus releasing the deadlocked resources.
    </para>
-   <note>
+   <important>
     Now that the default task executor is unbounded, it is possible that an out of
     memory condition might occur with high rates of incoming messages, if message
     processing takes extended time. If your application exhibits this type of
     behavior, you are advised to use a pooled task executor with an appropriate
-    pool size.
-   </note>
+    pool size, but see the next section.
+   </important>
+   <section>
+     <title>Thread Pool Task Executor with CALLER_RUNS Policy</title>
+     <para>
+      There are some important considerations when using a fixed thread pool with
+      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>
+      With NIO connections there are 3 distinct task types; the IO Selector processing
+      is performed on one dedicated thread - detecting events, accepting new connections,
+      and dispatching the IO read operations to other threads, using the task
+      executor. When an IO reader thread (to which the read operation is dispatched)
+      reads data, it hands off to another thread
+      to assemble the incoming message; large messages may take several reads to complete.
+      These "assembler" threads can block waiting for data. When a new read event
+      occurs, the reader determines if this socket already has an assembler and
+      runs a new one if not. When the assembly process is complete, the assembler
+      thread is returned to the pool.
+     </para>
+     <para>
+      This can cause a deadlock when the pool is exhausted and the CALLER_RUNS
+      rejection policy is in use, and the task queue is full.
+      When the pool is empty and there is no room in the queue, the IO selector thread
+      receives an <code>OP_READ</code> event and dispatches the read using the
+      executor; the queue is full, so the selector thread itself starts the
+      read process; now, it detects that there is not an assembler for this
+      socket and, before it does the read, fires off an assembler; again, the
+      queue is full, and the selector thread becomes the assembler. The assembler
+      is now blocked awaiting the data to be read, which will never happen.
+      The connection factory is now deadlocked because the selector thread
+      can't handle new events.
+     </para>
+     <para>
+      We must avoid the selector (or reader) threads performing the
+      assembly task to avoid this deadlock.
+     </para>
+     <para>
+      Two classes are provided by the framework to avoid this problem. The
+      <classname>CompositeExecutor</classname> 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 both 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.
+      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>
+     <programlisting language="java"><![CDATA[@Bean
+private CompositeExecutor compositeExecutor() {
+    ThreadPoolTaskExecutor ioExec = new ThreadPoolTaskExecutor();
+    ioExec.setCorePoolSize(4);
+    ioExec.setMaxPoolSize(8);
+    ioExec.setQueueCapacity(10);
+    ioExec.setThreadNamePrefix("io-");
+    ioExec.setRejectedExecutionHandler(new CallerRunsPolicy());
+    ioExec.initialize();
+    ThreadPoolTaskExecutor assemblerExec = new ThreadPoolTaskExecutor();
+    assemblerExec.setCorePoolSize(2);
+    assemblerExec.setMaxPoolSize(10);
+    assemblerExec.setQueueCapacity(12);
+    assemblerExec.setThreadNamePrefix("assembler-");
+    assemblerExec.setRejectedExecutionHandler(new CallerBlocksPolicy(10000));
+    assemblerExec.initialize();
+    return new CompositeExecutor(ioExec, assemblerExec);
+}]]></programlisting>
+<programlisting language="xml"><![CDATA[<bean id="myTaskExecutor" class="org.springframework.integration.util.CompositeExecutor">
+    <constructor-arg>
+        <bean class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
+            <property name="threadNamePrefix" value="io-" />
+            <property name="corePoolSize" value="4" />
+            <property name="maxPoolSize" value="8" />
+            <property name="queueCapacity" value="10" />
+            <property name="rejectedExecutionHandler">
+                <bean class="org.springframework.integration.util.CallerBlocksPolicy">
+                    <constructor-arg value="10000" />
+                </bean>
+            </property>
+        </bean>
+    </constructor-arg>
+    <constructor-arg>
+        <bean class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
+            <property name="threadNamePrefix" value="assembler-" />
+            <property name="corePoolSize" value="4" />
+            <property name="maxPoolSize" value="10" />
+            <property name="queueCapacity" value="10" />
+            <property name="rejectedExecutionHandler">
+                <bean class="org.springframework.integration.util.CallerBlocksPolicy">
+                    <constructor-arg value="10000" />
+                </bean>
+            </property>
+        </bean>
+    </constructor-arg>
+</bean>]]></programlisting>
+   </section>
   </section>
   <section id="ssl-tls">
     <title>SSL/TLS Support</title>