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

457 lines
17 KiB
C++
Raw Permalink Blame History

/*
* TaskBucketCorrectness.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2026 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 "flow/UnitTest.h"
#include "flow/Error.h"
#include "fdbclient/Tuple.h"
#include "fdbclient/TaskBucket.h"
#include "fdbclient/ReadYourWrites.h"
#include "fdbserver/tester/workloads.h"
struct SayHelloTaskFunc : TaskFuncBase {
static StringRef name;
static constexpr uint32_t version = 1;
StringRef getName() const override { return name; };
Future<Void> execute(Database cx,
Reference<TaskBucket> tb,
Reference<FutureBucket> fb,
Reference<Task> task) override {
return Void();
};
Future<Void> finish(Reference<ReadYourWritesTransaction> tr,
Reference<TaskBucket> tb,
Reference<FutureBucket> fb,
Reference<Task> task) override {
return _finish(tr, tb, fb, task);
};
static Future<Void> _finish(Reference<ReadYourWritesTransaction> tr,
Reference<TaskBucket> taskBucket,
Reference<FutureBucket> futureBucket,
Reference<Task> task) {
// check task version
uint32_t taskVersion = task->getVersion();
if (taskVersion > SayHelloTaskFunc::version) {
uint32_t v = SayHelloTaskFunc::version;
TraceEvent("TaskBucketCorrectnessSayHello")
.detail("CheckTaskVersion", "taskVersion is larger than the funcVersion")
.detail("TaskVersion", taskVersion)
.detail("FuncVersion", v);
}
Reference<TaskFuture> done = futureBucket->unpack(task->params[Task::reservedTaskParamKeyDone]);
co_await taskBucket->finish(tr, task);
if (buggify())
co_await delay(10);
Key key = StringRef("Hello_" + deterministicRandom()->randomUniqueID().toString());
Key value;
auto itor = task->params.find("name"_sr);
if (itor != task->params.end()) {
value = itor->value;
TraceEvent("TaskBucketCorrectnessSayHello").detail("SayHelloTaskFunc", printable(itor->value));
} else {
ASSERT(false);
}
if (!task->params["chained"_sr].compare("false"_sr)) {
co_await done->set(tr, taskBucket);
} else {
int subtaskCount = atoi(task->params["subtaskCount"_sr].toString().c_str());
int currTaskNumber = atoi(value.removePrefix("task_"_sr).toString().c_str());
TraceEvent("TaskBucketCorrectnessSayHello")
.detail("SubtaskCount", subtaskCount)
.detail("CurrTaskNumber", currTaskNumber);
if (currTaskNumber < subtaskCount - 1) {
std::vector<Reference<TaskFuture>> vectorFuture;
auto new_task = makeReference<Task>(SayHelloTaskFunc::name,
SayHelloTaskFunc::version,
StringRef(),
deterministicRandom()->randomInt(0, 2));
new_task->params["name"_sr] = StringRef(format("task_%d", currTaskNumber + 1));
new_task->params["chained"_sr] = task->params["chained"_sr];
new_task->params["subtaskCount"_sr] = task->params["subtaskCount"_sr];
Reference<TaskFuture> taskDone = futureBucket->future(tr);
new_task->params[Task::reservedTaskParamKeyDone] = taskDone->key;
taskBucket->addTask(tr, new_task);
vectorFuture.push_back(taskDone);
co_await done->join(tr, taskBucket, vectorFuture);
} else {
co_await done->set(tr, taskBucket);
}
}
tr->set(key, value);
}
};
StringRef SayHelloTaskFunc::name = "SayHello"_sr;
REGISTER_TASKFUNC(SayHelloTaskFunc);
struct SayHelloToEveryoneTaskFunc : TaskFuncBase {
static StringRef name;
static constexpr uint32_t version = 1;
StringRef getName() const override { return name; };
Future<Void> execute(Database cx,
Reference<TaskBucket> tb,
Reference<FutureBucket> fb,
Reference<Task> task) override {
return Void();
};
Future<Void> finish(Reference<ReadYourWritesTransaction> tr,
Reference<TaskBucket> tb,
Reference<FutureBucket> fb,
Reference<Task> task) override {
return _finish(tr, tb, fb, task);
};
static Future<Void> _finish(Reference<ReadYourWritesTransaction> tr,
Reference<TaskBucket> taskBucket,
Reference<FutureBucket> futureBucket,
Reference<Task> task) {
Reference<TaskFuture> done = futureBucket->unpack(task->params[Task::reservedTaskParamKeyDone]);
std::vector<Reference<TaskFuture>> vectorFuture;
int subtaskCount = 1;
if (!task->params["chained"_sr].compare("false"_sr)) {
subtaskCount = atoi(task->params["subtaskCount"_sr].toString().c_str());
}
for (int i = 0; i < subtaskCount; ++i) {
auto new_task = makeReference<Task>(
SayHelloTaskFunc::name, SayHelloTaskFunc::version, StringRef(), deterministicRandom()->randomInt(0, 2));
new_task->params["name"_sr] = StringRef(format("task_%d", i));
new_task->params["chained"_sr] = task->params["chained"_sr];
new_task->params["subtaskCount"_sr] = task->params["subtaskCount"_sr];
Reference<TaskFuture> taskDone = futureBucket->future(tr);
new_task->params[Task::reservedTaskParamKeyDone] = taskDone->key;
taskBucket->addTask(tr, new_task);
vectorFuture.push_back(taskDone);
}
co_await done->join(tr, taskBucket, vectorFuture);
co_await taskBucket->finish(tr, task);
Key key = StringRef("Hello_" + deterministicRandom()->randomUniqueID().toString());
Value value = "Hello, Everyone!"_sr;
TraceEvent("TaskBucketCorrectnessSayHello").detail("SayHelloToEveryoneTaskFunc", printable(value));
tr->set(key, value);
}
};
StringRef SayHelloToEveryoneTaskFunc::name = "SayHelloToEveryone"_sr;
REGISTER_TASKFUNC(SayHelloToEveryoneTaskFunc);
struct SaidHelloTaskFunc : TaskFuncBase {
static StringRef name;
static constexpr uint32_t version = 1;
StringRef getName() const override { return name; };
Future<Void> execute(Database cx,
Reference<TaskBucket> tb,
Reference<FutureBucket> fb,
Reference<Task> task) override {
return Void();
};
Future<Void> finish(Reference<ReadYourWritesTransaction> tr,
Reference<TaskBucket> tb,
Reference<FutureBucket> fb,
Reference<Task> task) override {
return _finish(tr, tb, fb, task);
};
static Future<Void> _finish(Reference<ReadYourWritesTransaction> tr,
Reference<TaskBucket> taskBucket,
Reference<FutureBucket> futureBucket,
Reference<Task> task) {
co_await taskBucket->finish(tr, task);
Key key = StringRef("Hello_" + deterministicRandom()->randomUniqueID().toString());
Value value = "Said hello to everyone!"_sr;
TraceEvent("TaskBucketCorrectnessSayHello").detail("SaidHelloTaskFunc", printable(value));
tr->set(key, value);
}
};
StringRef SaidHelloTaskFunc::name = "SaidHello"_sr;
REGISTER_TASKFUNC(SaidHelloTaskFunc);
struct CancelledTaskFunc : TaskFuncBase {
static StringRef name;
static constexpr uint32_t version = 1;
StringRef getName() const override { return name; }
Future<Void> execute(Database cx,
Reference<TaskBucket> tb,
Reference<FutureBucket> fb,
Reference<Task> task) override {
return actor_cancelled();
}
Future<Void> finish(Reference<ReadYourWritesTransaction> tr,
Reference<TaskBucket> tb,
Reference<FutureBucket> fb,
Reference<Task> task) override {
return Void();
}
};
StringRef CancelledTaskFunc::name = "TaskBucketCorrectnessCancelled"_sr;
REGISTER_TASKFUNC(CancelledTaskFunc);
// A workload which test the correctness of TaskBucket
struct TaskBucketCorrectnessWorkload : TestWorkload {
static constexpr auto NAME = "TaskBucketCorrectness";
bool chained;
int subtaskCount;
explicit TaskBucketCorrectnessWorkload(WorkloadContext const& wcx) : TestWorkload(wcx) {
chained = getOption(options, "chained"_sr, false);
subtaskCount = getOption(options, "subtaskCount"_sr, 20);
}
void getMetrics(std::vector<PerfMetric>& m) override {}
Future<Void> addInitTasks(Reference<ReadYourWritesTransaction> tr,
Reference<TaskBucket> taskBucket,
Reference<FutureBucket> futureBucket,
bool chained,
int subtaskCount) {
Key addedInitKey("addedInitTasks"_sr);
Optional<Standalone<StringRef>> res = co_await tr->get(addedInitKey);
if (res.present())
co_return;
tr->set(addedInitKey, "true"_sr);
Reference<TaskFuture> allDone = futureBucket->future(tr);
auto task = makeReference<Task>(SayHelloToEveryoneTaskFunc::name,
SayHelloToEveryoneTaskFunc::version,
allDone->key,
deterministicRandom()->randomInt(0, 2));
task->params["chained"_sr] = chained ? "true"_sr : "false"_sr;
task->params["subtaskCount"_sr] = StringRef(format("%d", subtaskCount));
taskBucket->addTask(tr, task);
auto taskDone = makeReference<Task>(
SaidHelloTaskFunc::name, SaidHelloTaskFunc::version, StringRef(), deterministicRandom()->randomInt(0, 2));
co_await allDone->onSetAddTask(tr, taskBucket, taskDone);
}
Future<Void> start(Database const& cx) override {
Reference<ReadYourWritesTransaction> tr(new ReadYourWritesTransaction(cx));
Subspace taskSubspace("backup-agent"_sr);
Reference<TaskBucket> taskBucket(new TaskBucket(taskSubspace.get("tasks"_sr)));
Reference<FutureBucket> futureBucket(new FutureBucket(taskSubspace.get("futures"_sr)));
Error err;
try {
if (clientId == 0) {
TraceEvent("TaskBucketCorrectness").detail("ClearingDb", "...");
co_await taskBucket->clear(cx);
auto cancelledTask =
makeReference<Task>(CancelledTaskFunc::name, CancelledTaskFunc::version, StringRef(), 0);
Future<bool> cancelledResult = taskBucket->doTask(cx, futureBucket, cancelledTask);
ASSERT(cancelledResult.isReady() && cancelledResult.isError());
ASSERT_EQ(cancelledResult.getError().code(), error_code_actor_cancelled);
TraceEvent("TaskBucketCorrectness").detail("AddingTasks", "...");
co_await runRYWTransaction(cx, [=](Reference<ReadYourWritesTransaction> tr) {
return addInitTasks(tr, taskBucket, futureBucket, chained, subtaskCount);
});
TraceEvent("TaskBucketCorrectness").detail("RunningTasks", "...");
}
while (true) {
{
Error err;
try {
bool oneTaskDone = co_await taskBucket->doOne(cx, futureBucket);
if (!oneTaskDone) {
bool isEmpty = co_await taskBucket->isEmpty(cx);
if (isEmpty) {
co_await delay(5.0);
bool isFutureEmpty = co_await futureBucket->isEmpty(cx);
if (isFutureEmpty)
break;
else {
co_await TaskBucket::debugPrintRange(
cx, taskSubspace.key(), StringRef(format("client_%d", clientId)));
TraceEvent("TaskBucketCorrectness").detail("FutureIsNotEmpty", "...");
}
} else {
co_await delay(1.0);
}
}
} catch (Error& e) {
err = e;
}
if (err.isValid()) {
if (err.code() == error_code_timed_out)
TraceEvent(SevWarn, "TaskBucketCorrectness").error(err);
else
co_await tr->onError(err);
}
}
}
if (clientId == 0) {
TraceEvent("TaskBucketCorrectness").detail("NotTasksRemain", "...");
co_await TaskBucket::debugPrintRange(cx, StringRef(), StringRef());
}
} catch (Error& e) {
err = e;
}
if (err.isValid()) {
TraceEvent(SevError, "TaskBucketCorrectness").error(err);
co_await tr->onError(err);
}
}
Future<bool> check(Database const& cx) override {
bool ret = co_await runRYWTransaction(
cx, [=](Reference<ReadYourWritesTransaction> tr) { return checkSayHello(tr, subtaskCount); });
co_return ret;
}
Future<bool> checkSayHello(Reference<ReadYourWritesTransaction> tr, int subTaskCount) {
std::set<std::string> data = { "Hello, Everyone!", "Said hello to everyone!" };
for (int i = 0; i < subTaskCount; i++) {
data.insert(format("task_%d", i));
}
RangeResult values =
co_await tr->getRange(KeyRangeRef("Hello_\x00"_sr, "Hello_\xff"_sr), CLIENT_KNOBS->TOO_MANY);
if (values.size() != data.size()) {
TraceEvent(SevError, "CheckSayHello")
.detail("CountNotMatchIs", values.size())
.detail("ShouldBe", data.size());
for (auto& s : values) {
TraceEvent("CheckSayHello").detail("Item", printable(s)).detail("Value", printable(s.value));
}
co_return false;
}
for (auto& s : values) {
// TraceEvent("CheckSayHello").detail("Item", printable(s)).detail("Value", printable(s.value));
data.erase(s.value.toString());
}
if (!data.empty()) {
TraceEvent(SevError, "CheckSayHello").detail("DataNotMatch", data.size());
co_return false;
}
co_return true;
}
};
WorkloadFactory<TaskBucketCorrectnessWorkload> TaskBucketCorrectnessWorkloadFactory;
void print_subspace_key(const Subspace& subspace, int id) {
printf("%d==========%s===%d\n", id, printable(StringRef(subspace.key())).c_str(), subspace.key().size());
}
TEST_CASE("/fdbclient/TaskBucket/Subspace") {
Subspace subspace_test;
print_subspace_key(subspace_test, 0);
ASSERT(subspace_test.key().toString().empty());
Subspace subspace_test1("abc"_sr);
print_subspace_key(subspace_test1, 1);
ASSERT(subspace_test1.key() == "abc"_sr);
Tuple t = Tuple::makeTuple("user"_sr);
Subspace subspace_test2(t);
print_subspace_key(subspace_test2, 2);
ASSERT(subspace_test2.key() == "\x01user\x00"_sr);
Subspace subspace_test3(t, "abc"_sr);
print_subspace_key(subspace_test3, 3);
ASSERT(subspace_test3.key() == "abc\x01user\x00"_sr);
Tuple t1 = Tuple::makeTuple(1);
Subspace subspace_test4(t1);
print_subspace_key(subspace_test4, 4);
ASSERT(subspace_test4.key() == "\x15\x01"_sr);
t.append(123);
Subspace subspace_test5(t, "abc"_sr);
print_subspace_key(subspace_test5, 5);
ASSERT(subspace_test5.key() == "abc\x01user\x00\x15\x7b"_sr);
// Subspace pack
printf("%d==========%s===%d\n", 6, printable(subspace_test5.pack(t)).c_str(), subspace_test5.pack(t).size());
ASSERT(subspace_test5.pack(t) == "abc\x01user\x00\x15\x7b\x01user\x00\x15\x7b"_sr);
printf("%d==========%s===%d\n", 7, printable(subspace_test5.pack(t1)).c_str(), subspace_test5.pack(t1).size());
ASSERT(subspace_test5.pack(t1) == "abc\x01user\x00\x15\x7b\x15\x01"_sr);
// Subspace getItem
Subspace subspace_test6(t);
Subspace subspace_test7 = subspace_test6.get("subitem"_sr);
print_subspace_key(subspace_test7, 8);
ASSERT(subspace_test7.key() == "\x01user\x00\x15\x7b\x01subitem\x00"_sr);
// Subspace unpack
Tuple t2 = subspace_test6.unpack(subspace_test7.key());
Subspace subspace_test8(t2);
print_subspace_key(subspace_test8, 9);
ASSERT(subspace_test8.key() == "\x01subitem\x00"_sr);
// pack
Tuple t3 = Tuple::makeTuple(""_sr);
printf("%d==========%s===%d\n", 10, printable(subspace_test5.pack(t3)).c_str(), subspace_test5.pack(t3).size());
ASSERT(subspace_test5.pack(t3) == subspace_test5.pack(StringRef()));
ASSERT(subspace_test5.pack(t3) == "abc\x01user\x00\x15\x7b\x01\x00"_sr);
printf("%d==========%s===%d\n",
11,
printable(subspace_test5.range(t3).begin).c_str(),
subspace_test5.range(t3).begin.size());
ASSERT(subspace_test5.range(t3).begin == subspace_test5.get(StringRef()).range().begin);
printf("%d==========%s===%d\n",
12,
printable(subspace_test5.range(t3).end).c_str(),
subspace_test5.range(t3).end.size());
ASSERT(subspace_test5.range(t3).end == subspace_test5.get(StringRef()).range().end);
StringRef def = "def"_sr;
StringRef ghi = "ghi"_sr;
t3.append(def);
t3.append(ghi);
printf("%d==========%s===%d\n", 13, printable(subspace_test5.pack(t3)).c_str(), subspace_test5.pack(t3).size());
ASSERT(subspace_test5.pack(t3) == subspace_test5.get(StringRef()).get(def).pack(ghi));
ASSERT(subspace_test5.pack(t3) == "abc\x01user\x00\x15\x7b\x01\x00\x01"
"def\x00\x01ghi\x00"_sr);
printf("%d==========%s===%d\n",
14,
printable(subspace_test5.range(t3).begin).c_str(),
subspace_test5.range(t3).begin.size());
ASSERT(subspace_test5.range(t3).begin == subspace_test5.get(StringRef()).get(def).get(ghi).range().begin);
printf("%d==========%s===%d\n",
15,
printable(subspace_test5.range(t3).end).c_str(),
subspace_test5.range(t3).end.size());
ASSERT(subspace_test5.range(t3).end == subspace_test5.get(StringRef()).get(def).get(ghi).range().end);
return Void();
}
Reference in New Issue View Git Blame Copy Permalink