From 80df75d72fbdaa5aac434b3a7c7a7bfe04ce816b Mon Sep 17 00:00:00 2001 From: larry-safran Date: Tue, 5 Sep 2023 16:34:49 -0700 Subject: [PATCH 1/5] netty: Eliminate buffer release when there is an error as caller still owns the buffer. --- .../src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java | 4 ---- .../test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java | 6 ++++-- 2 files changed, 4 insertions(+), 6 deletions(-) diff --git a/netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java b/netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java index b3a28c55c79..be846cedc9b 100644 --- a/netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java +++ b/netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java @@ -210,10 +210,6 @@ static void mergeWithCompositeTail( // the input and the new tail so that finally block can handles them properly. composite.readerIndex(prevReader); } finally { - // Input buffer was merged with the tail. - if (in != null) { - in.release(); - } // If new tail's ownership isn't transferred to the composite buf. // Release it to prevent a leak. if (newTail != null) { diff --git a/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java b/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java index 6a0c00bac0e..feda8036c67 100644 --- a/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java +++ b/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java @@ -537,7 +537,7 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, } catch (UnsupportedOperationException actualError) { assertSame(expectedError, actualError); // Input must be released unless its ownership has been to the composite cumulation. - assertEquals(0, in.refCnt()); + assertEquals(1, in.refCnt()); // Tail released assertEquals(0, tail.refCnt()); // Composite cumulation is retained @@ -546,6 +546,7 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, assertEquals(0, compositeThrows.numComponents()); } finally { compositeThrows.release(); + in.release(); } } @@ -573,7 +574,7 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, } catch (UnsupportedOperationException actualError) { assertSame(expectedError, actualError); // Input must be released unless its ownership has been to the composite cumulation. - assertEquals(0, in.refCnt()); + assertEquals(1, in.refCnt()); // New buffer released assertEquals(0, newTail.refCnt()); // Composite cumulation is retained @@ -582,6 +583,7 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, assertEquals(0, compositeRo.numComponents()); } finally { compositeRo.release(); + in.release(); } } } From 426f1e60cd20e5f7f1b55e5b3f4861dcf7d0d348 Mon Sep 17 00:00:00 2001 From: larry-safran Date: Wed, 6 Sep 2023 14:44:26 -0700 Subject: [PATCH 2/5] Move resetting the composite reader index to before the release as it that fails then the operation hasn't completed correctly and the ownership of the buffer should remain with the caller. --- .../src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java | 5 +---- .../test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java | 4 ++-- 2 files changed, 3 insertions(+), 6 deletions(-) diff --git a/netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java b/netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java index be846cedc9b..aa2a2a3e769 100644 --- a/netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java +++ b/netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java @@ -204,11 +204,8 @@ static void mergeWithCompositeTail( composite.addFlattenedComponents(true, newTail); // New tail's ownership transferred to the composite buf. newTail = null; - in.release(); - in = null; - // Restore the reader. In case it fails we restore the reader after releasing/forgetting - // the input and the new tail so that finally block can handles them properly. composite.readerIndex(prevReader); + in.release(); // Successful so took over ownership of in } finally { // If new tail's ownership isn't transferred to the composite buf. // Release it to prevent a leak. diff --git a/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java b/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java index feda8036c67..f507df5c99d 100644 --- a/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java +++ b/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java @@ -536,7 +536,7 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, fail("Cumulator didn't throw"); } catch (UnsupportedOperationException actualError) { assertSame(expectedError, actualError); - // Input must be released unless its ownership has been to the composite cumulation. + // Because of error, ownership shouldn't have changed so should not have been released. assertEquals(1, in.refCnt()); // Tail released assertEquals(0, tail.refCnt()); @@ -573,7 +573,7 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, fail("Cumulator didn't throw"); } catch (UnsupportedOperationException actualError) { assertSame(expectedError, actualError); - // Input must be released unless its ownership has been to the composite cumulation. + // Because of error, ownership shouldn't have changed so should not have been released. assertEquals(1, in.refCnt()); // New buffer released assertEquals(0, newTail.refCnt()); From 6d97e70c7fb05f11d31252a3e7404137ad53280c Mon Sep 17 00:00:00 2001 From: Larry Safran Date: Wed, 13 Sep 2023 17:53:23 -0700 Subject: [PATCH 3/5] Update netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java Co-authored-by: Sergii Tkachenko --- netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java b/netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java index aa2a2a3e769..58eabb2cf8d 100644 --- a/netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java +++ b/netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java @@ -205,7 +205,9 @@ static void mergeWithCompositeTail( // New tail's ownership transferred to the composite buf. newTail = null; composite.readerIndex(prevReader); - in.release(); // Successful so took over ownership of in + // Input buffer was successfully merged with the tail. + // Must be the last line in the try because we release it only on success. + in.release(); } finally { // If new tail's ownership isn't transferred to the composite buf. // Release it to prevent a leak. From 4696ad586f5c29a3d814b46b8988c3d9c5831b48 Mon Sep 17 00:00:00 2001 From: larry-safran Date: Thu, 14 Sep 2023 14:41:50 -0700 Subject: [PATCH 4/5] Move release in tests from finally to catch block. --- .../java/io/grpc/netty/NettyAdaptiveCumulatorTest.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java b/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java index f507df5c99d..5cbf37d856b 100644 --- a/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java +++ b/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java @@ -537,7 +537,9 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, } catch (UnsupportedOperationException actualError) { assertSame(expectedError, actualError); // Because of error, ownership shouldn't have changed so should not have been released. - assertEquals(1, in.refCnt()); + int inRefCnt = in.refCnt(); + in.release(); + assertEquals(1, inRefCnt); // Tail released assertEquals(0, tail.refCnt()); // Composite cumulation is retained @@ -546,7 +548,6 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, assertEquals(0, compositeThrows.numComponents()); } finally { compositeThrows.release(); - in.release(); } } @@ -572,9 +573,11 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, NettyAdaptiveCumulator.mergeWithCompositeTail(mockAlloc, compositeRo, in); fail("Cumulator didn't throw"); } catch (UnsupportedOperationException actualError) { + int inRefCnt = in.refCnt(); + in.release(); assertSame(expectedError, actualError); // Because of error, ownership shouldn't have changed so should not have been released. - assertEquals(1, in.refCnt()); + assertEquals(1, inRefCnt); // New buffer released assertEquals(0, newTail.refCnt()); // Composite cumulation is retained @@ -583,7 +586,6 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, assertEquals(0, compositeRo.numComponents()); } finally { compositeRo.release(); - in.release(); } } } From 9a46d9ed77464853bc9ce051cd70b6c16e8f04c8 Mon Sep 17 00:00:00 2001 From: larry-safran Date: Thu, 14 Sep 2023 14:50:34 -0700 Subject: [PATCH 5/5] Move release back to finally block in tests, but skip it if the mergeWithCompsiteTail unexpectedly succeeded. --- .../grpc/netty/NettyAdaptiveCumulatorTest.java | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java b/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java index 5cbf37d856b..0e524fd1c91 100644 --- a/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java +++ b/netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java @@ -533,13 +533,12 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, try { NettyAdaptiveCumulator.mergeWithCompositeTail(alloc, compositeThrows, in); + in = null; // On success it would be released fail("Cumulator didn't throw"); } catch (UnsupportedOperationException actualError) { assertSame(expectedError, actualError); // Because of error, ownership shouldn't have changed so should not have been released. - int inRefCnt = in.refCnt(); - in.release(); - assertEquals(1, inRefCnt); + assertEquals(1, in.refCnt()); // Tail released assertEquals(0, tail.refCnt()); // Composite cumulation is retained @@ -547,6 +546,9 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, // Composite cumulation loses the tail assertEquals(0, compositeThrows.numComponents()); } finally { + if (in != null) { + in.release(); + } compositeThrows.release(); } } @@ -571,13 +573,12 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, try { NettyAdaptiveCumulator.mergeWithCompositeTail(mockAlloc, compositeRo, in); + in = null; // On success it would be released fail("Cumulator didn't throw"); } catch (UnsupportedOperationException actualError) { - int inRefCnt = in.refCnt(); - in.release(); assertSame(expectedError, actualError); // Because of error, ownership shouldn't have changed so should not have been released. - assertEquals(1, inRefCnt); + assertEquals(1, in.refCnt()); // New buffer released assertEquals(0, newTail.refCnt()); // Composite cumulation is retained @@ -585,6 +586,9 @@ public CompositeByteBuf addFlattenedComponents(boolean increaseWriterIndex, // Composite cumulation loses the tail assertEquals(0, compositeRo.numComponents()); } finally { + if (in != null) { + in.release(); + } compositeRo.release(); } }