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

FastLzFrameDecoder should use allocator to allocate output buffer (#11499)

Browse Source 
Motivation:

FastLzFrameDecoder currently not use the allocator to alocate the output buffer. This means that if you use the PooledByteBufAllocator you still can't make use of the pooling. Beside this the decoder also does an uncessary memory copy when no compression is used.

Modifications:

- Allocate the output buffer via the allocator
- Don't allocate and copy if we handle an uncompressed chunk
- Make use of ByteBufChecksum for a few optimizations when running on a recent JDK

Result:

Less allocations when using FastLzFrameDecoder
This commit is contained in:
Norman Maurer 2021-07-21 22:09:12 +02:00 committed by GitHub
parent 3859805126
commit b14dfbb74c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 49 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
codec/src/main/java/io/netty/handler/codec/compression/FastLzFrameDecoder.java
View File

@ -16,10 +16,8 @@
package io.netty.handler.codec.compression; package io.netty.handler.codec.compression;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder; import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.util.internal.EmptyArrays;
import java.util.List; import java.util.List;
import java.util.zip.Adler32; import java.util.zip.Adler32;
@ -48,7 +46,7 @@ public class FastLzFrameDecoder extends ByteToMessageDecoder {
/** /**
* Underlying checksum calculator in use. * Underlying checksum calculator in use.
*/ */
private final Checksum checksum; private final ByteBufChecksum checksum;
/** /**
* Length of current received chunk of data. * Length of current received chunk of data.
@ -105,7 +103,7 @@ public class FastLzFrameDecoder extends ByteToMessageDecoder {
* You may set {@code null} if you do not want to validate checksum of each block. * You may set {@code null} if you do not want to validate checksum of each block.
*/ */
public FastLzFrameDecoder(Checksum checksum) { public FastLzFrameDecoder(Checksum checksum) {
this.checksum = checksum; this.checksum = checksum == null ? null : ByteBufChecksum.wrapChecksum(checksum);
} }
@Override @Override
@ -147,50 +145,61 @@ public class FastLzFrameDecoder extends ByteToMessageDecoder {
final int idx = in.readerIndex(); final int idx = in.readerIndex();
final int originalLength = this.originalLength; final int originalLength = this.originalLength;
final byte[] output = originalLength == 0? EmptyArrays.EMPTY_BYTES : new byte[originalLength]; ByteBuf output = null;
final int outputPtr = 0;
if (isCompressed) { try {
final byte[] input; if (isCompressed) {
final int inputPtr; final byte[] input;
if (in.hasArray()) { final int inputOffset;
input = in.array(); if (in.hasArray()) {
inputPtr = in.arrayOffset() + idx; input = in.array();
inputOffset = in.arrayOffset() + idx;
} else {
input = new byte[chunkLength];
in.getBytes(idx, input);
inputOffset = 0;
}
output = ctx.alloc().heapBuffer(originalLength);
int outputOffset = output.arrayOffset() + output.writerIndex();
final int decompressedBytes = decompress(input, inputOffset, chunkLength,
output.array(), outputOffset, originalLength);
if (originalLength != decompressedBytes) {
throw new DecompressionException(String.format(
"stream corrupted: originalLength(%d) and actual length(%d) mismatch",
originalLength, decompressedBytes));
}
output.writerIndex(output.writerIndex() + decompressedBytes);
} else { } else {
input = new byte[chunkLength]; output = in.retainedSlice(idx, chunkLength);
in.getBytes(idx, input);
inputPtr = 0;
} }
final int decompressedBytes = decompress(input, inputPtr, chunkLength, final ByteBufChecksum checksum = this.checksum;
output, outputPtr, originalLength); if (hasChecksum && checksum != null) {
if (originalLength != decompressedBytes) { checksum.reset();
throw new DecompressionException(String.format( checksum.update(output, output.readerIndex(), output.readableBytes());
"stream corrupted: originalLength(%d) and actual length(%d) mismatch", final int checksumResult = (int) checksum.getValue();
originalLength, decompressedBytes)); if (checksumResult != currentChecksum) {
throw new DecompressionException(String.format(
"stream corrupted: mismatching checksum: %d (expected: %d)",
checksumResult, currentChecksum));
}
} }
} else {
in.getBytes(idx, output, outputPtr, chunkLength);
}
final Checksum checksum = this.checksum; if (output.readableBytes() > 0) {
if (hasChecksum && checksum != null) { out.add(output);
checksum.reset(); } else {
checksum.update(output, outputPtr, originalLength); output.release();
final int checksumResult = (int) checksum.getValue(); }
if (checksumResult != currentChecksum) { output = null;
throw new DecompressionException(String.format( in.skipBytes(chunkLength);
"stream corrupted: mismatching checksum: %d (expected: %d)",
checksumResult, currentChecksum)); currentState = State.INIT_BLOCK;
} finally {
if (output != null) {
output.release();
} }
} }
if (output.length > 0) {
out.add(Unpooled.wrappedBuffer(output).writerIndex(originalLength));
}
in.skipBytes(chunkLength);
currentState = State.INIT_BLOCK;
break; break;
case CORRUPTED: case CORRUPTED:
in.skipBytes(in.readableBytes()); in.skipBytes(in.readableBytes());