wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

add test workload

This commit is contained in:
Nim Wijetunga 2021-02-09 12:32:37 -05:00
parent 83af53b1bb
commit 2560dc0227
5 changed files with 261 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
fdbclient/DatabaseContext.h
View File

@ -194,6 +194,7 @@ public:
Reference<WatchMetadata> getWatchMetadata(Key key) const;
void setWatchMetadata(Key key, Reference<WatchMetadata> metadata);
void deleteWatchMetadata(Key key);
void clearWatchMetadata();
void setOption( FDBDatabaseOptions::Option option, Optional<StringRef> value );
@ -384,7 +385,8 @@ public:
static bool debugUseTags;
static const std::vector<std::string> debugTransactionTagChoices;
std::map<Key, Reference<WatchMetadata>> watchMap;
private:
std::map<Key, Reference<WatchMetadata>> watchMap;
};
#endif

41
fdbclient/NativeAPI.actor.cpp
View File

@ -1387,6 +1387,10 @@ void DatabaseContext::deleteWatchMetadata(Key key) {
watchMap.erase(key);
}
void DatabaseContext::clearWatchMetadata() {
watchMap.clear();
}
WatchMetadata::WatchMetadata(Optional<Value> value, Version version, TransactionInfo info, TagSet tags): value(value), version(version), info(info), tags(tags) {
// create dummy future
watchFuture = watchPromise.getFuture();
@ -2228,31 +2232,35 @@ ACTOR Future<Void> watchStorageServerResp(Key key, Database cx) {
state Future<Version> watchFutureSS = watchValue(Future<Version>(metadata->version), key, metadata->value, cx, metadata->info, metadata->tags);
Version watchVersion = wait(watchFutureSS);
metadata = cx->getWatchMetadata(key);
if (!metadata.isValid()) return Void();
// TraceEvent("Nim_Case SS done wait").detail("version", watchVersion);
if (watchVersion >= metadata->version) { // case 1: version_1 (SS) >= version_2 (map)
// TraceEvent("Nim_Case SS 1").detail("Key", key).detail("Value", metadata->value).detail("numF", metadata->watchPromise.getFutureReferenceCount());
cx->deleteWatchMetadata(key);
metadata->watchPromise.send(watchVersion);
if(metadata->watchPromise.canBeSet()) metadata->watchPromise.send(watchVersion);
} else {
if (metadata->watchPromise.getFutureReferenceCount() == 1) { // case 2: version_1 < version_2 and future_count == 1
// TraceEvent("Nim_Case SS 2").detail("Key", key).detail("Value", metadata->value).detail("numF", metadata->watchPromise.getFutureReferenceCount());
cx->deleteWatchMetadata(key);
} else {
// TraceEvent("Nim_Case SS 3").detail("Key", key).detail("Value", metadata->value).detail("numF", metadata->watchPromise.getFutureReferenceCount());
}
}
} catch(Error &e) {
// TraceEvent("Nim_Case SS exit exception").detail("Code", e.code()).detail("key", key);
if (e.code() == error_code_actor_cancelled) {
if (e.code() == error_code_operation_cancelled) {
// TraceEvent("Nim_Case SS exit actor cancelled").detail("key", key);
throw e;
}
Reference<WatchMetadata> metadata = cx->getWatchMetadata(key);
if (!metadata.isValid() || metadata->watchPromise.getFutureReferenceCount() == 1) {
if (!metadata.isValid()) {
return Void();
} else if (metadata->watchPromise.getFutureReferenceCount() == 1) {
cx->deleteWatchMetadata(key);
return Void();
}
}
}
}
}
@ -2292,8 +2300,7 @@ ACTOR Future<Void> sameVersionDifKey(Version ver, Key key, Optional<Value> value
metadata = cx->getWatchMetadata(key);
if (metadata.isValid()) { // if val_3 == val_2
Future<Void> watchFuture = success(metadata->watchPromise.getFuture());
wait(watchFuture);
wait(success(metadata->watchPromise.getFuture()));
}
return Void();
@ -2318,9 +2325,7 @@ Future<Void> getWatchFuture(Version ver, Key key, Optional<Value> value, Databas
cx->setWatchMetadata(key, metadata);
metadata->watchFutureSS = watchStorageServerResp(key, cx);
Future<Void> watchFuture = success(metadata->watchPromise.getFuture());
return watchFuture;
return success(metadata->watchPromise.getFuture());
}
else if (metadata->value == value) { // case 2: val_1 == val_2
// TraceEvent("Nim_Case 2").detail("Key", key).detail("Value", value);
@ -2331,8 +2336,7 @@ Future<Void> getWatchFuture(Version ver, Key key, Optional<Value> value, Databas
metadata->tags = tags;
}
Future<Void> watchFuture = success(metadata->watchPromise.getFuture());
return watchFuture;
return success(metadata->watchPromise.getFuture());
} else if(ver > metadata->version) { // case 3: val_1 != val_2 && version_2 > version_1
// TraceEvent("Nim_Case 3").detail("Key", key).detail("Value", value);
@ -2346,15 +2350,14 @@ Future<Void> getWatchFuture(Version ver, Key key, Optional<Value> value, Databas
metadata->watchFutureSS = watchStorageServerResp(key, cx);
Future<Void> watchFuture = success(metadata->watchPromise.getFuture());
return watchFuture;
return success(metadata->watchPromise.getFuture());
} else if (metadata->version == ver) { // case 5: val_1 != val_2 && version_1 == version_2
// TraceEvent("Nim_Case 5").detail("Key", key).detail("Value", value);
return sameVersionDifKey(ver, key, value, cx, info, tags);
}
// TraceEvent("Nim_Case return").detail("Key", key).detail("Value", value);
// Note: We will reach here for case 3: val_1 != val_2 && version_2 < version_1
return Future<Void>();
// TraceEvent("Nim_Case 4").detail("Key", key).detail("Value", value);
// Note: We will reach here for case 4: val_1 != val_2 && version_2 < version_1
return Void();
}
ACTOR Future<Void> watchValueMap(Future<Version> version, Key key, Optional<Value> value, Database cx,
@ -3041,7 +3044,7 @@ ACTOR Future<Void> watch(Reference<Watch> watch, Database cx, TagSet tags, Trans
when(wait(cx->connectionFileChanged())) {
TEST(true); // Recreated a watch after switch
cx->watchMap.clear();
cx->clearWatchMetadata();
watch->watchFuture =
watchValueMap(cx->minAcceptableReadVersion, watch->key, watch->value, cx, info, tags);
}

1
fdbserver/CMakeLists.txt
View File

@ -225,6 +225,7 @@ set(FDBSERVER_SRCS
workloads/VersionStamp.actor.cpp
workloads/WatchAndWait.actor.cpp
workloads/Watches.actor.cpp
workloads/WatchesSameKeyCorrectness.actor.cpp
workloads/WorkerErrors.actor.cpp
workloads/workloads.actor.h
workloads/WriteBandwidth.actor.cpp

232
fdbserver/workloads/WatchesSameKeyCorrectness.actor.cpp Normal file
View File

@ -0,0 +1,232 @@
/*
* WatchesSameKeyCorrectness.actor.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2018 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include "fdbrpc/ContinuousSample.h"
#include "fdbclient/ReadYourWrites.h"
#include "fdbclient/NativeAPI.actor.h"
#include "fdbserver/TesterInterface.actor.h"
#include "flow/DeterministicRandom.h"
#include "fdbserver/workloads/workloads.actor.h"
#include "flow/actorcompiler.h" // This must be the last #include.
struct WatchesSameKeyWorkload : TestWorkload {
int numWatches;
std::vector<Future<Void>> cases;
WatchesSameKeyWorkload(WorkloadContext const& wcx)
: TestWorkload(wcx)
{
numWatches = getOption( options, LiteralStringRef("numWatches"), 3 );
}
std::string description() const override { return "WatchesSameKeyCorrectness"; }
Future<Void> setup(Database const& cx) override {
if ( clientId == 0 ) {
cases.push_back( case1(cx, StringRef("foo1"), this) );
cases.push_back( case2(cx, StringRef("foo2"), this) );
cases.push_back( case3(cx, StringRef("foo3"), this) );
cases.push_back( case4(cx, StringRef("foo4"), this) );
cases.push_back( case5(cx, StringRef("foo5"), this) );
}
return Void();
}
Future<Void> start(Database const& cx) override {
if (clientId == 0) return waitForAll( cases );
return Void();
}
Future<bool> check(Database const& cx) override {
if ( clientId != 0 ) return true;
bool ok = true;
for( int i = 0; i < cases.size(); i++ ) {
if ( cases[i].isError() ) ok = false;
}
cases.clear();
return ok;
}
ACTOR static Future<Void> setKeyRandomValue(Database cx, Key key, Optional<Value> val) {
try {
if ( !val.present() ) val = Value( deterministicRandom()->randomUniqueID().toString() );
state ReadYourWritesTransaction tr(cx);
tr.set(key, val.get());
wait( tr.commit() );
return Void();
} catch ( Error& e ) {
throw e;
}
}
ACTOR static Future<Optional<Value>> getValue(Database cx, Key key) {
try {
state ReadYourWritesTransaction tr(cx);
Optional<Value> val = wait(tr.get(key));
return val;
} catch ( Error& e ) {
throw e;
}
}
ACTOR static Future<Future<Void>> watchKey(Database cx, Key key) {
try {
state ReadYourWritesTransaction tr(cx);
state Future<Void> watchFuture = tr.watch(key);
wait( tr.commit() );
return watchFuture;
} catch ( Error& e ) {
throw e;
}
}
ACTOR static Future<Void> case1(Database cx, Key key, WatchesSameKeyWorkload* self) {
/**
* Tests case 2 in the design doc:
* - we get a watch that has the same value as a key in the watch map
* */
loop {
try {
state std::vector<Future<Void>> watchFutures;
state int i;
for ( i = 0; i < self->numWatches; i++ ) {
Future<Void> watchFuture = wait(watchKey(cx, key));
watchFutures.push_back(watchFuture);
}
wait( setKeyRandomValue(cx, key, Optional<Value>()) );
for ( i = 0; i < watchFutures.size(); i++) {
wait( watchFutures[i] );
}
return Void();
} catch( Error& e ) {}
}
}
ACTOR static Future<Void> case2(Database cx, Key key, WatchesSameKeyWorkload* self) {
/**
* Tests case 3 in the design doc:
* - we get a watch that has a different value than the key in the map but the version is larger
* */
loop {
try {
state std::vector<Future<Void>> watchFutures;
state Future<Void> watch1 = wait(watchKey(cx, key));
state int i;
state Value val = Value( deterministicRandom()->randomUniqueID().toString() );
state ReadYourWritesTransaction tr(cx);
tr.set(key, val);
for ( i = 0; i < self->numWatches; i++ ) {
watchFutures.push_back(tr.watch(key));
}
wait ( tr.commit() );
wait( watch1 );
wait( setKeyRandomValue(cx, key, Optional<Value>()) );
for ( i = 0; i < watchFutures.size(); i++) {
wait( watchFutures[i] );
}
return Void();
} catch ( Error& e ) {}
}
}
ACTOR static Future<Void> case3(Database cx, Key key, WatchesSameKeyWorkload* self) {
/**
* Tests case 2 for the storage server response:
* - i.e ABA but when the storage server responds the future count == 1 so we do nothing (no refire)
* */
loop {
try {
wait ( setKeyRandomValue(cx, key, Optional<Value>()) );
state Optional<Value> val = wait( getValue(cx, key) );
state Future<Void> watch1 = wait(watchKey(cx, key));
wait ( setKeyRandomValue(cx, key, Optional<Value>()) );
state ReadYourWritesTransaction tr(cx);
tr.set(key, val.get());
state Future<Void> watch2 = tr.watch(key);
wait( tr.commit() );
watch1.cancel();
watch2.cancel();
return Void();
} catch ( Error& e ) {}
}
}
ACTOR static Future<Void> case4(Database cx, Key key, WatchesSameKeyWorkload* self) {
/**
* Tests case 3 for the storage server response:
* - i.e ABA but when the storage server responds the future count > 1 so we refire request to SS
* */
loop {
try {
wait ( setKeyRandomValue(cx, key, Optional<Value>()) );
state Optional<Value> val = wait( getValue(cx, key) );
state Future<Void> watch1 = wait(watchKey(cx, key));
wait ( setKeyRandomValue(cx, key, Optional<Value>()) );
state ReadYourWritesTransaction tr(cx);
tr.set(key, val.get());
state Future<Void> watch2 = tr.watch(key);
wait( tr.commit() );
wait( setKeyRandomValue(cx, key, Optional<Value>()) );
wait( watch1 );
wait( watch2 );
return Void();
} catch ( Error& e ) {}
}
}
ACTOR static Future<Void> case5(Database cx, Key key, WatchesSameKeyWorkload* self) {
/**
* Tests case 5 in the design doc:
* - i.e values of watches are different but versions are the same
* */
loop {
try {
state Value val1 = Value( deterministicRandom()->randomUniqueID().toString() );
state Value val2 = Value( deterministicRandom()->randomUniqueID().toString() );
state ReadYourWritesTransaction tr1(cx);
tr1.setOption( FDBTransactionOptions::NEXT_WRITE_NO_WRITE_CONFLICT_RANGE );
state ReadYourWritesTransaction tr2(cx);
tr2.setOption( FDBTransactionOptions::NEXT_WRITE_NO_WRITE_CONFLICT_RANGE );
tr1.set(key, val1);
tr2.set(key, val2);
state Future<Void> watch1 = tr1.watch(key);
state Future<Void> watch2 = tr2.watch(key);
wait( tr1.commit() && tr2.commit() );
wait( watch1 || watch2 ); // since we enter case 5 at least one of the watches should be fired
wait( setKeyRandomValue(cx, key, Optional<Value>()) );
wait( watch1 && watch2 );
return Void();
} catch ( Error& e ) {}
}
}
void getMetrics(vector<PerfMetric>& m) override {}
};
WorkloadFactory<WatchesSameKeyWorkload> WatchesSameKeyWorkloadFactory("WatchesSameKeyCorrectness");

3
tests/fast/Watches.toml
View File

@ -3,3 +3,6 @@ testTitle = 'WatchesTest'
[[test.workload]]
testName = 'Watches'
[[test.workload]]
testName = 'WatchesSameKeyCorrectness'