client flowcontrol (#11029)
This commit is contained in:
xin gu
GitHub
parent
247eeb39a5
commit
b67db81e2e
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -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<>();
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
2
pom.xml
2
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>
|
||||
|
|
|
|||
Loading…
Reference in New Issue