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

Do not defer closing of Channel when in flush

Browse Source 
Motivation:

Previously, we deferred the closing of the Channel when we were flushing. This is problematic as this means that if the user adds a ChannelFutureListener, that will close the Channel, the closing will not happen until we are done with flushing. This can lead to more data is sent than expected.

Modifications:

- Do not defer closing when in flush

Result:

Correctly respect order of events and closing the Channel ASAP
This commit is contained in:
Norman Maurer 2015-05-06 12:35:53 +02:00
parent caac01b0f5
commit 0f8cf690cb
1 changed files with 64 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

127
transport/src/main/java/io/netty/channel/AbstractChannel.java
View File

@ -604,24 +604,17 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
return;
}
if (inFlush0) {
invokeLater(new OneTimeTask() {
@Override
public void run() {
close(promise);
}
});
return;
}
if (outboundBuffer == null) {
// This means close() was called before so we just register a listener and return
closeFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
promise.setSuccess();
}
});
// Only needed if no VoidChannelPromise.
if (!(promise instanceof VoidChannelPromise)) {
// This means close() was called before so we just register a listener and return
closeFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
promise.setSuccess();
}
});
}
return;
}
@ -639,64 +632,72 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
closeExecutor.execute(new OneTimeTask() {
@Override
public void run() {
Throwable cause = null;
try {
doClose();
} catch (Throwable t) {
cause = t;
// Execute the close.
doClose0(promise);
} finally {
// Call invokeLater so closeAndDeregister is executed in the EventLoop again!
invokeLater(new OneTimeTask() {
@Override
public void run() {
// Fail all the queued messages
buffer.failFlushed(CLOSED_CHANNEL_EXCEPTION);
buffer.close(CLOSED_CHANNEL_EXCEPTION);
fireChannelInactiveAndDeregister(wasActive);
}
});
}
final Throwable error = cause;
// Call invokeLater so closeAndDeregister is executed in the EventLoop again!
invokeLater(new OneTimeTask() {
@Override
public void run() {
closeAndDeregister(buffer, wasActive, promise, error);
}
});
}
});
} else {
Throwable error = null;
try {
doClose();
} catch (Throwable t) {
error = t;
// Close the channel and fail the queued messages in all cases.
doClose0(promise);
} finally {
// Fail all the queued messages.
buffer.failFlushed(CLOSED_CHANNEL_EXCEPTION);
buffer.close(CLOSED_CHANNEL_EXCEPTION);
}
if (inFlush0) {
invokeLater(new OneTimeTask() {
@Override
public void run() {
fireChannelInactiveAndDeregister(wasActive);
}
});
} else {
fireChannelInactiveAndDeregister(wasActive);
}
closeAndDeregister(buffer, wasActive, promise, error);
}
}
private void closeAndDeregister(ChannelOutboundBuffer outboundBuffer, final boolean wasActive,
ChannelPromise promise, Throwable error) {
// Fail all the queued messages
private void doClose0(ChannelPromise promise) {
try {
outboundBuffer.failFlushed(CLOSED_CHANNEL_EXCEPTION);
outboundBuffer.close(CLOSED_CHANNEL_EXCEPTION);
} finally {
if (wasActive && !isActive()) {
invokeLater(new OneTimeTask() {
@Override
public void run() {
pipeline.fireChannelInactive();
deregister(voidPromise());
}
});
} else {
invokeLater(new OneTimeTask() {
@Override
public void run() {
deregister(voidPromise());
}
});
}
// Now complete the closeFuture and promise.
doClose();
closeFuture.setClosed();
if (error != null) {
safeSetFailure(promise, error);
} else {
safeSetSuccess(promise);
}
safeSetSuccess(promise);
} catch (Throwable t) {
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
private void fireChannelInactiveAndDeregister(final boolean wasActive) {
if (wasActive && !isActive()) {
invokeLater(new OneTimeTask() {
@Override
public void run() {
pipeline.fireChannelInactive();
deregister(voidPromise());
}
});
} else {
invokeLater(new OneTimeTask() {
@Override
public void run() {
deregister(voidPromise());
}
});
}
}