oio and nio transport now make sure that a upstream event get only

executed from an io thread. See #140 and #187
This commit is contained in:
Norman Maurer 2012-02-25 15:12:58 +01:00
parent c4a437e16b
commit ef64e8c332
6 changed files with 164 additions and 34 deletions

View File

@ -298,6 +298,21 @@ public final class Channels {
ctx.getChannel(), message, remoteAddress)); ctx.getChannel(), message, remoteAddress));
} }
/**
* Sends a {@code "writeComplete"} event to the first
* {@link ChannelUpstreamHandler} in the {@link ChannelPipeline} of
* the specified {@link Channel} in the next io-thread.
*/
public static void fireWriteCompleteLater(Channel channel, long amount) {
if (amount == 0) {
return;
}
channel.getPipeline().sendUpstreamLater(
new DefaultWriteCompletionEvent(channel, amount));
}
/** /**
* Sends a {@code "writeComplete"} event to the first * Sends a {@code "writeComplete"} event to the first
* {@link ChannelUpstreamHandler} in the {@link ChannelPipeline} of * {@link ChannelUpstreamHandler} in the {@link ChannelPipeline} of

View File

@ -398,6 +398,7 @@ abstract class AbstractNioWorker implements Worker {
boolean open = true; boolean open = true;
boolean addOpWrite = false; boolean addOpWrite = false;
boolean removeOpWrite = false; boolean removeOpWrite = false;
boolean iothread = isIoThread(channel);
long writtenBytes = 0; long writtenBytes = 0;
@ -468,7 +469,11 @@ abstract class AbstractNioWorker implements Worker {
buf = null; buf = null;
evt = null; evt = null;
future.setFailure(t); future.setFailure(t);
fireExceptionCaught(channel, t); if (iothread) {
fireExceptionCaught(channel, t);
} else {
fireExceptionCaughtLater(channel, t);
}
if (t instanceof IOException) { if (t instanceof IOException) {
open = false; open = false;
close(channel, succeededFuture(channel)); close(channel, succeededFuture(channel));
@ -491,10 +496,17 @@ abstract class AbstractNioWorker implements Worker {
} }
} }
} }
if (iothread) {
fireWriteComplete(channel, writtenBytes); fireWriteComplete(channel, writtenBytes);
} else {
fireWriteCompleteLater(channel, writtenBytes);
}
} }
static boolean isIoThread(AbstractNioChannel<?> channel) {
return Thread.currentThread() == channel.worker.thread;
}
private void setOpWrite(AbstractNioChannel<?> channel) { private void setOpWrite(AbstractNioChannel<?> channel) {
Selector selector = this.selector; Selector selector = this.selector;
SelectionKey key = channel.channel.keyFor(selector); SelectionKey key = channel.channel.keyFor(selector);
@ -545,6 +557,8 @@ abstract class AbstractNioWorker implements Worker {
void close(AbstractNioChannel<?> channel, ChannelFuture future) { void close(AbstractNioChannel<?> channel, ChannelFuture future) {
boolean connected = channel.isConnected(); boolean connected = channel.isConnected();
boolean bound = channel.isBound(); boolean bound = channel.isBound();
boolean iothread = isIoThread(channel);
try { try {
channel.channel.close(); channel.channel.close();
cancelledKeys ++; cancelledKeys ++;
@ -552,20 +566,36 @@ abstract class AbstractNioWorker implements Worker {
if (channel.setClosed()) { if (channel.setClosed()) {
future.setSuccess(); future.setSuccess();
if (connected) { if (connected) {
fireChannelDisconnected(channel); if (iothread) {
fireChannelDisconnected(channel);
} else {
fireChannelDisconnectedLater(channel);
}
} }
if (bound) { if (bound) {
fireChannelUnbound(channel); if (iothread) {
fireChannelUnbound(channel);
} else {
fireChannelUnboundLater(channel);
}
} }
cleanUpWriteBuffer(channel); cleanUpWriteBuffer(channel);
fireChannelClosed(channel); if (iothread) {
fireChannelClosed(channel);
} else {
fireChannelClosedLater(channel);
}
} else { } else {
future.setSuccess(); future.setSuccess();
} }
} catch (Throwable t) { } catch (Throwable t) {
future.setFailure(t); future.setFailure(t);
fireExceptionCaught(channel, t); if (iothread) {
fireExceptionCaught(channel, t);
} else {
fireExceptionCaughtLater(channel, t);
}
} }
} }
@ -618,12 +648,17 @@ abstract class AbstractNioWorker implements Worker {
} }
if (fireExceptionCaught) { if (fireExceptionCaught) {
fireExceptionCaught(channel, cause); if (isIoThread(channel)) {
fireExceptionCaught(channel, cause);
} else {
fireExceptionCaughtLater(channel, cause);
}
} }
} }
void setInterestOps(AbstractNioChannel<?> channel, ChannelFuture future, int interestOps) { void setInterestOps(AbstractNioChannel<?> channel, ChannelFuture future, int interestOps) {
boolean changed = false; boolean changed = false;
boolean iothread = isIoThread(channel);
try { try {
// interestOps can change at any time and at any thread. // interestOps can change at any time and at any thread.
// Acquire a lock to avoid possible race condition. // Acquire a lock to avoid possible race condition.
@ -684,16 +719,28 @@ abstract class AbstractNioWorker implements Worker {
future.setSuccess(); future.setSuccess();
if (changed) { if (changed) {
fireChannelInterestChanged(channel); if (iothread) {
fireChannelInterestChanged(channel);
} else {
fireChannelInterestChangedLater(channel);
}
} }
} catch (CancelledKeyException e) { } catch (CancelledKeyException e) {
// setInterestOps() was called on a closed channel. // setInterestOps() was called on a closed channel.
ClosedChannelException cce = new ClosedChannelException(); ClosedChannelException cce = new ClosedChannelException();
future.setFailure(cce); future.setFailure(cce);
fireExceptionCaught(channel, cce); if (iothread) {
fireExceptionCaught(channel, cce);
} else {
fireExceptionCaughtLater(channel, cce);
}
} catch (Throwable t) { } catch (Throwable t) {
future.setFailure(t); future.setFailure(t);
fireExceptionCaught(channel, t); if (iothread) {
fireExceptionCaught(channel, t);
} else {
fireExceptionCaughtLater(channel, t);
}
} }
} }

View File

@ -16,7 +16,9 @@
package io.netty.channel.socket.nio; package io.netty.channel.socket.nio;
import static io.netty.channel.Channels.fireChannelDisconnected; import static io.netty.channel.Channels.fireChannelDisconnected;
import static io.netty.channel.Channels.fireChannelDisconnectedLater;
import static io.netty.channel.Channels.fireExceptionCaught; import static io.netty.channel.Channels.fireExceptionCaught;
import static io.netty.channel.Channels.fireExceptionCaughtLater;
import static io.netty.channel.Channels.fireMessageReceived; import static io.netty.channel.Channels.fireMessageReceived;
import static io.netty.channel.Channels.succeededFuture; import static io.netty.channel.Channels.succeededFuture;
import io.netty.buffer.ChannelBufferFactory; import io.netty.buffer.ChannelBufferFactory;
@ -126,15 +128,24 @@ class NioDatagramWorker extends AbstractNioWorker {
static void disconnect(NioDatagramChannel channel, ChannelFuture future) { static void disconnect(NioDatagramChannel channel, ChannelFuture future) {
boolean connected = channel.isConnected(); boolean connected = channel.isConnected();
boolean iothread = isIoThread(channel);
try { try {
channel.getDatagramChannel().disconnect(); channel.getDatagramChannel().disconnect();
future.setSuccess(); future.setSuccess();
if (connected) { if (connected) {
fireChannelDisconnected(channel); if (iothread) {
fireChannelDisconnected(channel);
} else {
fireChannelDisconnectedLater(channel);
}
} }
} catch (Throwable t) { } catch (Throwable t) {
future.setFailure(t); future.setFailure(t);
fireExceptionCaught(channel, t); if (iothread) {
fireExceptionCaught(channel, t);
} else {
fireExceptionCaughtLater(channel, t);
}
} }
} }

View File

@ -15,12 +15,7 @@
*/ */
package io.netty.channel.socket.oio; package io.netty.channel.socket.oio;
import static io.netty.channel.Channels.fireChannelClosed; import static io.netty.channel.Channels.*;
import static io.netty.channel.Channels.fireChannelDisconnected;
import static io.netty.channel.Channels.fireChannelInterestChanged;
import static io.netty.channel.Channels.fireChannelUnbound;
import static io.netty.channel.Channels.fireExceptionCaught;
import static io.netty.channel.Channels.succeededFuture;
import io.netty.channel.Channel; import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
import io.netty.channel.Channels; import io.netty.channel.Channels;
@ -89,6 +84,9 @@ abstract class AbstractOioWorker<C extends AbstractOioChannel> implements Worker
close(channel, succeededFuture(channel)); close(channel, succeededFuture(channel));
} }
static boolean isIoThead(AbstractOioChannel channel) {
return Thread.currentThread() == channel.workerThread;
}
@Override @Override
public void executeInIoThread(Runnable eventRunnable) { public void executeInIoThread(Runnable eventRunnable) {
@ -120,7 +118,8 @@ abstract class AbstractOioWorker<C extends AbstractOioChannel> implements Worker
static void setInterestOps( static void setInterestOps(
AbstractOioChannel channel, ChannelFuture future, int interestOps) { AbstractOioChannel channel, ChannelFuture future, int interestOps) {
boolean iothread = isIoThead(channel);
// Override OP_WRITE flag - a user cannot change this flag. // Override OP_WRITE flag - a user cannot change this flag.
interestOps &= ~Channel.OP_WRITE; interestOps &= ~Channel.OP_WRITE;
interestOps |= channel.getInterestOps() & Channel.OP_WRITE; interestOps |= channel.getInterestOps() & Channel.OP_WRITE;
@ -148,18 +147,27 @@ abstract class AbstractOioWorker<C extends AbstractOioChannel> implements Worker
workerThread.interrupt(); workerThread.interrupt();
} }
} }
if (iothread) {
fireChannelInterestChanged(channel); fireChannelInterestChanged(channel);
} else {
fireChannelInterestChangedLater(channel);
}
} }
} catch (Throwable t) { } catch (Throwable t) {
future.setFailure(t); future.setFailure(t);
fireExceptionCaught(channel, t); if (iothread) {
fireExceptionCaught(channel, t);
} else {
fireExceptionCaughtLater(channel, t);
}
} }
} }
static void close(AbstractOioChannel channel, ChannelFuture future) { static void close(AbstractOioChannel channel, ChannelFuture future) {
boolean connected = channel.isConnected(); boolean connected = channel.isConnected();
boolean bound = channel.isBound(); boolean bound = channel.isBound();
boolean iothread = isIoThead(channel);
try { try {
channel.closeSocket(); channel.closeSocket();
if (channel.setClosed()) { if (channel.setClosed()) {
@ -171,18 +179,34 @@ abstract class AbstractOioWorker<C extends AbstractOioChannel> implements Worker
if (workerThread != null && currentThread != workerThread) { if (workerThread != null && currentThread != workerThread) {
workerThread.interrupt(); workerThread.interrupt();
} }
fireChannelDisconnected(channel); if (iothread) {
fireChannelDisconnected(channel);
} else {
fireChannelDisconnectedLater(channel);
}
} }
if (bound) { if (bound) {
fireChannelUnbound(channel); if (iothread) {
fireChannelUnbound(channel);
} else {
fireChannelUnboundLater(channel);
}
}
if (iothread) {
fireChannelClosed(channel);
} else {
fireChannelClosedLater(channel);
} }
fireChannelClosed(channel);
} else { } else {
future.setSuccess(); future.setSuccess();
} }
} catch (Throwable t) { } catch (Throwable t) {
future.setFailure(t); future.setFailure(t);
fireExceptionCaught(channel, t); if (iothread) {
fireExceptionCaught(channel, t);
} else {
fireExceptionCaughtLater(channel, t);
}
} }
} }
} }

View File

@ -63,6 +63,8 @@ class OioDatagramWorker extends AbstractOioWorker<OioDatagramChannel> {
static void write( static void write(
OioDatagramChannel channel, ChannelFuture future, OioDatagramChannel channel, ChannelFuture future,
Object message, SocketAddress remoteAddress) { Object message, SocketAddress remoteAddress) {
boolean iothread = isIoThead(channel);
try { try {
ChannelBuffer buf = (ChannelBuffer) message; ChannelBuffer buf = (ChannelBuffer) message;
int offset = buf.readerIndex(); int offset = buf.readerIndex();
@ -84,27 +86,45 @@ class OioDatagramWorker extends AbstractOioWorker<OioDatagramChannel> {
packet.setSocketAddress(remoteAddress); packet.setSocketAddress(remoteAddress);
} }
channel.socket.send(packet); channel.socket.send(packet);
fireWriteComplete(channel, length); if (iothread) {
fireWriteComplete(channel, length);
} else {
fireWriteCompleteLater(channel, length);
}
future.setSuccess(); future.setSuccess();
} catch (Throwable t) { } catch (Throwable t) {
future.setFailure(t); future.setFailure(t);
fireExceptionCaught(channel, t); if (iothread) {
fireExceptionCaught(channel, t);
} else {
fireExceptionCaughtLater(channel, t);
}
} }
} }
static void disconnect(OioDatagramChannel channel, ChannelFuture future) { static void disconnect(OioDatagramChannel channel, ChannelFuture future) {
boolean connected = channel.isConnected(); boolean connected = channel.isConnected();
boolean iothread = isIoThead(channel);
try { try {
channel.socket.disconnect(); channel.socket.disconnect();
future.setSuccess(); future.setSuccess();
if (connected) { if (connected) {
// Notify. // Notify.
fireChannelDisconnected(channel); if (iothread) {
fireChannelDisconnected(channel);
} else {
fireChannelDisconnectedLater(channel);
}
} }
} catch (Throwable t) { } catch (Throwable t) {
future.setFailure(t); future.setFailure(t);
fireExceptionCaught(channel, t); if (iothread) {
fireExceptionCaught(channel, t);
} else {
fireExceptionCaughtLater(channel, t);
}
} }
} }

View File

@ -65,11 +65,16 @@ class OioWorker extends AbstractOioWorker<OioSocketChannel> {
OioSocketChannel channel, ChannelFuture future, OioSocketChannel channel, ChannelFuture future,
Object message) { Object message) {
boolean iothread = isIoThead(channel);
OutputStream out = channel.getOutputStream(); OutputStream out = channel.getOutputStream();
if (out == null) { if (out == null) {
Exception e = new ClosedChannelException(); Exception e = new ClosedChannelException();
future.setFailure(e); future.setFailure(e);
fireExceptionCaught(channel, e); if (iothread) {
fireExceptionCaught(channel, e);
} else {
fireExceptionCaughtLater(channel, e);
}
return; return;
} }
@ -106,7 +111,11 @@ class OioWorker extends AbstractOioWorker<OioSocketChannel> {
} }
} }
fireWriteComplete(channel, length); if (iothread) {
fireWriteComplete(channel, length);
} else {
fireWriteCompleteLater(channel, length);
}
future.setSuccess(); future.setSuccess();
} catch (Throwable t) { } catch (Throwable t) {
@ -118,7 +127,11 @@ class OioWorker extends AbstractOioWorker<OioSocketChannel> {
t = new ClosedChannelException(); t = new ClosedChannelException();
} }
future.setFailure(t); future.setFailure(t);
fireExceptionCaught(channel, t); if (iothread) {
fireExceptionCaught(channel, t);
} else {
fireExceptionCaughtLater(channel, t);
}
} }
} }