Skip to content

Commit

Permalink
cherry pick tikv#636 to release-3.3
Browse files Browse the repository at this point in the history
Signed-off-by: ti-srebot <ti-srebot@pingcap.com>
  • Loading branch information
iosmanthus authored and ti-srebot committed Jul 27, 2022
1 parent 52491a1 commit 6dac458
Show file tree
Hide file tree
Showing 5 changed files with 94 additions and 16 deletions.
4 changes: 1 addition & 3 deletions src/main/java/org/tikv/common/AbstractGRPCClient.java
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,6 @@
import org.tikv.common.util.BackOffFunction.BackOffFuncType;
import org.tikv.common.util.BackOffer;
import org.tikv.common.util.ChannelFactory;
import org.tikv.common.util.ConcreteBackOffer;

public abstract class AbstractGRPCClient<
BlockingStubT extends AbstractStub<BlockingStubT>,
Expand Down Expand Up @@ -198,8 +197,7 @@ private boolean doCheckHealth(BackOffer backOffer, String addressStr, HostMappin
}
}

protected boolean checkHealth(String addressStr, HostMapping hostMapping) {
BackOffer backOffer = ConcreteBackOffer.newCustomBackOff((int) (timeout * 2));
protected boolean checkHealth(BackOffer backOffer, String addressStr, HostMapping hostMapping) {
try {
return doCheckHealth(backOffer, addressStr, hostMapping);
} catch (Exception e) {
Expand Down
31 changes: 20 additions & 11 deletions src/main/java/org/tikv/common/PDClient.java
Original file line number Diff line number Diff line change
Expand Up @@ -440,6 +440,8 @@ PDClientWrapper getPdClientWrapper() {

private GetMembersResponse doGetMembers(BackOffer backOffer, URI uri) {
while (true) {
backOffer.checkTimeout();

try {
ManagedChannel probChan = channelFactory.getChannel(uriToAddr(uri), hostMapping);
PDGrpc.PDBlockingStub stub =
Expand All @@ -459,8 +461,7 @@ private GetMembersResponse doGetMembers(BackOffer backOffer, URI uri) {
}
}

private GetMembersResponse getMembers(URI uri) {
BackOffer backOffer = ConcreteBackOffer.newCustomBackOff(BackOffer.PD_INFO_BACKOFF);
private GetMembersResponse getMembers(BackOffer backOffer, URI uri) {
try {
return doGetMembers(backOffer, uri);
} catch (Exception e) {
Expand Down Expand Up @@ -497,11 +498,12 @@ private synchronized boolean createLeaderClientWrapper(String leaderUrlStr) {
return true;
}

synchronized boolean createFollowerClientWrapper(String followerUrlStr, String leaderUrls) {
synchronized boolean createFollowerClientWrapper(
BackOffer backOffer, String followerUrlStr, String leaderUrls) {
// TODO: Why not strip protocol info on server side since grpc does not need it

try {
if (!checkHealth(followerUrlStr, hostMapping)) {
if (!checkHealth(backOffer, followerUrlStr, hostMapping)) {
return false;
}

Expand All @@ -516,13 +518,13 @@ synchronized boolean createFollowerClientWrapper(String followerUrlStr, String l
return true;
}

public synchronized void updateLeaderOrForwardFollower() {
public synchronized void updateLeaderOrForwardFollower(BackOffer backOffer) {
if (System.currentTimeMillis() - lastUpdateLeaderTime < MIN_TRY_UPDATE_DURATION) {
return;
}
for (URI url : this.pdAddrs) {
// since resp is null, we need update leader's address by walking through all pd server.
GetMembersResponse resp = getMembers(url);
GetMembersResponse resp = getMembers(backOffer, url);
if (resp == null) {
continue;
}
Expand All @@ -534,7 +536,8 @@ public synchronized void updateLeaderOrForwardFollower() {
leaderUrlStr = uriToAddr(addrToUri(leaderUrlStr));

// if leader is switched, just return.
if (checkHealth(leaderUrlStr, hostMapping) && createLeaderClientWrapper(leaderUrlStr)) {
if (checkHealth(backOffer, leaderUrlStr, hostMapping)
&& createLeaderClientWrapper(leaderUrlStr)) {
lastUpdateLeaderTime = System.currentTimeMillis();
return;
}
Expand All @@ -561,7 +564,8 @@ public synchronized void updateLeaderOrForwardFollower() {
hasReachNextMember = true;
continue;
}
if (hasReachNextMember && createFollowerClientWrapper(followerUrlStr, leaderUrlStr)) {
if (hasReachNextMember
&& createFollowerClientWrapper(backOffer, followerUrlStr, leaderUrlStr)) {
logger.warn(
String.format("forward request to pd [%s] by pd [%s]", leaderUrlStr, followerUrlStr));
return;
Expand All @@ -577,8 +581,9 @@ public synchronized void updateLeaderOrForwardFollower() {

public void tryUpdateLeader() {
for (URI url : this.pdAddrs) {
BackOffer backOffer = defaultBackOffer();
// since resp is null, we need update leader's address by walking through all pd server.
GetMembersResponse resp = getMembers(url);
GetMembersResponse resp = getMembers(backOffer, url);
if (resp == null) {
continue;
}
Expand All @@ -591,7 +596,7 @@ public void tryUpdateLeader() {
leaderUrlStr = uriToAddr(addrToUri(leaderUrlStr));

// If leader is not change but becomes available, we can cancel follower forward.
if (checkHealth(leaderUrlStr, hostMapping) && trySwitchLeader(leaderUrlStr)) {
if (checkHealth(backOffer, leaderUrlStr, hostMapping) && trySwitchLeader(leaderUrlStr)) {
if (!urls.equals(this.pdAddrs)) {
tryUpdateMembers(urls);
}
Expand Down Expand Up @@ -705,7 +710,7 @@ private void initCluster() {
this.timeout = conf.getPdFirstGetMemberTimeout();
for (URI u : pdAddrs) {
logger.info("get members with pd " + u + ": start");
resp = getMembers(u);
resp = getMembers(defaultBackOffer(), u);
logger.info("get members with pd " + u + ": end");
if (resp != null) {
break;
Expand Down Expand Up @@ -825,4 +830,8 @@ public List<URI> getPdAddrs() {
public RequestKeyCodec getCodec() {
return codec;
}

private static BackOffer defaultBackOffer() {
return ConcreteBackOffer.newCustomBackOff(BackOffer.PD_INFO_BACKOFF);
}
}
4 changes: 2 additions & 2 deletions src/main/java/org/tikv/common/operation/PDErrorHandler.java
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ public boolean handleResponseError(BackOffer backOffer, RespT resp) {
case PD_ERROR:
backOffer.doBackOff(
BackOffFunction.BackOffFuncType.BoPDRPC, new GrpcException(error.toString()));
client.updateLeaderOrForwardFollower();
client.updateLeaderOrForwardFollower(backOffer);
return true;
case REGION_PEER_NOT_ELECTED:
logger.debug(error.getMessage());
Expand All @@ -80,7 +80,7 @@ public boolean handleRequestError(BackOffer backOffer, Exception e) {
return false;
}
backOffer.doBackOff(BackOffFunction.BackOffFuncType.BoPDRPC, e);
client.updateLeaderOrForwardFollower();
client.updateLeaderOrForwardFollower(backOffer);
return true;
}
}
11 changes: 11 additions & 0 deletions src/test/java/org/tikv/common/PDMockServerTest.java
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;
import org.junit.After;
import org.junit.Before;
import org.tikv.common.TiConfiguration.ApiVersion;
Expand Down Expand Up @@ -51,6 +52,16 @@ void upgradeToV2Cluster() throws Exception {
session = TiSession.create(conf);
}

void updateConf(Function<TiConfiguration, TiConfiguration> update) throws Exception {
if (session == null) {
throw new IllegalStateException("Cluster is not initialized");
}

session.close();

session = TiSession.create(update.apply(session.getConf()));
}

void setup(String addr) throws IOException {
int[] ports = new int[3];
for (int i = 0; i < ports.length; i++) {
Expand Down
60 changes: 60 additions & 0 deletions src/test/java/org/tikv/common/TimeoutTest.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,60 @@
/*
* Copyright 2022 TiKV Project Authors.
*
* Licensed 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.tikv.common;

import com.google.protobuf.ByteString;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.tikv.raw.RawKVClient;

public class TimeoutTest extends MockThreeStoresTest {
@Before
public void init() throws Exception {
updateConf(
conf -> {
conf.setEnableAtomicForCAS(true);
conf.setTimeout(150);
conf.setForwardTimeout(200);
conf.setRawKVReadTimeoutInMS(400);
conf.setRawKVWriteTimeoutInMS(400);
conf.setRawKVBatchReadTimeoutInMS(400);
conf.setRawKVBatchWriteTimeoutInMS(400);
conf.setRawKVWriteSlowLogInMS(50);
conf.setRawKVReadSlowLogInMS(50);
conf.setRawKVBatchReadSlowLogInMS(50);
conf.setRawKVBatchWriteSlowLogInMS(50);
return conf;
});
}

private RawKVClient createClient() {
return session.createRawClient();
}

@Test
public void testTimeoutInTime() {
try (RawKVClient client = createClient()) {
pdServers.get(0).stop();
long start = System.currentTimeMillis();
client.get(ByteString.copyFromUtf8("key"));
long end = System.currentTimeMillis();
Assert.assertTrue(end - start < session.getConf().getRawKVReadTimeoutInMS() * 2L);
}
}
}

0 comments on commit 6dac458

Please sign in to comment.