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

Implemented consolidate-on-flush in BufferedWriteHandler

Browse Source
This commit is contained in:
Trustin Lee 2010-02-16 08:41:02 +00:00
parent dd7f588916
commit 28fe7c67cb
1 changed files with 194 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

201
src/main/java/org/jboss/netty/handler/queue/BufferedWriteHandler.java
View File

@ -15,15 +15,23 @@
*/
package org.jboss.netty.handler.queue;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelConfig;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelDownstreamHandler;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.jboss.netty.channel.socket.nio.NioSocketChannelConfig;
import org.jboss.netty.util.HashedWheelTimer;
import org.jboss.netty.util.internal.LinkedTransferQueue;
@ -54,13 +62,15 @@ import org.jboss.netty.util.internal.LinkedTransferQueue;
* <h3>Auto-flush</h3>
* The write request queue is automatically flushed when the associated
* {@link Channel} is disconnected or closed. However, it does not flush the
* queue when the size of the queue increases. You can implement your own
* auto-flush strategy by extending this handler:
* queue otherwise. It means you have to call {@link #flush()} before the size
* of the queue increases too much. You can implement your own auto-flush
* strategy by extending this handler:
* <pre>
* public class AutoFlusher extends {@link BufferedWriteHandler} {
*
* private final AtomicLong bufferSize = new AtomicLong();
*
* {@literal @Override}
* public void writeRequested({@link ChannelHandlerContext} ctx, {@link MessageEvent} e) {
* super.writeRequested(ctx, e);
*
@ -76,41 +86,134 @@ import org.jboss.netty.util.internal.LinkedTransferQueue;
* }
* </pre>
*
* <h3>Prioritized Writes</h3>
* <h3>Consolidate on flush</h3>
*
* If there are two or more write requests in the queue and all their message
* type is {@link ChannelBuffer}, they can be merged into a single write request
* to save the number of system calls.
* <pre>
* BEFORE consolidation: AFTER consolidation:
* +-------+-------+-------+ +-------------+
* | Req C | Req B | Req A |------\\| Request ABC |
* | "789" | "456" | "123" |------//| "123456789" |
* +-------+-------+-------+ +-------------+
* </pre>
* This feature is disabled by default. You can override the default when you
* create this handler or call {@link #flush(boolean)}. If you specified
* {@code true} when you call the constructor, calling {@link #flush()} will
* always consolidate the queue. Otherwise, you have to call
* {@link #flush(boolean)} with {@code true} to enable this feature for each
* flush.
* <p>
* The disadvantage of consolidation is that the {@link ChannelFuture} and its
* {@link ChannelFutureListener}s associated with the original write requests
* might be notified later than when they are actually written out. They will
* always be notified when the consolidated write request is fully written.
* <p>
* The following example implements the consolidation strategy that reduces
* the number of write requests based on the writability of a channel:
* <pre>
* public class ConsolidatingAutoFlusher extends {@link BufferedWriteHandler} {
*
* public ConsolidatingAutoFlusher() {
* // Enable consolidation by default.
* super(true);
* }
*
* {@literal @Override}
* public void channelOpen({@link ChannelHandlerContext} ctx, {@link ChannelStateEvent} e) throws Exception {
* {@link ChannelConfig} cfg = e.getChannel().getConfig();
* if (cfg instanceof {@link NioSocketChannelConfig}) {
* // Lower the watermark to increase the chance of consolidation.
* cfg.setWriteBufferLowWaterMark(0);
* }
* super.channelOpen(e);
* }
*
* {@literal @Override}
* public void writeRequested({@link ChannelHandlerContext} ctx, {@link MessageEvent} e) throws Exception {
* super.writeRequested(ctx, et);
* if (e.getChannel().isWritable()) {
* flush();
* }
* }
*
* {@literal @Override}
* public void channelInterestChanged({@link ChannelHandlerContext} ctx, {@link ChannelStateEvent} e) throws Exception {
* if (e.getChannel().isWritable()) {
* flush();
* }
* }
* }
* </pre>
*
* <h3>Prioritized Writes</h3>
*
* You can implement prioritized writes by specifying an unbounded priority
* queue in the constructor of this handler. It will be required to design
* the proper strategy to determine how often {@link #flush()} should be called.
* In most cases, it should be enough to call {@link #flush()} periodically,
* using {@link HashedWheelTimer} for example.
* For example, you could call {@link #flush()} periodically, using
* {@link HashedWheelTimer} every second.
*
* @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
* @author <a href="http://gleamynode.net/">Trustin Lee</a>
* @version $Rev$, $Date$
*/
public class BufferedWriteHandler extends SimpleChannelDownstreamHandler {
public class BufferedWriteHandler extends SimpleChannelHandler {
private final Queue<MessageEvent> queue;
private final boolean consolidateOnFlush;
private volatile ChannelHandlerContext ctx;
/**
* Creates a new instance with the default unbounded {@link BlockingQueue}
* implementation.
* implementation and without buffer consolidation.
*/
public BufferedWriteHandler() {
this(new LinkedTransferQueue<MessageEvent>());
this(false);
}
/**
* Creates a new instance with the specified thread-safe unbounded
* {@link Queue} and without buffer consolidation. Please note that
* specifying a bounded {@link Queue} or a thread-unsafe {@link Queue} will
* result in an unspecified behavior.
*/
public BufferedWriteHandler(Queue<MessageEvent> queue) {
this(queue, false);
}
/**
* Creates a new instance with the default unbounded {@link BlockingQueue}
* implementation.
*
* @param consolidateOnFlush
* {@code true} if and only if the buffered write requests are merged
* into a single write request on {@link #flush()}
*/
public BufferedWriteHandler(boolean consolidateOnFlush) {
this(new LinkedTransferQueue<MessageEvent>(), consolidateOnFlush);
}
/**
* Creates a new instance with the specified thread-safe unbounded
* {@link Queue}. Please note that specifying a bounded {@link Queue} or
* a thread-unsafe {@link Queue} will result in an unspecified behavior.
*
* @param consolidateOnFlush
* {@code true} if and only if the buffered write requests are merged
* into a single write request on {@link #flush()}
*/
public BufferedWriteHandler(Queue<MessageEvent> queue) {
public BufferedWriteHandler(Queue<MessageEvent> queue, boolean consolidateOnFlush) {
if (queue == null) {
throw new NullPointerException("queue");
}
this.queue = queue;
this.consolidateOnFlush = consolidateOnFlush;
}
public boolean isConsolidateOnFlush() {
return consolidateOnFlush;
}
/**
@ -125,6 +228,17 @@ public class BufferedWriteHandler extends SimpleChannelDownstreamHandler {
* Sends the queued write requests to the downstream.
*/
public void flush() {
flush(consolidateOnFlush);
}
/**
* Sends the queued write requests to the downstream.
*
* @param consolidateOnFlush
* {@code true} if and only if the buffered write requests are merged
* into a single write request
*/
public void flush(boolean consolidateOnFlush) {
final ChannelHandlerContext ctx = this.ctx;
if (ctx == null) {
// No write request was made.
@ -132,6 +246,31 @@ public class BufferedWriteHandler extends SimpleChannelDownstreamHandler {
}
final Queue<MessageEvent> queue = getQueue();
if (consolidateOnFlush) {
if (queue.isEmpty()) {
return;
}
List<MessageEvent> pendingWrites = new ArrayList<MessageEvent>();
synchronized (this) {
for (;;) {
MessageEvent e = queue.poll();
if (e == null) {
break;
}
if (!(e.getMessage() instanceof ChannelBuffer)) {
if ((pendingWrites = consolidatedWrite(pendingWrites)) == null) {
pendingWrites = new ArrayList<MessageEvent>();
}
ctx.sendDownstream(e);
} else {
pendingWrites.add(e);
}
}
consolidatedWrite(pendingWrites);
}
} else {
synchronized (this) {
for (;;) {
MessageEvent e = queue.poll();
if (e == null) {
@ -140,6 +279,44 @@ public class BufferedWriteHandler extends SimpleChannelDownstreamHandler {
ctx.sendDownstream(e);
}
}
}
}
private List<MessageEvent> consolidatedWrite(final List<MessageEvent> pendingWrites) {
final int size = pendingWrites.size();
if (size == 1) {
ctx.sendDownstream(pendingWrites.remove(0));
return pendingWrites;
} else if (size == 0) {
return pendingWrites;
}
ChannelBuffer[] data = new ChannelBuffer[size];
for (int i = 0; i < data.length; i ++) {
data[i] = (ChannelBuffer) pendingWrites.get(i).getMessage();
}
ChannelBuffer composite = ChannelBuffers.wrappedBuffer(data);
ChannelFuture future = Channels.future(ctx.getChannel());
future.addListener(new ChannelFutureListener() {
public void operationComplete(ChannelFuture future)
throws Exception {
if (future.isSuccess()) {
for (MessageEvent e: pendingWrites) {
e.getFuture().setSuccess();
}
} else {
Throwable cause = future.getCause();
for (MessageEvent e: pendingWrites) {
e.getFuture().setFailure(cause);
}
}
}
});
Channels.write(ctx, future, composite);
return null;
}
/**
* Stores all write requests to the queue so that they are actually written
@ -161,7 +338,7 @@ public class BufferedWriteHandler extends SimpleChannelDownstreamHandler {
public void disconnectRequested(ChannelHandlerContext ctx,
ChannelStateEvent e) throws Exception {
try {
flush();
flush(consolidateOnFlush);
} finally {
ctx.sendDownstream(e);
}
@ -171,7 +348,7 @@ public class BufferedWriteHandler extends SimpleChannelDownstreamHandler {
public void closeRequested(ChannelHandlerContext ctx, ChannelStateEvent e)
throws Exception {
try {
flush();
flush(consolidateOnFlush);
} finally {
ctx.sendDownstream(e);
}