This is an automated email from the ASF dual-hosted git repository.
andor pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/zookeeper.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new 63e656d ZOOKEEPER-3274: Use CompositeByteBuf to queue data in
NettyServerCnxn
63e656d is described below
commit 63e656d47571471f5c283276f56cf54eb3414628
Author: Ilya Maykov <[email protected]>
AuthorDate: Mon Feb 18 17:05:51 2019 +0100
ZOOKEEPER-3274: Use CompositeByteBuf to queue data in NettyServerCnxn
This avoids unnecessary buffer copies and resizes.
Author: Ilya Maykov <[email protected]>
Reviewers: [email protected]
Closes #810 from ivmaykov/ZOOKEEPER-3274
(cherry picked from commit 5375c2542977223ff9beaeb62d6ee5aed783118e)
Signed-off-by: Andor Molnar <[email protected]>
---
.../apache/zookeeper/server/NettyServerCnxn.java | 35 +++++++++++++++++-----
1 file changed, 27 insertions(+), 8 deletions(-)
diff --git
a/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxn.java
b/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxn.java
index 423fc7f..e6f3495 100644
---
a/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxn.java
+++
b/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxn.java
@@ -32,6 +32,7 @@ import java.util.concurrent.atomic.AtomicLong;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.CompositeByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
@@ -53,7 +54,7 @@ import org.slf4j.LoggerFactory;
public class NettyServerCnxn extends ServerCnxn {
private static final Logger LOG =
LoggerFactory.getLogger(NettyServerCnxn.class);
private final Channel channel;
- private ByteBuf queuedBuffer;
+ private CompositeByteBuf queuedBuffer;
private final AtomicBoolean throttled = new AtomicBoolean(false);
private ByteBuffer bb;
private final ByteBuffer bbLen = ByteBuffer.allocate(4);
@@ -298,6 +299,24 @@ public class NettyServerCnxn extends ServerCnxn {
}
/**
+ * Appends <code>buf</code> to <code>queuedBuffer</code>. Does not
duplicate <code>buf</code>
+ * or call any flavor of {@link ByteBuf#retain()}. Caller must ensure that
<code>buf</code>
+ * is not owned by anyone else, as this call transfers ownership of
<code>buf</code> to the
+ * <code>queuedBuffer</code>.
+ *
+ * This method should only be called from the event loop thread.
+ * @param buf the buffer to append to the queue.
+ */
+ private void appendToQueuedBuffer(ByteBuf buf) {
+ checkIsInEventLoop("appendToQueuedBuffer");
+ if (queuedBuffer.numComponents() == queuedBuffer.maxNumComponents()) {
+ // queuedBuffer has reached its component limit, so combine the
existing components.
+ queuedBuffer.consolidate();
+ }
+ queuedBuffer.addComponent(true, buf);
+ }
+
+ /**
* Process incoming message. This should only be called from the event
* loop thread.
* Note that this method does not call <code>buf.release()</code>. The
caller
@@ -324,9 +343,9 @@ public class NettyServerCnxn extends ServerCnxn {
// we are throttled, so we need to queue
if (queuedBuffer == null) {
LOG.debug("allocating queue");
- queuedBuffer = channel.alloc().buffer(buf.readableBytes());
+ queuedBuffer = channel.alloc().compositeBuffer();
}
- queuedBuffer.writeBytes(buf);
+ appendToQueuedBuffer(buf.retainedDuplicate());
if (LOG.isTraceEnabled()) {
LOG.trace("0x{} queuedBuffer {}",
Long.toHexString(sessionId),
@@ -335,7 +354,7 @@ public class NettyServerCnxn extends ServerCnxn {
} else {
LOG.debug("not throttled");
if (queuedBuffer != null) {
- queuedBuffer.writeBytes(buf);
+ appendToQueuedBuffer(buf.retainedDuplicate());
processQueuedBuffer();
} else {
receiveMessage(buf);
@@ -346,9 +365,9 @@ public class NettyServerCnxn extends ServerCnxn {
LOG.trace("Before copy {}", buf);
}
if (queuedBuffer == null) {
- queuedBuffer =
channel.alloc().buffer(buf.readableBytes());
+ queuedBuffer = channel.alloc().compositeBuffer();
}
- queuedBuffer.writeBytes(buf);
+ appendToQueuedBuffer(buf.retainedSlice(buf.readerIndex(),
buf.readableBytes()));
if (LOG.isTraceEnabled()) {
LOG.trace("Copy is {}", queuedBuffer);
LOG.trace("0x{} queuedBuffer {}",
@@ -381,9 +400,9 @@ public class NettyServerCnxn extends ServerCnxn {
releaseQueuedBuffer();
} else {
LOG.debug("Processed queue - bytes remaining");
- // Possibly reduce memory consumption by freeing up buffer
space
+ // Try to reduce memory consumption by freeing up buffer space
// which is no longer needed.
- queuedBuffer.discardSomeReadBytes();
+ queuedBuffer.discardReadComponents();
}
} else {
LOG.debug("queue empty");
