You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@dubbo.apache.org by GitBox <gi...@apache.org> on 2022/09/26 03:13:43 UTC
[GitHub] [dubbo] asa3311 opened a new pull request, #10672: triple反压增加流控
asa3311 opened a new pull request, #10672:
URL: https://github.com/apache/dubbo/pull/10672
triple反压增加流控，服务端读完数据后才更新流控窗口。
主要改动点：
1、TripleHttp2Protocol增加加载自定义流控，见130～132，180～182行。
2、TripleHttp2FrameServerHandler读完数据body后，更新流控窗口，见74～76行。
3、TriHttp2LocalFlowController在消费流量更新流控窗口时，判断处理是否从TripleHttp2FrameServerHandler发起，非TripleHttp2FrameServerHandler发起的不再更新流控窗口，见187～190行
尚余在stream关闭时，流控还未从triple侧生效，仍从netty原生处发起。
## What is the purpose of the change
## Brief changelog
## Verifying this change
<!-- Follow this checklist to help us incorporate your contribution quickly and easily: -->
## Checklist
- [x] Make sure there is a [GitHub_issue](https://github.com/apache/dubbo/issues) field for the change (usually before you start working on it). Trivial changes like typos do not require a GitHub issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue.
- [ ] Each commit in the pull request should have a meaningful subject line and body.
- [ ] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
- [ ] Check if is necessary to patch to Dubbo 3 if you are work on Dubbo 2.7
- [ ] Write necessary unit-test to verify your logic correction, more mock a little better when cross module dependency exist. If the new feature or significant change is committed, please remember to add sample in [dubbo samples](https://github.com/apache/dubbo-samples) project.
- [ ] Add some description to [dubbo-website](https://github.com/apache/dubbo-website) project if you are requesting to add a feature.
- [ ] GitHub Actions works fine on your own branch.
- [ ] If this contribution is large, please follow the [Software Donation Guide](https://github.com/apache/dubbo/wiki/Software-donation-guide).
--
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
[GitHub] [dubbo] EarthChen commented on a diff in pull request #10672: triple反压增加流控
Posted by GitBox <gi...@apache.org>.
EarthChen commented on code in PR #10672:
URL: https://github.com/apache/dubbo/pull/10672#discussion_r979559476
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java:
##########
@@ -125,11 +126,18 @@ public void configServerProtocolHandler(URL url, ChannelOperator operator) {
DEFAULT_MAX_HEADER_LIST_SIZE)))
.frameLogger(SERVER_LOGGER)
.build();
+ //增加自定义流控
Review Comment:
plz use English
--
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
[GitHub] [dubbo] EarthChen commented on a diff in pull request #10672: triple反压增加流控
Posted by GitBox <gi...@apache.org>.
EarthChen commented on code in PR #10672:
URL: https://github.com/apache/dubbo/pull/10672#discussion_r979560004
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java:
##########
@@ -125,11 +126,18 @@ public void configServerProtocolHandler(URL url, ChannelOperator operator) {
DEFAULT_MAX_HEADER_LIST_SIZE)))
.frameLogger(SERVER_LOGGER)
.build();
+ //增加自定义流控
+ codec.connection().local().flowController(new TriHttp2LocalFlowController(codec.connection()));
+ codec.connection().remote().flowController(new TriHttp2RemoteFlowController(codec.connection()));
+ codec.connection().local().flowController().frameWriter(codec.encoder().frameWriter());
+
final Http2MultiplexHandler handler = new Http2MultiplexHandler(
new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) {
final ChannelPipeline p = ch.pipeline();
+ AttributeKey key = AttributeKey.valueOf("tri-connection");
Review Comment:
connection attr is already exist on channel
--
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
[GitHub] [dubbo] EarthChen commented on a diff in pull request #10672: triple反压增加流控
Posted by GitBox <gi...@apache.org>.
EarthChen commented on code in PR #10672:
URL: https://github.com/apache/dubbo/pull/10672#discussion_r979560004
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java:
##########
@@ -125,11 +126,18 @@ public void configServerProtocolHandler(URL url, ChannelOperator operator) {
DEFAULT_MAX_HEADER_LIST_SIZE)))
.frameLogger(SERVER_LOGGER)
.build();
+ //增加自定义流控
+ codec.connection().local().flowController(new TriHttp2LocalFlowController(codec.connection()));
+ codec.connection().remote().flowController(new TriHttp2RemoteFlowController(codec.connection()));
+ codec.connection().local().flowController().frameWriter(codec.encoder().frameWriter());
+
final Http2MultiplexHandler handler = new Http2MultiplexHandler(
new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) {
final ChannelPipeline p = ch.pipeline();
+ AttributeKey key = AttributeKey.valueOf("tri-connection");
Review Comment:
connection attr is already exist on channel
--
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
[GitHub] [dubbo] EarthChen commented on a diff in pull request #10672: triple反压增加流控
Posted by GitBox <gi...@apache.org>.
EarthChen commented on code in PR #10672:
URL: https://github.com/apache/dubbo/pull/10672#discussion_r991871561
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/BiStreamServerCallListener.java:
##########
@@ -41,13 +49,27 @@ public void onReturn(Object value) {
@Override
public void onMessage(Object message) {
- if (message instanceof Object[]) {
- message = ((Object[]) message)[0];
+ if (((TripleFlowControlFrame) message).getInstance() instanceof Object[]) {
Review Comment:
the message is must be TripleFlowControlFrame？
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleHttp2FrameServerHandler.java:
##########
@@ -112,15 +112,18 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
public void onDataRead(ChannelHandlerContext ctx, Http2DataFrame msg) throws Exception {
final TripleServerStream tripleServerStream = ctx.channel().attr(SERVER_STREAM_KEY)
.get();
- tripleServerStream.transportObserver.onData(msg.content(), msg.isEndStream());
+ tripleServerStream.transportObserver.onData(msg, msg.isEndStream());
}
public void onHeadersRead(ChannelHandlerContext ctx, Http2HeadersFrame msg) throws Exception {
TripleServerStream tripleServerStream = new TripleServerStream(ctx.channel(),
frameworkModel, executor,
pathResolver, acceptEncoding, filters);
ctx.channel().attr(SERVER_STREAM_KEY).set(tripleServerStream);
- tripleServerStream.transportObserver.onHeader(msg.headers(), msg.isEndStream());
+
+ //transmit connection to triple invoke to flowcontrol
+ Http2Connection connection = (Http2Connection)ctx.channel().attr(AttributeKey.valueOf("tri-connection")).get();
Review Comment:
Change AttributeKey to a constant
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/UnaryServerCallListener.java:
##########
@@ -37,11 +49,15 @@ public void onReturn(Object value) {
@Override
public void onMessage(Object message) {
- if (message instanceof Object[]) {
- invocation.setArguments((Object[]) message);
+ if (((TripleFlowControlFrame) message).getInstance() instanceof Object[]) {
Review Comment:
same
--
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
[GitHub] [dubbo] EarthChen commented on a diff in pull request #10672: triple反压增加流控
Posted by GitBox <gi...@apache.org>.
EarthChen commented on code in PR #10672:
URL: https://github.com/apache/dubbo/pull/10672#discussion_r985084600
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ServerStreamServerCallListener.java:
##########
@@ -34,11 +40,19 @@ public void onReturn(Object value) {
}
@Override
- public void onMessage(Object message) {
+ public void onMessage(Object message,DefaultHttp2WindowUpdateFrame stream, Http2Connection connection) {
Review Comment:
If it is necessary to add DefaultHttp2WindowUpdateFrame, can it be replaced with an interface?
--
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
[GitHub] [dubbo] EarthChen commented on a diff in pull request #10672: triple反压增加流控
Posted by GitBox <gi...@apache.org>.
EarthChen commented on code in PR #10672:
URL: https://github.com/apache/dubbo/pull/10672#discussion_r985084521
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ServerStreamServerCallListener.java:
##########
@@ -34,11 +40,19 @@ public void onReturn(Object value) {
}
@Override
- public void onMessage(Object message) {
+ public void onMessage(Object message,DefaultHttp2WindowUpdateFrame stream, Http2Connection connection) {
Review Comment:
Can these parameters not be added to the interface?
--
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
[GitHub] [dubbo] codecov-commenter commented on pull request #10672: triple反压增加流控
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #10672:
URL: https://github.com/apache/dubbo/pull/10672#issuecomment-1263869164
# [Codecov](https://codecov.io/gh/apache/dubbo/pull/10672?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#10672](https://codecov.io/gh/apache/dubbo/pull/10672?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (3dedaca) into [3.1](https://codecov.io/gh/apache/dubbo/commit/eb0e25e55f1516f1457f6acf9df8ee48cfa492c7?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (eb0e25e) will **decrease** coverage by `0.12%`.
> The diff coverage is `54.71%`.
```diff
@@ Coverage Diff @@
## 3.1 #10672 +/- ##
============================================
- Coverage 65.24% 65.11% -0.13%
- Complexity 493 494 +1
============================================
Files 1336 1338 +2
Lines 56918 57401 +483
Branches 8434 8501 +67
============================================
+ Hits 37134 37378 +244
- Misses 15813 15999 +186
- Partials 3971 4024 +53
```
| [Impacted Files](https://codecov.io/gh/apache/dubbo/pull/10672?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [.../protocol/tri/call/AbstractServerCallListener.java](https://codecov.io/gh/apache/dubbo/pull/10672/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvY2FsbC9BYnN0cmFjdFNlcnZlckNhbGxMaXN0ZW5lci5qYXZh) | `56.81% <0.00%> (-5.69%)` | :arrow_down: |
| [.../dubbo/rpc/protocol/tri/call/TripleClientCall.java](https://codecov.io/gh/apache/dubbo/pull/10672/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvY2FsbC9UcmlwbGVDbGllbnRDYWxsLmphdmE=) | `48.88% <ø> (+17.77%)` | :arrow_up: |
| [.../protocol/tri/call/BiStreamServerCallListener.java](https://codecov.io/gh/apache/dubbo/pull/10672/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvY2FsbC9CaVN0cmVhbVNlcnZlckNhbGxMaXN0ZW5lci5qYXZh) | `54.54% <16.66%> (-20.46%)` | :arrow_down: |
| [...tocol/tri/call/ServerStreamServerCallListener.java](https://codecov.io/gh/apache/dubbo/pull/10672/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvY2FsbC9TZXJ2ZXJTdHJlYW1TZXJ2ZXJDYWxsTGlzdGVuZXIuamF2YQ==) | `64.70% <50.00%> (-8.03%)` | :arrow_down: |
| [...rpc/protocol/tri/call/UnaryServerCallListener.java](https://codecov.io/gh/apache/dubbo/pull/10672/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvY2FsbC9VbmFyeVNlcnZlckNhbGxMaXN0ZW5lci5qYXZh) | `77.77% <50.00%> (-13.89%)` | :arrow_down: |
| [...rpc/protocol/tri/TriHttp2RemoteFlowController.java](https://codecov.io/gh/apache/dubbo/pull/10672/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvVHJpSHR0cDJSZW1vdGVGbG93Q29udHJvbGxlci5qYXZh) | `51.79% <51.79%> (ø)` | |
| [.../rpc/protocol/tri/TriHttp2LocalFlowController.java](https://codecov.io/gh/apache/dubbo/pull/10672/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvVHJpSHR0cDJMb2NhbEZsb3dDb250cm9sbGVyLmphdmE=) | `54.94% <54.94%> (ø)` | |
| [...bo/rpc/protocol/tri/stream/TripleClientStream.java](https://codecov.io/gh/apache/dubbo/pull/10672/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvc3RyZWFtL1RyaXBsZUNsaWVudFN0cmVhbS5qYXZh) | `55.69% <60.00%> (+1.89%)` | :arrow_up: |
| [...pache/dubbo/rpc/protocol/tri/frame/TriDecoder.java](https://codecov.io/gh/apache/dubbo/pull/10672/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvZnJhbWUvVHJpRGVjb2Rlci5qYXZh) | `83.33% <90.90%> (+1.97%)` | :arrow_up: |
| [...he/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java](https://codecov.io/gh/apache/dubbo/pull/10672/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZHViYm8tcnBjL2R1YmJvLXJwYy10cmlwbGUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2R1YmJvL3JwYy9wcm90b2NvbC90cmkvVHJpcGxlSHR0cDJQcm90b2NvbC5qYXZh) | `92.30% <100.00%> (+0.64%)` | :arrow_up: |
| ... and [49 more](https://codecov.io/gh/apache/dubbo/pull/10672/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
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
[GitHub] [dubbo] EarthChen commented on a diff in pull request #10672: triple反压增加流控
Posted by GitBox <gi...@apache.org>.
EarthChen commented on code in PR #10672:
URL: https://github.com/apache/dubbo/pull/10672#discussion_r985085048
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/frame/TriDecoder.java:
##########
@@ -37,23 +39,38 @@ public class TriDecoder implements Deframer {
private boolean inDelivery = false;
private boolean closing;
private boolean closed;
-
+ private Http2Connection connection;
private int requiredLength = HEADER_LENGTH;
+ private int flowControlledBytes;
+
+ private Http2FrameStream stream;
Review Comment:
It seems that these parameters can be obtained in `onRawMessage`
--
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
[GitHub] [dubbo] asa3311 closed pull request #10672: triple反压增加流控
Posted by GitBox <gi...@apache.org>.
asa3311 closed pull request #10672: triple反压增加流控
URL: https://github.com/apache/dubbo/pull/10672
--
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
[GitHub] [dubbo] EarthChen commented on a diff in pull request #10672: triple反压增加流控
Posted by GitBox <gi...@apache.org>.
EarthChen commented on code in PR #10672:
URL: https://github.com/apache/dubbo/pull/10672#discussion_r979560516
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleHttp2FrameServerHandler.java:
##########
@@ -71,6 +71,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
onHeadersRead(ctx, (Http2HeadersFrame) msg);
} else if (msg instanceof Http2DataFrame) {
onDataRead(ctx, (Http2DataFrame) msg);
+ //服务端读完数据，更新流控窗口
Review Comment:
onDataRead is async, so is unused
--
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
[GitHub] [dubbo] AlbumenJ commented on a diff in pull request #10672: triple反压增加流控
Posted by GitBox <gi...@apache.org>.
AlbumenJ commented on code in PR #10672:
URL: https://github.com/apache/dubbo/pull/10672#discussion_r984262241
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TriHttp2LocalFlowController.java:
##########
@@ -0,0 +1,640 @@
+/*
+ * 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;
+import static io.netty.handler.codec.http2.Http2CodecUtil.CONNECTION_STREAM_ID;
+import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_WINDOW_SIZE;
+import static io.netty.handler.codec.http2.Http2CodecUtil.MAX_INITIAL_WINDOW_SIZE;
+import static io.netty.handler.codec.http2.Http2CodecUtil.MIN_INITIAL_WINDOW_SIZE;
+import static io.netty.handler.codec.http2.Http2Error.FLOW_CONTROL_ERROR;
+import static io.netty.handler.codec.http2.Http2Error.INTERNAL_ERROR;
+import static io.netty.handler.codec.http2.Http2Exception.connectionError;
+import static io.netty.handler.codec.http2.Http2Exception.streamError;
+import static io.netty.util.internal.ObjectUtil.checkNotNull;
+import static io.netty.util.internal.ObjectUtil.checkPositiveOrZero;
+import static java.lang.Math.max;
+import static java.lang.Math.min;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http2.*;
+import io.netty.handler.codec.http2.Http2Exception.CompositeStreamException;
+import io.netty.handler.codec.http2.Http2Exception.StreamException;
+import io.netty.util.internal.PlatformDependent;
+import io.netty.util.internal.UnstableApi;
+
+@UnstableApi
+public class TriHttp2LocalFlowController implements Http2LocalFlowController {
+ /**
+ * The default ratio of window size to initial window size below which a {@code WINDOW_UPDATE}
+ * is sent to expand the window.
+ */
+ public static final float DEFAULT_WINDOW_UPDATE_RATIO = 0.5f;
+
+ private final Http2Connection connection;
+ private final Http2Connection.PropertyKey stateKey;
+ private Http2FrameWriter frameWriter;
+ private ChannelHandlerContext ctx;
+ private float windowUpdateRatio;
+ private int initialWindowSize = DEFAULT_WINDOW_SIZE;
+
+ public TriHttp2LocalFlowController(Http2Connection connection) {
+ this(connection, DEFAULT_WINDOW_UPDATE_RATIO, false);
+ }
+
+ public TriHttp2LocalFlowController(Http2Connection connection,
+ float windowUpdateRatio,
+ boolean autoRefillConnectionWindow) {
+ this.connection = checkNotNull(connection, "connection");
+ windowUpdateRatio(windowUpdateRatio);
+
+ // Add a flow state for the connection.
+ stateKey = connection.newKey();
+ org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.FlowState connectionState = autoRefillConnectionWindow ?
+ new org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.AutoRefillState(connection.connectionStream(), initialWindowSize) :
+ new org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.DefaultState(connection.connectionStream(), initialWindowSize);
+ connection.connectionStream().setProperty(stateKey, connectionState);
+
+ // Register for notification of new streams.
+ connection.addListener(new Http2ConnectionAdapter() {
+ @Override
+ public void onStreamAdded(Http2Stream stream) {
+ // Unconditionally used the reduced flow control state because it requires no object allocation
+ // and the DefaultFlowState will be allocated in onStreamActive.
+ stream.setProperty(stateKey, REDUCED_FLOW_STATE);
+ }
+
+ @Override
+ public void onStreamActive(Http2Stream stream) {
+ // Need to be sure the stream's initial window is adjusted for SETTINGS
+ // frames which may have been exchanged while it was in IDLE
+ stream.setProperty(stateKey, new org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.DefaultState(stream, initialWindowSize));
+ }
+
+ @Override
+ public void onStreamClosed(Http2Stream stream) {
+ try {
+ // When a stream is closed, consume any remaining bytes so that they
+ // are restored to the connection window.
+ org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.FlowState state = state(stream);
+ int unconsumedBytes = state.unconsumedBytes();
+ if (ctx != null && unconsumedBytes > 0) {
+ if (consumeAllBytes(state, unconsumedBytes)) {
+ // As the user has no real control on when this callback is used we should better
+ // call flush() if we produced any window update to ensure we not stale.
+ ctx.flush();
+ }
+ }
+ } catch (Http2Exception e) {
+ PlatformDependent.throwException(e);
+ } finally {
+ // Unconditionally reduce the amount of memory required for flow control because there is no
+ // object allocation costs associated with doing so and the stream will not have any more
+ // local flow control state to keep track of anymore.
+ stream.setProperty(stateKey, REDUCED_FLOW_STATE);
+ }
+ }
+ });
+ }
+
+ @Override
+ public TriHttp2LocalFlowController frameWriter(Http2FrameWriter frameWriter) {
+ this.frameWriter = checkNotNull(frameWriter, "frameWriter");
+ return this;
+ }
+
+ @Override
+ public void channelHandlerContext(ChannelHandlerContext ctx) {
+ this.ctx = checkNotNull(ctx, "ctx");
+ }
+
+ @Override
+ public void initialWindowSize(int newWindowSize) throws Http2Exception {
+ assert ctx == null || ctx.executor().inEventLoop();
+ int delta = newWindowSize - initialWindowSize;
+ initialWindowSize = newWindowSize;
+
+ org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.WindowUpdateVisitor visitor = new org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.WindowUpdateVisitor(delta);
+ connection.forEachActiveStream(visitor);
+ visitor.throwIfError();
+ }
+
+ @Override
+ public int initialWindowSize() {
+ return initialWindowSize;
+ }
+
+ @Override
+ public int windowSize(Http2Stream stream) {
+ return state(stream).windowSize();
+ }
+
+ @Override
+ public int initialWindowSize(Http2Stream stream) {
+ return state(stream).initialWindowSize();
+ }
+
+ @Override
+ public void incrementWindowSize(Http2Stream stream, int delta) throws Http2Exception {
+ assert ctx != null && ctx.executor().inEventLoop();
+ org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.FlowState state = state(stream);
+ // Just add the delta to the stream-specific initial window size so that the next time the window
+ // expands it will grow to the new initial size.
+ state.incrementInitialStreamWindow(delta);
+ state.writeWindowUpdateIfNeeded();
+ }
+
+ @Override
+ public boolean consumeBytes(Http2Stream stream, int numBytes) throws Http2Exception {
+ assert ctx != null && ctx.executor().inEventLoop();
+ checkPositiveOrZero(numBytes, "numBytes");
+ if (numBytes == 0) {
+ return false;
+ }
+ //use triple flowcontroller
+ if(!(Thread.currentThread().getStackTrace()[2].getClassName().endsWith("BiStreamServerCallListener") || Thread.currentThread().getStackTrace()[2].getClassName().endsWith("AbstractServerCallListener"))){
+ return false;
+ }
+ // Streams automatically consume all remaining bytes when they are closed, so just ignore
+ // if already closed.
+ if (stream != null && !isClosed(stream)) {
+ if (stream.id() == CONNECTION_STREAM_ID) {
+ throw new UnsupportedOperationException("Returning bytes for the connection window is not supported");
+ }
+
+ return consumeAllBytes(state(stream), numBytes);
+ }
+ return false;
+ }
+
+ private boolean consumeAllBytes(org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.FlowState state, int numBytes) throws Http2Exception {
+ return connectionState().consumeBytes(numBytes) | state.consumeBytes(numBytes);
+ }
+
+ @Override
+ public int unconsumedBytes(Http2Stream stream) {
+ return state(stream).unconsumedBytes();
+ }
+
+ private static void checkValidRatio(float ratio) {
+ if (Double.compare(ratio, 0.0) <= 0 || Double.compare(ratio, 1.0) >= 0) {
+ throw new IllegalArgumentException("Invalid ratio: " + ratio);
+ }
+ }
+
+ /**
+ * The window update ratio is used to determine when a window update must be sent. If the ratio
+ * of bytes processed since the last update has meet or exceeded this ratio then a window update will
+ * be sent. This is the global window update ratio that will be used for new streams.
+ * @param ratio the ratio to use when checking if a {@code WINDOW_UPDATE} is determined necessary for new streams.
+ * @throws IllegalArgumentException If the ratio is out of bounds (0, 1).
+ */
+ public void windowUpdateRatio(float ratio) {
+ assert ctx == null || ctx.executor().inEventLoop();
+ checkValidRatio(ratio);
+ windowUpdateRatio = ratio;
+ }
+
+ /**
+ * The window update ratio is used to determine when a window update must be sent. If the ratio
+ * of bytes processed since the last update has meet or exceeded this ratio then a window update will
+ * be sent. This is the global window update ratio that will be used for new streams.
+ */
+ public float windowUpdateRatio() {
+ return windowUpdateRatio;
+ }
+
+ /**
+ * The window update ratio is used to determine when a window update must be sent. If the ratio
+ * of bytes processed since the last update has meet or exceeded this ratio then a window update will
+ * be sent. This window update ratio will only be applied to {@code streamId}.
+ * <p>
+ * Note it is the responsibly of the caller to ensure that the the
+ * initial {@code SETTINGS} frame is sent before this is called. It would
+ * be considered a {@link Http2Error#PROTOCOL_ERROR} if a {@code WINDOW_UPDATE}
+ * was generated by this method before the initial {@code SETTINGS} frame is sent.
+ * @param stream the stream for which {@code ratio} applies to.
+ * @param ratio the ratio to use when checking if a {@code WINDOW_UPDATE} is determined necessary.
+ * @throws Http2Exception If a protocol-error occurs while generating {@code WINDOW_UPDATE} frames
+ */
+ public void windowUpdateRatio(Http2Stream stream, float ratio) throws Http2Exception {
+ assert ctx != null && ctx.executor().inEventLoop();
+ checkValidRatio(ratio);
+ org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.FlowState state = state(stream);
+ state.windowUpdateRatio(ratio);
+ state.writeWindowUpdateIfNeeded();
+ }
+
+ /**
+ * The window update ratio is used to determine when a window update must be sent. If the ratio
+ * of bytes processed since the last update has meet or exceeded this ratio then a window update will
+ * be sent. This window update ratio will only be applied to {@code streamId}.
+ * @throws Http2Exception If no stream corresponding to {@code stream} could be found.
+ */
+ public float windowUpdateRatio(Http2Stream stream) throws Http2Exception {
+ return state(stream).windowUpdateRatio();
+ }
+
+ @Override
+ public void receiveFlowControlledFrame(Http2Stream stream, ByteBuf data, int padding,
+ boolean endOfStream) throws Http2Exception {
+ assert ctx != null && ctx.executor().inEventLoop();
+ int dataLength = data.readableBytes() + padding;
+
+ // Apply the connection-level flow control
+ org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.FlowState connectionState = connectionState();
+ connectionState.receiveFlowControlledFrame(dataLength);
+
+ if (stream != null && !isClosed(stream)) {
+ // Apply the stream-level flow control
+ org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.FlowState state = state(stream);
+ state.endOfStream(endOfStream);
+ state.receiveFlowControlledFrame(dataLength);
+ } else if (dataLength > 0) {
+ // Immediately consume the bytes for the connection window.
+ connectionState.consumeBytes(dataLength);
+ }
+ }
+
+ private org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.FlowState connectionState() {
+ return connection.connectionStream().getProperty(stateKey);
+ }
+
+ private org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.FlowState state(Http2Stream stream) {
+ return stream.getProperty(stateKey);
+ }
+
+ private static boolean isClosed(Http2Stream stream) {
+ return stream.state() == Http2Stream.State.CLOSED;
+ }
+
+ /**
+ * Flow control state that does autorefill of the flow control window when the data is
+ * received.
+ */
+ private final class AutoRefillState extends org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.DefaultState {
+ AutoRefillState(Http2Stream stream, int initialWindowSize) {
+ super(stream, initialWindowSize);
+ }
+
+ @Override
+ public void receiveFlowControlledFrame(int dataLength) throws Http2Exception {
+ super.receiveFlowControlledFrame(dataLength);
+ // Need to call the super to consume the bytes, since this.consumeBytes does nothing.
+ super.consumeBytes(dataLength);
+ }
+
+ @Override
+ public boolean consumeBytes(int numBytes) throws Http2Exception {
+ // Do nothing, since the bytes are already consumed upon receiving the data.
+ return false;
+ }
+ }
+
+ /**
+ * Flow control window state for an individual stream.
+ */
+ private class DefaultState implements org.apache.dubbo.rpc.protocol.tri.TriHttp2LocalFlowController.FlowState {
+ private final Http2Stream stream;
+
+ /**
+ * The actual flow control window that is decremented as soon as {@code DATA} arrives.
+ */
+ private int window;
+
+ /**
+ * A view of {@link #window} that is used to determine when to send {@code WINDOW_UPDATE}
+ * frames. Decrementing this window for received {@code DATA} frames is delayed until the
+ * application has indicated that the data has been fully processed. This prevents sending
+ * a {@code WINDOW_UPDATE} until the number of processed bytes drops below the threshold.
+ */
+ private int processedWindow;
+
+ /**
+ * This is what is used to determine how many bytes need to be returned relative to {@link #processedWindow}.
+ * Each stream has their own initial window size.
+ */
+ private int initialStreamWindowSize;
+
+ /**
+ * This is used to determine when {@link #processedWindow} is sufficiently far away from
+ * {@link #initialStreamWindowSize} such that a {@code WINDOW_UPDATE} should be sent.
+ * Each stream has their own window update ratio.
+ */
+ private float streamWindowUpdateRatio;
+
+ private int lowerBound;
+ private boolean endOfStream;
+
+ DefaultState(Http2Stream stream, int initialWindowSize) {
+ this.stream = stream;
+ window(initialWindowSize);
+ streamWindowUpdateRatio = windowUpdateRatio;
+ }
+
+ @Override
+ public void window(int initialWindowSize) {
+ assert ctx == null || ctx.executor().inEventLoop();
+ window = processedWindow = initialStreamWindowSize = initialWindowSize;
+ }
+
+ @Override
+ public int windowSize() {
+ return window;
+ }
+
+ @Override
+ public int initialWindowSize() {
+ return initialStreamWindowSize;
+ }
+
+ @Override
+ public void endOfStream(boolean endOfStream) {
+ this.endOfStream = endOfStream;
+ }
+
+ @Override
+ public float windowUpdateRatio() {
+ return streamWindowUpdateRatio;
+ }
+
+ @Override
+ public void windowUpdateRatio(float ratio) {
+ assert ctx == null || ctx.executor().inEventLoop();
+ streamWindowUpdateRatio = ratio;
+ }
+
+ @Override
+ public void incrementInitialStreamWindow(int delta) {
+ // Clip the delta so that the resulting initialStreamWindowSize falls within the allowed range.
+ int newValue = (int) min(MAX_INITIAL_WINDOW_SIZE,
+ max(MIN_INITIAL_WINDOW_SIZE, initialStreamWindowSize + (long) delta));
+ delta = newValue - initialStreamWindowSize;
+
+ initialStreamWindowSize += delta;
+ }
+
+ @Override
+ public void incrementFlowControlWindows(int delta) throws Http2Exception {
+ if (delta > 0 && window > MAX_INITIAL_WINDOW_SIZE - delta) {
+ throw streamError(stream.id(), FLOW_CONTROL_ERROR,
+ "Flow control window overflowed for stream: %d", stream.id());
+ }
+
+ window += delta;
+ processedWindow += delta;
+ lowerBound = delta < 0 ? delta : 0;
+ }
+
+ @Override
+ public void receiveFlowControlledFrame(int dataLength) throws Http2Exception {
+ assert dataLength >= 0;
+
+ // Apply the delta. Even if we throw an exception we want to have taken this delta into account.
+ window -= dataLength;
+
+ // Window size can become negative if we sent a SETTINGS frame that reduces the
+ // size of the transfer window after the peer has written data frames.
+ // The value is bounded by the length that SETTINGS frame decrease the window.
+ // This difference is stored for the connection when writing the SETTINGS frame
+ // and is cleared once we send a WINDOW_UPDATE frame.
+ if (window < lowerBound) {
+ throw streamError(stream.id(), FLOW_CONTROL_ERROR,
+ "Flow control window exceeded for stream: %d", stream.id());
+ }
+ }
+
+ private void returnProcessedBytes(int delta) throws Http2Exception {
+ if (processedWindow - delta < window) {
+ throw streamError(stream.id(), INTERNAL_ERROR,
+ "Attempting to return too many bytes for stream %d", stream.id());
+ }
+ processedWindow -= delta;
+ }
+
+ @Override
+ public boolean consumeBytes(int numBytes) throws Http2Exception {
+ // Return the bytes processed and update the window.
+ returnProcessedBytes(numBytes);
+ return writeWindowUpdateIfNeeded();
+ }
+
+ @Override
+ public int unconsumedBytes() {
+ return processedWindow - window;
+ }
+
+ @Override
+ public boolean writeWindowUpdateIfNeeded() throws Http2Exception {
+ if (endOfStream || initialStreamWindowSize <= 0 ||
+ // If the stream is already closed there is no need to try to write a window update for it.
+ isClosed(stream)) {
+ return false;
+ }
+
+ int threshold = (int) (initialStreamWindowSize * streamWindowUpdateRatio);
+ if (processedWindow <= threshold) {
+ writeWindowUpdate();
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Called to perform a window update for this stream (or connection). Updates the window size back
+ * to the size of the initial window and sends a window update frame to the remote endpoint.
+ */
+ private void writeWindowUpdate() throws Http2Exception {
+ // Expand the window for this stream back to the size of the initial window.
+ int deltaWindowSize = initialStreamWindowSize - processedWindow;
+ try {
+ incrementFlowControlWindows(deltaWindowSize);
+ } catch (Throwable t) {
+ throw connectionError(INTERNAL_ERROR, t,
+ "Attempting to return too many bytes for stream %d", stream.id());
+ }
+
+ System.out.println("send window update id: " + stream.id() + " size: " + deltaWindowSize);
+ StackTraceElement[] str = Thread.currentThread().getStackTrace();
+ for(int i = 0; i< 15; i++){
+ System.out.println("send window update id: " + stream.id() + " class: " + str[i].getClassName());
Review Comment:
should be replaced to logger. sout here is not allow
--
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
[GitHub] [dubbo] EarthChen commented on a diff in pull request #10672: triple反压增加流控
Posted by GitBox <gi...@apache.org>.
EarthChen commented on code in PR #10672:
URL: https://github.com/apache/dubbo/pull/10672#discussion_r985084668
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/ServerStreamServerCallListener.java:
##########
@@ -34,11 +40,19 @@ public void onReturn(Object value) {
}
@Override
- public void onMessage(Object message) {
+ public void onMessage(Object message,DefaultHttp2WindowUpdateFrame stream, Http2Connection connection) {
if (message instanceof Object[]) {
message = ((Object[]) message)[0];
}
invocation.setArguments(new Object[]{message, responseObserver});
+ invocation.setAttachment("tri-connection",connection);
Review Comment:
Is it possible to pass parameters transparently without invocation?
--
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
[GitHub] [dubbo] asa3311 commented on a diff in pull request #10672: triple反压增加流控
Posted by GitBox <gi...@apache.org>.
asa3311 commented on code in PR #10672:
URL: https://github.com/apache/dubbo/pull/10672#discussion_r991978057
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/BiStreamServerCallListener.java:
##########
@@ -41,13 +49,27 @@ public void onReturn(Object value) {
@Override
public void onMessage(Object message) {
- if (message instanceof Object[]) {
- message = ((Object[]) message)[0];
+ if (((TripleFlowControlFrame) message).getInstance() instanceof Object[]) {
Review Comment:
in my test,it must be TripleFlowControlFrame,but I added type judgment in the latest version。
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/UnaryServerCallListener.java:
##########
@@ -37,11 +49,15 @@ public void onReturn(Object value) {
@Override
public void onMessage(Object message) {
- if (message instanceof Object[]) {
- invocation.setArguments((Object[]) message);
+ if (((TripleFlowControlFrame) message).getInstance() instanceof Object[]) {
Review Comment:
in my test,it must be TripleFlowControlFrame,but I added type judgment in the latest version。
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleHttp2FrameServerHandler.java:
##########
@@ -112,15 +112,18 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
public void onDataRead(ChannelHandlerContext ctx, Http2DataFrame msg) throws Exception {
final TripleServerStream tripleServerStream = ctx.channel().attr(SERVER_STREAM_KEY)
.get();
- tripleServerStream.transportObserver.onData(msg.content(), msg.isEndStream());
+ tripleServerStream.transportObserver.onData(msg, msg.isEndStream());
}
public void onHeadersRead(ChannelHandlerContext ctx, Http2HeadersFrame msg) throws Exception {
TripleServerStream tripleServerStream = new TripleServerStream(ctx.channel(),
frameworkModel, executor,
pathResolver, acceptEncoding, filters);
ctx.channel().attr(SERVER_STREAM_KEY).set(tripleServerStream);
- tripleServerStream.transportObserver.onHeader(msg.headers(), msg.isEndStream());
+
+ //transmit connection to triple invoke to flowcontrol
+ Http2Connection connection = (Http2Connection)ctx.channel().attr(AttributeKey.valueOf("tri-connection")).get();
Review Comment:
done
--
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