Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2837: TCP NIO Prioritize accept

Browse Source 
See https://github.com/spring-projects/spring-integration/pull/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
This commit is contained in:
Gary Russell
2019-03-28 11:57:06 -04:00
committed by Artem Bilan
parent 6e9bb60a61
commit 6f3fd0961f
4 changed files with 121 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@@ -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.lang.Nullable;
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 @@ import org.springframework.util.Assert;
*/
public class TcpNioServerConnectionFactory extends AbstractServerConnectionFactory {
private boolean multiAccept = true;
private volatile ServerSocketChannel serverChannel;
private volatile boolean usingDirectBuffers;
@@ -63,6 +66,18 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
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,47 +208,57 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
}
/**
* @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.");
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();
}
}
}
}
channel.close();
while (this.multiAccept && channel != null);
}
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(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();
}
catch (IOException e) {
throw new UncheckedIOException(e);
}
}
@@ -241,7 +266,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
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());

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

@@ -47,6 +47,7 @@ import java.nio.channels.Selector;
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.ApplicationEvent;
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 class TcpNioConnectionTests {
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 class TcpNioConnectionTests {
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();

4
src/reference/asciidoc/ip.adoc
View File

@@ -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.