From 94ad12ff646f84d4f663ce836d41a55253617ab8 Mon Sep 17 00:00:00 2001 From: Philipp Moritz Date: Tue, 11 Oct 2016 17:58:14 -0700 Subject: [PATCH] Get valgrind in place for plasma (#35) * plasma store: refactor and get valgrind in place * fix valgrind errors --- .travis.yml | 11 +++++ src/plasma_client.c | 1 + src/plasma_manager.c | 9 ++++ src/plasma_store.c | 109 ++++++++++++++++++++++++++----------------- src/plasma_store.h | 49 +++++++++++++------ test/test.py | 74 ++++++++++++++++++++++++----- 6 files changed, 184 insertions(+), 69 deletions(-) diff --git a/.travis.yml b/.travis.yml index 5873bfca2..464724a27 100644 --- a/.travis.yml +++ b/.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 diff --git a/src/plasma_client.c b/src/plasma_client.c index bd706ded7..43c448e40 100644 --- a/src/plasma_client.c +++ b/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) + diff --git a/src/plasma_manager.c b/src/plasma_manager.c index 6bda50d93..921d328ee 100644 --- a/src/plasma_manager.c +++ b/src/plasma_manager.c @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -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. */ diff --git a/src/plasma_store.c b/src/plasma_store.c index 128a13292..9edc2c258 100644 --- a/src/plasma_store.c +++ b/src/plasma_store.c @@ -18,6 +18,7 @@ #include #include #include +#include #include #include @@ -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(¬ify_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) { diff --git a/src/plasma_store.h b/src/plasma_store.h index 6b8a0df1d..ae3f03e25 100644 --- a/src/plasma_store.h +++ b/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 */ diff --git a/test/test.py b/test/test.py index fcf0be437..75c072c65 100644 --- a/test/test.py +++ b/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)