From 3c6686db08bd67e43641ff29ee30513165412919 Mon Sep 17 00:00:00 2001 From: Stephanie Wang Date: Mon, 23 Jan 2017 19:44:15 -0800 Subject: [PATCH] Photon optimizations (#219) * Optimizations: - Track mapping of missing object to dependent tasks to avoid iterating over task queue - Perform all fetch requests for missing objects using the same timer * Fix bug and add regression test * Record task dependencies and active fetch requests in the same hash table * fix typo * Fix memory leak and add test cases for scheduling when dependencies are evicted * Fix python3 test case * Minor details. --- python/photon/test/test.py | 55 +++++- src/photon/photon_algorithm.c | 323 +++++++++++++++++++++------------ src/photon/photon_algorithm.h | 12 ++ src/photon/photon_scheduler.c | 3 + src/photon/test/photon_tests.c | 77 +++++++- 5 files changed, 348 insertions(+), 122 deletions(-) diff --git a/python/photon/test/test.py b/python/photon/test/test.py index 3d6bdb0be..07c366484 100644 --- a/python/photon/test/test.py +++ b/python/photon/test/test.py @@ -139,11 +139,62 @@ class TestPhotonClient(unittest.TestCase): # Wait until the thread finishes so that we know the task was scheduled. t.join() + def test_scheduling_when_objects_evicted(self): + # Create a task with two dependencies and submit it. + object_id1 = random_object_id() + object_id2 = random_object_id() + task = photon.Task(random_function_id(), [object_id1, object_id2], 0, random_task_id(), 0) + self.photon_client.submit(task) + + # Launch a thread to get the task. + def get_task(): + self.photon_client.get_task() + t = threading.Thread(target=get_task) + t.start() + + # Make one of the dependencies available. + self.plasma_client.create(object_id1.id(), 1) + self.plasma_client.seal(object_id1.id()) + # Check that the thread is still waiting for a task. + time.sleep(0.1) + self.assertTrue(t.is_alive()) + + # Force eviction of the first dependency. + num_objects = 4 + object_size = plasma.DEFAULT_PLASMA_STORE_MEMORY // num_objects + for i in range(num_objects + 1): + object_id = random_object_id() + self.plasma_client.create(object_id.id(), object_size) + self.plasma_client.seal(object_id.id()) + # Check that the thread is still waiting for a task. + time.sleep(0.1) + self.assertTrue(t.is_alive()) + # Check that the first object dependency was evicted. + object1 = self.plasma_client.get([object_id1.id()], timeout_ms=0) + self.assertEqual(object1, [None]) + # Check that the thread is still waiting for a task. + time.sleep(0.1) + self.assertTrue(t.is_alive()) + + # Create the second dependency. + self.plasma_client.create(object_id2.id(), 1) + self.plasma_client.seal(object_id2.id()) + # Check that the thread is still waiting for a task. + time.sleep(0.1) + self.assertTrue(t.is_alive()) + + # Create the first dependency again. Both dependencies are now available. + self.plasma_client.create(object_id1.id(), 1) + self.plasma_client.seal(object_id1.id()) + + # Wait until the thread finishes so that we know the task was scheduled. + t.join() + if __name__ == "__main__": if len(sys.argv) > 1: # pop the argument so we don't mess with unittest's own argument parser - arg = sys.argv.pop() - if arg == "valgrind": + if sys.argv[-1] == "valgrind": + arg = sys.argv.pop() USE_VALGRIND = True print("Using valgrind for tests") unittest.main(verbosity=2) diff --git a/src/photon/photon_algorithm.c b/src/photon/photon_algorithm.c index 471f1199d..7dba09b77 100644 --- a/src/photon/photon_algorithm.c +++ b/src/photon/photon_algorithm.c @@ -17,26 +17,21 @@ typedef struct task_queue_entry { struct task_queue_entry *next; } task_queue_entry; +/** A data structure used to track which objects are available locally and + * which objects are being actively fetched. */ typedef struct { /** Object id of this object. */ object_id object_id; - /** Handle for the uthash table. */ - UT_hash_handle handle; -} available_object; - -/** A data structure used to track which objects are being fetched. */ -typedef struct { - /** The object ID that we are trying to fetch. */ - object_id object_id; - /** The local scheduler state. */ - local_scheduler_state *state; - /** The scheduling algorithm state. */ - scheduling_algorithm_state *algorithm_state; - /** The ID for the timer that will time out the current request. */ - int64_t timer; - /** Handle for the uthash table. */ + /** An array of the tasks dependent on this object. */ + UT_array *dependent_tasks; + /** Handle for the uthash table. NOTE: This handle is used for both the + * scheduling algorithm state's local_objects and remote_objects tables. + * We must enforce the uthash invariant that the entry be in at most one of + * the tables. */ UT_hash_handle hh; -} fetch_object_request; +} object_entry; + +UT_icd task_queue_entry_icd = {sizeof(task_queue_entry *), NULL, NULL, NULL}; /** Part of the photon state that is maintained by the scheduling algorithm. */ struct scheduling_algorithm_state { @@ -49,11 +44,14 @@ struct scheduling_algorithm_state { * waiting for tasks. */ UT_array *available_workers; /** A hash map of the objects that are available in the local Plasma store. - * This information could be a little stale. */ - available_object *local_objects; - /** A hash map of the objects that are currently being fetched by this local - * scheduler. The key is the object ID. */ - fetch_object_request *fetch_requests; + * The key is the object ID. This information could be a little stale. */ + object_entry *local_objects; + /** A hash map of the objects that are not available locally. These are + * currently being fetched by this local scheduler. The key is the object + * ID. Every LOCAL_SCHEDULER_FETCH_TIMEOUT_MILLISECONDS, a Plasma fetch + * request will be sent the object IDs in this table. Each entry also holds + * an array of queued tasks that are dependent on it. */ + object_entry *remote_objects; }; scheduling_algorithm_state *make_scheduling_algorithm_state(void) { @@ -61,12 +59,12 @@ scheduling_algorithm_state *make_scheduling_algorithm_state(void) { malloc(sizeof(scheduling_algorithm_state)); /* Initialize an empty hash map for the cache of local available objects. */ algorithm_state->local_objects = NULL; + /* Initialize the hash table of objects being fetched. */ + algorithm_state->remote_objects = NULL; /* Initialize the local data structures used for queuing tasks and workers. */ algorithm_state->waiting_task_queue = NULL; algorithm_state->dispatch_task_queue = NULL; utarray_new(algorithm_state->available_workers, &ut_int_icd); - /* Initialize the hash table of objects being fetched. */ - algorithm_state->fetch_requests = NULL; return algorithm_state; } @@ -84,15 +82,16 @@ void free_scheduling_algorithm_state( free(elt); } utarray_free(algorithm_state->available_workers); - available_object *available_obj, *tmp2; - HASH_ITER(handle, algorithm_state->local_objects, available_obj, tmp2) { - HASH_DELETE(handle, algorithm_state->local_objects, available_obj); - free(available_obj); + object_entry *obj_entry, *tmp_obj_entry; + HASH_ITER(hh, algorithm_state->local_objects, obj_entry, tmp_obj_entry) { + HASH_DELETE(hh, algorithm_state->local_objects, obj_entry); + CHECK(obj_entry->dependent_tasks == NULL); + free(obj_entry); } - fetch_object_request *fetch_elt, *tmp_fetch_elt; - HASH_ITER(hh, algorithm_state->fetch_requests, fetch_elt, tmp_fetch_elt) { - HASH_DELETE(hh, algorithm_state->fetch_requests, fetch_elt); - free(fetch_elt); + HASH_ITER(hh, algorithm_state->remote_objects, obj_entry, tmp_obj_entry) { + HASH_DELETE(hh, algorithm_state->remote_objects, obj_entry); + utarray_free(obj_entry->dependent_tasks); + free(obj_entry); } free(algorithm_state); } @@ -114,22 +113,93 @@ void provide_scheduler_info(local_scheduler_state *state, info->available_workers = utarray_len(algorithm_state->available_workers); } +/** + * Fetch a queued task's missing object dependency. The fetch request will be + * retried every LOCAL_SCHEDULER_FETCH_TIMEOUT_MILLISECONDS until the object is + * available locally. + * + * @param state The scheduler state. + * @param algorithm_state The scheduling algorithm state. + * @param task_entry The task's queue entry. + * @param obj_id The ID of the object that the task is dependent on. + * @returns Void. + */ +void fetch_missing_dependency(local_scheduler_state *state, + scheduling_algorithm_state *algorithm_state, + task_queue_entry *task_entry, + object_id obj_id) { + object_entry *entry; + HASH_FIND(hh, algorithm_state->remote_objects, &obj_id, sizeof(obj_id), + entry); + if (entry == NULL) { + /* We weren't actively fetching this object. Try the fetch once + * immediately. */ + if (plasma_manager_is_connected(state->plasma_conn)) { + plasma_fetch(state->plasma_conn, 1, &obj_id); + } + /* Create an entry and add it to the list of active fetch requests to + * ensure that the fetch actually happens. The entry will be moved to the + * hash table of locally available objects in handle_object_available when + * the object becomes available locally. It will get freed if the object is + * subsequently removed locally. */ + entry = malloc(sizeof(object_entry)); + entry->object_id = obj_id; + utarray_new(entry->dependent_tasks, &task_queue_entry_icd); + HASH_ADD(hh, algorithm_state->remote_objects, object_id, + sizeof(entry->object_id), entry); + } + utarray_push_back(entry->dependent_tasks, &task_entry); +} + +/** + * Fetch a queued task's missing object dependencies. The fetch requests will + * be retried every LOCAL_SCHEDULER_FETCH_TIMEOUT_MILLISECONDS until all + * objects are available locally. + * + * @param state The scheduler state. + * @param algorithm_state The scheduling algorithm state. + * @param task_entry The task's queue entry. + * @returns Void. + */ +void fetch_missing_dependencies(local_scheduler_state *state, + scheduling_algorithm_state *algorithm_state, + task_queue_entry *task_entry) { + task_spec *task = task_entry->spec; + int64_t num_args = task_num_args(task); + int num_missing_dependencies = 0; + 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_entry *entry; + HASH_FIND(hh, algorithm_state->local_objects, &obj_id, sizeof(obj_id), + entry); + if (entry == NULL) { + /* If the entry is not yet available locally, record the dependency. */ + fetch_missing_dependency(state, algorithm_state, task_entry, obj_id); + ++num_missing_dependencies; + } + } + } + CHECK(num_missing_dependencies > 0); +} + /** * Check if all of the remote object arguments for a task are available in the * local object store. * - * @param s The scheduler state. + * @param algorithm_state The scheduling algorithm state. * @param task Task specification of the task to check. - * @return This returns 1 if all of the remote object arguments for the task are - * present in the local object store, otherwise it returns 0. + * @return bool This returns true if all of the remote object arguments for the + * task are present in the local object store, otherwise it returns + * false. */ bool can_run(scheduling_algorithm_state *algorithm_state, 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); - available_object *entry; - HASH_FIND(handle, algorithm_state->local_objects, &obj_id, sizeof(obj_id), + object_entry *entry; + HASH_FIND(hh, algorithm_state->local_objects, &obj_id, sizeof(obj_id), entry); if (entry == NULL) { /* The object is not present locally, so this task cannot be scheduled @@ -142,43 +212,29 @@ bool can_run(scheduling_algorithm_state *algorithm_state, task_spec *task) { } /* TODO(rkn): This method will need to be changed to call reconstruct. */ +/* TODO(swang): This method is not covered by any valgrind tests. */ int fetch_object_timeout_handler(event_loop *loop, timer_id id, void *context) { - fetch_object_request *fetch_req = (fetch_object_request *) context; - object_id object_ids[1] = {fetch_req->object_id}; - plasma_fetch(fetch_req->state->plasma_conn, 1, object_ids); - return LOCAL_SCHEDULER_FETCH_TIMEOUT_MILLISECONDS; -} - -void fetch_missing_dependencies(local_scheduler_state *state, - scheduling_algorithm_state *algorithm_state, - task_spec *spec) { - int64_t num_args = task_num_args(spec); - for (int i = 0; i < num_args; ++i) { - if (task_arg_type(spec, i) == ARG_BY_REF) { - object_id obj_id = task_arg_id(spec, i); - available_object *entry; - HASH_FIND(handle, algorithm_state->local_objects, &obj_id, sizeof(obj_id), - entry); - if (entry == NULL) { - /* The object is not present locally, fetch the object. */ - object_id object_ids[1] = {obj_id}; - plasma_fetch(state->plasma_conn, 1, object_ids); - /* Create a fetch request and add a timer to the event loop to ensure - * that the fetch actually happens. */ - fetch_object_request *fetch_req = malloc(sizeof(fetch_object_request)); - fetch_req->object_id = obj_id; - fetch_req->state = state; - fetch_req->algorithm_state = algorithm_state; - fetch_req->timer = event_loop_add_timer( - state->loop, LOCAL_SCHEDULER_FETCH_TIMEOUT_MILLISECONDS, - fetch_object_timeout_handler, fetch_req); - /* The fetch request will be freed and removed from the hash table in - * handle_object_available when the object becomes available locally. */ - HASH_ADD(hh, algorithm_state->fetch_requests, object_id, - sizeof(fetch_req->object_id), fetch_req); - } - } + local_scheduler_state *state = context; + /* Only try the fetches if we are connected to the object store manager. */ + if (!plasma_manager_is_connected(state->plasma_conn)) { + LOG_INFO("Local scheduler is not connected to a object store manager"); + return LOCAL_SCHEDULER_FETCH_TIMEOUT_MILLISECONDS; } + + /* Allocate a buffer to hold all the object IDs for active fetch requests. */ + int num_object_ids = HASH_COUNT(state->algorithm_state->remote_objects); + object_id *object_ids = malloc(num_object_ids * sizeof(object_id)); + + /* Fill out the request with the object IDs for active fetches. */ + object_entry *fetch_request, *tmp; + int i = 0; + HASH_ITER(hh, state->algorithm_state->remote_objects, fetch_request, tmp) { + object_ids[i] = fetch_request->object_id; + ++i; + } + plasma_fetch(state->plasma_conn, num_object_ids, object_ids); + free(object_ids); + return LOCAL_SCHEDULER_FETCH_TIMEOUT_MILLISECONDS; } /** @@ -225,10 +281,10 @@ void dispatch_tasks(local_scheduler_state *state, * scheduler. If false, the task was submitted by a worker. * @return Void. */ -void queue_task(local_scheduler_state *state, - task_queue_entry **task_queue, - task_spec *spec, - bool from_global_scheduler) { +task_queue_entry *queue_task(local_scheduler_state *state, + task_queue_entry **task_queue, + task_spec *spec, + bool from_global_scheduler) { /* Copy the spec and add it to the task queue. The allocated spec will be * freed when it is assigned to a worker. */ task_queue_entry *elt = malloc(sizeof(task_queue_entry)); @@ -253,6 +309,8 @@ void queue_task(local_scheduler_state *state, NULL); } } + + return elt; } /** @@ -273,12 +331,11 @@ void queue_waiting_task(local_scheduler_state *state, task_spec *spec, bool from_global_scheduler) { LOG_DEBUG("Queueing task in waiting queue"); - /* Initiate fetch calls for any dependencies that are not present locally. */ - if (plasma_manager_is_connected(state->plasma_conn)) { - fetch_missing_dependencies(state, algorithm_state, spec); - } - queue_task(state, &algorithm_state->waiting_task_queue, spec, - from_global_scheduler); + task_queue_entry *task_entry = queue_task( + state, &algorithm_state->waiting_task_queue, spec, from_global_scheduler); + /* If we're queueing this task in the waiting queue, there must be at least + * one missing dependency, so record it. */ + fetch_missing_dependencies(state, algorithm_state, task_entry); } /** @@ -419,55 +476,83 @@ void handle_worker_available(local_scheduler_state *state, void handle_object_available(local_scheduler_state *state, scheduling_algorithm_state *algorithm_state, object_id object_id) { - /* Available object entries get freed if the object is removed. */ - available_object *entry = - (available_object *) malloc(sizeof(available_object)); - entry->object_id = object_id; - HASH_ADD(handle, algorithm_state->local_objects, object_id, sizeof(object_id), + /* Get the entry for this object from the active fetch request, or allocate + * one if needed. */ + object_entry *entry; + HASH_FIND(hh, algorithm_state->remote_objects, &object_id, sizeof(object_id), + entry); + if (entry != NULL) { + /* Remove the object from the active fetch requests. */ + HASH_DELETE(hh, algorithm_state->remote_objects, entry); + } else { + /* Allocate a new object entry. Object entries will get freed if the object + * is removed. */ + entry = (object_entry *) malloc(sizeof(object_entry)); + entry->object_id = object_id; + entry->dependent_tasks = NULL; + } + + /* Add the entry to the set of locally available objects. */ + HASH_ADD(hh, algorithm_state->local_objects, object_id, sizeof(object_id), entry); - /* If we were previously trying to fetch this object, remove the fetch request - * from the hash table. */ - fetch_object_request *fetch_req; - HASH_FIND(hh, algorithm_state->fetch_requests, &object_id, sizeof(object_id), - fetch_req); - if (fetch_req != NULL) { - HASH_DELETE(hh, algorithm_state->fetch_requests, fetch_req); - CHECK(event_loop_remove_timer(state->loop, fetch_req->timer) == AE_OK); - free(fetch_req); - } - - /* Move any tasks whose object dependencies are now ready to the dispatch - * queue. */ - /* TODO(swang): This can be optimized by keeping a lookup table from object - * ID to list of dependent tasks in the waiting queue. */ - task_queue_entry *elt, *tmp; - DL_FOREACH_SAFE(algorithm_state->waiting_task_queue, elt, tmp) { - if (can_run(algorithm_state, elt->spec)) { - LOG_DEBUG("Moved task to dispatch queue"); - DL_DELETE(algorithm_state->waiting_task_queue, elt); - DL_APPEND(algorithm_state->dispatch_task_queue, elt); + if (entry->dependent_tasks != NULL) { + /* Out of the tasks that were dependent on this object, if they were now + * ready to run, move them to the dispatch queue. */ + task_queue_entry *task_entry = NULL; + for (task_queue_entry **p = + (task_queue_entry **) utarray_front(entry->dependent_tasks); + p != NULL; + p = (task_queue_entry **) utarray_next(entry->dependent_tasks, p)) { + task_queue_entry *task_entry = *p; + if (can_run(algorithm_state, task_entry->spec)) { + LOG_DEBUG("Moved task to dispatch queue"); + DL_DELETE(algorithm_state->waiting_task_queue, task_entry); + DL_APPEND(algorithm_state->dispatch_task_queue, task_entry); + } } + /* Try to dispatch tasks, since we may have added some from the waiting + * queue. */ + dispatch_tasks(state, algorithm_state); + /* Clean up the records for dependent tasks. */ + utarray_free(entry->dependent_tasks); + entry->dependent_tasks = NULL; } - - /* Try to dispatch tasks, since we may have added some from the waiting - * queue. */ - dispatch_tasks(state, algorithm_state); } void handle_object_removed(local_scheduler_state *state, object_id removed_object_id) { + /* Remove the object from the set of locally available objects. */ scheduling_algorithm_state *algorithm_state = state->algorithm_state; - available_object *entry; - HASH_FIND(handle, algorithm_state->local_objects, &removed_object_id, + object_entry *entry; + HASH_FIND(hh, algorithm_state->local_objects, &removed_object_id, sizeof(removed_object_id), entry); - if (entry != NULL) { - HASH_DELETE(handle, algorithm_state->local_objects, entry); - free(entry); - } + CHECK(entry != NULL); + HASH_DELETE(hh, algorithm_state->local_objects, entry); + free(entry); - /* Move dependent tasks from the dispatch queue back to the waiting queue. */ + /* Track queued tasks that were dependent on this object. + * NOTE: Since objects often get removed in batches (e.g., during eviction), + * we may end up iterating through the queues many times in a row. If this + * turns out to be a bottleneck, consider tracking dependencies even for + * tasks in the dispatch queue, or batching object notifications. */ task_queue_entry *elt, *tmp; + /* Track the dependency for tasks that were in the waiting queue. */ + DL_FOREACH(algorithm_state->waiting_task_queue, elt) { + task_spec *task = elt->spec; + 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 arg_id = task_arg_id(task, i); + if (object_ids_equal(arg_id, removed_object_id)) { + fetch_missing_dependency(state, algorithm_state, elt, + removed_object_id); + } + } + } + } + /* Track the dependency for tasks that were in the dispatch queue. Remove + * these tasks from the dispatch queue and push them to the waiting queue. */ DL_FOREACH_SAFE(algorithm_state->dispatch_task_queue, elt, tmp) { task_spec *task = elt->spec; int64_t num_args = task_num_args(task); @@ -478,6 +563,8 @@ void handle_object_removed(local_scheduler_state *state, LOG_DEBUG("Moved task from dispatch queue back to waiting queue"); DL_DELETE(algorithm_state->dispatch_task_queue, elt); DL_APPEND(algorithm_state->waiting_task_queue, elt); + fetch_missing_dependency(state, algorithm_state, elt, + removed_object_id); } } } diff --git a/src/photon/photon_algorithm.h b/src/photon/photon_algorithm.h index 1e767a5bd..62453cb8e 100644 --- a/src/photon/photon_algorithm.h +++ b/src/photon/photon_algorithm.h @@ -108,6 +108,18 @@ void handle_worker_available(local_scheduler_state *state, scheduling_algorithm_state *algorithm_state, int worker_index); +/** + * This function fetches queued task's missing object dependencies. It is + * called every LOCAL_SCHEDULER_FETCH_TIMEOUT_MILLISECONDS. + * + * @param loop The local scheduler's event loop. + * @param id The ID of the timer that triggers this function. + * @param context The function's context. + * @return An integer representing the time interval in seconds before the + * next invocation of the function. + */ +int fetch_object_timeout_handler(event_loop *loop, timer_id id, void *context); + /** The following methods are for testing purposes only. */ #ifdef PHOTON_TEST /** diff --git a/src/photon/photon_scheduler.c b/src/photon/photon_scheduler.c index 8f283eafe..9f0185151 100644 --- a/src/photon/photon_scheduler.c +++ b/src/photon/photon_scheduler.c @@ -388,6 +388,9 @@ void start_server(const char *node_ip_address, event_loop_add_timer(loop, LOCAL_SCHEDULER_HEARTBEAT_TIMEOUT_MILLISECONDS, heartbeat_handler, g_state); } + /* Create a timer for fetching queued tasks' missing object dependencies. */ + event_loop_add_timer(loop, LOCAL_SCHEDULER_FETCH_TIMEOUT_MILLISECONDS, + fetch_object_timeout_handler, g_state); /* Run event loop. */ event_loop_run(loop); } diff --git a/src/photon/test/photon_tests.c b/src/photon/test/photon_tests.c index 51c4906ee..2e5788628 100644 --- a/src/photon/test/photon_tests.c +++ b/src/photon/test/photon_tests.c @@ -289,7 +289,7 @@ TEST object_reconstruction_suppression_test(void) { } } -TEST object_notifications_test(void) { +TEST task_dependency_test(void) { photon_mock *photon = init_photon_mock(false); local_scheduler_state *state = photon->photon_state; scheduling_algorithm_state *algorithm_state = state->algorithm_state; @@ -362,11 +362,84 @@ TEST object_notifications_test(void) { PASS(); } +TEST task_multi_dependency_test(void) { + photon_mock *photon = init_photon_mock(false); + local_scheduler_state *state = photon->photon_state; + scheduling_algorithm_state *algorithm_state = state->algorithm_state; + int worker_index = 0; + task_spec *spec = example_task_spec(2, 1); + object_id oid1 = task_arg_id(spec, 0); + object_id oid2 = task_arg_id(spec, 1); + + /* Check that the task gets queued in the waiting queue if the task is + * submitted, but the inputs and workers are not available. */ + handle_task_submitted(state, algorithm_state, spec); + ASSERT_EQ(num_waiting_tasks(algorithm_state), 1); + ASSERT_EQ(num_dispatch_tasks(algorithm_state), 0); + /* Check that the task stays in the waiting queue if only one input becomes + * available. */ + handle_object_available(state, algorithm_state, oid2); + ASSERT_EQ(num_waiting_tasks(algorithm_state), 1); + ASSERT_EQ(num_dispatch_tasks(algorithm_state), 0); + /* Once all inputs are available, the task is moved to the dispatch queue. */ + handle_object_available(state, algorithm_state, oid1); + ASSERT_EQ(num_waiting_tasks(algorithm_state), 0); + ASSERT_EQ(num_dispatch_tasks(algorithm_state), 1); + /* Once a worker is available, the task gets assigned. */ + handle_worker_available(state, algorithm_state, worker_index); + ASSERT_EQ(num_waiting_tasks(algorithm_state), 0); + ASSERT_EQ(num_dispatch_tasks(algorithm_state), 0); + reset_worker(photon, worker_index); + + /* Check that the task gets queued in the dispatch queue if the task is + * submitted and the inputs are available, but no worker is available yet. */ + handle_task_submitted(state, algorithm_state, spec); + ASSERT_EQ(num_waiting_tasks(algorithm_state), 0); + ASSERT_EQ(num_dispatch_tasks(algorithm_state), 1); + /* If any input is removed while a task is in the dispatch queue, the task + * gets moved back to the waiting queue. */ + handle_object_removed(state, oid1); + ASSERT_EQ(num_waiting_tasks(algorithm_state), 1); + ASSERT_EQ(num_dispatch_tasks(algorithm_state), 0); + handle_object_removed(state, oid2); + ASSERT_EQ(num_waiting_tasks(algorithm_state), 1); + ASSERT_EQ(num_dispatch_tasks(algorithm_state), 0); + /* Check that the task stays in the waiting queue if only one input becomes + * available. */ + handle_object_available(state, algorithm_state, oid2); + ASSERT_EQ(num_waiting_tasks(algorithm_state), 1); + ASSERT_EQ(num_dispatch_tasks(algorithm_state), 0); + /* Check that the task stays in the waiting queue if the one input is + * unavailable again. */ + handle_object_removed(state, oid2); + ASSERT_EQ(num_waiting_tasks(algorithm_state), 1); + ASSERT_EQ(num_dispatch_tasks(algorithm_state), 0); + /* Check that the task stays in the waiting queue if the other input becomes + * available. */ + handle_object_available(state, algorithm_state, oid1); + ASSERT_EQ(num_waiting_tasks(algorithm_state), 1); + ASSERT_EQ(num_dispatch_tasks(algorithm_state), 0); + /* Once all inputs are available, the task is moved to the dispatch queue. */ + handle_object_available(state, algorithm_state, oid2); + ASSERT_EQ(num_waiting_tasks(algorithm_state), 0); + ASSERT_EQ(num_dispatch_tasks(algorithm_state), 1); + /* Once a worker is available, the task gets assigned. */ + handle_worker_available(state, algorithm_state, worker_index); + ASSERT_EQ(num_waiting_tasks(algorithm_state), 0); + ASSERT_EQ(num_dispatch_tasks(algorithm_state), 0); + reset_worker(photon, worker_index); + + free_task_spec(spec); + destroy_photon_mock(photon); + PASS(); +} + SUITE(photon_tests) { RUN_REDIS_TEST(object_reconstruction_test); RUN_REDIS_TEST(object_reconstruction_recursive_test); RUN_REDIS_TEST(object_reconstruction_suppression_test); - RUN_TEST(object_notifications_test); + RUN_TEST(task_dependency_test); + RUN_TEST(task_multi_dependency_test); } GREATEST_MAIN_DEFS();