From 706db46b7ea30d51c7012400a3c6a8b11dd1462d Mon Sep 17 00:00:00 2001 From: Norman Maurer Date: Sat, 10 May 2014 15:21:25 +0200 Subject: [PATCH] [#2485] Use RecvByteBufAllocator for all allocations related to read from Channel Motivation: At the moment we sometimes use only RecvByteBufAllocator.guess() to guess the next size and the use the ByteBufAllocator.* directly to allocate the buffer. We should always use RecvByteBufAllocator.allocate(...) all the time as this makes the behavior easier to adjust. Modifications: Change the read() implementations to make use of RecvByteBufAllocator. Result: Behavior is more consistent. --- .../channel/epoll/EpollSocketChannel.java | 3 +-- .../channel/nio/AbstractNioByteChannel.java | 3 +-- .../channel/oio/AbstractOioByteChannel.java | 24 +++++++++++++++++-- 3 files changed, 24 insertions(+), 6 deletions(-) diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java index 2dda84aafd..11f8af80b0 100644 --- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java +++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannel.java @@ -597,12 +597,11 @@ public final class EpollSocketChannel extends AbstractEpollChannel implements So ByteBuf byteBuf = null; boolean close = false; try { - int byteBufCapacity = allocHandle.guess(); int totalReadAmount = 0; for (;;) { // we use a direct buffer here as the native implementations only be able // to handle direct buffers. - byteBuf = allocator.directBuffer(byteBufCapacity); + byteBuf = allocHandle.allocate(allocator); int writable = byteBuf.writableBytes(); int localReadAmount = doReadBytes(byteBuf); if (localReadAmount <= 0) { diff --git a/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java b/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java index 18ed477afe..d18cff47f4 100644 --- a/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java +++ b/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java @@ -106,11 +106,10 @@ public abstract class AbstractNioByteChannel extends AbstractNioChannel { int messages = 0; boolean close = false; try { - int byteBufCapacity = allocHandle.guess(); int totalReadAmount = 0; boolean readPendingReset = false; do { - byteBuf = allocator.ioBuffer(byteBufCapacity); + byteBuf = allocHandle.allocate(allocator); int writable = byteBuf.writableBytes(); int localReadAmount = doReadBytes(byteBuf); if (localReadAmount <= 0) { diff --git a/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java b/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java index e4fd801c4a..4215ad2771 100644 --- a/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java +++ b/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java @@ -23,6 +23,7 @@ import io.netty.channel.ChannelOption; import io.netty.channel.ChannelOutboundBuffer; import io.netty.channel.ChannelPipeline; import io.netty.channel.FileRegion; +import io.netty.channel.RecvByteBufAllocator; import io.netty.channel.socket.ChannelInputShutdownEvent; import io.netty.util.internal.StringUtil; @@ -32,6 +33,7 @@ import java.io.IOException; * Abstract base class for OIO which reads and writes bytes from/to a Socket */ public abstract class AbstractOioByteChannel extends AbstractOioChannel { + private RecvByteBufAllocator.Handle allocHandle; private volatile boolean inputShutdown; private static final ChannelMetadata METADATA = new ChannelMetadata(false); @@ -76,13 +78,20 @@ public abstract class AbstractOioByteChannel extends AbstractOioChannel { final ChannelConfig config = config(); final ChannelPipeline pipeline = pipeline(); - // TODO: calculate size as in 3.x - ByteBuf byteBuf = alloc().buffer(); + RecvByteBufAllocator.Handle allocHandle = this.allocHandle; + if (allocHandle == null) { + this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle(); + } + + ByteBuf byteBuf = allocHandle.allocate(alloc()); + boolean closed = false; boolean read = false; Throwable exception = null; int localReadAmount = 0; try { + int totalReadAmount = 0; + for (;;) { localReadAmount = doReadBytes(byteBuf); if (localReadAmount > 0) { @@ -114,12 +123,23 @@ public abstract class AbstractOioByteChannel extends AbstractOioChannel { } } } + + if (totalReadAmount >= Integer.MAX_VALUE - localReadAmount) { + // Avoid overflow. + totalReadAmount = Integer.MAX_VALUE; + break; + } + + totalReadAmount += localReadAmount; + if (!config.isAutoRead()) { // stop reading until next Channel.read() call // See https://github.com/netty/netty/issues/1363 break; } } + allocHandle.record(totalReadAmount); + } catch (Throwable t) { exception = t; } finally {