[#887] [#866] [#883] Add unified interface for Message oriented protocols and also use direct buffers for them

This commit is contained in:
Norman Maurer 2013-01-01 16:03:18 +01:00
parent 4a1aa37773
commit 37a3f2e3b8
18 changed files with 457 additions and 140 deletions

View File

@ -18,22 +18,9 @@ package io.netty.buffer;
/** /**
* A buffer to operate on * A buffer to operate on
*/ */
public interface Buf { public interface Buf extends Freeable {
/** /**
* The BufType which will be handled by the Buf implementation * The BufType which will be handled by the Buf implementation
*/ */
BufType type(); BufType type();
/**
* Returns {@code true} if and only if this buffer has been deallocated by {@link #free()}.
*/
boolean isFreed();
/**
* Deallocates the internal memory block of this buffer or returns it to the allocator or pool it came from.
* The result of accessing a released buffer is unspecified and can even cause JVM crash.
*
* @throws UnsupportedOperationException if this buffer is derived
*/
void free();
} }

View File

@ -1893,4 +1893,19 @@ public interface ByteBuf extends Buf, Comparable<ByteBuf> {
*/ */
@Override @Override
String toString(); String toString();
/**
* Deallocates the internal memory block of this buffer or returns it to the allocator or pool it came from.
* The result of accessing a released buffer is unspecified and can even cause JVM crash.
*
* @throws UnsupportedOperationException if this buffer is derived
*/
@Override
void free();
/**
* Returns {@code true} if and only if this buffer has been deallocated by {@link #free()}.
*/
@Override
boolean isFreed();
} }

View File

@ -0,0 +1,52 @@
/*
* Copyright 2013 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.buffer;
/**
* A packet which is send or receive. The contract for a {@link ByteBufHolder} is the
* following:
*
* When send a {@link ByteBufHolder} the {@link ByteBufHolder} will be freed by calling {@link #free()}
* in the actual transport implementation. When receive a {@link ByteBufHolder} the {@link #free()}
* must be called once is is processed.
*
*/
public interface ByteBufHolder extends Freeable {
/**
* Return the data which is held by this {@link ByteBufHolder}.
*
*/
ByteBuf data();
/**
* Create a copy of this {@link ByteBufHolder} which can be used even after {@link #free()}
* is called.
*/
ByteBufHolder copy();
/**
* Free of the resources that are hold by this instance. This includes the {@link ByteBuf}.
*/
@Override
void free();
/**
* Returns {@code true} if and only if this instances was freed.
*/
@Override
boolean isFreed();
}

View File

@ -0,0 +1,65 @@
/*
* Copyright 2013 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.buffer;
/**
* Default implementation of a {@link ByteBufHolder} that holds it's data in a {@link ByteBuf}.
*
*/
public class DefaultByteBufHolder implements ByteBufHolder {
private final ByteBuf data;
public DefaultByteBufHolder(ByteBuf data) {
if (data == null) {
throw new NullPointerException("data");
}
if (data.unwrap() != null && !(data instanceof SwappedByteBuf)) {
throw new IllegalArgumentException("Only not-derived ByteBuf instance are supported, you used: "
+ data.getClass().getSimpleName());
}
this.data = data;
}
@Override
public ByteBuf data() {
if (data.isFreed()) {
throw new IllegalBufferAccessException();
}
return data;
}
@Override
public void free() {
data.free();
}
@Override
public boolean isFreed() {
return data.isFreed();
}
@Override
public ByteBufHolder copy() {
return new DefaultByteBufHolder(data().copy());
}
@Override
public String toString() {
if (isFreed()) {
return "Message{data=(FREED)}";
}
return "Message{data=" + ByteBufUtil.hexDump(data()) + '}';
}
}

View File

@ -0,0 +1,32 @@
/*
* Copyright 2013 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.buffer;
public interface Freeable {
/**
* Returns {@code true} if and only if this resource has been deallocated by {@link #free()}.
*/
boolean isFreed();
/**
* Deallocates the resources.
*
* The result of accessing a freed resource is unspecified and can even cause JVM crash.
*
*/
void free();
}

View File

@ -43,6 +43,7 @@ import io.netty.channel.ChannelPipeline;
* } * }
* } * }
* </pre> * </pre>
*
*/ */
public abstract class MessageToMessageDecoder<I, O> public abstract class MessageToMessageDecoder<I, O>
extends ChannelInboundHandlerAdapter implements ChannelInboundMessageHandler<I> { extends ChannelInboundHandlerAdapter implements ChannelInboundMessageHandler<I> {
@ -81,15 +82,24 @@ public abstract class MessageToMessageDecoder<I, O>
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
I imsg = (I) msg; I imsg = (I) msg;
O omsg = decode(ctx, imsg); boolean free = true;
if (omsg == null) { try {
// Decoder consumed a message but returned null. O omsg = decode(ctx, imsg);
// Probably it needs more messages because it's an aggregator. if (omsg == null) {
continue; // Decoder consumed a message but returned null.
} // Probably it needs more messages because it's an aggregator.
continue;
if (ChannelHandlerUtil.unfoldAndAdd(ctx, omsg, true)) { }
notify = true; if (omsg == imsg) {
free = false;
}
if (ChannelHandlerUtil.unfoldAndAdd(ctx, omsg, true)) {
notify = true;
}
} finally {
if (free) {
freeInboundMessage(imsg);
}
} }
} catch (Throwable t) { } catch (Throwable t) {
if (t instanceof CodecException) { if (t instanceof CodecException) {
@ -122,4 +132,13 @@ public abstract class MessageToMessageDecoder<I, O>
* @throws Exception is thrown if an error accour * @throws Exception is thrown if an error accour
*/ */
protected abstract O decode(ChannelHandlerContext ctx, I msg) throws Exception; protected abstract O decode(ChannelHandlerContext ctx, I msg) throws Exception;
/**
* Is called after a message was processed via {@link #decode(ChannelHandlerContext, Object)} to free
* up any resources that is held by the inbound message. You may want to override this if your implementation
* just pass-through the input message or need it for later usage.
*/
protected void freeInboundMessage(I msg) throws Exception {
ChannelHandlerUtil.freeMessage(msg);
}
} }

View File

@ -42,6 +42,7 @@ import io.netty.channel.ChannelPromise;
* } * }
* } * }
* </pre> * </pre>
*
*/ */
public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundMessageHandlerAdapter<I> { public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundMessageHandlerAdapter<I> {
@ -72,14 +73,23 @@ public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundMessa
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
I imsg = (I) msg; I imsg = (I) msg;
O omsg = encode(ctx, imsg); boolean free = true;
if (omsg == null) { try {
// encode() might be waiting for more inbound messages to generate O omsg = encode(ctx, imsg);
// an aggregated message - keep polling. if (omsg == null) {
continue; // encode() might be waiting for more inbound messages to generate
// an aggregated message - keep polling.
continue;
}
if (omsg == imsg) {
free = false;
}
ChannelHandlerUtil.unfoldAndAdd(ctx, omsg, false);
} finally {
if (free) {
freeInboundMessage(imsg);
}
} }
ChannelHandlerUtil.unfoldAndAdd(ctx, omsg, false);
} catch (Throwable t) { } catch (Throwable t) {
if (t instanceof CodecException) { if (t instanceof CodecException) {
ctx.fireExceptionCaught(t); ctx.fireExceptionCaught(t);
@ -112,4 +122,13 @@ public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundMessa
* @throws Exception is thrown if an error accour * @throws Exception is thrown if an error accour
*/ */
protected abstract O encode(ChannelHandlerContext ctx, I msg) throws Exception; protected abstract O encode(ChannelHandlerContext ctx, I msg) throws Exception;
/**
* Is called after a message was processed via {@link #encode(ChannelHandlerContext, Object)} to free
* up any resources that is held by the inbound message. You may want to override this if your implementation
* just pass-through the input message or need it for later usage.
*/
protected void freeInboundMessage(I msg) throws Exception {
ChannelHandlerUtil.freeMessage(msg);
}
} }

View File

@ -35,28 +35,31 @@ public class SctpInboundByteStreamHandler extends ChannelInboundMessageHandlerAd
* @param protocolIdentifier supported application protocol. * @param protocolIdentifier supported application protocol.
*/ */
public SctpInboundByteStreamHandler(int protocolIdentifier, int streamIdentifier) { public SctpInboundByteStreamHandler(int protocolIdentifier, int streamIdentifier) {
super(SctpMessage.class);
this.protocolIdentifier = protocolIdentifier; this.protocolIdentifier = protocolIdentifier;
this.streamIdentifier = streamIdentifier; this.streamIdentifier = streamIdentifier;
} }
@Override
public boolean isSupported(Object msg) throws Exception {
if (super.isSupported(msg)) {
return isDecodable((SctpMessage) msg);
}
return false;
}
protected boolean isDecodable(SctpMessage msg) { protected boolean isDecodable(SctpMessage msg) {
return msg.protocolIdentifier() == protocolIdentifier && msg.streamIdentifier() == streamIdentifier; return msg.protocolIdentifier() == protocolIdentifier && msg.streamIdentifier() == streamIdentifier;
} }
@Override @Override
protected void messageReceived(ChannelHandlerContext ctx, SctpMessage msg) throws Exception { protected void messageReceived(ChannelHandlerContext ctx, SctpMessage msg) throws Exception {
if (!isDecodable(msg)) {
ctx.nextInboundMessageBuffer().add(msg);
ctx.fireInboundBufferUpdated();
return;
}
if (!msg.isComplete()) { if (!msg.isComplete()) {
throw new CodecException(String.format("Received SctpMessage is not complete, please add %s in the " + throw new CodecException(String.format("Received SctpMessage is not complete, please add %s in the " +
"pipeline before this handler", SctpMessageCompletionHandler.class.getSimpleName())); "pipeline before this handler", SctpMessageCompletionHandler.class.getSimpleName()));
} }
ctx.nextInboundByteBuffer().writeBytes(msg.payloadBuffer()); ctx.nextInboundByteBuffer().writeBytes(msg.data());
ctx.fireInboundBufferUpdated(); ctx.fireInboundBufferUpdated();
} }
} }

View File

@ -53,7 +53,7 @@ public class SctpMessageCompletionHandler extends ChannelInboundMessageHandlerAd
@Override @Override
protected void messageReceived(ChannelHandlerContext ctx, SctpMessage msg) throws Exception { protected void messageReceived(ChannelHandlerContext ctx, SctpMessage msg) throws Exception {
final ByteBuf byteBuf = msg.payloadBuffer(); final ByteBuf byteBuf = msg.data();
final int protocolIdentifier = msg.protocolIdentifier(); final int protocolIdentifier = msg.protocolIdentifier();
final int streamIdentifier = msg.streamIdentifier(); final int streamIdentifier = msg.streamIdentifier();
final boolean isComplete = msg.isComplete(); final boolean isComplete = msg.isComplete();
@ -90,4 +90,9 @@ public class SctpMessageCompletionHandler extends ChannelInboundMessageHandlerAd
ctx.nextInboundMessageBuffer().add(assembledMsg); ctx.nextInboundMessageBuffer().add(assembledMsg);
assembled = true; assembled = true;
} }
@Override
protected void freeInboundMessage(SctpMessage msg) throws Exception {
// It is an aggregator so not free it yet
}
} }

View File

@ -358,6 +358,8 @@
-XX:+UseStringCache -XX:+UseStringCache
-XX:+OptimizeStringConcat -XX:+OptimizeStringConcat
-XX:+HeapDumpOnOutOfMemoryError -XX:+HeapDumpOnOutOfMemoryError
-Xmx2048m
-Xms1024m
</argLine> </argLine>
</configuration> </configuration>
</plugin> </plugin>

View File

@ -16,6 +16,8 @@
package io.netty.channel; package io.netty.channel;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufHolder;
import io.netty.buffer.Freeable;
import io.netty.buffer.MessageBuf; import io.netty.buffer.MessageBuf;
/** /**
@ -166,6 +168,15 @@ public final class ChannelHandlerUtil {
} }
} }
/**
* Try to free up resources that are held by the message.
*/
public static void freeMessage(Object msg) throws Exception {
if (msg instanceof Freeable) {
((Freeable) msg).free();
}
}
private ChannelHandlerUtil() { private ChannelHandlerUtil() {
// Unused // Unused
} }

View File

@ -58,7 +58,6 @@ public abstract class ChannelInboundMessageHandlerAdapter<I>
return Unpooled.messageBuffer(); return Unpooled.messageBuffer();
} }
@SuppressWarnings("unchecked")
@Override @Override
public final void inboundBufferUpdated(ChannelHandlerContext ctx) throws Exception { public final void inboundBufferUpdated(ChannelHandlerContext ctx) throws Exception {
if (!beginMessageReceived(ctx)) { if (!beginMessageReceived(ctx)) {
@ -86,7 +85,14 @@ public abstract class ChannelInboundMessageHandlerAdapter<I>
unsupportedFound = false; unsupportedFound = false;
ctx.fireInboundBufferUpdated(); ctx.fireInboundBufferUpdated();
} }
messageReceived(ctx, (I) msg);
@SuppressWarnings("unchecked")
I imsg = (I) msg;
try {
messageReceived(ctx, imsg);
} finally {
freeInboundMessage(imsg);
}
} catch (Throwable t) { } catch (Throwable t) {
exceptionCaught(ctx, t); exceptionCaught(ctx, t);
} }
@ -144,4 +150,13 @@ public abstract class ChannelInboundMessageHandlerAdapter<I>
protected void endMessageReceived(ChannelHandlerContext ctx) throws Exception { protected void endMessageReceived(ChannelHandlerContext ctx) throws Exception {
// NOOP // NOOP
} }
/**
* Is called after a message was processed via {@link #messageReceived(ChannelHandlerContext, Object)} to free
* up any resources that is held by the inbound message. You may want to override this if your implementation
* just pass-through the input message or need it for later usage.
*/
protected void freeInboundMessage(I msg) throws Exception {
ChannelHandlerUtil.freeMessage(msg);
}
} }

View File

@ -16,15 +16,16 @@
package io.netty.channel.socket; package io.netty.channel.socket;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.DefaultByteBufHolder;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
/** /**
* The message container that is used for {@link DatagramChannel} to communicate with the remote peer. * The message container that is used for {@link DatagramChannel} to communicate with the remote peer.
*/ */
public final class DatagramPacket { public final class DatagramPacket extends DefaultByteBufHolder {
private final ByteBuf data;
private final InetSocketAddress remoteAddress; private final InetSocketAddress remoteAddress;
/** /**
@ -35,24 +36,13 @@ public final class DatagramPacket {
* packet will be send * packet will be send
*/ */
public DatagramPacket(ByteBuf data, InetSocketAddress remoteAddress) { public DatagramPacket(ByteBuf data, InetSocketAddress remoteAddress) {
if (data == null) { super(data);
throw new NullPointerException("data");
}
if (remoteAddress == null) { if (remoteAddress == null) {
throw new NullPointerException("remoteAddress"); throw new NullPointerException("remoteAddress");
} }
this.data = data;
this.remoteAddress = remoteAddress; this.remoteAddress = remoteAddress;
} }
/**
* Return the data which is container. May return an empty {@link ByteBuf}
*/
public ByteBuf data() {
return data;
}
/** /**
* The {@link InetSocketAddress} which this {@link DatagramPacket} will send to or was received from. * The {@link InetSocketAddress} which this {@link DatagramPacket} will send to or was received from.
*/ */
@ -60,8 +50,18 @@ public final class DatagramPacket {
return remoteAddress; return remoteAddress;
} }
@Override
public DatagramPacket copy() {
return new DatagramPacket(data().copy(), remoteAddress());
}
@Override @Override
public String toString() { public String toString() {
return "datagram(" + data.readableBytes() + "B, " + remoteAddress + ')'; if (isFreed()) {
return "DatagramPacket{remoteAddress=" + remoteAddress().toString() +
", data=(FREED)}";
}
return "DatagramPacket{remoteAddress=" + remoteAddress().toString() +
", data=" + ByteBufUtil.hexDump(data()) + '}';
} }
} }

View File

@ -18,16 +18,15 @@ package io.netty.channel.socket;
import com.sun.nio.sctp.MessageInfo; import com.sun.nio.sctp.MessageInfo;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil; import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled; import io.netty.buffer.DefaultByteBufHolder;
/** /**
* Representation of SCTP Data Chunk * Representation of SCTP Data Chunk
*/ */
public final class SctpMessage { public final class SctpMessage extends DefaultByteBufHolder {
private final int streamIdentifier; private final int streamIdentifier;
private final int protocolIdentifier; private final int protocolIdentifier;
private final ByteBuf payloadBuffer;
private final MessageInfo msgInfo; private final MessageInfo msgInfo;
/** /**
@ -37,9 +36,9 @@ public final class SctpMessage {
* @param payloadBuffer channel buffer * @param payloadBuffer channel buffer
*/ */
public SctpMessage(int protocolIdentifier, int streamIdentifier, ByteBuf payloadBuffer) { public SctpMessage(int protocolIdentifier, int streamIdentifier, ByteBuf payloadBuffer) {
super(payloadBuffer);
this.protocolIdentifier = protocolIdentifier; this.protocolIdentifier = protocolIdentifier;
this.streamIdentifier = streamIdentifier; this.streamIdentifier = streamIdentifier;
this.payloadBuffer = payloadBuffer;
msgInfo = null; msgInfo = null;
} }
@ -49,16 +48,13 @@ public final class SctpMessage {
* @param payloadBuffer channel buffer * @param payloadBuffer channel buffer
*/ */
public SctpMessage(MessageInfo msgInfo, ByteBuf payloadBuffer) { public SctpMessage(MessageInfo msgInfo, ByteBuf payloadBuffer) {
super(payloadBuffer);
if (msgInfo == null) { if (msgInfo == null) {
throw new NullPointerException("msgInfo"); throw new NullPointerException("msgInfo");
} }
if (payloadBuffer == null) {
throw new NullPointerException("payloadBuffer");
}
this.msgInfo = msgInfo; this.msgInfo = msgInfo;
streamIdentifier = msgInfo.streamNumber(); streamIdentifier = msgInfo.streamNumber();
protocolIdentifier = msgInfo.payloadProtocolID(); protocolIdentifier = msgInfo.payloadProtocolID();
this.payloadBuffer = payloadBuffer;
} }
/** /**
@ -75,17 +71,6 @@ public final class SctpMessage {
return protocolIdentifier; return protocolIdentifier;
} }
/**
* Return a view of the readable bytes of the payload.
*/
public ByteBuf payloadBuffer() {
if (payloadBuffer.readable()) {
return payloadBuffer.slice();
} else {
return Unpooled.EMPTY_BUFFER;
}
}
/** /**
* Return the {@link MessageInfo} for inbound messages or {@code null} for * Return the {@link MessageInfo} for inbound messages or {@code null} for
* outbound messages. * outbound messages.
@ -126,7 +111,7 @@ public final class SctpMessage {
return false; return false;
} }
if (!payloadBuffer.equals(sctpFrame.payloadBuffer)) { if (!data().equals(sctpFrame.data())) {
return false; return false;
} }
@ -137,14 +122,28 @@ public final class SctpMessage {
public int hashCode() { public int hashCode() {
int result = streamIdentifier; int result = streamIdentifier;
result = 31 * result + protocolIdentifier; result = 31 * result + protocolIdentifier;
result = 31 * result + payloadBuffer.hashCode(); result = 31 * result + data().hashCode();
return result; return result;
} }
@Override
public SctpMessage copy() {
if (msgInfo == null) {
return new SctpMessage(protocolIdentifier, streamIdentifier, data().copy());
} else {
return new SctpMessage(msgInfo, data().copy());
}
}
@Override @Override
public String toString() { public String toString() {
if (isFreed()) {
return "SctpFrame{" +
"streamIdentifier=" + streamIdentifier + ", protocolIdentifier=" + protocolIdentifier +
", data=(FREED)}";
}
return "SctpFrame{" + return "SctpFrame{" +
"streamIdentifier=" + streamIdentifier + ", protocolIdentifier=" + protocolIdentifier + "streamIdentifier=" + streamIdentifier + ", protocolIdentifier=" + protocolIdentifier +
", payloadBuffer=" + ByteBufUtil.hexDump(payloadBuffer()) + '}'; ", data=" + ByteBufUtil.hexDump(data()) + '}';
} }
} }

View File

@ -18,7 +18,6 @@ package io.netty.channel.socket.nio;
import io.netty.buffer.BufType; import io.netty.buffer.BufType;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.buffer.MessageBuf; import io.netty.buffer.MessageBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelException; import io.netty.channel.ChannelException;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelMetadata; import io.netty.channel.ChannelMetadata;
@ -192,15 +191,34 @@ public final class NioDatagramChannel
@Override @Override
protected int doReadMessages(MessageBuf<Object> buf) throws Exception { protected int doReadMessages(MessageBuf<Object> buf) throws Exception {
DatagramChannel ch = javaChannel(); DatagramChannel ch = javaChannel();
ByteBuffer data = ByteBuffer.allocate(config().getReceivePacketSize()); ByteBuf buffer = alloc().directBuffer(config().getReceivePacketSize());
InetSocketAddress remoteAddress = (InetSocketAddress) ch.receive(data); boolean free = true;
if (remoteAddress == null) { try {
return 0; ByteBuffer data = buffer.nioBuffer(buffer.writerIndex(), buffer.writableBytes());
}
data.flip(); InetSocketAddress remoteAddress = (InetSocketAddress) ch.receive(data);
buf.add(new DatagramPacket(Unpooled.wrappedBuffer(data), remoteAddress)); if (remoteAddress == null) {
return 1; return 0;
}
buf.add(new DatagramPacket(buffer.writerIndex(buffer.writerIndex() + data.remaining()), remoteAddress));
free = false;
return 1;
} catch (Throwable cause) {
if (cause instanceof Error) {
throw (Error) cause;
}
if (cause instanceof RuntimeException) {
throw (RuntimeException) cause;
}
if (cause instanceof Exception) {
throw (Exception) cause;
}
throw new ChannelException(cause);
} finally {
if (free) {
buffer.free();
}
}
} }
@Override @Override
@ -237,6 +255,10 @@ public final class NioDatagramChannel
// Wrote a packet. // Wrote a packet.
buf.remove(); buf.remove();
// packet was written free up buffer
packet.free();
if (buf.isEmpty()) { if (buf.isEmpty()) {
// Wrote the outbound buffer completely - clear OP_WRITE. // Wrote the outbound buffer completely - clear OP_WRITE.
if ((interestOps & SelectionKey.OP_WRITE) != 0) { if ((interestOps & SelectionKey.OP_WRITE) != 0) {

View File

@ -22,7 +22,6 @@ import com.sun.nio.sctp.SctpChannel;
import io.netty.buffer.BufType; import io.netty.buffer.BufType;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.buffer.MessageBuf; import io.netty.buffer.MessageBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel; import io.netty.channel.Channel;
import io.netty.channel.ChannelException; import io.netty.channel.ChannelException;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
@ -244,21 +243,41 @@ public class NioSctpChannel extends AbstractNioMessageChannel implements io.nett
@Override @Override
protected int doReadMessages(MessageBuf<Object> buf) throws Exception { protected int doReadMessages(MessageBuf<Object> buf) throws Exception {
SctpChannel ch = javaChannel(); SctpChannel ch = javaChannel();
ByteBuffer data = ByteBuffer.allocate(config().getReceiveBufferSize()); ByteBuf buffer = alloc().directBuffer(config().getReceiveBufferSize());
MessageInfo messageInfo = ch.receive(data, null, notificationHandler); boolean free = true;
if (messageInfo == null) { try {
return 0; ByteBuffer data = buffer.nioBuffer(buffer.writerIndex(), buffer.writableBytes());
} MessageInfo messageInfo = ch.receive(data, null, notificationHandler);
if (messageInfo == null) {
return 0;
}
data.flip(); data.flip();
buf.add(new SctpMessage(messageInfo, Unpooled.wrappedBuffer(data))); buf.add(new SctpMessage(messageInfo, buffer.writerIndex(buffer.writerIndex() + data.remaining())));
return 1; free = false;
return 1;
} catch (Throwable cause) {
if (cause instanceof Error) {
throw (Error) cause;
}
if (cause instanceof RuntimeException) {
throw (RuntimeException) cause;
}
if (cause instanceof Exception) {
throw (Exception) cause;
}
throw new ChannelException(cause);
} finally {
if (free) {
buffer.free();
}
}
} }
@Override @Override
protected int doWriteMessages(MessageBuf<Object> buf, boolean lastSpin) throws Exception { protected int doWriteMessages(MessageBuf<Object> buf, boolean lastSpin) throws Exception {
SctpMessage packet = (SctpMessage) buf.peek(); SctpMessage packet = (SctpMessage) buf.peek();
ByteBuf data = packet.payloadBuffer(); ByteBuf data = packet.data();
int dataLen = data.readableBytes(); int dataLen = data.readableBytes();
ByteBuffer nioData; ByteBuffer nioData;
if (data.nioBufferCount() == 1) { if (data.nioBufferCount() == 1) {
@ -293,6 +312,10 @@ public class NioSctpChannel extends AbstractNioMessageChannel implements io.nett
// Wrote a packet. // Wrote a packet.
buf.remove(); buf.remove();
// packet was written free up buffer
packet.free();
if (buf.isEmpty()) { if (buf.isEmpty()) {
// Wrote the outbound buffer completely - clear OP_WRITE. // Wrote the outbound buffer completely - clear OP_WRITE.
if ((interestOps & SelectionKey.OP_WRITE) != 0) { if ((interestOps & SelectionKey.OP_WRITE) != 0) {

View File

@ -183,17 +183,23 @@ public class OioDatagramChannel extends AbstractOioMessageChannel
@Override @Override
protected int doReadMessages(MessageBuf<Object> buf) throws Exception { protected int doReadMessages(MessageBuf<Object> buf) throws Exception {
int packetSize = config().getReceivePacketSize(); int packetSize = config().getReceivePacketSize();
byte[] data = new byte[packetSize]; // TODO: Use alloc().heapBuffer(..) but there seems to be a memory-leak, need to investigate
tmpPacket.setData(data); ByteBuf buffer = Unpooled.buffer(packetSize);
boolean free = true;
try { try {
int writerIndex = buffer.writerIndex();
tmpPacket.setData(buffer.array(), writerIndex + buffer.arrayOffset(), packetSize);
socket.receive(tmpPacket); socket.receive(tmpPacket);
InetSocketAddress remoteAddr = (InetSocketAddress) tmpPacket.getSocketAddress(); InetSocketAddress remoteAddr = (InetSocketAddress) tmpPacket.getSocketAddress();
if (remoteAddr == null) { if (remoteAddr == null) {
remoteAddr = remoteAddress(); remoteAddr = remoteAddress();
} }
buf.add(new DatagramPacket(Unpooled.wrappedBuffer( DatagramPacket packet = new DatagramPacket(buffer.writerIndex(writerIndex + tmpPacket.getLength())
data, tmpPacket.getOffset(), tmpPacket.getLength()), remoteAddr)); .readerIndex(writerIndex), remoteAddr);
buf.add(packet);
free = false;
return 1; return 1;
} catch (SocketTimeoutException e) { } catch (SocketTimeoutException e) {
// Expected // Expected
@ -203,27 +209,46 @@ public class OioDatagramChannel extends AbstractOioMessageChannel
throw e; throw e;
} }
return -1; return -1;
} catch (Throwable cause) {
if (cause instanceof Error) {
throw (Error) cause;
}
if (cause instanceof RuntimeException) {
throw (RuntimeException) cause;
}
if (cause instanceof Exception) {
throw (Exception) cause;
}
throw new ChannelException(cause);
} finally {
if (free) {
buffer.free();
}
} }
} }
@Override @Override
protected void doWriteMessages(MessageBuf<Object> buf) throws Exception { protected void doWriteMessages(MessageBuf<Object> buf) throws Exception {
DatagramPacket p = (DatagramPacket) buf.poll(); DatagramPacket p = (DatagramPacket) buf.poll();
ByteBuf data = p.data();
int length = data.readableBytes();
InetSocketAddress remote = p.remoteAddress();
if (remote != null) {
tmpPacket.setSocketAddress(remote);
}
if (data.hasArray()) {
tmpPacket.setData(data.array(), data.arrayOffset() + data.readerIndex(), length);
} else {
byte[] tmp = new byte[length];
data.getBytes(data.readerIndex(), tmp);
tmpPacket.setData(tmp);
}
socket.send(tmpPacket); try {
ByteBuf data = p.data();
int length = data.readableBytes();
InetSocketAddress remote = p.remoteAddress();
if (remote != null) {
tmpPacket.setSocketAddress(remote);
}
if (data.hasArray()) {
tmpPacket.setData(data.array(), data.arrayOffset() + data.readerIndex(), length);
} else {
byte[] tmp = new byte[length];
data.getBytes(data.readerIndex(), tmp);
tmpPacket.setData(tmp);
}
socket.send(tmpPacket);
} finally {
p.free();
}
} }
@Override @Override

View File

@ -22,7 +22,6 @@ import com.sun.nio.sctp.SctpChannel;
import io.netty.buffer.BufType; import io.netty.buffer.BufType;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.buffer.MessageBuf; import io.netty.buffer.MessageBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel; import io.netty.channel.Channel;
import io.netty.channel.ChannelException; import io.netty.channel.ChannelException;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
@ -165,16 +164,36 @@ public class OioSctpChannel extends AbstractOioMessageChannel
Set<SelectionKey> reableKeys = readSelector.selectedKeys(); Set<SelectionKey> reableKeys = readSelector.selectedKeys();
try { try {
for (SelectionKey ignored : reableKeys) { for (SelectionKey ignored : reableKeys) {
ByteBuffer data = ByteBuffer.allocate(config().getReceiveBufferSize()); ByteBuf buffer = alloc().directBuffer(config().getReceiveBufferSize());
MessageInfo messageInfo = ch.receive(data, null, notificationHandler); boolean free = true;
if (messageInfo == null) {
return readMessages; try {
ByteBuffer data = buffer.nioBuffer(buffer.writerIndex(), buffer.writableBytes());
MessageInfo messageInfo = ch.receive(data, null, notificationHandler);
if (messageInfo == null) {
return readMessages;
}
data.flip();
buf.add(new SctpMessage(messageInfo, buffer.writerIndex(buffer.writerIndex() + data.remaining())));
free = false;
readMessages ++;
} catch (Throwable cause) {
if (cause instanceof Error) {
throw (Error) cause;
}
if (cause instanceof RuntimeException) {
throw (RuntimeException) cause;
}
if (cause instanceof Exception) {
throw (Exception) cause;
}
throw new ChannelException(cause);
} finally {
if (free) {
buffer.free();
}
} }
data.flip();
buf.add(new SctpMessage(messageInfo, Unpooled.wrappedBuffer(data)));
readMessages ++;
} }
} finally { } finally {
reableKeys.clear(); reableKeys.clear();
@ -196,23 +215,27 @@ public class OioSctpChannel extends AbstractOioMessageChannel
if (packet == null) { if (packet == null) {
return; return;
} }
ByteBuf data = packet.payloadBuffer(); try {
int dataLen = data.readableBytes(); ByteBuf data = packet.data();
ByteBuffer nioData; int dataLen = data.readableBytes();
ByteBuffer nioData;
if (data.nioBufferCount() != -1) { if (data.nioBufferCount() != -1) {
nioData = data.nioBuffer(); nioData = data.nioBuffer();
} else { } else {
nioData = ByteBuffer.allocate(dataLen); nioData = ByteBuffer.allocate(dataLen);
data.getBytes(data.readerIndex(), nioData); data.getBytes(data.readerIndex(), nioData);
nioData.flip(); nioData.flip();
}
final MessageInfo mi = MessageInfo.createOutgoing(association(), null, packet.streamIdentifier());
mi.payloadProtocolID(packet.protocolIdentifier());
mi.streamNumber(packet.streamIdentifier());
ch.send(nioData, mi);
} finally {
packet.free();
} }
final MessageInfo mi = MessageInfo.createOutgoing(association(), null, packet.streamIdentifier());
mi.payloadProtocolID(packet.protocolIdentifier());
mi.streamNumber(packet.streamIdentifier());
ch.send(nioData, mi);
} }
writableKeys.clear(); writableKeys.clear();
} }