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

Make NioByteUnsafe.read() respect ChannelConfig.maxMessagesPerRead and adjust the default from 16 to 1

Browse Source 
- Fixes #1486
- Decreased the default from 16 to 1 because unnecessary extra read on req-res protocols results in lower throughput due to extra syscalls.
This commit is contained in:
Trustin Lee 2013-06-25 18:08:39 +09:00
parent a1632e7d15
commit 1f27c3b390
2 changed files with 36 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
transport/src/main/java/io/netty/channel/DefaultChannelConfig.java
View File

@ -39,7 +39,7 @@ public class DefaultChannelConfig implements ChannelConfig {
private volatile ByteBufAllocator allocator = DEFAULT_ALLOCATOR;
private volatile RecvByteBufAllocator rcvBufAllocator = DEFAULT_RCVBUF_ALLOCATOR;
private volatile int connectTimeoutMillis = DEFAULT_CONNECT_TIMEOUT;
private volatile int maxMessagesPerRead = 16;
private volatile int maxMessagesPerRead = 1;
private volatile int writeSpinCount = 16;
private volatile boolean autoRead = true;
private volatile int writeBufferHighWaterMark = 64 * 1024;

45
transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java
View File

@ -16,6 +16,7 @@
package io.netty.channel.nio;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelConfig;
import io.netty.channel.ChannelOption;
@ -75,25 +76,49 @@ public abstract class AbstractNioByteChannel extends AbstractNioChannel {
this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
}
ByteBuf byteBuf = allocHandle.allocate(config.getAllocator());
final ByteBufAllocator allocator = config.getAllocator();
final int maxMessagesPerRead = config.getMaxMessagesPerRead();
final MessageList<ByteBuf> messages = MessageList.newInstance();
boolean closed = false;
Throwable exception = null;
ByteBuf byteBuf = null;
try {
int localReadAmount = doReadBytes(byteBuf);
if (localReadAmount < 0) {
closed = true;
for (;;) {
byteBuf = allocHandle.allocate(allocator);
int localReadAmount = doReadBytes(byteBuf);
if (localReadAmount == 0) {
byteBuf.release();
byteBuf = null;
break;
}
if (localReadAmount < 0) {
closed = true;
byteBuf.release();
byteBuf = null;
break;
}
messages.add(byteBuf);
allocHandle.record(localReadAmount);
byteBuf = null;
if (messages.size() == maxMessagesPerRead) {
break;
}
}
} catch (Throwable t) {
exception = t;
} finally {
int readBytes = byteBuf.readableBytes();
allocHandle.record(readBytes);
if (readBytes != 0) {
pipeline.fireMessageReceived(byteBuf);
} else {
byteBuf.release();
if (byteBuf != null) {
if (byteBuf.isReadable()) {
messages.add(byteBuf);
} else {
byteBuf.release();
}
}
pipeline.fireMessageReceived(messages);
if (exception != null) {
if (exception instanceof IOException) {
closed = true;