andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fixed issue: NETTY-380 releaseExternalResources() hang indefinitely

Browse Source 
when called from a handler

* Replaced IoWorkerRunnable with DeadLockProofWorker
* ExecutorUtil now checks dead lock
This commit is contained in:
Trustin Lee 2011-02-01 10:56:59 +09:00
parent 339c2a6641
commit be6cdb4a11
12 changed files with 92 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/java/org/jboss/netty/channel/DefaultChannelFuture.java
View File

@ -24,7 +24,7 @@ import java.util.concurrent.TimeUnit;
import org.jboss.netty.logging.InternalLogger;
import org.jboss.netty.logging.InternalLoggerFactory;
import org.jboss.netty.util.internal.IoWorkerRunnable;
import org.jboss.netty.util.internal.DeadLockProofWorker;
/**
* The default {@link ChannelFuture} implementation. It is recommended to
@ -292,7 +292,7 @@ public class DefaultChannelFuture implements ChannelFuture {
}
private void checkDeadLock() {
if (isUseDeadLockChecker() && IoWorkerRunnable.IN_IO_THREAD.get()) {
if (isUseDeadLockChecker() && DeadLockProofWorker.PARENT.get() != null) {
throw new IllegalStateException(
"await*() in I/O thread causes a dead lock or " +
"sudden performance drop. Use addListener() instead or " +

4
src/main/java/org/jboss/netty/channel/group/DefaultChannelGroupFuture.java
View File

@ -31,7 +31,7 @@ import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.logging.InternalLogger;
import org.jboss.netty.logging.InternalLoggerFactory;
import org.jboss.netty.util.internal.IoWorkerRunnable;
import org.jboss.netty.util.internal.DeadLockProofWorker;
/**
* The default {@link ChannelGroupFuture} implementation.
@ -320,7 +320,7 @@ public class DefaultChannelGroupFuture implements ChannelGroupFuture {
}
private void checkDeadLock() {
if (IoWorkerRunnable.IN_IO_THREAD.get()) {
if (DeadLockProofWorker.PARENT.get() != null) {
throw new IllegalStateException(
"await*() in I/O thread causes a dead lock or " +
"sudden performance drop. Use addListener() instead or " +

10
src/main/java/org/jboss/netty/channel/socket/nio/NioClientSocketPipelineSink.java
View File

@ -43,7 +43,7 @@ import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.logging.InternalLogger;
import org.jboss.netty.logging.InternalLoggerFactory;
import org.jboss.netty.util.ThreadRenamingRunnable;
import org.jboss.netty.util.internal.IoWorkerRunnable;
import org.jboss.netty.util.internal.DeadLockProofWorker;
import org.jboss.netty.util.internal.LinkedTransferQueue;
/**
@ -194,10 +194,10 @@ class NioClientSocketPipelineSink extends AbstractChannelSink {
// Start the worker thread with the new Selector.
boolean success = false;
try {
bossExecutor.execute(
new IoWorkerRunnable(
new ThreadRenamingRunnable(
this, "New I/O client boss #" + id)));
DeadLockProofWorker.start(
bossExecutor,
new ThreadRenamingRunnable(
this, "New I/O client boss #" + id));
success = true;
} finally {
if (!success) {

13
src/main/java/org/jboss/netty/channel/socket/nio/NioServerSocketPipelineSink.java
View File

@ -40,7 +40,7 @@ import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.logging.InternalLogger;
import org.jboss.netty.logging.InternalLoggerFactory;
import org.jboss.netty.util.ThreadRenamingRunnable;
import org.jboss.netty.util.internal.IoWorkerRunnable;
import org.jboss.netty.util.internal.DeadLockProofWorker;
/**
*
@ -153,12 +153,11 @@ class NioServerSocketPipelineSink extends AbstractChannelSink {
Executor bossExecutor =
((NioServerSocketChannelFactory) channel.getFactory()).bossExecutor;
bossExecutor.execute(
new IoWorkerRunnable(
new ThreadRenamingRunnable(
new Boss(channel),
"New I/O server boss #" + id +
" (" + channel + ')')));
DeadLockProofWorker.start(
bossExecutor,
new ThreadRenamingRunnable(
new Boss(channel),
"New I/O server boss #" + id + " (" + channel + ')'));
bossStarted = true;
} catch (Throwable t) {
future.setFailure(t);

7
src/main/java/org/jboss/netty/channel/socket/nio/NioWorker.java
View File

@ -47,7 +47,7 @@ import org.jboss.netty.channel.socket.nio.SocketSendBufferPool.SendBuffer;
import org.jboss.netty.logging.InternalLogger;
import org.jboss.netty.logging.InternalLoggerFactory;
import org.jboss.netty.util.ThreadRenamingRunnable;
import org.jboss.netty.util.internal.IoWorkerRunnable;
import org.jboss.netty.util.internal.DeadLockProofWorker;
import org.jboss.netty.util.internal.LinkedTransferQueue;
/**
@ -112,9 +112,8 @@ class NioWorker implements Runnable {
boolean success = false;
try {
executor.execute(
new IoWorkerRunnable(
new ThreadRenamingRunnable(this, threadName)));
DeadLockProofWorker.start(
executor, new ThreadRenamingRunnable(this, threadName));
success = true;
} finally {
if (!success) {

12
src/main/java/org/jboss/netty/channel/socket/oio/OioClientSocketPipelineSink.java
View File

@ -30,7 +30,7 @@ import org.jboss.netty.channel.ChannelState;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.util.ThreadRenamingRunnable;
import org.jboss.netty.util.internal.IoWorkerRunnable;
import org.jboss.netty.util.internal.DeadLockProofWorker;
/**
*
@ -127,11 +127,11 @@ class OioClientSocketPipelineSink extends AbstractChannelSink {
fireChannelConnected(channel, channel.getRemoteAddress());
// Start the business.
workerExecutor.execute(
new IoWorkerRunnable(
new ThreadRenamingRunnable(
new OioWorker(channel),
"Old I/O client worker (" + channel + ')')));
DeadLockProofWorker.start(
workerExecutor,
new ThreadRenamingRunnable(
new OioWorker(channel),
"Old I/O client worker (" + channel + ')'));
workerStarted = true;
} catch (Throwable t) {
future.setFailure(t);

20
src/main/java/org/jboss/netty/channel/socket/oio/OioDatagramPipelineSink.java
View File

@ -29,7 +29,7 @@ import org.jboss.netty.channel.ChannelState;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.util.ThreadRenamingRunnable;
import org.jboss.netty.util.internal.IoWorkerRunnable;
import org.jboss.netty.util.internal.DeadLockProofWorker;
/**
*
@ -100,11 +100,11 @@ class OioDatagramPipelineSink extends AbstractChannelSink {
fireChannelBound(channel, channel.getLocalAddress());
// Start the business.
workerExecutor.execute(
new IoWorkerRunnable(
new ThreadRenamingRunnable(
new OioDatagramWorker(channel),
"Old I/O datagram worker (" + channel + ')')));
DeadLockProofWorker.start(
workerExecutor,
new ThreadRenamingRunnable(
new OioDatagramWorker(channel),
"Old I/O datagram worker (" + channel + ')'));
workerStarted = true;
} catch (Throwable t) {
future.setFailure(t);
@ -144,10 +144,10 @@ class OioDatagramPipelineSink extends AbstractChannelSink {
String threadName = "Old I/O datagram worker (" + channel + ')';
if (!bound) {
// Start the business.
workerExecutor.execute(
new IoWorkerRunnable(
new ThreadRenamingRunnable(
new OioDatagramWorker(channel), threadName)));
DeadLockProofWorker.start(
workerExecutor,
new ThreadRenamingRunnable(
new OioDatagramWorker(channel), threadName));
} else {
// Worker started by bind() - just rename.
Thread workerThread = channel.workerThread;

24
src/main/java/org/jboss/netty/channel/socket/oio/OioServerSocketPipelineSink.java
View File

@ -34,7 +34,7 @@ import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.logging.InternalLogger;
import org.jboss.netty.logging.InternalLoggerFactory;
import org.jboss.netty.util.ThreadRenamingRunnable;
import org.jboss.netty.util.internal.IoWorkerRunnable;
import org.jboss.netty.util.internal.DeadLockProofWorker;
/**
*
@ -143,11 +143,11 @@ class OioServerSocketPipelineSink extends AbstractChannelSink {
Executor bossExecutor =
((OioServerSocketChannelFactory) channel.getFactory()).bossExecutor;
bossExecutor.execute(
new IoWorkerRunnable(
new ThreadRenamingRunnable(
new Boss(channel),
"Old I/O server boss (" + channel + ')')));
DeadLockProofWorker.start(
bossExecutor,
new ThreadRenamingRunnable(
new Boss(channel),
"Old I/O server boss (" + channel + ')'));
bossStarted = true;
} catch (Throwable t) {
future.setFailure(t);
@ -210,12 +210,12 @@ class OioServerSocketPipelineSink extends AbstractChannelSink {
pipeline,
OioServerSocketPipelineSink.this,
acceptedSocket);
workerExecutor.execute(
new IoWorkerRunnable(
new ThreadRenamingRunnable(
new OioWorker(acceptedChannel),
"Old I/O server worker (parentId: " +
channel.getId() + ", " + channel + ')')));
DeadLockProofWorker.start(
workerExecutor,
new ThreadRenamingRunnable(
new OioWorker(acceptedChannel),
"Old I/O server worker (parentId: " +
channel.getId() + ", " + channel + ')'));
} catch (Exception e) {
logger.warn(
"Failed to initialize an accepted socket.", e);

4
src/main/java/org/jboss/netty/handler/queue/BlockingReadHandler.java
View File

@ -28,7 +28,7 @@ import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.util.internal.IoWorkerRunnable;
import org.jboss.netty.util.internal.DeadLockProofWorker;
import org.jboss.netty.util.internal.LinkedTransferQueue;
/**
@ -244,7 +244,7 @@ public class BlockingReadHandler<E> extends SimpleChannelUpstreamHandler {
}
private void detectDeadLock() {
if (IoWorkerRunnable.IN_IO_THREAD.get()) {
if (DeadLockProofWorker.PARENT.get() != null) {
throw new IllegalStateException(
"read*(...) in I/O thread causes a dead lock or " +
"sudden performance drop. Implement a state machine or " +

40
src/main/java/org/jboss/netty/util/internal/IoWorkerRunnable.java → src/main/java/org/jboss/netty/util/internal/DeadLockProofWorker.java
View File

@ -15,37 +15,41 @@
*/
package org.jboss.netty.util.internal;
import org.jboss.netty.channel.ChannelFuture;
import java.util.concurrent.Executor;
/**
* @author <a href="http://gleamynode.net/">Trustin Lee</a>
* @version $Rev: 2080 $, $Date: 2010-01-26 18:04:19 +0900 (Tue, 26 Jan 2010) $
*/
public class IoWorkerRunnable implements Runnable {
public final class DeadLockProofWorker {
/**
* An <em>internal use only</em> thread-local variable that determines if
* the caller is running on an I/O worker thread, which is the case where
* the caller enters a dead lock if the caller calls
* {@link ChannelFuture#await()} or {@link ChannelFuture#awaitUninterruptibly()}.
* An <em>internal use only</em> thread-local variable that tells the
* {@link Executor} that this worker acquired a worker thread from.
*/
public static final ThreadLocal<Boolean> IN_IO_THREAD = new ThreadLocalBoolean();
public static final ThreadLocal<Executor> PARENT = new ThreadLocal<Executor>();
private final Runnable runnable;
public IoWorkerRunnable(Runnable runnable) {
public static void start(final Executor parent, final Runnable runnable) {
if (parent == null) {
throw new NullPointerException("parent");
}
if (runnable == null) {
throw new NullPointerException("runnable");
}
this.runnable = runnable;
parent.execute(new Runnable() {
public void run() {
PARENT.set(parent);
try {
runnable.run();
} finally {
PARENT.remove();
}
}
});
}
public void run() {
IN_IO_THREAD.set(Boolean.TRUE);
try {
runnable.run();
} finally {
IN_IO_THREAD.remove();
}
private DeadLockProofWorker() {
super();
}
}

20
src/main/java/org/jboss/netty/util/internal/ExecutorUtil.java
View File

@ -50,6 +50,11 @@ public class ExecutorUtil {
* Shuts down the specified executors.
*/
public static void terminate(Executor... executors) {
// Check nulls.
if (executors == null) {
throw new NullPointerException("executors");
}
Executor[] executorsCopy = new Executor[executors.length];
for (int i = 0; i < executors.length; i ++) {
if (executors[i] == null) {
@ -58,6 +63,21 @@ public class ExecutorUtil {
executorsCopy[i] = executors[i];
}
// Check dead lock.
final Executor currentParent = DeadLockProofWorker.PARENT.get();
if (currentParent != null) {
for (Executor e: executorsCopy) {
if (e == currentParent) {
throw new IllegalStateException(
"An Executor cannot be shut down from the thread " +
"acquired from itself. Please make sure you are " +
"not calling releaseExternalResources() from an " +
"I/O worker thread.");
}
}
}
// Shut down all executors.
boolean interrupted = false;
for (Executor e: executorsCopy) {
if (!(e instanceof ExecutorService)) {

4
src/main/java/org/jboss/netty/util/internal/StackTraceSimplifier.java
View File

@ -42,8 +42,8 @@ public class StackTraceSimplifier {
private static final Pattern EXCLUDED_STACK_TRACE =
Pattern.compile(
"^org\\.jboss\\.netty\\." +
"(util\\.(ThreadRenamingRunnable)" +
"|channel\\.(SimpleChannel(Upstream|Downstream)?Handler|(Default|Static)ChannelPipeline.*))$");
"(util\\.(ThreadRenamingRunnable|internal\\.DeadLockProofWorker)" +
"|channel\\.(SimpleChannel(Upstream|Downstream)?Handler|(Default|Static)ChannelPipeline.*))(\\$.*)?$");
/**
* Removes unnecessary {@link StackTraceElement}s from the specified