Compare commits

...

1 Commits

Author SHA1 Message Date
Norman Maurer
f23e901c99 [#5028] Fix re-entrance issue with channelWritabilityChanged(...) and write(...)
Motivation:

When always triggered fireChannelWritabilityChanged() directly when the update the pending bytes in the ChannelOutboundBuffer was made from within the EventLoop. This is problematic as this can cause some re-entrance issue if the user has a custom ChannelOutboundHandler that does multiple writes from within the write(...) method and also has a handler that will intercept the channelWritabilityChanged event and trigger another write when the Channel is writable. This can also easily happen if the user just use a MessageToMessageEncoder subclass and triggers a write from channelWritabilityChanged().

Modification:

- Always invoke the fireChannelWritabilityChanged() later on the EventLoop.
- Only trigger the fireChannelWritabilityChanged() if the channel is still active.
- when write(...) is called from outside the EventLoop we only increase the pending bytes in the outbound buffer (so that Channel.isWritable() is updated directly) but not cause a fireChannelWritabilityChanged(). The fireChannelWritabilityChanged() is then triggered once the task is picked up by the EventLoop as usual.

Result:

No more re-entrance possible because of writes from within channelWritabilityChanged(...) method.
2016-12-05 10:24:19 +01:00
2 changed files with 47 additions and 37 deletions

View File

@ -96,7 +96,7 @@ public final class ChannelOutboundBuffer {
@SuppressWarnings("UnusedDeclaration")
private volatile int unwritable;
private volatile Runnable fireChannelWritabilityChangedTask;
private final Runnable fireChannelWritabilityChangedTask;
static {
AtomicIntegerFieldUpdater<ChannelOutboundBuffer> unwritableUpdater =
@ -116,6 +116,7 @@ public final class ChannelOutboundBuffer {
ChannelOutboundBuffer(AbstractChannel channel) {
this.channel = channel;
fireChannelWritabilityChangedTask = new ChannelWritabilityChangedTask(channel);
}
/**
@ -138,7 +139,7 @@ public final class ChannelOutboundBuffer {
// increment pending bytes after adding message to the unflushed arrays.
// See https://github.com/netty/netty/issues/1619
incrementPendingOutboundBytes(entry.pendingSize, false);
incrementPendingOutboundBytes(entry.pendingSize);
}
/**
@ -161,7 +162,7 @@ public final class ChannelOutboundBuffer {
if (!entry.promise.setUncancellable()) {
// Was cancelled so make sure we free up memory and notify about the freed bytes
int pending = entry.cancel();
decrementPendingOutboundBytes(pending, false, true);
decrementPendingOutboundBytes(pending);
}
entry = entry.next;
} while (entry != null);
@ -176,17 +177,13 @@ public final class ChannelOutboundBuffer {
* This method is thread-safe!
*/
void incrementPendingOutboundBytes(long size) {
incrementPendingOutboundBytes(size, true);
}
private void incrementPendingOutboundBytes(long size, boolean invokeLater) {
if (size == 0) {
return;
}
long newWriteBufferSize = TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, size);
if (newWriteBufferSize > channel.config().getWriteBufferHighWaterMark()) {
setUnwritable(invokeLater);
if (newWriteBufferSize >= channel.config().getWriteBufferHighWaterMark()) {
setUnwritable();
}
}
@ -195,17 +192,17 @@ public final class ChannelOutboundBuffer {
* This method is thread-safe!
*/
void decrementPendingOutboundBytes(long size) {
decrementPendingOutboundBytes(size, true, true);
decrementPendingOutboundBytes(size, true);
}
private void decrementPendingOutboundBytes(long size, boolean invokeLater, boolean notifyWritability) {
private void decrementPendingOutboundBytes(long size, boolean notifyWritability) {
if (size == 0) {
return;
}
long newWriteBufferSize = TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, -size);
if (notifyWritability && newWriteBufferSize < channel.config().getWriteBufferLowWaterMark()) {
setWritable(invokeLater);
setWritable(notifyWritability);
}
}
@ -270,7 +267,7 @@ public final class ChannelOutboundBuffer {
// only release message, notify and decrement if it was not canceled before.
ReferenceCountUtil.safeRelease(msg);
safeSuccess(promise);
decrementPendingOutboundBytes(size, false, true);
decrementPendingOutboundBytes(size);
}
// recycle the entry
@ -306,7 +303,7 @@ public final class ChannelOutboundBuffer {
ReferenceCountUtil.safeRelease(msg);
safeFail(promise, cause);
decrementPendingOutboundBytes(size, false, notifyWritability);
decrementPendingOutboundBytes(size, notifyWritability);
}
// recycle the entry
@ -528,7 +525,7 @@ public final class ChannelOutboundBuffer {
final int newValue = oldValue & mask;
if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
if (oldValue != 0 && newValue == 0) {
fireChannelWritabilityChanged(true);
fireChannelWritabilityChanged();
}
break;
}
@ -542,7 +539,7 @@ public final class ChannelOutboundBuffer {
final int newValue = oldValue | mask;
if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
if (oldValue == 0 && newValue != 0) {
fireChannelWritabilityChanged(true);
fireChannelWritabilityChanged();
}
break;
}
@ -556,48 +553,36 @@ public final class ChannelOutboundBuffer {
return 1 << index;
}
private void setWritable(boolean invokeLater) {
private void setWritable(boolean notify) {
for (;;) {
final int oldValue = unwritable;
final int newValue = oldValue & ~1;
if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
if (oldValue != 0 && newValue == 0) {
fireChannelWritabilityChanged(invokeLater);
if (notify && oldValue != 0 && newValue == 0) {
fireChannelWritabilityChanged();
}
break;
}
}
}
private void setUnwritable(boolean invokeLater) {
private void setUnwritable() {
for (;;) {
final int oldValue = unwritable;
final int newValue = oldValue | 1;
if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
if (oldValue == 0 && newValue != 0) {
fireChannelWritabilityChanged(invokeLater);
fireChannelWritabilityChanged();
}
break;
}
}
}
private void fireChannelWritabilityChanged(boolean invokeLater) {
final ChannelPipeline pipeline = channel.pipeline();
if (invokeLater) {
Runnable task = fireChannelWritabilityChangedTask;
if (task == null) {
fireChannelWritabilityChangedTask = task = new Runnable() {
@Override
public void run() {
pipeline.fireChannelWritabilityChanged();
}
};
}
channel.eventLoop().execute(task);
} else {
pipeline.fireChannelWritabilityChanged();
}
private void fireChannelWritabilityChanged() {
// Always invoke it later to prevent re-entrance bug.
// See https://github.com/netty/netty/issues/5028
channel.eventLoop().execute(fireChannelWritabilityChangedTask);
}
/**
@ -838,4 +823,20 @@ public final class ChannelOutboundBuffer {
return next;
}
}
private static final class ChannelWritabilityChangedTask implements Runnable {
private final Channel channel;
ChannelWritabilityChangedTask(Channel channel) {
this.channel = channel;
}
@Override
public void run() {
// Only need to fire if the Channel is still active otherwise it may be confusing to receive such an event.
if (channel.isActive()) {
channel.pipeline().fireChannelWritabilityChanged();
}
}
}
}

View File

@ -230,6 +230,9 @@ public class ChannelOutboundBufferTest {
// Ensure exceeding the high watermark makes channel unwritable.
ch.write(buffer().writeZero(127));
assertThat(buf.toString(), is(""));
ch.runPendingTasks();
assertThat(buf.toString(), is("false "));
// Ensure going down to the low watermark makes channel writable again by flushing the first write.
@ -237,6 +240,9 @@ public class ChannelOutboundBufferTest {
assertThat(ch.unsafe().outboundBuffer().remove(), is(true));
assertThat(ch.unsafe().outboundBuffer().totalPendingWriteBytes(),
is(127L + ChannelOutboundBuffer.CHANNEL_OUTBOUND_BUFFER_ENTRY_OVERHEAD));
assertThat(buf.toString(), is("false "));
ch.runPendingTasks();
assertThat(buf.toString(), is("false true "));
safeClose(ch);
@ -334,6 +340,9 @@ public class ChannelOutboundBufferTest {
// Trigger channelWritabilityChanged() by writing a lot.
ch.write(buffer().writeZero(257));
assertThat(buf.toString(), is(""));
ch.runPendingTasks();
assertThat(buf.toString(), is("false "));
// Ensure that setting a user-defined writability flag to false does not trigger channelWritabilityChanged()