Refactor getrange for read-write module and add a test to make sure we have consistent results
This commit is contained in:
parent
63a5f26a23
commit
43e3e320e3
|
@ -50,6 +50,9 @@ std::unordered_map<std::string, KeyRange> SpecialKeySpace::managementApiCommandT
|
|||
|
||||
std::set<std::string> SpecialKeySpace::options = { "excluded/force", "failed/force" };
|
||||
|
||||
Standalone<RangeResultRef> rywGetRange(ReadYourWritesTransaction* ryw, const KeyRangeRef& kr,
|
||||
const Standalone<RangeResultRef>& res);
|
||||
|
||||
// This function will move the given KeySelector as far as possible to the standard form:
|
||||
// orEqual == false && offset == 1 (Standard form)
|
||||
// If the corresponding key is not in the underlying key range, it will move over the range
|
||||
|
@ -458,6 +461,24 @@ Future<Void> SpecialKeySpace::commit(ReadYourWritesTransaction* ryw) {
|
|||
return commitActor(this, ryw);
|
||||
}
|
||||
|
||||
SKSCTestImpl::SKSCTestImpl(KeyRangeRef kr) : SpecialKeyRangeRWImpl(kr) {}
|
||||
|
||||
Future<Standalone<RangeResultRef>> SKSCTestImpl::getRange(ReadYourWritesTransaction* ryw, KeyRangeRef kr) const {
|
||||
ASSERT(range.contains(kr));
|
||||
auto resultFuture = ryw->getRange(kr, CLIENT_KNOBS->TOO_MANY);
|
||||
// all keys are written to RYW, since GRV is set, the read should happen locally
|
||||
ASSERT(resultFuture.isReady());
|
||||
auto result = resultFuture.getValue();
|
||||
ASSERT(!result.more && result.size() < CLIENT_KNOBS->TOO_MANY);
|
||||
auto kvs = resultFuture.getValue();
|
||||
return rywGetRange(ryw, kr, kvs);
|
||||
}
|
||||
|
||||
Future<Optional<std::string>> SKSCTestImpl::commit(ReadYourWritesTransaction* ryw) {
|
||||
ASSERT(false);
|
||||
return Optional<std::string>();
|
||||
}
|
||||
|
||||
ReadConflictRangeImpl::ReadConflictRangeImpl(KeyRangeRef kr) : SpecialKeyRangeReadImpl(kr) {}
|
||||
|
||||
ACTOR static Future<Standalone<RangeResultRef>> getReadConflictRangeImpl(ReadYourWritesTransaction* ryw, KeyRange kr) {
|
||||
|
|
|
@ -216,6 +216,14 @@ private:
|
|||
void modulesBoundaryInit();
|
||||
};
|
||||
|
||||
// Used for SpecialKeySpaceCorrectnessWorkload
|
||||
class SKSCTestImpl : public SpecialKeyRangeRWImpl {
|
||||
public:
|
||||
explicit SKSCTestImpl(KeyRangeRef kr);
|
||||
Future<Standalone<RangeResultRef>> getRange(ReadYourWritesTransaction* ryw, KeyRangeRef kr) const override;
|
||||
Future<Optional<std::string>> commit(ReadYourWritesTransaction* ryw) override;
|
||||
};
|
||||
|
||||
// Use special key prefix "\xff\xff/transaction/conflicting_keys/<some_key>",
|
||||
// to retrieve keys which caused latest not_committed(conflicting with another transaction) error.
|
||||
// The returned key value pairs are interpretted as :
|
||||
|
|
|
@ -27,27 +27,6 @@
|
|||
#include "fdbserver/workloads/workloads.actor.h"
|
||||
#include "flow/actorcompiler.h"
|
||||
|
||||
class SKSCTestImpl : public SpecialKeyRangeReadImpl {
|
||||
public:
|
||||
explicit SKSCTestImpl(KeyRangeRef kr) : SpecialKeyRangeReadImpl(kr) {}
|
||||
virtual Future<Standalone<RangeResultRef>> getRange(ReadYourWritesTransaction* ryw, KeyRangeRef kr) const {
|
||||
ASSERT(range.contains(kr));
|
||||
auto resultFuture = ryw->getRange(kr, CLIENT_KNOBS->TOO_MANY);
|
||||
// all keys are written to RYW, since GRV is set, the read should happen locally
|
||||
ASSERT(resultFuture.isReady());
|
||||
auto result = resultFuture.getValue();
|
||||
ASSERT(!result.more && result.size() < CLIENT_KNOBS->TOO_MANY);
|
||||
// To make the test more complext, instead of simply returning the k-v pairs, we reverse all the value strings
|
||||
auto kvs = resultFuture.getValue();
|
||||
for (int i = 0; i < kvs.size(); ++i) {
|
||||
std::string valStr(kvs[i].value.toString());
|
||||
std::reverse(valStr.begin(), valStr.end());
|
||||
kvs[i].value = ValueRef(kvs.arena(), valStr);
|
||||
}
|
||||
return kvs;
|
||||
}
|
||||
};
|
||||
|
||||
struct SpecialKeySpaceCorrectnessWorkload : TestWorkload {
|
||||
|
||||
int actorCount, minKeysPerRange, maxKeysPerRange, rangeCount, keyBytes, valBytes, conflictRangeSizeFactor;
|
||||
|
@ -86,6 +65,7 @@ struct SpecialKeySpaceCorrectnessWorkload : TestWorkload {
|
|||
cx->specialKeySpace = std::make_unique<SpecialKeySpace>();
|
||||
self->ryw = Reference(new ReadYourWritesTransaction(cx));
|
||||
self->ryw->setOption(FDBTransactionOptions::SPECIAL_KEY_SPACE_RELAXED);
|
||||
self->ryw->setOption(FDBTransactionOptions::SPECIAL_KEY_SPACE_ENABLE_WRITES);
|
||||
self->ryw->setVersion(100);
|
||||
self->ryw->clear(normalKeys);
|
||||
// generate key ranges
|
||||
|
@ -97,7 +77,7 @@ struct SpecialKeySpaceCorrectnessWorkload : TestWorkload {
|
|||
self->impls.push_back(std::make_shared<SKSCTestImpl>(KeyRangeRef(startKey, endKey)));
|
||||
// Although there are already ranges registered, the testing range will replace them
|
||||
cx->specialKeySpace->registerKeyRange(SpecialKeySpace::MODULE::TESTONLY,
|
||||
SpecialKeySpace::IMPLTYPE::READONLY, self->keys.back(),
|
||||
SpecialKeySpace::IMPLTYPE::READWRITE, self->keys.back(),
|
||||
self->impls.back().get());
|
||||
// generate keys in each key range
|
||||
int keysInRange = deterministicRandom()->randomInt(self->minKeysPerRange, self->maxKeysPerRange + 1);
|
||||
|
@ -157,6 +137,47 @@ struct SpecialKeySpaceCorrectnessWorkload : TestWorkload {
|
|||
.detail("Reverse", reverse);
|
||||
++self->wrongResults;
|
||||
}
|
||||
|
||||
// check ryw result consistency
|
||||
KeyRange rkr = self->randomKeyRange();
|
||||
KeyRef rkey1 = rkr.begin;
|
||||
KeyRef rkey2 = rkr.end;
|
||||
// randomly set/clear two keys or clear a key range
|
||||
if (deterministicRandom()->coinflip()) {
|
||||
Value rvalue1 = self->randomValue();
|
||||
cx->specialKeySpace->set(self->ryw.getPtr(), rkey1, rvalue1);
|
||||
self->ryw->set(rkey1, rvalue1);
|
||||
Value rvalue2 = self->randomValue();
|
||||
cx->specialKeySpace->set(self->ryw.getPtr(), rkey2, rvalue2);
|
||||
self->ryw->set(rkey2, rvalue2);
|
||||
} else if (deterministicRandom()->coinflip()) {
|
||||
cx->specialKeySpace->clear(self->ryw.getPtr(), rkey1);
|
||||
self->ryw->clear(rkey1);
|
||||
cx->specialKeySpace->clear(self->ryw.getPtr(), rkey2);
|
||||
self->ryw->clear(rkey2);
|
||||
} else {
|
||||
cx->specialKeySpace->clear(self->ryw.getPtr(), rkr);
|
||||
self->ryw->clear(rkr);
|
||||
}
|
||||
// use the same key selectors again to test consistency of ryw
|
||||
auto correctRywResultFuture = self->ryw->getRange(begin, end, limit, false, reverse);
|
||||
ASSERT(correctRywResultFuture.isReady());
|
||||
auto correctRywResult = correctRywResultFuture.getValue();
|
||||
auto testRywResultFuture = cx->specialKeySpace->getRange(self->ryw.getPtr(), begin, end, limit, reverse);
|
||||
ASSERT(testRywResultFuture.isReady());
|
||||
auto testRywResult = testRywResultFuture.getValue();
|
||||
|
||||
// check the consistency of results
|
||||
if (!self->compareRangeResult(correctRywResult, testRywResult)) {
|
||||
TraceEvent(SevError, "TestFailure")
|
||||
.detail("Reason", "Results from getRange(ryw) are inconsistent")
|
||||
.detail("Begin", begin.toString())
|
||||
.detail("End", end.toString())
|
||||
.detail("LimitRows", limit.rows)
|
||||
.detail("LimitBytes", limit.bytes)
|
||||
.detail("Reverse", reverse);
|
||||
++self->wrongResults;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -189,13 +210,9 @@ struct SpecialKeySpaceCorrectnessWorkload : TestWorkload {
|
|||
.detail("TestKey", printable(res2[i].key));
|
||||
return false;
|
||||
}
|
||||
// Value strings should be reversed pairs
|
||||
std::string valStr(res2[i].value.toString());
|
||||
std::reverse(valStr.begin(), valStr.end());
|
||||
Value valReversed(valStr);
|
||||
if (res1[i].value != valReversed) {
|
||||
if (res1[i].value != res2[i].value) {
|
||||
TraceEvent(SevError, "TestFailure")
|
||||
.detail("Reason", "Values are inconsistent, CorrectValue should be the reverse of the TestValue")
|
||||
.detail("Reason", "Values are inconsistent")
|
||||
.detail("Index", i)
|
||||
.detail("CorrectValue", printable(res1[i].value))
|
||||
.detail("TestValue", printable(res2[i].value));
|
||||
|
@ -206,7 +223,14 @@ struct SpecialKeySpaceCorrectnessWorkload : TestWorkload {
|
|||
return true;
|
||||
}
|
||||
|
||||
KeySelector randomKeySelector() {
|
||||
KeyRange randomKeyRange() {
|
||||
Key prefix = keys[deterministicRandom()->randomInt(0, rangeCount)].begin;
|
||||
Key rkey1 = Key(deterministicRandom()->randomAlphaNumeric(deterministicRandom()->randomInt(0, keyBytes))).withPrefix(prefix);
|
||||
Key rkey2 = Key(deterministicRandom()->randomAlphaNumeric(deterministicRandom()->randomInt(0, keyBytes))).withPrefix(prefix);
|
||||
return rkey1 <= rkey2 ? KeyRangeRef(rkey1, rkey2) : KeyRangeRef(rkey2, rkey1);
|
||||
}
|
||||
|
||||
Key randomKey() {
|
||||
Key randomKey;
|
||||
if (deterministicRandom()->random01() < absoluteRandomProb) {
|
||||
Key prefix;
|
||||
|
@ -224,9 +248,15 @@ struct SpecialKeySpaceCorrectnessWorkload : TestWorkload {
|
|||
KeyRangeRef randomKeyRangeRef = keys[deterministicRandom()->randomInt(0, keys.size())];
|
||||
randomKey = deterministicRandom()->coinflip() ? randomKeyRangeRef.begin : randomKeyRangeRef.end;
|
||||
}
|
||||
return randomKey;
|
||||
}
|
||||
|
||||
Value randomValue() { return Value(deterministicRandom()->randomAlphaNumeric(valBytes)); }
|
||||
|
||||
KeySelector randomKeySelector() {
|
||||
// covers corner cases where offset points outside the key space
|
||||
int offset = deterministicRandom()->randomInt(-keysCount.getValue() - 1, keysCount.getValue() + 2);
|
||||
return KeySelectorRef(randomKey, deterministicRandom()->coinflip(), offset);
|
||||
return KeySelectorRef(randomKey(), deterministicRandom()->coinflip(), offset);
|
||||
}
|
||||
|
||||
GetRangeLimits randomLimits() {
|
||||
|
@ -652,17 +682,17 @@ struct SpecialKeySpaceCorrectnessWorkload : TestWorkload {
|
|||
try {
|
||||
// test getRange
|
||||
state Standalone<RangeResultRef> class_source_result = wait(tx->getRange(
|
||||
KeyRangeRef(LiteralStringRef("process/class_source/"), LiteralStringRef("process/class_source0"))
|
||||
.withPrefix(SpecialKeySpace::getModuleRange(SpecialKeySpace::MODULE::CONFIGURATION).begin),
|
||||
CLIENT_KNOBS->TOO_MANY));
|
||||
KeyRangeRef(LiteralStringRef("process/class_source/"), LiteralStringRef("process/class_source0"))
|
||||
.withPrefix(SpecialKeySpace::getModuleRange(SpecialKeySpace::MODULE::CONFIGURATION).begin),
|
||||
CLIENT_KNOBS->TOO_MANY));
|
||||
ASSERT(!class_source_result.more && class_source_result.size() < CLIENT_KNOBS->TOO_MANY);
|
||||
ASSERT(self->getRangeResultInOrder(class_source_result));
|
||||
// check correctness of classType of each process
|
||||
vector<ProcessData> workers = wait(getWorkers(&tx->getTransaction()));
|
||||
for (const auto& worker : workers) {
|
||||
Key addr =
|
||||
Key("process/class_source/" + formatIpPort(worker.address.ip, worker.address.port))
|
||||
.withPrefix(SpecialKeySpace::getModuleRange(SpecialKeySpace::MODULE::CONFIGURATION).begin);
|
||||
Key("process/class_source/" + formatIpPort(worker.address.ip, worker.address.port))
|
||||
.withPrefix(SpecialKeySpace::getModuleRange(SpecialKeySpace::MODULE::CONFIGURATION).begin);
|
||||
bool found = false;
|
||||
for (const auto& kv : class_source_result) {
|
||||
if (kv.key == addr) {
|
||||
|
@ -680,11 +710,12 @@ struct SpecialKeySpaceCorrectnessWorkload : TestWorkload {
|
|||
state std::string address = formatIpPort(worker.address.ip, worker.address.port);
|
||||
tx->setOption(FDBTransactionOptions::SPECIAL_KEY_SPACE_ENABLE_WRITES);
|
||||
tx->set(Key("process/class_type/" + address)
|
||||
.withPrefix(SpecialKeySpace::getModuleRange(SpecialKeySpace::MODULE::CONFIGURATION).begin),
|
||||
LiteralStringRef("unset"));
|
||||
.withPrefix(SpecialKeySpace::getModuleRange(SpecialKeySpace::MODULE::CONFIGURATION).begin),
|
||||
LiteralStringRef("unset"));
|
||||
wait(tx->commit());
|
||||
Optional<Value> class_source = wait(tx->get(Key("process/class_source/" + address)
|
||||
.withPrefix(SpecialKeySpace::getModuleRange(SpecialKeySpace::MODULE::CONFIGURATION).begin)));
|
||||
Optional<Value> class_source = wait(tx->get(
|
||||
Key("process/class_source/" + address)
|
||||
.withPrefix(SpecialKeySpace::getModuleRange(SpecialKeySpace::MODULE::CONFIGURATION).begin)));
|
||||
ASSERT(class_source.present() && class_source.get() == LiteralStringRef("set_class"));
|
||||
tx->reset();
|
||||
} catch (Error& e) {
|
||||
|
|
Loading…
Reference in New Issue