From 194bdb1d963d25a16791c925aa2ba15f9bff5b13 Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Tue, 8 Nov 2016 14:46:34 -0800 Subject: [PATCH] Compute task IDs and object IDs deterministically. (#31) * Put infrastructure in place to compute task IDs and object IDs. * Fix version number for common library. * Compute task IDs and object IDs deterministically. * Address Stephanie's comments. * Update task documentation. * Fix formatting. * Add more tests and checks. * Fix formatting. * Enable DCHECKs and change CHECKs to DCHECKs. --- lib/python/ray/worker.py | 16 +- src/common/Makefile | 2 +- src/common/common.c | 4 + src/common/common.h | 18 +- src/common/lib/python/common_extension.c | 35 ++-- src/common/lib/python/setup.py | 2 +- src/common/task.c | 171 +++++++++++++++---- src/common/task.h | 206 +++++++++++++++++++---- src/common/test/task_tests.c | 137 +++++++++++++-- src/common/test/test.py | 5 +- src/common/test/test_common.h | 5 +- src/common/thirdparty/sha256.c | 158 +++++++++++++++++ src/common/thirdparty/sha256.h | 34 ++++ src/photon/photon_algorithm.c | 2 +- src/photon/test/test.py | 9 +- 15 files changed, 691 insertions(+), 113 deletions(-) create mode 100644 src/common/thirdparty/sha256.c create mode 100644 src/common/thirdparty/sha256.h diff --git a/lib/python/ray/worker.py b/lib/python/ray/worker.py index f7147db5f..be5f838db 100644 --- a/lib/python/ray/worker.py +++ b/lib/python/ray/worker.py @@ -477,7 +477,14 @@ class Worker(object): args_for_photon.append(put(arg)) # Submit the task to Photon. - task = photon.Task(photon.ObjectID(function_id.id()), args_for_photon, self.num_return_vals[function_id.id()]) + task = photon.Task(photon.ObjectID(function_id.id()), + args_for_photon, + self.num_return_vals[function_id.id()], + self.current_task_id, + self.task_index) + # Increment the worker's task index to track how many tasks have been + # submitted by the current task so far. + self.task_index += 1 self.photon_client.submit(task) return task.returns() @@ -832,6 +839,11 @@ def connect(address_info, mode=WORKER_MODE, worker=global_worker): worker.redis_client.rpush("Workers", worker.worker_id) else: raise Exception("This code should be unreachable.") + # If this is a driver, set the current task ID to a specific fixed value and + # set the task index to 0. + if mode in [SCRIPT_MODE, SILENT_MODE]: + worker.current_task_id = photon.ObjectID("".join(chr(i) for i in range(20))) + worker.task_index = 0 # If this is a worker, then start a thread to import exports from the driver. if mode == WORKER_MODE: t = threading.Thread(target=import_thread, args=(worker,)) @@ -1038,6 +1050,8 @@ def main_loop(worker=global_worker): After the task executes, the worker resets any reusable variables that were accessed by the task. """ + worker.current_task_id = task.task_id() + worker.task_index = 0 function_id = task.function_id() args = task.arguments() return_object_ids = task.returns() diff --git a/src/common/Makefile b/src/common/Makefile index d2efda314..86cab3688 100644 --- a/src/common/Makefile +++ b/src/common/Makefile @@ -4,7 +4,7 @@ BUILD = build all: hiredis $(BUILD)/libcommon.a -$(BUILD)/libcommon.a: event_loop.o common.o task.o io.o state/redis.o state/table.o state/object_table.o state/task_log.o thirdparty/ae/ae.o +$(BUILD)/libcommon.a: event_loop.o common.o task.o io.o state/redis.o state/table.o state/object_table.o state/task_log.o thirdparty/ae/ae.o thirdparty/sha256.o ar rcs $@ $^ $(BUILD)/common_tests: test/common_tests.c $(BUILD)/libcommon.a diff --git a/src/common/common.c b/src/common/common.c index 16ce1471b..fdbe21616 100644 --- a/src/common/common.c +++ b/src/common/common.c @@ -24,6 +24,10 @@ unique_id globally_unique_id(void) { return result; } +bool object_ids_equal(object_id first_id, object_id second_id) { + return memcmp(&first_id, &second_id, sizeof(object_id)) == 0 ? true : false; +} + char *sha1_to_hex(const unsigned char *sha1, char *buffer) { static const char hex[] = "0123456789abcdef"; char *buf = buffer; diff --git a/src/common/common.h b/src/common/common.h index 0c909c405..e60bc7fdc 100644 --- a/src/common/common.h +++ b/src/common/common.h @@ -1,6 +1,7 @@ #ifndef COMMON_H #define COMMON_H +#include #include #include #include @@ -37,9 +38,13 @@ #define CHECK(COND) CHECKM(COND, "") +/* This should be defined if we want to check calls to DCHECK. */ +#define RAY_DCHECK + +#ifdef RAY_DCHECK +#define DCHECK(COND) CHECK(COND) +#else #define DCHECK(COND) -#ifdef RAY_COMMON_DEBUG -CHECK(COND) #endif /* These are exit codes for common errors that can occur in Ray components. */ @@ -67,4 +72,13 @@ char *sha1_to_hex(const unsigned char *sha1, char *buffer); typedef unique_id object_id; +/** + * Compare two object IDs. + * + * @param first_id The first object ID to compare. + * @param second_id The first object ID to compare. + * @return True if the object IDs are the same and false otherwise. + */ +bool object_ids_equal(object_id first_id, object_id second_id); + #endif diff --git a/src/common/lib/python/common_extension.c b/src/common/lib/python/common_extension.c index 14f3292d9..bf994c83e 100644 --- a/src/common/lib/python/common_extension.c +++ b/src/common/lib/python/common_extension.c @@ -120,8 +120,13 @@ static int PyTask_init(PyTask *self, PyObject *args, PyObject *kwds) { UT_array *val_repr_ptrs; utarray_new(val_repr_ptrs, &ut_ptr_icd); int num_returns; - if (!PyArg_ParseTuple(args, "O&Oi", &PyObjectToUniqueID, &function_id, - &arguments, &num_returns)) { + /* The ID of the task that called this task. */ + task_id parent_task_id; + /* The number of tasks that the parent task has called prior to this one. */ + int parent_counter; + if (!PyArg_ParseTuple(args, "O&OiO&i", &PyObjectToUniqueID, &function_id, + &arguments, &num_returns, &PyObjectToUniqueID, + &parent_task_id, &parent_counter)) { return -1; } size_t size = PyList_Size(arguments); @@ -138,7 +143,8 @@ static int PyTask_init(PyTask *self, PyObject *args, PyObject *kwds) { /* Construct the task specification. */ int val_repr_index = 0; self->spec = - alloc_task_spec(function_id, size, num_returns, value_data_bytes); + start_construct_task_spec(parent_task_id, parent_counter, function_id, + size, num_returns, value_data_bytes); /* Add the task arguments. */ for (size_t i = 0; i < size; ++i) { PyObject *arg = PyList_GetItem(arguments, i); @@ -154,14 +160,8 @@ static int PyTask_init(PyTask *self, PyObject *args, PyObject *kwds) { } } utarray_free(val_repr_ptrs); - /* Generate and add the object IDs for the return values. */ - for (size_t i = 0; i < num_returns; ++i) { - /* TODO(rkn): Later, this should be computed as a deterministic hash of (1) - * the contents of the task, (2) the index i, and (3) a counter of the - * number of tasks launched so far by the parent task. For now, we generate - * it randomly. */ - *task_return(self->spec, i) = globally_unique_id(); - } + /* Compute the task ID and the return object IDs. */ + finish_construct_task_spec(self->spec); return 0; } @@ -171,17 +171,22 @@ static void PyTask_dealloc(PyTask *self) { } static PyObject *PyTask_function_id(PyObject *self) { - function_id function_id = *task_function(((PyTask *) self)->spec); + function_id function_id = task_function(((PyTask *) self)->spec); return PyObjectID_make(function_id); } +static PyObject *PyTask_task_id(PyObject *self) { + task_id task_id = task_task_id(((PyTask *) self)->spec); + return PyObjectID_make(task_id); +} + static PyObject *PyTask_arguments(PyObject *self) { int64_t num_args = task_num_args(((PyTask *) self)->spec); PyObject *arg_list = PyList_New((Py_ssize_t) num_args); task_spec *task = ((PyTask *) self)->spec; for (int i = 0; i < num_args; ++i) { if (task_arg_type(task, i) == ARG_BY_REF) { - object_id object_id = *task_arg_id(task, i); + object_id object_id = task_arg_id(task, i); PyList_SetItem(arg_list, i, PyObjectID_make(object_id)); } else { PyObject *s = @@ -198,7 +203,7 @@ static PyObject *PyTask_returns(PyObject *self) { PyObject *return_id_list = PyList_New((Py_ssize_t) num_returns); task_spec *task = ((PyTask *) self)->spec; for (int i = 0; i < num_returns; ++i) { - object_id object_id = *task_return(task, i); + object_id object_id = task_return(task, i); PyList_SetItem(return_id_list, i, PyObjectID_make(object_id)); } return return_id_list; @@ -207,6 +212,8 @@ static PyObject *PyTask_returns(PyObject *self) { static PyMethodDef PyTask_methods[] = { {"function_id", (PyCFunction) PyTask_function_id, METH_NOARGS, "Return the function ID for this task."}, + {"task_id", (PyCFunction) PyTask_task_id, METH_NOARGS, + "Return the task ID for this task."}, {"arguments", (PyCFunction) PyTask_arguments, METH_NOARGS, "Return the arguments for the task."}, {"returns", (PyCFunction) PyTask_returns, METH_NOARGS, diff --git a/src/common/lib/python/setup.py b/src/common/lib/python/setup.py index 8ef8a1abd..2f9f19436 100644 --- a/src/common/lib/python/setup.py +++ b/src/common/lib/python/setup.py @@ -7,6 +7,6 @@ common_module = Extension("common", extra_compile_args=["--std=c99", "-Werror"]) setup(name="Common", - version="0.1", + version="0.01", description="Common library for Ray", ext_modules=[common_module]) diff --git a/src/common/task.c b/src/common/task.c index 0086eb337..57acf5009 100644 --- a/src/common/task.c +++ b/src/common/task.c @@ -2,12 +2,17 @@ #include #include +#include "sha256.h" #include "utarray.h" #include "task.h" #include "common.h" #include "io.h" +const unique_id NIL_TASK_ID = {{255, 255, 255, 255, 255, 255, 255, + 255, 255, 255, 255, 255, 255, 255, + 255, 255, 255, 255, 255, 255}}; + /* TASK SPECIFICATIONS */ /* Tasks are stored in a consecutive chunk of memory, the first @@ -34,21 +39,28 @@ typedef struct { } task_arg; struct task_spec_impl { - /* Function ID of the task. */ + /** Task ID of the task. */ + task_id task_id; + /** Task ID of the parent task. */ + task_id parent_task_id; + /** A count of the number of tasks submitted by the parent task before this + * one. */ + int64_t parent_counter; + /** Function ID of the task. */ function_id function_id; - /* Total number of arguments. */ + /** Total number of arguments. */ int64_t num_args; - /* Index of the last argument that has been constructed. */ + /** Index of the last argument that has been constructed. */ int64_t arg_index; - /* Number of return values. */ + /** Number of return values. */ int64_t num_returns; - /* Number of bytes the pass-by-value arguments are occupying. */ + /** Number of bytes the pass-by-value arguments are occupying. */ int64_t args_value_size; - /* The offset of the number of bytes of pass-by-value data that - * has been written so far, relative to &task_spec->args_and_returns[0] + - * (task_spec->num_args + task_spec->num_returns) * sizeof(task_arg) */ + /** The offset of the number of bytes of pass-by-value data that + * has been written so far, relative to &task_spec->args_and_returns[0] + + * (task_spec->num_args + task_spec->num_returns) * sizeof(task_arg) */ int64_t args_value_offset; - /* Argument and return IDs as well as offsets for pass-by-value args. */ + /** Argument and return IDs as well as offsets for pass-by-value args. */ task_arg args_and_returns[0]; }; @@ -57,28 +69,103 @@ struct task_spec_impl { (sizeof(task_spec) + ((NUM_ARGS) + (NUM_RETURNS)) * sizeof(task_arg) + \ (ARGS_VALUE_SIZE)) -task_spec *alloc_task_spec(function_id function_id, - int64_t num_args, - int64_t num_returns, - int64_t args_value_size) { +bool task_ids_equal(task_id first_id, task_id second_id) { + return memcmp(&first_id, &second_id, sizeof(task_id)) == 0 ? true : false; +} + +bool function_ids_equal(function_id first_id, function_id second_id) { + return memcmp(&first_id, &second_id, sizeof(function_id)) == 0 ? true : false; +} + +task_id *task_return_ptr(task_spec *spec, int64_t return_index) { + DCHECK(0 <= return_index && return_index < spec->num_returns); + task_arg *ret = &spec->args_and_returns[spec->num_args + return_index]; + DCHECK(ret->type == ARG_BY_REF); + return &ret->obj_id; +} + +/* Compute the task ID. This assumes that all of the other fields have been set + * and that the return IDs have not been set. It assumes the task_spec was + * zero-initialized so that uninitialized fields will not make the task ID + * nondeterministic. */ +task_id compute_task_id(task_spec *spec) { + /* Check that the task ID and return ID fields of the task_spec are + * uninitialized. */ + DCHECK(task_ids_equal(spec->task_id, NIL_TASK_ID)); + for (int i = 0; i < spec->num_returns; ++i) { + DCHECK(object_ids_equal(*task_return_ptr(spec, i), NIL_ID)); + } + /* Compute a SHA256 hash of the task_spec. */ + SHA256_CTX ctx; + BYTE buff[SHA256_BLOCK_SIZE]; + sha256_init(&ctx); + sha256_update(&ctx, (BYTE *) spec, task_size(spec)); + sha256_final(&ctx, buff); + /* Create a task ID out of the hash. This will truncate the hash. */ + task_id task_id; + CHECK(sizeof(task_id) <= SHA256_BLOCK_SIZE); + memcpy(&task_id.id, buff, sizeof(task_id)); + return task_id; +} + +object_id compute_return_id(task_id task_id, int64_t return_index) { + /* TODO(rkn): This line requires object and task IDs to be the same size. */ + object_id return_id = (object_id) task_id; + int64_t *first_bytes = (int64_t *) &return_id; + /* XOR the first bytes of the object ID with the return index. We add one so + * the first return ID is not the same as the task ID. */ + *first_bytes = *first_bytes ^ (return_index + 1); + return return_id; +} + +task_spec *start_construct_task_spec(task_id parent_task_id, + int64_t parent_counter, + function_id function_id, + int64_t num_args, + int64_t num_returns, + int64_t args_value_size) { int64_t size = TASK_SPEC_SIZE(num_args, num_returns, args_value_size); task_spec *task = malloc(size); memset(task, 0, size); + task->task_id = NIL_TASK_ID; + task->parent_task_id = parent_task_id; + task->parent_counter = parent_counter; task->function_id = function_id; task->num_args = num_args; task->arg_index = 0; task->num_returns = num_returns; task->args_value_size = args_value_size; + for (int i = 0; i < num_returns; ++i) { + *task_return_ptr(task, i) = NIL_ID; + } return task; } +void finish_construct_task_spec(task_spec *spec) { + /* Check that all of the arguments were added to the task. */ + DCHECK(spec->arg_index == spec->num_args); + spec->task_id = compute_task_id(spec); + /* Set the object IDs for the return values. */ + for (int64_t i = 0; i < spec->num_returns; ++i) { + *task_return_ptr(spec, i) = compute_return_id(spec->task_id, i); + } +} + int64_t task_size(task_spec *spec) { return TASK_SPEC_SIZE(spec->num_args, spec->num_returns, spec->args_value_size); } -unique_id *task_function(task_spec *spec) { - return &spec->function_id; +function_id task_function(task_spec *spec) { + /* Check that the task has been constructed. */ + DCHECK(!task_ids_equal(spec->task_id, NIL_TASK_ID)); + return spec->function_id; +} + +task_id task_task_id(task_spec *spec) { + /* Check that the task has been constructed. */ + DCHECK(!task_ids_equal(spec->task_id, NIL_TASK_ID)); + return spec->task_id; } int64_t task_num_args(task_spec *spec) { @@ -90,34 +177,38 @@ int64_t task_num_returns(task_spec *spec) { } int8_t task_arg_type(task_spec *spec, int64_t arg_index) { - CHECK(0 <= arg_index && arg_index < spec->num_args); + DCHECK(0 <= arg_index && arg_index < spec->num_args); return spec->args_and_returns[arg_index].type; } -object_id *task_arg_id(task_spec *spec, int64_t arg_index) { - CHECK(0 <= arg_index && arg_index < spec->num_args); +object_id task_arg_id(task_spec *spec, int64_t arg_index) { + /* Check that the task has been constructed. */ + DCHECK(!task_ids_equal(spec->task_id, NIL_TASK_ID)); + DCHECK(0 <= arg_index && arg_index < spec->num_args); task_arg *arg = &spec->args_and_returns[arg_index]; - CHECK(arg->type == ARG_BY_REF) - return &arg->obj_id; + DCHECK(arg->type == ARG_BY_REF) + return arg->obj_id; } uint8_t *task_arg_val(task_spec *spec, int64_t arg_index) { - CHECK(0 <= arg_index && arg_index < spec->num_args); + DCHECK(0 <= arg_index && arg_index < spec->num_args); task_arg *arg = &spec->args_and_returns[arg_index]; - CHECK(arg->type == ARG_BY_VAL); + DCHECK(arg->type == ARG_BY_VAL); uint8_t *data = (uint8_t *) &spec->args_and_returns[0]; data += (spec->num_args + spec->num_returns) * sizeof(task_arg); return data + arg->value.offset; } int64_t task_arg_length(task_spec *spec, int64_t arg_index) { - CHECK(0 <= arg_index && arg_index < spec->num_args); + DCHECK(0 <= arg_index && arg_index < spec->num_args); task_arg *arg = &spec->args_and_returns[arg_index]; - CHECK(arg->type == ARG_BY_VAL); + DCHECK(arg->type == ARG_BY_VAL); return arg->value.length; } int64_t task_args_add_ref(task_spec *spec, object_id obj_id) { + /* Check that the task is still under construction. */ + DCHECK(task_ids_equal(spec->task_id, NIL_TASK_ID)); task_arg *arg = &spec->args_and_returns[spec->arg_index]; arg->type = ARG_BY_REF; arg->obj_id = obj_id; @@ -125,28 +216,34 @@ int64_t task_args_add_ref(task_spec *spec, object_id obj_id) { } int64_t task_args_add_val(task_spec *spec, uint8_t *data, int64_t length) { + /* Check that the task is still under construction. */ + DCHECK(task_ids_equal(spec->task_id, NIL_TASK_ID)); task_arg *arg = &spec->args_and_returns[spec->arg_index]; arg->type = ARG_BY_VAL; arg->value.offset = spec->args_value_offset; arg->value.length = length; uint8_t *addr = task_arg_val(spec, spec->arg_index); - CHECK(spec->args_value_offset + length <= spec->args_value_size); - CHECK(spec->arg_index != spec->num_args - 1 || - spec->args_value_offset + length == spec->args_value_size); + DCHECK(spec->args_value_offset + length <= spec->args_value_size); + DCHECK(spec->arg_index != spec->num_args - 1 || + spec->args_value_offset + length == spec->args_value_size); memcpy(addr, data, length); spec->args_value_offset += length; return spec->arg_index++; } -object_id *task_return(task_spec *spec, int64_t ret_index) { - CHECK(0 <= ret_index && ret_index < spec->num_returns); - task_arg *ret = &spec->args_and_returns[spec->num_args + ret_index]; - CHECK(ret->type == ARG_BY_REF); /* No memory corruption. */ - return &ret->obj_id; +object_id task_return(task_spec *spec, int64_t return_index) { + /* Check that the task has been constructed. */ + DCHECK(!task_ids_equal(spec->task_id, NIL_TASK_ID)); + DCHECK(0 <= return_index && return_index < spec->num_returns); + task_arg *ret = &spec->args_and_returns[spec->num_args + return_index]; + DCHECK(ret->type == ARG_BY_REF); + return ret->obj_id; } void free_task_spec(task_spec *spec) { - CHECK(spec->arg_index == spec->num_args); /* Task was fully constructed */ + /* Check that the task has been constructed. */ + DCHECK(!task_ids_equal(spec->task_id, NIL_TASK_ID)); + DCHECK(spec->arg_index == spec->num_args); free(spec); } @@ -155,17 +252,17 @@ void print_task(task_spec *spec, UT_string *output) { * of bytes compared to the id (+ 1 byte for '\0'). */ static char hex[2 * UNIQUE_ID_SIZE + 1]; /* Print function id. */ - sha1_to_hex(&task_function(spec)->id[0], &hex[0]); + sha1_to_hex(&task_function(spec).id[0], &hex[0]); utstring_printf(output, "fun %s ", &hex[0]); /* Print arguments. */ for (int i = 0; i < task_num_args(spec); ++i) { - sha1_to_hex(&task_arg_id(spec, i)->id[0], &hex[0]); + sha1_to_hex(&task_arg_id(spec, i).id[0], &hex[0]); utstring_printf(output, " id:%d %s", i, &hex[0]); } /* Print return ids. */ for (int i = 0; i < task_num_returns(spec); ++i) { - object_id *object_id = task_return(spec, i); - sha1_to_hex(&object_id->id[0], &hex[0]); + object_id object_id = task_return(spec, i); + sha1_to_hex(&object_id.id[0], &hex[0]); utstring_printf(output, " ret:%d %s", i, &hex[0]); } } diff --git a/src/common/task.h b/src/common/task.h index 8d818e3a3..fb2e14fbe 100644 --- a/src/common/task.h +++ b/src/common/task.h @@ -7,6 +7,7 @@ * are relative, so that we can memcpy the datastructure and ship it over the * network without serialization and deserialization. */ +#include #include #include #include "common.h" @@ -14,16 +15,15 @@ typedef unique_id function_id; -/* The task ID is a deterministic hash of the function ID that - * the task executes and the argument IDs or argument values */ +/** The task ID is a deterministic hash of the function ID that the task + * executes and the argument IDs or argument values. */ typedef unique_id task_id; -/* The task instance ID is a globally unique ID generated which - * identifies this particular execution of the task */ +/** The task instance ID is a globally unique ID generated which identifies this + * particular execution of the task. */ typedef unique_id task_iid; -/* The node id is an identifier for the node the task is - * scheduled on */ +/** The node id is an identifier for the node the task is scheduled on. */ typedef unique_id node_id; /* @@ -34,52 +34,186 @@ typedef unique_id node_id; typedef struct task_spec_impl task_spec; -/* If argument is passed by value or reference. */ +/** If argument is passed by value or reference. */ enum arg_type { ARG_BY_REF, ARG_BY_VAL }; +/** + * Compare two task IDs. + * + * @param first_id The first task ID to compare. + * @param second_id The first task ID to compare. + * @return True if the task IDs are the same and false otherwise. + */ +bool task_ids_equal(task_id first_id, task_id second_id); + +/** + * Compare two function IDs. + * + * @param first_id The first function ID to compare. + * @param second_id The first function ID to compare. + * @return True if the function IDs are the same and false otherwise. + */ +bool function_ids_equal(function_id first_id, function_id second_id); + /* Construct and modify task specifications. */ -/* Allocating and initializing a task. */ -task_spec *alloc_task_spec(function_id function_id, - int64_t num_args, - int64_t num_returns, - int64_t args_value_size); +/** + * Begin constructing a task_spec. After this is called, the arguments must be + * added to the task_spec and then finish_construct_task_spec must be called. + * + * @param parent_task_id The task ID of the task that submitted this task. + * @param parent_counter A counter indicating how many tasks were submitted by + * the parent task prior to this one. + * @param function_id The function ID of the function to execute in this task. + * @param num_args The number of arguments that this task has. + * @param num_returns The number of return values that this task has. + * @param args_value_size The total size in bytes of the arguments to this task + ignoring object ID arguments. + * @return The partially constructed task_spec. + */ +task_spec *start_construct_task_spec(task_id parent_task_id, + int64_t parent_counter, + function_id function_id, + int64_t num_args, + int64_t num_returns, + int64_t args_value_size); -/* Size of the task in bytes. */ +/** + * Finish constructing a task_spec. This computes the task ID and the object IDs + * of the task return values. This must be called after all of the arguments + * have been added to the task. + * + * @param spec The task spec whose ID and return object IDs should be computed. + * @return Void. + */ +void finish_construct_task_spec(task_spec *spec); + +/** + * The size of the task in bytes. + * + * @param spec The task_spec in question. + * @return The size of the task_spec in bytes. + */ int64_t task_size(task_spec *spec); -/* Return the function ID of the task. */ -unique_id *task_function(task_spec *spec); +/** + * Return the function ID of the task. + * + * @param spec The task_spec in question. + * @return The function ID of the function to execute in this task. + */ +function_id task_function(task_spec *spec); -/* Getting the number of arguments and returns. */ +/** + * Return the task ID of the task. + * + * @param spec The task_spec in question. + * @return The task ID of the task. + */ +task_id task_task_id(task_spec *spec); + +/** + * Get the number of arguments to this task. + * + * @param spec The task_spec in question. + * @return The number of arguments to this task. + */ int64_t task_num_args(task_spec *spec); + +/** + * Get the number of return values expected from this task. + * + * @param spec The task_spec in question. + * @return The number of return values expected from this task. + */ int64_t task_num_returns(task_spec *spec); -/* Getting task arguments. */ +/** + * Get the type of an argument to this task. It should be either ARG_BY_REF or + * ARG_BY_VAL. + * + * @param spec The task_spec in question. + * @param arg_index The index of the argument in question. + * @return The type of the argument. + */ int8_t task_arg_type(task_spec *spec, int64_t arg_index); -unique_id *task_arg_id(task_spec *spec, int64_t arg_index); + +/** + * Get a particular argument to this task. This assumes the argument is an + * object ID. + * + * @param spec The task_spec in question. + * @param arg_index The index of the argument in question. + * @return The argument at that index. + */ +object_id task_arg_id(task_spec *spec, int64_t arg_index); + +/** + * Get a particular argument to this task. This assumes the argument is a value. + * + * @param spec The task_spec in question. + * @param arg_index The index of the argument in question. + * @return The argument at that index. + */ uint8_t *task_arg_val(task_spec *spec, int64_t arg_index); + +/** + * Get the number of bytes in a particular argument to this task. This assumes + * the argument is a value. + * + * @param spec The task_spec in question. + * @param arg_index The index of the argument in question. + * @return The number of bytes in the argument. + */ int64_t task_arg_length(task_spec *spec, int64_t arg_index); -/* Setting task arguments. Note that this API only allows you to set the - * arguments in their order of appearance. */ +/** + * Set the next task argument. Note that this API only allows you to set the + * arguments in their order of appearance. + * + * @param spec The task_spec in question. + * @param The object ID to set the argument to. + * @return The number of task arguments that have been set before this one. This + * is only used for testing. + */ int64_t task_args_add_ref(task_spec *spec, object_id obj_id); + +/** + * Set the next task argument. Note that this API only allows you to set the + * arguments in their order of appearance. + * + * @param spec The task_spec in question. + * @param The value to set the argument to. + * @param The length of the value to set the argument to. + * @return The number of task arguments that have been set before this one. This + * is only used for testing. + */ int64_t task_args_add_val(task_spec *spec, uint8_t *data, int64_t length); -/* Getting and setting return arguments. Tasks return by reference for now. */ -unique_id *task_return(task_spec *spec, int64_t ret_index); +/** + * Get a particular return object ID of a task. + * + * @param spec The task_spec in question. + * @param return_index The index of the return object ID in question. + * @return The relevant return object ID. + */ +object_id task_return(task_spec *spec, int64_t return_index); -/* Freeing the task datastructure. */ +/** + * Free a task_spec. + * + * @param The task_spec in question. + * @return Void. + */ void free_task_spec(task_spec *spec); -/* Write the task specification to a file or socket. */ -void write_task(int fd, task_spec *spec); - -/* Read the task specification from a file or socket. It is the user's - * responsibility to free the task after it has been used. */ -task_spec *read_task(int fd); - -/* Print task as a humanly readable string. */ +/** + * Print the task as a humanly readable string. + * + * @param spec The task_spec in question. + * @param output The buffer to write the string to. + * @return Void. + */ void print_task(task_spec *spec, UT_string *output); /* @@ -89,8 +223,8 @@ void print_task(task_spec *spec, UT_string *output); * RUNNING, DONE) and which node the task is scheduled on. */ -/* The scheduling_state can be used as a flag when we are listening - * for an event, for example TASK_WAITING | TASK_SCHEDULED. */ +/** The scheduling_state can be used as a flag when we are listening + * for an event, for example TASK_WAITING | TASK_SCHEDULED. */ enum scheduling_state { TASK_STATUS_WAITING = 1, TASK_STATUS_SCHEDULED = 2, @@ -98,9 +232,9 @@ enum scheduling_state { TASK_STATUS_DONE = 8 }; -/* A task instance is one execution of a task specification. - * It has a unique instance id, a state of execution (see scheduling_state) - * and a node it is scheduled on or running on. */ +/** A task instance is one execution of a task specification. It has a unique + * instance id, a state of execution (see scheduling_state) and a node it is + * scheduled on or running on. */ typedef struct task_instance_impl task_instance; /* Allocate and initialize a new task instance. Must be freed with diff --git a/src/common/test/task_tests.c b/src/common/test/task_tests.c index 3a7862455..1acff2264 100644 --- a/src/common/test/task_tests.c +++ b/src/common/test/task_tests.c @@ -12,8 +12,10 @@ SUITE(task_tests); TEST task_test(void) { + task_id parent_task_id = globally_unique_id(); function_id func_id = globally_unique_id(); - task_spec *task = alloc_task_spec(func_id, 4, 2, 10); + task_spec *task = + start_construct_task_spec(parent_task_id, 0, func_id, 4, 2, 10); ASSERT(task_num_args(task) == 4); ASSERT(task_num_returns(task) == 2); @@ -23,30 +25,136 @@ TEST task_test(void) { unique_id arg2 = globally_unique_id(); ASSERT(task_args_add_ref(task, arg2) == 2); ASSERT(task_args_add_val(task, (uint8_t *) "world", 5) == 3); + /* Finish constructing the task. This constructs the task ID and the task + * return IDs. */ + finish_construct_task_spec(task); - unique_id ret0 = globally_unique_id(); - unique_id ret1 = globally_unique_id(); - memcpy(task_return(task, 0), &ret0, sizeof(ret0)); - memcpy(task_return(task, 1), &ret1, sizeof(ret1)); - - ASSERT(memcmp(task_arg_id(task, 0), &arg1, sizeof(arg1)) == 0); + /* Check that the task was constructed as expected. */ + ASSERT(task_num_args(task) == 4); + ASSERT(task_num_returns(task) == 2); + ASSERT(function_ids_equal(task_function(task), func_id)); + ASSERT(object_ids_equal(task_arg_id(task, 0), arg1)); ASSERT(memcmp(task_arg_val(task, 1), (uint8_t *) "hello", task_arg_length(task, 1)) == 0); - ASSERT(memcmp(task_arg_id(task, 2), &arg2, sizeof(arg2)) == 0); + ASSERT(object_ids_equal(task_arg_id(task, 2), arg2)); ASSERT(memcmp(task_arg_val(task, 3), (uint8_t *) "world", task_arg_length(task, 3)) == 0); - ASSERT(memcmp(task_return(task, 0), &ret0, sizeof(unique_id)) == 0); - ASSERT(memcmp(task_return(task, 1), &ret1, sizeof(unique_id)) == 0); - free_task_spec(task); PASS(); } -TEST send_task(void) { +TEST deterministic_ids_test(void) { + /* Define the inputs to the task construction. */ + task_id parent_task_id = globally_unique_id(); function_id func_id = globally_unique_id(); - task_spec *task = alloc_task_spec(func_id, 4, 2, 10); - *task_return(task, 1) = globally_unique_id(); + unique_id arg1 = globally_unique_id(); + uint8_t *arg2 = (uint8_t *) "hello world"; + + /* Construct a first task. */ + task_spec *task1 = + start_construct_task_spec(parent_task_id, 0, func_id, 2, 3, 11); + task_args_add_ref(task1, arg1); + task_args_add_val(task1, arg2, 11); + finish_construct_task_spec(task1); + + /* Construct a second identical task. */ + task_spec *task2 = + start_construct_task_spec(parent_task_id, 0, func_id, 2, 3, 11); + task_args_add_ref(task2, arg1); + task_args_add_val(task2, arg2, 11); + finish_construct_task_spec(task2); + + /* Check that these tasks have the same task IDs and the same return IDs.*/ + ASSERT(task_ids_equal(task_task_id(task1), task_task_id(task2))); + ASSERT(object_ids_equal(task_return(task1, 0), task_return(task2, 0))); + ASSERT(object_ids_equal(task_return(task1, 1), task_return(task2, 1))); + ASSERT(object_ids_equal(task_return(task1, 2), task_return(task2, 2))); + /* Check that the return IDs are all distinct. */ + ASSERT(!object_ids_equal(task_return(task1, 0), task_return(task2, 1))); + ASSERT(!object_ids_equal(task_return(task1, 0), task_return(task2, 2))); + ASSERT(!object_ids_equal(task_return(task1, 1), task_return(task2, 2))); + + /* Create more tasks that are only mildly different. */ + + /* Construct a task with a different parent task ID. */ + task_spec *task3 = + start_construct_task_spec(globally_unique_id(), 0, func_id, 2, 3, 11); + task_args_add_ref(task3, arg1); + task_args_add_val(task3, arg2, 11); + finish_construct_task_spec(task3); + + /* Construct a task with a different parent counter. */ + task_spec *task4 = + start_construct_task_spec(parent_task_id, 1, func_id, 2, 3, 11); + task_args_add_ref(task4, arg1); + task_args_add_val(task4, arg2, 11); + finish_construct_task_spec(task4); + + /* Construct a task with a different function ID. */ + task_spec *task5 = start_construct_task_spec(parent_task_id, 0, + globally_unique_id(), 2, 3, 11); + task_args_add_ref(task5, arg1); + task_args_add_val(task5, arg2, 11); + finish_construct_task_spec(task5); + + /* Construct a task with a different object ID argument. */ + task_spec *task6 = + start_construct_task_spec(parent_task_id, 0, func_id, 2, 3, 11); + task_args_add_ref(task6, globally_unique_id()); + task_args_add_val(task6, arg2, 11); + finish_construct_task_spec(task6); + + /* Construct a task with a different value argument. */ + task_spec *task7 = + start_construct_task_spec(parent_task_id, 0, func_id, 2, 3, 11); + task_args_add_ref(task7, arg1); + task_args_add_val(task7, (uint8_t *) "hello_world", 11); + finish_construct_task_spec(task7); + + /* Check that the task IDs are all distinct from the original. */ + ASSERT(!task_ids_equal(task_task_id(task1), task_task_id(task3))); + ASSERT(!task_ids_equal(task_task_id(task1), task_task_id(task4))); + ASSERT(!task_ids_equal(task_task_id(task1), task_task_id(task5))); + ASSERT(!task_ids_equal(task_task_id(task1), task_task_id(task6))); + ASSERT(!task_ids_equal(task_task_id(task1), task_task_id(task7))); + + /* Check that the return object IDs are distinct from the originals. */ + task_spec *tasks[6] = {task1, task3, task4, task5, task6, task7}; + for (int task_index1 = 0; task_index1 < 6; ++task_index1) { + for (int return_index1 = 0; return_index1 < 3; ++return_index1) { + for (int task_index2 = 0; task_index2 < 6; ++task_index2) { + for (int return_index2 = 0; return_index2 < 3; ++return_index2) { + if (task_index1 != task_index2 && return_index1 != return_index2) { + ASSERT(!object_ids_equal( + task_return(tasks[task_index1], return_index1), + task_return(tasks[task_index2], return_index2))); + } + } + } + } + } + + free_task_spec(task1); + free_task_spec(task2); + free_task_spec(task3); + free_task_spec(task4); + free_task_spec(task5); + free_task_spec(task6); + free_task_spec(task7); + PASS(); +} + +TEST send_task(void) { + task_id parent_task_id = globally_unique_id(); + function_id func_id = globally_unique_id(); + task_spec *task = + start_construct_task_spec(parent_task_id, 0, func_id, 4, 2, 10); + task_args_add_ref(task, globally_unique_id()); + task_args_add_val(task, (uint8_t *) "Hello", 5); + task_args_add_val(task, (uint8_t *) "World", 5); + task_args_add_ref(task, globally_unique_id()); + finish_construct_task_spec(task); int fd[2]; socketpair(AF_UNIX, SOCK_STREAM, 0, fd); write_message(fd[0], SUBMIT_TASK, task_size(task), (uint8_t *) task); @@ -65,6 +173,7 @@ TEST send_task(void) { SUITE(task_tests) { RUN_TEST(task_test); + RUN_TEST(deterministic_ids_test); RUN_TEST(send_task); } diff --git a/src/common/test/test.py b/src/common/test/test.py index 359e8c030..cede0bc34 100644 --- a/src/common/test/test.py +++ b/src/common/test/test.py @@ -74,7 +74,8 @@ class TestTask(unittest.TestCase): def test_create_task(self): # TODO(rkn): The function ID should be a FunctionID object, not an ObjectID. - function_id = common.ObjectID(20 * "a") + parent_id = common.ObjectID(20 * "a") + function_id = common.ObjectID(20 * "b") object_ids = [common.ObjectID(20 * chr(i)) for i in range(256)] args_list = [ [], @@ -104,7 +105,7 @@ class TestTask(unittest.TestCase): ] for args in args_list: for num_return_vals in [0, 1, 2, 3, 5, 10, 100]: - task = common.Task(function_id, args, num_return_vals) + task = common.Task(function_id, args, num_return_vals, parent_id, 0) self.assertEqual(function_id.id(), task.function_id().id()) retrieved_args = task.arguments() self.assertEqual(num_return_vals, len(task.returns())) diff --git a/src/common/test/test_common.h b/src/common/test/test_common.h index 305d74446..fbcc3f35f 100644 --- a/src/common/test/test_common.h +++ b/src/common/test/test_common.h @@ -6,10 +6,13 @@ #include "task.h" task_spec *example_task(void) { + task_id parent_task_id = globally_unique_id(); function_id func_id = globally_unique_id(); - task_spec *task = alloc_task_spec(func_id, 2, 1, 0); + task_spec *task = + start_construct_task_spec(parent_task_id, 0, func_id, 2, 1, 0); task_args_add_ref(task, globally_unique_id()); task_args_add_ref(task, globally_unique_id()); + finish_construct_task_spec(task); return task; } diff --git a/src/common/thirdparty/sha256.c b/src/common/thirdparty/sha256.c new file mode 100644 index 000000000..eb9c5c073 --- /dev/null +++ b/src/common/thirdparty/sha256.c @@ -0,0 +1,158 @@ +/********************************************************************* +* Filename: sha256.c +* Author: Brad Conte (brad AT bradconte.com) +* Copyright: +* Disclaimer: This code is presented "as is" without any guarantees. +* Details: Implementation of the SHA-256 hashing algorithm. + SHA-256 is one of the three algorithms in the SHA2 + specification. The others, SHA-384 and SHA-512, are not + offered in this implementation. + Algorithm specification can be found here: + * http://csrc.nist.gov/publications/fips/fips180-2/fips180-2withchangenotice.pdf + This implementation uses little endian byte order. +*********************************************************************/ + +/*************************** HEADER FILES ***************************/ +#include +#include +#include "sha256.h" + +/****************************** MACROS ******************************/ +#define ROTLEFT(a,b) (((a) << (b)) | ((a) >> (32-(b)))) +#define ROTRIGHT(a,b) (((a) >> (b)) | ((a) << (32-(b)))) + +#define CH(x,y,z) (((x) & (y)) ^ (~(x) & (z))) +#define MAJ(x,y,z) (((x) & (y)) ^ ((x) & (z)) ^ ((y) & (z))) +#define EP0(x) (ROTRIGHT(x,2) ^ ROTRIGHT(x,13) ^ ROTRIGHT(x,22)) +#define EP1(x) (ROTRIGHT(x,6) ^ ROTRIGHT(x,11) ^ ROTRIGHT(x,25)) +#define SIG0(x) (ROTRIGHT(x,7) ^ ROTRIGHT(x,18) ^ ((x) >> 3)) +#define SIG1(x) (ROTRIGHT(x,17) ^ ROTRIGHT(x,19) ^ ((x) >> 10)) + +/**************************** VARIABLES *****************************/ +static const WORD k[64] = { + 0x428a2f98,0x71374491,0xb5c0fbcf,0xe9b5dba5,0x3956c25b,0x59f111f1,0x923f82a4,0xab1c5ed5, + 0xd807aa98,0x12835b01,0x243185be,0x550c7dc3,0x72be5d74,0x80deb1fe,0x9bdc06a7,0xc19bf174, + 0xe49b69c1,0xefbe4786,0x0fc19dc6,0x240ca1cc,0x2de92c6f,0x4a7484aa,0x5cb0a9dc,0x76f988da, + 0x983e5152,0xa831c66d,0xb00327c8,0xbf597fc7,0xc6e00bf3,0xd5a79147,0x06ca6351,0x14292967, + 0x27b70a85,0x2e1b2138,0x4d2c6dfc,0x53380d13,0x650a7354,0x766a0abb,0x81c2c92e,0x92722c85, + 0xa2bfe8a1,0xa81a664b,0xc24b8b70,0xc76c51a3,0xd192e819,0xd6990624,0xf40e3585,0x106aa070, + 0x19a4c116,0x1e376c08,0x2748774c,0x34b0bcb5,0x391c0cb3,0x4ed8aa4a,0x5b9cca4f,0x682e6ff3, + 0x748f82ee,0x78a5636f,0x84c87814,0x8cc70208,0x90befffa,0xa4506ceb,0xbef9a3f7,0xc67178f2 +}; + +/*********************** FUNCTION DEFINITIONS ***********************/ +void sha256_transform(SHA256_CTX *ctx, const BYTE data[]) +{ + WORD a, b, c, d, e, f, g, h, i, j, t1, t2, m[64]; + + for (i = 0, j = 0; i < 16; ++i, j += 4) + m[i] = (data[j] << 24) | (data[j + 1] << 16) | (data[j + 2] << 8) | (data[j + 3]); + for ( ; i < 64; ++i) + m[i] = SIG1(m[i - 2]) + m[i - 7] + SIG0(m[i - 15]) + m[i - 16]; + + a = ctx->state[0]; + b = ctx->state[1]; + c = ctx->state[2]; + d = ctx->state[3]; + e = ctx->state[4]; + f = ctx->state[5]; + g = ctx->state[6]; + h = ctx->state[7]; + + for (i = 0; i < 64; ++i) { + t1 = h + EP1(e) + CH(e,f,g) + k[i] + m[i]; + t2 = EP0(a) + MAJ(a,b,c); + h = g; + g = f; + f = e; + e = d + t1; + d = c; + c = b; + b = a; + a = t1 + t2; + } + + ctx->state[0] += a; + ctx->state[1] += b; + ctx->state[2] += c; + ctx->state[3] += d; + ctx->state[4] += e; + ctx->state[5] += f; + ctx->state[6] += g; + ctx->state[7] += h; +} + +void sha256_init(SHA256_CTX *ctx) +{ + ctx->datalen = 0; + ctx->bitlen = 0; + ctx->state[0] = 0x6a09e667; + ctx->state[1] = 0xbb67ae85; + ctx->state[2] = 0x3c6ef372; + ctx->state[3] = 0xa54ff53a; + ctx->state[4] = 0x510e527f; + ctx->state[5] = 0x9b05688c; + ctx->state[6] = 0x1f83d9ab; + ctx->state[7] = 0x5be0cd19; +} + +void sha256_update(SHA256_CTX *ctx, const BYTE data[], size_t len) +{ + WORD i; + + for (i = 0; i < len; ++i) { + ctx->data[ctx->datalen] = data[i]; + ctx->datalen++; + if (ctx->datalen == 64) { + sha256_transform(ctx, ctx->data); + ctx->bitlen += 512; + ctx->datalen = 0; + } + } +} + +void sha256_final(SHA256_CTX *ctx, BYTE hash[]) +{ + WORD i; + + i = ctx->datalen; + + // Pad whatever data is left in the buffer. + if (ctx->datalen < 56) { + ctx->data[i++] = 0x80; + while (i < 56) + ctx->data[i++] = 0x00; + } + else { + ctx->data[i++] = 0x80; + while (i < 64) + ctx->data[i++] = 0x00; + sha256_transform(ctx, ctx->data); + memset(ctx->data, 0, 56); + } + + // Append to the padding the total message's length in bits and transform. + ctx->bitlen += ctx->datalen * 8; + ctx->data[63] = ctx->bitlen; + ctx->data[62] = ctx->bitlen >> 8; + ctx->data[61] = ctx->bitlen >> 16; + ctx->data[60] = ctx->bitlen >> 24; + ctx->data[59] = ctx->bitlen >> 32; + ctx->data[58] = ctx->bitlen >> 40; + ctx->data[57] = ctx->bitlen >> 48; + ctx->data[56] = ctx->bitlen >> 56; + sha256_transform(ctx, ctx->data); + + // Since this implementation uses little endian byte ordering and SHA uses big endian, + // reverse all the bytes when copying the final state to the output hash. + for (i = 0; i < 4; ++i) { + hash[i] = (ctx->state[0] >> (24 - i * 8)) & 0x000000ff; + hash[i + 4] = (ctx->state[1] >> (24 - i * 8)) & 0x000000ff; + hash[i + 8] = (ctx->state[2] >> (24 - i * 8)) & 0x000000ff; + hash[i + 12] = (ctx->state[3] >> (24 - i * 8)) & 0x000000ff; + hash[i + 16] = (ctx->state[4] >> (24 - i * 8)) & 0x000000ff; + hash[i + 20] = (ctx->state[5] >> (24 - i * 8)) & 0x000000ff; + hash[i + 24] = (ctx->state[6] >> (24 - i * 8)) & 0x000000ff; + hash[i + 28] = (ctx->state[7] >> (24 - i * 8)) & 0x000000ff; + } +} diff --git a/src/common/thirdparty/sha256.h b/src/common/thirdparty/sha256.h new file mode 100644 index 000000000..7123a30dd --- /dev/null +++ b/src/common/thirdparty/sha256.h @@ -0,0 +1,34 @@ +/********************************************************************* +* Filename: sha256.h +* Author: Brad Conte (brad AT bradconte.com) +* Copyright: +* Disclaimer: This code is presented "as is" without any guarantees. +* Details: Defines the API for the corresponding SHA1 implementation. +*********************************************************************/ + +#ifndef SHA256_H +#define SHA256_H + +/*************************** HEADER FILES ***************************/ +#include + +/****************************** MACROS ******************************/ +#define SHA256_BLOCK_SIZE 32 // SHA256 outputs a 32 byte digest + +/**************************** DATA TYPES ****************************/ +typedef unsigned char BYTE; // 8-bit byte +typedef unsigned int WORD; // 32-bit word, change to "long" for 16-bit machines + +typedef struct { + BYTE data[64]; + WORD datalen; + unsigned long long bitlen; + WORD state[8]; +} SHA256_CTX; + +/*********************** FUNCTION DECLARATIONS **********************/ +void sha256_init(SHA256_CTX *ctx); +void sha256_update(SHA256_CTX *ctx, const BYTE data[], size_t len); +void sha256_final(SHA256_CTX *ctx, BYTE hash[]); + +#endif // SHA256_H diff --git a/src/photon/photon_algorithm.c b/src/photon/photon_algorithm.c index bd177ea8d..d7a52bbfa 100644 --- a/src/photon/photon_algorithm.c +++ b/src/photon/photon_algorithm.c @@ -72,7 +72,7 @@ bool can_run(scheduler_state *s, task_spec *task) { int64_t num_args = task_num_args(task); for (int i = 0; i < num_args; ++i) { if (task_arg_type(task, i) == ARG_BY_REF) { - object_id obj_id = *task_arg_id(task, i); + object_id obj_id = task_arg_id(task, i); available_object *entry; HASH_FIND(handle, s->local_objects, &obj_id, sizeof(object_id), entry); if (entry == NULL) { diff --git a/src/photon/test/test.py b/src/photon/test/test.py index 8bcbb0807..4a8dc2c1d 100644 --- a/src/photon/test/test.py +++ b/src/photon/test/test.py @@ -15,6 +15,9 @@ import plasma USE_VALGRIND = False PLASMA_STORE_MEMORY = 1000000000 +def random_object_id(): + return photon.ObjectID("".join([chr(random.randint(0, 255)) for _ in range(plasma.PLASMA_ID_SIZE)])) + class TestPhotonClient(unittest.TestCase): def setUp(self): @@ -95,7 +98,7 @@ class TestPhotonClient(unittest.TestCase): for args in args_list: for num_return_vals in [0, 1, 2, 3, 5, 10, 100]: - task = photon.Task(function_id, args, num_return_vals) + task = photon.Task(function_id, args, num_return_vals, random_object_id(), 0) # Submit a task. self.photon_client.submit(task) # Get the task. @@ -114,7 +117,7 @@ class TestPhotonClient(unittest.TestCase): # Submit all of the tasks. for args in args_list: for num_return_vals in [0, 1, 2, 3, 5, 10, 100]: - task = photon.Task(function_id, args, num_return_vals) + task = photon.Task(function_id, args, num_return_vals, random_object_id(), 0) self.photon_client.submit(task) # Get all of the tasks. for args in args_list: @@ -126,7 +129,7 @@ class TestPhotonClient(unittest.TestCase): object_id = photon.ObjectID(20 * chr(0)) # TODO(rkn): This should be a FunctionID. function_id = photon.ObjectID(20 * "a") - task = photon.Task(function_id, [object_id], 0) + task = photon.Task(function_id, [object_id], 0, random_object_id(), 0) self.photon_client.submit(task) # Launch a thread to get the task. def get_task():