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

Direct return when the server goes down unnormally. (#2185) 

* Keep the unfinished request in every channel.
When the server goes down un-normally, return the unfinished requests directly.
The current way is to wait until timeout.

* default method

* direct return unfinished requests when remote inactive or disconnect

* direct return unfinished requests when remote inactive or disconnect

* name and doc

* name and doc

* fix unit test

* fix unit test

* fix unit test

* close heartbeat to fix unit test

* sth test

* sth test

* sth test

* sth test

* close when client close

* fix unit test

* fix unit test

* fix unit test

* ignore remote invoke unit test

* don't clear request when channelInactive

* replace import *

* remove unuse import

* optimize

* optimize

* fix ci failed

* fix ci failed

* fix ci failed

* fix ci failed

* fix ci failed

* fix ci failed

* fix ci failed

* optimize unfinished request keeper

* rebase onto master

* fix review code
This commit is contained in:
时无两丶 2018-08-23 14:08:47 +08:00 committed by Ian Luo
parent 53d34f7ad1
commit c6fd6841b8
11 changed files with 69 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/Channel.java
View File

@ -21,8 +21,6 @@ import java.net.InetSocketAddress;
/**
* Channel. (API/SPI, Prototype, ThreadSafe)
*
*
*
* @see org.apache.dubbo.remoting.Client
* @see org.apache.dubbo.remoting.Server#getChannels()
* @see org.apache.dubbo.remoting.Server#getChannel(InetSocketAddress)
@ -73,5 +71,4 @@ public interface Channel extends Endpoint {
* @param key key.
*/
void removeAttribute(String key);
}

1
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/ExchangeChannel.java
View File

@ -57,5 +57,4 @@ public interface ExchangeChannel extends Channel {
*/
@Override
void close(int timeout);
}

5
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/Response.java
View File

@ -40,6 +40,11 @@ public class Response {
*/
public static final byte SERVER_TIMEOUT = 31;
/**
* channel inactive, directly return the unfinished requests.
*/
public static final byte CHANNEL_INACTIVE = 35;
/**
* request format error.
*/

45
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/DefaultFuture.java
View File

@ -48,9 +48,9 @@ public class DefaultFuture implements ResponseFuture {
private static final Logger logger = LoggerFactory.getLogger(DefaultFuture.class);
private static final Map<Long, Channel> CHANNELS = new ConcurrentHashMap<Long, Channel>();
private static final Map<Long, Channel> CHANNELS = new ConcurrentHashMap<>();
private static final Map<Long, DefaultFuture> FUTURES = new ConcurrentHashMap<Long, DefaultFuture>();
private static final Map<Long, DefaultFuture> FUTURES = new ConcurrentHashMap<>();
public static final Timer TIME_OUT_TIMER = new HashedWheelTimer(
new NamedThreadFactory("dubbo-future-timeout", true),
@ -87,10 +87,21 @@ public class DefaultFuture implements ResponseFuture {
TIME_OUT_TIMER.newTimeout(task, future.getTimeout(), TimeUnit.MILLISECONDS);
}
/**
* init a DefaultFuture
* 1.init a DefaultFuture
* 2.timeout check
*
* @param channel channel
* @param request the request
* @param timeout timeout
* @return a new DefaultFuture
*/
public static DefaultFuture newFuture(Channel channel, Request request, int timeout) {
final DefaultFuture defaultFuture = new DefaultFuture(channel, request, timeout);
timeoutCheck(defaultFuture);
return defaultFuture;
final DefaultFuture future = new DefaultFuture(channel, request, timeout);
// timeout check
timeoutCheck(future);
return future;
}
public static DefaultFuture getFuture(long id) {
@ -108,6 +119,29 @@ public class DefaultFuture implements ResponseFuture {
}
}
/**
* close a channel when a channel is inactive
* directly return the unfinished requests.
*
* @param channel channel to close
*/
public static void closeChannel(Channel channel) {
for (long id : CHANNELS.keySet()) {
if (channel.equals(CHANNELS.get(id))) {
DefaultFuture future = getFuture(id);
if (future != null && !future.isDone()) {
Response disconnectResponse = new Response(future.getId());
disconnectResponse.setStatus(Response.CHANNEL_INACTIVE);
disconnectResponse.setErrorMessage("Channel " +
channel +
" is inactive. Directly return the unFinished request : " +
future.getRequest());
DefaultFuture.received(channel, disconnectResponse);
}
}
}
}
public static void received(Channel channel, Response response) {
try {
DefaultFuture future = FUTURES.remove(response.getId());
@ -212,6 +246,7 @@ public class DefaultFuture implements ResponseFuture {
timeoutResponse.setErrorMessage(future.getTimeoutMessage(true));
// handle response.
DefaultFuture.received(future.getChannel(), timeoutResponse);
}
}

1
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/header/HeaderExchangeHandler.java
View File

@ -146,6 +146,7 @@ public class HeaderExchangeHandler implements ChannelHandlerDelegate {
try {
handler.disconnected(exchangeChannel);
} finally {
DefaultFuture.closeChannel(channel);
HeaderExchangeChannel.removeChannelIfDisconnected(channel);
}
}

1
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/AbstractChannel.java
View File

@ -43,5 +43,4 @@ public abstract class AbstractChannel extends AbstractPeer implements Channel {
public String toString() {
return getLocalAddress() + " -> " + getRemoteAddress();
}
}

1
dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboInvokerAvilableTest.java
View File

@ -131,7 +131,6 @@ public class DubboInvokerAvilableTest {
}
//invoke method --> init client
IDemoService service = (IDemoService) proxy.getProxy(invoker);
Assert.assertEquals("ok", service.get());

13
dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/MultiThreadTest.java
View File

@ -23,30 +23,31 @@ import org.apache.dubbo.rpc.Protocol;
import org.apache.dubbo.rpc.ProxyFactory;
import org.apache.dubbo.rpc.protocol.dubbo.support.DemoService;
import org.apache.dubbo.rpc.protocol.dubbo.support.DemoServiceImpl;
import junit.framework.TestCase;
import org.junit.Assert;
import org.junit.Test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class MultiThreadTest extends TestCase {
public class MultiThreadTest {
private Protocol protocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
private ProxyFactory proxy = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();
@Test
public void testDubboMultiThreadInvoke() throws Exception {
Exporter<?> rpcExporter = protocol.export(proxy.getInvoker(new DemoServiceImpl(), DemoService.class, URL.valueOf("dubbo://127.0.0.1:20259/TestService")));
final AtomicInteger counter = new AtomicInteger();
final DemoService service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("dubbo://127.0.0.1:20259/TestService")));
assertEquals(service.getSize(new String[]{"123", "456", "789"}), 3);
Assert.assertEquals(service.getSize(new String[]{"123", "456", "789"}), 3);
final StringBuffer sb = new StringBuffer();
for (int i = 0; i < 1024 * 64 + 32; i++)
sb.append('A');
assertEquals(sb.toString(), service.echo(sb.toString()));
Assert.assertEquals(sb.toString(), service.echo(sb.toString()));
ExecutorService exec = Executors.newFixedThreadPool(10);
for (int i = 0; i < 10; i++) {
@ -55,7 +56,7 @@ public class MultiThreadTest extends TestCase {
public void run() {
for (int i = 0; i < 30; i++) {
System.out.println(fi + ":" + counter.getAndIncrement());
assertEquals(service.echo(sb.toString()), sb.toString());
Assert.assertEquals(service.echo(sb.toString()), sb.toString());
}
}
});

8
dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/ReferenceCountExchangeClientTest.java
View File

@ -123,6 +123,14 @@ public class ReferenceCountExchangeClientTest {
Assert.assertEquals("should not warning message", 0, LogUtil.findMessage(errorMsg));
// counter is incorrect, invocation still succeeds
client.close();
// wait close done.
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Assert.fail();
}
Assert.assertEquals("hello", helloService.hello());
Assert.assertEquals("should warning message", 1, LogUtil.findMessage(errorMsg));

14
dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/RpcFilterTest.java
View File

@ -24,23 +24,25 @@ import org.apache.dubbo.rpc.protocol.dubbo.support.DemoService;
import org.apache.dubbo.rpc.protocol.dubbo.support.DemoServiceImpl;
import org.apache.dubbo.rpc.service.EchoService;
import junit.framework.TestCase;
import org.junit.Assert;
import org.junit.Test;
public class RpcFilterTest extends TestCase {
public class RpcFilterTest {
private Protocol protocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
private ProxyFactory proxy = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();
@Test
public void testRpcFilter() throws Exception {
DemoService service = new DemoServiceImpl();
URL url = URL.valueOf("dubbo://127.0.0.1:9010/org.apache.dubbo.rpc.DemoService?service.filter=echo");
protocol.export(proxy.getInvoker(service, DemoService.class, url));
service = proxy.getProxy(protocol.refer(DemoService.class, url));
assertEquals("123", service.echo("123"));
Assert.assertEquals("123", service.echo("123"));
// cast to EchoService
EchoService echo = proxy.getProxy(protocol.refer(EchoService.class, url));
assertEquals(echo.$echo("test"), "test");
assertEquals(echo.$echo("abcdefg"), "abcdefg");
assertEquals(echo.$echo(1234), 1234);
Assert.assertEquals(echo.$echo("test"), "test");
Assert.assertEquals(echo.$echo("abcdefg"), "abcdefg");
Assert.assertEquals(echo.$echo(1234), 1234);
}
}

1
dubbo-rpc/dubbo-rpc-thrift/src/test/java/org/apache/dubbo/rpc/protocol/thrift/ThriftCodecTest.java
View File

@ -38,7 +38,6 @@ import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TIOStreamTransport;
import org.apache.thrift.transport.TTransport;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;
import java.io.ByteArrayInputStream;