apple/foundationdb
1
0
Fork 0
mirror of https://github.com/apple/foundationdb.git synced 2025-06-02 11:15:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

Factor out known replica update step

Browse Source
This commit is contained in:
Lukas Joswiak 2022-02-08 16:05:10 -08:00
parent 1d496b7b30
commit b5a3312a26
1 changed files with 18 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
fdbserver/ConfigBroadcaster.actor.cpp
View File

@ -321,6 +321,22 @@ public:
return registerNode(&self, this, w, lastSeenVersion, configClassSet, watcher, broadcastInterface); return registerNode(&self, this, w, lastSeenVersion, configClassSet, watcher, broadcastInterface);
} }
// Updates the broadcasters knowledge of which replicas are fully up to
// date, based on data gathered by the consumer.
void updateKnownReplicas(std::vector<ConfigFollowerInterface> const& readReplicas) {
if (!newConfigNodesAllowed.canBeSet()) {
return;
}
for (const auto& cfi : readReplicas) {
this->activeConfigNodes.insert(cfi.address());
}
if (activeConfigNodes.size() >= coordinators / 2 + 1) {
disallowUnregistered = true;
newConfigNodesAllowed.send(Void());
}
}
void applyChanges(Standalone<VectorRef<VersionedConfigMutationRef>> const& changes, void applyChanges(Standalone<VectorRef<VersionedConfigMutationRef>> const& changes,
Version mostRecentVersion, Version mostRecentVersion,
Standalone<VectorRef<VersionedConfigCommitAnnotationRef>> const& annotations, Standalone<VectorRef<VersionedConfigCommitAnnotationRef>> const& annotations,
@ -334,15 +350,7 @@ public:
addChanges(changes, mostRecentVersion, annotations); addChanges(changes, mostRecentVersion, annotations);
} }
if (newConfigNodesAllowed.canBeSet()) { updateKnownReplicas(readReplicas);
for (const auto& cfi : readReplicas) {
this->activeConfigNodes.insert(cfi.address());
}
if (activeConfigNodes.size() >= coordinators / 2 + 1) {
disallowUnregistered = true;
newConfigNodesAllowed.send(Void());
}
}
} }
template <class Snapshot> template <class Snapshot>
@ -361,13 +369,7 @@ public:
.detail("ActiveReplicas", readReplicas.size()); .detail("ActiveReplicas", readReplicas.size());
actors.add(pushSnapshotAndChanges(this, snapshot, snapshotVersion, changes, changesVersion, annotations)); actors.add(pushSnapshotAndChanges(this, snapshot, snapshotVersion, changes, changesVersion, annotations));
for (const auto& cfi : readReplicas) { updateKnownReplicas(readReplicas);
this->activeConfigNodes.insert(cfi.address());
}
if (activeConfigNodes.size() >= coordinators / 2 + 1 && newConfigNodesAllowed.canBeSet()) {
disallowUnregistered = true;
newConfigNodesAllowed.send(Void());
}
} }
ConfigBroadcasterImpl(ConfigFollowerInterface const& cfi) : ConfigBroadcasterImpl() { ConfigBroadcasterImpl(ConfigFollowerInterface const& cfi) : ConfigBroadcasterImpl() {