From 0b6fecddbc244c3d7d8b8b97ca0595f19d8b05ba Mon Sep 17 00:00:00 2001 From: Suraj Gupta Date: Mon, 13 Sep 2021 09:58:38 -0400 Subject: [PATCH] Refactor logic for recruiting singletons. This commit refactors the logic for recruiting singletons, which is done by the ClusterController. This allows for far easier additions of new singletons in the future, and also cleans up the code. Also, the logic for recruiting DD was changed to mirror the logic for recruiting RK. Although the logic for RK allows there to be many RKs existing at once, the moveKeysLock mechanism used by DD still prevents multiple DDs existing at once. --- fdbclient/ServerKnobs.cpp | 2 +- fdbclient/ServerKnobs.h | 2 +- fdbserver/ClusterController.actor.cpp | 495 +++++++++++++++++--------- fdbserver/DataDistributorInterface.h | 6 +- fdbserver/worker.actor.cpp | 2 +- 5 files changed, 341 insertions(+), 166 deletions(-) diff --git a/fdbclient/ServerKnobs.cpp b/fdbclient/ServerKnobs.cpp index c1599fc604..b8ec2c4177 100644 --- a/fdbclient/ServerKnobs.cpp +++ b/fdbclient/ServerKnobs.cpp @@ -241,7 +241,6 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi init( DD_LOCATION_CACHE_SIZE, 2000000 ); if( randomize && BUGGIFY ) DD_LOCATION_CACHE_SIZE = 3; init( MOVEKEYS_LOCK_POLLING_DELAY, 5.0 ); init( DEBOUNCE_RECRUITING_DELAY, 5.0 ); - init( DD_FAILURE_TIME, 1.0 ); if( randomize && BUGGIFY ) DD_FAILURE_TIME = 10.0; init( DD_ZERO_HEALTHY_TEAM_DELAY, 1.0 ); init( REBALANCE_MAX_RETRIES, 100 ); init( DD_OVERLAP_PENALTY, 10000 ); @@ -466,6 +465,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi init( VERSION_LAG_METRIC_INTERVAL, 0.5 ); if( randomize && BUGGIFY ) VERSION_LAG_METRIC_INTERVAL = 10.0; init( MAX_VERSION_DIFFERENCE, 20 * VERSIONS_PER_SECOND ); init( FORCE_RECOVERY_CHECK_DELAY, 5.0 ); + init( DATA_DISTRIBUTOR_FAILURE_TIME, 1.0 ); if( randomize && BUGGIFY ) DATA_DISTRIBUTOR_FAILURE_TIME = 10.0; init( RATEKEEPER_FAILURE_TIME, 1.0 ); init( REPLACE_INTERFACE_DELAY, 60.0 ); init( REPLACE_INTERFACE_CHECK_DELAY, 5.0 ); diff --git a/fdbclient/ServerKnobs.h b/fdbclient/ServerKnobs.h index e509246024..dbd4dd9e1d 100644 --- a/fdbclient/ServerKnobs.h +++ b/fdbclient/ServerKnobs.h @@ -224,7 +224,6 @@ public: // Remove wrong storage engines double DD_REMOVE_STORE_ENGINE_DELAY; // wait for the specified time before remove the next batch - double DD_FAILURE_TIME; double DD_ZERO_HEALTHY_TEAM_DELAY; // KeyValueStore SQLITE @@ -390,6 +389,7 @@ public: double VERSION_LAG_METRIC_INTERVAL; int64_t MAX_VERSION_DIFFERENCE; double FORCE_RECOVERY_CHECK_DELAY; + double DATA_DISTRIBUTOR_FAILURE_TIME; double RATEKEEPER_FAILURE_TIME; double REPLACE_INTERFACE_DELAY; double REPLACE_INTERFACE_CHECK_DELAY; diff --git a/fdbserver/ClusterController.actor.cpp b/fdbserver/ClusterController.actor.cpp index b0b7ea0227..b82b9803cb 100644 --- a/fdbserver/ClusterController.actor.cpp +++ b/fdbserver/ClusterController.actor.cpp @@ -53,6 +53,8 @@ void failAfter(Future trigger, Endpoint e); +static const int PID_USED_AMP_FOR_NON_SINGLETON = 100; + struct WorkerInfo : NonCopyable { Future watcher; ReplyPromise reply; @@ -2711,7 +2713,9 @@ public: return false; } - bool isUsedNotMaster(Optional processId) { + // Returns true iff processId is currently being used + // for any non-singleton role other than master + bool isUsedNotMaster(Optional processId) const { ASSERT(masterProcessId.present()); if (processId == masterProcessId) return false; @@ -2741,7 +2745,10 @@ public: return false; } - bool onMasterIsBetter(const WorkerDetails& worker, ProcessClass::ClusterRole role) { + // Returns true iff + // - role is master, or + // - role is a singleton AND worker's pid is being used for any non-singleton role + bool onMasterIsBetter(const WorkerDetails& worker, ProcessClass::ClusterRole role) const { ASSERT(masterProcessId.present()); const auto& pid = worker.interf.locality.processId(); if ((role != ProcessClass::DataDistributor && role != ProcessClass::Ratekeeper) || @@ -2751,6 +2758,7 @@ public: return isUsedNotMaster(pid); } + // Returns a map of for all non-singleton roles std::map>, int> getUsedIds() { std::map>, int> idUsed; updateKnownIds(&idUsed); @@ -3022,9 +3030,14 @@ public: Version datacenterVersionDifference; PromiseStream> addActor; bool versionDifferenceUpdated; - bool recruitingDistributor; - Optional recruitingRatekeeperID; + + // recruitX is used to signal when role X needs to be (re)recruited. + // recruitingXID is used to track the ID of X's interface which is being recruited. + // We use AsyncVars to kill (i.e. halt) singletons that have been replaced. + AsyncVar recruitDistributor; + Optional recruitingDistributorID; AsyncVar recruitRatekeeper; + Optional recruitingRatekeeperID; // Stores the health information from a particular worker's perspective. struct WorkerHealth { @@ -3060,7 +3073,7 @@ public: clusterControllerDcId(locality.dcId()), id(ccInterface.id()), ac(false), outstandingRequestChecker(Void()), outstandingRemoteRequestChecker(Void()), startTime(now()), goodRecruitmentTime(Never()), goodRemoteRecruitmentTime(Never()), datacenterVersionDifference(0), versionDifferenceUpdated(false), - recruitingDistributor(false), recruitRatekeeper(false), + recruitDistributor(false), recruitRatekeeper(false), clusterControllerMetrics("ClusterController", id.toString()), openDatabaseRequests("OpenDatabaseRequests", clusterControllerMetrics), registerWorkerRequests("RegisterWorkerRequests", clusterControllerMetrics), @@ -3084,6 +3097,63 @@ public: } }; +// Wrapper for singleton interfaces +template +struct Singleton { + const Optional& interface; + + Singleton(const Optional& interface) : interface(interface) {} + + virtual Role getRole() const = 0; + virtual ProcessClass::ClusterRole getClusterRole() const = 0; + + virtual void setOnDb(ClusterControllerData* cc) const = 0; + virtual void halt(ClusterControllerData* cc, Optional> pid) const = 0; + virtual void recruit(ClusterControllerData* cc) const = 0; +}; + +struct RatekeeperSingleton : Singleton { + + RatekeeperSingleton(const Optional& interface) : Singleton(interface) {} + + Role getRole() const { return Role::RATEKEEPER; } + ProcessClass::ClusterRole getClusterRole() const { return ProcessClass::Ratekeeper; } + + void setOnDb(ClusterControllerData* cc) const { + if (interface.present()) { + cc->db.setRatekeeper(interface.get()); + } + } + void halt(ClusterControllerData* cc, Optional> pid) const { + if (interface.present()) { + cc->id_worker[pid].haltRatekeeper = + brokenPromiseToNever(interface.get().haltRatekeeper.getReply(HaltRatekeeperRequest(cc->id))); + } + } + void recruit(ClusterControllerData* cc) const { cc->recruitRatekeeper.set(true); } +}; + +struct DataDistributorSingleton : Singleton { + + DataDistributorSingleton(const Optional& interface) : Singleton(interface) {} + + Role getRole() const { return Role::DATA_DISTRIBUTOR; } + ProcessClass::ClusterRole getClusterRole() const { return ProcessClass::DataDistributor; } + + void setOnDb(ClusterControllerData* cc) const { + if (interface.present()) { + cc->db.setDistributor(interface.get()); + } + } + void halt(ClusterControllerData* cc, Optional> pid) const { + if (interface.present()) { + cc->id_worker[pid].haltDistributor = + brokenPromiseToNever(interface.get().haltDataDistributor.getReply(HaltDataDistributorRequest(cc->id))); + } + } + void recruit(ClusterControllerData* cc) const { cc->recruitDistributor.set(true); } +}; + ACTOR Future clusterWatchDatabase(ClusterControllerData* cluster, ClusterControllerData::DBInfo* db) { state MasterInterface iMaster; @@ -3295,108 +3365,167 @@ void checkOutstandingStorageRequests(ClusterControllerData* self) { } } -void checkBetterDDOrRK(ClusterControllerData* self) { +// Finds and returns a new process for role +WorkerDetails findNewProcessForSingleton(ClusterControllerData* self, + const ProcessClass::ClusterRole role, + std::map>, int>& id_used) { + // find new process in cluster for role + WorkerDetails newWorker = + self->getWorkerForRoleInDatacenter( + self->clusterControllerDcId, role, ProcessClass::NeverAssign, self->db.config, id_used, {}, true) + .worker; + + // check if master's process is actually better suited for role + if (self->onMasterIsBetter(newWorker, role)) { + newWorker = self->id_worker[self->masterProcessId.get()].details; + } + + // acknowledge that the pid is now potentially used by this role as well + id_used[newWorker.interf.locality.processId()]++; + + return newWorker; +} + +// Return best possible fitness for singleton. Note that lower fitness is better. +ProcessClass::Fitness findBestFitnessForSingleton(const ClusterControllerData* self, + const WorkerDetails& worker, + const ProcessClass::ClusterRole& role) { + auto bestFitness = worker.processClass.machineClassFitness(role); + // If the process has been marked as excluded, we take the max with ExcludeFit to ensure its fit + // is at least as bad as ExcludeFit. This assists with successfully offboarding such processes + // and removing them from the cluster. + if (self->db.config.isExcludedServer(worker.interf.addresses())) { + bestFitness = std::max(bestFitness, ProcessClass::ExcludeFit); + } + return bestFitness; +} + +// Returns true iff the singleton is healthy. "Healthy" here means that +// the singleton is stable (see below) and doesn't need to be rerecruited. +// Side effects: (possibly) initiates recruitment +template +bool rerecruitSingleton(ClusterControllerData* self, + const WorkerDetails& newWorker, + const Singleton& singleton, + const ProcessClass::Fitness& bestFitness, + const Optional recruitingID) { + // A singleton is stable if it exists in cluster, has not been killed off of proc and is not being recruited + bool isStableSingleton = singleton.interface.present() && + self->id_worker.count(singleton.interface.get().locality.processId()) && + (!recruitingID.present() || (recruitingID.get() == singleton.interface.get().id())); + + if (!isStableSingleton) { + return false; // not healthy because unstable + } + + auto& currWorker = self->id_worker[singleton.interface.get().locality.processId()]; + auto currFitness = currWorker.details.processClass.machineClassFitness(singleton.getClusterRole()); + if (currWorker.priorityInfo.isExcluded) { + currFitness = ProcessClass::ExcludeFit; + } + // If any of the following conditions are met, we will switch the singleton's process: + // - if the current proc is used by some non-master, non-singleton role + // - if the current fitness is less than optimal (lower fitness is better) + // - if currently at peak fitness but on same process as master, and the new worker is on different process + bool shouldRerecruit = + self->isUsedNotMaster(currWorker.details.interf.locality.processId()) || bestFitness < currFitness || + (currFitness == bestFitness && currWorker.details.interf.locality.processId() == self->masterProcessId && + newWorker.interf.locality.processId() != self->masterProcessId); + if (shouldRerecruit) { + std::string roleAbbr = singleton.getRole().abbreviation; + TraceEvent(("CCHalt" + roleAbbr).c_str(), self->id) + .detail(roleAbbr + "ID", singleton.interface.get().id()) + .detail("Excluded", currWorker.priorityInfo.isExcluded) + .detail("Fitness", currFitness) + .detail("BestFitness", bestFitness); + singleton.recruit(self); + return false; // not healthy since needed to be rerecruited + } else { + return true; // healthy because doesn't need to be rerecruited + } +} + +// Returns a mapping from pid->pidCount for pids +std::map>, int> getColocCounts(const vector>>& pids) { + std::map>, int> counts; + for (const auto& pid : pids) { + if (counts.find(pid) == counts.end()) { + counts[pid] = 1; + } else { + ++counts[pid]; + } + } + return counts; +} + +// Checks if there exists a better process for each singleton (e.g. DD) compared +// to the process it is currently on. +void checkBetterSingletons(ClusterControllerData* self) { if (!self->masterProcessId.present() || self->db.serverInfo->get().recoveryState < RecoveryState::ACCEPTING_COMMITS) { return; } + // note: this map doesn't consider pids used by existing singletons std::map>, int> id_used = self->getUsedIds(); - WorkerDetails newRKWorker = self->getWorkerForRoleInDatacenter(self->clusterControllerDcId, - ProcessClass::Ratekeeper, - ProcessClass::NeverAssign, - self->db.config, - id_used, - {}, - true) - .worker; - if (self->onMasterIsBetter(newRKWorker, ProcessClass::Ratekeeper)) { - newRKWorker = self->id_worker[self->masterProcessId.get()].details; - } - id_used = self->getUsedIds(); - for (auto& it : id_used) { - it.second *= 2; - } - id_used[newRKWorker.interf.locality.processId()]++; - WorkerDetails newDDWorker = self->getWorkerForRoleInDatacenter(self->clusterControllerDcId, - ProcessClass::DataDistributor, - ProcessClass::NeverAssign, - self->db.config, - id_used, - {}, - true) - .worker; - if (self->onMasterIsBetter(newDDWorker, ProcessClass::DataDistributor)) { - newDDWorker = self->id_worker[self->masterProcessId.get()].details; - } - auto bestFitnessForRK = newRKWorker.processClass.machineClassFitness(ProcessClass::Ratekeeper); - if (self->db.config.isExcludedServer(newRKWorker.interf.addresses())) { - bestFitnessForRK = std::max(bestFitnessForRK, ProcessClass::ExcludeFit); - } - auto bestFitnessForDD = newDDWorker.processClass.machineClassFitness(ProcessClass::DataDistributor); - if (self->db.config.isExcludedServer(newDDWorker.interf.addresses())) { - bestFitnessForDD = std::max(bestFitnessForDD, ProcessClass::ExcludeFit); - } - //TraceEvent("CheckBetterDDorRKNewRecruits", self->id).detail("MasterProcessId", self->masterProcessId) - //.detail("NewRecruitRKProcessId", newRKWorker.interf.locality.processId()).detail("NewRecruiteDDProcessId", - // newDDWorker.interf.locality.processId()); - Optional> currentRKProcessId; - Optional> currentDDProcessId; + // TODO: moved this amplification to before getWorker, verify this is valid + // 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. + id_used = self->getUsedIds(); // don't need this if the move was correct + for (auto& it : id_used) { + it.second *= PID_USED_AMP_FOR_NON_SINGLETON; + } + + // Try to find a new process for each singleton. + WorkerDetails newRKWorker = findNewProcessForSingleton(self, ProcessClass::Ratekeeper, id_used); + WorkerDetails newDDWorker = findNewProcessForSingleton(self, ProcessClass::DataDistributor, id_used); + + // Find best possible fitnesses for each singleton. + auto bestFitnessForRK = findBestFitnessForSingleton(self, newRKWorker, ProcessClass::Ratekeeper); + auto bestFitnessForDD = findBestFitnessForSingleton(self, newDDWorker, ProcessClass::DataDistributor); auto& db = self->db.serverInfo->get(); - bool ratekeeperHealthy = false; - if (db.ratekeeper.present() && self->id_worker.count(db.ratekeeper.get().locality.processId()) && - (!self->recruitingRatekeeperID.present() || (self->recruitingRatekeeperID.get() == db.ratekeeper.get().id()))) { - auto& rkWorker = self->id_worker[db.ratekeeper.get().locality.processId()]; - currentRKProcessId = rkWorker.details.interf.locality.processId(); - auto rkFitness = rkWorker.details.processClass.machineClassFitness(ProcessClass::Ratekeeper); - if (rkWorker.priorityInfo.isExcluded) { - rkFitness = ProcessClass::ExcludeFit; - } - if (self->isUsedNotMaster(rkWorker.details.interf.locality.processId()) || bestFitnessForRK < rkFitness || - (rkFitness == bestFitnessForRK && rkWorker.details.interf.locality.processId() == self->masterProcessId && - newRKWorker.interf.locality.processId() != self->masterProcessId)) { - TraceEvent("CCHaltRK", self->id) - .detail("RKID", db.ratekeeper.get().id()) - .detail("Excluded", rkWorker.priorityInfo.isExcluded) - .detail("Fitness", rkFitness) - .detail("BestFitness", bestFitnessForRK); - self->recruitRatekeeper.set(true); - } else { - ratekeeperHealthy = true; - } + auto rkSingleton = RatekeeperSingleton(db.ratekeeper); + auto ddSingleton = DataDistributorSingleton(db.distributor); + + // Try to rerecruit the singletons to more optimal processes + bool rkHealthy = rerecruitSingleton( + self, newRKWorker, rkSingleton, bestFitnessForRK, self->recruitingRatekeeperID); + + bool ddHealthy = rerecruitSingleton( + self, newDDWorker, ddSingleton, bestFitnessForDD, self->recruitingDistributorID); + + // if any of the singletons are unhealthy (rerecruited or not stable), then do not + // consider any further re-recruitments + if (!(rkHealthy && ddHealthy)) { + return; } - if (!self->recruitingDistributor && db.distributor.present() && - self->id_worker.count(db.distributor.get().locality.processId())) { - auto& ddWorker = self->id_worker[db.distributor.get().locality.processId()]; - auto ddFitness = ddWorker.details.processClass.machineClassFitness(ProcessClass::DataDistributor); - currentDDProcessId = ddWorker.details.interf.locality.processId(); - if (ddWorker.priorityInfo.isExcluded) { - ddFitness = ProcessClass::ExcludeFit; - } - if (self->isUsedNotMaster(ddWorker.details.interf.locality.processId()) || bestFitnessForDD < ddFitness || - (ddFitness == bestFitnessForDD && ddWorker.details.interf.locality.processId() == self->masterProcessId && - newDDWorker.interf.locality.processId() != self->masterProcessId) || - (ddFitness == bestFitnessForDD && - newRKWorker.interf.locality.processId() != newDDWorker.interf.locality.processId() && ratekeeperHealthy && - currentRKProcessId.present() && currentDDProcessId == currentRKProcessId && - (newRKWorker.interf.locality.processId() != self->masterProcessId && - newDDWorker.interf.locality.processId() != self->masterProcessId))) { - TraceEvent("CCHaltDD", self->id) - .detail("DDID", db.distributor.get().id()) - .detail("Excluded", ddWorker.priorityInfo.isExcluded) - .detail("Fitness", ddFitness) - .detail("BestFitness", bestFitnessForDD) - .detail("CurrentRateKeeperProcessId", - currentRKProcessId.present() ? currentRKProcessId.get() : LiteralStringRef("None")) - .detail("CurrentDDProcessId", currentDDProcessId) - .detail("MasterProcessID", self->masterProcessId) - .detail("NewRKWorkers", newRKWorker.interf.locality.processId()) - .detail("NewDDWorker", newDDWorker.interf.locality.processId()); - ddWorker.haltDistributor = brokenPromiseToNever( - db.distributor.get().haltDataDistributor.getReply(HaltDataDistributorRequest(self->id))); + // if we reach here, we know that the singletons are healthy so let's + // check if we can colocate the singletons in a more optimal way + + // TODO: verify that we don't need to get the pid from the worker like we were doing before + Optional> currentRKProcessId = rkSingleton.interface.get().locality.processId(); + Optional> currentDDProcessId = ddSingleton.interface.get().locality.processId(); + Optional> newRKProcessId = newRKWorker.interf.locality.processId(); + Optional> newDDProcessId = newRKWorker.interf.locality.processId(); + + auto currColocMap = getColocCounts({ currentRKProcessId, currentDDProcessId }); + auto newColocMap = getColocCounts({ newRKProcessId, newDDProcessId }); + + auto currColocCounts = std::make_tuple(currColocMap[newRKProcessId], currColocMap[newDDProcessId]); + auto newColocCounts = std::make_tuple(newColocMap[newRKProcessId], newColocMap[newDDProcessId]); + + // if the new coloc counts are collectively better (i.e. each singleton's coloc count has not increased) + if (newColocCounts <= currColocCounts) { + // rerecruit the singleton for which we have found a better process, if any + if (newColocMap[newRKProcessId] < currColocMap[currentRKProcessId]) { + rkSingleton.recruit(self); + } else if (newColocMap[newDDProcessId] < currColocMap[currentDDProcessId]) { + ddSingleton.recruit(self); } } } @@ -3410,7 +3539,7 @@ ACTOR Future doCheckOutstandingRequests(ClusterControllerData* self) { checkOutstandingRecruitmentRequests(self); checkOutstandingStorageRequests(self); - checkBetterDDOrRK(self); + checkBetterSingletons(self); self->checkRecoveryStalled(); if (self->betterMasterExists()) { @@ -3740,6 +3869,49 @@ void clusterRegisterMaster(ClusterControllerData* self, RegisterMasterRequest co checkOutstandingRequests(self); } +// Halts the registering (i.e. requesting) singleton if one is already in the process of being recruited +// or, halts the existing singleton in favour of the requesting one +template +void haltRegisteringOrCurrentSingleton(ClusterControllerData* self, + const WorkerInterface& worker, + const Singleton& currSingleton, + const Singleton& registeringSingleton, + const Optional recruitingID) { + ASSERT(currSingleton.getRole() == registeringSingleton.getRole()); + const UID registeringID = registeringSingleton.interface.get().id(); + const std::string roleName = currSingleton.getRole().roleName; + const std::string roleAbbr = currSingleton.getRole().abbreviation; + + // halt the requesting singleton if it isn't the one currently being recruited + if ((recruitingID.present() && recruitingID.get() != registeringID) || + self->clusterControllerDcId != worker.locality.dcId()) { + TraceEvent(("CCHaltRegistering" + roleName).c_str(), self->id) + .detail(roleAbbr + "ID", registeringID) + .detail("DcID", printable(self->clusterControllerDcId)) + .detail("ReqDcID", printable(worker.locality.dcId())) + .detail("Recruiting" + roleAbbr + "ID", recruitingID.present() ? recruitingID.get() : UID()); + if (registeringSingleton.getClusterRole() == ProcessClass::DataDistributor) { + } + registeringSingleton.halt(self, worker.locality.processId()); + } else if (!recruitingID.present()) { + // if not currently recruiting, then halt previous one in favour of requesting one + TraceEvent(("CCRegister" + roleName).c_str(), self->id).detail(roleAbbr + "ID", registeringID); + if (currSingleton.interface.present() && currSingleton.interface.get().id() != registeringID && + self->id_worker.count(currSingleton.interface.get().locality.processId())) { + TraceEvent(("CCHaltPrevious" + roleName).c_str(), self->id) + .detail(roleAbbr + "ID", currSingleton.interface.get().id()) + .detail("DcID", printable(self->clusterControllerDcId)) + .detail("ReqDcID", printable(worker.locality.dcId())) + .detail("Recruiting" + roleAbbr + "ID", recruitingID.present() ? recruitingID.get() : UID()); + currSingleton.halt(self, currSingleton.interface.get().locality.processId()); + } + // set the curr singleton if it doesn't exist or its different from the requesting one + if (!currSingleton.interface.present() || currSingleton.interface.get().id() != registeringID) { + registeringSingleton.setOnDb(self); + } + } +} + void registerWorker(RegisterWorkerRequest req, ClusterControllerData* self, ConfigBroadcaster* configBroadcaster) { const WorkerInterface& w = req.wi; ProcessClass newProcessClass = req.processClass; @@ -3866,43 +4038,21 @@ void registerWorker(RegisterWorkerRequest req, ClusterControllerData* self, Conf TEST(true); // Received an old worker registration request. } - if (req.distributorInterf.present() && !self->db.serverInfo->get().distributor.present() && - self->clusterControllerDcId == req.distributorInterf.get().locality.dcId() && !self->recruitingDistributor) { - const DataDistributorInterface& di = req.distributorInterf.get(); - TraceEvent("CCRegisterDataDistributor", self->id).detail("DDID", di.id()); - self->db.setDistributor(di); + // For each singleton + // - if the registering singleton conflicts with the singleton being recruited, kill the registering one + // - if the singleton is not being recruited, kill the existing one in favour of the registering one + if (req.distributorInterf.present()) { + auto currSingleton = DataDistributorSingleton(self->db.serverInfo->get().distributor); + auto registeringSingleton = DataDistributorSingleton(req.distributorInterf); + haltRegisteringOrCurrentSingleton( + self, w, currSingleton, registeringSingleton, self->recruitingDistributorID); } + if (req.ratekeeperInterf.present()) { - if ((self->recruitingRatekeeperID.present() && - self->recruitingRatekeeperID.get() != req.ratekeeperInterf.get().id()) || - self->clusterControllerDcId != w.locality.dcId()) { - TraceEvent("CCHaltRegisteringRatekeeper", self->id) - .detail("RKID", req.ratekeeperInterf.get().id()) - .detail("DcID", printable(self->clusterControllerDcId)) - .detail("ReqDcID", printable(w.locality.dcId())) - .detail("RecruitingRKID", - self->recruitingRatekeeperID.present() ? self->recruitingRatekeeperID.get() : UID()); - self->id_worker[w.locality.processId()].haltRatekeeper = brokenPromiseToNever( - req.ratekeeperInterf.get().haltRatekeeper.getReply(HaltRatekeeperRequest(self->id))); - } else if (!self->recruitingRatekeeperID.present()) { - const RatekeeperInterface& rki = req.ratekeeperInterf.get(); - const auto& ratekeeper = self->db.serverInfo->get().ratekeeper; - TraceEvent("CCRegisterRatekeeper", self->id).detail("RKID", rki.id()); - if (ratekeeper.present() && ratekeeper.get().id() != rki.id() && - self->id_worker.count(ratekeeper.get().locality.processId())) { - TraceEvent("CCHaltPreviousRatekeeper", self->id) - .detail("RKID", ratekeeper.get().id()) - .detail("DcID", printable(self->clusterControllerDcId)) - .detail("ReqDcID", printable(w.locality.dcId())) - .detail("RecruitingRKID", - self->recruitingRatekeeperID.present() ? self->recruitingRatekeeperID.get() : UID()); - self->id_worker[ratekeeper.get().locality.processId()].haltRatekeeper = - brokenPromiseToNever(ratekeeper.get().haltRatekeeper.getReply(HaltRatekeeperRequest(self->id))); - } - if (!ratekeeper.present() || ratekeeper.get().id() != rki.id()) { - self->db.setRatekeeper(rki); - } - } + auto currSingleton = RatekeeperSingleton(self->db.serverInfo->get().ratekeeper); + auto registeringSingleton = RatekeeperSingleton(req.ratekeeperInterf); + haltRegisteringOrCurrentSingleton( + self, w, currSingleton, registeringSingleton, self->recruitingRatekeeperID); } // Notify the worker to register again with new process class/exclusive property @@ -4537,41 +4687,64 @@ ACTOR Future handleForcedRecoveries(ClusterControllerData* self, ClusterCo } } -ACTOR Future startDataDistributor(ClusterControllerData* self) { +ACTOR Future startDataDistributor(ClusterControllerData* self) { wait(delay(0.0)); // If master fails at the same time, give it a chance to clear master PID. TraceEvent("CCStartDataDistributor", self->id).log(); loop { try { - state bool no_distributor = !self->db.serverInfo->get().distributor.present(); + state bool noDistributor = !self->db.serverInfo->get().distributor.present(); while (!self->masterProcessId.present() || self->masterProcessId != self->db.serverInfo->get().master.locality.processId() || self->db.serverInfo->get().recoveryState < RecoveryState::ACCEPTING_COMMITS) { wait(self->db.serverInfo->onChange() || delay(SERVER_KNOBS->WAIT_FOR_GOOD_RECRUITMENT_DELAY)); } - if (no_distributor && self->db.serverInfo->get().distributor.present()) { - return self->db.serverInfo->get().distributor.get(); + if (noDistributor && self->db.serverInfo->get().distributor.present()) { + // Existing distributor registers while waiting, so skip. + return Void(); } - std::map>, int> id_used = self->getUsedIds(); - WorkerFitnessInfo data_distributor = self->getWorkerForRoleInDatacenter(self->clusterControllerDcId, - ProcessClass::DataDistributor, - ProcessClass::NeverAssign, - self->db.config, - id_used); - state WorkerDetails worker = data_distributor.worker; + std::map>, int> idUsed = self->getUsedIds(); + WorkerFitnessInfo ddWorker = self->getWorkerForRoleInDatacenter(self->clusterControllerDcId, + ProcessClass::DataDistributor, + ProcessClass::NeverAssign, + self->db.config, + idUsed); + InitializeDataDistributorRequest req(deterministicRandom()->randomUniqueID()); + state WorkerDetails worker = ddWorker.worker; if (self->onMasterIsBetter(worker, ProcessClass::DataDistributor)) { worker = self->id_worker[self->masterProcessId.get()].details; } - InitializeDataDistributorRequest req(deterministicRandom()->randomUniqueID()); - TraceEvent("CCDataDistributorRecruit", self->id).detail("Addr", worker.interf.address()); + self->recruitingDistributorID = req.reqId; + TraceEvent("CCRecruitDataDistributor", self->id) + .detail("Addr", worker.interf.address()) + .detail("DDID", req.reqId); - ErrorOr distributor = wait(worker.interf.dataDistributor.getReplyUnlessFailedFor( + ErrorOr ddInterf = wait(worker.interf.dataDistributor.getReplyUnlessFailedFor( req, SERVER_KNOBS->WAIT_FOR_DISTRIBUTOR_JOIN_DELAY, 0)); - if (distributor.present()) { - TraceEvent("CCDataDistributorRecruited", self->id).detail("Addr", worker.interf.address()); - return distributor.get(); + + if (ddInterf.present()) { + self->recruitDistributor.set(false); + self->recruitingDistributorID = ddInterf.get().id(); + const auto& distributor = self->db.serverInfo->get().distributor; + TraceEvent("CCDataDistributorRecruited", self->id) + .detail("Addr", worker.interf.address()) + .detail("DDID", ddInterf.get().id()); + if (distributor.present() && distributor.get().id() != ddInterf.get().id() && + self->id_worker.count(distributor.get().locality.processId())) { + + TraceEvent("CCHaltDataDistributorAfterRecruit", self->id) + .detail("DDID", distributor.get().id()) + .detail("DcID", printable(self->clusterControllerDcId)); + + DataDistributorSingleton(distributor).halt(self, distributor.get().locality.processId()); + } + if (!distributor.present() || distributor.get().id() != ddInterf.get().id()) { + self->db.setDistributor(ddInterf.get()); + } + checkOutstandingRequests(self); + return Void(); } } catch (Error& e) { TraceEvent("CCDataDistributorRecruitError", self->id).error(e); @@ -4589,17 +4762,18 @@ ACTOR Future monitorDataDistributor(ClusterControllerData* self) { } loop { - if (self->db.serverInfo->get().distributor.present()) { - wait(waitFailureClient(self->db.serverInfo->get().distributor.get().waitFailure, - SERVER_KNOBS->DD_FAILURE_TIME)); - TraceEvent("CCDataDistributorDied", self->id) - .detail("DistributorId", self->db.serverInfo->get().distributor.get().id()); - self->db.clearInterf(ProcessClass::DataDistributorClass); + if (self->db.serverInfo->get().distributor.present() && !self->recruitDistributor.get()) { + choose { + when(wait(waitFailureClient(self->db.serverInfo->get().distributor.get().waitFailure, + SERVER_KNOBS->DATA_DISTRIBUTOR_FAILURE_TIME))) { + TraceEvent("CCDataDistributorDied", self->id) + .detail("DDID", self->db.serverInfo->get().distributor.get().id()); + self->db.clearInterf(ProcessClass::DataDistributorClass); + } + when(wait(self->recruitDistributor.onChange())) {} + } } else { - self->recruitingDistributor = true; - DataDistributorInterface distributorInterf = wait(startDataDistributor(self)); - self->recruitingDistributor = false; - self->db.setDistributor(distributorInterf); + wait(startDataDistributor(self)); } } } @@ -4652,8 +4826,7 @@ ACTOR Future startRatekeeper(ClusterControllerData* self) { TraceEvent("CCHaltRatekeeperAfterRecruit", self->id) .detail("RKID", ratekeeper.get().id()) .detail("DcID", printable(self->clusterControllerDcId)); - self->id_worker[ratekeeper.get().locality.processId()].haltRatekeeper = - brokenPromiseToNever(ratekeeper.get().haltRatekeeper.getReply(HaltRatekeeperRequest(self->id))); + RatekeeperSingleton(ratekeeper).halt(self, ratekeeper.get().locality.processId()); } if (!ratekeeper.present() || ratekeeper.get().id() != interf.get().id()) { self->db.setRatekeeper(interf.get()); diff --git a/fdbserver/DataDistributorInterface.h b/fdbserver/DataDistributorInterface.h index 2c78028f99..666918f3b7 100644 --- a/fdbserver/DataDistributorInterface.h +++ b/fdbserver/DataDistributorInterface.h @@ -30,15 +30,16 @@ struct DataDistributorInterface { RequestStream> waitFailure; RequestStream haltDataDistributor; struct LocalityData locality; + UID myId; RequestStream distributorSnapReq; RequestStream distributorExclCheckReq; RequestStream dataDistributorMetrics; DataDistributorInterface() {} - explicit DataDistributorInterface(const struct LocalityData& l) : locality(l) {} + explicit DataDistributorInterface(const struct LocalityData& l, UID id) : locality(l), myId(id) {} void initEndpoints() {} - UID id() const { return waitFailure.getEndpoint().token; } + UID id() const { return myId; } NetworkAddress address() const { return waitFailure.getEndpoint().getPrimaryAddress(); } bool operator==(const DataDistributorInterface& r) const { return id() == r.id(); } bool operator!=(const DataDistributorInterface& r) const { return !(*this == r); } @@ -49,6 +50,7 @@ struct DataDistributorInterface { waitFailure, haltDataDistributor, locality, + myId, distributorSnapReq, distributorExclCheckReq, dataDistributorMetrics); diff --git a/fdbserver/worker.actor.cpp b/fdbserver/worker.actor.cpp index 0ed71e75d8..b175d4f3d0 100644 --- a/fdbserver/worker.actor.cpp +++ b/fdbserver/worker.actor.cpp @@ -1617,7 +1617,7 @@ ACTOR Future workerServer(Reference connFile, when(InitializeDataDistributorRequest req = waitNext(interf.dataDistributor.getFuture())) { LocalLineage _; getCurrentLineage()->modify(&RoleLineage::role) = ProcessClass::ClusterRole::DataDistributor; - DataDistributorInterface recruited(locality); + DataDistributorInterface recruited(locality, req.reqId); recruited.initEndpoints(); if (ddInterf->get().present()) {