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()) {