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

Better handling of canceling. See #210 and #209

Browse Source
This commit is contained in:
Norman Maurer 2012-02-29 21:37:26 +01:00
parent a545157f4b
commit 1589dadcce
4 changed files with 24 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
transport-sctp/src/main/java/io/netty/channel/sctp/SctpWorker.java
View File

@ -74,7 +74,7 @@ class SctpWorker implements Worker {
private final Object startStopLock = new Object();
private final Queue<Runnable> registerTaskQueue = QueueFactory.createQueue(Runnable.class);
private final Queue<Runnable> writeTaskQueue = QueueFactory.createQueue(Runnable.class);
private final Queue<ChannelRunnableWrapper> eventQueue = QueueFactory.createQueue(ChannelRunnableWrapper.class);
private final Queue<Runnable> eventQueue = QueueFactory.createQueue(Runnable.class);
private volatile int cancelledKeys; // should use AtomicInteger but we just need approximation
@ -301,13 +301,11 @@ class SctpWorker implements Worker {
private void processEventQueue() throws IOException {
for (;;) {
final ChannelRunnableWrapper task = eventQueue.poll();
final Runnable task = eventQueue.poll();
if (task == null) {
break;
}
if (!task.isCancelled()) {
task.run();
}
task.run();
cleanUpCancelledKeys();
}
}

18
transport/src/main/java/io/netty/channel/socket/ChannelRunnableWrapper.java
View File

@ -21,7 +21,8 @@ import io.netty.channel.DefaultChannelFuture;
public class ChannelRunnableWrapper extends DefaultChannelFuture implements Runnable {
private final Runnable task;
private boolean started = false;
public ChannelRunnableWrapper(Channel channel, Runnable task) {
super(channel, true);
this.task = task;
@ -29,6 +30,13 @@ public class ChannelRunnableWrapper extends DefaultChannelFuture implements Runn
@Override
public void run() {
synchronized(this) {
if (!isCancelled()) {
started = true;
} else {
return;
}
}
try {
task.run();
setSuccess();
@ -36,6 +44,14 @@ public class ChannelRunnableWrapper extends DefaultChannelFuture implements Runn
setFailure(t);
}
}
@Override
public synchronized boolean cancel() {
if (started) {
return false;
}
return super.cancel();
}

8
transport/src/main/java/io/netty/channel/socket/nio/AbstractNioWorker.java
View File

@ -109,7 +109,7 @@ abstract class AbstractNioWorker implements Worker {
*/
protected final Queue<Runnable> writeTaskQueue = QueueFactory.createQueue(Runnable.class);
private final Queue<ChannelRunnableWrapper> eventQueue = QueueFactory.createQueue(ChannelRunnableWrapper.class);
private final Queue<Runnable> eventQueue = QueueFactory.createQueue(Runnable.class);
private volatile int cancelledKeys; // should use AtomicInteger but we just need approximation
@ -324,13 +324,11 @@ abstract class AbstractNioWorker implements Worker {
private void processEventQueue() throws IOException {
for (;;) {
final ChannelRunnableWrapper task = eventQueue.poll();
final Runnable task = eventQueue.poll();
if (task == null) {
break;
}
if (!task.isCancelled()) {
task.run();
}
task.run();
cleanUpCancelledKeys();
}
}

4
transport/src/main/java/io/netty/channel/socket/oio/AbstractOioWorker.java
View File

@ -119,9 +119,7 @@ abstract class AbstractOioWorker<C extends AbstractOioChannel> implements Worker
if (task == null) {
break;
}
if (!task.isCancelled()) {
task.run();
}
task.run();
}
}