From 9e4a3e497231b5af02e6e96d8473bed73463d34b Mon Sep 17 00:00:00 2001 From: Robert Nishihara Date: Mon, 19 Jun 2017 14:58:42 -0700 Subject: [PATCH] Replace some UT data structures in local scheduler with C++ STL. (#680) * Replace a local scheduler ut_array with a std::vector. * Replace vector of sizes in local scheduler with std::pair. * Remove utarray include. * Replace utarray with std::vector for reading local scheduler input messages. * Remove more UT data structures. * Remove UT includes. * Fix linting. * Include stdlib.h to find size_t. * Remove includes of stdbool.h. * Replace std::pair with TaskQueueEntry. * Fix redis tests. * Reinstate tests. --- src/common/common.cc | 2 + src/common/common.h | 1 - src/common/io.cc | 44 ------------- src/common/io.h | 24 +------ src/common/redis_module/ray_redis_module.cc | 1 - src/common/state/redis.cc | 1 - src/common/state/table.h | 1 - src/common/task.h | 1 - src/common/test/redis_tests.cc | 13 ++++ src/local_scheduler/local_scheduler.cc | 30 ++------- .../local_scheduler_algorithm.cc | 64 ++++++------------- src/local_scheduler/local_scheduler_shared.h | 7 +- src/plasma/plasma.h | 1 - src/plasma/plasma_client.h | 1 - src/plasma/plasma_manager.cc | 1 - 15 files changed, 44 insertions(+), 148 deletions(-) diff --git a/src/common/common.cc b/src/common/common.cc index e032ae3ec..d22a873ce 100644 --- a/src/common/common.cc +++ b/src/common/common.cc @@ -7,6 +7,8 @@ #include #include +#include "utarray.h" + #include "io.h" #include diff --git a/src/common/common.h b/src/common/common.h index 6c09913ac..436f5fb44 100644 --- a/src/common/common.h +++ b/src/common/common.h @@ -1,7 +1,6 @@ #ifndef COMMON_H #define COMMON_H -#include #include #include #include diff --git a/src/common/io.cc b/src/common/io.cc index e94a11127..4ec28850c 100644 --- a/src/common/io.cc +++ b/src/common/io.cc @@ -10,7 +10,6 @@ #include #include #include -#include #include #include "common.h" @@ -354,36 +353,6 @@ uint8_t *read_message_async(event_loop *loop, int sock) { return message; } -int64_t read_buffer(int fd, int64_t *type, UT_array *buffer) { - int64_t version; - int closed = read_bytes(fd, (uint8_t *) &version, sizeof(version)); - if (closed) { - goto disconnected; - } - CHECK(version == RAY_PROTOCOL_VERSION); - int64_t length; - closed = read_bytes(fd, (uint8_t *) type, sizeof(*type)); - if (closed) { - goto disconnected; - } - closed = read_bytes(fd, (uint8_t *) &length, sizeof(length)); - if (closed) { - goto disconnected; - } - if (length > utarray_len(buffer)) { - utarray_resize(buffer, length); - } - closed = read_bytes(fd, (uint8_t *) utarray_front(buffer), length); - if (closed) { - goto disconnected; - } - return length; -disconnected: - /* Handle the case in which the socket is closed. */ - *type = DISCONNECT_CLIENT; - return 0; -} - int64_t read_vector(int fd, int64_t *type, std::vector &buffer) { int64_t version; int closed = read_bytes(fd, (uint8_t *) &version, sizeof(version)); @@ -427,16 +396,3 @@ char *read_log_message(int fd) { CHECK(type == LOG_MESSAGE); return (char *) bytes; } - -void write_formatted_log_message(int socket_fd, const char *format, ...) { - UT_string *cmd; - va_list ap; - - utstring_new(cmd); - va_start(ap, format); - utstring_printf_va(cmd, format, ap); - va_end(ap); - - write_log_message(socket_fd, utstring_body(cmd)); - utstring_free(cmd); -} diff --git a/src/common/io.h b/src/common/io.h index a6a77e3c5..1ad717fd9 100644 --- a/src/common/io.h +++ b/src/common/io.h @@ -1,11 +1,10 @@ #ifndef IO_H #define IO_H -#include #include +#include #include -#include "utarray.h" #define RAY_PROTOCOL_VERSION 0x0000000000000000 @@ -169,25 +168,6 @@ void read_message(int fd, int64_t *type, int64_t *length, uint8_t **bytes); */ uint8_t *read_message_async(event_loop *loop, int sock); -/** - * Read a sequence of bytes written by write_message from a file descriptor. - * This does not allocate space for the message if the provided buffer is - * large enough and can therefore often avoid allocations. - * - * @note The caller must create and free the buffer. - * - * @param fd The file descriptor to read from. It can be non-blocking. - * @param type The type of the message that is read will be written at this - * address. If there was an error while reading, this will be - * DISCONNECT_CLIENT. - * @param buffer The array the message will be written to. If it is not - * large enough to hold the message, it will be enlarged by read_buffer. - * @return Number of bytes of the message that were read. This size does not - * include the bytes used to encode the type and length. If there was - * an error while reading, this will be 0. - */ -int64_t read_buffer(int fd, int64_t *type, UT_array *buffer); - /** * Read a sequence of bytes written by write_message from a file descriptor. * This does not allocate space for the message if the provided buffer is @@ -210,8 +190,6 @@ int64_t read_vector(int fd, int64_t *type, std::vector &buffer); */ void write_log_message(int fd, const char *message); -void write_formatted_log_message(int fd, const char *format, ...); - /** * Reads a null-terminated string from the file descriptor that has been * written by write_log_message. Allocates and returns a pointer to the string. diff --git a/src/common/redis_module/ray_redis_module.cc b/src/common/redis_module/ray_redis_module.cc index a8d54207e..bd749fcce 100644 --- a/src/common/redis_module/ray_redis_module.cc +++ b/src/common/redis_module/ray_redis_module.cc @@ -1,5 +1,4 @@ #include "redismodule.h" -#include #include #include "redis_string.h" diff --git a/src/common/state/redis.cc b/src/common/state/redis.cc index e0fed8430..44b65c0c8 100644 --- a/src/common/state/redis.cc +++ b/src/common/state/redis.cc @@ -1,7 +1,6 @@ /* Redis implementation of the global state store */ #include -#include #include #include #include diff --git a/src/common/state/table.h b/src/common/state/table.h index 760cc3dcc..5c247dd3e 100644 --- a/src/common/state/table.h +++ b/src/common/state/table.h @@ -2,7 +2,6 @@ #define TABLE_H #include "uthash.h" -#include "stdbool.h" #include "common.h" #include "db.h" diff --git a/src/common/task.h b/src/common/task.h index 5bb4f3fee..557921462 100644 --- a/src/common/task.h +++ b/src/common/task.h @@ -1,7 +1,6 @@ #ifndef TASK_H #define TASK_H -#include #include #include #include "common.h" diff --git a/src/common/test/redis_tests.cc b/src/common/test/redis_tests.cc index 78f80f986..ecef159cb 100644 --- a/src/common/test/redis_tests.cc +++ b/src/common/test/redis_tests.cc @@ -20,6 +20,19 @@ const char *test_key = "foo"; const char *test_value = "bar"; UT_array *connections = NULL; +void write_formatted_log_message(int socket_fd, const char *format, ...) { + UT_string *cmd; + va_list ap; + + utstring_new(cmd); + va_start(ap, format); + utstring_printf_va(cmd, format, ap); + va_end(ap); + + write_log_message(socket_fd, utstring_body(cmd)); + utstring_free(cmd); +} + int async_redis_socket_test_callback_called = 0; void async_redis_socket_test_callback(redisAsyncContext *ac, diff --git a/src/local_scheduler/local_scheduler.cc b/src/local_scheduler/local_scheduler.cc index a3337f270..7e30e84ab 100644 --- a/src/local_scheduler/local_scheduler.cc +++ b/src/local_scheduler/local_scheduler.cc @@ -23,13 +23,6 @@ #include "state/task_table.h" #include "state/object_table.h" #include "state/error_table.h" -#include "utarray.h" - -UT_icd task_ptr_icd = {sizeof(Task *), NULL, NULL, NULL}; - -UT_icd pid_t_icd = {sizeof(pid_t), NULL, NULL, NULL}; - -UT_icd byte_icd = {sizeof(uint8_t), NULL, NULL, NULL}; /** * A helper function for printing available and requested resource information. @@ -200,9 +193,6 @@ void LocalSchedulerState_free(LocalSchedulerState *state) { /* Free the algorithm state. */ SchedulingAlgorithmState_free(state->algorithm_state); state->algorithm_state = NULL; - /* Free the input buffer. */ - utarray_free(state->input_buffer); - state->input_buffer = NULL; /* Destroy the event loop. */ event_loop_destroy(state->loop); state->loop = NULL; @@ -384,9 +374,6 @@ LocalSchedulerState *LocalSchedulerState_init( process_plasma_notification, state); /* Add scheduler state. */ state->algorithm_state = SchedulingAlgorithmState_init(); - /* Add the input buffer. This is used to read in messages from clients without - * having to reallocate a new buffer every time. */ - utarray_new(state->input_buffer, &byte_icd); /* Initialize resource vectors. */ for (int i = 0; i < ResourceIndex_MAX; i++) { @@ -877,13 +864,14 @@ void process_message(event_loop *loop, LocalSchedulerState *state = worker->local_scheduler_state; int64_t type; - int64_t length = read_buffer(client_sock, &type, state->input_buffer); + int64_t length = read_vector(client_sock, &type, state->input_buffer); + uint8_t *input = state->input_buffer.data(); LOG_DEBUG("New event of type %" PRId64, type); switch (type) { case MessageType_SubmitTask: { - TaskSpec *spec = (TaskSpec *) utarray_front(state->input_buffer); + TaskSpec *spec = (TaskSpec *) input; /* Update the result table, which holds mappings of object ID -> ID of the * task that created it. */ if (state->db != NULL) { @@ -917,8 +905,7 @@ void process_message(event_loop *loop, } break; case MessageType_EventLogMessage: { /* Parse the message. */ - auto message = flatbuffers::GetRoot( - utarray_front(state->input_buffer)); + auto message = flatbuffers::GetRoot(input); if (state->db != NULL) { RayLogger_log_event( state->db, (uint8_t *) message->key()->data(), message->key()->size(), @@ -926,8 +913,7 @@ void process_message(event_loop *loop, } } break; case MessageType_RegisterClientRequest: { - auto message = flatbuffers::GetRoot( - utarray_front(state->input_buffer)); + auto message = flatbuffers::GetRoot(input); handle_client_register(state, worker, message); send_client_register_reply(state, worker); } break; @@ -943,8 +929,7 @@ void process_message(event_loop *loop, } } break; case MessageType_ReconstructObject: { - auto message = flatbuffers::GetRoot( - utarray_front(state->input_buffer)); + auto message = flatbuffers::GetRoot(input); if (worker->task_in_progress != NULL && !worker->is_blocked) { /* TODO(swang): For now, we don't handle blocked actors. */ if (ActorID_equal(worker->actor_id, NIL_ACTOR_ID)) { @@ -994,8 +979,7 @@ void process_message(event_loop *loop, print_worker_info("Worker unblocked", state->algorithm_state); } break; case MessageType_PutObject: { - auto message = - flatbuffers::GetRoot(utarray_front(state->input_buffer)); + auto message = flatbuffers::GetRoot(input); result_table_add(state->db, from_flatbuf(message->object_id()), from_flatbuf(message->task_id()), true, NULL, NULL, NULL); } break; diff --git a/src/local_scheduler/local_scheduler_algorithm.cc b/src/local_scheduler/local_scheduler_algorithm.cc index 7c4132ae5..e10755cd0 100644 --- a/src/local_scheduler/local_scheduler_algorithm.cc +++ b/src/local_scheduler/local_scheduler_algorithm.cc @@ -1,8 +1,5 @@ #include "local_scheduler_algorithm.h" -#include -#include "utarray.h" - #include #include #include @@ -38,12 +35,6 @@ struct ObjectEntry { std::vector::iterator> dependent_tasks; }; -/** This is used to define the queue of actor task specs for which the - * corresponding local scheduler is unknown. */ -UT_icd task_spec_icd = {sizeof(TaskSpec *), NULL, NULL, NULL}; -/** This is used to keep track of task spec sizes in the above queue. */ -UT_icd task_spec_size_icd = {sizeof(int64_t), NULL, NULL, NULL}; - /** This struct contains information about a specific actor. This struct will be * used inside of a hash table. */ typedef struct { @@ -78,14 +69,12 @@ struct SchedulingAlgorithmState { * particular, a queue of tasks that are waiting to execute on that actor. * This is only used for actors that exist locally. */ std::unordered_map local_actor_infos; - /** An array of actor tasks that have been submitted but this local scheduler + /** A vector of actor tasks that have been submitted but this local scheduler * doesn't know which local scheduler is responsible for them, so cannot * assign them to the correct local scheduler yet. Whenever a notification * about a new local scheduler arrives, we will resubmit all of these tasks * locally. */ - UT_array *cached_submitted_actor_tasks; - /** An array of task sizes of cached_submitted_actor_tasks. */ - UT_array *cached_submitted_actor_task_sizes; + std::vector cached_submitted_actor_tasks; /** An array of pointers to workers in the worker pool. These are workers * that have registered a PID with us and that are now waiting to be * assigned a task to execute. */ @@ -127,10 +116,6 @@ SchedulingAlgorithmState *SchedulingAlgorithmState_init(void) { algorithm_state->waiting_task_queue = new std::list(); algorithm_state->dispatch_task_queue = new std::list(); - utarray_new(algorithm_state->cached_submitted_actor_tasks, &task_spec_icd); - utarray_new(algorithm_state->cached_submitted_actor_task_sizes, - &task_spec_size_icd); - return algorithm_state; } @@ -154,14 +139,11 @@ void SchedulingAlgorithmState_free(SchedulingAlgorithmState *algorithm_state) { remove_actor(algorithm_state, actor_id); } /* Free the list of cached actor task specs and the task specs themselves. */ - for (int i = 0; - i < utarray_len(algorithm_state->cached_submitted_actor_tasks); ++i) { - TaskSpec **spec = (TaskSpec **) utarray_eltptr( - algorithm_state->cached_submitted_actor_tasks, i); - free(*spec); + for (int i = 0; i < algorithm_state->cached_submitted_actor_tasks.size(); + ++i) { + TaskQueueEntry task = algorithm_state->cached_submitted_actor_tasks[i]; + TaskQueueEntry_free(&task); } - utarray_free(algorithm_state->cached_submitted_actor_tasks); - utarray_free(algorithm_state->cached_submitted_actor_task_sizes); /* Free the algorithm state. */ delete algorithm_state; } @@ -831,8 +813,7 @@ void handle_actor_task_submitted(LocalSchedulerState *state, SchedulingAlgorithmState *algorithm_state, TaskSpec *task_spec, int64_t task_spec_size) { - TaskSpec *spec = TaskSpec_copy(task_spec, task_spec_size); - ActorID actor_id = TaskSpec_actor_id(spec); + ActorID actor_id = TaskSpec_actor_id(task_spec); CHECK(!ActorID_equal(actor_id, NIL_ACTOR_ID)); if (state->actor_mapping.count(actor_id) == 0) { @@ -841,9 +822,8 @@ void handle_actor_task_submitted(LocalSchedulerState *state, * will be resubmitted (internally by the local scheduler) whenever a new * actor notification arrives. NOTE(swang): These tasks have not yet been * added to the task table. */ - utarray_push_back(algorithm_state->cached_submitted_actor_tasks, &spec); - utarray_push_back(algorithm_state->cached_submitted_actor_task_sizes, - &task_spec_size); + TaskQueueEntry task_entry = TaskQueueEntry_init(task_spec, task_spec_size); + algorithm_state->cached_submitted_actor_tasks.push_back(task_entry); return; } @@ -851,7 +831,7 @@ void handle_actor_task_submitted(LocalSchedulerState *state, get_db_client_id(state->db))) { /* This local scheduler is responsible for the actor, so handle the task * locally. */ - add_task_to_actor_queue(state, algorithm_state, spec, task_spec_size, + add_task_to_actor_queue(state, algorithm_state, task_spec, task_spec_size, false); /* Attempt to dispatch tasks to this actor. */ dispatch_actor_task(state, algorithm_state, actor_id); @@ -860,10 +840,9 @@ void handle_actor_task_submitted(LocalSchedulerState *state, * scheduler that is responsible for this actor and assign the task directly * to that local scheduler. */ give_task_to_local_scheduler( - state, algorithm_state, spec, task_spec_size, + state, algorithm_state, task_spec, task_spec_size, state->actor_mapping[actor_id].local_scheduler_id); } - TaskSpec_free(spec); } void handle_actor_creation_notification( @@ -871,24 +850,21 @@ void handle_actor_creation_notification( SchedulingAlgorithmState *algorithm_state, ActorID actor_id) { int num_cached_actor_tasks = - utarray_len(algorithm_state->cached_submitted_actor_tasks); - CHECK(num_cached_actor_tasks == - utarray_len(algorithm_state->cached_submitted_actor_task_sizes)); + algorithm_state->cached_submitted_actor_tasks.size(); + for (int i = 0; i < num_cached_actor_tasks; ++i) { - TaskSpec **spec = (TaskSpec **) utarray_eltptr( - algorithm_state->cached_submitted_actor_tasks, i); - int64_t *task_spec_size = (int64_t *) utarray_eltptr( - algorithm_state->cached_submitted_actor_task_sizes, i); + TaskQueueEntry task = algorithm_state->cached_submitted_actor_tasks[i]; + TaskSpec *spec = task.spec; /* Note that handle_actor_task_submitted may append the spec to the end of * the cached_submitted_actor_tasks array. */ - handle_actor_task_submitted(state, algorithm_state, *spec, *task_spec_size); + handle_actor_task_submitted(state, algorithm_state, task.spec, + task.task_spec_size); } /* Remove all the tasks that were resubmitted. This does not erase the tasks * that were newly appended to the cached_submitted_actor_tasks array. */ - utarray_erase(algorithm_state->cached_submitted_actor_tasks, 0, - num_cached_actor_tasks); - utarray_erase(algorithm_state->cached_submitted_actor_task_sizes, 0, - num_cached_actor_tasks); + auto begin = algorithm_state->cached_submitted_actor_tasks.begin(); + algorithm_state->cached_submitted_actor_tasks.erase( + begin, begin + num_cached_actor_tasks); } void handle_task_scheduled(LocalSchedulerState *state, diff --git a/src/local_scheduler/local_scheduler_shared.h b/src/local_scheduler/local_scheduler_shared.h index 8b35c5ec4..df3372061 100644 --- a/src/local_scheduler/local_scheduler_shared.h +++ b/src/local_scheduler/local_scheduler_shared.h @@ -5,17 +5,12 @@ #include "common/state/table.h" #include "common/state/db.h" #include "plasma_client.h" -#include "utarray.h" -#include "uthash.h" #include #include #include #include -/* These are needed to define the UT_arrays. */ -extern UT_icd task_ptr_icd; - /** This struct is used to maintain a mapping from actor IDs to the ID of the * local scheduler that is responsible for the actor. */ struct ActorMapEntry { @@ -67,7 +62,7 @@ struct LocalSchedulerState { SchedulingAlgorithmState *algorithm_state; /** Input buffer, used for reading input in process_message to avoid * allocation for each call to process_message. */ - UT_array *input_buffer; + std::vector input_buffer; /** Vector of static attributes associated with the node owned by this local * scheduler. */ double static_resources[ResourceIndex_MAX]; diff --git a/src/plasma/plasma.h b/src/plasma/plasma.h index 4cd3f46dd..c3b2d89c7 100644 --- a/src/plasma/plasma.h +++ b/src/plasma/plasma.h @@ -4,7 +4,6 @@ #include #include #include -#include #include #include #include /* pid_t */ diff --git a/src/plasma/plasma_client.h b/src/plasma/plasma_client.h index cd0384041..f8abbd43e 100644 --- a/src/plasma/plasma_client.h +++ b/src/plasma/plasma_client.h @@ -1,7 +1,6 @@ #ifndef PLASMA_CLIENT_H #define PLASMA_CLIENT_H -#include #include #include diff --git a/src/plasma/plasma_manager.cc b/src/plasma/plasma_manager.cc index 6447df544..4b4de038c 100644 --- a/src/plasma/plasma_manager.cc +++ b/src/plasma/plasma_manager.cc @@ -11,7 +11,6 @@ #include #include #include -#include #include #include #include