move methods related to plasma requests and replies to plasma.c (#60)

src/plasma/CMakeLists.txt

@@ -80,6 +80,7 @@ include_directories("${CMAKE_SOURCE_DIR}/../common/thirdparty/")
 include_directories("${CMAKE_SOURCE_DIR}/../common/lib/python/")
 
 add_library(plasma SHARED
+  plasma.c
   plasma_extension.c
   plasma_client.c
   fling.c)

src/plasma/Makefile

@@ -13,23 +13,23 @@ clean:
 	cd ../common; make clean
 	rm -rf $(BUILD)/*
 
-$(BUILD)/manager_tests: test/manager_tests.c plasma.h plasma_client.h plasma_client.c plasma_manager.h plasma_manager.c fling.h fling.c common
-	$(CC) $(CFLAGS) $(TEST_CFLAGS) -o $@ test/manager_tests.c plasma_manager.c plasma_client.c fling.c ../common/build/libcommon.a ../common/thirdparty/hiredis/libhiredis.a
+$(BUILD)/manager_tests: test/manager_tests.c plasma.h plasma.c plasma_client.h plasma_client.c plasma_manager.h plasma_manager.c fling.h fling.c common
+	$(CC) $(CFLAGS) $(TEST_CFLAGS) -o $@ test/manager_tests.c plasma.c plasma_manager.c plasma_client.c fling.c ../common/build/libcommon.a ../common/thirdparty/hiredis/libhiredis.a
 
-$(BUILD)/plasma_store: plasma_store.c plasma.h eviction_policy.c fling.h fling.c malloc.c malloc.h thirdparty/dlmalloc.c common
-	$(CC) $(CFLAGS) plasma_store.c eviction_policy.c fling.c malloc.c ../common/build/libcommon.a -o $(BUILD)/plasma_store
+$(BUILD)/plasma_store: plasma_store.c plasma.h plasma.c eviction_policy.c fling.h fling.c malloc.c malloc.h thirdparty/dlmalloc.c common
+	$(CC) $(CFLAGS) plasma_store.c plasma.c eviction_policy.c fling.c malloc.c ../common/build/libcommon.a -o $(BUILD)/plasma_store
 
-$(BUILD)/plasma_manager: plasma_manager.c plasma.h plasma_client.c fling.h fling.c common
-	$(CC) $(CFLAGS) plasma_manager.c plasma_client.c fling.c ../common/build/libcommon.a ../common/thirdparty/hiredis/libhiredis.a -o $(BUILD)/plasma_manager
+$(BUILD)/plasma_manager: plasma_manager.c plasma.h plasma.c plasma_client.c fling.h fling.c common
+	$(CC) $(CFLAGS) plasma_manager.c plasma.c plasma_client.c fling.c ../common/build/libcommon.a ../common/thirdparty/hiredis/libhiredis.a -o $(BUILD)/plasma_manager
 
-$(BUILD)/plasma_client.so: plasma_client.c fling.h fling.c common
-	$(CC) $(CFLAGS) plasma_client.c fling.c ../common/build/libcommon.a -fPIC -shared -o $(BUILD)/plasma_client.so
+$(BUILD)/plasma_client.so: plasma.h plasma.c plasma_client.c fling.h fling.c common
+	$(CC) $(CFLAGS) plasma.c plasma_client.c fling.c ../common/build/libcommon.a -fPIC -shared -o $(BUILD)/plasma_client.so
 
-$(BUILD)/libplasma_client.a: plasma_client.o fling.o
+$(BUILD)/libplasma_client.a: plasma.o plasma_client.o fling.o
 	ar rcs $@ $^
 
-$(BUILD)/example: plasma_client.c plasma.h example.c fling.h fling.c common
-	$(CC) $(CFLAGS) plasma_client.c example.c fling.c ../common/build/libcommon.a -o $(BUILD)/example
+$(BUILD)/example: plasma_client.c plasma.h plasma.c example.c fling.h fling.c common
+	$(CC) $(CFLAGS) plasma_client.c plasma.c example.c fling.c ../common/build/libcommon.a -o $(BUILD)/example
 
 common: FORCE
 	cd ../common; make

src/plasma/fling.c

@@ -17,7 +17,7 @@ void init_msg(struct msghdr *msg,
   msg->msg_namelen = 0;
 }
 
-int send_fd(int conn, int fd, const char *payload, int size) {
+int send_fd(int conn, int fd) {
   struct msghdr msg;
   struct iovec iov;
   char buf[CMSG_SPACE(sizeof(int))];
@@ -31,11 +31,11 @@ int send_fd(int conn, int fd, const char *payload, int size) {
   header->cmsg_len = CMSG_LEN(sizeof(int));
   *(int *) CMSG_DATA(header) = fd;
 
-  /* send file descriptor and payload */
-  return sendmsg(conn, &msg, 0) != -1 && send(conn, payload, size, 0) == -1;
+  /* Send file descriptor. */
+  return sendmsg(conn, &msg, 0);
 }
 
-int recv_fd(int conn, char *payload, int size) {
+int recv_fd(int conn) {
   struct msghdr msg;
   struct iovec iov;
   char buf[CMSG_SPACE(sizeof(int))];
@@ -72,10 +72,5 @@ int recv_fd(int conn, char *payload, int size) {
     return -1;
   }
 
-  ssize_t len = recv(conn, payload, size, 0);
-  if (len < 0) {
-    return -1;
-  }
-
   return found_fd;
 }

src/plasma/fling.h

@@ -25,11 +25,19 @@
 
 void init_msg(struct msghdr *msg, struct iovec *iov, char *buf, size_t buf_len);
 
-/* Send a file descriptor "fd" and a payload "payload" of size "size"
- * over the socket "conn". Return 0 on success. */
-int send_fd(int conn, int fd, const char *payload, int size);
+/**
+ * Send a file descriptor over a unix domain socket.
+ *
+ * @param conn Unix domain socket to send the file descriptor over.
+ * @param fd File descriptor to send over.
+ * @return Status code which is < 0 on failure.
+ */
+int send_fd(int conn, int fd);
 
-/* Receive a file descriptor and a payload of size up to "size" from a
- * socket "conn". The payload will be written to "payload" and the file
- * descriptor will be returned. Returns -1 on failure. */
-int recv_fd(int conn, char *payload, int size);
+/**
+ * Receive a file descriptor over a unix domain socket.
+ *
+ * @param conn Unix domain socket to receive the file descriptor from.
+ * @return File descriptor or a value < 0 on failure.
+ */
+int recv_fd(int conn);

src/plasma/plasma.c

@@ -0,0 +1,90 @@
+#include "plasma.h"
+
+#include "io.h"
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+plasma_request plasma_make_request(object_id object_id) {
+  plasma_request request;
+  memset(&request, 0, sizeof(request));
+  request.num_object_ids = 1;
+  request.object_ids[0] = object_id;
+  return request;
+}
+
+plasma_request *plasma_alloc_request(int num_object_ids,
+                                     object_id object_ids[]) {
+  DCHECK(num_object_ids >= 1);
+  int req_size = plasma_request_size(num_object_ids);
+  plasma_request *req = malloc(req_size);
+  memset(req, 0, req_size);
+  req->num_object_ids = num_object_ids;
+  memcpy(&req->object_ids, object_ids, num_object_ids * sizeof(object_ids[0]));
+  return req;
+}
+
+void plasma_free_request(plasma_request *request) {
+  free(request);
+}
+
+int64_t plasma_request_size(int num_object_ids) {
+  int64_t object_ids_size = (num_object_ids - 1) * sizeof(object_id);
+  return sizeof(plasma_request) + object_ids_size;
+}
+
+plasma_reply plasma_make_reply(object_id object_id) {
+  plasma_reply reply;
+  memset(&reply, 0, sizeof(reply));
+  reply.num_object_ids = 1;
+  reply.object_ids[0] = object_id;
+  return reply;
+}
+
+plasma_reply *plasma_alloc_reply(int num_object_ids) {
+  DCHECK(num_object_ids >= 1);
+  int64_t size = plasma_reply_size(num_object_ids);
+  plasma_reply *reply = malloc(size);
+  memset(reply, 0, size);
+  reply->num_object_ids = num_object_ids;
+  return reply;
+}
+
+void plasma_free_reply(plasma_reply *reply) {
+  free(reply);
+}
+
+int64_t plasma_reply_size(int num_object_ids) {
+  DCHECK(num_object_ids >= 1);
+  return sizeof(plasma_reply) + (num_object_ids - 1) * sizeof(object_id);
+}
+
+int plasma_send_reply(int sock, plasma_reply *reply) {
+  DCHECK(reply);
+  int64_t reply_size = plasma_reply_size(reply->num_object_ids);
+  int n = write(sock, (uint8_t *) reply, reply_size);
+  return n == reply_size ? 0 : -1;
+}
+
+int plasma_receive_reply(int sock, int64_t reply_size, plasma_reply *reply) {
+  int r = recv(sock, reply, reply_size, 0);
+  CHECKM(r != -1, "read error");
+  CHECKM(r != 0, "connection disconnected");
+  return r == reply_size ? 0 : -1;
+}
+
+int plasma_send_request(int sock, int64_t type, plasma_request *request) {
+  DCHECK(request);
+  int req_size = plasma_request_size(request->num_object_ids);
+  int error = write_message(sock, type, req_size, (uint8_t *) request);
+  return error ? -1 : 0;
+}
+
+int plasma_receive_request(int sock, int64_t *type, plasma_request **request) {
+  int64_t length;
+  read_message(sock, type, &length, (uint8_t **) request);
+  if (*request == NULL) {
+    return *type == DISCONNECT_CLIENT;
+  }
+  return length == plasma_request_size((*request)->num_object_ids) ? 0 : -1;
+}

src/plasma/plasma.h

@@ -2,6 +2,7 @@
 #define PLASMA_H
 
 #include <inttypes.h>
+#include <stdlib.h>
 #include <stdio.h>
 #include <errno.h>
 #include <stddef.h>
@@ -144,4 +145,114 @@ typedef struct {
   object_table_entry *objects;
 } plasma_store_info;
 
-#endif
+/**
+ * Create a plasma request with one object ID on the stack.
+ *
+ * @param object_id The object ID to include in the request.
+ * @return The plasma request.
+ */
+plasma_request plasma_make_request(object_id object_id);
+
+/**
+ * Create a plasma request with one or more object IDs on the heap. The caller
+ * must free the returned plasma request pointer with plasma_free_request.
+ *
+ * @param num_object_ids The number of object IDs to include in the request.
+ * @param object_ids The array of object IDs to include in the request. It must
+ *        have length at least equal to num_object_ids.
+ * @return A pointer to the newly created plasma request.
+ */
+plasma_request *plasma_alloc_request(int num_object_ids,
+                                     object_id object_ids[]);
+
+/**
+ * Free a plasma request.
+ *
+ * @param request Pointer to the plasma request to be freed.
+ * @return Void.
+ */
+void plasma_free_request(plasma_request *request);
+
+/**
+ * Size of a request in bytes.
+ *
+ * @param num_object_ids Number of object IDs in the request.
+ * @return The size of the request in bytes.
+ */
+int64_t plasma_request_size(int num_object_ids);
+
+/**
+ * Create a plasma reply with one object ID on the stack.
+ *
+ * @param object_id The object ID to include in the reply.
+ * @return The plasma reply.
+ */
+plasma_reply plasma_make_reply(object_id object_id);
+
+/**
+ * Create a plasma reply with one or more object IDs on the heap. The caller
+ * must free the returned plasma reply pointer with plasma_free_reply.
+ *
+ * @param num_object_ids The number of object IDs to include in the reply.
+ * @return A pointer to the newly created plasma reply.
+ */
+plasma_reply *plasma_alloc_reply(int num_object_ids);
+
+/**
+ * Free a plasma reply.
+ *
+ * @param request Pointer to the plasma reply to be freed.
+ * @return Void.
+ */
+void plasma_free_reply(plasma_reply *request);
+
+/**
+ * Size of a reply in bytes.
+ *
+ * @param num_returns Number of object IDs returned with this reply.
+ * @return The size of the reply in bytes.
+ */
+int64_t plasma_reply_size(int num_returns);
+
+/**
+ * Send a plasma reply.
+ *
+ * @param sock The file descriptor to use to send the request.
+ * @param reply Address of the reply that is sent.
+ * @return Returns a value >= 0 on success.
+ */
+int plasma_send_reply(int sock, plasma_reply *reply);
+
+/**
+ * Receive a plasma reply.
+ *
+ * @param sock The file descriptor to use to get the reply.
+ * @param reply Address of the reply that is received.
+ * @return Returns a value >= 0 on success.
+ */
+int plasma_receive_reply(int sock, int64_t receive_size, plasma_reply *reply);
+
+/**
+ * This is used to send a request to the Plasma Store or
+ * the Plasma Manager.
+ *
+ * @param sock The file descriptor to use to send the request.
+ * @param type The type of request.
+ * @param req The address of the request to send.
+ * @return Returns a value >= 0 on success.
+ */
+int plasma_send_request(int sock, int64_t type, plasma_request *request);
+
+/**
+ * Receive a plasma request. This allocates memory for the request which
+ * needs to be freed by the user.
+ *
+ * @param sock The file descriptor to use to get the reply.
+ * @param type Address where the type of the request is written to.
+ * @param request Address at which the address of the allocated request is
+ *        written.
+ * @return Returns a value >= 0 on success.
+ */
+int plasma_receive_request(int sock, int64_t *type, plasma_request **request);
+
+#endif /* PLASMA_H */

src/plasma/plasma_client.c

@@ -94,35 +94,6 @@ struct plasma_connection {
   plasma_client_config config;
 };
 
-int plasma_request_size(int num_object_ids) {
-  int object_ids_size = (num_object_ids - 1) * sizeof(object_id);
-  return sizeof(plasma_request) + object_ids_size;
-}
-
-void plasma_send_request(int fd, int type, plasma_request *req) {
-  int req_size = plasma_request_size(req->num_object_ids);
-  int error = write_message(fd, type, req_size, (uint8_t *) req);
-  /* TODO(swang): Actually handle the write error. */
-  CHECK(!error);
-}
-
-plasma_request make_plasma_request(object_id object_id) {
-  plasma_request req;
-  memset(&req, 0, sizeof(req));
-  req.num_object_ids = 1;
-  req.object_ids[0] = object_id;
-  return req;
-}
-
-plasma_request *make_plasma_multiple_request(int num_object_ids,
-                                             object_id object_ids[]) {
-  int req_size = plasma_request_size(num_object_ids);
-  plasma_request *req = malloc(req_size);
-  req->num_object_ids = num_object_ids;
-  memcpy(&req->object_ids, object_ids, num_object_ids * sizeof(object_id));
-  return req;
-}
-
 /* If the file descriptor fd has been mmapped in this client process before,
  * return the pointer that was returned by mmap, otherwise mmap it and store the
  * pointer in a hash table. */
@@ -196,12 +167,14 @@ void plasma_create(plasma_connection *conn,
             " and metadata size "
             "%" PRId64,
             conn->store_conn, data_size, metadata_size);
-  plasma_request req = make_plasma_request(object_id);
+  plasma_request req = plasma_make_request(object_id);
   req.data_size = data_size;
   req.metadata_size = metadata_size;
-  plasma_send_request(conn->store_conn, PLASMA_CREATE, &req);
+  CHECK(plasma_send_request(conn->store_conn, PLASMA_CREATE, &req) >= 0);
   plasma_reply reply;
-  int fd = recv_fd(conn->store_conn, (char *) &reply, sizeof(plasma_reply));
+  CHECK(plasma_receive_reply(conn->store_conn, sizeof(reply), &reply) >= 0);
+  int fd = recv_fd(conn->store_conn);
+  CHECK(fd >= 0);
   plasma_object *object = &reply.object;
   CHECK(object->data_size == data_size);
   CHECK(object->metadata_size == metadata_size);
@@ -230,11 +203,12 @@ void plasma_get(plasma_connection *conn,
                 uint8_t **data,
                 int64_t *metadata_size,
                 uint8_t **metadata) {
-  plasma_request req = make_plasma_request(object_id);
-  plasma_send_request(conn->store_conn, PLASMA_GET, &req);
+  plasma_request req = plasma_make_request(object_id);
+  CHECK(plasma_send_request(conn->store_conn, PLASMA_GET, &req) >= 0);
   plasma_reply reply;
-  int fd = recv_fd(conn->store_conn, (char *) &reply, sizeof(plasma_reply));
-  CHECKM(fd != -1, "recv not successful");
+  CHECK(plasma_receive_reply(conn->store_conn, sizeof(reply), &reply) >= 0);
+  int fd = recv_fd(conn->store_conn);
+  CHECK(fd >= 0);
   plasma_object *object = &reply.object;
   *data = lookup_or_mmap(conn, fd, object->handle.store_fd,
                          object->handle.mmap_size) +
@@ -279,8 +253,8 @@ void plasma_perform_release(plasma_connection *conn, object_id object_id) {
       free(entry);
     }
     /* Tell the store that the client no longer needs the object. */
-    plasma_request req = make_plasma_request(object_id);
-    plasma_send_request(conn->store_conn, PLASMA_RELEASE, &req);
+    plasma_request req = plasma_make_request(object_id);
+    CHECK(plasma_send_request(conn->store_conn, PLASMA_RELEASE, &req) >= 0);
     /* Remove the entry from the hash table of objects currently in use. */
     HASH_DELETE(hh, conn->objects_in_use, object_entry);
     free(object_entry);
@@ -309,26 +283,24 @@ void plasma_release(plasma_connection *conn, object_id obj_id) {
 void plasma_contains(plasma_connection *conn,
                      object_id object_id,
                      int *has_object) {
-  plasma_request req = make_plasma_request(object_id);
-  plasma_send_request(conn->store_conn, PLASMA_CONTAINS, &req);
+  plasma_request req = plasma_make_request(object_id);
+  CHECK(plasma_send_request(conn->store_conn, PLASMA_CONTAINS, &req) >= 0);
   plasma_reply reply;
-  int r = read(conn->store_conn, &reply, sizeof(reply));
-  CHECKM(r != -1, "read error");
-  CHECKM(r != 0, "connection disconnected");
+  CHECK(plasma_receive_reply(conn->store_conn, sizeof(reply), &reply) >= 0);
   *has_object = reply.has_object;
 }
 
 void plasma_seal(plasma_connection *conn, object_id object_id) {
-  plasma_request req = make_plasma_request(object_id);
-  plasma_send_request(conn->store_conn, PLASMA_SEAL, &req);
+  plasma_request req = plasma_make_request(object_id);
+  CHECK(plasma_send_request(conn->store_conn, PLASMA_SEAL, &req) >= 0);
   if (conn->manager_conn >= 0) {
-    plasma_send_request(conn->manager_conn, PLASMA_SEAL, &req);
+    CHECK(plasma_send_request(conn->manager_conn, PLASMA_SEAL, &req) >= 0);
   }
 }
 
 void plasma_delete(plasma_connection *conn, object_id object_id) {
-  plasma_request req = make_plasma_request(object_id);
-  plasma_send_request(conn->store_conn, PLASMA_DELETE, &req);
+  plasma_request req = plasma_make_request(object_id);
+  CHECK(plasma_send_request(conn->store_conn, PLASMA_DELETE, &req) >= 0);
 }
 
 int64_t plasma_evict(plasma_connection *conn, int64_t num_bytes) {
@@ -336,12 +308,10 @@ int64_t plasma_evict(plasma_connection *conn, int64_t num_bytes) {
   plasma_request req;
   memset(&req, 0, sizeof(req));
   req.num_bytes = num_bytes;
-  plasma_send_request(conn->store_conn, PLASMA_EVICT, &req);
+  CHECK(plasma_send_request(conn->store_conn, PLASMA_EVICT, &req) >= 0);
   /* Wait for a response with the number of bytes actually evicted. */
   plasma_reply reply;
-  int r = read(conn->store_conn, &reply, sizeof(reply));
-  CHECKM(r != -1, "read error");
-  CHECKM(r != 0, "connection disconnected");
+  CHECK(plasma_receive_reply(conn->store_conn, sizeof(reply), &reply) >= 0);
   return reply.num_bytes;
 }
 
@@ -357,12 +327,10 @@ int plasma_subscribe(plasma_connection *conn) {
   CHECK(fcntl(fd[1], F_SETFL, flags | O_NONBLOCK) == 0);
   /* Tell the Plasma store about the subscription. */
   plasma_request req = {0};
-  plasma_send_request(conn->store_conn, PLASMA_SUBSCRIBE, &req);
+  CHECK(plasma_send_request(conn->store_conn, PLASMA_SUBSCRIBE, &req) >= 0);
   /* Send the file descriptor that the Plasma store should use to push
-   * notifications about sealed objects to this client. We include a one byte
-   * message because otherwise it seems to hang on Linux. */
-  char dummy = '\0';
-  send_fd(conn->store_conn, fd[1], &dummy, 1);
+   * notifications about sealed objects to this client. */
+  CHECK(send_fd(conn->store_conn, fd[1]) >= 0);
   close(fd[1]);
   /* Return the file descriptor that the client should use to read notifications
    * about sealed objects. */
@@ -474,7 +442,7 @@ void plasma_transfer(plasma_connection *conn,
                      const char *addr,
                      int port,
                      object_id object_id) {
-  plasma_request req = make_plasma_request(object_id);
+  plasma_request req = plasma_make_request(object_id);
   req.port = port;
   char *end = NULL;
   for (int i = 0; i < 4; ++i) {
@@ -482,7 +450,7 @@ void plasma_transfer(plasma_connection *conn,
     /* skip the '.' */
     end += 1;
   }
-  plasma_send_request(conn->manager_conn, PLASMA_TRANSFER, &req);
+  CHECK(plasma_send_request(conn->manager_conn, PLASMA_TRANSFER, &req) >= 0);
 }
 
 void plasma_fetch(plasma_connection *conn,
@@ -490,10 +458,9 @@ void plasma_fetch(plasma_connection *conn,
                   object_id object_ids[],
                   int is_fetched[]) {
   CHECK(conn->manager_conn >= 0);
-  plasma_request *req =
-      make_plasma_multiple_request(num_object_ids, object_ids);
+  plasma_request *req = plasma_alloc_request(num_object_ids, object_ids);
   LOG_DEBUG("Requesting fetch");
-  plasma_send_request(conn->manager_conn, PLASMA_FETCH, req);
+  CHECK(plasma_send_request(conn->manager_conn, PLASMA_FETCH, req) >= 0);
   free(req);
 
   plasma_reply reply;
@@ -533,17 +500,14 @@ int plasma_wait(plasma_connection *conn,
                 int num_returns,
                 object_id return_object_ids[]) {
   CHECK(conn->manager_conn >= 0);
-  plasma_request *req =
-      make_plasma_multiple_request(num_object_ids, object_ids);
+  plasma_request *req = plasma_alloc_request(num_object_ids, object_ids);
   req->num_returns = num_returns;
   req->timeout = timeout;
-  plasma_send_request(conn->manager_conn, PLASMA_WAIT, req);
-  free(req);
-  int64_t return_size =
-      sizeof(plasma_reply) + (num_returns - 1) * sizeof(object_id);
+  CHECK(plasma_send_request(conn->manager_conn, PLASMA_WAIT, req) >= 0);
+  plasma_free_request(req);
+  int64_t return_size = plasma_reply_size(num_returns);
   plasma_reply *reply = malloc(return_size);
-  int nbytes = recv(conn->manager_conn, (uint8_t *) reply, return_size, 0);
-  CHECK(nbytes == return_size);
+  CHECK(plasma_receive_reply(conn->manager_conn, return_size, reply) >= 0);
   memcpy(return_object_ids, reply->object_ids, num_returns * sizeof(object_id));
   int num_objects_returned = reply->num_objects_returned;
   free(reply);

src/plasma/plasma_client.h

@@ -9,37 +9,6 @@
 
 typedef struct plasma_connection plasma_connection;
 
-/**
- * This is used by the Plasma Client to send a request to the Plasma Store or
- * the Plasma Manager.
- *
- * @param conn The file descriptor to use to send the request.
- * @param type The type of request.
- * @param req The address of the request to send.
- * @return Void.
- */
-void plasma_send_request(int fd, int type, plasma_request *req);
-
-/**
- * Create a plasma request to be sent with a single object ID.
- *
- * @param object_id The object ID to include in the request.
- * @return The plasma request.
- */
-plasma_request make_plasma_request(object_id object_id);
-
-/**
- * Create a plasma request to be sent with multiple object ID. Caller must free
- * the returned plasma request pointer.
- *
- * @param num_object_ids The number of object IDs to include in the request.
- * @param object_ids The array of object IDs to include in the request. It must
- *        have length at least equal to num_object_ids.
- * @return A pointer to the newly created plasma request.
- */
-plasma_request *make_plasma_multiple_request(int num_object_ids,
-                                             object_id object_ids[]);
-
 /**
  * Try to connect to the socket several times. If unsuccessful, fail.
  *

src/plasma/plasma_manager.c

@@ -147,21 +147,16 @@ void free_client_object_connection(client_object_connection *object_conn) {
   free(object_conn);
 }
 
-int send_client_reply(client_connection *conn, plasma_reply *reply) {
+void send_client_reply(client_connection *conn, plasma_reply *reply) {
   CHECK(conn->num_return_objects >= 0);
   --conn->num_return_objects;
-  /* TODO(swang): Handle errors in write. */
-  int n = write(conn->fd, (uint8_t *) reply, sizeof(*reply));
-  return (n != sizeof(*reply));
+  CHECK(plasma_send_reply(conn->fd, reply) >= 0);
 }
 
-int send_client_failure_reply(object_id object_id, client_connection *conn) {
-  plasma_reply reply;
-  memset(&reply, 0, sizeof(reply));
-  reply.object_ids[0] = object_id;
-  reply.num_object_ids = 1;
+void send_client_failure_reply(object_id object_id, client_connection *conn) {
+  plasma_reply reply = plasma_make_reply(object_id);
   reply.has_object = 0;
-  return send_client_reply(conn, &reply);
+  send_client_reply(conn, &reply);
 }
 
 /**
@@ -350,13 +345,13 @@ void send_queued_request(event_loop *loop,
   }
 
   plasma_request_buffer *buf = conn->transfer_queue;
-  plasma_request manager_req = make_plasma_request(buf->object_id);
+  plasma_request manager_req = plasma_make_request(buf->object_id);
   switch (buf->type) {
   case PLASMA_TRANSFER:
     memcpy(manager_req.addr, conn->manager_state->addr,
            sizeof(manager_req.addr));
     manager_req.port = conn->manager_state->port;
-    plasma_send_request(conn->fd, buf->type, &manager_req);
+    CHECK(plasma_send_request(conn->fd, PLASMA_TRANSFER, &manager_req) >= 0);
     break;
   case PLASMA_DATA:
     LOG_DEBUG("Transferring object to manager");
@@ -366,7 +361,7 @@ void send_queued_request(event_loop *loop,
        * so send the initial PLASMA_DATA request. */
       manager_req.data_size = buf->data_size;
       manager_req.metadata_size = buf->metadata_size;
-      plasma_send_request(conn->fd, PLASMA_DATA, &manager_req);
+      CHECK(plasma_send_request(conn->fd, PLASMA_DATA, &manager_req) >= 0);
     }
     write_object_chunk(conn, buf);
     break;
@@ -587,10 +582,7 @@ int manager_timeout_handler(event_loop *loop, timer_id id, void *context) {
     object_conn->num_retries--;
     return MANAGER_TIMEOUT;
   }
-  plasma_reply reply;
-  memset(&reply, 0, sizeof(reply));
-  reply.object_ids[0] = object_conn->object_id;
-  reply.num_object_ids = 1;
+  plasma_reply reply = plasma_make_reply(object_conn->object_id);
   reply.has_object = 0;
   send_client_reply(client_conn, &reply);
   remove_object_connection(client_conn, object_conn);
@@ -652,10 +644,7 @@ void process_fetch_request(client_connection *client_conn,
                            object_id object_id) {
   client_conn->is_wait = false;
   client_conn->wait_reply = NULL;
-  plasma_reply reply;
-  memset(&reply, 0, sizeof(reply));
-  reply.object_ids[0] = object_id;
-  reply.num_object_ids = 1;
+  plasma_reply reply = plasma_make_reply(object_id);
   if (client_conn->manager_state->db == NULL) {
     reply.has_object = 0;
     send_client_reply(client_conn, &reply);
@@ -691,14 +680,10 @@ void process_fetch_requests(client_connection *client_conn,
 
 void return_from_wait(client_connection *client_conn) {
   CHECK(client_conn->is_wait);
-  int64_t size =
-      sizeof(plasma_reply) +
-      (client_conn->wait_reply->num_object_ids - 1) * sizeof(object_id);
   client_conn->wait_reply->num_objects_returned =
       client_conn->wait_reply->num_object_ids - client_conn->num_return_objects;
-  int n = write(client_conn->fd, (uint8_t *) client_conn->wait_reply, size);
-  CHECK(n == size);
-  free(client_conn->wait_reply);
+  CHECK(plasma_send_reply(client_conn->fd, client_conn->wait_reply) >= 0);
+  plasma_free_reply(client_conn->wait_reply);
   /* Clean the remaining object connections. */
   client_object_connection *object_conn, *tmp;
   HASH_ITER(active_hh, client_conn->active_objects, object_conn, tmp) {
@@ -723,10 +708,7 @@ void process_wait_request(client_connection *client_conn,
   client_conn->is_wait = true;
   client_conn->timer_id = event_loop_add_timer(
       manager_state->loop, timeout, wait_timeout_handler, client_conn);
-  int64_t size = sizeof(plasma_reply) + (num_returns - 1) * sizeof(object_id);
-  client_conn->wait_reply = malloc(size);
-  memset(client_conn->wait_reply, 0, size);
-  client_conn->wait_reply->num_object_ids = num_returns;
+  client_conn->wait_reply = plasma_alloc_reply(num_returns);
   for (int i = 0; i < num_object_ids; ++i) {
     available_object *entry;
     HASH_FIND(hh, manager_state->local_available_objects, &object_ids[i],
@@ -777,10 +759,7 @@ void process_object_notification(event_loop *loop,
   client_connection *client_conn;
   HASH_FIND(fetch_hh, state->fetch_connections, &obj_id, sizeof(object_id),
             object_conn);
-  plasma_reply reply;
-  memset(&reply, 0, sizeof(reply));
-  reply.object_ids[0] = obj_id;
-  reply.num_object_ids = 1;
+  plasma_reply reply = plasma_make_reply(obj_id);
   reply.has_object = 1;
   while (object_conn) {
     next = object_conn->next;
@@ -811,9 +790,8 @@ void process_message(event_loop *loop,
   client_connection *conn = (client_connection *) context;
 
   int64_t type;
-  int64_t length;
   plasma_request *req;
-  read_message(client_sock, &type, &length, (uint8_t **) &req);
+  CHECK(plasma_receive_request(client_sock, &type, &req) >= 0);
 
   switch (type) {
   case PLASMA_TRANSFER:

src/plasma/plasma_store.c

@@ -36,16 +36,6 @@
 void *dlmalloc(size_t);
 void dlfree(void *);
 
-/**
- * This is used by the Plasma Store to send a reply to the Plasma Client.
- */
-void plasma_send_reply(int fd, plasma_reply *reply) {
-  int reply_count = sizeof(*reply);
-  if (write(fd, reply, reply_count) != reply_count) {
-    LOG_FATAL("write error, fd = %d", fd);
-  }
-}
-
 typedef struct {
   /* Object id of this object. */
   object_id object_id;
@@ -306,8 +296,7 @@ void seal_object(client *client_context, object_id object_id) {
   HASH_FIND(handle, plasma_state->objects_notify, &object_id, sizeof(object_id),
             notify_entry);
   if (notify_entry) {
-    plasma_reply reply;
-    memset(&reply, 0, sizeof(reply));
+    plasma_reply reply = plasma_make_reply(NIL_OBJECT_ID);
     plasma_object *result = &reply.object;
     result->handle.store_fd = entry->fd;
     result->handle.mmap_size = entry->map_size;
@@ -319,8 +308,8 @@ void seal_object(client *client_context, object_id object_id) {
     /* Send notifications to the clients that were waiting for this object. */
     for (int i = 0; i < utarray_len(notify_entry->waiting_clients); ++i) {
       client **c = (client **) utarray_eltptr(notify_entry->waiting_clients, i);
-      send_fd((*c)->sock, reply.object.handle.store_fd, (char *) &reply,
-              sizeof(reply));
+      CHECK(plasma_send_reply((*c)->sock, &reply) >= 0);
+      CHECK(send_fd((*c)->sock, reply.object.handle.store_fd) >= 0);
       /* Record that the client is using this object. */
       add_client_to_object_clients(entry, *c);
     }
@@ -411,8 +400,8 @@ void send_notifications(event_loop *loop,
 void subscribe_to_updates(client *client_context, int conn) {
   LOG_DEBUG("subscribing to updates");
   plasma_store_state *plasma_state = client_context->plasma_state;
-  char dummy;
-  int fd = recv_fd(conn, &dummy, 1);
+  int fd = recv_fd(conn);
+  CHECK(fd >= 0);
   CHECKM(HASH_CNT(handle, plasma_state->plasma_store_info->objects) == 0,
          "plasma_subscribe should be called before any objects are created.");
   /* Create a new array to buffer notifications that can't be sent to the
@@ -436,21 +425,20 @@ void process_message(event_loop *loop,
   plasma_request *req = (plasma_request *) utarray_front(state->input_buffer);
 
   /* We're only sending a single object ID at a time for now. */
-  plasma_reply reply;
-  memset(&reply, 0, sizeof(reply));
+  plasma_reply reply = plasma_make_reply(NIL_OBJECT_ID);
   /* Process the different types of requests. */
   switch (type) {
   case PLASMA_CREATE:
     create_object(client_context, req->object_ids[0], req->data_size,
                   req->metadata_size, &reply.object);
-    send_fd(client_sock, reply.object.handle.store_fd, (char *) &reply,
-            sizeof(reply));
+    CHECK(plasma_send_reply(client_sock, &reply) >= 0);
+    CHECK(send_fd(client_sock, reply.object.handle.store_fd) >= 0);
     break;
   case PLASMA_GET:
     if (get_object(client_context, client_sock, req->object_ids[0],
                    &reply.object) == OBJECT_FOUND) {
-      send_fd(client_sock, reply.object.handle.store_fd, (char *) &reply,
-              sizeof(reply));
+      CHECK(plasma_send_reply(client_sock, &reply) >= 0);
+      CHECK(send_fd(client_sock, reply.object.handle.store_fd) >= 0);
     }
     break;
   case PLASMA_RELEASE:
@@ -460,7 +448,7 @@ void process_message(event_loop *loop,
     if (contains_object(client_context, req->object_ids[0]) == OBJECT_FOUND) {
       reply.has_object = 1;
     }
-    plasma_send_reply(client_sock, &reply);
+    CHECK(plasma_send_reply(client_sock, &reply) >= 0);
     break;
   case PLASMA_SEAL:
     seal_object(client_context, req->object_ids[0]);
@@ -480,7 +468,7 @@ void process_message(event_loop *loop,
     remove_objects(client_context->plasma_state, num_objects_to_evict,
                    objects_to_evict);
     reply.num_bytes = num_bytes_evicted;
-    plasma_send_reply(client_sock, &reply);
+    CHECK(plasma_send_reply(client_sock, &reply) >= 0);
     break;
   }
   case PLASMA_SUBSCRIBE: