Skip to content

Commit 7860144

Browse files
authored
mem: ensure Reader buffers are reused on early return (grpc#8886)
If there was an error, e.g. while decompressing the data, we still want to return all buffers to the pool. RELEASE NOTES: None
1 parent 0f69b6e commit 7860144

1 file changed

Lines changed: 16 additions & 6 deletions

File tree

rpc_util.go

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -961,24 +961,32 @@ func recvAndDecompress(p *parser, s recvCompressor, dc Decompressor, maxReceiveM
961961
return out, nil
962962
}
963963

964-
// decompress processes the given data by decompressing it using either a custom decompressor or a standard compressor.
965-
// If a custom decompressor is provided, it takes precedence. The function validates that the decompressed data
966-
// does not exceed the specified maximum size and returns an error if this limit is exceeded.
967-
// On success, it returns the decompressed data. Otherwise, it returns an error if decompression fails or the data exceeds the size limit.
964+
// decompress processes the given data by decompressing it using either
965+
// a custom decompressor or a standard compressor. If a custom decompressor
966+
// is provided, it takes precedence. The function validates that
967+
// the decompressed data does not exceed the specified maximum size and returns
968+
// an error if this limit is exceeded. On success, it returns the decompressed
969+
// data. Otherwise, it returns an error if decompression fails or the data
970+
// exceeds the size limit.
968971
func decompress(compressor encoding.Compressor, d mem.BufferSlice, dc Decompressor, maxReceiveMessageSize int, pool mem.BufferPool) (mem.BufferSlice, error) {
969972
if dc != nil {
970-
uncompressed, err := dc.Do(d.Reader())
973+
r := d.Reader()
974+
uncompressed, err := dc.Do(r)
971975
if err != nil {
976+
r.Close() // ensure buffers are reused
972977
return nil, status.Errorf(codes.Internal, "grpc: failed to decompress the received message: %v", err)
973978
}
974979
if len(uncompressed) > maxReceiveMessageSize {
980+
r.Close() // ensure buffers are reused
975981
return nil, status.Errorf(codes.ResourceExhausted, "grpc: message after decompression larger than max (%d vs. %d)", len(uncompressed), maxReceiveMessageSize)
976982
}
977983
return mem.BufferSlice{mem.SliceBuffer(uncompressed)}, nil
978984
}
979985
if compressor != nil {
980-
dcReader, err := compressor.Decompress(d.Reader())
986+
r := d.Reader()
987+
dcReader, err := compressor.Decompress(r)
981988
if err != nil {
989+
r.Close() // ensure buffers are reused
982990
return nil, status.Errorf(codes.Internal, "grpc: failed to decompress the message: %v", err)
983991
}
984992

@@ -990,11 +998,13 @@ func decompress(compressor encoding.Compressor, d mem.BufferSlice, dc Decompress
990998
}
991999
out, err := mem.ReadAll(dcReader, pool)
9921000
if err != nil {
1001+
r.Close() // ensure buffers are reused
9931002
out.Free()
9941003
return nil, status.Errorf(codes.Internal, "grpc: failed to read decompressed data: %v", err)
9951004
}
9961005

9971006
if out.Len() > maxReceiveMessageSize {
1007+
r.Close() // ensure buffers are reused
9981008
out.Free()
9991009
return nil, status.Errorf(codes.ResourceExhausted, "grpc: received message after decompression larger than max %d", maxReceiveMessageSize)
10001010
}

0 commit comments

Comments
 (0)