Skip to content

Release ByteBuf when handle onData failed #13102

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
EarthChen merged 8 commits into from
Oct 10, 2023
Merged

Release ByteBuf when handle onData failed #13102

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
d1e8a6d
Release ByteBuf when handle onData failed
finefuture Sep 22, 2023
4ae0bbe
Runnable with clean function
finefuture Sep 26, 2023
bde8a97
Fix license
finefuture Sep 26, 2023
64b7d22
Merge branch '3.2' into 3.2-fix-netty-memory-leak
EarthChen Sep 27, 2023
cfdc986
Release ByteBuf in onData method, and refactor some sonar issue.
finefuture Oct 9, 2023
eb7a1d3
Remove unused dependency
finefuture Oct 9, 2023
f10d41e
Merge branch '3.2' into 3.2-fix-netty-memory-leak
finefuture Oct 9, 2023
1f29b39
Merge branch '3.2' into 3.2-fix-netty-memory-leak
AlbumenJ Oct 10, 2023
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
13 changes: 10 additions & 3 deletions ...o-common/src/main/java/org/apache/dubbo/common/threadpool/serial/SerializingExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,11 +60,15 @@ public SerializingExecutor(Executor executor) {
*/
@Override
public void execute(Runnable r) {
execute(r, null);
}

public void execute(Runnable r, Runnable cleanIfFailed) {
runQueue.add(r);
schedule(r);
schedule(r, cleanIfFailed);
}

private void schedule(Runnable removable) {
private void schedule(Runnable removable, Runnable cleanIfFailed) {
if (atomicBoolean.compareAndSet(false, true)) {
boolean success = false;
try {
Expand All @@ -86,6 +90,9 @@ private void schedule(Runnable removable) {
// to execute don't succeed and accidentally run a previous runnable.
runQueue.remove(removable);
}
if (cleanIfFailed != null) {
cleanIfFailed.run();
}
atomicBoolean.set(false);
}
}
Expand All @@ -111,7 +118,7 @@ public void run() {
}
if (!runQueue.isEmpty()) {
// we didn't enqueue anything but someone else did.
schedule(null);
schedule(null, null);
}
}
}
2 changes: 1 addition & 1 deletion ...bbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/AbstractStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@
*/
public abstract class AbstractStream implements Stream {

protected Executor executor;
protected SerializingExecutor executor;
protected final FrameworkModel frameworkModel;


Expand Down
34 changes: 18 additions & 16 deletions ...rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleClientStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -449,23 +449,25 @@ public void onHeader(Http2Headers headers, boolean endStream) {

@Override
public void onData(ByteBuf data, boolean endStream) {
executor.execute(() -> {
if (transportError != null) {
transportError.appendDescription(
"Data:" + data.toString(StandardCharsets.UTF_8));
ReferenceCountUtil.release(data);
if (transportError.description.length() > 512 || endStream) {
handleH2TransportError(transportError);
}
return;
}
if (!headerReceived) {
handleH2TransportError(TriRpcStatus.INTERNAL.withDescription(
"headers not received before payload"));
return;
executor.execute(() -> doOnData(data, endStream), () -> ReferenceCountUtil.release(data));
Copy link
Member

@EarthChen EarthChen Sep 26, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But is there anything better here?

Copy link
Member

@EarthChen EarthChen Sep 26, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Still using Executor

Copy link
Contributor Author

@finefuture finefuture Sep 26, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is SerializingExecutor, not a real Executor, should replace method name?

Copy link
Member

@EarthChen EarthChen Sep 26, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What I mean is that the failed operation can not destroy the serialExecutor interface?

Copy link
Contributor Author

@finefuture finefuture Sep 26, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe i can make CloseableRunnable, if execute failed, clear resource

Copy link
Member

@EarthChen EarthChen Sep 26, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

maybe，go ahead

Copy link
Contributor Author

@finefuture finefuture Sep 26, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Resubmitted, please review.

Copy link
Member

@EarthChen EarthChen Sep 26, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it possible to reuse jdk's closeable?

Copy link
Contributor Author

@finefuture finefuture Sep 26, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

image
It would throw a IOException, and SerializingExecutor need handle it, but SerializingExecutor shouldn't do the job

Copy link
Member

@EarthChen EarthChen Sep 27, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok.fine

}

private void doOnData(ByteBuf data, boolean endStream) {
if (transportError != null) {
transportError.appendDescription(
"Data:" + data.toString(StandardCharsets.UTF_8));
ReferenceCountUtil.release(data);
if (transportError.description.length() > 512 || endStream) {
handleH2TransportError(transportError);
}
deframer.deframe(data);
});
return;
}
if (!headerReceived) {
handleH2TransportError(TriRpcStatus.INTERNAL.withDescription(
"headers not received before payload"));
return;
}
deframer.deframe(data);
}

@Override
Expand Down
3 changes: 2 additions & 1 deletion ...rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/TripleServerStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

package org.apache.dubbo.rpc.protocol.tri.stream;

import io.netty.util.ReferenceCountUtil;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.logger.Logger;
Expand Down Expand Up @@ -431,7 +432,7 @@ private void processHeader(Http2Headers headers, boolean endStream) {

@Override
public void onData(ByteBuf data, boolean endStream) {
executor.execute(() -> doOnData(data, endStream));
executor.execute(() -> doOnData(data, endStream), () -> ReferenceCountUtil.release(data));
}

private void doOnData(ByteBuf data, boolean endStream) {
Expand Down