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

@@ -7,6 +7,8 @@
 #include <sys/stat.h>
 #include <fcntl.h>
 
+#include "utarray.h"
+
 #include "io.h"
 #include <functional>
 

src/common/common.h

@@ -1,7 +1,6 @@
 #ifndef COMMON_H
 #define COMMON_H
 
-#include <stdbool.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>

src/common/io.cc

@@ -10,7 +10,6 @@
 #include <stdarg.h>
 #include <sys/ioctl.h>
 #include <netinet/in.h>
-#include <utstring.h>
 #include <netdb.h>
 
 #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<uint8_t> &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);
-}

src/common/io.h

@@ -1,11 +1,10 @@
 #ifndef IO_H
 #define IO_H
 
-#include <stdbool.h>
 #include <stdint.h>
+#include <stdlib.h>
 
 #include <vector>
-#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<uint8_t> &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.

src/common/redis_module/ray_redis_module.cc

@@ -1,5 +1,4 @@
 #include "redismodule.h"
-#include <stdbool.h>
 #include <string.h>
 
 #include "redis_string.h"

src/common/state/redis.cc

@@ -1,7 +1,6 @@
 /* Redis implementation of the global state store */
 
 #include <assert.h>
-#include <stdbool.h>
 #include <stdlib.h>
 #include <unistd.h>
 #include <vector>

src/common/state/table.h

@@ -2,7 +2,6 @@
 #define TABLE_H
 
 #include "uthash.h"
-#include "stdbool.h"
 
 #include "common.h"
 #include "db.h"

src/common/task.h

@@ -1,7 +1,6 @@
 #ifndef TASK_H
 #define TASK_H
 
-#include <stdbool.h>
 #include <stddef.h>
 #include <stdint.h>
 #include "common.h"

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,

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<EventLogMessage>(
-        utarray_front(state->input_buffer));
+    auto message = flatbuffers::GetRoot<EventLogMessage>(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<RegisterClientRequest>(
-        utarray_front(state->input_buffer));
+    auto message = flatbuffers::GetRoot<RegisterClientRequest>(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<ReconstructObject>(
-        utarray_front(state->input_buffer));
+    auto message = flatbuffers::GetRoot<ReconstructObject>(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<PutObject>(utarray_front(state->input_buffer));
+    auto message = flatbuffers::GetRoot<PutObject>(input);
     result_table_add(state->db, from_flatbuf(message->object_id()),
                      from_flatbuf(message->task_id()), true, NULL, NULL, NULL);
   } break;

src/local_scheduler/local_scheduler_algorithm.cc

@@ -1,8 +1,5 @@
 #include "local_scheduler_algorithm.h"
 
-#include <stdbool.h>
-#include "utarray.h"
-
 #include <list>
 #include <vector>
 #include <unordered_map>
@@ -38,12 +35,6 @@ struct ObjectEntry {
   std::vector<std::list<TaskQueueEntry>::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<ActorID, LocalActorInfo, UniqueIDHasher> 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<TaskQueueEntry> 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<TaskQueueEntry>();
   algorithm_state->dispatch_task_queue = new std::list<TaskQueueEntry>();
 
-  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,

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 <list>
 #include <unordered_map>
 #include <unordered_set>
 #include <vector>
 
-/* 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<uint8_t> input_buffer;
   /** Vector of static attributes associated with the node owned by this local
    *  scheduler. */
   double static_resources[ResourceIndex_MAX];

src/plasma/plasma.h

@@ -4,7 +4,6 @@
 #include <stdlib.h>
 #include <stdio.h>
 #include <errno.h>
-#include <stdbool.h>
 #include <stddef.h>
 #include <string.h>
 #include <unistd.h> /* pid_t */

src/plasma/plasma_client.h

@@ -1,7 +1,6 @@
 #ifndef PLASMA_CLIENT_H
 #define PLASMA_CLIENT_H
 
-#include <stdbool.h>
 #include <time.h>
 
 #include <deque>

src/plasma/plasma_manager.cc

@@ -11,7 +11,6 @@
 #include <unistd.h>
 #include <signal.h>
 #include <stdlib.h>
-#include <stdbool.h>
 #include <sys/mman.h>
 #include <sys/types.h>
 #include <sys/socket.h>