apple/foundationdb
1
0
Fork 0
mirror of https://github.com/apple/foundationdb.git synced 2025-05-14 18:02:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add broadcaster error check to unit tests

Browse Source
This commit is contained in:
Lukas Joswiak 2021-08-10 11:21:57 -07:00
parent 564a3d69b7
commit 72e55ef72e
3 changed files with 11 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
fdbserver/ConfigBroadcaster.actor.cpp
View File

@ -70,6 +70,7 @@ class ConfigBroadcasterImpl {
Version lastCompactedVersion; Version lastCompactedVersion;
Version mostRecentVersion; Version mostRecentVersion;
std::unique_ptr<IConfigConsumer> consumer; std::unique_ptr<IConfigConsumer> consumer;
Future<Void> consumerFuture;
ActorCollection actors{ false }; ActorCollection actors{ false };
std::vector<BroadcastClientDetails> clients; std::vector<BroadcastClientDetails> clients;
@ -83,10 +84,6 @@ class ConfigBroadcasterImpl {
template <class Changes> template <class Changes>
Future<Void> pushChanges(BroadcastClientDetails& client, Changes const& changes) { Future<Void> pushChanges(BroadcastClientDetails& client, Changes const& changes) {
if (client.watcher.isReady()) {
clients.erase(std::remove(clients.begin(), clients.end(), client));
}
// Skip if client has already seen the latest version. // Skip if client has already seen the latest version.
if (client.lastSeenVersion >= mostRecentVersion) { if (client.lastSeenVersion >= mostRecentVersion) {
return Void(); return Void();
@ -172,7 +169,7 @@ public:
ConfigClassSet configClassSet, ConfigClassSet configClassSet,
Future<Void> watcher, Future<Void> watcher,
ConfigBroadcastInterface broadcastInterface) { ConfigBroadcastInterface broadcastInterface) {
actors.add(consumer->consume(*self)); consumerFuture = consumer->consume(*self);
clients.push_back(BroadcastClientDetails{ clients.push_back(BroadcastClientDetails{
watcher, std::move(configClassSet), lastSeenVersion, std::move(broadcastInterface) }); watcher, std::move(configClassSet), lastSeenVersion, std::move(broadcastInterface) });
this->actors.add(waitForFailure(this, watcher, &clients.back())); this->actors.add(waitForFailure(this, watcher, &clients.back()));
@ -297,6 +294,8 @@ public:
} }
} }
Future<Void> getError() const { return consumerFuture; }
UID getID() const { return id; } UID getID() const { return id; }
static void runPendingRequestStoreTest(bool includeGlobalMutation, int expectedMatches); static void runPendingRequestStoreTest(bool includeGlobalMutation, int expectedMatches);
@ -345,6 +344,10 @@ void ConfigBroadcaster::applySnapshotAndChanges(
impl().applySnapshotAndChanges(std::move(snapshot), snapshotVersion, changes, changesVersion, annotations); impl().applySnapshotAndChanges(std::move(snapshot), snapshotVersion, changes, changesVersion, annotations);
} }
Future<Void> ConfigBroadcaster::getError() const {
return impl().getError();
}
UID ConfigBroadcaster::getID() const { UID ConfigBroadcaster::getID() const {
return impl().getID(); return impl().getID();
} }

1
fdbserver/ConfigBroadcaster.h
View File

@ -61,6 +61,7 @@ public:
Standalone<VectorRef<VersionedConfigMutationRef>> const& changes, Standalone<VectorRef<VersionedConfigMutationRef>> const& changes,
Version changesVersion, Version changesVersion,
Standalone<VectorRef<VersionedConfigCommitAnnotationRef>> const& annotations); Standalone<VectorRef<VersionedConfigCommitAnnotationRef>> const& annotations);
Future<Void> getError() const;
UID getID() const; UID getID() const;
JsonBuilderObject getStatus() const; JsonBuilderObject getStatus() const;
void compact(Version compactionVersion); void compact(Version compactionVersion);

4
fdbserver/ConfigDatabaseUnitTests.actor.cpp
View File

@ -279,7 +279,7 @@ public:
void compact() { broadcaster.compact(lastWrittenVersion); } void compact() { broadcaster.compact(lastWrittenVersion); }
Future<Void> getError() const { return readFrom.getError(); } Future<Void> getError() const { return readFrom.getError() || broadcaster.getError(); }
}; };
class TransactionEnvironment { class TransactionEnvironment {
@ -411,7 +411,7 @@ public:
} }
Future<Void> clear(Optional<KeyRef> configClass) { return writeTo.clear(configClass); } Future<Void> clear(Optional<KeyRef> configClass) { return writeTo.clear(configClass); }
Future<Void> check(Optional<int64_t> value) const { return readFrom.checkEventually(value); } Future<Void> check(Optional<int64_t> value) const { return readFrom.checkEventually(value); }
Future<Void> getError() const { return writeTo.getError() || readFrom.getError(); } Future<Void> getError() const { return writeTo.getError() || readFrom.getError() || broadcaster.getError(); }
}; };
// These functions give a common interface to all environments, to improve code reuse // These functions give a common interface to all environments, to improve code reuse