Make global scheduler periodically resubmit tasks that can't be sched… (#306)

* Make global scheduler periodically resubmit tasks that can't be scheduled because their resource requirements are not met.

* Address comments and fix bug.

* Rename impossible_tasks -> pending_tasks.

* Fix formatting.

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);
 }

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;
 
 /**

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,

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);