Get valgrind in place for plasma (#35)

* plasma store: refactor and get valgrind in place

* fix valgrind errors

.travis.yml

@@ -29,6 +29,17 @@ matrix:
       install: []
       script:
         - .travis/check-git-clang-format-output.sh
+    - os: linux
+      dist: trusty
+      python: "2.7"
+      env: VALGRIND=1
+      before_install:
+        - sudo apt-get update -qq
+        - sudo apt-get install -qq valgrind
+      script:
+        - make
+        - source setup-env.sh
+        - python test/test.py valgrind
 
 install:
   - make

src/plasma_client.c

@@ -115,6 +115,7 @@ void plasma_get(plasma_store_conn *conn,
   plasma_send_request(conn->conn, PLASMA_GET, &req);
   plasma_reply reply;
   int fd = recv_fd(conn->conn, (char *) &reply, sizeof(plasma_reply));
+  CHECKM(fd != -1, "recv not successful");
   plasma_object *object = &reply.object;
   *data = lookup_or_mmap(conn, fd, object->handle.store_fd,
                          object->handle.mmap_size) +

src/plasma_manager.c

@@ -8,6 +8,7 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <unistd.h>
+#include <signal.h>
 #include <stdlib.h>
 #include <sys/mman.h>
 #include <sys/types.h>
@@ -343,7 +344,15 @@ void start_server(const char *store_socket_name,
   event_loop_run(loop);
 }
 
+/* Report "success" to valgrind. */
+void signal_handler(int signal) {
+  if (signal == SIGTERM) {
+    exit(0);
+  }
+}
+
 int main(int argc, char *argv[]) {
+  signal(SIGTERM, signal_handler);
   /* Socket name of the plasma store this manager is connected to. */
   char *store_socket_name = NULL;
   /* IP address of this node. */

src/plasma_store.c

@@ -18,6 +18,7 @@
 #include <sys/un.h>
 #include <getopt.h>
 #include <string.h>
+#include <signal.h>
 #include <limits.h>
 #include <poll.h>
 
@@ -28,7 +29,7 @@
 #include "utarray.h"
 #include "fling.h"
 #include "malloc.h"
-#include "plasma.h"
+#include "plasma_store.h"
 
 void *dlmalloc(size_t);
 void dlfree(void *);
@@ -61,13 +62,6 @@ typedef struct {
   uint8_t *pointer;
 } object_table_entry;
 
-/* Objects that are still being written by their owner process. */
-object_table_entry *open_objects = NULL;
-
-/* Objects that have already been sealed by their owner process and
- * can now be shared with other processes. */
-object_table_entry *sealed_objects = NULL;
-
 typedef struct {
   /* Object id of this object. */
   object_id object_id;
@@ -77,11 +71,29 @@ typedef struct {
   UT_hash_handle handle;
 } object_notify_entry;
 
-/* Objects that processes are waiting for. */
-object_notify_entry *objects_notify = NULL;
+struct plasma_store_state {
+  /* Event loop of the plasma store. */
+  event_loop *loop;
+  /* Objects that are still being written by their owner process. */
+  object_table_entry *open_objects;
+  /* Objects that have already been sealed by their owner process and
+   * can now be shared with other processes. */
+  object_table_entry *sealed_objects;
+  /* Objects that processes are waiting for. */
+  object_notify_entry *objects_notify;
+};
+
+plasma_store_state *init_plasma_store(event_loop *loop) {
+  plasma_store_state *state = malloc(sizeof(plasma_store_state));
+  state->loop = loop;
+  state->open_objects = NULL;
+  state->sealed_objects = NULL;
+  state->objects_notify = NULL;
+  return state;
+}
 
 /* Create a new object buffer in the hash table. */
-plasma_object create_object(int conn,
+plasma_object create_object(plasma_store_state *s,
                             object_id object_id,
                             int64_t data_size,
                             int64_t metadata_size,
@@ -89,7 +101,7 @@ plasma_object create_object(int conn,
   LOG_DEBUG("creating object"); /* TODO(pcm): add object_id here */
 
   object_table_entry *entry;
-  HASH_FIND(handle, open_objects, &object_id, sizeof(object_id), entry);
+  HASH_FIND(handle, s->open_objects, &object_id, sizeof(object_id), entry);
   CHECKM(entry == NULL, "Cannot create object twice.");
 
   uint8_t *pointer = dlmalloc(data_size + metadata_size);
@@ -108,9 +120,9 @@ plasma_object create_object(int conn,
   entry->fd = fd;
   entry->map_size = map_size;
   entry->offset = offset;
-  HASH_ADD(handle, open_objects, object_id, sizeof(object_id), entry);
-  object_handle handle = {.store_fd = fd, .mmap_size = map_size};
-  result->handle = handle;
+  HASH_ADD(handle, s->open_objects, object_id, sizeof(object_id), entry);
+  result->handle.store_fd = fd;
+  result->handle.mmap_size = map_size;
   result->data_offset = offset;
   result->metadata_offset = offset + data_size;
   result->data_size = data_size;
@@ -118,13 +130,15 @@ plasma_object create_object(int conn,
 }
 
 /* Get an object from the hash table. */
-int get_object(int conn, object_id object_id, plasma_object *result) {
+int get_object(plasma_store_state *s,
+               int conn,
+               object_id object_id,
+               plasma_object *result) {
   object_table_entry *entry;
-  HASH_FIND(handle, sealed_objects, &object_id, sizeof(object_id), entry);
+  HASH_FIND(handle, s->sealed_objects, &object_id, sizeof(object_id), entry);
   if (entry) {
-    object_handle handle = {.store_fd = entry->fd,
-                            .mmap_size = entry->map_size};
-    result->handle = handle;
+    result->handle.store_fd = entry->fd;
+    result->handle.mmap_size = entry->map_size;
     result->data_offset = entry->offset;
     result->metadata_offset = entry->offset + entry->info.data_size;
     result->data_size = entry->info.data_size;
@@ -133,14 +147,14 @@ int get_object(int conn, object_id object_id, plasma_object *result) {
   } else {
     object_notify_entry *notify_entry;
     LOG_DEBUG("object not in hash table of sealed objects");
-    HASH_FIND(handle, objects_notify, &object_id, sizeof(object_id),
+    HASH_FIND(handle, s->objects_notify, &object_id, sizeof(object_id),
               notify_entry);
     if (!notify_entry) {
       notify_entry = malloc(sizeof(object_notify_entry));
       memset(notify_entry, 0, sizeof(object_notify_entry));
       utarray_new(notify_entry->conns, &ut_int_icd);
       memcpy(&notify_entry->object_id, &object_id, 20);
-      HASH_ADD(handle, objects_notify, object_id, sizeof(object_id),
+      HASH_ADD(handle, s->objects_notify, object_id, sizeof(object_id),
                notify_entry);
     }
     utarray_push_back(notify_entry->conns, &conn);
@@ -149,62 +163,64 @@ int get_object(int conn, object_id object_id, plasma_object *result) {
 }
 
 /* Check if an object is present. */
-int contains_object(int conn, object_id object_id) {
+int contains_object(plasma_store_state *s, object_id object_id) {
   object_table_entry *entry;
-  HASH_FIND(handle, sealed_objects, &object_id, sizeof(object_id), entry);
+  HASH_FIND(handle, s->sealed_objects, &object_id, sizeof(object_id), entry);
   return entry ? OBJECT_FOUND : OBJECT_NOT_FOUND;
 }
 
 /* Seal an object that has been created in the hash table. */
-void seal_object(int conn,
+void seal_object(plasma_store_state *s,
                  object_id object_id,
                  UT_array **conns,
                  plasma_object *result) {
   LOG_DEBUG("sealing object");  // TODO(pcm): add object_id here
   object_table_entry *entry;
-  HASH_FIND(handle, open_objects, &object_id, sizeof(object_id), entry);
+  HASH_FIND(handle, s->open_objects, &object_id, sizeof(object_id), entry);
   if (!entry) {
     return; /* TODO(pcm): return error */
   }
-  HASH_DELETE(handle, open_objects, entry);
-  HASH_ADD(handle, sealed_objects, object_id, sizeof(object_id), entry);
+  HASH_DELETE(handle, s->open_objects, entry);
+  HASH_ADD(handle, s->sealed_objects, object_id, sizeof(object_id), entry);
   /* Inform processes that the object is ready now. */
   object_notify_entry *notify_entry;
-  HASH_FIND(handle, objects_notify, &object_id, sizeof(object_id),
+  HASH_FIND(handle, s->objects_notify, &object_id, sizeof(object_id),
             notify_entry);
   if (!notify_entry) {
     *conns = NULL;
     return;
   }
-  object_handle handle = {.store_fd = entry->fd, .mmap_size = entry->map_size};
-  result->handle = handle;
+  result->handle.store_fd = entry->fd;
+  result->handle.mmap_size = entry->map_size;
   result->data_offset = entry->offset;
   result->metadata_offset = entry->offset + entry->info.data_size;
   result->data_size = entry->info.data_size;
   result->metadata_size = entry->info.metadata_size;
-  HASH_DELETE(handle, objects_notify, notify_entry);
+  HASH_DELETE(handle, s->objects_notify, notify_entry);
   *conns = notify_entry->conns;
   free(notify_entry);
 }
 
 /* Delete an object that has been created in the hash table. */
-void delete_object(int conn, object_id object_id) {
+void delete_object(plasma_store_state *s, object_id object_id) {
   LOG_DEBUG("deleting object");  // TODO(rkn): add object_id here
   object_table_entry *entry;
-  HASH_FIND(handle, sealed_objects, &object_id, sizeof(object_id), entry);
+  HASH_FIND(handle, s->sealed_objects, &object_id, sizeof(object_id), entry);
   /* TODO(rkn): This should probably not fail, but should instead throw an
    * error. Maybe we should also support deleting objects that have been created
    * but not sealed. */
   CHECKM(entry != NULL, "To delete an object it must have been sealed.");
   uint8_t *pointer = entry->pointer;
-  HASH_DELETE(handle, sealed_objects, entry);
+  HASH_DELETE(handle, s->sealed_objects, entry);
   dlfree(pointer);
+  free(entry);
 }
 
 void process_message(event_loop *loop,
                      int client_sock,
                      void *context,
                      int events) {
+  plasma_store_state *s = context;
   int64_t type;
   int64_t length;
   plasma_request *req;
@@ -215,26 +231,26 @@ void process_message(event_loop *loop,
 
   switch (type) {
   case PLASMA_CREATE:
-    create_object(client_sock, req->object_id, req->data_size,
-                  req->metadata_size, &reply.object);
+    create_object(s, req->object_id, req->data_size, req->metadata_size,
+                  &reply.object);
     send_fd(client_sock, reply.object.handle.store_fd, (char *) &reply,
             sizeof(reply));
     break;
   case PLASMA_GET:
-    if (get_object(client_sock, req->object_id, &reply.object) ==
+    if (get_object(s, client_sock, req->object_id, &reply.object) ==
         OBJECT_FOUND) {
       send_fd(client_sock, reply.object.handle.store_fd, (char *) &reply,
               sizeof(reply));
     }
     break;
   case PLASMA_CONTAINS:
-    if (contains_object(client_sock, req->object_id) == OBJECT_FOUND) {
+    if (contains_object(s, req->object_id) == OBJECT_FOUND) {
       reply.has_object = 1;
     }
     plasma_send_reply(client_sock, &reply);
     break;
   case PLASMA_SEAL:
-    seal_object(client_sock, req->object_id, &conns, &reply.object);
+    seal_object(s, req->object_id, &conns, &reply.object);
     if (conns) {
       for (int *c = (int *) utarray_front(conns); c != NULL;
            c = (int *) utarray_next(conns, c)) {
@@ -245,7 +261,7 @@ void process_message(event_loop *loop,
     }
     break;
   case PLASMA_DELETE:
-    delete_object(client_sock, req->object_id);
+    delete_object(s, req->object_id);
     break;
   case DISCONNECT_CLIENT: {
     LOG_DEBUG("Disconnecting client on fd %d", client_sock);
@@ -269,16 +285,25 @@ void new_client_connection(event_loop *loop,
   LOG_DEBUG("new connection with fd %d", new_socket);
 }
 
+/* Report "success" to valgrind. */
+void signal_handler(int signal) {
+  if (signal == SIGTERM) {
+    exit(0);
+  }
+}
+
 void start_server(char *socket_name) {
   int socket = bind_ipc_sock(socket_name);
   CHECK(socket >= 0);
   event_loop *loop = event_loop_create();
+  plasma_store_state *state = init_plasma_store(loop);
   event_loop_add_file(loop, socket, EVENT_LOOP_READ, new_client_connection,
-                      NULL);
+                      state);
   event_loop_run(loop);
 }
 
 int main(int argc, char *argv[]) {
+  signal(SIGTERM, signal_handler);
   char *socket_name = NULL;
   int c;
   while ((c = getopt(argc, argv, "s:")) != -1) {

src/plasma_store.h

@@ -3,47 +3,66 @@
 
 #include "plasma.h"
 
+typedef struct plasma_store_state plasma_store_state;
+
 /**
  * Create a new object:
  *
+ * @param s The plasma store state.
  * @param object_id Object ID of the object to be created.
  * @param data_size Size in bytes of the object to be created.
  * @param metadata_size Size in bytes of the object metadata.
+ * @return The new plasma object.
  */
-void create_object(int conn,
-                   object_id object_id,
-                   int64_t data_size,
-                   int64_t metadata_size,
-                   plasma_object *result);
+plasma_object create_object(plasma_store_state *s,
+                            object_id object_id,
+                            int64_t data_size,
+                            int64_t metadata_size,
+                            plasma_object *result);
 
 /**
  * Get an object:
  *
+ * @param s The plasma store state.
+ * @param conn The client connection that requests the object.
  * @param object_id Object ID of the object to be gotten.
- *
- * Returns the status of the object (object_status in plasma.h).
+ * @return The status of the object (object_status in plasma.h).
  */
-int get_object(int conn, object_id object_id, plasma_object *result);
+int get_object(plasma_store_state *s,
+               int conn,
+               object_id object_id,
+               plasma_object *result);
 
 /**
  * Seal an object:
  *
+ * @param s The plasma store state.
  * @param object_id Object ID of the object to be sealed.
  * @param conns Returns the connection that are waiting for this object.
                 The caller is responsible for destroying this array.
- *
- * Should notify all the sockets waiting for the object.
+ * @return Void.
  */
-plasma_object seal_object(int conn,
-                          object_id object_id,
-                          UT_array **conns,
-                          plasma_object *result);
+void seal_object(plasma_store_state *s,
+                 object_id object_id,
+                 UT_array **conns,
+                 plasma_object *result);
 
 /**
  * Check if the plasma store contains an object:
  *
+ * @param s The plasma store state.
  * @param object_id Object ID that will be checked.
+ * @return OBJECT_FOUND if the object is in the store, OBJECT_NOT_FOUND if not
  */
-int contains_object(int conn, object_id object_id);
+int contains_object(plasma_store_state *s, object_id object_id);
+
+/**
+ * Delete an object from the plasma store:
+ *
+ * @param s The plasma store state.
+ * @param object_id Object ID of the object to be deleted.
+ * @return Void.
+ */
+void delete_object(plasma_store_state *s, object_id object_id);
 
 #endif /* PLASMA_STORE_H */

test/test.py

@@ -1,6 +1,7 @@
 from __future__ import print_function
 
 import os
+import signal
 import socket
 import subprocess
 import sys
@@ -11,6 +12,8 @@ import tempfile
 
 import plasma
 
+USE_VALGRIND = False
+
 def random_object_id():
   return "".join([chr(random.randint(0, 255)) for _ in range(20)])
 
@@ -45,13 +48,24 @@ class TestPlasmaClient(unittest.TestCase):
     # Start Plasma.
     plasma_store_executable = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../build/plasma_store")
     store_name = "/tmp/store{}".format(random.randint(0, 10000))
-    self.p = subprocess.Popen([plasma_store_executable, "-s", store_name])
+    command = [plasma_store_executable, "-s", store_name]
+    if USE_VALGRIND:
+      self.p = subprocess.Popen(["valgrind", "--track-origins=yes", "--leak-check=full"] + command)
+      time.sleep(2.0)
+    else:
+      self.p = subprocess.Popen(command)
     # Connect to Plasma.
     self.plasma_client = plasma.PlasmaClient(store_name)
 
   def tearDown(self):
     # Kill the plasma store process.
-    self.p.kill()
+    if USE_VALGRIND:
+      self.p.send_signal(signal.SIGTERM)
+      self.p.wait()
+      if self.p.returncode != 0:
+        os._exit(-1)
+    else:
+      self.p.kill()
 
   def test_create(self):
     # Create an object id string.
@@ -174,15 +188,32 @@ class TestPlasmaManager(unittest.TestCase):
     plasma_store_executable = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../build/plasma_store")
     store_name1 = "/tmp/store{}".format(random.randint(0, 10000))
     store_name2 = "/tmp/store{}".format(random.randint(0, 10000))
-    self.p2 = subprocess.Popen([plasma_store_executable, "-s", store_name1])
-    self.p3 = subprocess.Popen([plasma_store_executable, "-s", store_name2])
+    plasma_store_command1 = [plasma_store_executable, "-s", store_name1]
+    plasma_store_command2 = [plasma_store_executable, "-s", store_name2]
+
+    if USE_VALGRIND:
+      self.p2 = subprocess.Popen(["valgrind", "--track-origins=yes", "--error-exitcode=1"] + plasma_store_command1)
+      self.p3 = subprocess.Popen(["valgrind", "--track-origins=yes", "--error-exitcode=1"] + plasma_store_command2)
+    else:
+      self.p2 = subprocess.Popen(plasma_store_command1)
+      self.p3 = subprocess.Popen(plasma_store_command2)
+
     # Start two PlasmaManagers.
     self.port1 = random.randint(10000, 50000)
     self.port2 = random.randint(10000, 50000)
     plasma_manager_executable = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../build/plasma_manager")
-    self.p4 = subprocess.Popen([plasma_manager_executable, "-s", store_name1, "-m", "127.0.0.1", "-p", str(self.port1)])
-    self.p5 = subprocess.Popen([plasma_manager_executable, "-s", store_name2, "-m", "127.0.0.1", "-p", str(self.port2)])
-    time.sleep(0.1)
+    plasma_manager_command1 = [plasma_manager_executable, "-s", store_name1, "-m", "127.0.0.1", "-p", str(self.port1)]
+    plasma_manager_command2 = [plasma_manager_executable, "-s", store_name2, "-m", "127.0.0.1", "-p", str(self.port2)]
+
+    if USE_VALGRIND:
+      self.p4 = subprocess.Popen(["valgrind", "--track-origins=yes", "--error-exitcode=1"] + plasma_manager_command1)
+      self.p5 = subprocess.Popen(["valgrind", "--track-origins=yes", "--error-exitcode=1"] + plasma_manager_command2)
+      time.sleep(2.0)
+    else:
+      self.p4 = subprocess.Popen(plasma_manager_command1)
+      self.p5 = subprocess.Popen(plasma_manager_command2)
+      time.sleep(0.1)
+
     # Connect two PlasmaClients.
     self.client1 = plasma.PlasmaClient(store_name1, "127.0.0.1", self.port1)
     self.client2 = plasma.PlasmaClient(store_name2, "127.0.0.1", self.port2)
@@ -190,10 +221,23 @@ class TestPlasmaManager(unittest.TestCase):
 
   def tearDown(self):
     # Kill the PlasmaStore and PlasmaManager processes.
-    self.p2.kill()
-    self.p3.kill()
-    self.p4.kill()
-    self.p5.kill()
+    if USE_VALGRIND:
+      self.p4.send_signal(signal.SIGTERM)
+      self.p4.wait()
+      self.p5.send_signal(signal.SIGTERM)
+      self.p5.wait()
+      self.p2.send_signal(signal.SIGTERM)
+      self.p2.wait()
+      self.p3.send_signal(signal.SIGTERM)
+      self.p3.wait()
+      if self.p2.returncode != 0 or self.p3.returncode != 0 or self.p4.returncode != 0 or self.p5.returncode != 0:
+        print("aborting due to valgrind error")
+        os._exit(-1)
+    else:
+      self.p2.kill()
+      self.p3.kill()
+      self.p4.kill()
+      self.p5.kill()
 
   def test_transfer(self):
     for _ in range(100):
@@ -226,7 +270,7 @@ class TestPlasmaManager(unittest.TestCase):
     # Create an object id string.
     object_id = random_object_id()
     # Create a new buffer.
-    memory_buffer = self.client1.create(object_id, 20000)
+    # memory_buffer = self.client1.create(object_id, 20000)
     # This test is commented out because it currently fails.
     # # Transferring the buffer before sealing it should fail.
     # self.assertRaises(Exception, lambda : self.manager1.transfer(1, object_id))
@@ -246,4 +290,10 @@ class TestPlasmaManager(unittest.TestCase):
     print("it took", b, "seconds to put and transfer the objects")
 
 if __name__ == "__main__":
+  if len(sys.argv) > 1:
+    # pop the argument so we don't mess with unittest's own argument parser
+    arg = sys.argv.pop()
+    if arg == "valgrind":
+      USE_VALGRIND = True
+      print("Using valgrind for tests")
   unittest.main(verbosity=2)