1354 lines
49 KiB
C++
1354 lines
49 KiB
C++
/*
|
|
* TaskBucket.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 "fdbclient/TaskBucket.h"
|
|
#include "fdbclient/ReadYourWrites.h"
|
|
#include "flow/actorcompiler.h" // has to be last include
|
|
|
|
Reference<TaskFuture> Task::getDoneFuture(Reference<FutureBucket> fb) {
|
|
return fb->unpack(params[reservedTaskParamKeyDone]);
|
|
}
|
|
|
|
struct UnblockFutureTaskFunc : TaskFuncBase {
|
|
static StringRef name;
|
|
|
|
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);
|
|
};
|
|
|
|
ACTOR static Future<Void> _finish(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<FutureBucket> futureBucket,
|
|
Reference<Task> task) {
|
|
state Reference<TaskFuture> future = futureBucket->unpack(task->params[Task::reservedTaskParamKeyFuture]);
|
|
|
|
futureBucket->setOptions(tr);
|
|
|
|
tr->clear(future->blocks.pack(task->params[Task::reservedTaskParamKeyBlockID]));
|
|
|
|
bool is_set = wait(future->isSet(tr));
|
|
if (is_set) {
|
|
wait(future->performAllActions(tr, taskBucket));
|
|
}
|
|
|
|
return Void();
|
|
}
|
|
};
|
|
StringRef UnblockFutureTaskFunc::name = LiteralStringRef("UnblockFuture");
|
|
REGISTER_TASKFUNC(UnblockFutureTaskFunc);
|
|
|
|
struct AddTaskFunc : TaskFuncBase {
|
|
static StringRef name;
|
|
|
|
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 {
|
|
task->params[Task::reservedTaskParamKeyType] = task->params[Task::reservedTaskParamKeyAddTask];
|
|
tb->addTask(tr, task);
|
|
return Void();
|
|
};
|
|
};
|
|
StringRef AddTaskFunc::name = LiteralStringRef("AddTask");
|
|
REGISTER_TASKFUNC(AddTaskFunc);
|
|
|
|
struct IdleTaskFunc : 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 tb->finish(tr, task);
|
|
};
|
|
};
|
|
StringRef IdleTaskFunc::name = LiteralStringRef("idle");
|
|
REGISTER_TASKFUNC(IdleTaskFunc);
|
|
|
|
Key Task::reservedTaskParamKeyType = LiteralStringRef("type");
|
|
Key Task::reservedTaskParamKeyAddTask = LiteralStringRef("_add_task");
|
|
Key Task::reservedTaskParamKeyDone = LiteralStringRef("done");
|
|
Key Task::reservedTaskParamKeyPriority = LiteralStringRef("priority");
|
|
Key Task::reservedTaskParamKeyFuture = LiteralStringRef("future");
|
|
Key Task::reservedTaskParamKeyBlockID = LiteralStringRef("blockid");
|
|
Key Task::reservedTaskParamKeyVersion = LiteralStringRef("version");
|
|
Key Task::reservedTaskParamValidKey = LiteralStringRef("_validkey");
|
|
Key Task::reservedTaskParamValidValue = LiteralStringRef("_validvalue");
|
|
|
|
// IMPORTANT: Task() must result in an EMPTY parameter set, so params should only
|
|
// be set for non-default constructor arguments. To change this behavior look at all
|
|
// Task() default constructions to see if they require params to be empty and call clear.
|
|
Task::Task(Value type, uint32_t version, Value done, unsigned int priority) : extendMutex(1) {
|
|
if (type.size())
|
|
params[Task::reservedTaskParamKeyType] = type;
|
|
|
|
if (version > 0)
|
|
params[Task::reservedTaskParamKeyVersion] = BinaryWriter::toValue(version, Unversioned());
|
|
|
|
if (done.size())
|
|
params[Task::reservedTaskParamKeyDone] = done;
|
|
|
|
priority = std::min<int64_t>(priority, CLIENT_KNOBS->TASKBUCKET_MAX_PRIORITY);
|
|
if (priority != 0)
|
|
params[Task::reservedTaskParamKeyPriority] = BinaryWriter::toValue<int64_t>(priority, Unversioned());
|
|
}
|
|
|
|
uint32_t Task::getVersion() const {
|
|
uint32_t version(0);
|
|
auto itor = params.find(Task::reservedTaskParamKeyVersion);
|
|
if (itor != params.end()) {
|
|
version = BinaryReader::fromStringRef<uint32_t>(itor->value, Unversioned());
|
|
} else {
|
|
TraceEvent(SevWarn, "InvalidTaskVersion").detail("TaskHasNoVersion", version);
|
|
}
|
|
|
|
return version;
|
|
}
|
|
|
|
unsigned int Task::getPriority() const {
|
|
unsigned int priority = 0;
|
|
auto i = params.find(Task::reservedTaskParamKeyPriority);
|
|
if (i != params.end())
|
|
priority = std::min<int64_t>(BinaryReader::fromStringRef<int64_t>(i->value, Unversioned()),
|
|
CLIENT_KNOBS->TASKBUCKET_MAX_PRIORITY);
|
|
return priority;
|
|
}
|
|
|
|
class TaskBucketImpl {
|
|
public:
|
|
ACTOR static Future<Optional<Key>> getTaskKey(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
int priority = 0) {
|
|
Standalone<StringRef> uid = StringRef(deterministicRandom()->randomUniqueID().toString());
|
|
|
|
// Get keyspace for the specified priority level
|
|
state Subspace space = taskBucket->getAvailableSpace(priority);
|
|
|
|
{
|
|
// Get a task key that is <= a random UID task key, if successful then return it
|
|
Key k = wait(tr->getKey(lastLessOrEqual(space.pack(uid)), true));
|
|
if (space.contains(k))
|
|
return Optional<Key>(k);
|
|
}
|
|
|
|
{
|
|
// Get a task key that is <= the maximum possible UID, if successful return it.
|
|
Key k = wait(tr->getKey(lastLessOrEqual(space.pack(maxUIDKey)), true));
|
|
if (space.contains(k))
|
|
return Optional<Key>(k);
|
|
}
|
|
|
|
return Optional<Key>();
|
|
}
|
|
|
|
ACTOR static Future<Reference<Task>> getOne(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket) {
|
|
if (taskBucket->priority_batch)
|
|
tr->setOption(FDBTransactionOptions::PRIORITY_BATCH);
|
|
|
|
taskBucket->setOptions(tr);
|
|
|
|
// give it some chances for the timed out tasks to get into the task loop in the case of
|
|
// many other new tasks get added so that the timed out tasks never get chances to re-run
|
|
if (deterministicRandom()->random01() < CLIENT_KNOBS->TASKBUCKET_CHECK_TIMEOUT_CHANCE) {
|
|
bool anyTimeouts = wait(requeueTimedOutTasks(tr, taskBucket));
|
|
TEST(anyTimeouts); // Found a task that timed out
|
|
}
|
|
|
|
state std::vector<Future<Optional<Key>>> taskKeyFutures(CLIENT_KNOBS->TASKBUCKET_MAX_PRIORITY + 1);
|
|
|
|
// Start looking for a task at each priority, highest first
|
|
state int pri;
|
|
for (pri = CLIENT_KNOBS->TASKBUCKET_MAX_PRIORITY; pri >= 0; --pri)
|
|
taskKeyFutures[pri] = getTaskKey(tr, taskBucket, pri);
|
|
|
|
// Task key and subspace it is located in.
|
|
state Optional<Key> taskKey;
|
|
state Subspace availableSpace;
|
|
|
|
// In priority order from highest to lowest, wait for fetch to finish and if it found a task then cancel the
|
|
// rest.
|
|
for (pri = CLIENT_KNOBS->TASKBUCKET_MAX_PRIORITY; pri >= 0; --pri) {
|
|
// If we already have a task key then cancel this fetch
|
|
if (taskKey.present())
|
|
taskKeyFutures[pri].cancel();
|
|
else {
|
|
Optional<Key> key = wait(taskKeyFutures[pri]);
|
|
if (key.present()) {
|
|
taskKey = key;
|
|
availableSpace = taskBucket->getAvailableSpace(pri);
|
|
}
|
|
}
|
|
}
|
|
|
|
// If we don't have a task key, requeue timed out tasks and try again by calling self.
|
|
if (!taskKey.present()) {
|
|
bool anyTimeouts = wait(requeueTimedOutTasks(tr, taskBucket));
|
|
// If there were timeouts, try to get a task since there should now be one in one of the available spaces.
|
|
if (anyTimeouts) {
|
|
TEST(true); // Try to get one task from timeouts subspace
|
|
Reference<Task> task = wait(getOne(tr, taskBucket));
|
|
return task;
|
|
}
|
|
return Reference<Task>();
|
|
}
|
|
|
|
// Now we know the task key is present and we have the available space for the task's priority
|
|
state Tuple t = availableSpace.unpack(taskKey.get());
|
|
state Key taskUID = t.getString(0);
|
|
state Subspace taskAvailableSpace = availableSpace.get(taskUID);
|
|
|
|
state Reference<Task> task(new Task());
|
|
task->key = taskUID;
|
|
|
|
state Standalone<RangeResultRef> values =
|
|
wait(tr->getRange(taskAvailableSpace.range(), CLIENT_KNOBS->TOO_MANY));
|
|
Version version = wait(tr->getReadVersion());
|
|
task->timeoutVersion =
|
|
version + (uint64_t)(taskBucket->timeout *
|
|
(CLIENT_KNOBS->TASKBUCKET_TIMEOUT_JITTER_OFFSET +
|
|
CLIENT_KNOBS->TASKBUCKET_TIMEOUT_JITTER_RANGE * deterministicRandom()->random01()));
|
|
|
|
Subspace timeoutSpace = taskBucket->timeouts.get(task->timeoutVersion).get(taskUID);
|
|
|
|
for (auto& s : values) {
|
|
Key param = taskAvailableSpace.unpack(s.key).getString(0);
|
|
task->params[param] = s.value;
|
|
tr->set(timeoutSpace.pack(param), s.value);
|
|
}
|
|
|
|
// Clear task definition in the available keyspace
|
|
tr->clear(taskAvailableSpace.range());
|
|
tr->set(taskBucket->active.key(), deterministicRandom()->randomUniqueID().toString());
|
|
|
|
return task;
|
|
}
|
|
|
|
// Verify that the user configured task verification key still has the user specified value
|
|
ACTOR static Future<bool> taskVerify(Reference<TaskBucket> tb,
|
|
Reference<ReadYourWritesTransaction> tr,
|
|
Reference<Task> task) {
|
|
|
|
if (task->params.find(Task::reservedTaskParamValidKey) == task->params.end()) {
|
|
TraceEvent("TaskBucketTaskVerifyInvalidTask")
|
|
.detail("Task", task->params[Task::reservedTaskParamKeyType])
|
|
.detail("ReservedTaskParamValidKey", "missing");
|
|
return false;
|
|
}
|
|
|
|
if (task->params.find(Task::reservedTaskParamValidValue) == task->params.end()) {
|
|
TraceEvent("TaskBucketTaskVerifyInvalidTask")
|
|
.detail("Task", task->params[Task::reservedTaskParamKeyType])
|
|
.detail("ReservedTaskParamValidKey", task->params[Task::reservedTaskParamValidKey])
|
|
.detail("ReservedTaskParamValidValue", "missing");
|
|
return false;
|
|
}
|
|
|
|
tb->setOptions(tr);
|
|
|
|
Optional<Value> keyValue = wait(tr->get(task->params[Task::reservedTaskParamValidKey]));
|
|
|
|
if (!keyValue.present()) {
|
|
TraceEvent("TaskBucketTaskVerifyInvalidTask")
|
|
.detail("Task", task->params[Task::reservedTaskParamKeyType])
|
|
.detail("ReservedTaskParamValidKey", task->params[Task::reservedTaskParamValidKey])
|
|
.detail("ReservedTaskParamValidValue", task->params[Task::reservedTaskParamValidValue])
|
|
.detail("KeyValue", "missing");
|
|
return false;
|
|
}
|
|
|
|
if (keyValue.get().compare(StringRef(task->params[Task::reservedTaskParamValidValue]))) {
|
|
TraceEvent("TaskBucketTaskVerifyAbortedTask")
|
|
.detail("Task", task->params[Task::reservedTaskParamKeyType])
|
|
.detail("ReservedTaskParamValidKey", task->params[Task::reservedTaskParamValidKey])
|
|
.detail("ReservedTaskParamValidValue", task->params[Task::reservedTaskParamValidValue])
|
|
.detail("KeyValue", keyValue.get());
|
|
return false;
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
ACTOR static Future<bool> taskVerify(Reference<TaskBucket> tb, Database cx, Reference<Task> task) {
|
|
loop {
|
|
state Reference<ReadYourWritesTransaction> tr(new ReadYourWritesTransaction(cx));
|
|
|
|
try {
|
|
bool verified = wait(taskVerify(tb, tr, task));
|
|
return verified;
|
|
} catch (Error& e) {
|
|
wait(tr->onError(e));
|
|
}
|
|
}
|
|
}
|
|
|
|
ACTOR static Future<Void> finishTaskRun(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<FutureBucket> futureBucket,
|
|
Reference<Task> task,
|
|
Reference<TaskFuncBase> taskFunc,
|
|
bool verifyTask) {
|
|
bool isFinished = wait(taskBucket->isFinished(tr, task));
|
|
if (isFinished) {
|
|
return Void();
|
|
}
|
|
|
|
state bool validTask = true;
|
|
if (verifyTask) {
|
|
bool _validTask = wait(taskVerify(taskBucket, tr, task));
|
|
validTask = _validTask;
|
|
}
|
|
if (!validTask) {
|
|
wait(taskBucket->finish(tr, task));
|
|
} else {
|
|
wait(taskFunc->finish(tr, taskBucket, futureBucket, task));
|
|
}
|
|
|
|
return Void();
|
|
}
|
|
|
|
ACTOR static Future<bool> doOne(Database cx,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<FutureBucket> futureBucket) {
|
|
state Reference<Task> task = wait(taskBucket->getOne(cx));
|
|
bool result = wait(taskBucket->doTask(cx, futureBucket, task));
|
|
return result;
|
|
}
|
|
|
|
ACTOR static Future<Void> extendTimeoutRepeatedly(Database cx,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<Task> task) {
|
|
state Reference<ReadYourWritesTransaction> tr(new ReadYourWritesTransaction(cx));
|
|
state double start = now();
|
|
state Version versionNow = wait(runRYWTransaction(cx, [=](Reference<ReadYourWritesTransaction> tr) {
|
|
taskBucket->setOptions(tr);
|
|
return map(tr->getReadVersion(), [=](Version v) { return v; });
|
|
}));
|
|
|
|
loop {
|
|
state FlowLock::Releaser releaser;
|
|
|
|
// Wait until we are half way to the timeout version of this task
|
|
wait(delay(0.8 * (BUGGIFY ? (2 * deterministicRandom()->random01()) : 1.0) *
|
|
(double)(task->timeoutVersion - (uint64_t)versionNow) / CLIENT_KNOBS->CORE_VERSIONSPERSECOND));
|
|
|
|
if (now() - start > 300) {
|
|
TraceEvent(SevWarnAlways, "TaskBucketLongExtend")
|
|
.detail("Duration", now() - start)
|
|
.detail("TaskUID", task->key)
|
|
.detail("TaskType", task->params[Task::reservedTaskParamKeyType])
|
|
.detail("Priority", task->getPriority());
|
|
}
|
|
// Take the extendMutex lock until we either succeed or stop trying to extend due to failure
|
|
wait(task->extendMutex.take());
|
|
releaser = FlowLock::Releaser(task->extendMutex, 1);
|
|
|
|
loop {
|
|
try {
|
|
tr->reset();
|
|
taskBucket->setOptions(tr);
|
|
|
|
// Attempt to extend the task's timeout
|
|
state Version newTimeout = wait(taskBucket->extendTimeout(tr, task, false));
|
|
wait(tr->commit());
|
|
task->timeoutVersion = newTimeout;
|
|
versionNow = tr->getCommittedVersion();
|
|
break;
|
|
} catch (Error& e) {
|
|
wait(tr->onError(e));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
ACTOR static Future<bool> doTask(Database cx,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<FutureBucket> futureBucket,
|
|
Reference<Task> task) {
|
|
if (!task || !TaskFuncBase::isValidTask(task))
|
|
return false;
|
|
|
|
state Reference<TaskFuncBase> taskFunc;
|
|
|
|
try {
|
|
taskFunc = TaskFuncBase::create(task->params[Task::reservedTaskParamKeyType]);
|
|
if (taskFunc) {
|
|
state bool verifyTask = (task->params.find(Task::reservedTaskParamValidKey) != task->params.end());
|
|
|
|
if (verifyTask) {
|
|
loop {
|
|
state Reference<ReadYourWritesTransaction> tr(new ReadYourWritesTransaction(cx));
|
|
taskBucket->setOptions(tr);
|
|
|
|
try {
|
|
bool validTask = wait(taskVerify(taskBucket, tr, task));
|
|
|
|
if (!validTask) {
|
|
bool isFinished = wait(taskBucket->isFinished(tr, task));
|
|
if (!isFinished) {
|
|
wait(taskBucket->finish(tr, task));
|
|
}
|
|
wait(tr->commit());
|
|
return true;
|
|
}
|
|
break;
|
|
} catch (Error& e) {
|
|
wait(tr->onError(e));
|
|
}
|
|
}
|
|
}
|
|
|
|
wait(taskFunc->execute(cx, taskBucket, futureBucket, task) ||
|
|
extendTimeoutRepeatedly(cx, taskBucket, task));
|
|
|
|
if (BUGGIFY)
|
|
wait(delay(10.0));
|
|
wait(runRYWTransaction(cx, [=](Reference<ReadYourWritesTransaction> tr) {
|
|
return finishTaskRun(tr, taskBucket, futureBucket, task, taskFunc, verifyTask);
|
|
}));
|
|
}
|
|
} catch (Error& e) {
|
|
TraceEvent(SevWarn, "TaskBucketExecuteFailure")
|
|
.error(e)
|
|
.detail("TaskUID", task->key)
|
|
.detail("TaskType", task->params[Task::reservedTaskParamKeyType].printable())
|
|
.detail("Priority", task->getPriority());
|
|
try {
|
|
wait(taskFunc->handleError(cx, task, e));
|
|
} catch (Error& e) {
|
|
TraceEvent(SevWarn, "TaskBucketExecuteFailureLogErrorFailed")
|
|
.error(e) // output handleError() error instead of original task error
|
|
.detail("TaskUID", task->key.printable())
|
|
.detail("TaskType", task->params[Task::reservedTaskParamKeyType].printable())
|
|
.detail("Priority", task->getPriority());
|
|
}
|
|
}
|
|
|
|
// Return true to indicate that we did work.
|
|
return true;
|
|
}
|
|
|
|
ACTOR static Future<Void> dispatch(Database cx,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<FutureBucket> futureBucket,
|
|
double* pollDelay,
|
|
int maxConcurrentTasks) {
|
|
state std::vector<Future<bool>> tasks(maxConcurrentTasks);
|
|
for (auto& f : tasks)
|
|
f = Never();
|
|
|
|
// Since the futures have to be kept in a vector to be compatible with waitForAny(), we'll keep a queue
|
|
// of available slots in it. Initially, they're all available.
|
|
state std::vector<int> availableSlots;
|
|
availableSlots.reserve(tasks.size());
|
|
for (int i = 0; i < tasks.size(); ++i)
|
|
availableSlots.push_back(i);
|
|
|
|
state std::vector<Future<Reference<Task>>> getTasks;
|
|
state unsigned int getBatchSize = 1;
|
|
|
|
loop {
|
|
// Start running tasks while slots are available and we keep finding work to do
|
|
++taskBucket->dispatchSlotChecksStarted;
|
|
while (!availableSlots.empty()) {
|
|
getTasks.clear();
|
|
for (int i = 0, imax = std::min<unsigned int>(getBatchSize, availableSlots.size()); i < imax; ++i)
|
|
getTasks.push_back(taskBucket->getOne(cx));
|
|
wait(waitForAllReady(getTasks));
|
|
|
|
bool done = false;
|
|
for (int i = 0; i < getTasks.size(); ++i) {
|
|
if (getTasks[i].isError()) {
|
|
++taskBucket->dispatchErrors;
|
|
done = true;
|
|
continue;
|
|
}
|
|
Reference<Task> task = getTasks[i].get();
|
|
if (task) {
|
|
// Start the task
|
|
++taskBucket->dispatchDoTasks;
|
|
int slot = availableSlots.back();
|
|
availableSlots.pop_back();
|
|
tasks[slot] = taskBucket->doTask(cx, futureBucket, task);
|
|
} else {
|
|
++taskBucket->dispatchEmptyTasks;
|
|
done = true;
|
|
}
|
|
}
|
|
|
|
if (done) {
|
|
getBatchSize = 1;
|
|
break;
|
|
} else
|
|
getBatchSize = std::min<unsigned int>(getBatchSize * 2, maxConcurrentTasks);
|
|
}
|
|
++taskBucket->dispatchSlotChecksComplete;
|
|
|
|
// Wait for a task to be done. Also, if we have any slots available then stop waiting after pollDelay at
|
|
// the latest.
|
|
Future<Void> w = ready(waitForAny(tasks));
|
|
if (!availableSlots.empty()) {
|
|
if (*pollDelay > 600) {
|
|
TraceEvent(SevWarnAlways, "TaskBucketLongPollDelay").suppressFor(1.0).detail("Delay", *pollDelay);
|
|
}
|
|
w = w ||
|
|
delay(*pollDelay * (0.9 + deterministicRandom()->random01() / 5)); // Jittered by 20 %, so +/- 10%
|
|
}
|
|
wait(w);
|
|
|
|
// Check all of the task slots, any that are finished should be replaced with Never() and their slots added
|
|
// back to availableSlots
|
|
for (int i = 0; i < tasks.size(); ++i) {
|
|
if (tasks[i].isReady()) {
|
|
availableSlots.push_back(i);
|
|
tasks[i] = Never();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
ACTOR static Future<Void> watchPaused(Database cx,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<AsyncVar<bool>> paused) {
|
|
loop {
|
|
state Reference<ReadYourWritesTransaction> tr(new ReadYourWritesTransaction(cx));
|
|
try {
|
|
taskBucket->setOptions(tr);
|
|
Optional<Value> pausedVal = wait(tr->get(taskBucket->pauseKey));
|
|
paused->set(pausedVal.present());
|
|
state Future<Void> watchPausedFuture = tr->watch(taskBucket->pauseKey);
|
|
wait(tr->commit());
|
|
wait(watchPausedFuture);
|
|
} catch (Error& e) {
|
|
wait(tr->onError(e));
|
|
}
|
|
}
|
|
}
|
|
|
|
ACTOR static Future<Void> run(Database cx,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<FutureBucket> futureBucket,
|
|
double* pollDelay,
|
|
int maxConcurrentTasks) {
|
|
state Reference<AsyncVar<bool>> paused = makeReference<AsyncVar<bool>>(true);
|
|
state Future<Void> watchPausedFuture = watchPaused(cx, taskBucket, paused);
|
|
taskBucket->metricLogger = traceCounters(
|
|
"TaskBucketMetrics", taskBucket->dbgid, CLIENT_KNOBS->TASKBUCKET_LOGGING_DELAY, &taskBucket->cc);
|
|
loop {
|
|
while (paused->get()) {
|
|
wait(paused->onChange() || watchPausedFuture);
|
|
}
|
|
|
|
wait(dispatch(cx, taskBucket, futureBucket, pollDelay, maxConcurrentTasks) || paused->onChange() ||
|
|
watchPausedFuture);
|
|
}
|
|
}
|
|
|
|
static Future<Standalone<StringRef>> addIdle(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket) {
|
|
taskBucket->setOptions(tr);
|
|
|
|
auto newTask = makeReference<Task>(IdleTaskFunc::name, IdleTaskFunc::version);
|
|
return taskBucket->addTask(tr, newTask);
|
|
}
|
|
|
|
static Future<Standalone<StringRef>> addIdle(Database cx, Reference<TaskBucket> taskBucket) {
|
|
return runRYWTransaction(cx, [=](Reference<ReadYourWritesTransaction> tr) { return addIdle(tr, taskBucket); });
|
|
}
|
|
|
|
ACTOR static Future<bool> isEmpty(Reference<ReadYourWritesTransaction> tr, Reference<TaskBucket> taskBucket) {
|
|
taskBucket->setOptions(tr);
|
|
|
|
// Check all available priorities for keys
|
|
state std::vector<Future<Standalone<RangeResultRef>>> resultFutures;
|
|
for (int pri = 0; pri <= CLIENT_KNOBS->TASKBUCKET_MAX_PRIORITY; ++pri)
|
|
resultFutures.push_back(tr->getRange(taskBucket->getAvailableSpace(pri).range(), 1));
|
|
|
|
// If any priority levels have any keys then the taskbucket is not empty so return false
|
|
state int i;
|
|
for (i = 0; i < resultFutures.size(); ++i) {
|
|
Standalone<RangeResultRef> results = wait(resultFutures[i]);
|
|
if (results.size() > 0)
|
|
return false;
|
|
}
|
|
|
|
Standalone<RangeResultRef> values = wait(tr->getRange(taskBucket->timeouts.range(), 1));
|
|
if (values.size() > 0)
|
|
return false;
|
|
|
|
return true;
|
|
}
|
|
|
|
ACTOR static Future<bool> isBusy(Reference<ReadYourWritesTransaction> tr, Reference<TaskBucket> taskBucket) {
|
|
taskBucket->setOptions(tr);
|
|
|
|
// Check all available priorities for emptiness
|
|
state std::vector<Future<Standalone<RangeResultRef>>> resultFutures;
|
|
for (int pri = 0; pri <= CLIENT_KNOBS->TASKBUCKET_MAX_PRIORITY; ++pri)
|
|
resultFutures.push_back(tr->getRange(taskBucket->getAvailableSpace(pri).range(), 1));
|
|
|
|
// If any priority levels have any keys then return true as the level is 'busy'
|
|
state int i;
|
|
for (i = 0; i < resultFutures.size(); ++i) {
|
|
Standalone<RangeResultRef> results = wait(resultFutures[i]);
|
|
if (results.size() > 0)
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
// Verify that the task's keys are still in the timeout space at the expected timeout prefix
|
|
ACTOR static Future<bool> isFinished(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<Task> task) {
|
|
taskBucket->setOptions(tr);
|
|
|
|
Tuple t;
|
|
t.append(task->timeoutVersion);
|
|
t.append(task->key);
|
|
|
|
Standalone<RangeResultRef> values = wait(tr->getRange(taskBucket->timeouts.range(t), 1));
|
|
if (values.size() > 0)
|
|
return false;
|
|
|
|
return true;
|
|
}
|
|
|
|
ACTOR static Future<bool> getActiveKey(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Optional<Value> startingValue) {
|
|
taskBucket->setOptions(tr);
|
|
|
|
Optional<Value> new_value = wait(tr->get(taskBucket->active.key()));
|
|
if (new_value != startingValue) {
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
ACTOR static Future<bool> checkActive(Database cx, Reference<TaskBucket> taskBucket) {
|
|
state Reference<ReadYourWritesTransaction> tr(new ReadYourWritesTransaction(cx));
|
|
state Optional<Value> startingValue;
|
|
|
|
loop {
|
|
try {
|
|
taskBucket->setOptions(tr);
|
|
|
|
bool is_busy = wait(isBusy(tr, taskBucket));
|
|
if (!is_busy) {
|
|
wait(success(addIdle(tr, taskBucket)));
|
|
}
|
|
|
|
Optional<Value> val = wait(tr->get(taskBucket->active.key()));
|
|
startingValue = val;
|
|
|
|
wait(tr->commit());
|
|
break;
|
|
} catch (Error& e) {
|
|
wait(tr->onError(e));
|
|
}
|
|
}
|
|
|
|
state int idx = 0;
|
|
for (; idx < CLIENT_KNOBS->TASKBUCKET_CHECK_ACTIVE_AMOUNT; ++idx) {
|
|
tr = Reference<ReadYourWritesTransaction>(new ReadYourWritesTransaction(cx));
|
|
loop {
|
|
try {
|
|
taskBucket->setOptions(tr);
|
|
|
|
wait(delay(CLIENT_KNOBS->TASKBUCKET_CHECK_ACTIVE_DELAY));
|
|
bool isActiveKey = wait(getActiveKey(tr, taskBucket, startingValue));
|
|
if (isActiveKey) {
|
|
TEST(true); // checkActive return true
|
|
return true;
|
|
}
|
|
break;
|
|
} catch (Error& e) {
|
|
wait(tr->onError(e));
|
|
}
|
|
}
|
|
}
|
|
|
|
TEST(true); // checkActive return false
|
|
return false;
|
|
}
|
|
|
|
ACTOR static Future<int64_t> getTaskCount(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket) {
|
|
taskBucket->setOptions(tr);
|
|
|
|
Optional<Value> val = wait(tr->get(taskBucket->prefix.pack(LiteralStringRef("task_count"))));
|
|
|
|
if (!val.present())
|
|
return 0;
|
|
|
|
ASSERT(val.get().size() == sizeof(int64_t));
|
|
|
|
int64_t intValue = 0;
|
|
memcpy(&intValue, val.get().begin(), val.get().size());
|
|
|
|
return intValue;
|
|
}
|
|
|
|
// Looks for tasks that have timed out and returns them to be available tasks.
|
|
// Returns True if any tasks were affected.
|
|
ACTOR static Future<bool> requeueTimedOutTasks(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket) {
|
|
TEST(true); // Looks for tasks that have timed out and returns them to be available tasks.
|
|
Version end = wait(tr->getReadVersion());
|
|
state KeyRange range(
|
|
KeyRangeRef(taskBucket->timeouts.get(0).range().begin, taskBucket->timeouts.get(end).range().end));
|
|
|
|
Standalone<RangeResultRef> values = wait(tr->getRange(range, CLIENT_KNOBS->TASKBUCKET_MAX_TASK_KEYS));
|
|
|
|
// Keys will be tuples of (taskUID, param) -> paramValue
|
|
// Unfortunately we need to know the priority parameter for a taskUID before we can know which available-tasks
|
|
// subspace to move its keys to. The cleanest way to do this is to load a new Task() with parameters and once a
|
|
// new task id is encountered flush the old one using taskBucket->getAvailableSpace(task->getPriority())
|
|
|
|
Task task;
|
|
Key lastKey;
|
|
|
|
for (auto& iter : values) {
|
|
Tuple t = taskBucket->timeouts.unpack(iter.key);
|
|
Key uid = t.getString(1);
|
|
Key param = t.getString(2);
|
|
|
|
// If a new UID is seen, finish moving task to new available space. Safe if task == Task()
|
|
if (uid != task.key) {
|
|
// Get the space for this specific task within its available keyspace for its priority
|
|
Subspace space = taskBucket->getAvailableSpace(task.getPriority()).get(task.key);
|
|
for (auto& p : task.params) {
|
|
tr->set(space.pack(p.key), p.value);
|
|
}
|
|
task.params.clear();
|
|
task.key = uid;
|
|
lastKey = iter.key;
|
|
}
|
|
|
|
task.params[param] = iter.value;
|
|
}
|
|
|
|
// Move the final task, if complete, to its new available keyspace. Safe if task == Task()
|
|
if (!values.more) {
|
|
Subspace space = taskBucket->getAvailableSpace(task.getPriority()).get(task.key);
|
|
for (auto& p : task.params)
|
|
tr->set(space.pack(p.key), p.value);
|
|
|
|
if (values.size() > 0) {
|
|
tr->clear(range);
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
ASSERT(lastKey != Key());
|
|
tr->clear(KeyRangeRef(range.begin, lastKey));
|
|
return true;
|
|
}
|
|
|
|
ACTOR static Future<Void> debugPrintRange(Reference<ReadYourWritesTransaction> tr, Subspace subspace, Key msg) {
|
|
tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
|
|
tr->setOption(FDBTransactionOptions::LOCK_AWARE);
|
|
Standalone<RangeResultRef> values = wait(tr->getRange(subspace.range(), CLIENT_KNOBS->TOO_MANY));
|
|
TraceEvent("TaskBucketDebugPrintRange")
|
|
.detail("Key", subspace.key())
|
|
.detail("Count", values.size())
|
|
.detail("Msg", msg);
|
|
|
|
/*printf("debugPrintRange key: (%d) %s\n", values.size(), printable(subspace.key()).c_str());
|
|
for (auto & s : values) {
|
|
printf(" key: %-40s value: %s\n", printable(s.key).c_str(), s.value.c_str());
|
|
TraceEvent("TaskBucketDebugPrintKV").detail("Msg", msg)
|
|
.detail("Key", s.key)
|
|
.detail("Value", s.value);
|
|
}*/
|
|
|
|
return Void();
|
|
}
|
|
|
|
ACTOR static Future<Version> extendTimeout(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<Task> task,
|
|
bool updateParams,
|
|
Version newTimeoutVersion) {
|
|
taskBucket->setOptions(tr);
|
|
|
|
// First make sure it's safe to keep running
|
|
wait(taskBucket->keepRunning(tr, task));
|
|
|
|
// This is where the task definition currently exists
|
|
state Subspace oldTimeoutSpace = taskBucket->timeouts.get(task->timeoutVersion).get(task->key);
|
|
// Update the task's timeout
|
|
Version version = wait(tr->getReadVersion());
|
|
|
|
if (newTimeoutVersion == invalidVersion)
|
|
newTimeoutVersion = version + taskBucket->timeout;
|
|
else if (newTimeoutVersion <= version) // Ensure that the time extension is to the future
|
|
newTimeoutVersion = version + 1;
|
|
|
|
// This can happen if extendTimeout is called shortly after task start and the task's timeout was jittered to be
|
|
// longer
|
|
if (newTimeoutVersion <= task->timeoutVersion) {
|
|
newTimeoutVersion = task->timeoutVersion + 1;
|
|
}
|
|
|
|
// This is where the task definition is being moved to
|
|
state Subspace newTimeoutSpace = taskBucket->timeouts.get(newTimeoutVersion).get(task->key);
|
|
|
|
tr->addReadConflictRange(oldTimeoutSpace.range());
|
|
tr->addWriteConflictRange(newTimeoutSpace.range());
|
|
|
|
// If we're updating the task params the clear the old space and write params to the new space
|
|
if (updateParams) {
|
|
TEST(true); // Extended a task while updating parameters
|
|
for (auto& p : task->params) {
|
|
tr->set(newTimeoutSpace.pack(p.key), p.value);
|
|
}
|
|
} else {
|
|
TEST(true); // Extended a task without updating parameters
|
|
// Otherwise, read and transplant the params from the old to new timeout spaces
|
|
Standalone<RangeResultRef> params = wait(tr->getRange(oldTimeoutSpace.range(), CLIENT_KNOBS->TOO_MANY));
|
|
for (auto& kv : params) {
|
|
Tuple paramKey = oldTimeoutSpace.unpack(kv.key);
|
|
tr->set(newTimeoutSpace.pack(paramKey), kv.value);
|
|
}
|
|
}
|
|
|
|
tr->clear(oldTimeoutSpace.range());
|
|
|
|
return newTimeoutVersion;
|
|
}
|
|
};
|
|
|
|
TaskBucket::TaskBucket(const Subspace& subspace, bool sysAccess, bool priorityBatch, bool lockAware)
|
|
: prefix(subspace), active(prefix.get(LiteralStringRef("ac"))), available(prefix.get(LiteralStringRef("av"))),
|
|
available_prioritized(prefix.get(LiteralStringRef("avp"))), timeouts(prefix.get(LiteralStringRef("to"))),
|
|
pauseKey(prefix.pack(LiteralStringRef("pause"))), timeout(CLIENT_KNOBS->TASKBUCKET_TIMEOUT_VERSIONS),
|
|
system_access(sysAccess), priority_batch(priorityBatch), lock_aware(lockAware), cc("TaskBucket"),
|
|
dbgid(deterministicRandom()->randomUniqueID()), dispatchSlotChecksStarted("DispatchSlotChecksStarted", cc),
|
|
dispatchErrors("DispatchErrors", cc), dispatchDoTasks("DispatchDoTasks", cc),
|
|
dispatchEmptyTasks("DispatchEmptyTasks", cc), dispatchSlotChecksComplete("DispatchSlotChecksComplete", cc) {}
|
|
|
|
TaskBucket::~TaskBucket() {}
|
|
|
|
Future<Void> TaskBucket::clear(Reference<ReadYourWritesTransaction> tr) {
|
|
setOptions(tr);
|
|
|
|
tr->clear(prefix.range());
|
|
|
|
return Void();
|
|
}
|
|
|
|
Future<Void> TaskBucket::changePause(Reference<ReadYourWritesTransaction> tr, bool pause) {
|
|
setOptions(tr);
|
|
|
|
if (pause) {
|
|
tr->set(pauseKey, StringRef());
|
|
} else {
|
|
tr->clear(pauseKey);
|
|
}
|
|
|
|
return Void();
|
|
}
|
|
|
|
Key TaskBucket::addTask(Reference<ReadYourWritesTransaction> tr, Reference<Task> task) {
|
|
setOptions(tr);
|
|
|
|
Key key(deterministicRandom()->randomUniqueID().toString());
|
|
|
|
Subspace taskSpace;
|
|
|
|
// If scheduledVersion is valid then place the task directly into the timeout
|
|
// space for its scheduled time, otherwise place it in the available space by priority.
|
|
Version scheduledVersion = ReservedTaskParams::scheduledVersion().getOrDefault(task, invalidVersion);
|
|
if (scheduledVersion != invalidVersion) {
|
|
taskSpace = timeouts.get(scheduledVersion).get(key);
|
|
} else {
|
|
taskSpace = getAvailableSpace(task->getPriority()).get(key);
|
|
}
|
|
|
|
for (auto& param : task->params)
|
|
tr->set(taskSpace.pack(param.key), param.value);
|
|
|
|
tr->atomicOp(prefix.pack(LiteralStringRef("task_count")),
|
|
LiteralStringRef("\x01\x00\x00\x00\x00\x00\x00\x00"),
|
|
MutationRef::AddValue);
|
|
|
|
return key;
|
|
}
|
|
|
|
void TaskBucket::setValidationCondition(Reference<Task> task, KeyRef vKey, KeyRef vValue) {
|
|
task->params[Task::reservedTaskParamValidKey] = vKey;
|
|
task->params[Task::reservedTaskParamValidValue] = vValue;
|
|
}
|
|
|
|
ACTOR static Future<Key> actorAddTask(TaskBucket* tb,
|
|
Reference<ReadYourWritesTransaction> tr,
|
|
Reference<Task> task,
|
|
KeyRef validationKey) {
|
|
tb->setOptions(tr);
|
|
|
|
Optional<Value> validationValue = wait(tr->get(validationKey));
|
|
|
|
if (!validationValue.present()) {
|
|
TraceEvent(SevError, "TaskBucketAddTaskInvalidKey")
|
|
.detail("Task", task->params[Task::reservedTaskParamKeyType])
|
|
.detail("ValidationKey", validationKey);
|
|
throw invalid_option_value();
|
|
}
|
|
|
|
TaskBucket::setValidationCondition(task, validationKey, validationValue.get());
|
|
|
|
return tb->addTask(tr, task);
|
|
}
|
|
|
|
Future<Key> TaskBucket::addTask(Reference<ReadYourWritesTransaction> tr, Reference<Task> task, KeyRef validationKey) {
|
|
return actorAddTask(this, tr, task, validationKey);
|
|
}
|
|
|
|
Key TaskBucket::addTask(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<Task> task,
|
|
KeyRef validationKey,
|
|
KeyRef validationValue) {
|
|
setValidationCondition(task, validationKey, validationValue);
|
|
return addTask(tr, task);
|
|
}
|
|
|
|
Future<Reference<Task>> TaskBucket::getOne(Reference<ReadYourWritesTransaction> tr) {
|
|
return TaskBucketImpl::getOne(tr, Reference<TaskBucket>::addRef(this));
|
|
}
|
|
|
|
Future<bool> TaskBucket::doOne(Database cx, Reference<FutureBucket> futureBucket) {
|
|
return TaskBucketImpl::doOne(cx, Reference<TaskBucket>::addRef(this), futureBucket);
|
|
}
|
|
|
|
Future<bool> TaskBucket::doTask(Database cx, Reference<FutureBucket> futureBucket, Reference<Task> task) {
|
|
return TaskBucketImpl::doTask(cx, Reference<TaskBucket>::addRef(this), futureBucket, task);
|
|
}
|
|
|
|
Future<Void> TaskBucket::run(Database cx,
|
|
Reference<FutureBucket> futureBucket,
|
|
double* pollDelay,
|
|
int maxConcurrentTasks) {
|
|
return TaskBucketImpl::run(cx, Reference<TaskBucket>::addRef(this), futureBucket, pollDelay, maxConcurrentTasks);
|
|
}
|
|
|
|
Future<Void> TaskBucket::watchPaused(Database cx, Reference<AsyncVar<bool>> paused) {
|
|
return TaskBucketImpl::watchPaused(cx, Reference<TaskBucket>::addRef(this), paused);
|
|
}
|
|
|
|
Future<bool> TaskBucket::isEmpty(Reference<ReadYourWritesTransaction> tr) {
|
|
return TaskBucketImpl::isEmpty(tr, Reference<TaskBucket>::addRef(this));
|
|
}
|
|
|
|
Future<Void> TaskBucket::finish(Reference<ReadYourWritesTransaction> tr, Reference<Task> task) {
|
|
setOptions(tr);
|
|
|
|
Tuple t;
|
|
t.append(task->timeoutVersion);
|
|
t.append(task->key);
|
|
|
|
tr->atomicOp(prefix.pack(LiteralStringRef("task_count")),
|
|
LiteralStringRef("\xff\xff\xff\xff\xff\xff\xff\xff"),
|
|
MutationRef::AddValue);
|
|
tr->clear(timeouts.range(t));
|
|
|
|
return Void();
|
|
}
|
|
|
|
Future<Version> TaskBucket::extendTimeout(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<Task> task,
|
|
bool updateParams,
|
|
Version newTimeoutVersion) {
|
|
return TaskBucketImpl::extendTimeout(
|
|
tr, Reference<TaskBucket>::addRef(this), task, updateParams, newTimeoutVersion);
|
|
}
|
|
|
|
Future<bool> TaskBucket::isFinished(Reference<ReadYourWritesTransaction> tr, Reference<Task> task) {
|
|
return TaskBucketImpl::isFinished(tr, Reference<TaskBucket>::addRef(this), task);
|
|
}
|
|
|
|
Future<bool> TaskBucket::isVerified(Reference<ReadYourWritesTransaction> tr, Reference<Task> task) {
|
|
return TaskBucketImpl::taskVerify(Reference<TaskBucket>::addRef(this), tr, task);
|
|
}
|
|
|
|
Future<bool> TaskBucket::checkActive(Database cx) {
|
|
return TaskBucketImpl::checkActive(cx, Reference<TaskBucket>::addRef(this));
|
|
}
|
|
|
|
Future<int64_t> TaskBucket::getTaskCount(Reference<ReadYourWritesTransaction> tr) {
|
|
return TaskBucketImpl::getTaskCount(tr, Reference<TaskBucket>::addRef(this));
|
|
}
|
|
|
|
Future<Void> TaskBucket::watchTaskCount(Reference<ReadYourWritesTransaction> tr) {
|
|
return tr->watch(prefix.pack(LiteralStringRef("task_count")));
|
|
}
|
|
|
|
Future<Void> TaskBucket::debugPrintRange(Reference<ReadYourWritesTransaction> tr, Subspace subspace, Key msg) {
|
|
return TaskBucketImpl::debugPrintRange(tr, subspace, msg);
|
|
}
|
|
|
|
class FutureBucketImpl {
|
|
public:
|
|
ACTOR static Future<bool> isEmpty(Reference<ReadYourWritesTransaction> tr, Reference<FutureBucket> futureBucket) {
|
|
futureBucket->setOptions(tr);
|
|
|
|
Key lastKey = wait(tr->getKey(lastLessOrEqual(futureBucket->prefix.pack(maxUIDKey))));
|
|
return !futureBucket->prefix.contains(lastKey);
|
|
}
|
|
};
|
|
|
|
FutureBucket::FutureBucket(const Subspace& subspace, bool sysAccess, bool lockAware)
|
|
: prefix(subspace), system_access(sysAccess), lock_aware(lockAware) {}
|
|
|
|
FutureBucket::~FutureBucket() {}
|
|
|
|
Future<Void> FutureBucket::clear(Reference<ReadYourWritesTransaction> tr) {
|
|
setOptions(tr);
|
|
tr->clear(prefix.range());
|
|
|
|
return Void();
|
|
}
|
|
|
|
Reference<TaskFuture> FutureBucket::future(Reference<ReadYourWritesTransaction> tr) {
|
|
setOptions(tr);
|
|
|
|
auto taskFuture = makeReference<TaskFuture>(Reference<FutureBucket>::addRef(this));
|
|
taskFuture->addBlock(tr, StringRef());
|
|
|
|
return taskFuture;
|
|
}
|
|
|
|
Future<bool> FutureBucket::isEmpty(Reference<ReadYourWritesTransaction> tr) {
|
|
return FutureBucketImpl::isEmpty(tr, Reference<FutureBucket>::addRef(this));
|
|
}
|
|
|
|
Reference<TaskFuture> FutureBucket::unpack(Key key) {
|
|
return makeReference<TaskFuture>(Reference<FutureBucket>::addRef(this), key);
|
|
}
|
|
|
|
class TaskFutureImpl {
|
|
public:
|
|
ACTOR static Future<Void> join(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<TaskFuture> taskFuture,
|
|
std::vector<Reference<TaskFuture>> vectorFuture) {
|
|
taskFuture->futureBucket->setOptions(tr);
|
|
|
|
bool is_set = wait(isSet(tr, taskFuture));
|
|
if (is_set) {
|
|
return Void();
|
|
}
|
|
|
|
tr->clear(taskFuture->blocks.pack(StringRef()));
|
|
|
|
wait(_join(tr, taskBucket, taskFuture, vectorFuture));
|
|
|
|
return Void();
|
|
}
|
|
|
|
ACTOR static Future<Void> _join(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<TaskFuture> taskFuture,
|
|
std::vector<Reference<TaskFuture>> vectorFuture) {
|
|
std::vector<Future<Void>> onSetFutures;
|
|
for (int i = 0; i < vectorFuture.size(); ++i) {
|
|
Key key = StringRef(deterministicRandom()->randomUniqueID().toString());
|
|
taskFuture->addBlock(tr, key);
|
|
auto task = makeReference<Task>();
|
|
task->params[Task::reservedTaskParamKeyType] = LiteralStringRef("UnblockFuture");
|
|
task->params[Task::reservedTaskParamKeyFuture] = taskFuture->key;
|
|
task->params[Task::reservedTaskParamKeyBlockID] = key;
|
|
onSetFutures.push_back(vectorFuture[i]->onSet(tr, taskBucket, task));
|
|
}
|
|
|
|
wait(waitForAll(onSetFutures));
|
|
|
|
return Void();
|
|
}
|
|
|
|
ACTOR static Future<bool> isSet(Reference<ReadYourWritesTransaction> tr, Reference<TaskFuture> taskFuture) {
|
|
taskFuture->futureBucket->setOptions(tr);
|
|
|
|
Standalone<RangeResultRef> values = wait(tr->getRange(taskFuture->blocks.range(), 1));
|
|
if (values.size() > 0)
|
|
return false;
|
|
|
|
return true;
|
|
}
|
|
|
|
ACTOR static Future<Void> onSet(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<TaskFuture> taskFuture,
|
|
Reference<Task> task) {
|
|
taskFuture->futureBucket->setOptions(tr);
|
|
|
|
bool is_set = wait(isSet(tr, taskFuture));
|
|
|
|
if (is_set) {
|
|
TEST(true); // is_set == true
|
|
wait(performAction(tr, taskBucket, taskFuture, task));
|
|
} else {
|
|
TEST(true); // is_set == false
|
|
Subspace callbackSpace =
|
|
taskFuture->callbacks.get(StringRef(deterministicRandom()->randomUniqueID().toString()));
|
|
for (auto& v : task->params) {
|
|
tr->set(callbackSpace.pack(v.key), v.value);
|
|
}
|
|
}
|
|
|
|
return Void();
|
|
}
|
|
|
|
ACTOR static Future<Void> set(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<TaskFuture> taskFuture) {
|
|
taskFuture->futureBucket->setOptions(tr);
|
|
|
|
tr->clear(taskFuture->blocks.range());
|
|
|
|
wait(performAllActions(tr, taskBucket, taskFuture));
|
|
|
|
return Void();
|
|
}
|
|
|
|
ACTOR static Future<Void> performAction(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<TaskFuture> taskFuture,
|
|
Reference<Task> task) {
|
|
taskFuture->futureBucket->setOptions(tr);
|
|
|
|
if (task && TaskFuncBase::isValidTask(task)) {
|
|
Reference<TaskFuncBase> taskFunc = TaskFuncBase::create(task->params[Task::reservedTaskParamKeyType]);
|
|
if (taskFunc.getPtr()) {
|
|
wait(taskFunc->finish(tr, taskBucket, taskFuture->futureBucket, task));
|
|
}
|
|
}
|
|
|
|
return Void();
|
|
}
|
|
|
|
ACTOR static Future<Void> performAllActions(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<TaskFuture> taskFuture) {
|
|
taskFuture->futureBucket->setOptions(tr);
|
|
|
|
Standalone<RangeResultRef> values = wait(tr->getRange(taskFuture->callbacks.range(), CLIENT_KNOBS->TOO_MANY));
|
|
tr->clear(taskFuture->callbacks.range());
|
|
|
|
std::vector<Future<Void>> actions;
|
|
|
|
if (values.size() != 0) {
|
|
state Reference<Task> task(new Task());
|
|
Key lastTaskID;
|
|
for (auto& s : values) {
|
|
Tuple t = taskFuture->callbacks.unpack(s.key);
|
|
Key taskID = t.getString(0);
|
|
Key key = t.getString(1);
|
|
// If we see a new task ID and the old one isn't empty then process the task accumulated so far and make
|
|
// a new task
|
|
if (taskID.size() != 0 && taskID != lastTaskID) {
|
|
actions.push_back(performAction(tr, taskBucket, taskFuture, task));
|
|
task = makeReference<Task>();
|
|
}
|
|
task->params[key] = s.value;
|
|
lastTaskID = taskID;
|
|
}
|
|
// Process the last task
|
|
actions.push_back(performAction(tr, taskBucket, taskFuture, task));
|
|
}
|
|
|
|
wait(waitForAll(actions));
|
|
|
|
return Void();
|
|
}
|
|
|
|
ACTOR static Future<Void> onSetAddTask(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<TaskFuture> taskFuture,
|
|
Reference<Task> task) {
|
|
taskFuture->futureBucket->setOptions(tr);
|
|
|
|
task->params[Task::reservedTaskParamKeyAddTask] = task->params[Task::reservedTaskParamKeyType];
|
|
task->params[Task::reservedTaskParamKeyType] = LiteralStringRef("AddTask");
|
|
wait(onSet(tr, taskBucket, taskFuture, task));
|
|
|
|
return Void();
|
|
}
|
|
|
|
ACTOR static Future<Void> onSetAddTask(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<TaskFuture> taskFuture,
|
|
Reference<Task> task,
|
|
KeyRef validationKey) {
|
|
taskFuture->futureBucket->setOptions(tr);
|
|
|
|
Optional<Value> validationValue = wait(tr->get(validationKey));
|
|
|
|
if (!validationValue.present()) {
|
|
TraceEvent(SevError, "TaskBucketOnSetAddTaskInvalidKey")
|
|
.detail("Task", task->params[Task::reservedTaskParamKeyType])
|
|
.detail("ValidationKey", validationKey);
|
|
throw invalid_option_value();
|
|
}
|
|
|
|
task->params[Task::reservedTaskParamValidKey] = validationKey;
|
|
task->params[Task::reservedTaskParamValidValue] = validationValue.get();
|
|
|
|
wait(onSetAddTask(tr, taskBucket, taskFuture, task));
|
|
|
|
return Void();
|
|
}
|
|
|
|
static Future<Void> onSetAddTask(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<TaskFuture> taskFuture,
|
|
Reference<Task> task,
|
|
KeyRef validationKey,
|
|
KeyRef validationValue) {
|
|
taskFuture->futureBucket->setOptions(tr);
|
|
|
|
task->params[Task::reservedTaskParamValidKey] = validationKey;
|
|
task->params[Task::reservedTaskParamValidValue] = validationValue;
|
|
|
|
return onSetAddTask(tr, taskBucket, taskFuture, task);
|
|
}
|
|
|
|
ACTOR static Future<Reference<TaskFuture>> joinedFuture(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<TaskFuture> taskFuture) {
|
|
taskFuture->futureBucket->setOptions(tr);
|
|
|
|
std::vector<Reference<TaskFuture>> vectorFuture;
|
|
state Reference<TaskFuture> future = taskFuture->futureBucket->future(tr);
|
|
vectorFuture.push_back(future);
|
|
wait(join(tr, taskBucket, taskFuture, vectorFuture));
|
|
return future;
|
|
}
|
|
};
|
|
|
|
TaskFuture::TaskFuture() {}
|
|
|
|
TaskFuture::TaskFuture(const Reference<FutureBucket> bucket, Key k) : futureBucket(bucket), key(k) {
|
|
if (k.size() == 0) {
|
|
key = deterministicRandom()->randomUniqueID().toString();
|
|
}
|
|
|
|
prefix = futureBucket->prefix.get(key);
|
|
blocks = prefix.get(LiteralStringRef("bl"));
|
|
callbacks = prefix.get(LiteralStringRef("cb"));
|
|
}
|
|
|
|
TaskFuture::~TaskFuture() {}
|
|
|
|
void TaskFuture::addBlock(Reference<ReadYourWritesTransaction> tr, StringRef block_id) {
|
|
tr->set(blocks.pack(block_id), LiteralStringRef(""));
|
|
}
|
|
|
|
Future<Void> TaskFuture::set(Reference<ReadYourWritesTransaction> tr, Reference<TaskBucket> taskBucket) {
|
|
return TaskFutureImpl::set(tr, taskBucket, Reference<TaskFuture>::addRef(this));
|
|
}
|
|
|
|
Future<Void> TaskFuture::performAllActions(Reference<ReadYourWritesTransaction> tr, Reference<TaskBucket> taskBucket) {
|
|
return TaskFutureImpl::performAllActions(tr, taskBucket, Reference<TaskFuture>::addRef(this));
|
|
}
|
|
|
|
Future<Void> TaskFuture::join(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
std::vector<Reference<TaskFuture>> vectorFuture) {
|
|
return TaskFutureImpl::join(tr, taskBucket, Reference<TaskFuture>::addRef(this), vectorFuture);
|
|
}
|
|
|
|
Future<bool> TaskFuture::isSet(Reference<ReadYourWritesTransaction> tr) {
|
|
return TaskFutureImpl::isSet(tr, Reference<TaskFuture>::addRef(this));
|
|
}
|
|
|
|
Future<Void> TaskFuture::onSet(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<Task> task) {
|
|
return TaskFutureImpl::onSet(tr, taskBucket, Reference<TaskFuture>::addRef(this), task);
|
|
}
|
|
|
|
Future<Void> TaskFuture::onSetAddTask(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<Task> task) {
|
|
return TaskFutureImpl::onSetAddTask(tr, taskBucket, Reference<TaskFuture>::addRef(this), task);
|
|
}
|
|
|
|
Future<Void> TaskFuture::onSetAddTask(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<Task> task,
|
|
KeyRef validationKey) {
|
|
return TaskFutureImpl::onSetAddTask(tr, taskBucket, Reference<TaskFuture>::addRef(this), task, validationKey);
|
|
}
|
|
|
|
Future<Void> TaskFuture::onSetAddTask(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket,
|
|
Reference<Task> task,
|
|
KeyRef validationKey,
|
|
KeyRef validationValue) {
|
|
return TaskFutureImpl::onSetAddTask(
|
|
tr, taskBucket, Reference<TaskFuture>::addRef(this), task, validationKey, validationValue);
|
|
}
|
|
|
|
Future<Reference<TaskFuture>> TaskFuture::joinedFuture(Reference<ReadYourWritesTransaction> tr,
|
|
Reference<TaskBucket> taskBucket) {
|
|
return TaskFutureImpl::joinedFuture(tr, taskBucket, Reference<TaskFuture>::addRef(this));
|
|
}
|
|
|
|
ACTOR Future<Key> getCompletionKey(TaskCompletionKey* self, Future<Reference<TaskFuture>> f) {
|
|
Reference<TaskFuture> taskFuture = wait(f);
|
|
self->joinFuture.clear();
|
|
self->key = taskFuture->key;
|
|
return self->key.get();
|
|
}
|
|
|
|
Future<Key> TaskCompletionKey::get(Reference<ReadYourWritesTransaction> tr, Reference<TaskBucket> taskBucket) {
|
|
ASSERT(key.present() == (joinFuture.getPtr() == nullptr));
|
|
return key.present() ? key.get() : getCompletionKey(this, joinFuture->joinedFuture(tr, taskBucket));
|
|
}
|