fdbserver/ClusterController.actor.cpp
// Checks if there exists a better process for each singleton (e.g. DD) compared
// to the process it is currently on.
// Note: there is a lot of extra logic here to only recruit the blob manager when gate is open.
// When adding new singletons, just follow the ratekeeper/data distributor examples.
Checks if there exists a better process for each singleton  compared
to the process it is currently on.
Note: there is a lot of extra logic here to only recruit the blob manager when gate is open.
When adding new singletons, just follow the ratekeeper/data distributor examples.
Checks if there exists a better process for each singleton compared to the process it is currently on . Note : there is a lot of extra logic here to only recruit the blob manager when gate is open . When adding new singletons just follow the ratekeeper / data distributor examples .
void checkBetterSingletons(ClusterControllerData* self) {
	if (!self->masterProcessId.present() ||
	    self->db.serverInfo->get().recoveryState < RecoveryState::ACCEPTING_COMMITS) {
		return;
	}
	std::map<Optional<Standalone<StringRef>>, int> id_used = self->getUsedIds();
	for (auto& it : id_used) {
		it.second *= PID_USED_AMP_FOR_NON_SINGLETON;
	}
	WorkerDetails newRKWorker = findNewProcessForSingleton(self, ProcessClass::Ratekeeper, id_used);
	WorkerDetails newDDWorker = findNewProcessForSingleton(self, ProcessClass::DataDistributor, id_used);
	WorkerDetails newBMWorker;
	if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
		newBMWorker = findNewProcessForSingleton(self, ProcessClass::BlobManager, id_used);
	}
	auto bestFitnessForRK = findBestFitnessForSingleton(self, newRKWorker, ProcessClass::Ratekeeper);
	auto bestFitnessForDD = findBestFitnessForSingleton(self, newDDWorker, ProcessClass::DataDistributor);
	ProcessClass::Fitness bestFitnessForBM;
	if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
		bestFitnessForBM = findBestFitnessForSingleton(self, newBMWorker, ProcessClass::BlobManager);
	}
	auto& db = self->db.serverInfo->get();
	auto rkSingleton = RatekeeperSingleton(db.ratekeeper);
	auto ddSingleton = DataDistributorSingleton(db.distributor);
	BlobManagerSingleton bmSingleton(db.blobManager);
	bool rkHealthy = isHealthySingleton<RatekeeperInterface>(
	    self, newRKWorker, rkSingleton, bestFitnessForRK, self->recruitingRatekeeperID);
	bool ddHealthy = isHealthySingleton<DataDistributorInterface>(
	    self, newDDWorker, ddSingleton, bestFitnessForDD, self->recruitingDistributorID);
	bool bmHealthy = true;
	if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
		bmHealthy = isHealthySingleton<BlobManagerInterface>(
		    self, newBMWorker, bmSingleton, bestFitnessForBM, self->recruitingBlobManagerID);
	}
	if (!(rkHealthy && ddHealthy && bmHealthy)) {
		return;
	}
	Optional<Standalone<StringRef>> currRKProcessId = rkSingleton.interface.get().locality.processId();
	Optional<Standalone<StringRef>> currDDProcessId = ddSingleton.interface.get().locality.processId();
	Optional<Standalone<StringRef>> newRKProcessId = newRKWorker.interf.locality.processId();
	Optional<Standalone<StringRef>> newDDProcessId = newDDWorker.interf.locality.processId();
	Optional<Standalone<StringRef>> currBMProcessId, newBMProcessId;
	if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
		currBMProcessId = bmSingleton.interface.get().locality.processId();
		newBMProcessId = newBMWorker.interf.locality.processId();
	}
	std::vector<Optional<Standalone<StringRef>>> currPids = { currRKProcessId, currDDProcessId };
	std::vector<Optional<Standalone<StringRef>>> newPids = { newRKProcessId, newDDProcessId };
	if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
		currPids.emplace_back(currBMProcessId);
		newPids.emplace_back(newBMProcessId);
	}
	auto currColocMap = getColocCounts(currPids);
	auto newColocMap = getColocCounts(newPids);
	if (!CLIENT_KNOBS->ENABLE_BLOB_GRANULES) {
		ASSERT(currColocMap[currBMProcessId] == 0);
		ASSERT(newColocMap[newBMProcessId] == 0);
	}
	if (newColocMap[newRKProcessId] <= currColocMap[currRKProcessId] &&
	    newColocMap[newDDProcessId] <= currColocMap[currDDProcessId] &&
	    newColocMap[newBMProcessId] <= currColocMap[currBMProcessId]) {
		if (newColocMap[newRKProcessId] < currColocMap[currRKProcessId]) {
			rkSingleton.recruit(self);
		} else if (newColocMap[newDDProcessId] < currColocMap[currDDProcessId]) {
			ddSingleton.recruit(self);
		} else if (CLIENT_KNOBS->ENABLE_BLOB_GRANULES && newColocMap[newBMProcessId] < currColocMap[currBMProcessId]) {
			bmSingleton.recruit(self);
		}
	}
}
void checkBetterSingletons ( ClusterControllerData * self ) { if ( ! self -> masterProcessId . present ( ) || self -> db . serverInfo -> get ( ) . recoveryState < RecoveryState :: ACCEPTING_COMMITS ) { return ; } std :: map < Optional < Standalone < StringRef > > , int > id_used = self -> getUsedIds ( ) ; for ( auto & it : id_used ) { it . second *= PID_USED_AMP_FOR_NON_SINGLETON ; } WorkerDetails newRKWorker = findNewProcessForSingleton ( self , ProcessClass :: Ratekeeper , id_used ) ; WorkerDetails newDDWorker = findNewProcessForSingleton ( self , ProcessClass :: DataDistributor , id_used ) ; WorkerDetails newBMWorker ; if ( CLIENT_KNOBS -> ENABLE_BLOB_GRANULES ) { newBMWorker = findNewProcessForSingleton ( self , ProcessClass :: BlobManager , id_used ) ; } auto bestFitnessForRK = findBestFitnessForSingleton ( self , newRKWorker , ProcessClass :: Ratekeeper ) ; auto bestFitnessForDD = findBestFitnessForSingleton ( self , newDDWorker , ProcessClass :: DataDistributor ) ; ProcessClass :: Fitness bestFitnessForBM ; if ( CLIENT_KNOBS -> ENABLE_BLOB_GRANULES ) { bestFitnessForBM = findBestFitnessForSingleton ( self , newBMWorker , ProcessClass :: BlobManager ) ; } auto & db = self -> db . serverInfo -> get ( ) ; auto rkSingleton = RatekeeperSingleton ( db . ratekeeper ) ; auto ddSingleton = DataDistributorSingleton ( db . distributor ) ; BlobManagerSingleton bmSingleton ( db . blobManager ) ; bool rkHealthy = isHealthySingleton < RatekeeperInterface > ( self , newRKWorker , rkSingleton , bestFitnessForRK , self -> recruitingRatekeeperID ) ; bool ddHealthy = isHealthySingleton < DataDistributorInterface > ( self , newDDWorker , ddSingleton , bestFitnessForDD , self -> recruitingDistributorID ) ; bool bmHealthy = true ; if ( CLIENT_KNOBS -> ENABLE_BLOB_GRANULES ) { bmHealthy = isHealthySingleton < BlobManagerInterface > ( self , newBMWorker , bmSingleton , bestFitnessForBM , self -> recruitingBlobManagerID ) ; } if ( ! ( rkHealthy && ddHealthy && bmHealthy ) ) { return ; } Optional < Standalone < StringRef > > currRKProcessId = rkSingleton . interface . get ( ) . locality . processId ( ) ; Optional < Standalone < StringRef > > currDDProcessId = ddSingleton . interface . get ( ) . locality . processId ( ) ; Optional < Standalone < StringRef > > newRKProcessId = newRKWorker . interf . locality . processId ( ) ; Optional < Standalone < StringRef > > newDDProcessId = newDDWorker . interf . locality . processId ( ) ; Optional < Standalone < StringRef > > currBMProcessId , newBMProcessId ; if ( CLIENT_KNOBS -> ENABLE_BLOB_GRANULES ) { currBMProcessId = bmSingleton . interface . get ( ) . locality . processId ( ) ; newBMProcessId = newBMWorker . interf . locality . processId ( ) ; } std :: vector < Optional < Standalone < StringRef > > > currPids = { currRKProcessId , currDDProcessId } ; std :: vector < Optional < Standalone < StringRef > > > newPids = { newRKProcessId , newDDProcessId } ; if ( CLIENT_KNOBS -> ENABLE_BLOB_GRANULES ) { currPids . emplace_back ( currBMProcessId ) ; newPids . emplace_back ( newBMProcessId ) ; } auto currColocMap = getColocCounts ( currPids ) ; auto newColocMap = getColocCounts ( newPids ) ; if ( ! CLIENT_KNOBS -> ENABLE_BLOB_GRANULES ) { ASSERT ( currColocMap [ currBMProcessId ] == 0 ) ; ASSERT ( newColocMap [ newBMProcessId ] == 0 ) ; } if ( newColocMap [ newRKProcessId ] <= currColocMap [ currRKProcessId ] && newColocMap [ newDDProcessId ] <= currColocMap [ currDDProcessId ] && newColocMap [ newBMProcessId ] <= currColocMap [ currBMProcessId ] ) { if ( newColocMap [ newRKProcessId ] < currColocMap [ currRKProcessId ] ) { rkSingleton . recruit ( self ) ; } else if ( newColocMap [ newDDProcessId ] < currColocMap [ currDDProcessId ] ) { ddSingleton . recruit ( self ) ; } else if ( CLIENT_KNOBS -> ENABLE_BLOB_GRANULES && newColocMap [ newBMProcessId ] < currColocMap [ currBMProcessId ] ) { bmSingleton . recruit ( self ) ; } } }
Checks if there exists a better process for each singleton (e.g.
Checks if there exists a better process for each singleton ( e . g .
// note: this map doesn't consider pids used by existing singletons // 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. // Try to find a new process for each singleton. // Find best possible fitnesses for each singleton. // Check if the singletons are healthy. // side effect: try to rerecruit the singletons to more optimal processes // if any of the singletons are unhealthy (rerecruited or not stable), then do not // consider any further re-recruitments // 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 // if the knob is disabled, the BM coloc counts should have no affect on the coloc counts check below // if the new coloc counts are collectively better (i.e. each singleton's coloc count has not increased) // rerecruit the singleton for which we have found a better process, if any