Skip to content

Optimize ReadResponse for small entry sizes #3597

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merlimat merged 2 commits into from
Nov 2, 2022
Merged

Optimize ReadResponse for small entry sizes #3597

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0b38fbc
Optimize ReadResponse for small entry sizes
merlimat Oct 18, 2022
e48e521
Fixed checkstyle
merlimat Nov 2, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 32 additions & 17 deletions bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieProtoEncoding.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.ByteBufInputStream;
import io.netty.buffer.ByteBufOutputStream;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
Expand All @@ -51,6 +50,13 @@
public class BookieProtoEncoding {
private static final Logger LOG = LoggerFactory.getLogger(BookieProtoEncoding.class);

/**
* Threshold under which an entry is considered to be "small".
*
* Small entries payloads are copied instead of being passed around as references.
*/
public static final int SMALL_ENTRY_SIZE_THRESHOLD = 16 * 1024;

/**
* An encoder/decoder interface for the Bookkeeper protocol.
*/
Expand Down Expand Up @@ -246,38 +252,47 @@ public Object encode(Object msg, ByteBufAllocator allocator)
return msg;
}
BookieProtocol.Response r = (BookieProtocol.Response) msg;
ByteBuf buf = allocator.buffer(RESPONSE_HEADERS_SIZE + 4 /* frame size */);
buf.writerIndex(4); // Leave the placeholder for the frame size
buf.writeInt(PacketHeader.toInt(r.getProtocolVersion(), r.getOpCode(), (short) 0));

try {
if (msg instanceof BookieProtocol.ReadResponse) {
BookieProtocol.ReadResponse rr = (BookieProtocol.ReadResponse) r;
int payloadSize = rr.getData().readableBytes();
int responseSize = RESPONSE_HEADERS_SIZE + payloadSize;
boolean isSmallEntry = payloadSize < SMALL_ENTRY_SIZE_THRESHOLD;

int bufferSize = 4 /* frame size */ + RESPONSE_HEADERS_SIZE
+ (isSmallEntry ? payloadSize : 0);
ByteBuf buf = allocator.buffer(bufferSize);
buf.writeInt(responseSize);
buf.writeInt(PacketHeader.toInt(r.getProtocolVersion(), r.getOpCode(), (short) 0));
buf.writeInt(r.getErrorCode());
buf.writeLong(r.getLedgerId());
buf.writeLong(r.getEntryId());

BookieProtocol.ReadResponse rr = (BookieProtocol.ReadResponse) r;
if (rr.hasData()) {
int frameSize = RESPONSE_HEADERS_SIZE + rr.getData().readableBytes();
buf.setInt(0, frameSize);
return ByteBufList.get(buf, rr.getData());
} else {
buf.setInt(0, RESPONSE_HEADERS_SIZE); // Frame size
if (isSmallEntry) {
buf.writeBytes(rr.getData());
return buf;
} else {
return ByteBufList.get(buf, rr.getData());
}
} else if (msg instanceof BookieProtocol.AddResponse) {
ByteBuf buf = allocator.buffer(RESPONSE_HEADERS_SIZE + 4 /* frame size */);
buf.writeInt(RESPONSE_HEADERS_SIZE);
buf.writeInt(PacketHeader.toInt(r.getProtocolVersion(), r.getOpCode(), (short) 0));
buf.writeInt(r.getErrorCode());
buf.writeLong(r.getLedgerId());
buf.writeLong(r.getEntryId());
buf.setInt(0, RESPONSE_HEADERS_SIZE); // Frame size

return buf;
} else if (msg instanceof BookieProtocol.AuthResponse) {
BookkeeperProtocol.AuthMessage am = ((BookieProtocol.AuthResponse) r).getAuthMessage();
ByteBuf payload = Unpooled.wrappedBuffer(am.toByteArray());
int frameSize = 4 + payload.readableBytes();
buf.setInt(0, frameSize);
return ByteBufList.get(buf, payload);
int payloadSize = 4 + am.getSerializedSize();
int bufferSize = payloadSize + 4 /* frame size */;

ByteBuf buf = allocator.buffer(bufferSize);
buf.writeInt(payloadSize);
buf.writeInt(PacketHeader.toInt(r.getProtocolVersion(), r.getOpCode(), (short) 0));
buf.writeBytes(am.toByteArray());
return buf;
} else {
LOG.error("Cannot encode unknown response type {}", msg.getClass().getName());
return msg;
Expand Down