natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix timeout 6008 (#6136)

This commit is contained in:
ken.lj 2020-05-12 01:29:09 +08:00 committed by GitHub
parent 9bab51e7a6
commit 5da840dcdc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 34 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java
View File

@ -110,6 +110,10 @@ public interface CommonConstants {
int DEFAULT_TIMEOUT = 1000;
// used by invocation attachments to transfer timeout from Consumer to Provider.
// works as a replacement of TIMEOUT_KEY on wire, which seems to be totally useless in previous releases).
String TIMEOUT_ATTACHENT_KEY = "_TO";
String TIME_COUNTDOWN_KEY = "timeout-countdown";
String ENABLE_TIMEOUT_COUNTDOWN_KEY = "enable-timeout-countdown";

12
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/ContextFilter.java
View File

@ -40,6 +40,7 @@ import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER;
import static org.apache.dubbo.common.constants.CommonConstants.REMOTE_APPLICATION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_ATTACHENT_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIME_COUNTDOWN_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
@ -70,6 +71,7 @@ public class ContextFilter implements Filter, Filter.Listener {
UNLOADING_KEYS.add(DUBBO_VERSION_KEY);
UNLOADING_KEYS.add(TOKEN_KEY);
UNLOADING_KEYS.add(TIMEOUT_KEY);
UNLOADING_KEYS.add(TIMEOUT_ATTACHENT_KEY);
// Remove async property to avoid being passed to the following invoke chain.
UNLOADING_KEYS.add(ASYNC_KEY);
@ -103,6 +105,11 @@ public class ContextFilter implements Filter, Filter.Listener {
context.setRemoteApplicationName((String) context.getAttachment(REMOTE_APPLICATION_KEY));
}
long timeout = RpcUtils.getTimeout(invocation, -1);
if (timeout != -1) {
context.set(TIME_COUNTDOWN_KEY, TimeoutCountDown.newCountDown(timeout, TimeUnit.MILLISECONDS));
}
// merged from dubbox
// we may already added some attachments into RpcContext before this filter (e.g. in rest protocol)
if (attachments != null) {
@ -117,11 +124,6 @@ public class ContextFilter implements Filter, Filter.Listener {
((RpcInvocation) invocation).setInvoker(invoker);
}
long timeout = RpcUtils.getTimeout(invocation, -1);
if (timeout != -1) {
context.set(TIME_COUNTDOWN_KEY, TimeoutCountDown.newCountDown(timeout, TimeUnit.MILLISECONDS));
}
try {
context.clearAfterEachInvoke(false);
return invoker.invoke(invocation);

3
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/support/RpcUtils.java
View File

@ -34,6 +34,7 @@ import java.util.concurrent.atomic.AtomicLong;
import static org.apache.dubbo.common.constants.CommonConstants.$INVOKE;
import static org.apache.dubbo.common.constants.CommonConstants.$INVOKE_ASYNC;
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_ATTACHENT_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
import static org.apache.dubbo.rpc.Constants.$ECHO;
import static org.apache.dubbo.rpc.Constants.ASYNC_KEY;
@ -216,7 +217,7 @@ public class RpcUtils {
public static long getTimeout(Invocation invocation, long defaultTimeout) {
long timeout = defaultTimeout;
Object genericTimeout = invocation.getObjectAttachment(TIMEOUT_KEY);
Object genericTimeout = invocation.getObjectAttachment(TIMEOUT_ATTACHENT_KEY);
if (genericTimeout != null) {
timeout = convertToNumber(genericTimeout, defaultTimeout);
}

3
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/ChannelWrappedInvoker.java
View File

@ -38,7 +38,6 @@ import java.util.concurrent.CompletableFuture;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
import static org.apache.dubbo.remoting.Constants.SENT_KEY;
import static org.apache.dubbo.rpc.Constants.TOKEN_KEY;
import static org.apache.dubbo.rpc.protocol.dubbo.Constants.CALLBACK_SERVICE_KEY;
@ -54,7 +53,7 @@ class ChannelWrappedInvoker<T> extends AbstractInvoker<T> {
private final ExchangeClient currentClient;
ChannelWrappedInvoker(Class<T> serviceType, Channel channel, URL url, String serviceKey) {
super(serviceType, url, new String[]{GROUP_KEY, TOKEN_KEY, TIMEOUT_KEY});
super(serviceType, url, new String[]{GROUP_KEY, TOKEN_KEY});
this.channel = channel;
this.serviceKey = serviceKey;
this.currentClient = new HeaderExchangeClient(new ChannelWrapper(this.channel), false);

33
dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvoker.java
View File

@ -47,7 +47,7 @@ import static org.apache.dubbo.common.constants.CommonConstants.ENABLE_TIMEOUT_C
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_ATTACHENT_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIME_COUNTDOWN_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
import static org.apache.dubbo.rpc.Constants.TOKEN_KEY;
@ -72,7 +72,7 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
}
public DubboInvoker(Class<T> serviceType, URL url, ExchangeClient[] clients, Set<Invoker<?>> invokers) {
super(serviceType, url, new String[]{INTERFACE_KEY, GROUP_KEY, TOKEN_KEY, TIMEOUT_KEY});
super(serviceType, url, new String[]{INTERFACE_KEY, GROUP_KEY, TOKEN_KEY});
this.clients = clients;
// get version.
this.version = url.getParameter(VERSION_KEY, "0.0.0");
@ -94,18 +94,7 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
}
try {
boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
Object countdown = RpcContext.getContext().get(TIME_COUNTDOWN_KEY);
int timeout = DEFAULT_TIMEOUT;
if (countdown == null) {
timeout = (int) RpcUtils.getTimeout(getUrl(), methodName, RpcContext.getContext(), DEFAULT_TIMEOUT);
if (getUrl().getParameter(ENABLE_TIMEOUT_COUNTDOWN_KEY, false)) {
invocation.setObjectAttachment(TIMEOUT_KEY, timeout); // pass timeout to remote server
}
} else {
TimeoutCountDown timeoutCountDown = (TimeoutCountDown) countdown;
timeout = (int) timeoutCountDown.timeRemaining(TimeUnit.MILLISECONDS);
invocation.setObjectAttachment(TIMEOUT_KEY, timeout);// pass timeout to remote server
}
int timeout = calculateTimeout(invocation, methodName);
if (isOneway) {
boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
currentClient.send(inv, isSent);
@ -172,4 +161,20 @@ public class DubboInvoker<T> extends AbstractInvoker<T> {
}
}
}
private int calculateTimeout(Invocation invocation, String methodName) {
Object countdown = RpcContext.getContext().get(TIME_COUNTDOWN_KEY);
int timeout = DEFAULT_TIMEOUT;
if (countdown == null) {
timeout = (int) RpcUtils.getTimeout(getUrl(), methodName, RpcContext.getContext(), DEFAULT_TIMEOUT);
if (getUrl().getParameter(ENABLE_TIMEOUT_COUNTDOWN_KEY, false)) {
invocation.setObjectAttachment(TIMEOUT_ATTACHENT_KEY, timeout); // pass timeout to remote server
}
} else {
TimeoutCountDown timeoutCountDown = (TimeoutCountDown) countdown;
timeout = (int) timeoutCountDown.timeRemaining(TimeUnit.MILLISECONDS);
invocation.setObjectAttachment(TIMEOUT_ATTACHENT_KEY, timeout);// pass timeout to remote server
}
return timeout;
}
}

2
dubbo-rpc/dubbo-rpc-thrift/src/main/java/org/apache/dubbo/rpc/protocol/thrift/ThriftInvoker.java
View File

@ -64,7 +64,7 @@ public class ThriftInvoker<T> extends AbstractInvoker<T> {
}
public ThriftInvoker(Class<T> type, URL url, ExchangeClient[] clients, Set<Invoker<?>> invokers) {
super(type, url, new String[]{INTERFACE_KEY, GROUP_KEY, TOKEN_KEY, TIMEOUT_KEY});
super(type, url, new String[]{INTERFACE_KEY, GROUP_KEY, TOKEN_KEY});
this.clients = clients;
this.invokers = invokers;
}