From 8147b629643ac98454fb02c9ddc3493c3c91a90e Mon Sep 17 00:00:00 2001 From: Philipp Moritz Date: Wed, 23 Nov 2016 21:13:19 -0800 Subject: [PATCH] move methods related to plasma requests and replies to plasma.c (#60) --- src/plasma/CMakeLists.txt | 1 + src/plasma/Makefile | 22 +++---- src/plasma/fling.c | 13 ++--- src/plasma/fling.h | 22 ++++--- src/plasma/plasma.c | 90 ++++++++++++++++++++++++++++ src/plasma/plasma.h | 113 +++++++++++++++++++++++++++++++++++- src/plasma/plasma_client.c | 104 +++++++++++---------------------- src/plasma/plasma_client.h | 31 ---------- src/plasma/plasma_manager.c | 52 +++++------------ src/plasma/plasma_store.c | 36 ++++-------- 10 files changed, 294 insertions(+), 190 deletions(-) create mode 100644 src/plasma/plasma.c diff --git a/src/plasma/CMakeLists.txt b/src/plasma/CMakeLists.txt index a9c955ff8..085d52e7c 100644 --- a/src/plasma/CMakeLists.txt +++ b/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) diff --git a/src/plasma/Makefile b/src/plasma/Makefile index f0dfe7497..53f36b516 100644 --- a/src/plasma/Makefile +++ b/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 diff --git a/src/plasma/fling.c b/src/plasma/fling.c index e1417aeeb..379b896b5 100644 --- a/src/plasma/fling.c +++ b/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; } diff --git a/src/plasma/fling.h b/src/plasma/fling.h index 56f08dd46..efc41d801 100644 --- a/src/plasma/fling.h +++ b/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); diff --git a/src/plasma/plasma.c b/src/plasma/plasma.c new file mode 100644 index 000000000..550df0f8f --- /dev/null +++ b/src/plasma/plasma.c @@ -0,0 +1,90 @@ +#include "plasma.h" + +#include "io.h" +#include +#include +#include + +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; +} diff --git a/src/plasma/plasma.h b/src/plasma/plasma.h index 5c48f12f3..037a8b75b 100644 --- a/src/plasma/plasma.h +++ b/src/plasma/plasma.h @@ -2,6 +2,7 @@ #define PLASMA_H #include +#include #include #include #include @@ -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 */ diff --git a/src/plasma/plasma_client.c b/src/plasma/plasma_client.c index 85de81148..dd971984f 100644 --- a/src/plasma/plasma_client.c +++ b/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); diff --git a/src/plasma/plasma_client.h b/src/plasma/plasma_client.h index ac167f872..f28770615 100644 --- a/src/plasma/plasma_client.h +++ b/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. * diff --git a/src/plasma/plasma_manager.c b/src/plasma/plasma_manager.c index 1e99e9015..30a8b7df7 100644 --- a/src/plasma/plasma_manager.c +++ b/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: diff --git a/src/plasma/plasma_store.c b/src/plasma/plasma_store.c index 830febe38..88615b3fe 100644 --- a/src/plasma/plasma_store.c +++ b/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: