Skip to content

Commit

Permalink
Move the methods that's only used by DefaultChannelPipeline to Defaul…
Browse files Browse the repository at this point in the history 
…tChannelPipeline
  • Loading branch information
@trustin
trustin committed Mar 14, 2013
1 parent d55567e commit 9c96791
Show file tree
Hide file tree
Showing 2 changed files with 92 additions and 86 deletions.
109 changes: 35 additions & 74 deletions transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,20 +24,20 @@
import io.netty.util.DefaultAttributeMap;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.internal.PlatformDependent;

import java.net.SocketAddress;
import java.nio.channels.ClosedChannelException;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

import static io.netty.channel.DefaultChannelPipeline.*;

final class DefaultChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext {

private static final int FLAG_REMOVED = 1;
private static final int FLAG_FREED = 2;

volatile DefaultChannelHandlerContext next;
volatile DefaultChannelHandlerContext prev;

Expand All @@ -56,6 +56,8 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
private MessageBuf<Object> outMsgBuf;
private ByteBuf outByteBuf;

private int flags;

// When the two handlers run in a different thread and they are next to each other,
// each other's buffers can be accessed at the same time resulting in a race condition.
// To avoid such situation, we lazily creates an additional thread-safe buffer called
Expand Down Expand Up @@ -93,8 +95,6 @@ final class DefaultChannelHandlerContext extends DefaultAttributeMap implements
private Runnable fireInboundBufferUpdated0Task;
private Runnable invokeChannelReadSuspendedTask;
private Runnable invokeRead0Task;
private boolean freed;
boolean removed;

@SuppressWarnings("unchecked")
DefaultChannelHandlerContext(
Expand Down Expand Up @@ -383,29 +383,38 @@ private boolean flushOutboundBridge() {
return true;
}

void freeHandlerBuffersAfterRemoval() {
if (!removed || freed) {
return;
void setRemoved() {
flags |= FLAG_REMOVED;

// Free all buffers before completing removal.
if (!channel.isRegistered()) {
freeHandlerBuffersAfterRemoval();
}
freed = true;
final ChannelHandler handler = handler();
}

if (handler instanceof ChannelInboundHandler) {
try {
((ChannelInboundHandler) handler).freeInboundBuffer(this);
} catch (Exception e) {
notifyHandlerException(e);
} finally {
freeInboundBridge();
private void freeHandlerBuffersAfterRemoval() {
if (flags == FLAG_REMOVED) { // Removed, but not freed yet
flags |= FLAG_FREED;

final ChannelHandler handler = handler();

if (handler instanceof ChannelInboundHandler) {
try {
((ChannelInboundHandler) handler).freeInboundBuffer(this);
} catch (Exception e) {
notifyHandlerException(e);
} finally {
freeInboundBridge();
}
}
}
if (handler instanceof ChannelOutboundHandler) {
try {
((ChannelOutboundHandler) handler).freeOutboundBuffer(this);
} catch (Exception e) {
notifyHandlerException(e);
} finally {
freeOutboundBridge();
if (handler instanceof ChannelOutboundHandler) {
try {
((ChannelOutboundHandler) handler).freeOutboundBuffer(this);
} catch (Exception e) {
notifyHandlerException(e);
} finally {
freeOutboundBridge();
}
}
}
}
Expand Down Expand Up @@ -530,54 +539,6 @@ public <T> MessageBuf<T> outboundMessageBuffer() {
return (MessageBuf<T>) outMsgBuf;
}

/**
* Executes a task on the event loop and waits for it to finish. If the task is interrupted, then the
* current thread will be interrupted. It is expected that the task performs any appropriate locking.
* <p>
* If the {@link Runnable#run()} call throws a {@link Throwable}, but it is not an instance of
* {@link Error} or {@link RuntimeException}, then it is wrapped inside a
* {@link ChannelPipelineException} and that is thrown instead.</p>
*
* @param r execute this runnable
* @see Runnable#run()
* @see Future#get()
* @throws Error if the task threw this.
* @throws RuntimeException if the task threw this.
* @throws ChannelPipelineException with a {@link Throwable} as a cause, if the task threw another type of
* {@link Throwable}.
*/
void executeOnEventLoop(Runnable r) {
waitForFuture(executor().submit(r));
}

/**
* Waits for a future to finish. If the task is interrupted, then the current thread will be interrupted.
* It is expected that the task performs any appropriate locking.
* <p>
* If the internal call throws a {@link Throwable}, but it is not an instance of {@link Error} or
* {@link RuntimeException}, then it is wrapped inside a {@link ChannelPipelineException} and that is
* thrown instead.</p>
*
* @param future wait for this future
* @see Future#get()
* @throws Error if the task threw this.
* @throws RuntimeException if the task threw this.
* @throws ChannelPipelineException with a {@link Throwable} as a cause, if the task threw another type of
* {@link Throwable}.
*/
static void waitForFuture(Future<?> future) {
try {
future.get();
} catch (ExecutionException ex) {
// In the arbitrary case, we can throw Error, RuntimeException, and Exception
PlatformDependent.throwException(ex.getCause());
} catch (InterruptedException ex) {
// Interrupt the calling thread (note that this method is not called from the event loop)

Thread.currentThread().interrupt();
}
}

@Override
public ByteBuf nextInboundByteBuffer() {
DefaultChannelHandlerContext ctx = next;
Expand Down Expand Up @@ -939,7 +900,7 @@ private void invokeInboundBufferUpdated() {
} catch (Throwable t) {
notifyHandlerException(t);
} finally {
if (!freed) {
if ((flags & FLAG_FREED) == 0) {
if (handler instanceof ChannelInboundByteHandler && !pipeline.isInboundShutdown()) {
try {
((ChannelInboundByteHandler) handler).discardInboundReadBytes(this);
Expand Down
69 changes: 57 additions & 12 deletions transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import io.netty.channel.Channel.Unsafe;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.internal.PlatformDependent;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;

Expand All @@ -36,10 +37,9 @@
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.WeakHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import static io.netty.channel.DefaultChannelHandlerContext.*;

/**
* The default {@link ChannelPipeline} implementation. It is usually created
* by a {@link Channel} implementation when the {@link Channel} is created.
Expand Down Expand Up @@ -128,7 +128,7 @@ public ChannelPipeline addFirst(EventExecutorGroup group, final String name, Cha
// Run the following 'waiting' code outside of the above synchronized block
// in order to avoid deadlock

newCtx.executeOnEventLoop(new Runnable() {
executeOnEventLoop(newCtx, new Runnable() {
@Override
public void run() {
synchronized (DefaultChannelPipeline.this) {
Expand Down Expand Up @@ -178,7 +178,7 @@ public ChannelPipeline addLast(EventExecutorGroup group, final String name, Chan
// Run the following 'waiting' code outside of the above synchronized block
// in order to avoid deadlock

newCtx.executeOnEventLoop(new Runnable() {
executeOnEventLoop(newCtx, new Runnable() {
@Override
public void run() {
synchronized (DefaultChannelPipeline.this) {
Expand Down Expand Up @@ -232,7 +232,7 @@ public ChannelPipeline addBefore(
// Run the following 'waiting' code outside of the above synchronized block
// in order to avoid deadlock

newCtx.executeOnEventLoop(new Runnable() {
executeOnEventLoop(newCtx, new Runnable() {
@Override
public void run() {
synchronized (DefaultChannelPipeline.this) {
Expand Down Expand Up @@ -284,7 +284,7 @@ public ChannelPipeline addAfter(
// Run the following 'waiting' code outside of the above synchronized block
// in order to avoid deadlock

newCtx.executeOnEventLoop(new Runnable() {
executeOnEventLoop(newCtx, new Runnable() {
@Override
public void run() {
synchronized (DefaultChannelPipeline.this) {
Expand Down Expand Up @@ -676,8 +676,10 @@ private static void callBeforeRemove(ChannelHandlerContext ctx) {
}
}

private void callAfterRemove(final DefaultChannelHandlerContext ctx, DefaultChannelHandlerContext ctxPrev,
DefaultChannelHandlerContext ctxNext, boolean forward) {
private static void callAfterRemove(
final DefaultChannelHandlerContext ctx, DefaultChannelHandlerContext ctxPrev,
DefaultChannelHandlerContext ctxNext, boolean forward) {

final ChannelHandler handler = ctx.handler();

// Notify the complete removal.
Expand All @@ -695,11 +697,54 @@ private void callAfterRemove(final DefaultChannelHandlerContext ctx, DefaultChan
ctx.clearBuffer();
}

ctx.removed = true;
ctx.setRemoved();
}

/**
* Executes a task on the event loop and waits for it to finish. If the task is interrupted, then the
* current thread will be interrupted. It is expected that the task performs any appropriate locking.
* <p>
* If the {@link Runnable#run()} call throws a {@link Throwable}, but it is not an instance of
* {@link Error} or {@link RuntimeException}, then it is wrapped inside a
* {@link ChannelPipelineException} and that is thrown instead.</p>
*
* @param r execute this runnable
* @see Runnable#run()
* @see Future#get()
* @throws Error if the task threw this.
* @throws RuntimeException if the task threw this.
* @throws ChannelPipelineException with a {@link Throwable} as a cause, if the task threw another type of
* {@link Throwable}.
*/
private static void executeOnEventLoop(DefaultChannelHandlerContext ctx, Runnable r) {
waitForFuture(ctx.executor().submit(r));
}

// Free all buffers before completing removal.
if (!channel.isRegistered()) {
ctx.freeHandlerBuffersAfterRemoval();
/**
* Waits for a future to finish. If the task is interrupted, then the current thread will be interrupted.
* It is expected that the task performs any appropriate locking.
* <p>
* If the internal call throws a {@link Throwable}, but it is not an instance of {@link Error} or
* {@link RuntimeException}, then it is wrapped inside a {@link ChannelPipelineException} and that is
* thrown instead.</p>
*
* @param future wait for this future
* @see Future#get()
* @throws Error if the task threw this.
* @throws RuntimeException if the task threw this.
* @throws ChannelPipelineException with a {@link Throwable} as a cause, if the task threw another type of
* {@link Throwable}.
*/
private static void waitForFuture(Future<?> future) {
try {
future.get();
} catch (ExecutionException ex) {
// In the arbitrary case, we can throw Error, RuntimeException, and Exception
PlatformDependent.throwException(ex.getCause());
} catch (InterruptedException ex) {
// Interrupt the calling thread (note that this method is not called from the event loop)

Thread.currentThread().interrupt();
}
}

Expand Down

0 comments on commit 9c96791

Please sign in to comment.