From 35b9dedb48756eb2d4a68b1394482542c66407db Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Sun, 4 Dec 2016 15:51:03 -0800 Subject: [PATCH] Remove scheduler_info. (#84) --- src/photon/CMakeLists.txt | 1 + src/photon/Makefile | 4 +- src/photon/photon.h | 31 +++++++- src/photon/photon_algorithm.c | 133 ++++++++++++++++++---------------- src/photon/photon_algorithm.h | 51 ++++++------- src/photon/photon_scheduler.c | 80 ++++++-------------- src/photon/photon_scheduler.h | 4 +- 7 files changed, 152 insertions(+), 152 deletions(-) diff --git a/src/photon/CMakeLists.txt b/src/photon/CMakeLists.txt index 6e8a87925..fa5b6cc16 100644 --- a/src/photon/CMakeLists.txt +++ b/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 diff --git a/src/photon/Makefile b/src/photon/Makefile index e0334bd27..4c2f78964 100644 --- a/src/photon/Makefile +++ b/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 diff --git a/src/photon/photon.h b/src/photon/photon.h index 298a27383..dec47db1c 100644 --- a/src/photon/photon.h +++ b/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 */ diff --git a/src/photon/photon_algorithm.c b/src/photon/photon_algorithm.c index e5eab90e1..982a2cda4 100644 --- a/src/photon/photon_algorithm.c +++ b/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); } diff --git a/src/photon/photon_algorithm.h b/src/photon/photon_algorithm.h index cff36b27d..328dc9a3b 100644 --- a/src/photon/photon_algorithm.h +++ b/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 */ diff --git a/src/photon/photon_scheduler.c b/src/photon/photon_scheduler.c index c1f7c14a8..1141a980a 100644 --- a/src/photon/photon_scheduler.c +++ b/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); } diff --git a/src/photon/photon_scheduler.h b/src/photon/photon_scheduler.h index 0c81ce97e..58cf59829 100644 --- a/src/photon/photon_scheduler.h +++ b/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);