diff --git a/src/global_scheduler/global_scheduler.c b/src/global_scheduler/global_scheduler.c index 815d618b8..645a5b3e5 100644 --- a/src/global_scheduler/global_scheduler.c +++ b/src/global_scheduler/global_scheduler.c @@ -18,6 +18,9 @@ * global_scheduler_state type. */ UT_icd local_scheduler_icd = {sizeof(local_scheduler), NULL, NULL, NULL}; +/* This is used to define the array of tasks that haven't been scheduled yet. */ +UT_icd pending_tasks_icd = {sizeof(task *), NULL, NULL, NULL}; + /** * Assign the given task to the local scheduler, update Redis and scheduler data * structures. @@ -72,6 +75,8 @@ global_scheduler_state *init_global_scheduler(event_loop *loop, db_attach(state->db, loop, false); utarray_new(state->local_schedulers, &local_scheduler_icd); state->policy_state = init_global_scheduler_policy(); + /* Initialize the array of tasks that have not been scheduled yet. */ + utarray_new(state->pending_tasks, &pending_tasks_icd); return state; } @@ -103,6 +108,18 @@ void free_global_scheduler(global_scheduler_state *state) { utarray_free(object_entry->object_locations); free(object_entry); } + /* Free the array of unschedulable tasks. */ + int64_t num_pending_tasks = utarray_len(state->pending_tasks); + if (num_pending_tasks > 0) { + LOG_WARN("There are %" PRId64 + " remaining tasks in the pending tasks array.", + num_pending_tasks); + } + for (int i = 0; i < num_pending_tasks; ++i) { + task **pending_task = (task **) utarray_eltptr(state->pending_tasks, i); + free_task(*pending_task); + } + utarray_free(state->pending_tasks); /* Free the global scheduler state. */ free(state); } @@ -134,10 +151,18 @@ local_scheduler *get_local_scheduler(global_scheduler_state *state, return NULL; } -void process_task_waiting(task *task, void *user_context) { +void process_task_waiting(task *waiting_task, void *user_context) { global_scheduler_state *state = (global_scheduler_state *) user_context; LOG_DEBUG("Task waiting callback is called."); - handle_task_waiting(state, state->policy_state, task); + bool successfully_assigned = + handle_task_waiting(state, state->policy_state, waiting_task); + /* If the task was not successfully submitted to a local scheduler, add the + * task to the array of pending tasks. The global scheduler will periodically + * resubmit the tasks in this array. */ + if (!successfully_assigned) { + task *task_copy = copy_task(waiting_task); + utarray_push_back(state->pending_tasks, &task_copy); + } } /** @@ -288,6 +313,26 @@ void local_scheduler_table_handler(db_client_id client_id, } } +int task_cleanup_handler(event_loop *loop, timer_id id, void *context) { + global_scheduler_state *state = context; + /* Loop over the pending tasks and resubmit them. */ + int64_t num_pending_tasks = utarray_len(state->pending_tasks); + for (int64_t i = num_pending_tasks - 1; i >= 0; --i) { + task **pending_task = (task **) utarray_eltptr(state->pending_tasks, i); + /* Pretend that the task has been resubmitted. */ + bool successfully_assigned = + handle_task_waiting(state, state->policy_state, *pending_task); + if (successfully_assigned) { + /* The task was successfully assigned, so remove it from this list and + * free it. */ + utarray_erase(state->pending_tasks, i, 1); + free(*pending_task); + } + } + /* Reset the timer. */ + return GLOBAL_SCHEDULER_TASK_CLEANUP_MILLISECONDS; +} + void start_server(const char *redis_addr, int redis_port) { event_loop *loop = event_loop_create(); g_state = init_global_scheduler(loop, redis_addr, redis_port); @@ -315,6 +360,13 @@ void start_server(const char *redis_addr, int redis_port) { * schedulers. */ local_scheduler_table_subscribe(g_state->db, local_scheduler_table_handler, g_state, NULL); + /* Start a timer that periodically checks if there are queued tasks that can + * be scheduled. Currently this is only used to handle the special case in + * which a task is waiting and no node meets its static resource requirements. + * If a new node joins the cluster that does have enough resources, then this + * timer should notice and schedule the task. */ + event_loop_add_timer(loop, GLOBAL_SCHEDULER_TASK_CLEANUP_MILLISECONDS, + task_cleanup_handler, g_state); /* Start the event loop. */ event_loop_run(loop); } diff --git a/src/global_scheduler/global_scheduler.h b/src/global_scheduler/global_scheduler.h index 56290bf8c..a0a67e84f 100644 --- a/src/global_scheduler/global_scheduler.h +++ b/src/global_scheduler/global_scheduler.h @@ -8,6 +8,10 @@ #include "utarray.h" #include "uthash.h" +/* The frequency with which the global scheduler checks if there are any tasks + * that haven't been scheduled yet. */ +#define GLOBAL_SCHEDULER_TASK_CLEANUP_MILLISECONDS 100 + /** Contains all information that is associated with a local scheduler. */ typedef struct { /** The ID of the local scheduler in Redis. */ @@ -73,6 +77,8 @@ typedef struct { aux_address_entry *photon_plasma_map; /** Objects cached by this global scheduler instance. */ scheduler_object_info *scheduler_object_info_table; + /** An array of tasks that haven't been scheduled yet. */ + UT_array *pending_tasks; } global_scheduler_state; /** diff --git a/src/global_scheduler/global_scheduler_algorithm.c b/src/global_scheduler/global_scheduler_algorithm.c index 77fd0937e..9b62015e1 100644 --- a/src/global_scheduler/global_scheduler_algorithm.c +++ b/src/global_scheduler/global_scheduler_algorithm.c @@ -267,15 +267,7 @@ double calculate_cost_pending(const global_scheduler_state *state, return scheduler->num_recent_tasks_sent + scheduler->info.task_queue_length; } -/** - * Main new task handling function in the global scheduler. - * - * @param state Global scheduler state. - * @param policy_state State specific to the scheduling policy. - * @param task New task to be scheduled. - * @return Void. - */ -void handle_task_waiting(global_scheduler_state *state, +bool handle_task_waiting(global_scheduler_state *state, global_scheduler_policy_state *policy_state, task *task) { task_spec *task_spec = task_task_spec(task); @@ -324,12 +316,13 @@ void handle_task_waiting(global_scheduler_state *state, object_id_to_string(task_task_id(task), id_string, ID_STRING_SIZE)); /* TODO(atumanov): propagate this error to the task's driver and/or * cache the task in case new local schedulers satisfy it in the future. */ - return; + return false; } CHECKM(!IS_NIL_ID(best_photon_id), "Task is feasible, but doesn't have a local scheduler assigned."); /* A local scheduler ID was found, so assign the task. */ assign_task_to_local_scheduler(state, task, best_photon_id); + return true; } void handle_object_available(global_scheduler_state *state, diff --git a/src/global_scheduler/global_scheduler_algorithm.h b/src/global_scheduler/global_scheduler_algorithm.h index 0f3291c1d..159dee2a8 100644 --- a/src/global_scheduler/global_scheduler_algorithm.h +++ b/src/global_scheduler/global_scheduler_algorithm.h @@ -55,9 +55,10 @@ void destroy_global_scheduler_policy( * @param state Global scheduler state. * @param policy_state State specific to the scheduling policy. * @param task New task to be scheduled. - * @return Void. + * @return True if the task was assigned to a local scheduler and false + * otherwise. */ -void handle_task_waiting(global_scheduler_state *state, +bool handle_task_waiting(global_scheduler_state *state, global_scheduler_policy_state *policy_state, task *task);