You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by al...@apache.org on 2022/11/28 06:20:44 UTC
[dubbo] branch 3.2 updated: client flowcontrol (#11029)
This is an automated email from the ASF dual-hosted git repository.
albumenj pushed a commit to branch 3.2
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.2 by this push:
new b67db81e2e client flowcontrol (#11029)
b67db81e2e is described below
commit b67db81e2ea0a35ac4b9ee18e93d418d9bfb5426
Author: xin gu <41...@qq.com>
AuthorDate: Mon Nov 28 14:20:19 2022 +0800
client flowcontrol (#11029)
---
.licenserc.yaml | 1 +
.../protocol/tri/TriHttp2RemoteFlowController.java | 795 +++++++++++++++++++++
.../rpc/protocol/tri/TripleHttp2Protocol.java | 4 +
.../rpc/protocol/tri/call/TripleClientCall.java | 17 +-
pom.xml | 2 +
5 files changed, 818 insertions(+), 1 deletion(-)
diff --git a/.licenserc.yaml b/.licenserc.yaml
index dc186df636..561562b7f0 100644
--- a/.licenserc.yaml
+++ b/.licenserc.yaml
@@ -77,6 +77,7 @@ header:
- 'dubbo-common/src/main/java/org/apache/dubbo/common/utils/Utf8Utils.java'
- 'dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/EmbeddedZooKeeper.java'
- 'dubbo-test/dubbo-test-common/src/main/java/org/apache/dubbo/test/common/utils/TestSocketUtils.java'
+ - 'dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TriHttp2RemoteFlowController.java'
comment: on-failure
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TriHttp2RemoteFlowController.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TriHttp2RemoteFlowController.java
new file mode 100644
index 0000000000..3cbb08bf09
--- /dev/null
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TriHttp2RemoteFlowController.java
@@ -0,0 +1,795 @@
+/*
+ * Copyright 2014 The Netty Project
+ *
+ * The Netty Project 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 io.netty.channel.ChannelHandlerContext;
+import io.netty.util.internal.UnstableApi;
+import io.netty.util.internal.logging.InternalLogger;
+import io.netty.util.internal.logging.InternalLoggerFactory;
+import java.util.ArrayDeque;
+import java.util.Deque;
+import io.netty.handler.codec.http2.Http2Error;
+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.Http2Error.STREAM_CLOSED;
+import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_WINDOW_SIZE;
+import static io.netty.handler.codec.http2.Http2CodecUtil.MIN_WEIGHT;
+import static io.netty.handler.codec.http2.Http2CodecUtil.MAX_WEIGHT;
+import io.netty.handler.codec.http2.WeightedFairQueueByteDistributor;
+import io.netty.handler.codec.http2.StreamByteDistributor;
+import io.netty.handler.codec.http2.Http2StreamVisitor;
+import io.netty.handler.codec.http2.Http2Exception;
+import io.netty.handler.codec.http2.Http2ConnectionAdapter;
+import io.netty.handler.codec.http2.Http2Connection;
+import io.netty.handler.codec.http2.Http2RemoteFlowController;
+import io.netty.handler.codec.http2.Http2Stream;
+import org.apache.dubbo.common.config.Configuration;
+import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import static io.netty.handler.codec.http2.Http2Exception.streamError;
+import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_LOCAL;
+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 static org.apache.dubbo.rpc.Constants.H2_SETTINGS_INITIAL_WINDOW_SIZE_KEY;
+
+/**
+ * This design is learning from {@see io.netty.handler.codec.http2.DefaultHttp2RemoteFlowController} which is in Netty.
+ */
+@UnstableApi
+public class TriHttp2RemoteFlowController implements Http2RemoteFlowController {
+ private static final InternalLogger logger =
+ InternalLoggerFactory.getInstance(TriHttp2RemoteFlowController.class);
+ private static final int MIN_WRITABLE_CHUNK = 32 * 1024;
+ private final Http2Connection connection;
+ private final Http2Connection.PropertyKey stateKey;
+ private final StreamByteDistributor streamByteDistributor;
+ private final FlowState connectionState;
+ private Configuration config = ConfigurationUtils.getGlobalConfiguration(
+ ApplicationModel.defaultModel());
+ private int initialWindowSize = config.getInt(H2_SETTINGS_INITIAL_WINDOW_SIZE_KEY, DEFAULT_WINDOW_SIZE);
+ private WritabilityMonitor monitor;
+ private ChannelHandlerContext ctx;
+
+ public TriHttp2RemoteFlowController(Http2Connection connection) {
+ this(connection, (Listener) null);
+ }
+
+ public TriHttp2RemoteFlowController(Http2Connection connection,
+ StreamByteDistributor streamByteDistributor) {
+ this(connection, streamByteDistributor, null);
+ }
+
+ public TriHttp2RemoteFlowController(Http2Connection connection, final Listener listener) {
+ this(connection, new WeightedFairQueueByteDistributor(connection), listener);
+ }
+
+ public TriHttp2RemoteFlowController(Http2Connection connection,
+ StreamByteDistributor streamByteDistributor,
+ final Listener listener) {
+ this.connection = checkNotNull(connection, "connection");
+ this.streamByteDistributor = checkNotNull(streamByteDistributor, "streamWriteDistributor");
+
+ // Add a flow state for the connection.
+ stateKey = connection.newKey();
+ connectionState = new FlowState(connection.connectionStream());
+ connection.connectionStream().setProperty(stateKey, connectionState);
+
+ // Monitor may depend upon connectionState, and so initialize after connectionState
+ listener(listener);
+ monitor.windowSize(connectionState, initialWindowSize);
+
+ // Register for notification of new streams.
+ connection.addListener(new Http2ConnectionAdapter() {
+ @Override
+ public void onStreamAdded(Http2Stream stream) {
+ // If the stream state is not open then the stream is not yet eligible for flow controlled frames and
+ // only requires the ReducedFlowState. Otherwise the full amount of memory is required.
+ stream.setProperty(stateKey, new FlowState(stream));
+ }
+
+ @Override
+ public void onStreamActive(Http2Stream stream) {
+ // If the object was previously created, but later activated then we have to ensure the proper
+ // initialWindowSize is used.
+ monitor.windowSize(state(stream), initialWindowSize);
+ }
+
+ @Override
+ public void onStreamClosed(Http2Stream stream) {
+ // Any pending frames can never be written, cancel and
+ // write errors for any pending frames.
+ state(stream).cancel(STREAM_CLOSED, null);
+ }
+
+ @Override
+ public void onStreamHalfClosed(Http2Stream stream) {
+ if (HALF_CLOSED_LOCAL == stream.state()) {
+ /*
+ * When this method is called there should not be any
+ * pending frames left if the API is used correctly. However,
+ * it is possible that a erroneous application can sneak
+ * in a frame even after having already written a frame with the
+ * END_STREAM flag set, as the stream state might not transition
+ * immediately to HALF_CLOSED_LOCAL / CLOSED due to flow control
+ * delaying the write.
+ *
+ * This is to cancel any such illegal writes.
+ */
+ state(stream).cancel(STREAM_CLOSED, null);
+ }
+ }
+ });
+ }
+
+ /**
+ * {@inheritDoc}
+ * <p>
+ * Any queued {@link FlowControlled} objects will be sent.
+ */
+ @Override
+ public void channelHandlerContext(ChannelHandlerContext ctx) throws Http2Exception {
+ this.ctx = checkNotNull(ctx, "ctx");
+
+ // Writing the pending bytes will not check writability change and instead a writability change notification
+ // to be provided by an explicit call.
+ channelWritabilityChanged();
+
+ // Don't worry about cleaning up queued frames here if ctx is null. It is expected that all streams will be
+ // closed and the queue cleanup will occur when the stream state transitions occur.
+
+ // If any frames have been queued up, we should send them now that we have a channel context.
+ if (isChannelWritable()) {
+ writePendingBytes();
+ }
+ }
+
+ @Override
+ public ChannelHandlerContext channelHandlerContext() {
+ return ctx;
+ }
+
+ @Override
+ public void initialWindowSize(int newWindowSize) throws Http2Exception {
+ assert ctx == null || ctx.executor().inEventLoop();
+ monitor.initialWindowSize(newWindowSize);
+ }
+
+ @Override
+ public int initialWindowSize() {
+ return initialWindowSize;
+ }
+
+ @Override
+ public int windowSize(Http2Stream stream) {
+ return state(stream).windowSize();
+ }
+
+ @Override
+ public boolean isWritable(Http2Stream stream) {
+ return monitor.isWritable(state(stream));
+ }
+
+ @Override
+ public void channelWritabilityChanged() throws Http2Exception {
+ monitor.channelWritabilityChange();
+ }
+
+ @Override
+ public void updateDependencyTree(int childStreamId, int parentStreamId, short weight, boolean exclusive) {
+ // It is assumed there are all validated at a higher level. For example in the Http2FrameReader.
+ assert weight >= MIN_WEIGHT && weight <= MAX_WEIGHT : "Invalid weight";
+ assert childStreamId != parentStreamId : "A stream cannot depend on itself";
+ assert childStreamId > 0 && parentStreamId >= 0 : "childStreamId must be > 0. parentStreamId must be >= 0.";
+
+ streamByteDistributor.updateDependencyTree(childStreamId, parentStreamId, weight, exclusive);
+ }
+
+ private boolean isChannelWritable() {
+ return ctx != null && isChannelWritable0();
+ }
+
+ private boolean isChannelWritable0() {
+ return ctx.channel().isWritable();
+ }
+
+ @Override
+ public void listener(Listener listener) {
+ monitor = listener == null ? new WritabilityMonitor() : new ListenerWritabilityMonitor(listener);
+ }
+
+ @Override
+ public void incrementWindowSize(Http2Stream stream, int delta) throws Http2Exception {
+ assert ctx == null || ctx.executor().inEventLoop();
+ monitor.incrementWindowSize(state(stream), delta);
+ }
+
+ @Override
+ public void addFlowControlled(Http2Stream stream, FlowControlled frame) {
+ // The context can be null assuming the frame will be queued and send later when the context is set.
+ assert ctx == null || ctx.executor().inEventLoop();
+ checkNotNull(frame, "frame");
+ try {
+ monitor.enqueueFrame(state(stream), frame);
+ } catch (Throwable t) {
+ frame.error(ctx, t);
+ }
+ }
+
+ @Override
+ public boolean hasFlowControlled(Http2Stream stream) {
+ return state(stream).hasFrame();
+ }
+
+ private FlowState state(Http2Stream stream) {
+ return (FlowState) stream.getProperty(stateKey);
+ }
+
+ /**
+ * Returns the flow control window for the entire connection.
+ */
+ private int connectionWindowSize() {
+ return connectionState.windowSize();
+ }
+
+ private int minUsableChannelBytes() {
+ // The current allocation algorithm values "fairness" and doesn't give any consideration to "goodput". It
+ // is possible that 1 byte will be allocated to many streams. In an effort to try to make "goodput"
+ // reasonable with the current allocation algorithm we have this "cheap" check up front to ensure there is
+ // an "adequate" amount of connection window before allocation is attempted. This is not foolproof as if the
+ // number of streams is >= this minimal number then we may still have the issue, but the idea is to narrow the
+ // circumstances in which this can happen without rewriting the allocation algorithm.
+ return max(ctx.channel().config().getWriteBufferLowWaterMark(), MIN_WRITABLE_CHUNK);
+ }
+
+ private int maxUsableChannelBytes() {
+ // If the channel isWritable, allow at least minUsableChannelBytes.
+ int channelWritableBytes = (int) min(Integer.MAX_VALUE, ctx.channel().bytesBeforeUnwritable());
+ int usableBytes = channelWritableBytes > 0 ? max(channelWritableBytes, minUsableChannelBytes()) : 0;
+
+ // Clip the usable bytes by the connection window.
+ return min(connectionState.windowSize(), usableBytes);
+ }
+
+ /**
+ * The amount of bytes that can be supported by underlying {@link io.netty.channel.Channel} without
+ * queuing "too-much".
+ */
+ private int writableBytes() {
+ return min(connectionWindowSize(), maxUsableChannelBytes());
+ }
+
+ @Override
+ public void writePendingBytes() throws Http2Exception {
+ monitor.writePendingBytes();
+ }
+
+ /**
+ * The remote flow control state for a single stream.
+ */
+ private final class FlowState implements StreamByteDistributor.StreamState {
+ private final Http2Stream stream;
+ private final Deque<FlowControlled> pendingWriteQueue;
+ private int window;
+ private long pendingBytes;
+ private boolean markedWritable;
+
+ /**
+ * Set to true while a frame is being written, false otherwise.
+ */
+ private boolean writing;
+ /**
+ * Set to true if cancel() was called.
+ */
+ private boolean cancelled;
+
+ FlowState(Http2Stream stream) {
+ this.stream = stream;
+ pendingWriteQueue = new ArrayDeque<FlowControlled>(2);
+ }
+
+ /**
+ * Determine if the stream associated with this object is writable.
+ * @return {@code true} if the stream associated with this object is writable.
+ */
+ boolean isWritable() {
+ return windowSize() > pendingBytes() && !cancelled;
+ }
+
+ /**
+ * The stream this state is associated with.
+ */
+ @Override
+ public Http2Stream stream() {
+ return stream;
+ }
+
+ /**
+ * Returns the parameter from the last call to {@link #markedWritability(boolean)}.
+ */
+ boolean markedWritability() {
+ return markedWritable;
+ }
+
+ /**
+ * Save the state of writability.
+ */
+ void markedWritability(boolean isWritable) {
+ this.markedWritable = isWritable;
+ }
+
+ @Override
+ public int windowSize() {
+ return window;
+ }
+
+ /**
+ * Reset the window size for this stream.
+ */
+ void windowSize(int initialWindowSize) {
+ window = initialWindowSize;
+ }
+
+ /**
+ * Write the allocated bytes for this stream.
+ * @return the number of bytes written for a stream or {@code -1} if no write occurred.
+ */
+ int writeAllocatedBytes(int allocated) {
+ final int initialAllocated = allocated;
+ int writtenBytes;
+ // In case an exception is thrown we want to remember it and pass it to cancel(Throwable).
+ Throwable cause = null;
+ FlowControlled frame;
+ try {
+ assert !writing;
+ writing = true;
+
+ // Write the remainder of frames that we are allowed to
+ boolean writeOccurred = false;
+ while (!cancelled && (frame = peek()) != null) {
+ int maxBytes = min(allocated, writableWindow());
+ if (maxBytes <= 0 && frame.size() > 0) {
+ // The frame still has data, but the amount of allocated bytes has been exhausted.
+ // Don't write needless empty frames.
+ break;
+ }
+ writeOccurred = true;
+ int initialFrameSize = frame.size();
+ try {
+ frame.write(ctx, max(0, maxBytes));
+ if (frame.size() == 0) {
+ // This frame has been fully written, remove this frame and notify it.
+ // Since we remove this frame first, we're guaranteed that its error
+ // method will not be called when we call cancel.
+ pendingWriteQueue.remove();
+ frame.writeComplete();
+ }
+ } finally {
+ // Decrement allocated by how much was actually written.
+ allocated -= initialFrameSize - frame.size();
+ }
+ }
+
+ if (!writeOccurred) {
+ // Either there was no frame, or the amount of allocated bytes has been exhausted.
+ return -1;
+ }
+
+ } catch (Throwable t) {
+ // Mark the state as cancelled, we'll clear the pending queue via cancel() below.
+ cancelled = true;
+ cause = t;
+ } finally {
+ writing = false;
+ // Make sure we always decrement the flow control windows
+ // by the bytes written.
+ writtenBytes = initialAllocated - allocated;
+
+ decrementPendingBytes(writtenBytes, false);
+ decrementFlowControlWindow(writtenBytes);
+
+ // If a cancellation occurred while writing, call cancel again to
+ // clear and error all of the pending writes.
+ if (cancelled) {
+ cancel(INTERNAL_ERROR, cause);
+ }
+ if(monitor.isOverFlowControl()){
+ cause = new Throwable();
+ cancel(FLOW_CONTROL_ERROR,cause);
+ }
+ }
+ return writtenBytes;
+ }
+
+ /**
+ * Increments the flow control window for this stream by the given delta and returns the new value.
+ */
+ int incrementStreamWindow(int delta) throws Http2Exception {
+ if (delta > 0 && Integer.MAX_VALUE - delta < window) {
+ throw streamError(stream.id(), FLOW_CONTROL_ERROR,
+ "Window size overflow for stream: %d", stream.id());
+ }
+ window += delta;
+ streamByteDistributor.updateStreamableBytes(this);
+ return window;
+ }
+
+ /**
+ * Returns the maximum writable window (minimum of the stream and connection windows).
+ */
+ private int writableWindow() {
+ return min(window, connectionWindowSize());
+ }
+
+ @Override
+ public long pendingBytes() {
+ return pendingBytes;
+ }
+
+ /**
+ * Adds the {@code frame} to the pending queue and increments the pending byte count.
+ */
+ void enqueueFrame(FlowControlled frame) {
+ FlowControlled last = pendingWriteQueue.peekLast();
+ if (last == null) {
+ enqueueFrameWithoutMerge(frame);
+ return;
+ }
+
+ int lastSize = last.size();
+ if (last.merge(ctx, frame)) {
+ incrementPendingBytes(last.size() - lastSize, true);
+ return;
+ }
+ enqueueFrameWithoutMerge(frame);
+ }
+
+ private void enqueueFrameWithoutMerge(FlowControlled frame) {
+ pendingWriteQueue.offer(frame);
+ // This must be called after adding to the queue in order so that hasFrame() is
+ // updated before updating the stream state.
+ incrementPendingBytes(frame.size(), true);
+ }
+
+ @Override
+ public boolean hasFrame() {
+ return !pendingWriteQueue.isEmpty();
+ }
+
+ /**
+ * Returns the head of the pending queue, or {@code null} if empty.
+ */
+ private FlowControlled peek() {
+ return pendingWriteQueue.peek();
+ }
+
+ /**
+ * Clears the pending queue and writes errors for each remaining frame.
+ * @param error the {@link Http2Error} to use.
+ * @param cause the {@link Throwable} that caused this method to be invoked.
+ */
+ void cancel(Http2Error error, Throwable cause) {
+ cancelled = true;
+ // Ensure that the queue can't be modified while we are writing.
+ if (writing) {
+ return;
+ }
+
+ FlowControlled frame = pendingWriteQueue.poll();
+ if (frame != null) {
+ // Only create exception once and reuse to reduce overhead of filling in the stacktrace.
+ final Http2Exception exception = streamError(stream.id(), error, cause,
+ "Stream closed before write could take place");
+ do {
+ writeError(frame, exception);
+ frame = pendingWriteQueue.poll();
+ } while (frame != null);
+ }
+
+ streamByteDistributor.updateStreamableBytes(this);
+
+ monitor.stateCancelled(this);
+ }
+
+ /**
+ * Increments the number of pending bytes for this node and optionally updates the
+ * {@link StreamByteDistributor}.
+ */
+ private void incrementPendingBytes(int numBytes, boolean updateStreamableBytes) {
+ pendingBytes += numBytes;
+ monitor.incrementPendingBytes(numBytes);
+ if (updateStreamableBytes) {
+ streamByteDistributor.updateStreamableBytes(this);
+ }
+ }
+
+ /**
+ * If this frame is in the pending queue, decrements the number of pending bytes for the stream.
+ */
+ private void decrementPendingBytes(int bytes, boolean updateStreamableBytes) {
+ incrementPendingBytes(-bytes, updateStreamableBytes);
+ }
+
+ /**
+ * Decrement the per stream and connection flow control window by {@code bytes}.
+ */
+ private void decrementFlowControlWindow(int bytes) {
+ try {
+ int negativeBytes = -bytes;
+ connectionState.incrementStreamWindow(negativeBytes);
+ incrementStreamWindow(negativeBytes);
+ } catch (Http2Exception e) {
+ // Should never get here since we're decrementing.
+ throw new IllegalStateException("Invalid window state when writing frame: " + e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Discards this {@link FlowControlled}, writing an error. If this frame is in the pending queue,
+ * the unwritten bytes are removed from this branch of the priority tree.
+ */
+ private void writeError(FlowControlled frame, Http2Exception cause) {
+ assert ctx != null;
+ decrementPendingBytes(frame.size(), true);
+ frame.error(ctx, cause);
+ }
+ }
+
+ /**
+ * Abstract class which provides common functionality for writability monitor implementations.
+ */
+ private class WritabilityMonitor implements StreamByteDistributor.Writer {
+ private boolean inWritePendingBytes;
+ private long totalPendingBytes;
+
+ @Override
+ public final void write(Http2Stream stream, int numBytes) {
+ state(stream).writeAllocatedBytes(numBytes);
+ }
+
+ /**
+ * Called when the writability of the underlying channel changes.
+ * @throws Http2Exception If a write occurs and an exception happens in the write operation.
+ */
+ void channelWritabilityChange() throws Http2Exception { }
+
+ /**
+ * Called when the state is cancelled.
+ * @param state the state that was cancelled.
+ */
+ void stateCancelled(FlowState state) { }
+
+ /**
+ * Set the initial window size for {@code state}.
+ * @param state the state to change the initial window size for.
+ * @param initialWindowSize the size of the window in bytes.
+ */
+ void windowSize(FlowState state, int initialWindowSize) {
+ state.windowSize(initialWindowSize);
+ }
+
+ /**
+ * Increment the window size for a particular stream.
+ * @param state the state associated with the stream whose window is being incremented.
+ * @param delta The amount to increment by.
+ * @throws Http2Exception If this operation overflows the window for {@code state}.
+ */
+ void incrementWindowSize(FlowState state, int delta) throws Http2Exception {
+ state.incrementStreamWindow(delta);
+ }
+
+ /**
+ * Add a frame to be sent via flow control.
+ * @param state The state associated with the stream which the {@code frame} is associated with.
+ * @param frame the frame to enqueue.
+ * @throws Http2Exception If a writability error occurs.
+ */
+ void enqueueFrame(FlowState state, FlowControlled frame) throws Http2Exception {
+ state.enqueueFrame(frame);
+ }
+
+ /**
+ * Increment the total amount of pending bytes for all streams. When any stream's pending bytes changes
+ * method should be called.
+ * @param delta The amount to increment by.
+ */
+ final void incrementPendingBytes(int delta) {
+ totalPendingBytes += delta;
+
+ // Notification of writibilty change should be delayed until the end of the top level event.
+ // This is to ensure the flow controller is more consistent state before calling external listener methods.
+ }
+
+ /**
+ * Determine if the stream associated with {@code state} is writable.
+ * @param state The state which is associated with the stream to test writability for.
+ * @return {@code true} if {@link FlowState#stream()} is writable. {@code false} otherwise.
+ */
+ final boolean isWritable(FlowState state) {
+ return isWritableConnection() && state.isWritable();
+ }
+
+ final void writePendingBytes() throws Http2Exception {
+ // Reentry is not permitted during the byte distribution process. It may lead to undesirable distribution of
+ // bytes and even infinite loops. We protect against reentry and make sure each call has an opportunity to
+ // cause a distribution to occur. This may be useful for example if the channel's writability changes from
+ // Writable -> Not Writable (because we are writing) -> Writable (because the user flushed to make more room
+ // in the channel outbound buffer).
+ if (inWritePendingBytes) {
+ return;
+ }
+ inWritePendingBytes = true;
+ try {
+ int bytesToWrite = writableBytes();
+ // Make sure we always write at least once, regardless if we have bytesToWrite or not.
+ // This ensures that zero-length frames will always be written.
+ for (;;) {
+ if (!streamByteDistributor.distribute(bytesToWrite, this) ||
+ (bytesToWrite = writableBytes()) <= 0 ||
+ !isChannelWritable0()) {
+ break;
+ }
+ }
+ } finally {
+ inWritePendingBytes = false;
+ }
+ }
+
+ void initialWindowSize(int newWindowSize) throws Http2Exception {
+ checkPositiveOrZero(newWindowSize, "newWindowSize");
+
+ final int delta = newWindowSize - initialWindowSize;
+ initialWindowSize = newWindowSize;
+ connection.forEachActiveStream(new Http2StreamVisitor() {
+ @Override
+ public boolean visit(Http2Stream stream) throws Http2Exception {
+ state(stream).incrementStreamWindow(delta);
+ return true;
+ }
+ });
+
+ if (delta > 0 && isChannelWritable()) {
+ // The window size increased, send any pending frames for all streams.
+ writePendingBytes();
+ }
+ }
+
+ final boolean isWritableConnection() {
+ return connectionState.windowSize() - totalPendingBytes > 0 && isChannelWritable();
+ }
+
+ final boolean isOverFlowControl() {
+ if(connectionState.windowSize() == 0){
+ return true;
+ }else {
+ return false;
+ }
+ }
+ }
+
+ /**
+ * Writability of a {@code stream} is calculated using the following:
+ * <pre>
+ * Connection Window - Total Queued Bytes > 0 &&
+ * Stream Window - Bytes Queued for Stream > 0 &&
+ * isChannelWritable()
+ * </pre>
+ */
+ private final class ListenerWritabilityMonitor extends WritabilityMonitor implements Http2StreamVisitor {
+ private final Listener listener;
+
+ ListenerWritabilityMonitor(Listener listener) {
+ this.listener = listener;
+ }
+
+ @Override
+ public boolean visit(Http2Stream stream) throws Http2Exception {
+ FlowState state = state(stream);
+ if (isWritable(state) != state.markedWritability()) {
+ notifyWritabilityChanged(state);
+ }
+ return true;
+ }
+
+ @Override
+ void windowSize(FlowState state, int initialWindowSize) {
+ super.windowSize(state, initialWindowSize);
+ try {
+ checkStateWritability(state);
+ } catch (Http2Exception e) {
+ throw new RuntimeException("Caught unexpected exception from window", e);
+ }
+ }
+
+ @Override
+ void incrementWindowSize(FlowState state, int delta) throws Http2Exception {
+ super.incrementWindowSize(state, delta);
+ checkStateWritability(state);
+ }
+
+ @Override
+ void initialWindowSize(int newWindowSize) throws Http2Exception {
+ super.initialWindowSize(newWindowSize);
+ if (isWritableConnection()) {
+ // If the write operation does not occur we still need to check all streams because they
+ // may have transitioned from writable to not writable.
+ checkAllWritabilityChanged();
+ }
+ }
+
+ @Override
+ void enqueueFrame(FlowState state, FlowControlled frame) throws Http2Exception {
+ super.enqueueFrame(state, frame);
+ checkConnectionThenStreamWritabilityChanged(state);
+ }
+
+ @Override
+ void stateCancelled(FlowState state) {
+ try {
+ checkConnectionThenStreamWritabilityChanged(state);
+ } catch (Http2Exception e) {
+ throw new RuntimeException("Caught unexpected exception from checkAllWritabilityChanged", e);
+ }
+ }
+
+ @Override
+ void channelWritabilityChange() throws Http2Exception {
+ if (connectionState.markedWritability() != isChannelWritable()) {
+ checkAllWritabilityChanged();
+ }
+ }
+
+ private void checkStateWritability(FlowState state) throws Http2Exception {
+ if (isWritable(state) != state.markedWritability()) {
+ if (state == connectionState) {
+ checkAllWritabilityChanged();
+ } else {
+ notifyWritabilityChanged(state);
+ }
+ }
+ }
+
+ private void notifyWritabilityChanged(FlowState state) {
+ state.markedWritability(!state.markedWritability());
+ try {
+ listener.writabilityChanged(state.stream);
+ } catch (Throwable cause) {
+ logger.error("Caught Throwable from listener.writabilityChanged", cause);
+ }
+ }
+
+ private void checkConnectionThenStreamWritabilityChanged(FlowState state) throws Http2Exception {
+ // It is possible that the connection window and/or the individual stream writability could change.
+ if (isWritableConnection() != connectionState.markedWritability()) {
+ checkAllWritabilityChanged();
+ } else if (isWritable(state) != state.markedWritability()) {
+ notifyWritabilityChanged(state);
+ }else if(isOverFlowControl()){
+ throw streamError(state.stream().id(), FLOW_CONTROL_ERROR,
+ "TotalPendingBytes size overflow for stream: %d", state.stream().id());
+ }
+ }
+
+ private void checkAllWritabilityChanged() throws Http2Exception {
+ // Make sure we mark that we have notified as a result of this change.
+ connectionState.markedWritability(isWritableConnection());
+ connection.forEachActiveStream(this);
+ }
+ }
+}
+
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java
index 4516f27a24..3270ed5c27 100644
--- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java
@@ -129,6 +129,8 @@ public class TripleHttp2Protocol extends AbstractWireProtocol implements ScopeMo
.frameLogger(SERVER_LOGGER)
.build();
ExecutorSupport executorSupport = ExecutorRepository.getInstance(url.getOrDefaultApplicationModel()).getExecutorSupport(url);
+ codec.connection().remote().flowController(new TriHttp2RemoteFlowController(codec.connection()));
+ codec.connection().local().flowController().frameWriter(codec.encoder().frameWriter());
TripleWriteQueue writeQueue = new TripleWriteQueue();
final Http2MultiplexHandler handler = new Http2MultiplexHandler(
new ChannelInitializer<Http2StreamChannel>() {
@@ -173,6 +175,8 @@ public class TripleHttp2Protocol extends AbstractWireProtocol implements ScopeMo
DEFAULT_MAX_HEADER_LIST_SIZE)))
.frameLogger(CLIENT_LOGGER)
.build();
+ codec.connection().remote().flowController(new TriHttp2RemoteFlowController(codec.connection()));
+ codec.connection().local().flowController().frameWriter(codec.encoder().frameWriter());
final Http2MultiplexHandler handler = new Http2MultiplexHandler(
new TripleClientHandler(frameworkModel));
List<ChannelHandler> handlers = new ArrayList<>();
diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java
index 507c35773a..e4ff156d8e 100644
--- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java
+++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/call/TripleClientCall.java
@@ -18,6 +18,7 @@
package org.apache.dubbo.rpc.protocol.tri.call;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+import io.netty.handler.codec.http2.Http2Exception;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.remoting.api.connection.AbstractConnectionClient;
@@ -50,6 +51,7 @@ import java.util.concurrent.Executor;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_RESPONSE;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_SERIALIZE_TRIPLE;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_STREAM_LISTENER;
+import static io.netty.handler.codec.http2.Http2Error.FLOW_CONTROL_ERROR;
public class TripleClientCall implements ClientCall, ClientStream.Listener {
private static final ErrorTypeAwareLogger LOGGER = LoggerFactory.getErrorTypeAwareLogger(TripleClientCall.class);
@@ -64,6 +66,7 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener {
private boolean headerSent;
private boolean autoRequest = true;
private boolean done;
+ private Http2Exception.StreamException streamException;
public TripleClientCall(AbstractConnectionClient connectionClient, Executor executor,
FrameworkModel frameworkModel, TripleWriteQueue writeQueue) {
@@ -203,6 +206,16 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener {
if (stream == null) {
return;
}
+ if(t instanceof Http2Exception.StreamException && ((Http2Exception.StreamException) t).error().equals(FLOW_CONTROL_ERROR)){
+ TriRpcStatus status = TriRpcStatus.CANCELLED.withCause(t)
+ .withDescription("Due flowcontrol over pendingbytes, Cancelled by client");
+ stream.cancelByLocal(status);
+ streamException = (Http2Exception.StreamException) t;
+ }else{
+ TriRpcStatus status = TriRpcStatus.CANCELLED.withCause(t)
+ .withDescription("Cancelled by client");
+ stream.cancelByLocal(status);
+ }
TriRpcStatus status = TriRpcStatus.CANCELLED.withCause(t)
.withDescription("Cancelled by client");
stream.cancelByLocal(status);
@@ -218,7 +231,9 @@ public class TripleClientCall implements ClientCall, ClientStream.Listener {
@Override
public void sendMessage(Object message) {
- if (canceled) {
+ if (canceled && null != streamException) {
+ throw new IllegalStateException("Due flowcontrol over pendingbytes, Call already canceled");
+ }else if (canceled) {
throw new IllegalStateException("Call already canceled");
}
if (!headerSent) {
diff --git a/pom.xml b/pom.xml
index 2b8b2b7e89..a8c4662b81 100644
--- a/pom.xml
+++ b/pom.xml
@@ -329,6 +329,7 @@
**/org/apache/dubbo/common/serialize/protobuf/support/wrapper/ThrowablePB.java,
**/org/apache/dubbo/test/common/utils/TestSocketUtils.java,
**/org/apache/dubbo/triple/TripleWrapper.java,
+ **/org/apache/dubbo/rpc/protocol/tri/TriHttp2RemoteFlowController.java,
**/istio/v1/auth/**/*,
**/com/google/rpc/*,
**/generated/**/*,
@@ -759,6 +760,7 @@
<exclude>
**/org/apache/dubbo/common/utils/Utf8Utils.java
</exclude>
+ <exclude>**/org/apache/dubbo/rpc/protocol/tri/TriHttp2RemoteFlowController.java</exclude>
<exclude>.github/**</exclude>
<exclude>compiler/**</exclude>
<exclude>**/generated/**</exclude>