/*
 * 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 { return name; };
	Future<Void> execute(Database cx, Reference<TaskBucket> tb, Reference<FutureBucket> fb, Reference<Task> task) { return Void(); };
	Future<Void> finish(Reference<ReadYourWritesTransaction> tr, Reference<TaskBucket> tb, Reference<FutureBucket> fb, Reference<Task> task) { 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 { return name; };
	Future<Void> execute(Database cx, Reference<TaskBucket> tb, Reference<FutureBucket> fb, Reference<Task> task) { return Void(); };
	Future<Void> finish(Reference<ReadYourWritesTransaction> tr, Reference<TaskBucket> tb, Reference<FutureBucket> fb, Reference<Task> task) { 
		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 { return name; };
	Future<Void> execute(Database cx, Reference<TaskBucket> tb, Reference<FutureBucket> fb, Reference<Task> task) { return Void(); };
	Future<Void> finish(Reference<ReadYourWritesTransaction> tr, Reference<TaskBucket> tb, Reference<FutureBucket> fb, Reference<Task> task) { 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;
		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));
}