wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Rename more places from proxy to commit proxy

This commit is contained in:
Young Liu 2020-09-15 22:29:49 -07:00
parent 35bef73a1c
commit cc5bc16bd8
22 changed files with 100 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
fdbcli/fdbcli.actor.cpp
View File

@ -471,8 +471,7 @@ void initHelp() {
helpMap["configure"] = CommandHelp(
"configure [new] "
"<single|double|triple|three_data_hall|three_datacenter|ssd|memory|memory-radixtree-beta|commit_proxies=<"
"COMMIT_PROXIES>|grv_"
"proxies=<GRV_PROXIES>|logs=<LOGS>|resolvers=<RESOLVERS>>*",
"COMMIT_PROXIES>|grv_proxies=<GRV_PROXIES>|logs=<LOGS>|resolvers=<RESOLVERS>>*",
"change the database configuration",
"The `new' option, if present, initializes a new database with the given configuration rather than changing "
"the configuration of an existing one. When used, both a redundancy mode and a storage engine must be "
@ -481,15 +480,13 @@ void initHelp() {
"See the Admin Guide.\n three_datacenter - See the Admin Guide.\n\nStorage engine:\n ssd - B-Tree storage "
"engine optimized for solid state disks.\n memory - Durable in-memory storage engine for small "
"datasets.\n\ncommit_proxies=<COMMIT_PROXIES>: Sets the desired number of commit proxies in the cluster. Must "
"be at least 1, or set "
"to -1 which restores the number of commit proxies to the default value.\n\ngrv_proxies=<GRV_PROXIES>: Sets "
"the "
"desired number of GRV proxies in the cluster. Must be at least 1, or set to -1 which restores the number of "
"GRV proxies to the default value.\n\nlogs=<LOGS>: Sets the desired number of log servers in the cluster. Must "
"be "
"at least 1, or set to -1 which restores the number of logs to the default value.\n\nresolvers=<RESOLVERS>: "
"Sets the desired number of resolvers in the cluster. Must be at least 1, or set to -1 which restores the "
"number of resolvers to the default value.\n\nSee the FoundationDB Administration Guide for more information.");
"be at least 1, or set to -1 which restores the number of commit proxies to the default "
"value.\n\ngrv_proxies=<GRV_PROXIES>: Sets the desired number of GRV proxies in the cluster. Must be at least "
"1, or set to -1 which restores the number of GRV proxies to the default value.\n\nlogs=<LOGS>: Sets the "
"desired number of log servers in the cluster. Must be at least 1, or set to -1 which restores the number of "
"logs to the default value.\n\nresolvers=<RESOLVERS>: Sets the desired number of resolvers in the cluster. "
"Must be at least 1, or set to -1 which restores the number of resolvers to the default value.\n\nSee the "
"FoundationDB Administration Guide for more information.");
helpMap["fileconfigure"] = CommandHelp(
"fileconfigure [new] <FILENAME>",
"change the database configuration from a file",

4
fdbclient/CommitProxyInterface.h
View File

@ -100,10 +100,10 @@ struct CommitProxyInterface {
struct ClientDBInfo {
constexpr static FileIdentifier file_identifier = 5355080;
UID id; // Changes each time anything else changes
vector< GrvProxyInterface > grvProxies;
vector<GrvProxyInterface> grvProxies;
vector<CommitProxyInterface> commitProxies;
Optional<CommitProxyInterface>
firstCommitProxy; // not serialized, used for commitOnFirstProxy when the proxies vector has been shrunk
firstCommitProxy; // not serialized, used for commitOnFirstProxy when the commit proxies vector has been shrunk
double clientTxnInfoSampleRate;
int64_t clientTxnInfoSizeLimit;
Optional<Value> forward;

44
fdbclient/DatabaseConfiguration.cpp
View File

@ -164,24 +164,40 @@ void DatabaseConfiguration::setDefaultReplicationPolicy() {
}
bool DatabaseConfiguration::isValid() const {
if (!(initialized && tLogWriteAntiQuorum >= 0 && tLogWriteAntiQuorum <= tLogReplicationFactor / 2 &&
tLogReplicationFactor >= 1 && storageTeamSize >= 1 && getDesiredCommitProxies() >= 1 &&
getDesiredGrvProxies() >= 1 && getDesiredLogs() >= 1 && getDesiredResolvers() >= 1 &&
tLogVersion != TLogVersion::UNSET && tLogVersion >= TLogVersion::MIN_RECRUITABLE &&
tLogVersion <= TLogVersion::MAX_SUPPORTED && tLogDataStoreType != KeyValueStoreType::END &&
if( !(initialized &&
tLogWriteAntiQuorum >= 0 &&
tLogWriteAntiQuorum <= tLogReplicationFactor/2 &&
tLogReplicationFactor >= 1 &&
storageTeamSize >= 1 &&
getDesiredCommitProxies() >= 1 &&
getDesiredGrvProxies() >= 1 &&
getDesiredLogs() >= 1 &&
getDesiredResolvers() >= 1 &&
tLogVersion != TLogVersion::UNSET &&
tLogVersion >= TLogVersion::MIN_RECRUITABLE &&
tLogVersion <= TLogVersion::MAX_SUPPORTED &&
tLogDataStoreType != KeyValueStoreType::END &&
tLogSpillType != TLogSpillType::UNSET &&
!(tLogSpillType == TLogSpillType::REFERENCE && tLogVersion < TLogVersion::V3) &&
storageServerStoreType != KeyValueStoreType::END && autoCommitProxyCount >= 1 && autoGrvProxyCount >= 1 &&
autoResolverCount >= 1 && autoDesiredTLogCount >= 1 && storagePolicy && tLogPolicy &&
getDesiredRemoteLogs() >= 1 && remoteTLogReplicationFactor >= 0 && repopulateRegionAntiQuorum >= 0 &&
repopulateRegionAntiQuorum <= 1 && usableRegions >= 1 && usableRegions <= 2 && regions.size() <= 2 &&
(usableRegions == 1 || regions.size() == 2) && (regions.size() == 0 || regions[0].priority >= 0) &&
(regions.size() == 0 ||
tLogPolicy->info() !=
"dcid^2 x zoneid^2 x 1"))) { // We cannot specify regions with three_datacenter replication
storageServerStoreType != KeyValueStoreType::END &&
autoCommitProxyCount >= 1 &&
autoGrvProxyCount >= 1 &&
autoResolverCount >= 1 &&
autoDesiredTLogCount >= 1 &&
storagePolicy &&
tLogPolicy &&
getDesiredRemoteLogs() >= 1 &&
remoteTLogReplicationFactor >= 0 &&
repopulateRegionAntiQuorum >= 0 &&
repopulateRegionAntiQuorum <= 1 &&
usableRegions >= 1 &&
usableRegions <= 2 &&
regions.size() <= 2 &&
( usableRegions == 1 || regions.size() == 2 ) &&
( regions.size() == 0 || regions[0].priority >= 0 ) &&
( regions.size() == 0 || tLogPolicy->info() != "dcid^2 x zoneid^2 x 1") ) ) { //We cannot specify regions with three_datacenter replication
return false;
}
std::set<Key> dcIds;
dcIds.insert(Key());
for(auto& r : regions) {

2
fdbclient/DatabaseContext.h
View File

@ -221,7 +221,7 @@ public:
Future<Void> monitorProxiesInfoChange;
Reference<CommitProxyInfo> commitProxies;
Reference<GrvProxyInfo> grvProxies;
bool proxyProvisional;
bool proxyProvisional; // Provisional commit proxy and grv proxy are used at the same time.
UID proxiesLastChange;
LocalityData clientLocality;
QueueModel queueModel;

1
fdbclient/ManagementAPI.actor.cpp
View File

@ -747,7 +747,6 @@ ConfigureAutoResult parseConfig( StatusObject const& status ) {
proxyCount = result.old_commit_proxies;
}
// Need to configure a good number.
result.desired_grv_proxies = std::max(std::min(4, processCount / 20), 1);
int grvProxyCount;
if (!statusObjConfig.get("grv_proxies", result.old_grv_proxies)) {

6
fdbclient/NativeAPI.actor.cpp
View File

@ -1597,9 +1597,9 @@ Reference<GrvProxyInfo> DatabaseContext::getGrvProxies(bool useProvisionalProxie
// Actor which will wait until the MultiInterface<CommitProxyInterface> returned by the DatabaseContext cx is not NULL
ACTOR Future<Reference<CommitProxyInfo>> getCommitProxiesFuture(DatabaseContext* cx, bool useProvisionalProxies) {
loop{
Reference<CommitProxyInfo> proxies = cx->getCommitProxies(useProvisionalProxies);
if (proxies)
return proxies;
Reference<CommitProxyInfo> commitProxies = cx->getCommitProxies(useProvisionalProxies);
if (commitProxies)
return commitProxies;
wait( cx->onProxiesChanged() );
}
}

2
fdbclient/vexillographer/fdb.options
View File

@ -195,7 +195,7 @@ description is not currently required but encouraged.
<Option name="next_write_no_write_conflict_range" code="30"
description="The next write performed on this transaction will not generate a write conflict range. As a result, other transactions which read the key(s) being modified by the next write will not conflict with this transaction. Care needs to be taken when using this option on a transaction that is shared between multiple threads. When setting this option, write conflict ranges will be disabled on the next write operation, regardless of what thread it is on." />
<Option name="commit_on_first_proxy" code="40"
description="Committing this transaction will bypass the normal load balancing across proxies and go directly to the specifically nominated 'first proxy'."
description="Committing this transaction will bypass the normal load balancing across commit proxies and go directly to the specifically nominated 'first commit proxy'."
hidden="true" />
<Option name="check_writes_enable" code="50"
hidden="true" />

3
fdbrpc/Locality.h
View File

@ -77,7 +77,6 @@ public:
if (s=="storage") _class = StorageClass;
else if (s=="transaction") _class = TransactionClass;
else if (s=="resolution") _class = ResolutionClass;
// else if (s=="proxy") _class = CommitProxyClass;
else if (s=="commit_proxy") _class = CommitProxyClass;
else if (s=="grv_proxy") _class = GrvProxyClass;
else if (s=="master") _class = MasterClass;
@ -100,7 +99,6 @@ public:
if (classStr=="storage") _class = StorageClass;
else if (classStr=="transaction") _class = TransactionClass;
else if (classStr=="resolution") _class = ResolutionClass;
// else if (classStr=="proxy") _class = CommitProxyClass;
else if (classStr=="commit_proxy") _class = CommitProxyClass;
else if (classStr=="grv_proxy") _class = GrvProxyClass;
else if (classStr=="master") _class = MasterClass;
@ -344,7 +342,6 @@ struct LBLocalityData {
template <class Interface>
struct LBLocalityData<Interface, typename std::enable_if< Interface::LocationAwareLoadBalance >::type> {
enum { Present = 1 };
// TODO: figure out why some interfaces don't have locality.
static LocalityData getLocality( Interface const& i ) { return i.locality; }
static NetworkAddress getAddress( Interface const& i ) { return i.address(); }
static bool alwaysFresh() { return Interface::AlwaysFresh; }

4
fdbserver/ApplyMetadataMutation.cpp
View File

@ -43,8 +43,8 @@ Reference<StorageInfo> getStorageInfo(UID id, std::map<UID, Reference<StorageInf
}
// It is incredibly important that any modifications to txnStateStore are done in such a way that
// the same operations will be done on all proxies at the same time. Otherwise, the data stored in
// txnStateStore will become corrupted.
// the same operations will be done on all commit proxies at the same time. Otherwise, the data
// stored in txnStateStore will become corrupted.
void applyMetadataMutations(UID const& dbgid, Arena& arena, VectorRef<MutationRef> const& mutations,
IKeyValueStore* txnStateStore, LogPushData* toCommit, bool& confChange,
Reference<ILogSystem> logSystem, Version popVersion,

4
fdbserver/ClusterController.actor.cpp
View File

@ -801,7 +801,7 @@ public:
RoleFitness(SERVER_KNOBS->EXPECTED_TLOG_FITNESS, req.configuration.getDesiredSatelliteLogs(dcId),
ProcessClass::TLog)
.betterCount(RoleFitness(satelliteLogs, ProcessClass::TLog))) ||
RoleFitness(SERVER_KNOBS->EXPECTED_PROXY_FITNESS, req.configuration.getDesiredCommitProxies(),
RoleFitness(SERVER_KNOBS->EXPECTED_COMMIT_PROXY_FITNESS, req.configuration.getDesiredCommitProxies(),
ProcessClass::CommitProxy)
.betterCount(RoleFitness(commit_proxies, ProcessClass::CommitProxy)) ||
RoleFitness(SERVER_KNOBS->EXPECTED_GRV_PROXY_FITNESS, req.configuration.getDesiredGrvProxies(),
@ -994,7 +994,7 @@ public:
(RoleFitness(SERVER_KNOBS->EXPECTED_TLOG_FITNESS, req.configuration.getDesiredLogs(),
ProcessClass::TLog)
.betterCount(RoleFitness(tlogs, ProcessClass::TLog)) ||
RoleFitness(SERVER_KNOBS->EXPECTED_PROXY_FITNESS, req.configuration.getDesiredCommitProxies(),
RoleFitness(SERVER_KNOBS->EXPECTED_COMMIT_PROXY_FITNESS, req.configuration.getDesiredCommitProxies(),
ProcessClass::CommitProxy)
.betterCount(bestFitness.proxy) ||
RoleFitness(SERVER_KNOBS->EXPECTED_GRV_PROXY_FITNESS, req.configuration.getDesiredGrvProxies(),

2
fdbserver/Knobs.cpp
View File

@ -438,7 +438,7 @@ void ServerKnobs::initialize(bool randomize, ClientKnobs* clientKnobs, bool isSi
init( EXPECTED_MASTER_FITNESS, ProcessClass::UnsetFit );
init( EXPECTED_TLOG_FITNESS, ProcessClass::UnsetFit );
init( EXPECTED_LOG_ROUTER_FITNESS, ProcessClass::UnsetFit );
init( EXPECTED_PROXY_FITNESS, ProcessClass::UnsetFit );
init( EXPECTED_COMMIT_PROXY_FITNESS, ProcessClass::UnsetFit );
init( EXPECTED_GRV_PROXY_FITNESS, ProcessClass::UnsetFit );
init( EXPECTED_RESOLVER_FITNESS, ProcessClass::UnsetFit );
init( RECRUITMENT_TIMEOUT, 600 ); if( randomize && BUGGIFY ) RECRUITMENT_TIMEOUT = deterministicRandom()->coinflip() ? 60.0 : 1.0;

2
fdbserver/Knobs.h
View File

@ -369,7 +369,7 @@ public:
int EXPECTED_MASTER_FITNESS;
int EXPECTED_TLOG_FITNESS;
int EXPECTED_LOG_ROUTER_FITNESS;
int EXPECTED_PROXY_FITNESS;
int EXPECTED_COMMIT_PROXY_FITNESS;
int EXPECTED_GRV_PROXY_FITNESS;
int EXPECTED_RESOLVER_FITNESS;
double RECRUITMENT_TIMEOUT;

2
fdbserver/MasterInterface.h
View File

@ -38,7 +38,7 @@ struct MasterInterface {
RequestStream< struct ChangeCoordinatorsRequest > changeCoordinators;
RequestStream< struct GetCommitVersionRequest > getCommitVersion;
RequestStream<struct BackupWorkerDoneRequest> notifyBackupWorkerDone;
// Get the centralized live committed version reported by proxies.
// Get the centralized live committed version reported by commit proxies.
RequestStream< struct GetRawCommittedVersionRequest > getLiveCommittedVersion;
// Report a proxy's committed version.
RequestStream< struct ReportRawCommittedVersionRequest> reportLiveCommittedVersion;

18
fdbserver/Ratekeeper.actor.cpp
View File

@ -527,7 +527,7 @@ struct RatekeeperLimits {
{}
};
struct CommitProxyInfo {
struct GrvProxyInfo {
int64_t totalTransactions;
int64_t batchTransactions;
uint64_t lastThrottledTagChangeId;
@ -535,7 +535,7 @@ struct CommitProxyInfo {
double lastUpdateTime;
double lastTagPushTime;
CommitProxyInfo()
GrvProxyInfo()
: totalTransactions(0), batchTransactions(0), lastUpdateTime(0), lastThrottledTagChangeId(0), lastTagPushTime(0) {
}
};
@ -547,7 +547,7 @@ struct RatekeeperData {
Map<UID, StorageQueueInfo> storageQueueInfo;
Map<UID, TLogQueueInfo> tlogQueueInfo;
std::map<UID, CommitProxyInfo> commitProxyInfo;
std::map<UID, GrvProxyInfo> grvProxyInfo;
Smoother smoothReleasedTransactions, smoothBatchReleasedTransactions, smoothTotalDurableBytes;
HealthMetrics healthMetrics;
DatabaseConfiguration configuration;
@ -1269,7 +1269,7 @@ void updateRate(RatekeeperData* self, RatekeeperLimits* limits) {
.detail("ReleasedBatchTPS", self->smoothBatchReleasedTransactions.smoothRate())
.detail("TPSBasis", actualTps)
.detail("StorageServers", sscount)
.detail("GrvProxies", self->commitProxyInfo.size())
.detail("GrvProxies", self->grvProxyInfo.size())
.detail("TLogs", tlcount)
.detail("WorstFreeSpaceStorageServer", worstFreeSpaceStorageServer)
.detail("WorstFreeSpaceTLog", worstFreeSpaceTLog)
@ -1371,9 +1371,9 @@ ACTOR Future<Void> ratekeeper(RatekeeperInterface rkInterf, Reference<AsyncVar<S
lastLimited = self.smoothReleasedTransactions.smoothRate() > SERVER_KNOBS->LAST_LIMITED_RATIO * self.batchLimits.tpsLimit;
double tooOld = now() - 1.0;
for (auto p = self.commitProxyInfo.begin(); p != self.commitProxyInfo.end();) {
for (auto p = self.grvProxyInfo.begin(); p != self.grvProxyInfo.end();) {
if (p->second.lastUpdateTime < tooOld)
p = self.commitProxyInfo.erase(p);
p = self.grvProxyInfo.erase(p);
else
++p;
}
@ -1382,7 +1382,7 @@ ACTOR Future<Void> ratekeeper(RatekeeperInterface rkInterf, Reference<AsyncVar<S
when (GetRateInfoRequest req = waitNext(rkInterf.getRateInfo.getFuture())) {
GetRateInfoReply reply;
auto& p = self.commitProxyInfo[req.requesterID];
auto& p = self.grvProxyInfo[req.requesterID];
//TraceEvent("RKMPU", req.requesterID).detail("TRT", req.totalReleasedTransactions).detail("Last", p.totalTransactions).detail("Delta", req.totalReleasedTransactions - p.totalTransactions);
if (p.totalTransactions > 0) {
self.smoothReleasedTransactions.addDelta( req.totalReleasedTransactions - p.totalTransactions );
@ -1399,8 +1399,8 @@ ACTOR Future<Void> ratekeeper(RatekeeperInterface rkInterf, Reference<AsyncVar<S
p.batchTransactions = req.batchReleasedTransactions;
p.lastUpdateTime = now();
reply.transactionRate = self.normalLimits.tpsLimit / self.commitProxyInfo.size();
reply.batchTransactionRate = self.batchLimits.tpsLimit / self.commitProxyInfo.size();
reply.transactionRate = self.normalLimits.tpsLimit / self.grvProxyInfo.size();
reply.batchTransactionRate = self.batchLimits.tpsLimit / self.grvProxyInfo.size();
reply.leaseDuration = SERVER_KNOBS->METRIC_UPDATE_RATE;
if(p.lastThrottledTagChangeId != self.throttledTagChangeId || now() > p.lastTagPushTime + SERVER_KNOBS->TAG_THROTTLE_PUSH_INTERVAL) {

10
fdbserver/Resolver.actor.cpp
View File

@ -44,7 +44,7 @@ struct ProxyRequestsInfo {
namespace{
struct Resolver : ReferenceCounted<Resolver> {
UID dbgid;
int proxyCount, resolverCount;
int commitProxyCount, resolverCount;
NotifiedVersion version;
AsyncVar<Version> neededVersion;
@ -77,8 +77,8 @@ struct Resolver : ReferenceCounted<Resolver> {
Future<Void> logger;
Resolver( UID dbgid, int proxyCount, int resolverCount )
: dbgid(dbgid), proxyCount(proxyCount), resolverCount(resolverCount), version(-1), conflictSet( newConflictSet() ), iopsSample( SERVER_KNOBS->KEY_BYTES_PER_SAMPLE ), debugMinRecentStateVersion(0),
Resolver( UID dbgid, int commitProxyCount, int resolverCount )
: dbgid(dbgid), commitProxyCount(commitProxyCount), resolverCount(resolverCount), version(-1), conflictSet( newConflictSet() ), iopsSample( SERVER_KNOBS->KEY_BYTES_PER_SAMPLE ), debugMinRecentStateVersion(0),
cc("Resolver", dbgid.toString()),
resolveBatchIn("ResolveBatchIn", cc), resolveBatchStart("ResolveBatchStart", cc), resolvedTransactions("ResolvedTransactions", cc), resolvedBytes("ResolvedBytes", cc),
resolvedReadConflictRanges("ResolvedReadConflictRanges", cc), resolvedWriteConflictRanges("ResolvedWriteConflictRanges", cc), transactionsAccepted("TransactionsAccepted", cc),
@ -238,7 +238,7 @@ ACTOR Future<Void> resolveBatch(
//TraceEvent("ResolveBatch", self->dbgid).detail("PrevVersion", req.prevVersion).detail("Version", req.version).detail("StateTransactionVersions", self->recentStateTransactionSizes.size()).detail("StateBytes", stateBytes).detail("FirstVersion", self->recentStateTransactionSizes.empty() ? -1 : self->recentStateTransactionSizes.front().first).detail("StateMutationsIn", req.txnStateTransactions.size()).detail("StateMutationsOut", reply.stateMutations.size()).detail("From", proxyAddress);
ASSERT(!proxyInfo.outstandingBatches.empty());
ASSERT(self->proxyInfoMap.size() <= self->proxyCount+1);
ASSERT(self->proxyInfoMap.size() <= self->commitProxyCount+1);
// SOMEDAY: This is O(n) in number of proxies. O(log n) solution using appropriate data structure?
Version oldestProxyVersion = req.version;
@ -257,7 +257,7 @@ ACTOR Future<Void> resolveBatch(
TEST(oldestProxyVersion != req.version); // The proxy that sent this request does not have the oldest current version
bool anyPopped = false;
if(firstUnseenVersion <= oldestProxyVersion && self->proxyInfoMap.size() == self->proxyCount+1) {
if(firstUnseenVersion <= oldestProxyVersion && self->proxyInfoMap.size() == self->commitProxyCount+1) {
TEST(true); // Deleting old state transactions
self->recentStateTransactions.erase( self->recentStateTransactions.begin(), self->recentStateTransactions.upper_bound( oldestProxyVersion ) );
self->debugMinRecentStateVersion = oldestProxyVersion + 1;

16
fdbserver/Status.actor.cpp
View File

@ -1063,14 +1063,14 @@ ACTOR static Future<JsonBuilderObject> recoveryStateStatusFetcher(WorkerDetails
// Add additional metadata for certain statuses
if (mStatusCode == RecoveryStatus::recruiting_transaction_servers) {
int requiredLogs = atoi( md.getValue("RequiredTLogs").c_str() );
int requiredProxies = atoi(md.getValue("RequiredCommitProxies").c_str());
int requiredCommitProxies = atoi(md.getValue("RequiredCommitProxies").c_str());
int requiredGrvProxies = atoi(md.getValue("RequiredGrvProxies").c_str());
int requiredResolvers = atoi( md.getValue("RequiredResolvers").c_str() );
//int requiredProcesses = std::max(requiredLogs, std::max(requiredResolvers, requiredProxies));
//int requiredProcesses = std::max(requiredLogs, std::max(requiredResolvers, requiredCommitProxies));
//int requiredMachines = std::max(requiredLogs, 1);
message["required_logs"] = requiredLogs;
message["required_commit_proxies"] = requiredProxies;
message["required_commit_proxies"] = requiredCommitProxies;
message["required_grv_proxies"] = requiredGrvProxies;
message["required_resolvers"] = requiredResolvers;
} else if (mStatusCode == RecoveryStatus::locking_old_transaction_servers) {
@ -2443,7 +2443,7 @@ ACTOR Future<StatusReply> clusterGetStatus(
getProcessIssuesAsMessages(workerIssues);
state vector<std::pair<StorageServerInterface, EventMap>> storageServers;
state vector<std::pair<TLogInterface, EventMap>> tLogs;
state vector<std::pair<CommitProxyInterface, EventMap>> commit_proxies;
state vector<std::pair<CommitProxyInterface, EventMap>> commitProxies;
state vector<std::pair<GrvProxyInterface, EventMap>> grvProxies;
state JsonBuilderObject qos;
state JsonBuilderObject data_overlay;
@ -2592,9 +2592,9 @@ ACTOR Future<StatusReply> clusterGetStatus(
}
// ...also commit proxies
ErrorOr<vector<std::pair<CommitProxyInterface, EventMap>>> _commit_proxies = wait(commitProxyFuture);
if (_commit_proxies.present()) {
commit_proxies = _commit_proxies.get();
ErrorOr<vector<std::pair<CommitProxyInterface, EventMap>>> _commitProxies = wait(commitProxyFuture);
if (_commitProxies.present()) {
commitProxies = _commitProxies.get();
} else {
messages.push_back(
JsonBuilder::makeMessage("commit_proxies_error", "Timed out trying to retrieve commit proxies."));
@ -2620,7 +2620,7 @@ ACTOR Future<StatusReply> clusterGetStatus(
JsonBuilderObject processStatus = wait(processStatusFetcher(
db, workers, pMetrics, mMetrics, networkMetrics, latestError, traceFileOpenErrors, programStarts,
processIssues, storageServers, tLogs, commit_proxies, grvProxies, coordinators, cx, configuration,
processIssues, storageServers, tLogs, commitProxies, grvProxies, coordinators, cx, configuration,
loadResult.present() ? loadResult.get().healthyZone : Optional<Key>(), &status_incomplete_reasons));
statusObj["processes"] = processStatus;
statusObj["clients"] = clientStatusFetcher(clientStatus);

26
fdbserver/masterserver.actor.cpp
View File

@ -52,18 +52,18 @@ using std::vector;
using std::min;
using std::max;
struct ProxyVersionReplies {
struct CommitProxyVersionReplies {
std::map<uint64_t, GetCommitVersionReply> replies;
NotifiedVersion latestRequestNum;
ProxyVersionReplies(ProxyVersionReplies&& r) noexcept
CommitProxyVersionReplies(CommitProxyVersionReplies&& r) noexcept
: replies(std::move(r.replies)), latestRequestNum(std::move(r.latestRequestNum)) {}
void operator=(ProxyVersionReplies&& r) noexcept {
void operator=(CommitProxyVersionReplies&& r) noexcept {
replies = std::move(r.replies);
latestRequestNum = std::move(r.latestRequestNum);
}
ProxyVersionReplies() : latestRequestNum(0) {}
CommitProxyVersionReplies() : latestRequestNum(0) {}
};
ACTOR Future<Void> masterTerminateOnConflict( UID dbgid, Promise<Void> fullyRecovered, Future<Void> onConflict, Future<Void> switchedState ) {
@ -177,7 +177,7 @@ struct MasterData : NonCopyable, ReferenceCounted<MasterData> {
recoveryTransactionVersion; // The first version in this epoch
double lastCommitTime;
Version liveCommittedVersion; // The largest live committed version reported by proxies.
Version liveCommittedVersion; // The largest live committed version reported by commit proxies.
bool databaseLocked;
Optional<Value> proxyMetadataVersion;
Version minKnownCommittedVersion;
@ -213,7 +213,7 @@ struct MasterData : NonCopyable, ReferenceCounted<MasterData> {
std::vector<GrvProxyInterface> provisionalGrvProxies;
std::vector<ResolverInterface> resolvers;
std::map<UID, ProxyVersionReplies> lastProxyVersionReplies;
std::map<UID, CommitProxyVersionReplies> lastCommitProxyVersionReplies;
Standalone<StringRef> dbId;
@ -299,7 +299,7 @@ ACTOR Future<Void> newCommitProxies(Reference<MasterData> self, RecruitFromConfi
}
vector<CommitProxyInterface> newRecruits = wait(getAll(initializationReplies));
// It is required for the correctness of COMMIT_ON_FIRST_PROXY that self->proxies[0] is the firstCommitProxy.
// It is required for the correctness of COMMIT_ON_FIRST_PROXY that self->commitProxies[0] is the firstCommitProxy.
self->commitProxies = newRecruits;
return Void();
@ -502,14 +502,14 @@ ACTOR Future<Void> updateLogsValue( Reference<MasterData> self, Database cx ) {
}
Future<Void> sendMasterRegistration(MasterData* self, LogSystemConfig const& logSystemConfig,
vector<CommitProxyInterface> proxies, vector<GrvProxyInterface> grvProxies,
vector<CommitProxyInterface> commitProxies, vector<GrvProxyInterface> grvProxies,
vector<ResolverInterface> resolvers, DBRecoveryCount recoveryCount,
vector<UID> priorCommittedLogServers) {
RegisterMasterRequest masterReq;
masterReq.id = self->myInterface.id();
masterReq.mi = self->myInterface.locality;
masterReq.logSystemConfig = logSystemConfig;
masterReq.commitProxies = proxies;
masterReq.commitProxies = commitProxies;
masterReq.grvProxies = grvProxies;
masterReq.resolvers = resolvers;
masterReq.recoveryCount = recoveryCount;
@ -978,9 +978,9 @@ ACTOR Future<Void> recoverFrom( Reference<MasterData> self, Reference<ILogSystem
ACTOR Future<Void> getVersion(Reference<MasterData> self, GetCommitVersionRequest req) {
state Span span("M:getVersion"_loc, { req.spanContext });
state std::map<UID, ProxyVersionReplies>::iterator proxyItr = self->lastProxyVersionReplies.find(req.requestingProxy); // lastProxyVersionReplies never changes
state std::map<UID, CommitProxyVersionReplies>::iterator proxyItr = self->lastCommitProxyVersionReplies.find(req.requestingProxy); // lastCommitProxyVersionReplies never changes
if (proxyItr == self->lastProxyVersionReplies.end()) {
if (proxyItr == self->lastCommitProxyVersionReplies.end()) {
// Request from invalid proxy (e.g. from duplicate recruitment request)
req.reply.send(Never());
return Void();
@ -1047,7 +1047,7 @@ ACTOR Future<Void> getVersion(Reference<MasterData> self, GetCommitVersionReques
ACTOR Future<Void> provideVersions(Reference<MasterData> self) {
state ActorCollection versionActors(false);
for (auto& p : self->commitProxies) self->lastProxyVersionReplies[p.id()] = ProxyVersionReplies();
for (auto& p : self->commitProxies) self->lastCommitProxyVersionReplies[p.id()] = CommitProxyVersionReplies();
loop {
choose {
@ -1626,7 +1626,7 @@ ACTOR Future<Void> masterCore( Reference<MasterData> self ) {
state Future<ErrorOr<CommitID>> recoveryCommit = self->commitProxies[0].commit.tryGetReply(recoveryCommitRequest);
self->addActor.send( self->logSystem->onError() );
self->addActor.send( waitResolverFailure( self->resolvers ) );
self->addActor.send(waitCommitProxyFailure(self->commitProxies));
self->addActor.send( waitCommitProxyFailure(self->commitProxies));
self->addActor.send( waitGrvProxyFailure( self->grvProxies ) );
self->addActor.send( provideVersions(self) );
self->addActor.send( serveLiveCommittedVersion(self) );

6
fdbserver/storageserver.actor.cpp
View File

@ -3974,11 +3974,11 @@ ACTOR Future<Void> replaceInterface( StorageServer* self, StorageServerInterface
loop {
state Future<Void> infoChanged = self->db->onChange();
state Reference<CommitProxyInfo> proxies(new CommitProxyInfo(self->db->get().client.commitProxies));
state Reference<CommitProxyInfo> commitProxies(new CommitProxyInfo(self->db->get().client.commitProxies));
choose {
when(GetStorageServerRejoinInfoReply _rep =
wait(proxies->size()
? basicLoadBalance(proxies, &CommitProxyInterface::getStorageServerRejoinInfo,
wait(commitProxies->size()
? basicLoadBalance(commitProxies, &CommitProxyInterface::getStorageServerRejoinInfo,
GetStorageServerRejoinInfoRequest(ssi.id(), ssi.locality.dcId()))
: Never())) {
state GetStorageServerRejoinInfoReply rep = _rep;

2
fdbserver/workloads/ConfigureDatabase.actor.cpp
View File

@ -302,7 +302,7 @@ struct ConfigureDatabaseWorkload : TestWorkload {
config += generateRegions();
if (deterministicRandom()->random01() < 0.5) config += " logs=" + format("%d", randomRoleNumber());
if (deterministicRandom()->random01() < 0.5) config += " proxies=" + format("%d", randomRoleNumber());
if (deterministicRandom()->random01() < 0.5) config += " commit_proxies=" + format("%d", randomRoleNumber());
if (deterministicRandom()->random01() < 0.5)
config += " grv_proxies=" + format("%d", randomRoleNumber());
if (deterministicRandom()->random01() < 0.5) config += " resolvers=" + format("%d", randomRoleNumber());

10
fdbserver/workloads/TargetedKill.actor.cpp
View File

@ -88,14 +88,14 @@ struct TargetedKillWorkload : TestWorkload {
if( self->machineToKill == "master" ) {
machine = self->dbInfo->get().master.address();
} else if (self->machineToKill == "commitproxy") {
auto proxies = cx->getCommitProxies(false);
int o = deterministicRandom()->randomInt(0, proxies->size());
for( int i = 0; i < proxies->size(); i++) {
CommitProxyInterface mpi = proxies->getInterface(o);
auto commitProxies = cx->getCommitProxies(false);
int o = deterministicRandom()->randomInt(0, commitProxies->size());
for( int i = 0; i < commitProxies->size(); i++) {
CommitProxyInterface mpi = commitProxies->getInterface(o);
machine = mpi.address();
if(machine != self->dbInfo->get().clusterInterface.getWorkers.getEndpoint().getPrimaryAddress())
break;
o = ++o%proxies->size();
o = ++o%commitProxies->size();
}
} else if (self->machineToKill == "grvproxy") {
auto grvProxies = cx->getGrvProxies(false);

2
tests/status/separate_not_enough_servers.txt
View File

@ -2,6 +2,6 @@ Using cluster file `tcf/separatecoordinator.cluster'.
Recruiting new transaction servers.
Need at least 3 log servers, 1 proxies, 1 grv proxies and 1 resolvers.
Need at least 3 log servers, 1 commit proxies, 1 grv proxies and 1 resolvers.
Have 1 processes on 1 machines.