mirror of
https://github.com/apple/foundationdb.git
synced 2025-05-15 02:18:39 +08:00
After a proxy dies, it will remain alive for an additional 10 seconds to forward clients to the new proxies
This commit is contained in:
parent
e4193b125b
commit
c348b3da51
@ -8,7 +8,6 @@ set(FDBCLIENT_SRCS
|
||||
BackupContainer.actor.cpp
|
||||
BackupContainer.h
|
||||
BlobStore.actor.cpp
|
||||
ClientDBInfo.h
|
||||
ClientLogEvents.h
|
||||
ClientWorkerInterface.h
|
||||
ClusterInterface.h
|
||||
|
@ -1,49 +0,0 @@
|
||||
/*
|
||||
* ClientDBInfo.h
|
||||
*
|
||||
* This source file is part of the FoundationDB open source project
|
||||
*
|
||||
* Copyright 2013-2018 Apple Inc. and the FoundationDB 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.
|
||||
*/
|
||||
|
||||
#ifndef FDBCLIENT_CLIENTDBINFO_H
|
||||
#define FDBCLIENT_CLIENTDBINFO_H
|
||||
#pragma once
|
||||
|
||||
#include "fdbclient/MasterProxyInterface.h"
|
||||
|
||||
// ClientDBInfo is all the information needed by a database client to access the database
|
||||
// It is returned (and kept up to date) by the OpenDatabaseRequest interface of ClusterInterface
|
||||
struct ClientDBInfo {
|
||||
constexpr static FileIdentifier file_identifier = 5355080;
|
||||
UID id; // Changes each time anything else changes
|
||||
vector< MasterProxyInterface > proxies;
|
||||
double clientTxnInfoSampleRate;
|
||||
int64_t clientTxnInfoSizeLimit;
|
||||
ClientDBInfo() : clientTxnInfoSampleRate(std::numeric_limits<double>::infinity()), clientTxnInfoSizeLimit(-1) {}
|
||||
|
||||
bool operator == (ClientDBInfo const& r) const { return id == r.id; }
|
||||
bool operator != (ClientDBInfo const& r) const { return id != r.id; }
|
||||
|
||||
template <class Archive>
|
||||
void serialize(Archive& ar) {
|
||||
if constexpr (!is_fb_function<Archive>) {
|
||||
ASSERT(ar.protocolVersion().isValid());
|
||||
}
|
||||
serializer(ar, proxies, id, clientTxnInfoSampleRate, clientTxnInfoSizeLimit);
|
||||
}
|
||||
};
|
||||
|
||||
#endif
|
@ -25,7 +25,7 @@
|
||||
#include "fdbclient/FDBTypes.h"
|
||||
#include "fdbrpc/FailureMonitor.h"
|
||||
#include "fdbclient/Status.h"
|
||||
#include "fdbclient/ClientDBInfo.h"
|
||||
#include "fdbclient/MasterProxyInterface.h"
|
||||
|
||||
// Streams from WorkerInterface that are safe and useful to call from a client.
|
||||
// A ClientWorkerInterface is embedded as the first element of a WorkerInterface.
|
||||
|
@ -25,7 +25,7 @@
|
||||
#include "fdbclient/FDBTypes.h"
|
||||
#include "fdbrpc/FailureMonitor.h"
|
||||
#include "fdbclient/Status.h"
|
||||
#include "fdbclient/ClientDBInfo.h"
|
||||
#include "fdbclient/MasterProxyInterface.h"
|
||||
#include "fdbclient/ClientWorkerInterface.h"
|
||||
|
||||
struct ClusterInterface {
|
||||
|
@ -25,7 +25,6 @@
|
||||
#include "fdbclient/NativeAPI.actor.h"
|
||||
#include "fdbclient/KeyRangeMap.h"
|
||||
#include "fdbclient/MasterProxyInterface.h"
|
||||
#include "fdbclient/ClientDBInfo.h"
|
||||
#include "fdbrpc/QueueModel.h"
|
||||
#include "fdbrpc/MultiInterface.h"
|
||||
#include "flow/TDMetric.actor.h"
|
||||
|
@ -75,7 +75,39 @@ struct MasterProxyInterface {
|
||||
}
|
||||
};
|
||||
|
||||
struct CommitID {
|
||||
// ClientDBInfo is all the information needed by a database client to access the database
|
||||
// It is returned (and kept up to date) by the OpenDatabaseRequest interface of ClusterInterface
|
||||
struct ClientDBInfo {
|
||||
constexpr static FileIdentifier file_identifier = 5355080;
|
||||
UID id; // Changes each time anything else changes
|
||||
vector< MasterProxyInterface > proxies;
|
||||
double clientTxnInfoSampleRate;
|
||||
int64_t clientTxnInfoSizeLimit;
|
||||
ClientDBInfo() : clientTxnInfoSampleRate(std::numeric_limits<double>::infinity()), clientTxnInfoSizeLimit(-1) {}
|
||||
|
||||
bool operator == (ClientDBInfo const& r) const { return id == r.id; }
|
||||
bool operator != (ClientDBInfo const& r) const { return id != r.id; }
|
||||
|
||||
template <class Archive>
|
||||
void serialize(Archive& ar) {
|
||||
if constexpr (!is_fb_function<Archive>) {
|
||||
ASSERT(ar.protocolVersion().isValid());
|
||||
}
|
||||
serializer(ar, proxies, id, clientTxnInfoSampleRate, clientTxnInfoSizeLimit);
|
||||
}
|
||||
};
|
||||
|
||||
struct ProxyForwardReply {
|
||||
Optional<ClientDBInfo> newClientInfo;
|
||||
ProxyForwardReply() {}
|
||||
|
||||
template <class Ar>
|
||||
void serialize(Ar &ar) {
|
||||
serializer(ar, newClientInfo);
|
||||
}
|
||||
};
|
||||
|
||||
struct CommitID : public ProxyForwardReply {
|
||||
constexpr static FileIdentifier file_identifier = 14254927;
|
||||
Version version; // returns invalidVersion if transaction conflicts
|
||||
uint16_t txnBatchId;
|
||||
@ -83,7 +115,7 @@ struct CommitID {
|
||||
|
||||
template <class Ar>
|
||||
void serialize(Ar& ar) {
|
||||
serializer(ar, version, txnBatchId, metadataVersion);
|
||||
serializer(ar, *(ProxyForwardReply*)this, version, txnBatchId, metadataVersion);
|
||||
}
|
||||
|
||||
CommitID() : version(invalidVersion), txnBatchId(0) {}
|
||||
@ -127,7 +159,7 @@ static inline int getBytes( CommitTransactionRequest const& r ) {
|
||||
return total;
|
||||
}
|
||||
|
||||
struct GetReadVersionReply {
|
||||
struct GetReadVersionReply : public ProxyForwardReply {
|
||||
constexpr static FileIdentifier file_identifier = 15709388;
|
||||
Version version;
|
||||
bool locked;
|
||||
@ -135,7 +167,7 @@ struct GetReadVersionReply {
|
||||
|
||||
template <class Ar>
|
||||
void serialize(Ar& ar) {
|
||||
serializer(ar, version, locked, metadataVersion);
|
||||
serializer(ar, *(ProxyForwardReply*)this, version, locked, metadataVersion);
|
||||
}
|
||||
};
|
||||
|
||||
@ -169,14 +201,14 @@ struct GetReadVersionRequest : TimedRequest {
|
||||
}
|
||||
};
|
||||
|
||||
struct GetKeyServerLocationsReply {
|
||||
struct GetKeyServerLocationsReply : public ProxyForwardReply {
|
||||
constexpr static FileIdentifier file_identifier = 10636023;
|
||||
Arena arena;
|
||||
std::vector<std::pair<KeyRangeRef, vector<StorageServerInterface>>> results;
|
||||
|
||||
template <class Ar>
|
||||
void serialize(Ar& ar) {
|
||||
serializer(ar, results, arena);
|
||||
serializer(ar, *(ProxyForwardReply*)this, results, arena);
|
||||
}
|
||||
};
|
||||
|
||||
|
@ -573,38 +573,44 @@ ACTOR static Future<Void> monitorClientInfo( Reference<AsyncVar<Optional<Cluster
|
||||
|
||||
choose {
|
||||
when( ClientDBInfo ni = wait( clusterInterface->get().present() ? brokenPromiseToNever( clusterInterface->get().get().openDatabase.getReply( req ) ) : Never() ) ) {
|
||||
TraceEvent("ClientInfoChange").detail("ChangeID", ni.id);
|
||||
outInfo->set(ni);
|
||||
|
||||
if (ni.proxies.empty()) {
|
||||
TraceEvent("ClientInfo_NoProxiesReturned").detail("ChangeID", ni.id);
|
||||
continue;
|
||||
} else if (!FlowTransport::transport().isClient()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
vector<Future<Void>> onProxyFailureVec;
|
||||
bool skipWaitForProxyFail = false;
|
||||
for (const auto& proxy : ni.proxies) {
|
||||
if (proxy.provisional) {
|
||||
skipWaitForProxyFail = true;
|
||||
loop {
|
||||
TraceEvent("ClientInfoChange").detail("ChangeID", outInfo->get().id);
|
||||
if (outInfo->get().proxies.empty()) {
|
||||
TraceEvent("ClientInfo_NoProxiesReturned").detail("ChangeID", outInfo->get().id);
|
||||
break;
|
||||
} else if (!FlowTransport::transport().isClient()) {
|
||||
break;
|
||||
}
|
||||
|
||||
onProxyFailureVec.push_back(
|
||||
IFailureMonitor::failureMonitor().onDisconnectOrFailure(
|
||||
proxy.getConsistentReadVersion.getEndpoint()) ||
|
||||
IFailureMonitor::failureMonitor().onDisconnectOrFailure(proxy.commit.getEndpoint()) ||
|
||||
IFailureMonitor::failureMonitor().onDisconnectOrFailure(
|
||||
proxy.getKeyServersLocations.getEndpoint()) ||
|
||||
IFailureMonitor::failureMonitor().onDisconnectOrFailure(
|
||||
proxy.getStorageServerRejoinInfo.getEndpoint()));
|
||||
}
|
||||
if (skipWaitForProxyFail) continue;
|
||||
state vector<Future<Void>> onProxyFailureVec;
|
||||
bool skipWaitForProxyFail = false;
|
||||
for (const auto& proxy : outInfo->get().proxies) {
|
||||
if (proxy.provisional) {
|
||||
skipWaitForProxyFail = true;
|
||||
break;
|
||||
}
|
||||
|
||||
leaderMon = Void();
|
||||
wait(waitForAny(onProxyFailureVec));
|
||||
leaderMon = ccf ? monitorLeader(ccf, clusterInterface) : Void();
|
||||
onProxyFailureVec.push_back(
|
||||
IFailureMonitor::failureMonitor().onDisconnectOrFailure(
|
||||
proxy.getConsistentReadVersion.getEndpoint()) ||
|
||||
IFailureMonitor::failureMonitor().onDisconnectOrFailure(proxy.commit.getEndpoint()) ||
|
||||
IFailureMonitor::failureMonitor().onDisconnectOrFailure(
|
||||
proxy.getKeyServersLocations.getEndpoint()) ||
|
||||
IFailureMonitor::failureMonitor().onDisconnectOrFailure(
|
||||
proxy.getStorageServerRejoinInfo.getEndpoint()));
|
||||
}
|
||||
if (skipWaitForProxyFail) break;
|
||||
|
||||
leaderMon = Void();
|
||||
state Future<Void> anyFailures = waitForAny(onProxyFailureVec);
|
||||
wait(anyFailures || outInfo->onChange());
|
||||
if(anyFailures.isReady()) {
|
||||
leaderMon = ccf ? monitorLeader(ccf, clusterInterface) : Void();
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
when( wait( clusterInterface->onChange() ) ) {
|
||||
if(clusterInterface->get().present())
|
||||
@ -1241,6 +1247,10 @@ ACTOR Future< pair<KeyRange,Reference<LocationInfo>> > getKeyLocation_internal(
|
||||
choose {
|
||||
when ( wait( cx->onMasterProxiesChanged() ) ) {}
|
||||
when ( GetKeyServerLocationsReply rep = wait( loadBalance( cx->getMasterProxies(info.useProvisionalProxies), &MasterProxyInterface::getKeyServersLocations, GetKeyServerLocationsRequest(key, Optional<KeyRef>(), 100, isBackward, key.arena()), TaskPriority::DefaultPromiseEndpoint ) ) ) {
|
||||
if(rep.newClientInfo.present()) {
|
||||
cx->clientInfo->set(rep.newClientInfo.get());
|
||||
continue;
|
||||
}
|
||||
if( info.debugID.present() )
|
||||
g_traceBatch.addEvent("TransactionDebug", info.debugID.get().first(), "NativeAPI.getKeyLocation.After");
|
||||
ASSERT( rep.results.size() == 1 );
|
||||
@ -1278,6 +1288,11 @@ ACTOR Future< vector< pair<KeyRange,Reference<LocationInfo>> > > getKeyRangeLoca
|
||||
choose {
|
||||
when ( wait( cx->onMasterProxiesChanged() ) ) {}
|
||||
when ( GetKeyServerLocationsReply _rep = wait( loadBalance( cx->getMasterProxies(info.useProvisionalProxies), &MasterProxyInterface::getKeyServersLocations, GetKeyServerLocationsRequest(keys.begin, keys.end, limit, reverse, keys.arena()), TaskPriority::DefaultPromiseEndpoint ) ) ) {
|
||||
if(_rep.newClientInfo.present()) {
|
||||
cx->clientInfo->set(_rep.newClientInfo.get());
|
||||
continue;
|
||||
}
|
||||
|
||||
state GetKeyServerLocationsReply rep = _rep;
|
||||
if( info.debugID.present() )
|
||||
g_traceBatch.addEvent("TransactionDebug", info.debugID.get().first(), "NativeAPI.getKeyLocations.After");
|
||||
@ -1490,6 +1505,11 @@ ACTOR Future<Version> waitForCommittedVersion( Database cx, Version version ) {
|
||||
choose {
|
||||
when ( wait( cx->onMasterProxiesChanged() ) ) {}
|
||||
when ( GetReadVersionReply v = wait( loadBalance( cx->getMasterProxies(false), &MasterProxyInterface::getConsistentReadVersion, GetReadVersionRequest( 0, GetReadVersionRequest::PRIORITY_SYSTEM_IMMEDIATE ), cx->taskID ) ) ) {
|
||||
if(v.newClientInfo.present()) {
|
||||
cx->clientInfo->set(v.newClientInfo.get());
|
||||
continue;
|
||||
}
|
||||
|
||||
if (v.version >= version)
|
||||
return v.version;
|
||||
// SOMEDAY: Do the wait on the server side, possibly use less expensive source of committed version (causal consistency is not needed for this purpose)
|
||||
@ -2703,6 +2723,11 @@ ACTOR static Future<Void> tryCommit( Database cx, Reference<TransactionLogInfo>
|
||||
throw request_maybe_delivered();
|
||||
}
|
||||
when (CommitID ci = wait( reply )) {
|
||||
if(ci.newClientInfo.present()) {
|
||||
cx->clientInfo->set(ci.newClientInfo.get());
|
||||
throw not_committed();
|
||||
}
|
||||
|
||||
Version v = ci.version;
|
||||
if (v != invalidVersion) {
|
||||
if (info.debugID.present())
|
||||
@ -3034,6 +3059,10 @@ ACTOR Future<GetReadVersionReply> getConsistentReadVersion( DatabaseContext *cx,
|
||||
choose {
|
||||
when ( wait( cx->onMasterProxiesChanged() ) ) {}
|
||||
when ( GetReadVersionReply v = wait( loadBalance( cx->getMasterProxies(flags & GetReadVersionRequest::FLAG_USE_PROVISIONAL_PROXIES), &MasterProxyInterface::getConsistentReadVersion, req, cx->taskID ) ) ) {
|
||||
if(v.newClientInfo.present()) {
|
||||
cx->clientInfo->set(v.newClientInfo.get());
|
||||
continue;
|
||||
}
|
||||
if( debugID.present() )
|
||||
g_traceBatch.addEvent("TransactionDebug", debugID.get().first(), "NativeAPI.getConsistentReadVersion.After");
|
||||
ASSERT( v.version > 0 );
|
||||
|
@ -31,7 +31,6 @@
|
||||
<EnableCompile Condition="'$(Configuration)|$(Platform)'=='Release|X64'">false</EnableCompile>
|
||||
</ActorCompiler>
|
||||
<ClInclude Include="BlobStore.h" />
|
||||
<ClInclude Include="ClientDBInfo.h" />
|
||||
<ClInclude Include="ClientLogEvents.h" />
|
||||
<ClInclude Include="ClientWorkerInterface.h" />
|
||||
<ClInclude Include="ClusterInterface.h" />
|
||||
|
@ -282,6 +282,7 @@ ServerKnobs::ServerKnobs(bool randomize, ClientKnobs* clientKnobs) {
|
||||
init( PROXY_SPIN_DELAY, 0.01 );
|
||||
init( UPDATE_REMOTE_LOG_VERSION_INTERVAL, 2.0 );
|
||||
init( MAX_TXS_POP_VERSION_HISTORY, 1e5 );
|
||||
init( PROXY_FORWARD_DELAY, 10.0 );
|
||||
|
||||
// Master Server
|
||||
// masterCommitter() in the master server will allow lower priority tasks (e.g. DataDistibution)
|
||||
|
@ -227,6 +227,7 @@ public:
|
||||
double PROXY_SPIN_DELAY;
|
||||
double UPDATE_REMOTE_LOG_VERSION_INTERVAL;
|
||||
int MAX_TXS_POP_VERSION_HISTORY;
|
||||
double PROXY_FORWARD_DELAY;
|
||||
|
||||
// Master Server
|
||||
double COMMIT_SLEEP_TIME;
|
||||
|
@ -965,7 +965,7 @@ ACTOR Future<Void> commitBatch(
|
||||
break;
|
||||
}
|
||||
when(GetReadVersionReply v = wait(self->getConsistentReadVersion.getReply(GetReadVersionRequest(0, GetReadVersionRequest::PRIORITY_SYSTEM_IMMEDIATE | GetReadVersionRequest::FLAG_CAUSAL_READ_RISKY)))) {
|
||||
if(v.version > self->committedVersion.get()) {
|
||||
if(!v.newClientInfo.present() && v.version > self->committedVersion.get()) {
|
||||
self->locked = v.locked;
|
||||
self->metadataVersion = v.metadataVersion;
|
||||
self->committedVersion.set(v.version);
|
||||
@ -1782,33 +1782,69 @@ ACTOR Future<Void> masterProxyServerCore(
|
||||
|
||||
ACTOR Future<Void> checkRemoved(Reference<AsyncVar<ServerDBInfo>> db, uint64_t recoveryCount, MasterProxyInterface myInterface) {
|
||||
loop{
|
||||
if (db->get().recoveryCount >= recoveryCount && !std::count(db->get().client.proxies.begin(), db->get().client.proxies.end(), myInterface))
|
||||
throw worker_removed();
|
||||
if (db->get().recoveryCount >= recoveryCount && !std::count(db->get().client.proxies.begin(), db->get().client.proxies.end(), myInterface)) {
|
||||
throw worker_removed();
|
||||
}
|
||||
wait(db->onChange());
|
||||
}
|
||||
}
|
||||
|
||||
ACTOR Future<Void> forwardProxy(ClientDBInfo info, RequestStream<CommitTransactionRequest> commit, RequestStream<GetReadVersionRequest> getConsistentReadVersion, RequestStream<GetKeyServerLocationsRequest> getKeyServersLocations) {
|
||||
loop {
|
||||
choose {
|
||||
when(CommitTransactionRequest req = waitNext(commit.getFuture())) {
|
||||
CommitID rep;
|
||||
rep.newClientInfo = info;
|
||||
req.reply.send(rep);
|
||||
}
|
||||
when(GetReadVersionRequest req = waitNext(getConsistentReadVersion.getFuture())) {
|
||||
GetReadVersionReply rep;
|
||||
rep.newClientInfo = info;
|
||||
req.reply.send(rep);
|
||||
}
|
||||
when(GetKeyServerLocationsRequest req = waitNext(getKeyServersLocations.getFuture())) {
|
||||
GetKeyServerLocationsReply rep;
|
||||
rep.newClientInfo = info;
|
||||
req.reply.send(rep);
|
||||
}
|
||||
}
|
||||
wait(yield());
|
||||
}
|
||||
}
|
||||
|
||||
ACTOR Future<Void> masterProxyServer(
|
||||
MasterProxyInterface proxy,
|
||||
InitializeMasterProxyRequest req,
|
||||
Reference<AsyncVar<ServerDBInfo>> db,
|
||||
std::string whitelistBinPaths)
|
||||
{
|
||||
state Future<Void> core;
|
||||
try {
|
||||
state Future<Void> core = masterProxyServerCore(proxy, req.master, db, req.recoveryCount, req.recoveryTransactionVersion, req.firstProxy, whitelistBinPaths);
|
||||
loop choose{
|
||||
when(wait(core)) { return Void(); }
|
||||
when(wait(checkRemoved(db, req.recoveryCount, proxy))) {}
|
||||
}
|
||||
core = masterProxyServerCore(proxy, req.master, db, req.recoveryCount, req.recoveryTransactionVersion, req.firstProxy, whitelistBinPaths);
|
||||
wait(core || checkRemoved(db, req.recoveryCount, proxy));
|
||||
}
|
||||
catch (Error& e) {
|
||||
if (e.code() == error_code_actor_cancelled || e.code() == error_code_worker_removed || e.code() == error_code_tlog_stopped ||
|
||||
e.code() == error_code_master_tlog_failed || e.code() == error_code_coordinators_changed || e.code() == error_code_coordinated_state_conflict ||
|
||||
e.code() == error_code_new_coordinators_timed_out)
|
||||
{
|
||||
TraceEvent("MasterProxyTerminated", proxy.id()).error(e, true);
|
||||
TraceEvent("MasterProxyTerminated", proxy.id()).error(e, true);
|
||||
|
||||
if (e.code() != error_code_worker_removed && e.code() != error_code_tlog_stopped &&
|
||||
e.code() != error_code_master_tlog_failed && e.code() != error_code_coordinators_changed &&
|
||||
e.code() != error_code_coordinated_state_conflict && e.code() != error_code_new_coordinators_timed_out) {
|
||||
throw;
|
||||
}
|
||||
}
|
||||
core.cancel();
|
||||
state Future<Void> finishForward = delay(SERVER_KNOBS->PROXY_FORWARD_DELAY);
|
||||
loop {
|
||||
if(finishForward.isReady()) {
|
||||
return Void();
|
||||
}
|
||||
throw;
|
||||
if(db->get().client.proxies.size() > 0 && !db->get().client.proxies[0].provisional && db->get().recoveryCount >= req.recoveryCount
|
||||
&& !std::count(db->get().client.proxies.begin(), db->get().client.proxies.end(), proxy)) {
|
||||
core = forwardProxy(db->get().client, proxy.commit, proxy.getConsistentReadVersion, proxy.getKeyServersLocations);
|
||||
proxy = MasterProxyInterface();
|
||||
wait(finishForward);
|
||||
return Void();
|
||||
}
|
||||
wait(db->onChange() || finishForward);
|
||||
}
|
||||
}
|
||||
|
@ -1357,7 +1357,7 @@ ACTOR Future<Void> masterCore( Reference<MasterData> self ) {
|
||||
// SOMEDAY: For faster recovery, do this and setDBState asynchronously and don't wait for them
|
||||
// unless we want to change TLogs
|
||||
wait((success(recoveryCommit) && sendInitialCommitToResolvers(self)) );
|
||||
if(recoveryCommit.isReady() && recoveryCommit.get().isError()) {
|
||||
if(recoveryCommit.isReady() && ( recoveryCommit.get().isError() || recoveryCommit.get().get().newClientInfo.present() )) {
|
||||
TEST(true); // Master recovery failed because of the initial commit failed
|
||||
throw master_recovery_failed();
|
||||
}
|
||||
|
@ -354,7 +354,7 @@ struct ConsistencyCheckWorkload : TestWorkload
|
||||
ErrorOr<GetKeyServerLocationsReply> shards = keyServerLocationFutures[i].get();
|
||||
|
||||
//If performing quiescent check, then all master proxies should be reachable. Otherwise, only one needs to be reachable
|
||||
if (self->performQuiescentChecks && !shards.present())
|
||||
if (self->performQuiescentChecks && (!shards.present() || shards.get().newClientInfo.present()))
|
||||
{
|
||||
TraceEvent("ConsistencyCheck_MasterProxyUnavailable").detail("MasterProxyID", proxyInfo->getId(i));
|
||||
self->testFailure("Master proxy unavailable");
|
||||
@ -363,7 +363,7 @@ struct ConsistencyCheckWorkload : TestWorkload
|
||||
|
||||
//Get the list of shards if one was returned. If not doing a quiescent check, we can break if it is.
|
||||
//If we are doing a quiescent check, then we only need to do this for the first shard.
|
||||
if (shards.present() && !keyServersInsertedForThisIteration)
|
||||
if (shards.present() && !shards.get().newClientInfo.present() && !keyServersInsertedForThisIteration)
|
||||
{
|
||||
keyServers.insert(keyServers.end(), shards.get().results.begin(), shards.get().results.end());
|
||||
keyServersInsertedForThisIteration = true;
|
||||
|
Loading…
x
Reference in New Issue
Block a user