Remove scheduler_info. (#84)

src/photon/CMakeLists.txt

@@ -76,6 +76,7 @@ include_directories("${CMAKE_SOURCE_DIR}/../")
 include_directories("${CMAKE_SOURCE_DIR}/../common/")
 include_directories("${CMAKE_SOURCE_DIR}/../common/thirdparty/")
 include_directories("${CMAKE_SOURCE_DIR}/../common/lib/python/")
+include_directories("${CMAKE_SOURCE_DIR}/../plasma/")
 
 add_library(photon SHARED
   photon_extension.c

src/photon/Makefile

@@ -1,5 +1,5 @@
 CC = gcc
-CFLAGS = -g -Wall -Wextra -Werror=implicit-function-declaration -Wno-sign-compare -Wno-unused-parameter -Wno-type-limits -Wno-missing-field-initializers --std=c99 -D_XOPEN_SOURCE=500 -D_POSIX_C_SOURCE=200809L -I.. -I../common -I../common/thirdparty -fPIC
+CFLAGS = -g -Wall -Wextra -Werror=implicit-function-declaration -Wno-sign-compare -Wno-unused-parameter -Wno-type-limits -Wno-missing-field-initializers --std=c99 -D_XOPEN_SOURCE=500 -D_POSIX_C_SOURCE=200809L -I.. -I../common -I../common/thirdparty -I../plasma/ -fPIC
 BUILD = build
 
 all: $(BUILD)/photon_scheduler $(BUILD)/photon_client.a
@@ -8,7 +8,7 @@ $(BUILD)/photon_client.a: photon_client.o
 	ar rcs $(BUILD)/photon_client.a photon_client.o
 
 $(BUILD)/photon_scheduler: photon.h photon_scheduler.c photon_algorithm.c common
-	$(CC) $(CFLAGS) -o $@ photon_scheduler.c photon_algorithm.c ../common/build/libcommon.a ../common/thirdparty/hiredis/libhiredis.a -I../common/thirdparty/ -I../common/ ../plasma/build/libplasma_client.a -I../plasma/
+	$(CC) $(CFLAGS) -o $@ photon_scheduler.c photon_algorithm.c ../common/build/libcommon.a ../common/thirdparty/hiredis/libhiredis.a -I../common/thirdparty/ -I../common/ ../plasma/build/libplasma_client.a
 
 common: FORCE
 	cd ../common; make

src/photon/photon.h

@@ -3,6 +3,7 @@
 
 #include "common/task.h"
 #include "common/state/db.h"
+#include "plasma_client.h"
 #include "utarray.h"
 #include "uthash.h"
 
@@ -27,14 +28,38 @@ typedef struct {
 UT_icd task_ptr_icd;
 UT_icd worker_icd;
 
-/** Resources that are exposed to the scheduling algorithm. */
+/** Association between the socket fd of a worker and its worker_index. */
 typedef struct {
+  /** The socket fd of a worker. */
+  int sock;
+  /** The index of the worker in scheduler_info->workers. */
+  int64_t worker_index;
+  /** Handle for the hash table. */
+  UT_hash_handle hh;
+} worker_index;
+
+/** Internal state of the scheduling algorithm. */
+typedef struct scheduling_algorithm_state scheduling_algorithm_state;
+
+/** The state of the local scheduler. */
+typedef struct {
+  /** The local scheduler event loop. */
+  event_loop *loop;
+  /** Association between client socket and worker index. */
+  worker_index *worker_index;
   /** List of workers available to this node. The index into this array
    *  is the worker_index and is used to identify workers throughout
    *  the program. */
   UT_array *workers;
-  /* The handle to the database. */
+  /** The handle to the database. */
   db_handle *db;
-} scheduler_info;
+  /** The Plasma client. */
+  plasma_connection *plasma_conn;
+  /** State for the scheduling algorithm. */
+  scheduling_algorithm_state *algorithm_state;
+  /** Input buffer, used for reading input in process_message to avoid
+   *  allocation for each call to process_message. */
+  UT_array *input_buffer;
+} local_scheduler_state;
 
 #endif /* PHOTON_H */

src/photon/photon_algorithm.c

@@ -29,7 +29,7 @@ typedef struct {
 } available_object;
 
 /** Part of the photon state that is maintained by the scheduling algorithm. */
-struct scheduler_state {
+struct scheduling_algorithm_state {
   /** An array of pointers to tasks that are waiting to be scheduled. */
   task_queue_entry *task_queue;
   /** An array of worker indices corresponding to clients that are
@@ -40,30 +40,32 @@ struct scheduler_state {
   available_object *local_objects;
 };
 
-scheduler_state *make_scheduler_state(void) {
-  scheduler_state *state = malloc(sizeof(scheduler_state));
+scheduling_algorithm_state *make_scheduling_algorithm_state(void) {
+  scheduling_algorithm_state *algorithm_state =
+      malloc(sizeof(scheduling_algorithm_state));
   /* Initialize an empty hash map for the cache of local available objects. */
-  state->local_objects = NULL;
+  algorithm_state->local_objects = NULL;
   /* Initialize the local data structures used for queuing tasks and workers. */
-  state->task_queue = NULL;
-  utarray_new(state->available_workers, &ut_int_icd);
-  return state;
+  algorithm_state->task_queue = NULL;
+  utarray_new(algorithm_state->available_workers, &ut_int_icd);
+  return algorithm_state;
 }
 
-void free_scheduler_state(scheduler_state *s) {
+void free_scheduling_algorithm_state(
+    scheduling_algorithm_state *algorithm_state) {
   task_queue_entry *elt, *tmp1;
-  DL_FOREACH_SAFE(s->task_queue, elt, tmp1) {
-    DL_DELETE(s->task_queue, elt);
+  DL_FOREACH_SAFE(algorithm_state->task_queue, elt, tmp1) {
+    DL_DELETE(algorithm_state->task_queue, elt);
     free_task_spec(elt->spec);
     free(elt);
   }
-  utarray_free(s->available_workers);
+  utarray_free(algorithm_state->available_workers);
   available_object *available_obj, *tmp2;
-  HASH_ITER(handle, s->local_objects, available_obj, tmp2) {
-    HASH_DELETE(handle, s->local_objects, available_obj);
+  HASH_ITER(handle, algorithm_state->local_objects, available_obj, tmp2) {
+    HASH_DELETE(handle, algorithm_state->local_objects, available_obj);
     free(available_obj);
   }
-  free(s);
+  free(algorithm_state);
 }
 
 /**
@@ -75,13 +77,14 @@ void free_scheduler_state(scheduler_state *s) {
  * @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.
  */
-bool can_run(scheduler_state *s, task_spec *task) {
+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, s->local_objects, &obj_id, sizeof(obj_id), entry);
+      HASH_FIND(handle, 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
          * right now. */
@@ -101,14 +104,15 @@ bool can_run(scheduler_state *s, task_spec *task) {
  * @return This returns 1 if it successfully assigned a task to the worker,
  *         otherwise it returns 0.
  */
-bool find_and_schedule_task_if_possible(scheduler_info *info,
-                                        scheduler_state *state,
-                                        int worker_index) {
+bool find_and_schedule_task_if_possible(
+    local_scheduler_state *state,
+    scheduling_algorithm_state *algorithm_state,
+    int worker_index) {
   task_queue_entry *elt, *tmp;
   bool found_task_to_schedule = false;
   /* Find the first task whose dependencies are available locally. */
-  DL_FOREACH_SAFE(state->task_queue, elt, tmp) {
-    if (can_run(state, elt->spec)) {
+  DL_FOREACH_SAFE(algorithm_state->task_queue, elt, tmp) {
+    if (can_run(algorithm_state, elt->spec)) {
       found_task_to_schedule = true;
       break;
     }
@@ -116,30 +120,30 @@ bool find_and_schedule_task_if_possible(scheduler_info *info,
   if (found_task_to_schedule) {
     /* This task's dependencies are available locally, so assign the task to the
      * worker. */
-    assign_task_to_worker(info, elt->spec, worker_index,
+    assign_task_to_worker(state, elt->spec, worker_index,
                           elt->from_global_scheduler);
     /* Update the task queue data structure and free the task. */
-    DL_DELETE(state->task_queue, elt);
+    DL_DELETE(algorithm_state->task_queue, elt);
     free_task_spec(elt->spec);
     free(elt);
   }
   return found_task_to_schedule;
 }
 
-void run_task_immediately(scheduler_info *info,
-                          scheduler_state *s,
+void run_task_immediately(local_scheduler_state *state,
+                          scheduling_algorithm_state *algorithm_state,
                           task_spec *spec,
                           bool from_global_scheduler) {
   /* Get the last available worker in the available worker queue. */
-  int *worker_index = (int *) utarray_back(s->available_workers);
+  int *worker_index = (int *) utarray_back(algorithm_state->available_workers);
   /* Tell the available worker to execute the task. */
-  assign_task_to_worker(info, spec, *worker_index, from_global_scheduler);
+  assign_task_to_worker(state, spec, *worker_index, from_global_scheduler);
   /* Remove the available worker from the queue and free the struct. */
-  utarray_pop_back(s->available_workers);
+  utarray_pop_back(algorithm_state->available_workers);
 }
 
-void queue_task_locally(scheduler_info *info,
-                        scheduler_state *s,
+void queue_task_locally(local_scheduler_state *state,
+                        scheduling_algorithm_state *algorithm_state,
                         task_spec *spec,
                         bool from_global_scheduler) {
   /* Copy the spec and add it to the task queue. The allocated spec will be
@@ -148,93 +152,100 @@ void queue_task_locally(scheduler_info *info,
   elt->spec = (task_spec *) malloc(task_spec_size(spec));
   memcpy(elt->spec, spec, task_spec_size(spec));
   elt->from_global_scheduler = from_global_scheduler;
-  DL_APPEND(s->task_queue, elt);
+  DL_APPEND(algorithm_state->task_queue, elt);
 }
 
-void give_task_to_global_scheduler(scheduler_info *info,
-                                   scheduler_state *s,
+void give_task_to_global_scheduler(local_scheduler_state *state,
+                                   scheduling_algorithm_state *algorithm_state,
                                    task_spec *spec,
                                    bool from_global_scheduler) {
   /* Pass on the task to the global scheduler. */
   DCHECK(!from_global_scheduler);
   task *task = alloc_task(spec, TASK_STATUS_WAITING, NIL_ID);
-  DCHECK(info->db != NULL);
-  task_table_add_task(info->db, task, (retry_info *) &photon_retry, NULL, NULL);
+  DCHECK(state->db != NULL);
+  task_table_add_task(state->db, task, (retry_info *) &photon_retry, NULL,
+                      NULL);
 }
 
-void handle_task_submitted(scheduler_info *info,
-                           scheduler_state *s,
+void handle_task_submitted(local_scheduler_state *state,
+                           scheduling_algorithm_state *algorithm_state,
                            task_spec *spec) {
   /* If this task's dependencies are available locally, and if there is an
    * available worker, then assign this task to an available worker. If we
    * cannot assign the task to a worker immediately, we either queue the task in
    * the local task queue or we pass the task to the global scheduler. For now,
    * we pass the task along to the global scheduler if there is one. */
-  if ((utarray_len(s->available_workers) > 0) && can_run(s, spec)) {
-    run_task_immediately(info, s, spec, false);
-  } else if (info->db == NULL) {
-    queue_task_locally(info, s, spec, false);
+  if ((utarray_len(algorithm_state->available_workers) > 0) &&
+      can_run(algorithm_state, spec)) {
+    run_task_immediately(state, algorithm_state, spec, false);
+  } else if (state->db == NULL) {
+    queue_task_locally(state, algorithm_state, spec, false);
   } else {
-    give_task_to_global_scheduler(info, s, spec, false);
+    give_task_to_global_scheduler(state, algorithm_state, spec, false);
   }
 }
 
-void handle_task_scheduled(scheduler_info *info,
-                           scheduler_state *s,
+void handle_task_scheduled(local_scheduler_state *state,
+                           scheduling_algorithm_state *algorithm_state,
                            task_spec *spec) {
   /* This callback handles tasks that were assigned to this local scheduler by
    * the global scheduler, so we can safely assert that there is a connection
    * to the database. */
-  DCHECK(info->db != NULL);
+  DCHECK(state->db != NULL);
   /* If this task's dependencies are available locally, and if there is an
    * available worker, then assign this task to an available worker. If we
    * cannot assign the task to a worker immediately, queue the task locally. */
-  if ((utarray_len(s->available_workers) > 0) && can_run(s, spec)) {
-    run_task_immediately(info, s, spec, true);
+  if ((utarray_len(algorithm_state->available_workers) > 0) &&
+      can_run(algorithm_state, spec)) {
+    run_task_immediately(state, algorithm_state, spec, true);
   } else {
-    queue_task_locally(info, s, spec, true);
+    queue_task_locally(state, algorithm_state, spec, true);
   }
 }
 
-void handle_worker_available(scheduler_info *info,
-                             scheduler_state *state,
+void handle_worker_available(local_scheduler_state *state,
+                             scheduling_algorithm_state *algorithm_state,
                              int worker_index) {
   int scheduled_task =
-      find_and_schedule_task_if_possible(info, state, worker_index);
+      find_and_schedule_task_if_possible(state, algorithm_state, worker_index);
   /* If we couldn't find a task to schedule, add the worker to the queue of
    * available workers. */
   if (!scheduled_task) {
-    for (int *p = (int *) utarray_front(state->available_workers); p != NULL;
-         p = (int *) utarray_next(state->available_workers, p)) {
+    for (int *p = (int *) utarray_front(algorithm_state->available_workers);
+         p != NULL;
+         p = (int *) utarray_next(algorithm_state->available_workers, p)) {
       DCHECK(*p != worker_index);
     }
     /* Add client_sock to a list of available workers. This struct will be freed
      * when a task is assigned to this worker. */
-    utarray_push_back(state->available_workers, &worker_index);
+    utarray_push_back(algorithm_state->available_workers, &worker_index);
     LOG_DEBUG("Adding worker_index %d to available workers.\n", worker_index);
   }
 }
 
-void handle_object_available(scheduler_info *info,
-                             scheduler_state *state,
+void handle_object_available(local_scheduler_state *state,
+                             scheduling_algorithm_state *algorithm_state,
                              object_id object_id) {
   /* TODO(rkn): When does this get freed? */
   available_object *entry =
       (available_object *) malloc(sizeof(available_object));
   entry->object_id = object_id;
-  HASH_ADD(handle, state->local_objects, object_id, sizeof(object_id), entry);
+  HASH_ADD(handle, algorithm_state->local_objects, object_id, sizeof(object_id),
+           entry);
 
   /* Check if we can schedule any tasks. */
   int num_tasks_scheduled = 0;
-  for (int *p = (int *) utarray_front(state->available_workers); p != NULL;
-       p = (int *) utarray_next(state->available_workers, p)) {
+  for (int *p = (int *) utarray_front(algorithm_state->available_workers);
+       p != NULL;
+       p = (int *) utarray_next(algorithm_state->available_workers, p)) {
     /* Schedule a task on this worker if possible. */
-    int scheduled_task = find_and_schedule_task_if_possible(info, state, *p);
+    int scheduled_task =
+        find_and_schedule_task_if_possible(state, algorithm_state, *p);
     if (!scheduled_task) {
       /* There are no tasks we can schedule, so exit the loop. */
       break;
     }
     num_tasks_scheduled += 1;
   }
-  utarray_erase(state->available_workers, 0, num_tasks_scheduled);
+  utarray_erase(algorithm_state->available_workers, 0, num_tasks_scheduled);
 }

src/photon/photon_algorithm.h

@@ -4,6 +4,11 @@
 #include "photon.h"
 #include "common/task.h"
 
+/* The duration that the local scheduler will wait before reinitiating a fetch
+ * request for a missing task dependency. TODO(rkn): We may want this to be
+ * adaptable based on the load on the local scheduler. */
+#define LOCAL_SCHEDULER_FETCH_TIMEOUT_MILLISECONDS 1000
+
 /* ==== The scheduling algorithm ====
  *
  * This file contains declaration for all functions and data structures
@@ -12,77 +17,69 @@
  *
  */
 
-/** Internal state of the scheduling algorithm. */
-typedef struct scheduler_state scheduler_state;
-
 /**
  * Initialize the scheduler state.
  *
- * @return Internal state of the scheduling algorithm.
+ * @return State managed by the scheduling algorithm.
  */
-scheduler_state *make_scheduler_state(void);
+scheduling_algorithm_state *make_scheduling_algorithm_state(void);
 
 /**
  * Free the scheduler state.
  *
- * @param state Internal state of the scheduling algorithm.
+ * @param algorithm_state State maintained by the scheduling algorithm.
  * @return Void.
  */
-void free_scheduler_state(scheduler_state *state);
+void free_scheduling_algorithm_state(
+    scheduling_algorithm_state *algorithm_state);
 
 /**
  * This function will be called when a new task is submitted by a worker for
  * execution.
  *
- * @param info Info about resources exposed by photon to the scheduling
- *        algorithm.
- * @param state State of the scheduling algorithm.
  * @param task Task that is submitted by the worker.
  * @return Void.
  */
-void handle_task_submitted(scheduler_info *info,
-                           scheduler_state *state,
+void handle_task_submitted(local_scheduler_state *state,
+                           scheduling_algorithm_state *algorithm_state,
                            task_spec *spec);
 
 /**
  * This function will be called when a task is assigned by the global scheduler
  * for execution on this local scheduler.
  *
- * @param info Info about resources exposed by photon to the scheduling
- *        algorithm.
- * @param state State of the scheduling algorithm.
+ * @param state The state of the local scheduler.
+ * @param algorithm_state State maintained by the scheduling algorithm.
  * @param task Task that is assigned by the global scheduler.
  * @return Void.
  */
-void handle_task_scheduled(scheduler_info *info,
-                           scheduler_state *state,
+void handle_task_scheduled(local_scheduler_state *state,
+                           scheduling_algorithm_state *algorithm_state,
                            task_spec *spec);
 
 /**
  * This function is called if a new object becomes available in the local
  * plasma store.
  *
- * @param info Info about resources exposed by photon to the scheduling
- *        algorithm.
- * @param state State of the scheduling algorithm.
+ * @param state The state of the local scheduler.
+ * @param algorithm_state State maintained by the scheduling algorithm.
  * @param object_id ID of the object that became available.
  * @return Void.
  */
-void handle_object_available(scheduler_info *info,
-                             scheduler_state *state,
+void handle_object_available(local_scheduler_state *state,
+                             scheduling_algorithm_state *algorithm_state,
                              object_id object_id);
 
 /**
  * This function is called when a new worker becomes available
  *
- * @param info Info about resources exposed by photon to the scheduling
- *        algorithm.
- * @param state State of the scheduling algorithm.
+ * @param state The state of the local scheduler.
+ * @param algorithm_state State maintained by the scheduling algorithm.
  * @param worker_index The index of the worker that becomes available.
  * @return Void.
  */
-void handle_worker_available(scheduler_info *info,
-                             scheduler_state *state,
+void handle_worker_available(local_scheduler_state *state,
+                             scheduling_algorithm_state *algorithm_state,
                              int worker_index);
 
 #endif /* PHOTON_ALGORITHM_H */

src/photon/photon_scheduler.c

@@ -12,7 +12,6 @@
 #include "photon.h"
 #include "photon_algorithm.h"
 #include "photon_scheduler.h"
-#include "plasma_client.h"
 #include "state/db.h"
 #include "state/task_table.h"
 #include "state/object_table.h"
@@ -22,32 +21,6 @@
 UT_icd task_ptr_icd = {sizeof(task *), NULL, NULL, NULL};
 UT_icd worker_icd = {sizeof(worker), NULL, NULL, NULL};
 
-/** Association between the socket fd of a worker and its worker_index. */
-typedef struct {
-  /** The socket fd of a worker. */
-  int sock;
-  /** The index of the worker in scheduler_info->workers. */
-  int64_t worker_index;
-  /** Handle for the hash table. */
-  UT_hash_handle hh;
-} worker_index;
-
-struct local_scheduler_state {
-  /** The local scheduler event loop. */
-  event_loop *loop;
-  /** The Plasma client. */
-  plasma_connection *plasma_conn;
-  /** Association between client socket and worker index. */
-  worker_index *worker_index;
-  /** Info that is exposed to the scheduling algorithm. */
-  scheduler_info *scheduler_info;
-  /** State for the scheduling algorithm. */
-  scheduler_state *scheduler_state;
-  /** Input buffer, used for reading input in process_message to avoid
-   *  allocation for each call to process_message. */
-  UT_array *input_buffer;
-};
-
 UT_icd byte_icd = {sizeof(uint8_t), NULL, NULL, NULL};
 
 local_scheduler_state *init_local_scheduler(event_loop *loop,
@@ -67,25 +40,23 @@ local_scheduler_state *init_local_scheduler(event_loop *loop,
                       process_plasma_notification, state);
   state->worker_index = NULL;
   /* Add scheduler info. */
-  state->scheduler_info = malloc(sizeof(scheduler_info));
-  utarray_new(state->scheduler_info->workers, &worker_icd);
+  utarray_new(state->workers, &worker_icd);
   /* Connect to Redis if a Redis address is provided. */
   if (redis_addr != NULL) {
-    state->scheduler_info->db =
-        db_connect(redis_addr, redis_port, "photon", "", -1);
-    db_attach(state->scheduler_info->db, loop);
+    state->db = db_connect(redis_addr, redis_port, "photon", "", -1);
+    db_attach(state->db, loop);
   } else {
-    state->scheduler_info->db = NULL;
+    state->db = NULL;
   }
   /* Add scheduler state. */
-  state->scheduler_state = make_scheduler_state();
+  state->algorithm_state = make_scheduling_algorithm_state();
   utarray_new(state->input_buffer, &byte_icd);
   return state;
 };
 
 void free_local_scheduler(local_scheduler_state *s) {
-  if (s->scheduler_info->db != NULL) {
-    db_disconnect(s->scheduler_info->db);
+  if (s->db != NULL) {
+    db_disconnect(s->db);
   }
   plasma_disconnect(s->plasma_conn);
   worker_index *current_worker_index, *temp_worker_index;
@@ -93,34 +64,33 @@ void free_local_scheduler(local_scheduler_state *s) {
     HASH_DEL(s->worker_index, current_worker_index);
     free(current_worker_index);
   }
-  utarray_free(s->scheduler_info->workers);
-  free(s->scheduler_info);
-  free_scheduler_state(s->scheduler_state);
+  utarray_free(s->workers);
+  free_scheduling_algorithm_state(s->algorithm_state);
   utarray_free(s->input_buffer);
   event_loop_destroy(s->loop);
   free(s);
 }
 
-void assign_task_to_worker(scheduler_info *info,
+void assign_task_to_worker(local_scheduler_state *state,
                            task_spec *spec,
                            int worker_index,
                            bool from_global_scheduler) {
-  CHECK(worker_index < utarray_len(info->workers));
-  worker *w = (worker *) utarray_eltptr(info->workers, worker_index);
+  CHECK(worker_index < utarray_len(state->workers));
+  worker *w = (worker *) utarray_eltptr(state->workers, worker_index);
   write_message(w->sock, EXECUTE_TASK, task_spec_size(spec), (uint8_t *) spec);
   /* Update the global task table. */
-  if (info->db != NULL) {
+  if (state->db != NULL) {
     retry_info retry;
     memset(&retry, 0, sizeof(retry));
     retry.num_retries = 0;
     retry.timeout = 100;
     retry.fail_callback = NULL;
     task *task =
-        alloc_task(spec, TASK_STATUS_RUNNING, get_db_client_id(info->db));
+        alloc_task(spec, TASK_STATUS_RUNNING, get_db_client_id(state->db));
     if (from_global_scheduler) {
-      task_table_update(info->db, task, (retry_info *) &retry, NULL, NULL);
+      task_table_update(state->db, task, (retry_info *) &retry, NULL, NULL);
     } else {
-      task_table_add_task(info->db, task, (retry_info *) &retry, NULL, NULL);
+      task_table_add_task(state->db, task, (retry_info *) &retry, NULL, NULL);
     }
   }
 }
@@ -142,7 +112,7 @@ void process_plasma_notification(event_loop *loop,
     close(client_sock);
     return;
   }
-  handle_object_available(s->scheduler_info, s->scheduler_state, obj_id);
+  handle_object_available(s, s->algorithm_state, obj_id);
 }
 
 void process_message(event_loop *loop, int client_sock, void *context,
@@ -157,15 +127,14 @@ void process_message(event_loop *loop, int client_sock, void *context,
   switch (type) {
   case SUBMIT_TASK: {
     task_spec *spec = (task_spec *) utarray_front(s->input_buffer);
-    handle_task_submitted(s->scheduler_info, s->scheduler_state, spec);
+    handle_task_submitted(s, s->algorithm_state, spec);
   } break;
   case TASK_DONE: {
   } break;
   case GET_TASK: {
     worker_index *wi;
     HASH_FIND_INT(s->worker_index, &client_sock, wi);
-    handle_worker_available(s->scheduler_info, s->scheduler_state,
-                            wi->worker_index);
+    handle_worker_available(s, s->algorithm_state, wi->worker_index);
   } break;
   case DISCONNECT_CLIENT: {
     LOG_INFO("Disconnecting client on fd %d", client_sock);
@@ -189,12 +158,12 @@ void new_client_connection(event_loop *loop, int listener_sock, void *context,
   /* TODO(pcm): Where shall we free this? */
   worker_index *new_worker_index = malloc(sizeof(worker_index));
   new_worker_index->sock = new_socket;
-  new_worker_index->worker_index = utarray_len(s->scheduler_info->workers);
+  new_worker_index->worker_index = utarray_len(s->workers);
   HASH_ADD_INT(s->worker_index, sock, new_worker_index);
   worker worker;
   memset(&worker, 0, sizeof(worker));
   worker.sock = new_socket;
-  utarray_push_back(s->scheduler_info->workers, &worker);
+  utarray_push_back(s->workers, &worker);
 }
 
 /* We need this code so we can clean up when we get a SIGTERM signal. */
@@ -211,7 +180,7 @@ void signal_handler(int signal) {
 /* End of the cleanup code. */
 
 void handle_task_scheduled_callback(task *original_task, void *user_context) {
-  handle_task_scheduled(g_state->scheduler_info, g_state->scheduler_state,
+  handle_task_scheduled(g_state, g_state->algorithm_state,
                         task_task_spec(original_task));
 }
 
@@ -236,9 +205,8 @@ void start_server(const char *socket_name,
   retry.num_retries = 0;
   retry.timeout = 100;
   retry.fail_callback = NULL;
-  if (g_state->scheduler_info->db != NULL) {
-    task_table_subscribe(g_state->scheduler_info->db,
-                         get_db_client_id(g_state->scheduler_info->db),
+  if (g_state->db != NULL) {
+    task_table_subscribe(g_state->db, get_db_client_id(g_state->db),
                          TASK_STATUS_SCHEDULED, handle_task_scheduled_callback,
                          NULL, &retry, NULL, NULL);
   }

src/photon/photon_scheduler.h

@@ -4,8 +4,6 @@
 #include "task.h"
 #include "event_loop.h"
 
-typedef struct local_scheduler_state local_scheduler_state;
-
 /**
  * Establish a connection to a new client.
  *
@@ -32,7 +30,7 @@ void new_client_connection(event_loop *loop,
  *        scheduler by the global scheduler and false otherwise.
  * @return Void.
  */
-void assign_task_to_worker(scheduler_info *info,
+void assign_task_to_worker(local_scheduler_state *state,
                            task_spec *task,
                            int worker_index,
                            bool from_global_scheduler);