GH-2837: TCP NIO Prioritize accept

See #2837

Give priority to accepting new connections over reading data from
existing connections.
Add a flag to allow users to revert to the previous behavior of
giving reads priority.

**cherry-pick to 5.1.x**

# Conflicts:
#	spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java
#	src/reference/asciidoc/whats-new.adoc
#   TcpNioConnectionTests.java

Author: garyrussell

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

@@ -476,6 +476,7 @@ public void setSslHandshakeTimeout(int sslHandshakeTimeout) {
 	 * @see #setSslHandshakeTimeout(int)
 	 * @since 4.3.6
 	 */
+	@Nullable
 	protected Integer getSslHandshakeTimeout() {
 		return this.sslHandshakeTimeout;
 	}
@@ -724,7 +725,7 @@ else if (key.isAcceptable()) {
 							doAccept(selector, server, now);
 						}
 						catch (Exception e) {
-							logger.error("Exception accepting new connection", e);
+							logger.error("Exception accepting new connection(s)", e);
 						}
 					}
 					else {

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

@@ -17,6 +17,7 @@
 package org.springframework.integration.ip.tcp.connection;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.Socket;
@@ -34,7 +35,7 @@
 import org.springframework.util.Assert;
 
 /**
-/**
+ /**
  * Implements a server connection factory that produces {@link TcpNioConnection}s using
  * a {@link ServerSocketChannel}. Must have a {@link TcpListener} registered.
  *
@@ -45,6 +46,8 @@
  */
 public class TcpNioServerConnectionFactory extends AbstractServerConnectionFactory {
 
+	private boolean multiAccept = true;
+
 	private volatile ServerSocketChannel serverChannel;
 
 	private volatile boolean usingDirectBuffers;
@@ -63,6 +66,18 @@ public TcpNioServerConnectionFactory(int port) {
 		super(port);
 	}
 
+	/**
+	 * Set to false to only accept one connection per iteration over the
+	 * selector keys. This might be necessary to avoid accepts overwhelming
+	 * reads of existing sockets. By default when the {@code OP_ACCEPT} operation
+	 * is ready, we will keep accepting connections in a loop until no more arrive.
+	 * @param multiAccept false to accept connections one-at-a-time.
+	 * @since 5.1.4
+	 */
+	public void setMultiAccept(boolean multiAccept) {
+		this.multiAccept = multiAccept;
+	}
+
 	@Override
 	public String getComponentType() {
 		return "tcp-nio-server-connection-factory";
@@ -193,55 +208,65 @@ private void doSelect(ServerSocketChannel server, final Selector selector) throw
 	}
 
 	/**
-	 * @param selector The selector.
+	 * @param selectorForNewSocket The selector.
 	 * @param server The server socket channel.
 	 * @param now The current time.
-	 * @throws IOException Any IOException.
 	 */
 	@Override
-	protected void doAccept(final Selector selector, ServerSocketChannel server, long now) throws IOException {
+	protected void doAccept(final Selector selectorForNewSocket, ServerSocketChannel server, long now) {
 		logger.debug("New accept");
-		SocketChannel channel = server.accept();
-		if (isShuttingDown()) {
-			if (logger.isInfoEnabled()) {
-				logger.info("New connection from " + channel.socket().getInetAddress().getHostAddress()
-						+ ":" + channel.socket().getPort()
-						+ " rejected; the server is in the process of shutting down.");
-			}
-			channel.close();
-		}
-		else {
-			try {
-				channel.configureBlocking(false);
-				Socket socket = channel.socket();
-				setSocketAttributes(socket);
-				TcpNioConnection connection = createTcpNioConnection(channel);
-				if (connection == null) {
-					return;
-				}
-				connection.setTaskExecutor(getTaskExecutor());
-				connection.setLastRead(now);
-				if (getSslHandshakeTimeout() != null && connection instanceof TcpNioSSLConnection) {
-					((TcpNioSSLConnection) connection).setHandshakeTimeout(getSslHandshakeTimeout());
+		try {
+			SocketChannel channel = null;
+			do {
+				channel = server.accept();
+				if (channel != null) {
+					if (isShuttingDown()) {
+						if (logger.isInfoEnabled()) {
+							logger.info("New connection from " + channel.socket().getInetAddress().getHostAddress()
+									+ ":" + channel.socket().getPort()
+									+ " rejected; the server is in the process of shutting down.");
+						}
+						channel.close();
+					}
+					else {
+						try {
+							channel.configureBlocking(false);
+							Socket socket = channel.socket();
+							setSocketAttributes(socket);
+							TcpNioConnection connection = createTcpNioConnection(channel);
+							if (connection == null) {
+								return;
+							}
+							connection.setTaskExecutor(getTaskExecutor());
+							connection.setLastRead(now);
+							if (getSslHandshakeTimeout() != null && connection instanceof TcpNioSSLConnection) {
+								((TcpNioSSLConnection) connection).setHandshakeTimeout(getSslHandshakeTimeout());
+							}
+							this.channelMap.put(channel, connection);
+							channel.register(selectorForNewSocket, SelectionKey.OP_READ, connection);
+							connection.publishConnectionOpenEvent();
+						}
+						catch (IOException e) {
+							logger.error("Exception accepting new connection from "
+									+ channel.socket().getInetAddress().getHostAddress()
+									+ ":" + channel.socket().getPort(), e);
+							channel.close();
+						}
+					}
 				}
-				this.channelMap.put(channel, connection);
-				channel.register(selector, SelectionKey.OP_READ, connection);
-				connection.publishConnectionOpenEvent();
-			}
-			catch (Exception e) {
-				logger.error("Exception accepting new connection from "
-						+ channel.socket().getInetAddress().getHostAddress()
-						+ ":" + channel.socket().getPort(), e);
-				channel.close();
 			}
+			while (this.multiAccept && channel != null);
+		}
+		catch (IOException e) {
+			throw new UncheckedIOException(e);
 		}
 	}
 
 	@Nullable
 	private TcpNioConnection createTcpNioConnection(SocketChannel socketChannel) {
 		try {
 			TcpNioConnection connection = this.tcpNioConnectionSupport.createNewConnection(socketChannel, true,
-							isLookupHost(), getApplicationEventPublisher(), getComponentName());
+					isLookupHost(), getApplicationEventPublisher(), getComponentName());
 			connection.setUsingDirectBuffers(this.usingDirectBuffers);
 			TcpConnectionSupport wrappedConnection = wrapConnection(connection);
 			initializeConnection(wrappedConnection, socketChannel.socket());

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

@@ -47,6 +47,7 @@
 import java.nio.channels.SocketChannel;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -79,6 +80,7 @@
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.core.task.AsyncTaskExecutor;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.integration.ip.event.IpIntegrationEvent;
 import org.springframework.integration.ip.tcp.connection.TcpNioConnection.ChannelInputStream;
 import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
 import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer;
@@ -510,7 +512,7 @@ public void transferHeaders() throws Exception {
 		doAnswer(new Answer<Integer>() {
 
 			@Override
-			public Integer answer(InvocationOnMock invocation) throws Throwable {
+			public Integer answer(InvocationOnMock invocation) {
 				ByteBuffer buff = invocation.getArgument(0);
 				byte[] bytes = written.toByteArray();
 				buff.put(bytes);
@@ -840,6 +842,57 @@ public void testNoDelayOnClose() throws Exception {
 		assertThat(watch.getLastTaskTimeMillis(), lessThan(950L));
 	}
 
+	@Test
+	public void testMultiAccept() throws InterruptedException, IOException {
+		testMulti(true);
+	}
+
+	@Test
+	public void testNoMultiAccept() throws InterruptedException, IOException {
+		testMulti(false);
+	}
+
+	private void testMulti(boolean multiAccept) throws InterruptedException, IOException {
+		CountDownLatch serverReadyLatch = new CountDownLatch(1);
+		CountDownLatch latch = new CountDownLatch(21);
+		List<Socket> sockets = new ArrayList<>();
+		TcpNioServerConnectionFactory server = new TcpNioServerConnectionFactory(0);
+		try {
+			List<IpIntegrationEvent> events = Collections.synchronizedList(new ArrayList<>());
+			List<Message<?>> messages = Collections.synchronizedList(new ArrayList<>());
+			server.setMultiAccept(multiAccept);
+			server.setApplicationEventPublisher(e -> {
+				if (e instanceof TcpConnectionServerListeningEvent) {
+					serverReadyLatch.countDown();
+				}
+				events.add((IpIntegrationEvent) e);
+				latch.countDown();
+			});
+			server.registerListener(m -> {
+				messages.add(m);
+				latch.countDown();
+				return false;
+			});
+			server.afterPropertiesSet();
+			server.start();
+			assertTrue(serverReadyLatch.await(10, TimeUnit.SECONDS));
+			for (int i = 0; i < 10; i++) {
+				Socket socket = SocketFactory.getDefault().createSocket("localhost", server.getPort());
+				socket.getOutputStream().write("foo\r\n".getBytes());
+				sockets.add(socket);
+			}
+			assertTrue(latch.await(10, TimeUnit.SECONDS));
+			assertEquals(11, events.size()); // server ready + 10 opens
+			assertEquals(10, messages.size());
+		}
+		finally {
+			for (Socket socket : sockets) {
+				socket.close();
+			}
+			server.stop();
+		}
+	}
+
 	private void readFully(InputStream is, byte[] buff) throws IOException {
 		for (int i = 0; i < buff.length; i++) {
 			buff[i] = (byte) is.read();

src/reference/asciidoc/ip.adoc

@@ -971,6 +971,10 @@ Alternatively, you can insert a resequencer downstream of the inbound endpoint t
 If you set `apply-sequence` to `true` on the connection factory, messages arriving on a TCP connection have `sequenceNumber` and `correlationId` headers set.
 The resequencer uses these headers to return the messages to their proper sequence.
 
+IMPORTANT: Starting with version 5.1.4, priority is given to accepting new connections over reading from existing connections.
+This should, generally, have little impact unless you have a very high rate of new incoming connections.
+If you wish to revert to the previous behavior of giving reads priority, set the `multiAccept` property on the `TcpNioServerConnectionFactory` to `false`.
+
 ==== Pool Size
 
 The pool size attribute is no longer used.