Refactor ChunkedWriteHandler to remove synchronization which can have bad side effects like deadlocks. See #297 and #301
This commit is contained in:
norman
parent
2a249c14b1
commit
3a99550132
@ -20,6 +20,7 @@ import static org.jboss.netty.channel.Channels.*;
|
||||
import java.io.IOException;
|
||||
import java.nio.channels.ClosedChannelException;
|
||||
import java.util.Queue;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import org.jboss.netty.buffer.ChannelBuffers;
|
||||
import org.jboss.netty.channel.Channel;
|
||||
@ -80,6 +81,7 @@ public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDowns
|
||||
private final Queue<MessageEvent> queue = QueueFactory.createQueue(MessageEvent.class);
|
||||
|
||||
private volatile ChannelHandlerContext ctx;
|
||||
private final AtomicBoolean flush = new AtomicBoolean(false);
|
||||
private MessageEvent currentEvent;
|
||||
|
||||
/**
|
||||
@ -111,10 +113,9 @@ public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDowns
|
||||
assert offered;
|
||||
|
||||
final Channel channel = ctx.getChannel();
|
||||
if (channel.isWritable()) {
|
||||
this.ctx = ctx;
|
||||
flush(ctx, false);
|
||||
} else if (!channel.isConnected()) {
|
||||
// call flush if the channel is writable or not connected. flush(..) will take care of the rest
|
||||
|
||||
if (channel.isWritable() || !channel.isConnected()) {
|
||||
this.ctx = ctx;
|
||||
discard(ctx, false);
|
||||
}
|
||||
@ -142,7 +143,6 @@ public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDowns
|
||||
|
||||
private void discard(ChannelHandlerContext ctx, boolean fireNow) {
|
||||
ClosedChannelException cause = null;
|
||||
boolean fireExceptionCaught = false;
|
||||
|
||||
for (;;) {
|
||||
MessageEvent currentEvent = this.currentEvent;
|
||||
@ -168,13 +168,12 @@ public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDowns
|
||||
cause = new ClosedChannelException();
|
||||
}
|
||||
currentEvent.getFuture().setFailure(cause);
|
||||
fireExceptionCaught = true;
|
||||
|
||||
currentEvent = null;
|
||||
}
|
||||
|
||||
|
||||
if (fireExceptionCaught) {
|
||||
if (cause != null) {
|
||||
if (fireNow) {
|
||||
Channels.fireExceptionCaught(ctx.getChannel(), cause);
|
||||
} else {
|
||||
@ -183,105 +182,120 @@ public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDowns
|
||||
}
|
||||
}
|
||||
|
||||
private synchronized void flush(ChannelHandlerContext ctx, boolean fireNow) throws Exception {
|
||||
private void flush(ChannelHandlerContext ctx, boolean fireNow) throws Exception {
|
||||
boolean acquired = false;
|
||||
final Channel channel = ctx.getChannel();
|
||||
if (!channel.isConnected()) {
|
||||
discard(ctx, fireNow);
|
||||
}
|
||||
|
||||
while (channel.isWritable()) {
|
||||
if (currentEvent == null) {
|
||||
currentEvent = queue.poll();
|
||||
}
|
||||
|
||||
if (currentEvent == null) {
|
||||
break;
|
||||
}
|
||||
|
||||
if (currentEvent.getFuture().isDone()) {
|
||||
// Skip the current request because the previous partial write
|
||||
// attempt for the current request has been failed.
|
||||
currentEvent = null;
|
||||
} else {
|
||||
final MessageEvent currentEvent = this.currentEvent;
|
||||
Object m = currentEvent.getMessage();
|
||||
if (m instanceof ChunkedInput) {
|
||||
final ChunkedInput chunks = (ChunkedInput) m;
|
||||
Object chunk;
|
||||
boolean endOfInput;
|
||||
boolean suspend;
|
||||
try {
|
||||
chunk = chunks.nextChunk();
|
||||
endOfInput = chunks.isEndOfInput();
|
||||
if (chunk == null) {
|
||||
chunk = ChannelBuffers.EMPTY_BUFFER;
|
||||
// No need to suspend when reached at the end.
|
||||
suspend = !endOfInput;
|
||||
} else {
|
||||
suspend = false;
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
this.currentEvent = null;
|
||||
|
||||
currentEvent.getFuture().setFailure(t);
|
||||
if (fireNow) {
|
||||
fireExceptionCaught(ctx, t);
|
||||
} else {
|
||||
fireExceptionCaughtLater(ctx, t);
|
||||
}
|
||||
|
||||
closeInput(chunks);
|
||||
break;
|
||||
}
|
||||
|
||||
if (suspend) {
|
||||
// ChunkedInput.nextChunk() returned null and it has
|
||||
// not reached at the end of input. Let's wait until
|
||||
// more chunks arrive. Nothing to write or notify.
|
||||
break;
|
||||
} else {
|
||||
ChannelFuture writeFuture;
|
||||
if (endOfInput) {
|
||||
this.currentEvent = null;
|
||||
writeFuture = currentEvent.getFuture();
|
||||
|
||||
// Register a listener which will close the input once the write is complete. This is needed because the Chunk may have
|
||||
// some resource bound that can not be closed before its not written
|
||||
//
|
||||
// See https://github.com/netty/netty/issues/303
|
||||
writeFuture.addListener(new ChannelFutureListener() {
|
||||
|
||||
public void operationComplete(ChannelFuture future) throws Exception {
|
||||
closeInput(chunks);
|
||||
}
|
||||
});
|
||||
} else {
|
||||
writeFuture = future(channel);
|
||||
writeFuture.addListener(new ChannelFutureListener() {
|
||||
public void operationComplete(ChannelFuture future)
|
||||
throws Exception {
|
||||
if (!future.isSuccess()) {
|
||||
currentEvent.getFuture().setFailure(future.getCause());
|
||||
closeInput((ChunkedInput) currentEvent.getMessage());
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
Channels.write(
|
||||
ctx, writeFuture, chunk,
|
||||
currentEvent.getRemoteAddress());
|
||||
}
|
||||
} else {
|
||||
this.currentEvent = null;
|
||||
ctx.sendDownstream(currentEvent);
|
||||
// use CAS to see if the have flush already running, if so we don't need to take futher actions
|
||||
if (acquired = flush.compareAndSet(false, true)) {
|
||||
try {
|
||||
|
||||
if (!channel.isConnected()) {
|
||||
discard(ctx, fireNow);
|
||||
}
|
||||
}
|
||||
|
||||
if (!channel.isConnected()) {
|
||||
discard(ctx, fireNow);
|
||||
break;
|
||||
while (channel.isWritable()) {
|
||||
if (currentEvent == null) {
|
||||
currentEvent = queue.poll();
|
||||
}
|
||||
|
||||
if (currentEvent == null) {
|
||||
break;
|
||||
}
|
||||
|
||||
if (currentEvent.getFuture().isDone()) {
|
||||
// Skip the current request because the previous partial write
|
||||
// attempt for the current request has been failed.
|
||||
currentEvent = null;
|
||||
} else {
|
||||
final MessageEvent currentEvent = this.currentEvent;
|
||||
Object m = currentEvent.getMessage();
|
||||
if (m instanceof ChunkedInput) {
|
||||
final ChunkedInput chunks = (ChunkedInput) m;
|
||||
Object chunk;
|
||||
boolean endOfInput;
|
||||
boolean suspend;
|
||||
try {
|
||||
chunk = chunks.nextChunk();
|
||||
endOfInput = chunks.isEndOfInput();
|
||||
if (chunk == null) {
|
||||
chunk = ChannelBuffers.EMPTY_BUFFER;
|
||||
// No need to suspend when reached at the end.
|
||||
suspend = !endOfInput;
|
||||
} else {
|
||||
suspend = false;
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
this.currentEvent = null;
|
||||
|
||||
currentEvent.getFuture().setFailure(t);
|
||||
if (fireNow) {
|
||||
fireExceptionCaught(ctx, t);
|
||||
} else {
|
||||
fireExceptionCaughtLater(ctx, t);
|
||||
}
|
||||
|
||||
closeInput(chunks);
|
||||
break;
|
||||
}
|
||||
|
||||
if (suspend) {
|
||||
// ChunkedInput.nextChunk() returned null and it has
|
||||
// not reached at the end of input. Let's wait until
|
||||
// more chunks arrive. Nothing to write or notify.
|
||||
break;
|
||||
} else {
|
||||
ChannelFuture writeFuture;
|
||||
if (endOfInput) {
|
||||
this.currentEvent = null;
|
||||
writeFuture = currentEvent.getFuture();
|
||||
|
||||
// Register a listener which will close the input once the write is complete. This is needed because the Chunk may have
|
||||
// some resource bound that can not be closed before its not written
|
||||
//
|
||||
// See https://github.com/netty/netty/issues/303
|
||||
writeFuture.addListener(new ChannelFutureListener() {
|
||||
|
||||
public void operationComplete(ChannelFuture future) throws Exception {
|
||||
closeInput(chunks);
|
||||
}
|
||||
});
|
||||
} else {
|
||||
writeFuture = future(channel);
|
||||
writeFuture.addListener(new ChannelFutureListener() {
|
||||
public void operationComplete(ChannelFuture future) throws Exception {
|
||||
if (!future.isSuccess()) {
|
||||
currentEvent.getFuture().setFailure(future.getCause());
|
||||
closeInput((ChunkedInput) currentEvent.getMessage());
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
Channels.write(
|
||||
ctx, writeFuture, chunk,
|
||||
currentEvent.getRemoteAddress());
|
||||
}
|
||||
} else {
|
||||
this.currentEvent = null;
|
||||
ctx.sendDownstream(currentEvent);
|
||||
}
|
||||
}
|
||||
|
||||
if (!channel.isConnected()) {
|
||||
discard(ctx, fireNow);
|
||||
break;
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
// mark the flush as done
|
||||
flush.set(false);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
if (acquired && !channel.isConnected() || (channel.isWritable() && !queue.isEmpty())) {
|
||||
flush(ctx, fireNow);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user