Fix a regression in SslHandler where delegated tasks run in a different executor makes the session hang
- Fixes #2098
This commit is contained in:
parent
fc9a794613
commit
8198f23dbb
@ -49,7 +49,9 @@ import java.nio.ByteBuffer;
|
|||||||
import java.nio.channels.ClosedChannelException;
|
import java.nio.channels.ClosedChannelException;
|
||||||
import java.nio.channels.DatagramChannel;
|
import java.nio.channels.DatagramChannel;
|
||||||
import java.nio.channels.SocketChannel;
|
import java.nio.channels.SocketChannel;
|
||||||
|
import java.util.ArrayList;
|
||||||
import java.util.LinkedList;
|
import java.util.LinkedList;
|
||||||
|
import java.util.List;
|
||||||
import java.util.Queue;
|
import java.util.Queue;
|
||||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||||
import java.util.concurrent.Executor;
|
import java.util.concurrent.Executor;
|
||||||
@ -431,7 +433,7 @@ public class SslHandler extends FrameDecoder
|
|||||||
handshaking = true;
|
handshaking = true;
|
||||||
try {
|
try {
|
||||||
engine.beginHandshake();
|
engine.beginHandshake();
|
||||||
runDelegatedTasks();
|
runDelegatedTasks(false);
|
||||||
handshakeFuture = this.handshakeFuture = future(channel);
|
handshakeFuture = this.handshakeFuture = future(channel);
|
||||||
if (handshakeTimeoutInMillis > 0) {
|
if (handshakeTimeoutInMillis > 0) {
|
||||||
handshakeTimeout = timer.newTimeout(new TimerTask() {
|
handshakeTimeout = timer.newTimeout(new TimerTask() {
|
||||||
@ -1037,7 +1039,7 @@ public class SslHandler extends FrameDecoder
|
|||||||
needsUnwrap = true;
|
needsUnwrap = true;
|
||||||
break loop;
|
break loop;
|
||||||
case NEED_TASK:
|
case NEED_TASK:
|
||||||
runDelegatedTasks();
|
runDelegatedTasks(false);
|
||||||
break;
|
break;
|
||||||
case FINISHED:
|
case FINISHED:
|
||||||
case NOT_HANDSHAKING:
|
case NOT_HANDSHAKING:
|
||||||
@ -1182,10 +1184,10 @@ public class SslHandler extends FrameDecoder
|
|||||||
switch (handshakeStatus) {
|
switch (handshakeStatus) {
|
||||||
case FINISHED:
|
case FINISHED:
|
||||||
setHandshakeSuccess(channel);
|
setHandshakeSuccess(channel);
|
||||||
runDelegatedTasks();
|
runDelegatedTasks(false);
|
||||||
break;
|
break;
|
||||||
case NEED_TASK:
|
case NEED_TASK:
|
||||||
runDelegatedTasks();
|
runDelegatedTasks(false);
|
||||||
break;
|
break;
|
||||||
case NEED_UNWRAP:
|
case NEED_UNWRAP:
|
||||||
if (!Thread.holdsLock(handshakeLock)) {
|
if (!Thread.holdsLock(handshakeLock)) {
|
||||||
@ -1294,7 +1296,7 @@ public class SslHandler extends FrameDecoder
|
|||||||
wrapNonAppData(ctx, channel);
|
wrapNonAppData(ctx, channel);
|
||||||
break;
|
break;
|
||||||
case NEED_TASK:
|
case NEED_TASK:
|
||||||
runDelegatedTasks();
|
runDelegatedTasks(true);
|
||||||
break;
|
break;
|
||||||
case FINISHED:
|
case FINISHED:
|
||||||
setHandshakeSuccess(channel);
|
setHandshakeSuccess(channel);
|
||||||
@ -1393,21 +1395,64 @@ public class SslHandler extends FrameDecoder
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private void runDelegatedTasks() {
|
/**
|
||||||
for (;;) {
|
* Fetches all delegated tasks from the {@link SSLEngine} and runs them via the {@link #delegatedTaskExecutor}.
|
||||||
final Runnable task;
|
* If the {@link #delegatedTaskExecutor} is {@link ImmediateExecutor}, just call {@link Runnable#run()} directly
|
||||||
synchronized (handshakeLock) {
|
* instead of using {@link Executor#execute(Runnable)}. Otherwise, run the tasks via
|
||||||
task = engine.getDelegatedTask();
|
* the {@link #delegatedTaskExecutor} and continue unwrapping so that the handshake completes.
|
||||||
|
*/
|
||||||
|
private void runDelegatedTasks(final boolean unwrapLater) {
|
||||||
|
if (delegatedTaskExecutor == ImmediateExecutor.INSTANCE) {
|
||||||
|
for (;;) {
|
||||||
|
final Runnable task;
|
||||||
|
synchronized (handshakeLock) {
|
||||||
|
task = engine.getDelegatedTask();
|
||||||
|
}
|
||||||
|
|
||||||
|
if (task == null) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
delegatedTaskExecutor.execute(task);
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
final List<Runnable> tasks = new ArrayList<Runnable>(2);
|
||||||
|
for (;;) {
|
||||||
|
final Runnable task;
|
||||||
|
synchronized (handshakeLock) {
|
||||||
|
task = engine.getDelegatedTask();
|
||||||
|
}
|
||||||
|
|
||||||
|
if (task == null) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
tasks.add(task);
|
||||||
}
|
}
|
||||||
|
|
||||||
if (task == null) {
|
if (tasks.isEmpty()) {
|
||||||
break;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
delegatedTaskExecutor.execute(new Runnable() {
|
delegatedTaskExecutor.execute(new Runnable() {
|
||||||
public void run() {
|
public void run() {
|
||||||
synchronized (handshakeLock) {
|
try {
|
||||||
task.run();
|
for (Runnable task: tasks) {
|
||||||
|
task.run();
|
||||||
|
}
|
||||||
|
if (unwrapLater) {
|
||||||
|
ctx.getPipeline().execute(new Runnable() {
|
||||||
|
public void run() {
|
||||||
|
try {
|
||||||
|
unwrap(ctx, ctx.getChannel(), ChannelBuffers.EMPTY_BUFFER, 0, 0);
|
||||||
|
} catch (Exception e) {
|
||||||
|
fireExceptionCaught(ctx, e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
} catch (Exception e) {
|
||||||
|
fireExceptionCaught(ctx, e);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
@ -1664,7 +1709,7 @@ public class SslHandler extends FrameDecoder
|
|||||||
}
|
}
|
||||||
|
|
||||||
private final class SSLEngineInboundCloseFuture extends DefaultChannelFuture {
|
private final class SSLEngineInboundCloseFuture extends DefaultChannelFuture {
|
||||||
public SSLEngineInboundCloseFuture() {
|
SSLEngineInboundCloseFuture() {
|
||||||
super(null, true);
|
super(null, true);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -66,7 +66,29 @@ public abstract class AbstractSocketSslEchoTest {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void testSslEcho() throws Throwable {
|
public void testSslEcho1() throws Throwable {
|
||||||
|
testSslEcho(false, false);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testSslEcho2() throws Throwable {
|
||||||
|
testSslEcho(false, true);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testSslEcho3() throws Throwable {
|
||||||
|
testSslEcho(true, false);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testSslEcho4() throws Throwable {
|
||||||
|
testSslEcho(true, true);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
private void testSslEcho(
|
||||||
|
boolean serverUsesDelegatedTaskExecutor, boolean clientUsesDelegatedTaskExecutor) throws Throwable {
|
||||||
|
ExecutorService delegatedTaskExecutor = Executors.newCachedThreadPool();
|
||||||
ServerBootstrap sb = new ServerBootstrap(newServerSocketChannelFactory(Executors.newCachedThreadPool()));
|
ServerBootstrap sb = new ServerBootstrap(newServerSocketChannelFactory(Executors.newCachedThreadPool()));
|
||||||
ClientBootstrap cb = new ClientBootstrap(newClientSocketChannelFactory(Executors.newCachedThreadPool()));
|
ClientBootstrap cb = new ClientBootstrap(newClientSocketChannelFactory(Executors.newCachedThreadPool()));
|
||||||
|
|
||||||
@ -82,10 +104,22 @@ public abstract class AbstractSocketSslEchoTest {
|
|||||||
sb.setOption("receiveBufferSize", 1048576);
|
sb.setOption("receiveBufferSize", 1048576);
|
||||||
sb.setOption("receiveBufferSize", 1048576);
|
sb.setOption("receiveBufferSize", 1048576);
|
||||||
|
|
||||||
sb.getPipeline().addFirst("ssl", new SslHandler(sse));
|
// Configure the server pipeline.
|
||||||
|
if (serverUsesDelegatedTaskExecutor) {
|
||||||
|
sb.getPipeline().addFirst("ssl", new SslHandler(sse, delegatedTaskExecutor));
|
||||||
|
} else {
|
||||||
|
sb.getPipeline().addFirst("ssl", new SslHandler(sse));
|
||||||
|
}
|
||||||
sb.getPipeline().addLast("handler", sh);
|
sb.getPipeline().addLast("handler", sh);
|
||||||
cb.getPipeline().addFirst("ssl", new SslHandler(cse));
|
|
||||||
|
// Configure the client pipeline.
|
||||||
|
if (clientUsesDelegatedTaskExecutor) {
|
||||||
|
cb.getPipeline().addFirst("ssl", new SslHandler(cse, delegatedTaskExecutor));
|
||||||
|
} else {
|
||||||
|
cb.getPipeline().addFirst("ssl", new SslHandler(cse));
|
||||||
|
}
|
||||||
cb.getPipeline().addLast("handler", ch);
|
cb.getPipeline().addLast("handler", ch);
|
||||||
|
|
||||||
ExecutorService eventExecutor = null;
|
ExecutorService eventExecutor = null;
|
||||||
if (isExecutorRequired()) {
|
if (isExecutorRequired()) {
|
||||||
eventExecutor = new OrderedMemoryAwareThreadPoolExecutor(16, 0, 0);
|
eventExecutor = new OrderedMemoryAwareThreadPoolExecutor(16, 0, 0);
|
||||||
@ -159,6 +193,7 @@ public abstract class AbstractSocketSslEchoTest {
|
|||||||
sb.shutdown();
|
sb.shutdown();
|
||||||
cb.releaseExternalResources();
|
cb.releaseExternalResources();
|
||||||
sb.releaseExternalResources();
|
sb.releaseExternalResources();
|
||||||
|
delegatedTaskExecutor.shutdown();
|
||||||
|
|
||||||
if (eventExecutor != null) {
|
if (eventExecutor != null) {
|
||||||
eventExecutor.shutdown();
|
eventExecutor.shutdown();
|
||||||
|
Loading…
x
Reference in New Issue
Block a user