lhotari commented on code in PR #4780:
URL: https://github.com/apache/bookkeeper/pull/4780#discussion_r3206312868
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java:
##########
@@ -710,26 +690,21 @@ void writeLac(final long ledgerId, final byte[]
masterKey, final long lac, ByteB
ctx, ledgerId, this));
// Build the request
- BKPacketHeader.Builder headerBuilder = BKPacketHeader.newBuilder()
+ Request writeLacRequest = new Request();
+ writeLacRequest.setHeader()
.setVersion(ProtocolVersion.VERSION_THREE)
.setOperation(OperationType.WRITE_LAC)
.setTxnId(txnId);
- ByteString body = ByteStringUtil.byteBufListToByteString(toSend);
- toSend.retain();
- Runnable cleanupActionFailedBeforeWrite = toSend::release;
- Runnable cleanupActionAfterWrite = cleanupActionFailedBeforeWrite;
- WriteLacRequest.Builder writeLacBuilder = WriteLacRequest.newBuilder()
+ ByteBuf body = ByteBufList.coalesce(toSend);
Review Comment:
Use the suggested `ByteBufList.toByteBuf` (introduced in the `addEntry`
method comment) instead of the existing `ByteBufList.coalesce` method which
isn't optimized (the method is annotated with `@VisibleForTesting` ) and
doesn't handle `ByteBufList` reference counting.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java:
##########
@@ -813,38 +784,36 @@ void addEntry(final long ledgerId, byte[] masterKey,
final long entryId, Referen
completionKey = new TxnCompletionKey(txnId,
OperationType.ADD_ENTRY);
// Build the request and calculate the total size to be included
in the packet.
- BKPacketHeader.Builder headerBuilder = BKPacketHeader.newBuilder()
+ Request addEntryRequest = new Request();
+ BKPacketHeader header = addEntryRequest.setHeader()
.setVersion(ProtocolVersion.VERSION_THREE)
.setOperation(OperationType.ADD_ENTRY)
.setTxnId(txnId);
if (((short) options & BookieProtocol.FLAG_HIGH_PRIORITY) ==
BookieProtocol.FLAG_HIGH_PRIORITY) {
- headerBuilder.setPriority(DEFAULT_HIGH_PRIORITY_VALUE);
+ header.setPriority(DEFAULT_HIGH_PRIORITY_VALUE);
}
ByteBufList bufToSend = (ByteBufList) toSend;
- ByteString body =
ByteStringUtil.byteBufListToByteString(bufToSend);
- bufToSend.retain();
- cleanupActionFailedBeforeWrite = bufToSend::release;
+ ByteBuf body = ByteBufList.coalesce(bufToSend);
Review Comment:
This causes a performance regression since `ByteBufList.coalesce` is
implemented using an unpooled heap buffer. The method is marked with
`@VisibleForTesting` which signals that it's only meant for testing.
It could be useful to implement a new instance method in `ByteBufList`:
```java
/**
* Returns a coalesced {@link ByteBuf} containing all buffers in this
list, and
* decrements this {@code ByteBufList}'s reference count.
*
* <p>Ownership semantics: the returned {@link ByteBuf} owns a reference
to the
* underlying buffer data. The caller is responsible for releasing the
returned
* {@link ByteBuf} when it is no longer needed. This {@code ByteBufList}
is
* released as part of this call.
*
* <p>If this list is empty, {@link Unpooled#EMPTY_BUFFER} is returned.
* If it contains exactly one buffer, that buffer is returned directly
(with its
* reference count transferred to the caller) to avoid the overhead of
wrapping
* it in a {@link CompositeByteBuf}.
*
* @param allocator the {@link ByteBufAllocator} used to allocate the
* {@link CompositeByteBuf} when more than one buffer
is present
* @return a {@link ByteBuf} containing the coalesced contents of this
list
*/
public ByteBuf toByteBuf(ByteBufAllocator allocator) {
final int size = buffers.size();
if (size == 0) {
release();
return Unpooled.EMPTY_BUFFER;
}
if (size == 1) {
// Fast path: avoid wrapping a single buffer in a
CompositeByteBuf.
// Retain so the buffer survives our release() below; ownership
is
// transferred to the caller.
ByteBuf single = buffers.get(0).retain();
release();
return single;
}
CompositeByteBuf composite = allocator.compositeBuffer(size);
for (int i = 0; i < size; i++) {
// Buffers need to be retained because ownership is handed over
to
// the CompositeByteBuf, while this ByteBufList continues to hold
// its own references until release() below.
composite.addComponent(true, buffers.get(i).retain());
}
release();
return composite;
}
```
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadLacProcessorV3.java:
##########
@@ -66,7 +61,7 @@ private ReadLacResponse getReadLacResponse() {
try {
lac = requestProcessor.bookie.getExplicitLac(ledgerId);
if (lac != null) {
-
readLacResponse.setLacBody(ByteString.copyFrom(lac.nioBuffer()));
+ readLacResponse.setLacBody(ByteBufUtil.getBytes(lac));
Review Comment:
Pass ByteBuf directly? Also ensure whether .retain() also needs to be called.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadEntryProcessorV3.java:
##########
@@ -180,7 +175,7 @@ protected ReadResponse readEntry(ReadResponse.Builder
readResponseBuilder,
return null;
} else {
try {
-
readResponseBuilder.setBody(ByteString.copyFrom(entryBody.nioBuffer()));
+ readResponseBuilder.setBody(ByteBufUtil.getBytes(entryBody));
Review Comment:
Pass ByteBuf directly?
Special attention is needed for the ref count in this case since
ByteString.copyFrom is used in the previous code. Does .retain() need to be
called?
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadEntryProcessorV3.java:
##########
@@ -310,7 +304,7 @@ private void getFenceResponse(ReadResponse.Builder
readResponse,
registerFailedEvent(requestProcessor.getRequestStats().getFenceReadWaitStats(),
lastPhaseStartTime);
} else {
status = StatusCode.EOK;
- readResponse.setBody(ByteString.copyFrom(entryBody.nioBuffer()));
+ readResponse.setBody(ByteBufUtil.getBytes(entryBody));
Review Comment:
Pass ByteBuf directly? Also ensure whether .retain() also needs to be called.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ReadLacProcessorV3.java:
##########
@@ -93,7 +88,7 @@ private ReadLacResponse getReadLacResponse() {
try {
lastEntry = requestProcessor.bookie.readEntry(ledgerId,
BookieProtocol.LAST_ADD_CONFIRMED);
if (lastEntry != null) {
-
readLacResponse.setLastEntryBody(ByteString.copyFrom(lastEntry.nioBuffer()));
+
readLacResponse.setLastEntryBody(ByteBufUtil.getBytes(lastEntry));
Review Comment:
Pass ByteBuf directly? Also ensure whether .retain() also needs to be called.
##########
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndex.java:
##########
@@ -388,8 +387,8 @@ boolean setStorageStateFlags(int expected, int newFlags)
throws IOException {
void setExplicitLac(long ledgerId, ByteBuf lac) throws IOException {
LedgerData ledgerData = ledgers.get(ledgerId);
if (ledgerData != null) {
- LedgerData newLedgerData = LedgerData.newBuilder(ledgerData)
-
.setExplicitLac(ByteString.copyFrom(lac.nioBuffer())).build();
+ LedgerData newLedgerData = new LedgerData().copyFrom(ledgerData)
+ .setExplicitLac(ByteBufUtil.getBytes(lac));
Review Comment:
Pass ByteBuf directly?
Special attention is needed for the ref count in this case since
ByteString.copyFrom is used in the previous code. Does .retain() need to be
called?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
