diff --git a/fdbserver/clustercontroller/ClusterController.actor.cpp b/fdbserver/clustercontroller/ClusterController.actor.cpp index fff47196913..a8055f872f9 100644 --- a/fdbserver/clustercontroller/ClusterController.actor.cpp +++ b/fdbserver/clustercontroller/ClusterController.actor.cpp @@ -296,7 +296,7 @@ Future recruitFailedLogRouters(ClusterControllerData* cluster, db->recoveryData->remoteDcIds.size() ? db->recoveryData->remoteDcIds[0] : Optional(); // Use getWorkersForRoleInDatacenter to get workers for all log routers at once - std::map>, int> id_used; + ClusterControllerData::WorkerUsages id_used; cluster->updateKnownIds(&id_used); std::vector workers = @@ -812,7 +812,7 @@ void checkOutstandingStorageRequests(ClusterControllerData* self) { // Finds and returns a new process for role WorkerDetails findNewProcessForSingleton(ClusterControllerData* self, const ProcessClass::ClusterRole role, - std::map>, int>& id_used) { + ClusterControllerData::WorkerUsages& id_used) { // find new process in cluster for role WorkerDetails newWorker = self->getWorkerForRoleInDatacenter( @@ -825,7 +825,7 @@ WorkerDetails findNewProcessForSingleton(ClusterControllerData* self, } // acknowledge that the pid is now potentially used by this role as well - id_used[newWorker.interf.locality.processId()]++; + id_used[newWorker.interf.locality.processId()].addRole(role); return newWorker; } @@ -923,14 +923,14 @@ void checkBetterSingletons(ClusterControllerData* self) { } // note: this map doesn't consider pids used by existing singletons - std::map>, int> id_used = self->getUsedIds(); + ClusterControllerData::WorkerUsages id_used = self->getUsedIds(); // We prefer spreading out other roles more than separating singletons on their own process // so we artificially amplify the pid count for the processes used by non-singleton roles. // In other words, we make the processes used for other roles less desirable to be used // by singletons as well. for (auto& it : id_used) { - it.second *= PID_USED_AMP_FOR_NON_SINGLETON; + it.second.multiplier *= PID_USED_AMP_FOR_NON_SINGLETON; } // Try to find a new process for each singleton. @@ -2402,7 +2402,7 @@ Future startDataDistributor(ClusterControllerData* self, double waitTime) co_return; } - std::map>, int> idUsed = self->getUsedIds(); + auto idUsed = self->getUsedIds(); WorkerFitnessInfo ddWorker = self->getWorkerForRoleInDatacenter(self->clusterControllerDcId, ProcessClass::DataDistributor, ProcessClass::NeverAssign, @@ -2503,7 +2503,7 @@ Future startRatekeeper(ClusterControllerData* self, double waitTime) { co_return; } - std::map>, int> id_used = self->getUsedIds(); + ClusterControllerData::WorkerUsages id_used = self->getUsedIds(); WorkerFitnessInfo rkWorker = self->getWorkerForRoleInDatacenter(self->clusterControllerDcId, ProcessClass::Ratekeeper, ProcessClass::NeverAssign, @@ -2592,7 +2592,7 @@ Future startConsistencyScan(ClusterControllerData* self) { co_return; } - std::map>, int> id_used = self->getUsedIds(); + auto id_used = self->getUsedIds(); WorkerFitnessInfo csWorker = self->getWorkerForRoleInDatacenter(self->clusterControllerDcId, ProcessClass::ConsistencyScan, ProcessClass::NeverAssign, @@ -3833,4 +3833,107 @@ TEST_CASE("/fdbserver/clustercontroller/invalidateExcludedProcessComplaints") { return Void(); } +// Test for the fix described in PR #10411. +// Verifies that in a small cluster (3 stateless, 3 transaction, 3 storage processes), +// the role allocation correctly assigns all 3 commit_proxy roles to the stateless nodes. +// Previously, only 1 commit_proxy would be recruited due to a bug in the candidate +// selection logic within ClusterControllerData::getWorkersForRoleInDatacenter. +// This test ensures that the desired number of proxies (3) is now fully recruited +// when sufficient stateless processes exist. +TEST_CASE("/fdbserver/clustercontroller/proxyColocationOnStateless") { + ClusterControllerData data(ClusterControllerFullInterface(), + LocalityData(), + ServerCoordinators(Reference( + new ClusterConnectionMemoryRecord(ClusterConnectionString()))), + makeReference>>()); + + constexpr int numWorkersPerClass = 3; + + auto makeWorkers = [&](ProcessClass::ClassType classType, StringRef prefix, int count) { + std::vector workers; + for (int i = 0; i < count; i++) { + auto pid = prefix.toString() + std::to_string(i); + WorkerInterface wi; + wi.initEndpoints(); + wi.locality.set(LocalityData::keyZoneId, Standalone(pid + "_zone")); + wi.locality.set(LocalityData::keyProcessId, Standalone(pid)); + data.id_worker[wi.locality.processId()] = + WorkerInfo(Future(), + ReplyPromise(), + 0, + wi, + ProcessClass(classType, ProcessClass::CommandLineSource), + ProcessClass(classType, ProcessClass::CommandLineSource), + ClusterControllerPriorityInfo( + ProcessClass::UnsetFit, false, ClusterControllerPriorityInfo::FitnessUnknown), + false, + true, + Standalone>()); + workers.push_back(wi); + } + return workers; + }; + + auto statelessWorkers = makeWorkers(ProcessClass::StatelessClass, "sl"_sr, numWorkersPerClass); + auto transactionWorkers = makeWorkers(ProcessClass::TransactionClass, "tx"_sr, numWorkersPerClass); + auto storageWorkers = makeWorkers(ProcessClass::StorageClass, "ss"_sr, numWorkersPerClass); + + data.masterProcessId = statelessWorkers[0].locality.processId(); + data.clusterControllerProcessId = statelessWorkers[1].locality.processId(); + data.startTime = 0; + data.gotFullyRecoveredConfig = true; + data.gotProcessClasses = true; + + DatabaseConfiguration config; + config.initialized = true; + config.tLogReplicationFactor = numWorkersPerClass; + config.desiredTLogCount = numWorkersPerClass; + config.commitProxyCount = numWorkersPerClass; + config.grvProxyCount = numWorkersPerClass; + config.resolverCount = numWorkersPerClass; + config.tLogPolicy = makeReference(); + data.db.config = config; + data.db.fullyRecoveredConfig = config; + + // Use findWorkersForConfigurationDispatch — the production code path + // that handles full recruitment: TLogs + two-level proxy/resolver recruiting + RecruitFromConfigurationRequest req; + req.configuration = config; + req.recruitSeedServers = false; + req.maxOldLogRouters = 0; + + auto reply = data.findWorkersForConfigurationDispatch(req, false); + + ASSERT(reply.tLogs.size() == numWorkersPerClass); + ASSERT(reply.commitProxies.size() == numWorkersPerClass); + ASSERT(reply.grvProxies.size() == numWorkersPerClass); + ASSERT(reply.resolvers.size() == numWorkersPerClass); + + // TLogs are on transaction processes + std::set>> txPids; + for (const auto& w : transactionWorkers) { + txPids.insert(w.locality.processId()); + } + for (const auto& tlog : reply.tLogs) { + ASSERT(txPids.contains(tlog.locality.processId())); + } + + // check if proxy/resolvers are on stateless + std::set>> slPids; + for (const auto& w : statelessWorkers) { + slPids.insert(w.locality.processId()); + } + for (const auto& cp : reply.commitProxies) { + ASSERT(slPids.contains(cp.locality.processId())); + } + for (const auto& gp : reply.grvProxies) { + ASSERT(slPids.contains(gp.locality.processId())); + } + for (const auto& rs : reply.resolvers) { + ASSERT(slPids.contains(rs.locality.processId())); + } + + return Void(); +} + } // namespace diff --git a/fdbserver/clustercontroller/ClusterController.h b/fdbserver/clustercontroller/ClusterController.h index 44e4c690ae1..9b1bc92786f 100644 --- a/fdbserver/clustercontroller/ClusterController.h +++ b/fdbserver/clustercontroller/ClusterController.h @@ -280,6 +280,43 @@ class ClusterControllerData { } }; + struct WorkerUsage { + std::bitset roles; + unsigned weight = 0; + unsigned multiplier = 1; + + static unsigned roleWeight(ProcessClass::ClusterRole role) { + // TO DO: Introduce knobs for different weights of different + // roles + return 1; + } + + void addRole(ProcessClass::ClusterRole role) { + if (!roles.test(role)) { + roles.set(role); + weight += roleWeight(role); + } + } + + unsigned getWeight() const { return weight * multiplier; } + + unsigned getUniqueHash() const { return roles.to_ulong(); } + + std::string toString() const { + std::string roleCodes; + + for (unsigned r = 0; r < ProcessClass::NoRole; r++) + if (roles.test((ProcessClass::ClusterRole)r)) { + if (!roleCodes.empty()) + roleCodes.append(","); + roleCodes.append(Role::get((ProcessClass::ClusterRole)r).abbreviation); + } + return roleCodes; + } + }; + + using WorkerUsages = std::map>, WorkerUsage>; + bool workerAvailable(WorkerInfo const& worker, bool checkStable) const { return (now() - startTime < 2 * FLOW_KNOBS->SERVER_REQUEST_INTERVAL) || (IFailureMonitor::failureMonitor().getState(worker.details.interf.storage.getEndpoint()).isAvailable() && @@ -571,7 +608,7 @@ class ClusterControllerData { // It attempts to evenly recruit processes from across data_halls or datacenters std::vector getWorkersForTlogsComplex(DatabaseConfiguration const& conf, int32_t desired, - std::map>, int>& id_used, + WorkerUsages& id_used, StringRef field, int minFields, int minPerField, @@ -579,7 +616,7 @@ class ClusterControllerData { bool checkStable, const std::set>& dcIds, const std::vector& exclusionWorkerIds) { - std::map, std::vector> fitness_workers; + std::map, std::vector> fitness_workers; // Go through all the workers to list all the workers that can be recruited. for (const auto& [worker_process_id, worker_info] : id_worker) { @@ -646,8 +683,10 @@ class ClusterControllerData { continue; } - fitness_workers[std::make_tuple( - fitness, id_used[worker_process_id], isLongLivedStateless(worker_process_id))] + fitness_workers[std::make_tuple(fitness, + id_used[worker_process_id].getWeight(), + isLongLivedStateless(worker_process_id), + id_used[worker_process_id].getUniqueHash())] .push_back(worker_details); } @@ -756,7 +795,7 @@ class ClusterControllerData { } for (auto& result : resultSet) { - id_used[result.interf.locality.processId()]++; + id_used[result.interf.locality.processId()].addRole(ProcessClass::TLog); } return std::vector(resultSet.begin(), resultSet.end()); @@ -765,7 +804,7 @@ class ClusterControllerData { // Attempt to recruit TLogs without degraded processes and see if it improves the configuration std::vector getWorkersForTlogsComplex(DatabaseConfiguration const& conf, int32_t desired, - std::map>, int>& id_used, + WorkerUsages& id_used, StringRef field, int minFields, int minPerField, @@ -773,7 +812,7 @@ class ClusterControllerData { const std::set>& dcIds, const std::vector& exclusionWorkerIds) { desired = std::max(desired, minFields * minPerField); - std::map>, int> withDegradedUsed = id_used; + auto withDegradedUsed = id_used; auto withDegraded = getWorkersForTlogsComplex(conf, desired, withDegradedUsed, @@ -801,7 +840,7 @@ class ClusterControllerData { } try { - std::map>, int> withoutDegradedUsed = id_used; + auto withoutDegradedUsed = id_used; auto withoutDegraded = getWorkersForTlogsComplex(conf, desired, withoutDegradedUsed, @@ -835,11 +874,12 @@ class ClusterControllerData { std::vector getWorkersForTlogsSimple(DatabaseConfiguration const& conf, int32_t required, int32_t desired, - std::map>, int>& id_used, + WorkerUsages& id_used, bool checkStable, const std::set>& dcIds, const std::vector& exclusionWorkerIds) { - std::map, std::vector> fitness_workers; + std::map, std::vector> + fitness_workers; // Go through all the workers to list all the workers that can be recruited. for (const auto& [worker_process_id, worker_info] : id_worker) { @@ -909,10 +949,11 @@ class ClusterControllerData { } fitness_workers[std::make_tuple(fitness, - id_used[worker_process_id], + id_used[worker_process_id].getWeight(), worker_details.degraded, isLongLivedStateless(worker_process_id), - inCCDC)] + inCCDC, + id_used[worker_process_id].getUniqueHash())] .push_back(worker_details); } @@ -968,7 +1009,7 @@ class ClusterControllerData { ASSERT(resultSet.size() >= required && resultSet.size() <= desired); for (auto& result : resultSet) { - id_used[result.interf.locality.processId()]++; + id_used[result.interf.locality.processId()].addRole(ProcessClass::TLog); } return std::vector(resultSet.begin(), resultSet.end()); @@ -991,11 +1032,12 @@ class ClusterControllerData { int32_t required, int32_t desired, Reference const& policy, - std::map>, int>& id_used, + WorkerUsages& id_used, bool checkStable = false, const std::set>& dcIds = std::set>(), const std::vector& exclusionWorkerIds = {}) { - std::map, std::vector> fitness_workers; + std::map, std::vector> + fitness_workers; std::vector results; Reference logServerSet = makeReference>(); LocalityMap* logServerMap = (LocalityMap*)logServerSet.getPtr(); @@ -1070,7 +1112,11 @@ class ClusterControllerData { fitness = std::max(fitness, ProcessClass::GoodFit); } - fitness_workers[std::make_tuple(fitness, id_used[worker_process_id], worker_details.degraded, inCCDC)] + fitness_workers[std::make_tuple(fitness, + id_used[worker_process_id].getWeight(), + worker_details.degraded, + inCCDC, + id_used[worker_process_id].getUniqueHash())] .push_back(worker_details); } @@ -1120,7 +1166,7 @@ class ClusterControllerData { results.push_back(*object); } for (auto& result : results) { - id_used[result.interf.locality.processId()]++; + id_used[result.interf.locality.processId()].addRole(ProcessClass::TLog); } return results; } @@ -1194,7 +1240,7 @@ class ClusterControllerData { results.push_back(*object); } for (auto& result : results) { - id_used[result.interf.locality.processId()]++; + id_used[result.interf.locality.processId()].addRole(ProcessClass::TLog); } return results; } @@ -1219,7 +1265,7 @@ class ClusterControllerData { tLocalities.push_back(object->interf.locality); } for (auto& result : results) { - id_used[result.interf.locality.processId()]++; + id_used[result.interf.locality.processId()].addRole(ProcessClass::TLog); } TraceEvent("GetTLogTeamDone") .detail("Policy", policy->info()) @@ -1243,7 +1289,7 @@ class ClusterControllerData { int32_t required, int32_t desired, Reference const& policy, - std::map>, int>& id_used, + WorkerUsages& id_used, bool checkStable = false, const std::set>& dcIds = std::set>(), const std::vector& exclusionWorkerIds = {}) { @@ -1255,7 +1301,7 @@ class ClusterControllerData { if (embedded->name() == "Across") { PolicyAcross* pa2 = (PolicyAcross*)embedded.getPtr(); if (pa2->attributeKey() == "zoneid" && pa2->embeddedPolicyName() == "One") { - std::map>, int> testUsed = id_used; + auto testUsed = id_used; auto workers = getWorkersForTlogsComplex(conf, desired, @@ -1319,7 +1365,7 @@ class ClusterControllerData { useSimple = true; } if (useSimple) { - std::map>, int> testUsed = id_used; + auto testUsed = id_used; auto workers = getWorkersForTlogsSimple(conf, required, desired, id_used, checkStable, dcIds, exclusionWorkerIds); @@ -1362,7 +1408,7 @@ class ClusterControllerData { std::vector getWorkersForSatelliteLogs(const DatabaseConfiguration& conf, const RegionInfo& region, const RegionInfo& remoteRegion, - std::map>, int>& id_used, + WorkerUsages& id_used, bool& satelliteFallback, bool checkStable = false) { int startDC = 0; @@ -1405,7 +1451,7 @@ class ClusterControllerData { // TLogs can be recruited. It does not balance the number of desired TLogs across the satellite and // remote sides. if (remoteDCUsedAsSatellite) { - std::map>, int> tmpIdUsed; + WorkerUsages tmpIdUsed; auto remoteLogs = getWorkersForTlogs(conf, conf.getRemoteTLogReplicationFactor(), conf.getRemoteTLogReplicationFactor(), @@ -1468,10 +1514,11 @@ class ClusterControllerData { ProcessClass::ClusterRole role, ProcessClass::Fitness unacceptableFitness, DatabaseConfiguration const& conf, - std::map>, int>& id_used, + WorkerUsages& id_used, std::map>, int> preferredSharing = {}, bool checkStable = false) { - std::map, std::vector> fitness_workers; + std::map, std::vector> + fitness_workers; for (auto& it : id_worker) { auto fitness = it.second.details.processClass.machineClassFitness(role); @@ -1483,16 +1530,17 @@ class ClusterControllerData { it.second.details.interf.locality.dcId() == dcId) { auto sharing = preferredSharing.find(it.first); fitness_workers[std::make_tuple(fitness, - id_used[it.first], + id_used[it.first].getWeight(), isLongLivedStateless(it.first), - sharing != preferredSharing.end() ? sharing->second : 1e6)] + sharing != preferredSharing.end() ? sharing->second : 1e6, + id_used[it.first].getUniqueHash())] .push_back(it.second.details); } } if (fitness_workers.size()) { auto worker = deterministicRandom()->randomChoice(fitness_workers.begin()->second); - id_used[worker.interf.locality.processId()]++; + id_used[worker.interf.locality.processId()].addRole(role); return WorkerFitnessInfo(worker, std::max(ProcessClass::GoodFit, std::get<0>(fitness_workers.begin()->first)), std::get<1>(fitness_workers.begin()->first)); @@ -1506,13 +1554,17 @@ class ClusterControllerData { ProcessClass::ClusterRole role, int amount, DatabaseConfiguration const& conf, - std::map>, int>& id_used, + WorkerUsages& id_used, std::map>, int> preferredSharing = {}, Optional minWorker = Optional(), bool checkStable = false) { - std::map, std::vector> fitness_workers; + // for avoiding recruiting workers with worse fitness + ProcessClass::Fitness used_fitness = ProcessClass::NeverAssign; + std::map, std::vector> + fitness_workers; std::vector results; if (minWorker.present()) { + used_fitness = minWorker.get().fitness; results.push_back(minWorker.get().worker); } if (amount <= results.size()) { @@ -1525,24 +1577,27 @@ class ClusterControllerData { !conf.isExcludedServer(it.second.details.interf.addresses(), it.second.details.interf.locality) && !isExcludedDegradedServer(it.second.details.interf.addresses()) && it.second.details.interf.locality.dcId() == dcId && - (!minWorker.present() || - (it.second.details.interf.id() != minWorker.get().worker.interf.id() && - (fitness < minWorker.get().fitness || - (fitness == minWorker.get().fitness && id_used[it.first] <= minWorker.get().used))))) { + (!minWorker.present() || (it.second.details.interf.id() != minWorker.get().worker.interf.id()))) { auto sharing = preferredSharing.find(it.first); fitness_workers[std::make_tuple(fitness, - id_used[it.first], + id_used[it.first].getWeight(), isLongLivedStateless(it.first), - sharing != preferredSharing.end() ? sharing->second : 1e6)] + sharing != preferredSharing.end() ? sharing->second : 1e6, + id_used[it.first].getUniqueHash())] .push_back(it.second.details); } } for (auto& it : fitness_workers) { + ProcessClass::Fitness next_fitness = std::get<0>(it.first); + + if (next_fitness > used_fitness) + break; // do not recruit with a greater fitness + used_fitness = next_fitness; deterministicRandom()->randomShuffle(it.second); for (int i = 0; i < it.second.size(); i++) { results.push_back(it.second[i]); - id_used[it.second[i].interf.locality.processId()]++; + id_used[it.second[i].interf.locality.processId()].addRole(role); if (results.size() == amount) return results; } @@ -1559,7 +1614,7 @@ class ClusterControllerData { ProcessClass::Fitness worstFit; ProcessClass::ClusterRole role; int count; - int worstUsed = 1; + unsigned worstUsed = 1; bool degraded = false; RoleFitness(int bestFit, int worstFit, int count, ProcessClass::ClusterRole role) @@ -1576,7 +1631,7 @@ class ClusterControllerData { RoleFitness(const std::vector& workers, ProcessClass::ClusterRole role, - const std::map>, int>& id_used) + const WorkerUsages& id_used) : role(role) { // Every recruitment will attempt to recruit the preferred amount through GoodFit, // So a recruitment which only has BestFit is not better than one that has a GoodFit process @@ -1592,7 +1647,7 @@ class ClusterControllerData { TraceEvent(SevError, "UsedNotFound").detail("ProcessId", it.interf.locality.processId().get()); ASSERT(false); } - if (thisUsed->second == 0) { + if ((unsigned)thisUsed->second.getWeight() == 0) { TraceEvent(SevError, "UsedIsZero").detail("ProcessId", it.interf.locality.processId().get()); ASSERT(false); } @@ -1601,9 +1656,9 @@ class ClusterControllerData { if (thisFit > worstFit) { worstFit = thisFit; - worstUsed = thisUsed->second; + worstUsed = thisUsed->second.getWeight(); } else if (thisFit == worstFit) { - worstUsed = std::max(worstUsed, thisUsed->second); + worstUsed = std::max(worstUsed, thisUsed->second.getWeight()); } degraded = degraded || it.degraded; } @@ -1665,15 +1720,15 @@ class ClusterControllerData { return result; } - void updateKnownIds(std::map>, int>* id_used) { - (*id_used)[masterProcessId]++; - (*id_used)[clusterControllerProcessId]++; + void updateKnownIds(WorkerUsages* id_used) { + (*id_used)[masterProcessId].addRole(ProcessClass::Master); + (*id_used)[clusterControllerProcessId].addRole(ProcessClass::ClusterController); } RecruitRemoteFromConfigurationReply findRemoteWorkersForConfiguration( RecruitRemoteFromConfigurationRequest const& req) { RecruitRemoteFromConfigurationReply result; - std::map>, int> id_used; + WorkerUsages id_used; updateKnownIds(&id_used); @@ -1742,7 +1797,7 @@ class ClusterControllerData { Optional dcId, bool checkGoodRecruitment) { RecruitFromConfigurationReply result; - std::map>, int> id_used; + WorkerUsages id_used; updateKnownIds(&id_used); ASSERT(dcId.present()); @@ -1985,7 +2040,7 @@ class ClusterControllerData { throw no_more_servers(); } else { RecruitFromConfigurationReply result; - std::map>, int> id_used; + WorkerUsages id_used; updateKnownIds(&id_used); auto tlogs = getWorkersForTlogs(req.configuration, req.configuration.tLogReplicationFactor, @@ -2183,17 +2238,18 @@ class ClusterControllerData { } void updateIdUsed(const std::vector& workers, - std::map>, int>& id_used) { + ProcessClass::ClusterRole role, + WorkerUsages& id_used) { for (auto& it : workers) { - id_used[it.locality.processId()]++; + id_used[it.locality.processId()].addRole(role); } } void compareWorkers(const DatabaseConfiguration& conf, const std::vector& first, - const std::map>, int>& firstUsed, + WorkerUsages& firstUsed, const std::vector& second, - const std::map>, int>& secondUsed, + WorkerUsages& secondUsed, ProcessClass::ClusterRole role, std::string description) { std::vector firstDetails; @@ -2247,17 +2303,17 @@ class ClusterControllerData { if (!remoteDCUsedAsSatellite) { RecruitFromConfigurationReply compare = findWorkersForConfigurationDispatch(req, false); - std::map>, int> firstUsed; - std::map>, int> secondUsed; + WorkerUsages firstUsed; + WorkerUsages secondUsed; updateKnownIds(&firstUsed); updateKnownIds(&secondUsed); - updateIdUsed(rep.tLogs, firstUsed); - updateIdUsed(compare.tLogs, secondUsed); + updateIdUsed(rep.tLogs, ProcessClass::TLog, firstUsed); + updateIdUsed(compare.tLogs, ProcessClass::TLog, secondUsed); compareWorkers( req.configuration, rep.tLogs, firstUsed, compare.tLogs, secondUsed, ProcessClass::TLog, "TLog"); - updateIdUsed(rep.satelliteTLogs, firstUsed); - updateIdUsed(compare.satelliteTLogs, secondUsed); + updateIdUsed(rep.satelliteTLogs, ProcessClass::TLog, firstUsed); + updateIdUsed(compare.satelliteTLogs, ProcessClass::TLog, secondUsed); compareWorkers(req.configuration, rep.satelliteTLogs, firstUsed, @@ -2265,12 +2321,14 @@ class ClusterControllerData { secondUsed, ProcessClass::TLog, "Satellite"); - updateIdUsed(rep.commitProxies, firstUsed); - updateIdUsed(compare.commitProxies, secondUsed); - updateIdUsed(rep.grvProxies, firstUsed); - updateIdUsed(compare.grvProxies, secondUsed); - updateIdUsed(rep.resolvers, firstUsed); - updateIdUsed(compare.resolvers, secondUsed); + updateIdUsed(rep.commitProxies, ProcessClass::CommitProxy, firstUsed); + updateIdUsed(compare.commitProxies, ProcessClass::CommitProxy, secondUsed); + updateIdUsed(rep.grvProxies, ProcessClass::GrvProxy, firstUsed); + updateIdUsed(compare.grvProxies, ProcessClass::GrvProxy, secondUsed); + updateIdUsed(rep.resolvers, ProcessClass::Resolver, firstUsed); + updateIdUsed(compare.resolvers, ProcessClass::Resolver, secondUsed); + updateIdUsed(rep.backupWorkers, ProcessClass::Backup, firstUsed); + updateIdUsed(compare.backupWorkers, ProcessClass::Backup, secondUsed); compareWorkers(req.configuration, rep.commitProxies, firstUsed, @@ -2292,8 +2350,6 @@ class ClusterControllerData { secondUsed, ProcessClass::Resolver, "Resolver"); - updateIdUsed(rep.backupWorkers, firstUsed); - updateIdUsed(compare.backupWorkers, secondUsed); compareWorkers(req.configuration, rep.backupWorkers, firstUsed, @@ -2318,7 +2374,7 @@ class ClusterControllerData { } try { - std::map>, int> id_used; + WorkerUsages id_used; getWorkerForRoleInDatacenter(regions[0].dcId, ProcessClass::ClusterController, ProcessClass::ExcludeFit, @@ -2378,9 +2434,10 @@ class ClusterControllerData { } void updateIdUsed(const std::vector& workers, - std::map>, int>& id_used) { + ProcessClass::ClusterRole role, + WorkerUsages& id_used) { for (auto& it : workers) { - id_used[it.interf.locality.processId()]++; + id_used[it.interf.locality.processId()].addRole(role); } } @@ -2557,10 +2614,10 @@ class ClusterControllerData { oldMasterFit = std::max(oldMasterFit, ProcessClass::ExcludeFit); } - std::map>, int> id_used; - std::map>, int> old_id_used; - id_used[clusterControllerProcessId]++; - old_id_used[clusterControllerProcessId]++; + WorkerUsages id_used; + WorkerUsages old_id_used; + id_used[clusterControllerProcessId].addRole(ProcessClass::ClusterController); + old_id_used[clusterControllerProcessId].addRole(ProcessClass::ClusterController); WorkerFitnessInfo mworker = getWorkerForRoleInDatacenter( clusterControllerDcId, ProcessClass::Master, ProcessClass::NeverAssign, db.config, id_used, {}, true); auto newMasterFit = mworker.worker.processClass.machineClassFitness(ProcessClass::Master); @@ -2568,7 +2625,7 @@ class ClusterControllerData { newMasterFit = std::max(newMasterFit, ProcessClass::ExcludeFit); } - old_id_used[masterWorker->first]++; + old_id_used[masterWorker->first].addRole(ProcessClass::Master); if (oldMasterFit < newMasterFit) { TraceEvent("NewRecruitmentIsWorse", id) .detail("OldMasterFit", oldMasterFit) @@ -2608,7 +2665,7 @@ class ClusterControllerData { } // Check tLog fitness - updateIdUsed(tlogs, old_id_used); + updateIdUsed(tlogs, ProcessClass::TLog, old_id_used); RoleFitness oldTLogFit(tlogs, ProcessClass::TLog, old_id_used); auto newTLogs = getWorkersForTlogs(db.config, db.config.tLogReplicationFactor, @@ -2633,7 +2690,7 @@ class ClusterControllerData { } } - updateIdUsed(satellite_tlogs, old_id_used); + updateIdUsed(satellite_tlogs, ProcessClass::TLog, old_id_used); RoleFitness oldSatelliteTLogFit(satellite_tlogs, ProcessClass::TLog, old_id_used); bool newSatelliteFallback = false; auto newSatelliteTLogs = satellite_tlogs; @@ -2693,7 +2750,7 @@ class ClusterControllerData { return false; } - updateIdUsed(remote_tlogs, old_id_used); + updateIdUsed(remote_tlogs, ProcessClass::TLog, old_id_used); RoleFitness oldRemoteTLogFit(remote_tlogs, ProcessClass::TLog, old_id_used); std::vector exclusionWorkerIds; auto fn = [](const WorkerDetails& in) { return in.interf.id(); }; @@ -2717,7 +2774,7 @@ class ClusterControllerData { oldTLogFit.count * std::max(1, db.config.desiredLogRouterCount / std::max(1, oldTLogFit.count)); int newRouterCount = newTLogFit.count * std::max(1, db.config.desiredLogRouterCount / std::max(1, newTLogFit.count)); - updateIdUsed(log_routers, old_id_used); + updateIdUsed(log_routers, ProcessClass::LogRouter, old_id_used); RoleFitness oldLogRoutersFit(log_routers, ProcessClass::LogRouter, old_id_used); RoleFitness newLogRoutersFit = oldLogRoutersFit; if (db.config.usableRegions > 1 && dbi.recoveryState == RecoveryState::FULLY_RECOVERED) { @@ -2741,9 +2798,9 @@ class ClusterControllerData { } // Check proxy/grvProxy/resolver fitness - updateIdUsed(commitProxyClasses, old_id_used); - updateIdUsed(grvProxyClasses, old_id_used); - updateIdUsed(resolverClasses, old_id_used); + updateIdUsed(commitProxyClasses, ProcessClass::CommitProxy, old_id_used); + updateIdUsed(grvProxyClasses, ProcessClass::GrvProxy, old_id_used); + updateIdUsed(resolverClasses, ProcessClass::Resolver, old_id_used); RoleFitness oldCommitProxyFit(commitProxyClasses, ProcessClass::CommitProxy, old_id_used); RoleFitness oldGrvProxyFit(grvProxyClasses, ProcessClass::GrvProxy, old_id_used); RoleFitness oldResolverFit(resolverClasses, ProcessClass::Resolver, old_id_used); @@ -2807,7 +2864,7 @@ class ClusterControllerData { RoleFitness newResolverFit(resolvers, ProcessClass::Resolver, id_used); // Check backup worker fitness - updateIdUsed(backup_workers, old_id_used); + updateIdUsed(backup_workers, ProcessClass::Backup, old_id_used); RoleFitness oldBackupWorkersFit(backup_workers, ProcessClass::Backup, old_id_used); const int nBackup = backup_addresses.size(); RoleFitness newBackupWorkersFit(getWorkersForRoleInDatacenter(clusterControllerDcId, @@ -2936,30 +2993,30 @@ class ClusterControllerData { } // Returns a map of for all non-singleton roles - std::map>, int> getUsedIds() { - std::map>, int> idUsed; + WorkerUsages getUsedIds() { + WorkerUsages idUsed; updateKnownIds(&idUsed); auto& dbInfo = db.serverInfo->get(); for (const auto& tlogset : dbInfo.logSystemConfig.tLogs) { for (const auto& tlog : tlogset.tLogs) { if (tlog.present()) { - idUsed[tlog.interf().filteredLocality.processId()]++; + idUsed[tlog.interf().filteredLocality.processId()].addRole(ProcessClass::TLog); } } } for (const CommitProxyInterface& interf : dbInfo.client.commitProxies) { ASSERT(interf.processId.present()); - idUsed[interf.processId]++; + idUsed[interf.processId].addRole(ProcessClass::CommitProxy); } for (const GrvProxyInterface& interf : dbInfo.client.grvProxies) { ASSERT(interf.processId.present()); - idUsed[interf.processId]++; + idUsed[interf.processId].addRole(ProcessClass::GrvProxy); } for (const ResolverInterface& interf : dbInfo.resolvers) { ASSERT(interf.locality.processId().present()); - idUsed[interf.locality.processId()]++; + idUsed[interf.locality.processId()].addRole(ProcessClass::Resolver); } return idUsed; } diff --git a/fdbserver/clustercontroller/ClusterRecovery.cpp b/fdbserver/clustercontroller/ClusterRecovery.cpp index fd8a130f902..8446b5d9503 100644 --- a/fdbserver/clustercontroller/ClusterRecovery.cpp +++ b/fdbserver/clustercontroller/ClusterRecovery.cpp @@ -81,8 +81,8 @@ Future recruitNewMaster(ClusterControllerData* cluster, // We must recruit the master in the same data center as the cluster controller. // This should always be possible, because we can recruit the master on the same process as the cluster // controller. - std::map>, int> id_used; - id_used[cluster->clusterControllerProcessId]++; + ClusterControllerData::WorkerUsages id_used; + id_used[cluster->clusterControllerProcessId].addRole(ProcessClass::ClusterController); masterWorker = cluster->getWorkerForRoleInDatacenter( cluster->clusterControllerDcId, ProcessClass::Master, ProcessClass::NeverAssign, db->config, id_used); if ((masterWorker.worker.processClass.machineClassFitness(ProcessClass::Master) >