You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@dubbo.apache.org by "EarthChen (via GitHub)" <gi...@apache.org> on 2025/12/30 11:11:21 UTC
[PR] Support Tri backpress [dubbo]
EarthChen opened a new pull request, #15957:
URL: https://github.com/apache/dubbo/pull/15957
## What is the purpose of the change?
## Checklist
- [x] Make sure there is a [GitHub_issue](https://github.com/apache/dubbo/issues) field for the change.
- [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
- [x] Write necessary unit-test to verify your logic correction. If the new feature or significant change is committed, please remember to add sample in [dubbo samples](https://github.com/apache/dubbo-samples) project.
- [x] Make sure gitHub actions can pass. [Why the workflow is failing and how to fix it?](../CONTRIBUTING.md)
--
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: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] Support Tri backpress [dubbo]
Posted by "wangchengming666 (via GitHub)" <gi...@apache.org>.
wangchengming666 commented on PR #15957:
URL: https://github.com/apache/dubbo/pull/15957#issuecomment-3705739901
sample is https://github.com/apache/dubbo-samples/pull/1278
--
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: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] Support Tri backpress [dubbo]
Posted by "EarthChen (via GitHub)" <gi...@apache.org>.
EarthChen commented on PR #15957:
URL: https://github.com/apache/dubbo/pull/15957#issuecomment-3699782575
## 核心概念
### 为什么需要手动流控？
HTTP/2 协议使用**流控窗口**来防止接收端被发送端压垮。默认情况下，Netty 会自动管理流控：
- 收到数据后自动发送 `WINDOW_UPDATE` 帧
- 这种模式无法实现背压（Backpressure）控制
**手动流控**允许应用层控制何时发送 `WINDOW_UPDATE`：
- 只有当数据被真正处理后才返回窗口
- 实现真正的端到端背压控制
### 关键配置
| 配置项 | 描述 | 默认值 |
|--------|------|--------|
| `windowUpdateRatio` | 消费字节达到初始窗口的多少比例时发送 WINDOW_UPDATE | 0.5 (50%) |
| `initialWindowSize` | 初始流控窗口大小 | 8MB |
| `AUTO_STREAM_FLOW_CONTROL` | Netty 配置，禁用 Http2StreamChannel 自动流控 | false |
## 架构图
```
┌─────────────────────────────────────────────────────────────────────────────┐
│ HTTP/2 连接层 │
│ ┌─────────────────────────────────────────────────────────────────────┐ │
│ │ Http2FrameCodec │ │
│ │ ┌─────────────────────────────────────────────────────────────┐ │ │
│ │ │ TripleHttp2LocalFlowController │ │ │
│ │ │ - receiveFlowControlledFrame(): 减少窗口 │ │ │
│ │ │ - consumeBytes(): 返回窗口，触发 WINDOW_UPDATE │ │ │
│ │ │ - windowUpdateRatio: 控制何时发送 WINDOW_UPDATE │ │ │
│ │ └─────────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────────┘ │
│ │ │
│ Http2MultiplexHandler │
│ │ │
│ ┌─────────────────────┴─────────────────────┐ │
│ │ │ │
│ ┌───────────┴────────────┐ ┌──────────────┴───────────┐ │
│ │ Http2StreamChannel │ │ Http2StreamChannel │ │
│ │ (Stream 1) │ │ (Stream 3) │ │
│ │ AUTO_FLOW = false │ │ AUTO_FLOW = false │ │
│ └────────────────────────┘ └──────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘
```
## 服务端流控机制
### 数据流转图
```mermaid
sequenceDiagram
participant Client as 客户端
participant Netty as Netty Http2FrameCodec
participant FC as TripleHttp2LocalFlowController
participant Handler as NettyHttp2FrameHandler
participant Decoder as StreamingDecoder
participant Listener as FragmentListener
participant Channel as H2StreamChannel
participant App as 业务处理
Client->>Netty: DATA 帧
Netty->>FC: receiveFlowControlledFrame()
Note over FC: 减少窗口大小<br/>跟踪 unconsumedBytes
Netty->>Handler: onData(ByteBuf)
Handler->>Decoder: decode(inputStream)
loop 处理每个消息
Decoder->>Decoder: processBody()
Note over Decoder: 读取消息字节
Decoder->>Listener: bytesRead(numBytes)
Listener->>Channel: consumeBytes(numBytes)
Channel->>FC: consumeBytes(stream, numBytes)
Note over FC: 累加返回的字节<br/>检查是否达到阈值
alt 达到 windowUpdateRatio
FC->>Client: WINDOW_UPDATE 帧
end
Decoder->>Listener: onFragmentMessage(rawMessage)
Listener->>App: 业务处理
end
```
### 关键类与职责
#### 1. TripleHttp2Protocol（初始化）
```java
// 服务端创建连接时配置手动流控
private Http2Connection createHttp2ServerConnection(TripleConfig tripleConfig) {
Http2Connection connection = new DefaultHttp2Connection(true);
float windowUpdateRatio = tripleConfig.getWindowUpdateRatioOrDefault();
connection.local().flowController(
new TripleHttp2LocalFlowController(connection, windowUpdateRatio));
return connection;
}
// 禁用 Http2StreamChannel 自动流控
protected void initChannel(Http2StreamChannel ch) {
ch.config().setOption(Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL, false);
// ...
}
```
#### 2. TripleHttp2LocalFlowController（流控核心）
```java
public class TripleHttp2LocalFlowController extends DefaultHttp2LocalFlowController {
public TripleHttp2LocalFlowController(Http2Connection connection, float windowUpdateRatio) {
// windowUpdateRatio: 当消费字节达到初始窗口的该比例时发送 WINDOW_UPDATE
// true: 启用按帧跟踪
super(connection, windowUpdateRatio, true);
}
}
```
#### 3. GenericHttp2ServerTransportListener（服务端入口）
```java
@Override
protected HttpMessageListener buildHttpMessageListener() {
// 设置 FragmentListener 来处理流控
streamingDecoder.setFragmentListener(new DefaultFragmentListener(listeningDecoder));
return new StreamingHttpMessageListener(streamingDecoder);
}
final class DefaultFragmentListener implements FragmentListener {
@Override
public void bytesRead(int numBytes) {
// 数据被读取后，通知流控器返回窗口
getH2StreamChannel().consumeBytes(numBytes);
}
}
```
#### 4. LengthFieldStreamingDecoder / TriDecoder（解码触发）
```java
private void processBody() {
int totalBytesRead = lengthFieldOffset + lengthFieldLength + requiredLength;
byte[] rawMessage;
try {
rawMessage = readRawMessage(accumulate, requiredLength);
} finally {
// 在 finally 块中通知 bytesRead，确保流控一定被触发
listener.bytesRead(totalBytesRead);
}
// 处理消息
invokeListener(inputStream);
}
```
#### 5. NettyH2StreamChannel（执行流控）
```java
@Override
public void consumeBytes(int numBytes) throws Exception {
Http2LocalFlowController localFlowController = http2Connection.local().flowController();
int streamId = http2StreamChannel.stream().id();
Http2Stream stream = http2Connection.stream(streamId);
// 必须在 EventLoop 线程执行
if (http2StreamChannel.eventLoop().inEventLoop()) {
localFlowController.consumeBytes(stream, numBytes);
} else {
http2StreamChannel.eventLoop().execute(() -> {
localFlowController.consumeBytes(stream, numBytes);
});
}
}
```
### 服务端数据流转路径
```
收到 DATA 帧
│
▼
Http2FrameCodec.onDataRead() → 返回 0，不自动消费
│
▼
TripleHttp2LocalFlowController.receiveFlowControlledFrame()
│ - 减少流窗口
│ - 跟踪 unconsumedBytes
▼
NettyHttp2FrameHandler.onData()
│
▼
StreamingDecoder.decode()
│
▼
LengthFieldStreamingDecoder.processBody()
│
├──► 读取消息字节
│
├──► finally { listener.bytesRead(numBytes) } ◄─── 关键！在 finally 块中
│ │
│ ▼
│ DefaultFragmentListener.bytesRead()
│ │
│ ▼
│ H2StreamChannel.consumeBytes()
│ │
│ ▼
│ TripleHttp2LocalFlowController.consumeBytes()
│ │
│ ▼
│ [如果达到 windowUpdateRatio] → 发送 WINDOW_UPDATE
│
▼
listener.onFragmentMessage() → 业务处理
```
---
## 客户端流控机制
### 数据流转图
```mermaid
sequenceDiagram
participant Server as 服务端
participant Netty as Netty Http2FrameCodec
participant FC as TripleHttp2LocalFlowController
participant Handler as TripleHttp2ClientResponseHandler
participant Stream as AbstractTripleClientStream
participant Decoder as TriDecoder
participant Listener as TriDecoder.Listener
participant App as 业务回调
Server->>Netty: DATA 帧
Netty->>FC: receiveFlowControlledFrame()
Note over FC: 减少窗口大小
Netty->>Handler: onData(ByteBuf)
Handler->>Stream: H2TransportListener.onData()
Stream->>Decoder: deframer.deframe(data)
loop 处理每个消息
Decoder->>Decoder: processBody()
Note over Decoder: 读取消息字节
Decoder->>Listener: bytesRead(numBytes)
Listener->>Stream: consumeBytes(numBytes)
Stream->>FC: Http2LocalFlowController.consumeBytes()
Note over FC: 累加返回的字节
alt 达到 windowUpdateRatio
FC->>Server: WINDOW_UPDATE 帧
end
Decoder->>Listener: onRawMessage(data)
Listener->>App: 业务处理
end
```
### 关键类与职责
#### 1. TripleHttp2Protocol（客户端初始化）
```java
@Override
public void configClientPipeline(URL url, ChannelOperator operator, ContextOperator contextOperator) {
TripleConfig tripleConfig = ConfigManager.getProtocolOrDefault(url).getTripleOrDefault();
// 创建带有手动流控的连接
Http2Connection connection = createHttp2ClientConnection(tripleConfig);
// ...
}
private Http2Connection createHttp2ClientConnection(TripleConfig tripleConfig) {
Http2Connection connection = new DefaultHttp2Connection(false);
float windowUpdateRatio = tripleConfig.getWindowUpdateRatioOrDefault();
connection.local().flowController(
new TripleHttp2LocalFlowController(connection, windowUpdateRatio));
return connection;
}
```
#### 2. Http2TripleClientStream（客户端流）
```java
@Override
protected TripleStreamChannelFuture initStreamChannel(Channel parent) {
Http2StreamChannelBootstrap bootstrap = new Http2StreamChannelBootstrap(parent);
// 禁用自动流控
bootstrap.option(Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL, false);
// ...
}
@Override
protected void consumeBytes(int numBytes) {
// 从 parent channel 获取 Http2Connection
Http2Connection http2Connection = getHttp2Connection();
Http2LocalFlowController localFlowController = http2Connection.local().flowController();
// 调用流控器消费字节
localFlowController.consumeBytes(stream, numBytes);
}
private Http2Connection getHttp2Connection() {
ChannelHandlerContext ctx = parent.pipeline().context(Http2FrameCodec.class);
Http2FrameCodec codec = (Http2FrameCodec) ctx.handler();
return codec.connection();
}
```
#### 3. AbstractTripleClientStream.ClientTransportListener（设置解码监听器）
```java
void onHeaderReceived(Http2Headers headers) {
// 创建解码器监听器
TriDecoder.Listener listener = new TriDecoder.Listener() {
@Override
public void bytesRead(int numBytes) {
// 调用抽象方法，由子类实现
consumeBytes(numBytes);
}
@Override
public void onRawMessage(byte[] data) {
AbstractTripleClientStream.this.listener.onMessage(data, isReturnTriException);
}
};
deframer = new TriDecoder(decompressor, listener);
}
```
#### 4. TriDecoder（gRPC 解码器）
```java
private void processBody() {
int totalBytesRead = HEADER_LENGTH + requiredLength;
byte[] stream;
try {
stream = compressedFlag ? getCompressedBody() : getUncompressedBody();
} finally {
// 在 finally 块中通知流控
listener.bytesRead(totalBytesRead);
}
listener.onRawMessage(stream);
}
```
### 客户端数据流转路径
```
收到 DATA 帧（来自服务端响应）
│
▼
Http2FrameCodec.onDataRead() → 返回 0，不自动消费
│
▼
TripleHttp2LocalFlowController.receiveFlowControlledFrame()
│ - 减少流窗口
▼
TripleHttp2ClientResponseHandler.onData()
│
▼
ClientTransportListener.onData()
│
▼
TriDecoder.deframe(data)
│
▼
TriDecoder.processBody()
│
├──► 读取/解压消息
│
├──► finally { listener.bytesRead(numBytes) } ◄─── 关键！
│ │
│ ▼
│ TriDecoder.Listener.bytesRead()
│ │
│ ▼
│ AbstractTripleClientStream.consumeBytes()
│ │
│ ▼
│ Http2TripleClientStream.consumeBytes()
│ │
│ ▼
│ TripleHttp2LocalFlowController.consumeBytes()
│ │
│ ▼
│ [如果达到 windowUpdateRatio] → 发送 WINDOW_UPDATE
│
▼
listener.onRawMessage() → 业务回调
```
---
## 配置说明
### windowUpdateRatio 工作原理
假设 `initialWindowSize = 8MB`，`windowUpdateRatio = 0.5`：
1. 初始窗口 = 8MB
2. 接收数据，窗口减少
3. 调用 `consumeBytes()` 返回字节
4. 当累计返回 4MB (8MB × 0.5) 时，发送 `WINDOW_UPDATE` 帧
5. 远端收到后，可以继续发送数据
```
窗口状态变化：
初始: [████████████████████████████████] 8MB 可用
收到 2MB: [████████████████████████░░░░░░░░] 6MB 可用
consumeBytes(2MB) - 累计 2MB，未达阈值
收到 2MB: [████████████████░░░░░░░░░░░░░░░░] 4MB 可用
consumeBytes(2MB) - 累计 4MB，达到 50%！
→ 发送 WINDOW_UPDATE (4MB)
窗口恢复: [████████████████████████████████] 8MB 可用
```
---
## 设计要点总结
### 1. 两层禁用自动流控
| 层级 | 配置 | 作用 |
|------|------|------|
| Http2Connection | TripleHttp2LocalFlowController | 控制 WINDOW_UPDATE 发送时机 |
| Http2StreamChannel | AUTO_STREAM_FLOW_CONTROL=false | 禁止 Netty 自动发送 WINDOW_UPDATE |
### 2. bytesRead 在 finally 块中调用
确保即使消息处理失败，流控也能正常工作：
```java
try {
rawMessage = readRawMessage();
} finally {
listener.bytesRead(totalBytesRead); // 必须执行
}
```
### 3. EventLoop 线程安全
`consumeBytes()` 必须在 EventLoop 线程执行：
```java
if (eventLoop().inEventLoop()) {
localFlowController.consumeBytes(stream, numBytes);
} else {
eventLoop().execute(() -> localFlowController.consumeBytes(stream, numBytes));
}
```
---
## 类关系图
```mermaid
classDiagram
class TripleConfig {
+Float windowUpdateRatio
+Integer initialWindowSize
+getWindowUpdateRatioOrDefault() float
}
class TripleHttp2LocalFlowController {
+TripleHttp2LocalFlowController(connection, windowUpdateRatio)
+receiveFlowControlledFrame()
+consumeBytes(stream, numBytes)
}
class DefaultHttp2LocalFlowController {
+consumeBytes(stream, numBytes)
}
class H2StreamChannel {
<<interface>>
+consumeBytes(numBytes)
}
class NettyH2StreamChannel {
-Http2Connection http2Connection
+consumeBytes(numBytes)
}
class StreamingDecoder {
<<interface>>
+setFragmentListener(listener)
}
class FragmentListener {
<<interface>>
+bytesRead(numBytes)
+onFragmentMessage(rawMessage)
}
class AbstractTripleClientStream {
<<abstract>>
#consumeBytes(numBytes)*
}
class Http2TripleClientStream {
+consumeBytes(numBytes)
-getHttp2Connection()
}
TripleHttp2LocalFlowController --|> DefaultHttp2LocalFlowController
NettyH2StreamChannel ..|> H2StreamChannel
Http2TripleClientStream --|> AbstractTripleClientStream
NettyH2StreamChannel --> TripleHttp2LocalFlowController : uses
Http2TripleClientStream --> TripleHttp2LocalFlowController : uses
```
---
--
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: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] Support Tri backpress [dubbo]
Posted by "zrlw (via GitHub)" <gi...@apache.org>.
zrlw merged PR #15957:
URL: https://github.com/apache/dubbo/pull/15957
--
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: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] Support Tri backpress [dubbo]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #15957:
URL: https://github.com/apache/dubbo/pull/15957#issuecomment-3701315298
## [Codecov](https://app.codecov.io/gh/apache/dubbo/pull/15957?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `64.05229%` with `55 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 58.45%. Comparing base ([`6a70f6b`](https://app.codecov.io/gh/apache/dubbo/commit/6a70f6b217781fc35cc65bb464eb611512d91a3d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`b74c57a`](https://app.codecov.io/gh/apache/dubbo/commit/b74c57aad6cd61e15173e58246727b011ee0843f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 9 commits behind head on 3.3.
| [Files with missing lines](https://app.codecov.io/gh/apache/dubbo/pull/15957?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...rotocol/tri/h12/http2/Http2TripleClientStream.java](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&filepath=dubbo-rpc%2Fdubbo-rpc-triple%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fdubbo%2Frpc%2Fprotocol%2Ftri%2Fh12%2Fhttp2%2FHttp2TripleClientStream.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvaDEyL2h0dHAyL0h0dHAyVHJpcGxlQ2xpZW50U3RyZWFtLmphdmE=) | 60.00% | [11 Missing and 5 partials :warning: ](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...emoting/http12/netty4/h2/NettyH2StreamChannel.java](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&filepath=dubbo-remoting%2Fdubbo-remoting-http12%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fdubbo%2Fremoting%2Fhttp12%2Fnetty4%2Fh2%2FNettyH2StreamChannel.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZHViYm8tcmVtb3RpbmcvZHViYm8tcmVtb3RpbmctaHR0cDEyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9kdWJiby9yZW1vdGluZy9odHRwMTIvbmV0dHk0L2gyL05ldHR5SDJTdHJlYW1DaGFubmVsLmphdmE=) | 50.00% | [8 Missing and 4 partials :warning: ](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...a/org/apache/dubbo/config/nested/TripleConfig.java](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&filepath=dubbo-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fdubbo%2Fconfig%2Fnested%2FTripleConfig.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZHViYm8tY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9kdWJiby9jb25maWcvbmVzdGVkL1RyaXBsZUNvbmZpZy5qYXZh) | 16.66% | [4 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../dubbo/rpc/protocol/tri/call/TripleClientCall.java](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&filepath=dubbo-rpc%2Fdubbo-rpc-triple%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fdubbo%2Frpc%2Fprotocol%2Ftri%2Fcall%2FTripleClientCall.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvY2FsbC9UcmlwbGVDbGllbnRDYWxsLmphdmE=) | 60.00% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...he/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&filepath=dubbo-rpc%2Fdubbo-rpc-triple%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fdubbo%2Frpc%2Fprotocol%2Ftri%2FTripleHttp2Protocol.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvVHJpcGxlSHR0cDJQcm90b2NvbC5qYXZh) | 85.00% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...2/netty4/h2/NettyHttp2ProtocolSelectorHandler.java](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&filepath=dubbo-remoting%2Fdubbo-remoting-http12%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fdubbo%2Fremoting%2Fhttp12%2Fnetty4%2Fh2%2FNettyHttp2ProtocolSelectorHandler.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZHViYm8tcmVtb3RpbmcvZHViYm8tcmVtb3RpbmctaHR0cDEyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9kdWJiby9yZW1vdGluZy9odHRwMTIvbmV0dHk0L2gyL05ldHR5SHR0cDJQcm90b2NvbFNlbGVjdG9ySGFuZGxlci5qYXZh) | 60.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...tri/h12/grpc/GrpcHttp2ServerTransportListener.java](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&filepath=dubbo-rpc%2Fdubbo-rpc-triple%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fdubbo%2Frpc%2Fprotocol%2Ftri%2Fh12%2Fgrpc%2FGrpcHttp2ServerTransportListener.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvaDEyL2dycGMvR3JwY0h0dHAyU2VydmVyVHJhbnNwb3J0TGlzdGVuZXIuamF2YQ==) | 60.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...h12/http2/GenericHttp2ServerTransportListener.java](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&filepath=dubbo-rpc%2Fdubbo-rpc-triple%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fdubbo%2Frpc%2Fprotocol%2Ftri%2Fh12%2Fhttp2%2FGenericHttp2ServerTransportListener.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvaDEyL2h0dHAyL0dlbmVyaWNIdHRwMlNlcnZlclRyYW5zcG9ydExpc3RlbmVyLmphdmE=) | 87.50% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../rpc/protocol/tri/observer/CallStreamObserver.java](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&filepath=dubbo-rpc%2Fdubbo-rpc-triple%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fdubbo%2Frpc%2Fprotocol%2Ftri%2Fobserver%2FCallStreamObserver.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvb2JzZXJ2ZXIvQ2FsbFN0cmVhbU9ic2VydmVyLmphdmE=) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ocol/tri/observer/ClientCallToObserverAdapter.java](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&filepath=dubbo-rpc%2Fdubbo-rpc-triple%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fdubbo%2Frpc%2Fprotocol%2Ftri%2Fobserver%2FClientCallToObserverAdapter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvb2JzZXJ2ZXIvQ2xpZW50Q2FsbFRvT2JzZXJ2ZXJBZGFwdGVyLmphdmE=) | 50.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| ... and [5 more](https://app.codecov.io/gh/apache/dubbo/pull/15957?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## 3.3 #15957 +/- ##
============================================
- Coverage 60.74% 58.45% -2.29%
+ Complexity 11713 15 -11698
============================================
Files 1938 1939 +1
Lines 88698 88555 -143
Branches 13387 13275 -112
============================================
- Hits 53876 51769 -2107
- Misses 29290 31102 +1812
- Partials 5532 5684 +152
```
| [Flag](https://app.codecov.io/gh/apache/dubbo/pull/15957/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [integration-tests-java21](https://app.codecov.io/gh/apache/dubbo/pull/15957/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration-tests-java8](https://app.codecov.io/gh/apache/dubbo/pull/15957/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [samples-tests-java21](https://app.codecov.io/gh/apache/dubbo/pull/15957/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [samples-tests-java8](https://app.codecov.io/gh/apache/dubbo/pull/15957/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unit-tests-java11](https://app.codecov.io/gh/apache/dubbo/pull/15957/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unit-tests-java17](https://app.codecov.io/gh/apache/dubbo/pull/15957/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unit-tests-java21](https://app.codecov.io/gh/apache/dubbo/pull/15957/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unit-tests-java25](https://app.codecov.io/gh/apache/dubbo/pull/15957/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `58.45% <64.05%> (-0.02%)` | :arrow_down: |
| [unit-tests-java8](https://app.codecov.io/gh/apache/dubbo/pull/15957/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/dubbo/pull/15957?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] Support Tri backpress [dubbo]
Posted by "zrlw (via GitHub)" <gi...@apache.org>.
zrlw commented on PR #15957:
URL: https://github.com/apache/dubbo/pull/15957#issuecomment-3707818824
> sample is [apache/dubbo-samples#1278](https://github.com/apache/dubbo-samples/pull/1278)
lack of PR effectiveness checking asserations.
--
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: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] Support Tri backpress [dubbo]
Posted by "wangchengming666 (via GitHub)" <gi...@apache.org>.
wangchengming666 commented on PR #15957:
URL: https://github.com/apache/dubbo/pull/15957#issuecomment-3703680055
用户手动配置如下
dubbo:
protocol:
triple:
window-update-ratio: 0.5 # 默认值，可不配置
--
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: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] Support Tri backpress [dubbo]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #15957:
URL: https://github.com/apache/dubbo/pull/15957#discussion_r2653242494
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/CallStreamObserver.java:
##########
@@ -48,4 +48,11 @@ public interface CallStreamObserver<T> extends StreamObserver<T> {
* specified.
*/
void disableAutoFlowControl();
+
+ /**
+ * compatible method for gRPC
Review Comment:
The comment "compatible method for gRPC" is unclear. It should explain what gRPC API this is compatible with, and why the compatibility is needed. For example, mention the specific gRPC class/interface this aligns with.
```suggestion
* Compatibility method to mirror gRPC Java
* {@code io.grpc.stub.CallStreamObserver#disableAutoInboundFlowControl()}.
* <p>
* This allows code written against gRPC's {@code CallStreamObserver} API to be
* more easily reused with Dubbo by providing an equivalent entry point that
* delegates to {@link #disableAutoFlowControl()}.
```
##########
dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/frame/RecordListener.java:
##########
@@ -21,6 +21,9 @@ public class RecordListener implements TriDecoder.Listener {
int dataCount;
boolean close;
+ @Override
+ public void bytesRead(int numBytes) {}
Review Comment:
The empty bytesRead implementation should have a comment explaining why it's a no-op. For example, indicate that flow control is handled differently by the QUIC layer in HTTP/3, or reference the consumeBytes implementation below.
```suggestion
public void bytesRead(int numBytes) {
// Intentionally left blank in tests: byte-level flow control notifications
// are not needed for these assertions, only raw message contents are recorded.
}
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2TripleClientStream.java:
##########
@@ -72,4 +90,68 @@ public void handlerAdded(ChannelHandlerContext ctx) {
writeQueue.enqueue(CreateStreamQueueCommand.create(bootstrap, streamChannelFuture));
return streamChannelFuture;
}
+
+ @Override
+ protected void consumeBytes(int numBytes) {
+ if (numBytes <= 0) {
+ return;
+ }
+
+ // todo The current implementation is not optimal, and alternative implementations should be considered.
Review Comment:
The comment on line 100 contains a typo: "todo" should be "TODO" (all caps) to follow standard code annotation conventions. This makes it easier for IDEs and tools to recognize it as a task marker.
```suggestion
// TODO The current implementation is not optimal, and alternative implementations should be considered.
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/frame/TriDecoder.java:
##########
@@ -145,11 +145,23 @@ private void processHeader() {
* Processes the GRPC message body, which depending on frame header flags may be compressed.
*/
private void processBody() {
- // There is no reliable way to get the uncompressed size per message when it's compressed,
- // because the uncompressed bytes are provided through an InputStream whose total size is
- // unknown until all bytes are read, and we don't know when it happens.
- byte[] stream = compressedFlag ? getCompressedBody() : getUncompressedBody();
+ // Calculate total bytes read: header + payload (before decompression)
+ int totalBytesRead = HEADER_LENGTH + requiredLength;
+ byte[] stream;
+ try {
+ // There is no reliable way to get the uncompressed size per message when it's compressed,
+ // because the uncompressed bytes are provided through an InputStream whose total size is
+ // unknown until all bytes are read, and we don't know when it happens.
+ stream = compressedFlag ? getCompressedBody() : getUncompressedBody();
+ } finally {
+ // Notify listener about bytes read for flow control immediately after reading bytes
+ // This must be in finally block to ensure flow control works even if reading fails
+ // Following gRPC's pattern: bytesRead is called as soon as bytes are consumed from input
+ listener.bytesRead(totalBytesRead);
+ }
+
Review Comment:
The bytesRead notification in the finally block will be called even if an exception occurs during getCompressedBody or getUncompressedBody. However, if these methods fail partway through reading the buffer, the totalBytesRead value may not accurately reflect the actual bytes consumed. Consider whether this is the intended behavior or if bytesRead should only be called on successful completion.
```suggestion
// There is no reliable way to get the uncompressed size per message when it's compressed,
// because the uncompressed bytes are provided through an InputStream whose total size is
// unknown until all bytes are read, and we don't know when it happens.
stream = compressedFlag ? getCompressedBody() : getUncompressedBody();
// Notify listener about bytes read for flow control only after successfully reading bytes
// Following gRPC's pattern: bytesRead is called as soon as bytes are consumed from input
listener.bytesRead(totalBytesRead);
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ClientCall.java:
##########
@@ -32,6 +32,13 @@ public interface ClientCall {
*/
interface Listener {
+ /**
+ * Whether the response is streaming response.
+ *
+ * @return
Review Comment:
Missing Javadoc comment for the ClientCall.Listener.streamingResponse() method. The method should have proper documentation explaining what it returns and when it should return true vs false.
```suggestion
* Whether the response is a streaming response.
* <p>
* Implementations should return {@code true} when the server is expected to send
* zero or more response messages over time (server-streaming or bidi-streaming
* calls). Return {@code false} when at most a single response message is expected
* from the server (unary or non-streaming calls).
*
* @return {@code true} if the server may send multiple response messages,
* {@code false} if at most one response message is expected
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleHttp2LocalFlowController.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.rpc.protocol.tri.transport;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.handler.codec.http2.DefaultHttp2LocalFlowController;
+import io.netty.handler.codec.http2.Http2Connection;
+import io.netty.handler.codec.http2.Http2Exception;
+import io.netty.handler.codec.http2.Http2Stream;
+
+/**
+ * Custom HTTP/2 local flow controller for Triple protocol with manual flow control.
+ *
+ * <p>This flow controller works together with {@code Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL = false}
+ * to enable manual flow control. The complete mechanism requires two parts:
+ *
+ * <h3>1. Disable Netty's automatic WINDOW_UPDATE (at Http2StreamChannel level)</h3>
+ * <p>Set {@code Http2StreamChannelOption.AUTO_STREAM_FLOW_CONTROL = false} when creating Http2StreamChannel.
+ * This prevents Netty's AbstractHttp2StreamChannel from automatically sending WINDOW_UPDATE frames.
+ *
+ * <h3>2. Manual flow control pattern (at Http2Connection level)</h3>
+ * <ol>
+ * <li>Data is received and tracked via {@link #receiveFlowControlledFrame} - decreases window size</li>
+ * <li>Application processes the data (decoding, business logic)</li>
+ * <li>Application calls {@link #consumeBytes} to return processed bytes</li>
+ * <li>WINDOW_UPDATE frame is sent when consumed bytes reach threshold (default: 50% of initial window)</li>
+ * </ol>
+ *
+ * <h3>Flow Control Call Chain</h3>
+ * <pre>
+ * Server: StreamingDecoder.bytesRead() → FragmentListener.bytesRead() → H2StreamChannel.consumeBytes()
+ * → Http2LocalFlowController.consumeBytes() → WINDOW_UPDATE
+ * Client: TriDecoder.Listener.bytesRead() → AbstractTripleClientStream.consumeBytes()
+ * → Http2LocalFlowController.consumeBytes() → WINDOW_UPDATE
+ * </pre>
+ *
+ * @see org.apache.dubbo.remoting.http12.h2.H2StreamChannel#consumeBytes(int)
+ * @see io.netty.handler.codec.http2.Http2StreamChannelOption#AUTO_STREAM_FLOW_CONTROL
+ */
+public class TripleHttp2LocalFlowController extends DefaultHttp2LocalFlowController {
+
+ /**
+ * Creates a new flow controller with custom windowUpdateRatio.
+ *
+ * @param connection the HTTP/2 connection
+ * @param windowUpdateRatio the ratio of consumed bytes to initial window size at which
+ * WINDOW_UPDATE frames are sent. Must be between 0 (exclusive) and 1 (inclusive).
+ * For example, 0.5 means WINDOW_UPDATE is sent when 50% of the
+ * initial window has been consumed.
+ */
+ public TripleHttp2LocalFlowController(Http2Connection connection, float windowUpdateRatio) {
+ super(connection, windowUpdateRatio, true);
+ }
+
+ @Override
+ public void receiveFlowControlledFrame(Http2Stream stream, ByteBuf data, int padding, boolean endOfStream)
+ throws Http2Exception {
+ super.receiveFlowControlledFrame(stream, data, padding, endOfStream);
+ }
Review Comment:
The overridden receiveFlowControlledFrame method simply calls super without any additional logic. This override serves no purpose and should be removed to reduce unnecessary code.
```suggestion
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2TripleClientStream.java:
##########
@@ -72,4 +90,68 @@ public void handlerAdded(ChannelHandlerContext ctx) {
writeQueue.enqueue(CreateStreamQueueCommand.create(bootstrap, streamChannelFuture));
return streamChannelFuture;
}
+
+ @Override
+ protected void consumeBytes(int numBytes) {
+ if (numBytes <= 0) {
+ return;
+ }
+
+ // todo The current implementation is not optimal, and alternative implementations should be considered.
+
+ Channel streamChannel = getStreamChannelFuture().getNow();
+ if (!(streamChannel instanceof Http2StreamChannel)) {
+ return;
+ }
+
+ Http2StreamChannel http2StreamChannel = (Http2StreamChannel) streamChannel;
+
Review Comment:
Potential race condition: getStreamChannelFuture().getNow() may return null if the future hasn't completed yet. While the code handles the case where streamChannel is not an Http2StreamChannel, it doesn't explicitly handle the null case. Consider adding a null check or using a different approach to wait for the channel to be available.
```suggestion
TripleStreamChannelFuture streamChannelFuture = getStreamChannelFuture();
Channel streamChannel = streamChannelFuture.getNow();
if (streamChannel == null) {
// Stream channel not ready yet; defer consumeBytes until it becomes available.
streamChannelFuture.addListener(future -> {
Channel ch = streamChannelFuture.getNow();
if (ch instanceof Http2StreamChannel) {
doConsumeBytes((Http2StreamChannel) ch, numBytes);
}
});
return;
}
if (!(streamChannel instanceof Http2StreamChannel)) {
return;
}
Http2StreamChannel http2StreamChannel = (Http2StreamChannel) streamChannel;
doConsumeBytes(http2StreamChannel, numBytes);
}
private void doConsumeBytes(Http2StreamChannel http2StreamChannel, int numBytes) {
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ClientCall.java:
##########
@@ -32,6 +32,13 @@ public interface ClientCall {
*/
interface Listener {
+ /**
+ * Whether the response is streaming response.
+ *
+ * @return
Review Comment:
The Javadoc comment has an incomplete @return tag with no description. Either add the description or remove the @return tag if it follows a convention where boolean return types are self-explanatory.
```suggestion
* @return true if the response is a streaming response
```
##########
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h2/NettyH2StreamChannel.java:
##########
@@ -89,4 +103,40 @@ public CompletableFuture<Void> writeResetFrame(long errorCode) {
http2StreamChannel.write(resetFrame).addListener(nettyHttpChannelFutureListener);
return nettyHttpChannelFutureListener;
}
+
+ @Override
+ public void consumeBytes(int numBytes) throws Exception {
+ if (numBytes <= 0) {
+ return;
+ }
+
+ if (http2Connection == null) {
+ LOGGER.debug("Http2Connection not available, skip consumeBytes");
+ return;
+ }
+
+ Http2LocalFlowController localFlowController = http2Connection.local().flowController();
+
+ // Get the stream from connection using stream id
+ int streamId = http2StreamChannel.stream().id();
Review Comment:
Potential NullPointerException: If http2StreamChannel.stream() returns null at line 121, the code will throw a NullPointerException. While there's a check for the stream being null after retrieving it from the connection (line 123), the stream().id() call at line 121 could fail first if stream() itself returns null. Consider checking if stream() is null before calling id().
```suggestion
// Get the stream from Http2StreamChannel and then from connection using stream id
Http2Stream channelStream = http2StreamChannel.stream();
if (channelStream == null) {
LOGGER.debug("Http2StreamChannel.stream() is null, skip consumeBytes");
return;
}
int streamId = channelStream.id();
```
##########
dubbo-common/src/main/java/org/apache/dubbo/config/nested/TripleConfig.java:
##########
@@ -355,6 +367,19 @@ public void setMaxMessageSize(Integer maxMessageSize) {
this.maxMessageSize = maxMessageSize;
}
+ public Float getWindowUpdateRatio() {
+ return windowUpdateRatio;
+ }
+
+ @Parameter(excluded = true)
+ public float getWindowUpdateRatioOrDefault() {
+ return windowUpdateRatio == null ? DEFAULT_WINDOW_UPDATE_RATIO : windowUpdateRatio;
+ }
+
+ public void setWindowUpdateRatio(Float windowUpdateRatio) {
Review Comment:
The validation for windowUpdateRatio is missing. The documentation states it must be between 0 (exclusive) and 1 (inclusive), but there's no validation in the setter to enforce this range. Consider adding validation to prevent invalid values.
```suggestion
public void setWindowUpdateRatio(Float windowUpdateRatio) {
if (windowUpdateRatio != null && (windowUpdateRatio <= 0.0f || windowUpdateRatio > 1.0f)) {
throw new IllegalArgumentException("windowUpdateRatio must be > 0 and <= 1, but was: " + windowUpdateRatio);
}
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ClientCall.java:
##########
@@ -89,6 +96,8 @@ interface Listener {
*/
boolean isAutoRequest();
Review Comment:
Missing Javadoc comment for the setAutoRequestWithInitial method. This new API method should have proper documentation explaining what it does, the purpose of the initialRequest parameter, and how it differs from setAutoRequest.
```suggestion
/**
* Enable auto request for this call with an initial number of messages to request.
* <p>
* This variant of auto request allows specifying how many response messages should be
* requested from the server immediately when the call starts or auto request is enabled.
* It is similar to {@link #setAutoRequest(boolean)} but also configures the initial
* {@link #request(int) request} amount.
*
* @param initialRequest the initial number of messages to request from the server
*/
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/observer/ClientCallToObserverAdapter.java:
##########
@@ -23,10 +23,12 @@
public class ClientCallToObserverAdapter<T> extends CancelableStreamObserver<T> implements ClientStreamObserver<T> {
private final ClientCall call;
+ private final boolean streamingResponse;
private boolean terminated;
- public ClientCallToObserverAdapter(ClientCall call) {
+ public ClientCallToObserverAdapter(ClientCall call, boolean streamingResponse) {
this.call = call;
+ this.streamingResponse = streamingResponse;
Review Comment:
The unused field streamingResponse in ClientCallToObserverAdapter is stored but never used. Either remove the field or implement the intended functionality that requires it.
##########
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/LengthFieldStreamingDecoder.java:
##########
@@ -184,7 +184,20 @@ private void skipOffset(InputStream inputStream, int lengthFieldOffset) throws I
}
private void processBody() throws IOException {
- byte[] rawMessage = readRawMessage(accumulate, requiredLength);
+ // Calculate total bytes read: header (offset + length field) + payload
+ int totalBytesRead = lengthFieldOffset + lengthFieldLength + requiredLength;
+
+ byte[] rawMessage;
+ try {
+ rawMessage = readRawMessage(accumulate, requiredLength);
+ } finally {
+ // Notify listener about bytes read for flow control immediately after reading bytes
+ // This must be in finally block to ensure flow control works even if reading fails
+ // Following gRPC's pattern: bytesRead is called as soon as bytes are consumed from input
+ listener.bytesRead(totalBytesRead);
+ }
+
Review Comment:
The bytesRead notification in the finally block will be called even if an exception occurs during readRawMessage. However, if readRawMessage fails partway through reading the buffer, the totalBytesRead value may not accurately reflect the actual bytes consumed. Consider whether this is the intended behavior or if bytesRead should only be called on successful completion.
```suggestion
byte[] rawMessage = readRawMessage(accumulate, requiredLength);
// Notify listener about bytes read for flow control immediately after successfully reading bytes
// Following gRPC's pattern: bytesRead is called as soon as bytes are consumed from input
listener.bytesRead(totalBytesRead);
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http2/Http2TripleClientStream.java:
##########
@@ -72,4 +90,68 @@ public void handlerAdded(ChannelHandlerContext ctx) {
writeQueue.enqueue(CreateStreamQueueCommand.create(bootstrap, streamChannelFuture));
return streamChannelFuture;
}
+
+ @Override
+ protected void consumeBytes(int numBytes) {
+ if (numBytes <= 0) {
+ return;
+ }
+
+ // todo The current implementation is not optimal, and alternative implementations should be considered.
+
+ Channel streamChannel = getStreamChannelFuture().getNow();
+ if (!(streamChannel instanceof Http2StreamChannel)) {
+ return;
+ }
+
+ Http2StreamChannel http2StreamChannel = (Http2StreamChannel) streamChannel;
+
+ // Get Http2Connection from parent channel pipeline
+ Http2Connection http2Connection = getHttp2Connection();
+ if (http2Connection == null) {
+ LOGGER.debug("Http2Connection not available for flow control");
+ return;
+ }
+
+ Http2LocalFlowController localFlowController = http2Connection.local().flowController();
+ int streamId = http2StreamChannel.stream().id();
Review Comment:
Potential NullPointerException: If getStreamChannelFuture().getNow() returns null or if http2StreamChannel.stream() returns null, the subsequent calls will throw NullPointerException. The code checks if streamChannel is an instance of Http2StreamChannel but doesn't check for null before the instanceof check. Add an explicit null check before line 103.
##########
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h2/NettyH2StreamChannel.java:
##########
@@ -89,4 +103,40 @@ public CompletableFuture<Void> writeResetFrame(long errorCode) {
http2StreamChannel.write(resetFrame).addListener(nettyHttpChannelFutureListener);
return nettyHttpChannelFutureListener;
}
+
+ @Override
+ public void consumeBytes(int numBytes) throws Exception {
+ if (numBytes <= 0) {
+ return;
+ }
+
+ if (http2Connection == null) {
+ LOGGER.debug("Http2Connection not available, skip consumeBytes");
+ return;
+ }
+
+ Http2LocalFlowController localFlowController = http2Connection.local().flowController();
+
+ // Get the stream from connection using stream id
+ int streamId = http2StreamChannel.stream().id();
+ Http2Stream stream = http2Connection.stream(streamId);
+ if (stream == null) {
+ LOGGER.debug("Stream {} not found in connection, skip consumeBytes", streamId);
+ return;
+ }
+
+ // Consume bytes to trigger WINDOW_UPDATE frame
+ // This must be executed in the event loop thread
+ if (http2StreamChannel.eventLoop().inEventLoop()) {
+ localFlowController.consumeBytes(stream, numBytes);
+ } else {
+ http2StreamChannel.eventLoop().execute(() -> {
+ try {
+ localFlowController.consumeBytes(stream, numBytes);
+ } catch (Exception e) {
+ LOGGER.warn(PROTOCOL_FAILED_RESPONSE, "", "", "Failed to consumeBytes for stream " + streamId, e);
+ }
+ });
Review Comment:
Inconsistent error handling: When in the event loop (line 131), exceptions from consumeBytes are propagated (declared in the method signature), but when executed asynchronously (line 135), exceptions are caught and logged. This inconsistency could lead to different behavior depending on which thread calls the method. Consider catching the exception in both cases or propagating it in both cases for consistency.
```suggestion
http2StreamChannel.eventLoop().execute(() -> localFlowController.consumeBytes(stream, numBytes));
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ClientStreamObserver.java:
##########
@@ -30,4 +30,6 @@ public interface ClientStreamObserver<T> extends CallStreamObserver<T> {
default void disableAutoRequest() {
disableAutoFlowControl();
}
+
Review Comment:
Missing Javadoc comment for the disableAutoRequestWithInitial method. This new API method should have proper documentation explaining what it does, the purpose of the request parameter, and how it differs from disableAutoRequest or disableAutoFlowControl.
```suggestion
/**
* Swaps to manual flow control and configures an initial number of messages to request
* when the call starts.
* <p>
* This is similar to {@link #disableAutoRequest()} but allows specifying how many messages
* should be requested immediately once the call is established, as if {@link #request(int)}
* were invoked with the given {@code request} value after disabling automatic flow control.
*
* @param request the number of messages to request initially after auto flow control has been
* disabled; must be a non-negative number.
*/
```
--
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: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
Re: [PR] Support Tri backpress [dubbo]
Posted by "EarthChen (via GitHub)" <gi...@apache.org>.
EarthChen commented on PR #15957:
URL: https://github.com/apache/dubbo/pull/15957#issuecomment-3707973285
## 应用层背压
应用层背压机制允许应用程序在发送数据时感知底层传输层的状态，避免在传输层缓冲区满时继续发送数据导致内存溢出或性能下降。
### 核心 API
| API | 说明 |
|-----|------|
| `isReady()` | 检查当前流是否可写，返回 `true` 表示可以发送更多消息 |
| `setOnReadyHandler(Runnable)` | 设置回调，当流从不可写变为可写时触发 |
### 设计参考
本实现完全对齐 gRPC Java 的设计模式：
- `CallStreamObserver.isReady()` 和 `CallStreamObserver.setOnReadyHandler(Runnable)`
- `ClientCall.Listener.onReady()`
## 客户端实现
### 核心类关系
```
┌─────────────────────────────────────────────────────────────────────────────┐
│ 用户代码 │
│ CallStreamObserver<T> observer = ...; │
│ observer.setOnReadyHandler(() -> { /* 恢复发送 */ }); │
│ if (observer.isReady()) { observer.onNext(data); } │
└─────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ ClientCallToObserverAdapter │
│ - 实现 CallStreamObserver 接口 │
│ - 存储 onReadyHandler │
│ - isReady() 委托给 ClientCall.isReady() │
└─────────────────────────────────────────────────────────────────────────────┘
│
┌─────────────────┴─────────────────┐
▼ ▼
┌──────────────────────────────────┐ ┌──────────────────────────────────────┐
│ TripleClientCall │ │ ObserverToClientCallListenerAdapter │
│ - 实现 ClientCall │ │ - 实现 ClientCall.Listener │
│ - 实现 ClientStream.Listener │ │ - 持有 ClientCallToObserverAdapter │
│ - isReady() 委托给 stream │ │ - onReady() 触发 onReadyHandler │
└──────────────────────────────────┘ └──────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ AbstractTripleClientStream │
│ - 实现 ClientStream │
│ - isReady() 返回 channel.isWritable() │
│ - onWritabilityChanged() 调用 listener.onReady() │
└─────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ Netty Http2StreamChannel │
│ - isWritable() 底层可写性检查 │
│ - channelWritabilityChanged 事件 │
└─────────────────────────────────────────────────────────────────────────────┘
```
### isReady() 调用链
```
用户: observer.isReady()
│
▼
ClientCallToObserverAdapter.isReady()
│ return call.isReady()
▼
TripleClientCall.isReady()
│ 检查 canceled/done 状态
│ return stream.isReady()
▼
AbstractTripleClientStream.isReady()
│ Channel channel = streamChannelFuture.getNow()
│ return channel.isWritable()
▼
Netty: Http2StreamChannel.isWritable()
```
**代码实现：**
```java
// ClientCallToObserverAdapter.java
@Override
public boolean isReady() {
return call.isReady();
}
// TripleClientCall.java
@Override
public boolean isReady() {
if (canceled) return false;
if (done) return false;
return stream.isReady();
}
// AbstractTripleClientStream.java
@Override
public boolean isReady() {
Channel channel = streamChannelFuture.getNow();
if (channel == null) return false;
return channel.isWritable();
}
```
### onReadyHandler 触发链
当 Netty Channel 从不可写变为可写时，触发以下调用链：
```
Netty: channelWritabilityChanged 事件
│
▼
TripleHttp2ClientResponseHandler.channelWritabilityChanged(ctx)
│ transportListener.onWritabilityChanged()
▼
ClientTransportListener.onWritabilityChanged() [H2TransportListener 实现]
│ AbstractTripleClientStream.this.onWritabilityChanged()
▼
AbstractTripleClientStream.onWritabilityChanged()
│ 检查 channel.isWritable()
│ listener.onReady() [同步调用]
▼
TripleClientCall.onReady() [ClientStream.Listener 实现]
│ executor.execute(() -> listener.onReady()) [异步执行]
▼
ObserverToClientCallListenerAdapter.onReady() [ClientCall.Listener 实现]
│ requestAdapter.getOnReadyHandler().run()
▼
用户设置的 Runnable 被执行 [在业务 executor 线程中]
```
**代码实现：**
```java
// TripleHttp2ClientResponseHandler.java
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
transportListener.onWritabilityChanged();
super.channelWritabilityChanged(ctx);
}
// AbstractTripleClientStream.ClientTransportListener
@Override
public void onWritabilityChanged() {
AbstractTripleClientStream.this.onWritabilityChanged();
}
// AbstractTripleClientStream.java
protected void onWritabilityChanged() {
Channel channel = streamChannelFuture.getNow();
if (channel != null && channel.isWritable()) {
// 同步调用，由 TripleClientCall.onReady() 负责异步执行
listener.onReady();
}
}
// TripleClientCall.java
@Override
public void onReady() {
if (listener == null) {
return;
}
// ObserverToClientCallListenerAdapter.onReady() triggers the onReadyHandler
executor.execute(() -> {
try {
listener.onReady();
} catch (Throwable t) {
LOGGER.warn(PROTOCOL_STREAM_LISTENER, "", "",
"Error executing listener.onReady()", t);
}
});
}
// ObserverToClientCallListenerAdapter.java
@Override
public void onReady() {
if (requestAdapter == null) return;
Runnable handler = requestAdapter.getOnReadyHandler();
if (handler == null) return;
handler.run();
}
```
### setOnReadyHandler 设置流程
```
用户: observer.setOnReadyHandler(runnable)
│
▼
ClientCallToObserverAdapter.setOnReadyHandler(runnable)
│ this.onReadyHandler = runnable // 存储在本地
│
└──────────────────────────────────────────────────────
│
TripleInvoker.streamCall() 中建立关联： │
│ │
▼ │
listener.setRequestAdapter(adapter) │
│ │
▼ │
ObserverToClientCallListenerAdapter 持有 requestAdapter 引用 ──┘
│
▼ (当 onReady 被调用时)
requestAdapter.getOnReadyHandler().run()
```
**代码实现：**
```java
// ClientCallToObserverAdapter.java
private Runnable onReadyHandler;
@Override
public void setOnReadyHandler(Runnable onReadyHandler) {
this.onReadyHandler = onReadyHandler;
}
public Runnable getOnReadyHandler() {
return onReadyHandler;
}
// TripleInvoker.java
StreamObserver<Object> streamCall(...) {
ObserverToClientCallListenerAdapter listener =
new ObserverToClientCallListenerAdapter(responseObserver);
StreamObserver<Object> streamObserver = call.start(metadata, listener);
// 建立关联，让 listener 能够访问 adapter 的 onReadyHandler
if (streamObserver instanceof ClientCallToObserverAdapter) {
listener.setRequestAdapter((ClientCallToObserverAdapter<Object>) streamObserver);
}
// ...
}
```
## 服务端实现
服务端的背压机制通过 `H2StreamChannel` 和 `Http2ServerChannelObserver` 实现。
### 核心类关系
```
┌─────────────────────────────────────────────────────────────────────────────┐
│ Http2ServerChannelObserver │
│ - 实现 CallStreamObserver │
│ - isReady() 委托给 H2StreamChannel.isReady() │
│ - setOnReadyHandler() 设置回调 │
└─────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ NettyH2StreamChannel │
│ - 实现 H2StreamChannel │
│ - isReady() 返回 http2StreamChannel.isWritable() │
│ - onWritabilityChanged() 触发 onReadyHandler │
└─────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ Netty Http2StreamChannel │
│ - isWritable() 底层可写性检查 │
└─────────────────────────────────────────────────────────────────────────────┘
```
### 服务端 onWritabilityChanged 触发链
```
Netty: channelWritabilityChanged 事件
│
▼
NettyHttp2FrameHandler.channelWritabilityChanged(ctx)
│ transportListener.onWritabilityChanged()
▼
Http2TransportListener.onWritabilityChanged()
│ 由具体实现处理
▼
触发服务端的背压回调
```
## 使用示例
### 客户端使用背压
```java
// 获取请求 StreamObserver
StreamObserver<Request> requestObserver = stub.biStreamCall(responseObserver);
// 转换为 CallStreamObserver 以使用背压 API
CallStreamObserver<Request> callObserver = (CallStreamObserver<Request>) requestObserver;
// 禁用自动流控
callObserver.disableAutoFlowControl();
// 设置 ready 回调
callObserver.setOnReadyHandler(() -> {
// 当流变为可写时，恢复发送数据
while (callObserver.isReady() && hasMoreData()) {
callObserver.onNext(getNextData());
}
});
// 初始发送（如果可写）
while (callObserver.isReady() && hasMoreData()) {
callObserver.onNext(getNextData());
}
```
### 服务端使用背压
```java
@Override
public StreamObserver<Request> biStreamCall(StreamObserver<Response> responseObserver) {
// 转换为 ServerCallStreamObserver
ServerCallStreamObserver<Response> serverObserver =
(ServerCallStreamObserver<Response>) responseObserver;
// 设置 ready 回调
serverObserver.setOnReadyHandler(() -> {
while (serverObserver.isReady() && hasMoreResponses()) {
serverObserver.onNext(getNextResponse());
}
});
return new StreamObserver<Request>() {
@Override
public void onNext(Request request) {
// 处理请求
}
// ...
};
}
```
--
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: notifications-unsubscribe@dubbo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org