From 0ca0864856882b76b248b8f7572684c8ec62bcfa Mon Sep 17 00:00:00 2001 From: Philipp Moritz Date: Tue, 20 Dec 2016 14:46:25 -0800 Subject: [PATCH] Use flatcc for serialization of IPC messages. (#140) * added Phllipp's updates * Switch to using flatbuffers for IPC. * Various changes. * convert remaining messages and cleanups * fix * fix function signatures * fix valgrind errors * clang-format * final commit * Fix valgrind test. --- .gitignore | 5 + .travis.yml | 6 +- src/common/CMakeLists.txt | 16 + src/common/io.c | 130 +---- src/common/io.h | 124 +++++ src/photon/Makefile | 6 +- src/photon/photon_client.c | 2 +- src/plasma/CMakeLists.txt | 26 +- src/plasma/Makefile | 38 +- src/plasma/format/plasma.fbs | 266 ++++++++++ src/plasma/plasma.c | 79 +-- src/plasma/plasma.h | 214 +------- src/plasma/plasma_client.c | 238 ++++----- src/plasma/plasma_client.h | 6 +- src/plasma/plasma_extension.c | 7 +- src/plasma/plasma_manager.c | 186 +++---- src/plasma/plasma_protocol.c | 719 ++++++++++++++++++++++++++ src/plasma/plasma_protocol.h | 267 ++++++++++ src/plasma/plasma_store.c | 150 +++--- src/plasma/test/client_tests.c | 19 +- src/plasma/test/manager_tests.c | 45 +- src/plasma/test/serialization_tests.c | 490 ++++++++++++++++++ 22 files changed, 2305 insertions(+), 734 deletions(-) create mode 100644 src/common/CMakeLists.txt create mode 100644 src/plasma/format/plasma.fbs create mode 100644 src/plasma/plasma_protocol.c create mode 100644 src/plasma/plasma_protocol.h create mode 100644 src/plasma/test/serialization_tests.c diff --git a/.gitignore b/.gitignore index aee3fe885..f57b0110e 100644 --- a/.gitignore +++ b/.gitignore @@ -18,6 +18,11 @@ /lib/python/plasma /lib/python/webui +# Files generated by flatcc should be ignored +/src/plasma/format/*_builder.h +/src/plasma/format/*_reader.h +/src/plasma/format/*_verifier.h + # Redis temporary files *dump.rdb diff --git a/.travis.yml b/.travis.yml index 01382bbac..ba17b6fc3 100644 --- a/.travis.yml +++ b/.travis.yml @@ -34,7 +34,9 @@ matrix: before_install: - sudo apt-get update -qq - sudo apt-get install -qq valgrind - script: + install: + - ./.travis/install-dependencies.sh + - cd src/common - make valgrind - cd ../.. @@ -43,6 +45,8 @@ matrix: - make valgrind - cd ../.. + - ./.travis/install-ray.sh + script: - python src/plasma/test/test.py valgrind - python src/photon/test/test.py valgrind - python src/global_scheduler/test/test.py valgrind diff --git a/src/common/CMakeLists.txt b/src/common/CMakeLists.txt new file mode 100644 index 000000000..eb47576cc --- /dev/null +++ b/src/common/CMakeLists.txt @@ -0,0 +1,16 @@ +cmake_minimum_required(VERSION 2.8) + +project(common) + +include(ExternalProject) + +set(FLATCC_PREFIX "${CMAKE_CURRENT_LIST_DIR}/build/flatcc-prefix/src/flatcc") + +ExternalProject_Add(flatcc + URL "https://github.com/dvidelabs/flatcc/archive/v0.4.0.tar.gz" + INSTALL_COMMAND "" + CMAKE_ARGS "-DCMAKE_C_FLAGS=-fPIC") + +set(FLATBUFFERS_INCLUDE_DIR "${FLATCC_PREFIX}/include") +set(FLATBUFFERS_STATIC_LIB "${FLATCC_PREFIX}/lib/libflatcc.a") +set(FLATBUFFERS_COMPILER "${FLATCC_PREFIX}/bin/flatcc") diff --git a/src/common/io.c b/src/common/io.c index 5ec162544..07cba90f4 100644 --- a/src/common/io.c +++ b/src/common/io.c @@ -14,21 +14,6 @@ #include "common.h" -/** - * Binds to an Internet socket at the given port. Removes any existing file at - * the pathname. Returns a non-blocking file descriptor for the socket, or -1 - * if an error occurred. - * - * @note Since the returned file descriptor is non-blocking, it is not - * recommended to use the Linux read and write calls directly, since these - * might read or write a partial message. Instead, use the provided - * write_message and read_message methods. - * - * @param port The port to bind to. - * @param shall_listen Are we also starting to listen on the socket? - * @return A non-blocking file descriptor for the socket, or -1 if an error - * occurs. - */ int bind_inet_sock(const int port, bool shall_listen) { struct sockaddr_in name; int socket_fd = socket(PF_INET, SOCK_STREAM, 0); @@ -65,15 +50,6 @@ int bind_inet_sock(const int port, bool shall_listen) { return socket_fd; } -/** - * Binds to a Unix domain streaming socket at the given - * pathname. Removes any existing file at the pathname. - * - * @param socket_pathname The pathname for the socket. - * @param shall_listen Are we also starting to listen on the socket? - * @return A blocking file descriptor for the socket, or -1 if an error - * occurs. - */ int bind_ipc_sock(const char *socket_pathname, bool shall_listen) { struct sockaddr_un socket_address; int socket_fd = socket(AF_UNIX, SOCK_STREAM, 0); @@ -115,11 +91,6 @@ int bind_ipc_sock(const char *socket_pathname, bool shall_listen) { return socket_fd; } -/** - * Connects to a Unix domain streaming socket at the given - * pathname. Returns a file descriptor for the socket, or -1 if - * an error occurred. - */ int connect_ipc_sock(const char *socket_pathname) { struct sockaddr_un socket_address; int socket_fd; @@ -148,10 +119,6 @@ int connect_ipc_sock(const char *socket_pathname) { return socket_fd; } -/** - * Accept a new client connection on the given socket - * descriptor. Returns a descriptor for the new socket. - */ int accept_client(int socket_fd) { int client_fd = accept(socket_fd, NULL, NULL); if (client_fd < 0) { @@ -161,17 +128,6 @@ int accept_client(int socket_fd) { return client_fd; } -/** - * Write a sequence of bytes into a file descriptor. This will block until one - * of the following happens: (1) there is an error (2) end of file, or (3) all - * length bytes have been written. - * - * @param fd The file descriptor to write to. It can be non-blocking. - * @param cursor The cursor pointing to the beginning of the bytes to send. - * @param length The size of the bytes sequence to write. - * @return int Whether there was an error while writing. 0 corresponds to - * success and -1 corresponds to an error (errno will be set). - */ int write_bytes(int fd, uint8_t *cursor, size_t length) { ssize_t nbytes = 0; size_t bytesleft = length; @@ -197,19 +153,13 @@ int write_bytes(int fd, uint8_t *cursor, size_t length) { return 0; } -/** - * Write a sequence of bytes on a file descriptor. The bytes should then be read - * by read_message. - * - * @param fd The file descriptor to write to. It can be non-blocking. - * @param type The type of the message to send. - * @param length The size in bytes of the bytes parameter. - * @param bytes The address of the message to send. - * @return int Whether there was an error while writing. 0 corresponds to - * success and -1 corresponds to an error (errno will be set). - */ int write_message(int fd, int64_t type, int64_t length, uint8_t *bytes) { + int64_t version = RAY_PROTOCOL_VERSION; int closed; + closed = write_bytes(fd, (uint8_t *) &version, sizeof(version)); + if (closed) { + return closed; + } closed = write_bytes(fd, (uint8_t *) &type, sizeof(type)); if (closed) { return closed; @@ -225,20 +175,6 @@ int write_message(int fd, int64_t type, int64_t length, uint8_t *bytes) { return 0; } -/** - * Read a sequence of bytes from a file descriptor into a buffer. This will - * block until one of the following happens: (1) there is an error (2) end of - * file, or (3) all length bytes have been written. - * - * @note The buffer pointed to by cursor must already have length number of - * bytes allocated before calling this method. - * - * @param fd The file descriptor to read from. It can be non-blocking. - * @param cursor The cursor pointing to the beginning of the buffer. - * @param length The size of the byte sequence to read. - * @return int Whether there was an error while reading. 0 corresponds to - * success and -1 corresponds to an error (errno will be set). - */ int read_bytes(int fd, uint8_t *cursor, size_t length) { ssize_t nbytes = 0; /* Termination condition: EOF or read 'length' bytes total. */ @@ -263,27 +199,14 @@ int read_bytes(int fd, uint8_t *cursor, size_t length) { return 0; } -/** - * Read a sequence of bytes written by write_message from a file descriptor. - * This allocates space for the message. - * - * @note The caller must free the memory. - * - * @param fd The file descriptor to read from. It can be non-blocking. - * @param type The type of the message that is read will be written at this - * address. If there was an error while reading, this will be - * DISCONNECT_CLIENT. - * @param length The size in bytes of the message that is read will be written - * at this address. This size does not include the bytes used to encode - * the type and length. If there was an error while reading, this will - * be 0. - * @param bytes The address at which to write the pointer to the bytes that are - * read and allocated by this function. If there was an error while - * reading, this will be NULL. - * @return Void. - */ void read_message(int fd, int64_t *type, int64_t *length, uint8_t **bytes) { - int closed = read_bytes(fd, (uint8_t *) type, sizeof(*type)); + int64_t version; + int closed = read_bytes(fd, (uint8_t *) &version, sizeof(version)); + if (closed) { + goto disconnected; + } + CHECK(version == RAY_PROTOCOL_VERSION); + closed = read_bytes(fd, (uint8_t *) type, sizeof(*type)); if (closed) { goto disconnected; } @@ -307,26 +230,15 @@ disconnected: return; } -/** - * Read a sequence of bytes written by write_message from a file descriptor. - * This does not allocate space for the message if the provided buffer is - * large enough and can therefore often avoid allocations. - * - * @note The caller must create and free the buffer. - * - * @param fd The file descriptor to read from. It can be non-blocking. - * @param type The type of the message that is read will be written at this - * address. If there was an error while reading, this will be - * DISCONNECT_CLIENT. - * @param buffer The array the message will be written to. If it is not - * large enough to hold the message, it will be enlarged by read_buffer. - * @return Number of bytes of the message that were read. This size does not - * include the bytes used to encode the type and length. If there was - * an error while reading, this will be 0. - */ int64_t read_buffer(int fd, int64_t *type, UT_array *buffer) { + int64_t version; + int closed = read_bytes(fd, (uint8_t *) &version, sizeof(version)); + if (closed) { + goto disconnected; + } + CHECK(version == RAY_PROTOCOL_VERSION); int64_t length; - int closed = read_bytes(fd, (uint8_t *) type, sizeof(*type)); + closed = read_bytes(fd, (uint8_t *) type, sizeof(*type)); if (closed) { goto disconnected; } @@ -348,15 +260,11 @@ disconnected: return 0; } -/* Write a null-terminated string to a file descriptor. */ void write_log_message(int fd, char *message) { /* Account for the \0 at the end of the string. */ write_message(fd, LOG_MESSAGE, strlen(message) + 1, (uint8_t *) message); } -/* Reads a null-terminated string from the file descriptor that has been - * written by write_log_message. Allocates and returns a pointer to the string. - * NOTE: Caller must free the memory! */ char *read_log_message(int fd) { uint8_t *bytes; int64_t type; diff --git a/src/common/io.h b/src/common/io.h index d9a7e1025..6745dd007 100644 --- a/src/common/io.h +++ b/src/common/io.h @@ -6,6 +6,8 @@ #include "utarray.h" +#define RAY_PROTOCOL_VERSION 0x0000000000000000 + enum common_message_type { /** Disconnect a client. */ DISCONNECT_CLIENT, @@ -17,22 +19,144 @@ enum common_message_type { /* Helper functions for socket communication. */ +/** + * Binds to an Internet socket at the given port. Removes any existing file at + * the pathname. Returns a non-blocking file descriptor for the socket, or -1 + * if an error occurred. + * + * @note Since the returned file descriptor is non-blocking, it is not + * recommended to use the Linux read and write calls directly, since these + * might read or write a partial message. Instead, use the provided + * write_message and read_message methods. + * + * @param port The port to bind to. + * @param shall_listen Are we also starting to listen on the socket? + * @return A non-blocking file descriptor for the socket, or -1 if an error + * occurs. + */ int bind_inet_sock(const int port, bool shall_listen); + +/** + * Binds to a Unix domain streaming socket at the given + * pathname. Removes any existing file at the pathname. + * + * @param socket_pathname The pathname for the socket. + * @param shall_listen Are we also starting to listen on the socket? + * @return A blocking file descriptor for the socket, or -1 if an error + * occurs. + */ int bind_ipc_sock(const char *socket_pathname, bool shall_listen); + +/** + * Connects to a Unix domain streaming socket at the given + * pathname. Returns a file descriptor for the socket, or -1 if + * an error occurred. + */ int connect_ipc_sock(const char *socket_pathname); +/** + * Accept a new client connection on the given socket + * descriptor. Returns a descriptor for the new socket. + */ int accept_client(int socket_fd); /* Reading and writing data. */ +/** + * Write a sequence of bytes on a file descriptor. The bytes should then be read + * by read_message. + * + * @param fd The file descriptor to write to. It can be non-blocking. + * @param version The protocol version. + * @param type The type of the message to send. + * @param length The size in bytes of the bytes parameter. + * @param bytes The address of the message to send. + * @return int Whether there was an error while writing. 0 corresponds to + * success and -1 corresponds to an error (errno will be set). + */ int write_message(int fd, int64_t type, int64_t length, uint8_t *bytes); + +/** + * Read a sequence of bytes written by write_message from a file descriptor. + * This allocates space for the message. + * + * @note The caller must free the memory. + * + * @param fd The file descriptor to read from. It can be non-blocking. + * @param type The type of the message that is read will be written at this + * address. If there was an error while reading, this will be + * DISCONNECT_CLIENT. + * @param length The size in bytes of the message that is read will be written + * at this address. This size does not include the bytes used to encode + * the type and length. If there was an error while reading, this will + * be 0. + * @param bytes The address at which to write the pointer to the bytes that are + * read and allocated by this function. If there was an error while + * reading, this will be NULL. + * @return Void. + */ void read_message(int fd, int64_t *type, int64_t *length, uint8_t **bytes); + +/** + * Read a sequence of bytes written by write_message from a file descriptor. + * This does not allocate space for the message if the provided buffer is + * large enough and can therefore often avoid allocations. + * + * @note The caller must create and free the buffer. + * + * @param fd The file descriptor to read from. It can be non-blocking. + * @param type The type of the message that is read will be written at this + * address. If there was an error while reading, this will be + * DISCONNECT_CLIENT. + * @param buffer The array the message will be written to. If it is not + * large enough to hold the message, it will be enlarged by read_buffer. + * @return Number of bytes of the message that were read. This size does not + * include the bytes used to encode the type and length. If there was + * an error while reading, this will be 0. + */ int64_t read_buffer(int fd, int64_t *type, UT_array *buffer); +/** + * Write a null-terminated string to a file descriptor. + */ void write_log_message(int fd, char *message); + void write_formatted_log_message(int fd, const char *format, ...); + +/** + * Reads a null-terminated string from the file descriptor that has been + * written by write_log_message. Allocates and returns a pointer to the string. + * NOTE: Caller must free the memory! + */ char *read_log_message(int fd); + +/** + * Read a sequence of bytes from a file descriptor into a buffer. This will + * block until one of the following happens: (1) there is an error (2) end of + * file, or (3) all length bytes have been written. + * + * @note The buffer pointed to by cursor must already have length number of + * bytes allocated before calling this method. + * + * @param fd The file descriptor to read from. It can be non-blocking. + * @param cursor The cursor pointing to the beginning of the buffer. + * @param length The size of the byte sequence to read. + * @return int Whether there was an error while reading. 0 corresponds to + * success and -1 corresponds to an error (errno will be set). + */ int read_bytes(int fd, uint8_t *cursor, size_t length); + +/** + * Write a sequence of bytes into a file descriptor. This will block until one + * of the following happens: (1) there is an error (2) end of file, or (3) all + * length bytes have been written. + * + * @param fd The file descriptor to write to. It can be non-blocking. + * @param cursor The cursor pointing to the beginning of the bytes to send. + * @param length The size of the bytes sequence to write. + * @return int Whether there was an error while writing. 0 corresponds to + * success and -1 corresponds to an error (errno will be set). + */ int write_bytes(int fd, uint8_t *cursor, size_t length); #endif /* IO_H */ diff --git a/src/photon/Makefile b/src/photon/Makefile index 7f619e199..9be039ab5 100644 --- a/src/photon/Makefile +++ b/src/photon/Makefile @@ -6,16 +6,16 @@ BUILD = build all: $(BUILD)/photon_scheduler $(BUILD)/photon_client.a $(BUILD)/photon_tests: test/photon_tests.c photon.h photon_scheduler.h photon_scheduler.c photon_algorithm.h photon_algorithm.c photon_client.h photon_client.c common - $(CC) $(CFLAGS) $(TEST_CFLAGS) -o $@ test/photon_tests.c photon_scheduler.c photon_algorithm.c photon_client.c ../common/build/libcommon.a ../common/thirdparty/hiredis/libhiredis.a -I../common/thirdparty/ -I../common/ ../plasma/build/libplasma_client.a -I../plasma/ + $(CC) $(CFLAGS) $(TEST_CFLAGS) -o $@ test/photon_tests.c photon_scheduler.c photon_algorithm.c photon_client.c ../common/build/libcommon.a ../common/thirdparty/hiredis/libhiredis.a -I../common/thirdparty/ -I../common/ ../plasma/build/libplasma_client.a ../common/build/flatcc-prefix/src/flatcc/lib/libflatcc.a -I../plasma/ $(BUILD)/photon_client.a: photon_client.o ar rcs $(BUILD)/photon_client.a photon_client.o $(BUILD)/photon_scheduler: photon.h photon_scheduler.c photon_algorithm.c common - $(CC) $(CFLAGS) -o $@ photon_scheduler.c photon_algorithm.c ../common/build/libcommon.a ../common/thirdparty/hiredis/libhiredis.a -I../common/thirdparty/ -I../common/ ../plasma/build/libplasma_client.a + $(CC) $(CFLAGS) -o $@ photon_scheduler.c photon_algorithm.c ../common/build/libcommon.a ../common/thirdparty/hiredis/libhiredis.a -I../common/thirdparty/ -I../common/ ../plasma/build/libplasma_client.a ../common/build/flatcc-prefix/src/flatcc/lib/libflatcc.a common: FORCE - cd ../common; make + cd ../common; make; cd build; cmake ..; make clean: cd ../common; make clean diff --git a/src/photon/photon_client.c b/src/photon/photon_client.c index 6526f5984..18b030d76 100644 --- a/src/photon/photon_client.c +++ b/src/photon/photon_client.c @@ -29,7 +29,7 @@ task_spec *photon_get_task(photon_conn *conn) { * scheduler gives this client a task. */ read_message(conn->conn, &type, &length, &message); CHECK(type == EXECUTE_TASK); - task_spec *task = (task_spec *)message; + task_spec *task = (task_spec *) message; CHECK(length == task_spec_size(task)); return task; } diff --git a/src/plasma/CMakeLists.txt b/src/plasma/CMakeLists.txt index 1bff5fde4..ba8516ea9 100644 --- a/src/plasma/CMakeLists.txt +++ b/src/plasma/CMakeLists.txt @@ -2,6 +2,9 @@ cmake_minimum_required(VERSION 2.8) project(plasma) +# Recursively include common +include(${CMAKE_CURRENT_SOURCE_DIR}/../common/CMakeLists.txt) + message(STATUS "Trying custom approach for finding Python.") # Start off by figuring out which Python executable to use. find_program(CUSTOM_PYTHON_EXECUTABLE python) @@ -61,7 +64,21 @@ endif(APPLE) include_directories("${PYTHON_INCLUDE_DIRS}" thirdparty) -set(CMAKE_C_FLAGS "${CMAKE_CXX_FLAGS} --std=c99 -D_XOPEN_SOURCE=500 -D_POSIX_C_SOURCE=200809L") +set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} --std=c99 -D_XOPEN_SOURCE=500 -D_POSIX_C_SOURCE=200809L") + +# Compile flatbuffers + +set(PLASMA_FBS_SRC "${CMAKE_SOURCE_DIR}/format/plasma.fbs") +set(OUTPUT_DIR ${CMAKE_SOURCE_DIR}/format/) + +add_custom_target(gen_plasma_fbs ALL) + +add_custom_command( + TARGET gen_plasma_fbs + COMMAND ${FLATBUFFERS_COMPILER} -a -c -o ${OUTPUT_DIR} ${PLASMA_FBS_SRC} + DEPENDS ${FBS_DEPENDS} + COMMENT "Running flatc compiler on ${PLASMA_FBS_SRC}" + VERBATIM) if(UNIX AND NOT APPLE) link_libraries(rt) @@ -76,19 +93,24 @@ include_directories("${CMAKE_SOURCE_DIR}/../common/") include_directories("${CMAKE_SOURCE_DIR}/../common/thirdparty/") include_directories("${CMAKE_SOURCE_DIR}/../common/lib/python/") +include_directories("${FLATBUFFERS_INCLUDE_DIR}") + add_library(plasma SHARED plasma.c plasma_extension.c + plasma_protocol.c plasma_client.c thirdparty/xxhash.c fling.c) +add_dependencies(plasma gen_plasma_fbs) + get_filename_component(PYTHON_SHARED_LIBRARY ${PYTHON_LIBRARIES} NAME) if(APPLE) add_custom_command(TARGET plasma POST_BUILD COMMAND ${CMAKE_INSTALL_NAME_TOOL} -change ${PYTHON_SHARED_LIBRARY} ${PYTHON_LIBRARIES} libplasma.so) endif(APPLE) -target_link_libraries(plasma ${COMMON_LIB} ${PYTHON_LIBRARIES}) +target_link_libraries(plasma ${COMMON_LIB} ${PYTHON_LIBRARIES} ${FLATBUFFERS_STATIC_LIB}) install(TARGETS plasma DESTINATION ${CMAKE_SOURCE_DIR}/plasma) diff --git a/src/plasma/Makefile b/src/plasma/Makefile index 40892ca87..1d326ed4a 100644 --- a/src/plasma/Makefile +++ b/src/plasma/Makefile @@ -1,5 +1,6 @@ CC = gcc -CFLAGS = -g -Wall -Wextra -Werror=implicit-function-declaration -Wno-sign-compare -Wno-unused-parameter -Wno-type-limits -Wno-missing-field-initializers --std=c99 -D_XOPEN_SOURCE=500 -D_POSIX_C_SOURCE=200809L -I. -Ithirdparty -I../common -I../common/thirdparty +# The -rdynamic is here so we always get function names in backtraces. +CFLAGS = -g -Wall -rdynamic -Wextra -Werror=implicit-function-declaration -Wno-sign-compare -Wno-unused-parameter -Wno-type-limits -Wno-missing-field-initializers --std=c99 -D_XOPEN_SOURCE=500 -D_POSIX_C_SOURCE=200809L -I. -Ithirdparty -I../common -I../common/thirdparty -I../common/build/flatcc-prefix/src/flatcc/include TEST_CFLAGS = -DPLASMA_TEST=1 -I. BUILD = build @@ -14,35 +15,39 @@ clean: rm -f *.o rm -rf $(BUILD)/* -$(BUILD)/manager_tests: test/manager_tests.c plasma.h plasma.c plasma_client.h plasma_client.c thirdparty/xxhash.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 thirdparty/xxhash.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 thirdparty/xxhash.c plasma_protocol.h plasma_protocol.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 thirdparty/xxhash.c plasma_protocol.c fling.c ../common/build/libcommon.a ../common/build/flatcc-prefix/src/flatcc/lib/libflatcc.a ../common/thirdparty/hiredis/libhiredis.a -$(BUILD)/client_tests: test/client_tests.c plasma.h plasma.c plasma_client.h plasma_client.c thirdparty/xxhash.c plasma_manager.h plasma_manager.c fling.h fling.c common - $(CC) $(CFLAGS) $(TEST_CFLAGS) -o $@ test/client_tests.c plasma.c plasma_manager.c plasma_client.c thirdparty/xxhash.c fling.c ../common/build/libcommon.a ../common/thirdparty/hiredis/libhiredis.a +$(BUILD)/client_tests: test/client_tests.c plasma.h plasma.c plasma_client.h plasma_client.c thirdparty/xxhash.c plasma_protocol.h plasma_protocol.c plasma_manager.h plasma_manager.c fling.h fling.c common + $(CC) $(CFLAGS) $(TEST_CFLAGS) -o $@ test/client_tests.c plasma.c plasma_manager.c plasma_client.c thirdparty/xxhash.c plasma_protocol.c fling.c ../common/build/libcommon.a ../common/build/flatcc-prefix/src/flatcc/lib/libflatcc.a ../common/thirdparty/hiredis/libhiredis.a -$(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)/serialization_tests: test/serialization_tests.c plasma.h plasma.c plasma_protocol.h plasma_protocol.c common + $(CC) $(CFLAGS) $(TEST_CFLAGS) -o $@ test/serialization_tests.c plasma.c plasma_protocol.c ../common/build/libcommon.a ../common/build/flatcc-prefix/src/flatcc/lib/libflatcc.a -$(BUILD)/plasma_manager: plasma_manager.c plasma.h plasma.c plasma_client.c thirdparty/xxhash.c fling.h fling.c common - $(CC) $(CFLAGS) plasma_manager.c plasma.c plasma_client.c thirdparty/xxhash.c fling.c ../common/build/libcommon.a ../common/thirdparty/hiredis/libhiredis.a -o $(BUILD)/plasma_manager +$(BUILD)/plasma_store: plasma_store.c plasma.h plasma.c plasma_protocol.c eviction_policy.c fling.h fling.c malloc.c malloc.h thirdparty/dlmalloc.c common + $(CC) $(CFLAGS) plasma_store.c plasma.c plasma_protocol.c eviction_policy.c fling.c malloc.c ../common/build/libcommon.a ../common/build/flatcc-prefix/src/flatcc/lib/libflatcc.a -o $(BUILD)/plasma_store -$(BUILD)/plasma_client.so: plasma.h plasma.c plasma_client.c fling.h fling.c common - $(CC) $(CFLAGS) plasma.c plasma_client.c thirdparty/xxhash.c fling.c ../common/build/libcommon.a -fPIC -shared -o $(BUILD)/plasma_client.so +$(BUILD)/plasma_manager: plasma_manager.c plasma.h plasma.c plasma_protocol.c plasma_client.c thirdparty/xxhash.c fling.h fling.c common + $(CC) $(CFLAGS) plasma_manager.c plasma.c plasma_protocol.c plasma_client.c thirdparty/xxhash.c fling.c ../common/build/libcommon.a ../common/build/flatcc-prefix/src/flatcc/lib/libflatcc.a ../common/thirdparty/hiredis/libhiredis.a -o $(BUILD)/plasma_manager -$(BUILD)/libplasma_client.a: plasma.o plasma_client.o fling.o thirdparty/xxhash.o +$(BUILD)/plasma_client.so: plasma.h plasma.c plasma_protocol.c plasma_client.c thirdparty/xxhash.c fling.h fling.c common + $(CC) $(CFLAGS) plasma.c plasma_protocol.c plasma_client.c thirdparty/xxhash.c fling.c ../common/build/libcommon.a ../common/build/flatcc-prefix/src/flatcc/lib/libflatcc.a -fPIC -shared -o $(BUILD)/plasma_client.so + +$(BUILD)/libplasma_client.a: plasma.o plasma_protocol.o plasma_client.o thirdparty/xxhash.o fling.o ../common/build/flatcc-prefix/src/flatcc/lib/libflatcc.a ar rcs $@ $^ -$(BUILD)/example: plasma_client.c thirdparty/xxhash.c plasma.h plasma.c example.c fling.h fling.c common - $(CC) $(CFLAGS) plasma_client.c thirdparty/xxhash.c plasma.c example.c fling.c ../common/build/libcommon.a -o $(BUILD)/example +$(BUILD)/example: plasma_client.c thirdparty/xxhash.c plasma.h plasma.c plasma_protocol.c example.c fling.h fling.c common + $(CC) $(CFLAGS) plasma_client.c thirdparty/xxhash.c plasma.c plasma_protocol.c example.c fling.c ../common/build/libcommon.a ../common/build/flatcc-prefix/src/flatcc/lib/libflatcc.a -o $(BUILD)/example common: FORCE - cd ../common; make + cd ../common; make; cd build; cmake ..; make; cd ../../plasma/build; cmake ..; make # Set the request timeout low and logging level at FATAL for testing purposes. test: CFLAGS += -DRAY_TIMEOUT=50 -DRAY_COMMON_LOG_LEVEL=4 # First, build and run all the unit tests. -test: $(BUILD)/manager_tests $(BUILD)/client_tests FORCE +test: $(BUILD)/manager_tests $(BUILD)/client_tests $(BUILD)/serialization_tests FORCE ./build/manager_tests + ./build/serialization_tests ./test/run_client_tests.sh cd ../common; make redis # Next, build all the executables for Python testing. @@ -50,5 +55,6 @@ test: all valgrind: test valgrind --leak-check=full --error-exitcode=1 ./build/manager_tests + valgrind --leak-check=full --error-exitcode=1 ./build/serialization_tests FORCE: diff --git a/src/plasma/format/plasma.fbs b/src/plasma/format/plasma.fbs new file mode 100644 index 000000000..5b4402154 --- /dev/null +++ b/src/plasma/format/plasma.fbs @@ -0,0 +1,266 @@ +// Plasma protocol specification + +enum MessageType:int { + // Create a new object. + PlasmaCreateRequest = 1, + PlasmaCreateReply, + // Seal an object. + PlasmaSealRequest, + PlasmaSealReply, + // Get an object that is stored on the local Plasma store. + PlasmaGetRequest, + PlasmaGetReply, + // Non-blocking version of PlasmaGet + PlasmaGetLocalRequest, + PlasmaGetLocalReply, + // Release an object. + PlasmaReleaseRequest, + PlasmaReleaseReply, + // Delete an object. + PlasmaDeleteRequest, + PlasmaDeleteReply, + // Get status of an object. + PlasmaStatusRequest, + PlasmaStatusReply, + // See if the store contains an object (will be deprecated). + PlasmaContainsRequest, + PlasmaContainsReply, + // Make room for new objects in the plasma store. + PlasmaEvictRequest, + PlasmaEvictReply, + // Fetch objects from remote Plasma stores. + PlasmaFetchRequest, + // Wait for objects to be ready either from local or remote Plasma stores. + PlasmaWaitRequest, + PlasmaWaitReply, + // Subscribe to a list of objects or to all objects. + PlasmaSubscribeRequest, + // Unsubscribe. + PlasmaUnsubscribeRequest, + // Sending and receiving data. + PlasmaDataRequest, + PlasmaDataReply +} + +enum PlasmaError:int { + // Operation was successful. + OK, + // Trying to create an object that already exists. + ObjectExists, + // Trying to access an object that doesn't exist. + ObjectNonexistent +} + +// Plasma store messages + +struct PlasmaObject { + // Index of the memory segment (= memory mapped file) that + // this object is allocated in. + segment_index: int; + // Size in bytes of this segment (needed to call mmap). + mmap_size: ulong; + // The offset in bytes in the memory mapped file of the data. + data_offset: ulong; + // The size in bytes of the data. + data_size: ulong; + // The offset in bytes in the memory mapped file of the metadata. + metadata_offset: ulong; + // The size in bytes of the metadata. + metadata_size: ulong; +} + +table PlasmaCreateRequest { + // ID of the object to be created. + object_id: string; + // The size of the object's data in bytes. + data_size: ulong; + // The size of the object's metadata in bytes. + metadata_size: ulong; +} + +table PlasmaCreateReply { + // ID of the object that was created. + object_id: string; + // The object that is returned with this reply. + plasma_object: PlasmaObject; + // Error that occurred for this call. + error: PlasmaError; +} + +table PlasmaSealRequest { + // ID of the object to be sealed. + object_id: string; + // Hash of the object data. + digest: [ubyte]; +} + +table PlasmaSealReply { + // ID of the object that was sealed. + object_id: string; + // Error code. + error: PlasmaError; +} + +table PlasmaGetRequest { + // IDs of the objects stored at local Plasma store we are getting. + object_ids: [string]; +} + +table PlasmaGetReply { + // IDs of the objects being returned. + // This number can be smaller than the number of requested + // objects if not all requested objects are stored and sealed + // in the local Plasma store. + object_ids: [string]; + // Plasma object information, in the same order as their IDs. + plasma_objects: [PlasmaObject]; + // The number of elements in both object_ids and plasma_objects arrays must agree. +} + +table PlasmaGetLocalRequest { + // IDs of the objects stored at local Plasma store we are getting. + object_ids: [string]; +} + +table PlasmaGetLocalReply { + // IDs of the objects being returned. + object_ids: [string]; + // Plasma object information, in the same order as their IDs. + plasma_objects: [PlasmaObject]; + // Indication if the object is local it's file descriptor has been transfered, + // same order as their IDs. + has_object: [int]; +} + +table PlasmaReleaseRequest { + // ID of the object to be released. + object_id: string; +} + +table PlasmaReleaseReply { + // ID of the object that was released. + object_id: string; + // Error code. + error: PlasmaError; +} + +table PlasmaDeleteRequest { + // ID of the object to be deleted. + object_id: string; +} + +table PlasmaDeleteReply { + // ID of the object that was deleted. + object_id: string; + // Error code. + error: PlasmaError; +} + +table PlasmaStatusRequest { + // IDs of the objects stored at local Plasma store we request the status of. + object_ids: [string]; +} + +enum ObjectStatus:int { + // Object is stored in the local Plasma Store. + Local = 1, + // Object is stored on a remote Plasma store, and it is not stored on the + // local Plasma Store. + Remote, + // Object is not stored in the system. + Nonexistent, + // Object is currently transferred from a remote Plasma store the the local + // Plasma Store. + Transfer +} + +table PlasmaStatusReply { + // IDs of the objects being returned. + object_ids: [string]; + // Status of the object. + status: [ObjectStatus]; +} + +// PlasmaContains is a subset of PlasmaStatus which does not +// involve the plasma manager, only the store. We should consider +// unifying them in the future and deprecating PlasmaContains. + +table PlasmaContainsRequest { + // ID of the object we are querying. + object_id: string; +} + +table PlasmaContainsReply { + // ID of the object we are querying. + object_id: string; + // 1 if the object is in the store and 0 otherwise. + has_object: int; +} + +table PlasmaEvictRequest { + // Number of bytes that shall be freed. + num_bytes: ulong; +} + +table PlasmaEvictReply { + // Number of bytes that have been freed. + num_bytes: ulong; +} + +table PlasmaFetchRequest { + // IDs of objects to be gotten. + object_ids: [string]; +} + +table ObjectRequest { + // ID of the object. + object_id: string; + // The type of the object. This specifies whether we + // will be waiting for an object store in the local or + // global Plasma store. + type: int; +} + +table PlasmaWaitRequest { + // Array of object requests whose status we are asking for. + object_requests: [ObjectRequest]; + // Number of objects expected to be returned, if available. + num_ready_objects: int; + // timeout + timeout: long; +} + +table ObjectReply { + // ID of the object. + object_id: string; + // The object status. This specifies where the object is stored. + status: int; +} + +table PlasmaWaitReply { + // Array of object requests being returned. + object_requests: [ObjectReply]; + // Number of objects expected to be returned, if available. + num_ready_objects: int; +} + +table PlasmaSubscribeRequest { +} + +table PlasmaDataRequest { + // ID of the object that is requested. + object_id: string; + // The host address where the data shall be sent to. + address: string; + // The port of the manager the data shall be sent to. + port: int; +} + +table PlasmaDataReply { + // ID of the object that will be sent. + object_id: string; + // Size of the object data in bytes. + object_size: ulong; + // Size of the metadata in bytes. + metadata_size: ulong; +} diff --git a/src/plasma/plasma.c b/src/plasma/plasma.c index b6ad44768..4e3eb0c3b 100644 --- a/src/plasma/plasma.c +++ b/src/plasma/plasma.c @@ -5,84 +5,7 @@ #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_requests[0].object_id = object_id; - return request; -} - -plasma_request *plasma_alloc_request(int num_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; - 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_request); - 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_requests[0].object_id = 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_request); -} - -int plasma_send_reply(int sock, plasma_reply *reply) { - DCHECK(reply); - int64_t reply_size = plasma_reply_size(reply->num_object_ids); - return write_bytes(sock, (uint8_t *) reply, reply_size); -} - -int plasma_receive_reply(int sock, int64_t reply_size, plasma_reply *reply) { - return read_bytes(sock, (uint8_t *) reply, reply_size); -} - -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; - } - int req_size = plasma_request_size((*request)->num_object_ids); - return length == req_size ? 0 : -1; -} +#include "plasma_protocol.h" bool plasma_object_ids_distinct(int num_object_ids, object_id object_ids[]) { for (int i = 0; i < num_object_ids; ++i) { diff --git a/src/plasma/plasma.h b/src/plasma/plasma.h index a0cb531fe..51e7682f4 100644 --- a/src/plasma/plasma.h +++ b/src/plasma/plasma.h @@ -31,9 +31,9 @@ typedef struct { int type; /** Object status. Same as the status returned by plasma_status() function * call. This is filled in by plasma_wait_for_objects1(): - * - PLASMA_OBJECT_LOCAL: object is ready at the local Plasma Store. - * - PLASMA_OBJECT_REMOTE: object is ready at a remote Plasma Store. - * - PLASMA_OBJECT_NONEXISTENT: object does not exist in the system. + * - ObjectStatus_Local: object is ready at the local Plasma Store. + * - ObjectStatus_Remote: object is ready at a remote Plasma Store. + * - ObjectStatus_Nonexistent: object does not exist in the system. * - PLASMA_CLIENT_IN_TRANSFER, if the object is currently being scheduled * for being transferred or it is transferring. */ int status; @@ -76,19 +76,6 @@ typedef enum { OBJECT_FOUND = 1 } object_status; -typedef enum { - /** Object is stored in the local Plasma Store. */ - PLASMA_OBJECT_LOCAL = 1, - /** Object is stored on a remote Plasma store, and it is not stored on the - * local Plasma Store. */ - PLASMA_OBJECT_REMOTE, - /** Object is currently transferred from a remote Plasma store the the local - * Plasma Store. */ - PLASMA_OBJECT_IN_TRANSFER, - /** Object is not stored in the system. */ - PLASMA_OBJECT_NONEXISTENT -} object_status1; - typedef enum { /** Query for object in the local plasma store. */ PLASMA_QUERY_LOCAL = 1, @@ -96,94 +83,6 @@ typedef enum { PLASMA_QUERY_ANYWHERE } object_request_type; -enum plasma_message_type { - /** Create a new object. */ - PLASMA_CREATE = 128, - /** Get an object. */ - PLASMA_GET, - /** Get an object stored at the local Plasma Store. */ - PLASMA_GET_LOCAL, - /** Tell the store that the client no longer needs an object. */ - PLASMA_RELEASE, - /** Check if an object is present. */ - PLASMA_CONTAINS, - /** Seal an object. */ - PLASMA_SEAL, - /** Delete an object. */ - PLASMA_DELETE, - /** Evict objects from the store. */ - PLASMA_EVICT, - /** Subscribe to notifications about sealed objects. */ - PLASMA_SUBSCRIBE, - /** Request transfer to another store. */ - PLASMA_TRANSFER, - /** Header for sending data. */ - PLASMA_DATA, - /** Request a fetch of an object in another store. Non-blocking call. */ - PLASMA_FETCH, - /** Request status of an object, i.e., whether the object is stored in the - * local Plasma Store, in a remote Plasma Store, in transfer, or doesn't - * exist in the system. */ - PLASMA_STATUS, - /** Wait until an object becomes available. */ - PLASMA_WAIT -}; - -typedef struct { - /** The size of the object's data. */ - int64_t data_size; - /** The size of the object's metadata. */ - int64_t metadata_size; - /** The timeout of the request. */ - uint64_t timeout; - /** The number of objects we are waiting for to be ready. */ - int num_ready_objects; - /** In a transfer request, this is the IP address of the Plasma Manager to - * transfer the object to. */ - uint8_t addr[4]; - /** In a transfer request, this is the port of the Plasma Manager to transfer - * the object to. */ - int port; - /** A number of bytes. This is used for eviction requests. */ - int64_t num_bytes; - /** A digest describing the object. This is used for detecting - * nondeterministic tasks. */ - unsigned char digest[DIGEST_SIZE]; - /** The number of object IDs that will be included in this request. */ - int num_object_ids; - /** The object requests that the request is about. */ - object_request object_requests[1]; -} plasma_request; - -typedef enum { - /** There is no error. */ - PLASMA_REPLY_OK = 1, - /** The object already exists. */ - PLASMA_OBJECT_ALREADY_EXISTS -} plasma_error; - -typedef struct { - /** The object that is returned with this reply. */ - plasma_object object; - /** TODO: document this. */ - int object_status; - /** This is used only to respond to requests of type - * PLASMA_CONTAINS or PLASMA_FETCH. It is 1 if the object is - * present and 0 otherwise. Used for plasma_contains and - * plasma_fetch. */ - int has_object; - /** A number of bytes. This is used for replies to eviction requests. */ - int64_t num_bytes; - /** Number of object IDs a wait is returning. */ - int num_objects_returned; - /** The number of object IDs that will be included in this reply. */ - int num_object_ids; - /** The object requests that this reply refers to. */ - object_request object_requests[1]; - /** Return error code. */ - plasma_error error_code; -} plasma_reply; - /** This type is used by the Plasma store. It is here because it is exposed to * the eviction policy. */ typedef struct { @@ -224,113 +123,6 @@ typedef struct { unsigned char digest[DIGEST_SIZE]; } object_id_notification; -/** - * 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. - * @return A pointer to the newly created plasma request. - */ -plasma_request *plasma_alloc_request(int num_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); - /** * Check if a collection of object IDs contains any duplicates. * diff --git a/src/plasma/plasma_client.c b/src/plasma/plasma_client.c index 696220fc6..c4aa48305 100644 --- a/src/plasma/plasma_client.c +++ b/src/plasma/plasma_client.c @@ -23,6 +23,7 @@ #include "common.h" #include "io.h" #include "plasma.h" +#include "plasma_protocol.h" #include "plasma_client.h" #include "fling.h" #include "uthash.h" @@ -93,6 +94,8 @@ struct plasma_connection { * notifications for the objects it subscribes for when these objects are * sealed either locally or remotely. */ int manager_conn_subscribe; + /** Buffer that holds memory for serializing plasma protocol messages. */ + protocol_builder *builder; /** Table of dlmalloc buffer files that have been memory mapped so far. This * is a hash table mapping a file descriptor to a struct containing the * address of the corresponding memory-mapped file. */ @@ -184,7 +187,7 @@ void increment_object_count(plasma_connection *conn, } bool plasma_create(plasma_connection *conn, - object_id object_id, + object_id obj_id, int64_t data_size, uint8_t *metadata, int64_t metadata_size, @@ -192,52 +195,53 @@ bool plasma_create(plasma_connection *conn, LOG_DEBUG("called plasma_create on conn %d with size %" PRId64 " and metadata size %" PRId64, conn->store_conn, data_size, metadata_size); - plasma_request req = plasma_make_request(object_id); - req.data_size = data_size; - req.metadata_size = metadata_size; - CHECK(plasma_send_request(conn->store_conn, PLASMA_CREATE, &req) >= 0); - plasma_reply reply; - CHECK(plasma_receive_reply(conn->store_conn, sizeof(reply), &reply) >= 0); - int fd = recv_fd(conn->store_conn); - CHECKM(fd >= 0, "recv not successful"); - if (reply.error_code == PLASMA_OBJECT_ALREADY_EXISTS) { - LOG_DEBUG("returned from plasma_create with error %d", reply.error_code); - close(fd); + CHECK(plasma_send_CreateRequest(conn->store_conn, conn->builder, obj_id, + data_size, metadata_size) >= 0); + uint8_t *reply_data = + plasma_receive(conn->store_conn, MessageType_PlasmaCreateReply); + int error; + object_id id; + plasma_object object; + plasma_read_CreateReply(reply_data, &id, &object, &error); + free(reply_data); + if (error == PlasmaError_ObjectExists) { + LOG_DEBUG("returned from plasma_create with error %d", error); return false; } - plasma_object *object = &reply.object; - CHECK(object->data_size == data_size); - CHECK(object->metadata_size == metadata_size); + int fd = recv_fd(conn->store_conn); + CHECKM(fd >= 0, "recv not successful"); + CHECK(object.data_size == data_size); + CHECK(object.metadata_size == metadata_size); /* The metadata should come right after the data. */ - CHECK(object->metadata_offset == object->data_offset + data_size); - *data = lookup_or_mmap(conn, fd, object->handle.store_fd, - object->handle.mmap_size) + - object->data_offset; + CHECK(object.metadata_offset == object.data_offset + data_size); + *data = lookup_or_mmap(conn, fd, object.handle.store_fd, + object.handle.mmap_size) + + object.data_offset; /* If plasma_create is being called from a transfer, then we will not copy the * metadata here. The metadata will be written along with the data streamed * from the transfer. */ if (metadata != NULL) { /* Copy the metadata to the buffer. */ - memcpy(*data + object->data_size, metadata, metadata_size); + memcpy(*data + object.data_size, metadata, metadata_size); } /* Increment the count of the number of instances of this object that this * client is using. A call to plasma_release is required to decrement this * count. Cache the reference to the object. */ - increment_object_count(conn, object_id, object, false); + increment_object_count(conn, obj_id, &object, false); return true; } /* This method is used to get both the data and the metadata. */ void plasma_get(plasma_connection *conn, - object_id object_id, + object_id obj_id, int64_t *size, uint8_t **data, int64_t *metadata_size, uint8_t **metadata) { /* Check if we already have a reference to the object. */ object_in_use_entry *object_entry; - HASH_FIND(hh, conn->objects_in_use, &object_id, sizeof(object_id), - object_entry); + HASH_FIND(hh, conn->objects_in_use, &obj_id, sizeof(object_id), object_entry); + plasma_object object_data; plasma_object *object; if (object_entry) { /* If we have already have a reference to the object, use it to get the @@ -250,13 +254,17 @@ void plasma_get(plasma_connection *conn, *data = lookup_mmapped_file(conn, object->handle.store_fd); } else { /* Else, request a reference to the object data from the plasma store. */ - plasma_request req = plasma_make_request(object_id); - CHECK(plasma_send_request(conn->store_conn, PLASMA_GET, &req) >= 0); - plasma_reply reply; - CHECK(plasma_receive_reply(conn->store_conn, sizeof(reply), &reply) >= 0); + CHECK(plasma_send_GetRequest(conn->store_conn, conn->builder, &obj_id, 1) >= + 0); + uint8_t *reply_data = + plasma_receive(conn->store_conn, MessageType_PlasmaGetReply); + object_id received_obj_id; + plasma_read_GetReply(reply_data, &received_obj_id, &object_data, 1); + free(reply_data); + DCHECK(memcmp(&received_obj_id, &obj_id, sizeof(obj_id)) == 0); int fd = recv_fd(conn->store_conn); CHECK(fd >= 0); - object = &reply.object; + object = &object_data; *data = lookup_or_mmap(conn, fd, object->handle.store_fd, object->handle.mmap_size); } @@ -271,7 +279,7 @@ void plasma_get(plasma_connection *conn, /* Increment the count of the number of instances of this object that this * client is using. A call to plasma_release is required to decrement this * count. Cache the reference to the object. */ - increment_object_count(conn, object_id, object, true); + increment_object_count(conn, obj_id, object, true); } void plasma_perform_release(plasma_connection *conn, object_id object_id) { @@ -303,8 +311,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 = plasma_make_request(object_id); - CHECK(plasma_send_request(conn->store_conn, PLASMA_RELEASE, &req) >= 0); + CHECK(plasma_send_ReleaseRequest(conn->store_conn, conn->builder, + object_id) >= 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); @@ -331,22 +339,22 @@ void plasma_release(plasma_connection *conn, object_id obj_id) { /* This method is used to query whether the plasma store contains an object. */ void plasma_contains(plasma_connection *conn, - object_id object_id, + object_id obj_id, int *has_object) { /* Check if we already have a reference to the object. */ object_in_use_entry *object_entry; - HASH_FIND(hh, conn->objects_in_use, &object_id, sizeof(object_id), - object_entry); + HASH_FIND(hh, conn->objects_in_use, &obj_id, sizeof(obj_id), object_entry); if (object_entry) { *has_object = 1; } else { /* If we don't already have a reference to the object, check with the store * to see if we have the object. */ - plasma_request req = plasma_make_request(object_id); - CHECK(plasma_send_request(conn->store_conn, PLASMA_CONTAINS, &req) >= 0); - plasma_reply reply; - CHECK(plasma_receive_reply(conn->store_conn, sizeof(reply), &reply) >= 0); - *has_object = reply.has_object; + plasma_send_ContainsRequest(conn->store_conn, conn->builder, obj_id); + uint8_t *reply_data = + plasma_receive(conn->store_conn, MessageType_PlasmaContainsReply); + object_id object_id2; + plasma_read_ContainsReply(reply_data, &object_id2, has_object); + free(reply_data); } } @@ -390,26 +398,30 @@ void plasma_seal(plasma_connection *conn, object_id object_id) { "Plasma client called seal an already sealed object"); object_entry->is_sealed = true; /* Send the seal request to Plasma. */ - plasma_request req = plasma_make_request(object_id); - CHECK(plasma_compute_object_hash(conn, object_id, req.digest)); - CHECK(plasma_send_request(conn->store_conn, PLASMA_SEAL, &req) >= 0); + unsigned char digest[DIGEST_SIZE]; + CHECK(plasma_compute_object_hash(conn, object_id, &digest[0])); + CHECK(plasma_send_SealRequest(conn->store_conn, conn->builder, object_id, + &digest[0]) >= 0); } void plasma_delete(plasma_connection *conn, object_id object_id) { - plasma_request req = plasma_make_request(object_id); - CHECK(plasma_send_request(conn->store_conn, PLASMA_DELETE, &req) >= 0); + /* TODO(rkn): In the future, we can use this method to give hints to the + * eviction policy about when an object will no longer be needed. */ } int64_t plasma_evict(plasma_connection *conn, int64_t num_bytes) { /* Send a request to the store to evict objects. */ - plasma_request req; - memset(&req, 0, sizeof(req)); - req.num_bytes = num_bytes; - CHECK(plasma_send_request(conn->store_conn, PLASMA_EVICT, &req) >= 0); + CHECK(plasma_send_EvictRequest(conn->store_conn, conn->builder, num_bytes) >= + 0); /* Wait for a response with the number of bytes actually evicted. */ - plasma_reply reply; - CHECK(plasma_receive_reply(conn->store_conn, sizeof(reply), &reply) >= 0); - return reply.num_bytes; + int64_t type; + int64_t length; + uint8_t *reply_data; + read_message(conn->store_conn, &type, &length, &reply_data); + int64_t num_bytes_evicted; + plasma_read_EvictReply(reply_data, &num_bytes_evicted); + free(reply_data); + return num_bytes_evicted; } int plasma_subscribe(plasma_connection *conn) { @@ -423,8 +435,7 @@ int plasma_subscribe(plasma_connection *conn) { int flags = fcntl(fd[1], F_GETFL, 0); CHECK(fcntl(fd[1], F_SETFL, flags | O_NONBLOCK) == 0); /* Tell the Plasma store about the subscription. */ - plasma_request req = {0}; - CHECK(plasma_send_request(conn->store_conn, PLASMA_SUBSCRIBE, &req) >= 0); + CHECK(plasma_send_SubscribeRequest(conn->store_conn, conn->builder) >= 0); /* Send the file descriptor that the Plasma store should use to push * notifications about sealed objects to this client. */ CHECK(send_fd(conn->store_conn, fd[1]) >= 0); @@ -467,6 +478,7 @@ plasma_connection *plasma_connect(const char *store_socket_name, } else { result->manager_conn = -1; } + result->builder = make_protocol_builder(); result->mmap_table = NULL; result->objects_in_use = NULL; result->config.release_delay = release_delay; @@ -479,6 +491,7 @@ void plasma_disconnect(plasma_connection *conn) { while ((id = (object_id *) utringbuffer_next(conn->release_history, id))) { plasma_perform_release(conn, *id); } + free_protocol_builder(conn->builder); utringbuffer_free(conn->release_history); close(conn->store_conn); if (conn->manager_conn >= 0) { @@ -536,18 +549,11 @@ int plasma_manager_connect(const char *ip_addr, int port) { } void plasma_transfer(plasma_connection *conn, - const char *addr, + const char *address, int port, object_id object_id) { - plasma_request req = plasma_make_request(object_id); - req.port = port; - char *end = NULL; - for (int i = 0; i < 4; ++i) { - req.addr[i] = strtol(end ? end : addr, &end, 10); - /* skip the '.' */ - end += 1; - } - CHECK(plasma_send_request(conn->manager_conn, PLASMA_TRANSFER, &req) >= 0); + CHECK(plasma_send_DataRequest(conn->manager_conn, conn->builder, object_id, + address, port) >= 0); } void plasma_fetch(plasma_connection *conn, @@ -555,11 +561,8 @@ void plasma_fetch(plasma_connection *conn, object_id object_ids[]) { CHECK(conn != NULL); CHECK(conn->manager_conn >= 0); - plasma_request *req = plasma_alloc_request(num_object_ids); - for (int i = 0; i < num_object_ids; ++i) { - req->object_requests[i].object_id = object_ids[i]; - } - CHECK(plasma_send_request(conn->manager_conn, PLASMA_FETCH, req) >= 0); + CHECK(plasma_send_FetchRequest(conn->manager_conn, conn->builder, object_ids, + num_object_ids) >= 0); } int get_manager_fd(plasma_connection *conn) { @@ -567,13 +570,12 @@ int get_manager_fd(plasma_connection *conn) { } bool plasma_get_local(plasma_connection *conn, - object_id object_id, + object_id obj_id, object_buffer *object_buffer) { CHECK(conn != NULL); /* Check if we already have a reference to the object. */ object_in_use_entry *object_entry; - HASH_FIND(hh, conn->objects_in_use, &object_id, sizeof(object_id), - object_entry); + HASH_FIND(hh, conn->objects_in_use, &obj_id, sizeof(obj_id), object_entry); plasma_object *object; if (object_entry) { /* If we have already have a reference to the object, use it to get the @@ -586,20 +588,24 @@ bool plasma_get_local(plasma_connection *conn, object_buffer->data = lookup_mmapped_file(conn, object->handle.store_fd); } else { /* Else, request a reference to the object data from the plasma store. */ - plasma_request req = plasma_make_request(object_id); - CHECK(plasma_send_request(conn->store_conn, PLASMA_GET_LOCAL, &req) >= 0); + CHECK(plasma_send_GetLocalRequest(conn->store_conn, conn->builder, &obj_id, + 1) >= 0); + uint8_t *reply_data = + plasma_receive(conn->store_conn, MessageType_PlasmaGetLocalReply); + plasma_object object_data; + int has_object; + plasma_read_GetLocalReply(reply_data, &obj_id, &object_data, &has_object, + 1); + free(reply_data); - plasma_reply reply; - CHECK(plasma_receive_reply(conn->store_conn, sizeof(reply), &reply) >= 0); - int fd = recv_fd(conn->store_conn); - CHECKM(fd >= 0, "recv_fd not successful"); - - if (!reply.has_object) { + if (!has_object) { /* The object is not in our local store. */ - close(fd); return false; } - object = &reply.object; + + int fd = recv_fd(conn->store_conn); + CHECKM(fd >= 0, "recv_fd not successful"); + object = &object_data; object_buffer->data = lookup_or_mmap(conn, fd, object->handle.store_fd, object->handle.mmap_size); } @@ -611,7 +617,7 @@ bool plasma_get_local(plasma_connection *conn, /* Increment the count of the number of instances of this object that this * client is using. A call to plasma_release is required to decrement this * count. Cache the reference to the object. */ - increment_object_count(conn, object_id, object, true); + increment_object_count(conn, obj_id, object, true); return true; } @@ -619,13 +625,13 @@ int plasma_status(plasma_connection *conn, object_id object_id) { CHECK(conn != NULL); CHECK(conn->manager_conn >= 0); - plasma_request req = plasma_make_request(object_id); - CHECK(plasma_send_request(conn->manager_conn, PLASMA_STATUS, &req) >= 0); - - plasma_reply reply; - CHECK(plasma_receive_reply(conn->manager_conn, sizeof(reply), &reply) >= 0); - - return reply.object_status; + plasma_send_StatusRequest(conn->manager_conn, conn->builder, &object_id, 1); + uint8_t *reply_data = + plasma_receive(conn->manager_conn, MessageType_PlasmaStatusReply); + int object_status; + plasma_read_StatusReply(reply_data, &object_id, &object_status, 1); + free(reply_data); + return object_status; } int plasma_wait(plasma_connection *conn, @@ -639,46 +645,40 @@ int plasma_wait(plasma_connection *conn, CHECK(num_ready_objects > 0); CHECK(num_ready_objects <= num_object_requests); - plasma_request *req = plasma_alloc_request(num_object_requests); for (int i = 0; i < num_object_requests; ++i) { CHECK(object_requests[i].type == PLASMA_QUERY_LOCAL || object_requests[i].type == PLASMA_QUERY_ANYWHERE); - req->object_requests[i] = object_requests[i]; } - req->num_ready_objects = num_ready_objects; - req->timeout = timeout_ms; - CHECK(plasma_send_request(conn->manager_conn, PLASMA_WAIT, req) >= 0); - free(req); - plasma_reply *reply = plasma_alloc_reply(num_object_requests); - CHECK(plasma_receive_reply(conn->manager_conn, - plasma_reply_size(num_object_requests), - reply) >= 0); + CHECK(plasma_send_WaitRequest(conn->manager_conn, conn->builder, + object_requests, num_object_requests, + num_ready_objects, timeout_ms) >= 0); + uint8_t *reply_data = + plasma_receive(conn->manager_conn, MessageType_PlasmaWaitReply); + plasma_read_WaitReply(reply_data, object_requests, &num_ready_objects); + free(reply_data); + int num_objects_ready = 0; for (int i = 0; i < num_object_requests; ++i) { - int type = reply->object_requests[i].type; - int status = reply->object_requests[i].status; - object_requests[i].object_id = reply->object_requests[i].object_id; - object_requests[i].type = type; - object_requests[i].status = status; + int type = object_requests[i].type; + int status = object_requests[i].status; switch (type) { case PLASMA_QUERY_LOCAL: - if (status == PLASMA_OBJECT_LOCAL) { + if (status == ObjectStatus_Local) { num_objects_ready += 1; } break; case PLASMA_QUERY_ANYWHERE: - if (status == PLASMA_OBJECT_LOCAL || status == PLASMA_OBJECT_REMOTE) { + if (status == ObjectStatus_Local || status == ObjectStatus_Remote) { num_objects_ready += 1; } else { - CHECK(status == PLASMA_OBJECT_NONEXISTENT); + CHECK(status == ObjectStatus_Nonexistent); } break; default: LOG_FATAL("This code should be unreachable."); } } - free(reply); return num_objects_ready; } @@ -707,17 +707,17 @@ void plasma_client_get(plasma_connection *conn, object_id object_ids[1] = {obj_id}; plasma_fetch(conn, 1, object_ids); switch (plasma_status(conn, obj_id)) { - case PLASMA_OBJECT_LOCAL: + case ObjectStatus_Local: /* Object has finished being transfered just after calling * plasma_get_local(), and it is now in the local Plasma Store. Loop again * to call plasma_get_local() and eventually return. */ continue; - case PLASMA_OBJECT_REMOTE: + case ObjectStatus_Remote: /* A fetch request has been already scheduled for obj_id, so wait for * it to complete. */ request.type = PLASMA_QUERY_LOCAL; break; - case PLASMA_OBJECT_NONEXISTENT: + case ObjectStatus_Nonexistent: /* Object doesn’t exist in the system so ask local scheduler to create it. */ /* TODO: scheduler_create_object(obj_id); */ @@ -735,11 +735,11 @@ void plasma_client_get(plasma_connection *conn, * locally or remotely, if obj_id didn't exist in the system. * - if timeout, next iteration will retry plasma_fetch() or * scheduler_create_object() - * - if request.status == PLASMA_OBJECT_LOCAL, next iteration + * - if request.status == ObjectStatus_Local, next iteration * will get object and return - * - if request.status == PLASMA_OBJECT_REMOTE, next iteration + * - if request.status == ObjectStatus_Remote, next iteration * will call plasma_fetch() - * - if request.status == PLASMA_OBJECT_NONEXISTENT, next iteration + * - if request.status == ObjectStatus_Nonexistent, next iteration * will call scheduler_create_object() */ #define TIMEOUT_WAIT_MS 200 @@ -788,8 +788,8 @@ int plasma_client_wait(plasma_connection *conn, int idx_returns = 0; for (int i = 0; i < num_returns; ++i) { - if (requests[i].status == PLASMA_OBJECT_LOCAL || - requests[i].status == PLASMA_OBJECT_REMOTE) { + if (requests[i].status == ObjectStatus_Local || + requests[i].status == ObjectStatus_Remote) { return_object_ids[idx_returns] = requests[i].object_id; idx_returns += 1; } @@ -799,7 +799,7 @@ int plasma_client_wait(plasma_connection *conn, /* The timeout hasn't expired and we got less than num_returns in the * system. Trigger reconstruction of the missing objects. */ for (int i = 0; i < num_returns; ++i) { - if (requests[i].status == PLASMA_OBJECT_NONEXISTENT) { + if (requests[i].status == ObjectStatus_Nonexistent) { /* Object doesn’t exist in the system so ask local scheduler to create * object with ID requests[i].object_id. */ /* TODO: scheduler_create_object(object_id); */ @@ -816,7 +816,7 @@ void plasma_client_multiget(plasma_connection *conn, object_buffer object_buffers[]) { object_request requests[num_object_ids]; - /* Set all request types to PLASMA_OBJECT_LOCAL, as we want to get all objects + /* Set all request types to ObjectStatus_Local, as we want to get all objects * into the local Plasma Store. */ for (int i = 0; i < num_object_ids; ++i) { requests[i].object_id = object_ids[i]; diff --git a/src/plasma/plasma_client.h b/src/plasma/plasma_client.h index 0086120b2..7577e8d97 100644 --- a/src/plasma/plasma_client.h +++ b/src/plasma/plasma_client.h @@ -316,13 +316,13 @@ int plasma_info(plasma_connection *conn, * "type" field. * - A PLASMA_QUERY_LOCAL request is satisfied when object_id becomes * available in the local Plasma Store. In this case, this function - * sets the "status" field to PLASMA_OBJECT_LOCAL. Note, if the status - * is not PLASMA_OBJECT_LOCAL, it will be PLASMA_OBJECT_NONEXISTENT, + * sets the "status" field to ObjectStatus_Local. Note, if the status + * is not ObjectStatus_Local, it will be ObjectStatus_Nonexistent, * but it may exist elsewhere in the system. * - A PLASMA_QUERY_ANYWHERE request is satisfied when object_id becomes * available either at the local Plasma Store or on a remote Plasma * Store. In this case, the functions sets the "status" field to - * PLASMA_OBJECT_LOCAL or PLASMA_OBJECT_REMOTE. + * ObjectStatus_Local or ObjectStatus_Remote. * @param num_ready_objects The number of requests in object_requests array that * must be satisfied before the function returns, unless it timeouts. * The num_ready_objects should be no larger than num_object_requests. diff --git a/src/plasma/plasma_extension.c b/src/plasma/plasma_extension.c index 507010ef1..166f21d90 100644 --- a/src/plasma/plasma_extension.c +++ b/src/plasma/plasma_extension.c @@ -3,6 +3,7 @@ #include "common.h" #include "io.h" +#include "plasma_protocol.h" #include "plasma_client.h" #include "object_info.h" @@ -250,8 +251,8 @@ PyObject *PyPlasma_wait(PyObject *self, PyObject *args) { if (num_returned == num_to_return) { break; } - if (object_requests[i].status == PLASMA_OBJECT_LOCAL || - object_requests[i].status == PLASMA_OBJECT_REMOTE) { + if (object_requests[i].status == ObjectStatus_Local || + object_requests[i].status == ObjectStatus_Remote) { PyObject *ready = PyBytes_FromStringAndSize((char *) object_requests[i].object_id.id, sizeof(object_requests[i].object_id)); @@ -259,7 +260,7 @@ PyObject *PyPlasma_wait(PyObject *self, PyObject *args) { PySet_Discard(waiting_ids, ready); num_returned += 1; } else { - CHECK(object_requests[i].status == PLASMA_OBJECT_NONEXISTENT); + CHECK(object_requests[i].status == ObjectStatus_Nonexistent); } } CHECK(num_returned == num_to_return); diff --git a/src/plasma/plasma_manager.c b/src/plasma/plasma_manager.c index 32c9e3cf0..453304afd 100644 --- a/src/plasma/plasma_manager.c +++ b/src/plasma/plasma_manager.c @@ -31,6 +31,7 @@ #include "net.h" #include "event_loop.h" #include "plasma.h" +#include "plasma_protocol.h" #include "plasma_client.h" #include "plasma_manager.h" #include "state/db.h" @@ -66,7 +67,7 @@ int request_status(object_id object_id, /** * Send requested object_id back to the Plasma Manager identified * by (addr, port) which requested it. This is done via a - * PLASMA_DATA message. + * data Request message. * * @param loop * @param object_id The ID of the object being transferred to (addr, port). @@ -76,13 +77,13 @@ int request_status(object_id object_id, */ void process_transfer_request(event_loop *loop, object_id object_id, - uint8_t addr[4], + const char *addr, int port, client_connection *conn); /** * Receive object_id requested by this Plamsa Manager from the remote Plasma - * Manager identified by client_sock. The object_id is sent via the PLASMA_DATA + * Manager identified by client_sock. The object_id is sent via the data request * message. * * @param loop The event data structure. @@ -193,7 +194,7 @@ struct plasma_manager_state { client_connection *manager_connections; db_handle *db; /** Our address. */ - uint8_t addr[4]; + const char *addr; /** Our port. */ int port; /** Hash table of outstanding fetch requests. The key is the object ID. The @@ -207,6 +208,8 @@ struct plasma_manager_state { object_wait_requests *object_wait_requests_remote; /** Initialize an empty hash map for the cache of local available object. */ available_object *local_available_objects; + /** Buffer that holds memory for serializing plasma protocol messages. */ + protocol_builder *builder; }; plasma_manager_state *g_manager_state = NULL; @@ -263,9 +266,6 @@ struct client_connection { int fd; /** Timer id for timing out wait (or fetch). */ int64_t timer_id; - /** If this client is processing a wait, this contains the object ids that - * are already available. */ - plasma_reply *wait_reply; /** The objects that we are waiting for and their callback * contexts, for either a fetch or a wait operation. */ client_object_request *active_objects; @@ -356,14 +356,10 @@ void remove_wait_request(plasma_manager_state *manager_state, void return_from_wait(plasma_manager_state *manager_state, wait_request *wait_req) { - plasma_reply *reply = plasma_alloc_reply(wait_req->num_object_requests); - reply->num_object_ids = wait_req->num_object_requests; - for (int i = 0; i < wait_req->num_object_requests; ++i) { - reply->object_requests[i] = wait_req->object_requests[i]; - } /* Send the reply to the client. */ - CHECK(plasma_send_reply(wait_req->client_conn->fd, reply) >= 0); - free(reply); + CHECK(plasma_send_WaitReply(wait_req->client_conn->fd, manager_state->builder, + wait_req->object_requests, + wait_req->num_object_requests) >= 0); /* Remove the wait request from each of the relevant object_wait_requests hash * tables if it is present there. */ for (int i = 0; i < wait_req->num_object_requests; ++i) { @@ -397,7 +393,7 @@ void update_object_wait_requests(plasma_manager_state *manager_state, if (object_ids_equal(wait_req->object_requests[j].object_id, obj_id)) { /* Check that this object is currently nonexistent. */ CHECK(wait_req->object_requests[j].status == - PLASMA_OBJECT_NONEXISTENT); + ObjectStatus_Nonexistent); wait_req->object_requests[j].status = status; break; } @@ -468,8 +464,7 @@ plasma_manager_state *init_plasma_manager_state(const char *store_socket_name, state->db = NULL; LOG_DEBUG("No db connection specified"); } - sscanf(manager_addr, "%hhu.%hhu.%hhu.%hhu", &state->addr[0], &state->addr[1], - &state->addr[2], &state->addr[3]); + state->addr = manager_addr; state->port = manager_port; /* Initialize an empty hash map for the cache of local available objects. */ state->local_available_objects = NULL; @@ -478,6 +473,7 @@ plasma_manager_state *init_plasma_manager_state(const char *store_socket_name, /* Add the callback that processes the notification to the event loop. */ event_loop_add_file(state->loop, plasma_fd, EVENT_LOOP_READ, process_object_notification, state); + state->builder = make_protocol_builder(); return state; } @@ -505,6 +501,7 @@ void destroy_plasma_manager_state(plasma_manager_state *state) { plasma_disconnect(state->plasma_conn); event_loop_destroy(state->loop); + free_protocol_builder(state->builder); free(state); } @@ -554,32 +551,29 @@ void send_queued_request(event_loop *loop, void *context, int events) { client_connection *conn = (client_connection *) context; + plasma_manager_state *state = conn->manager_state; + if (conn->transfer_queue == NULL) { /* If there are no objects to transfer, temporarily remove this connection * from the event loop. It will be reawoken when we receive another - * PLASMA_TRANSFER request. */ + * data request. */ event_loop_remove_file(loop, conn->fd); return; } plasma_request_buffer *buf = conn->transfer_queue; - 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; - CHECK(plasma_send_request(conn->fd, PLASMA_TRANSFER, &manager_req) >= 0); + case MessageType_PlasmaDataRequest: + CHECK(plasma_send_DataRequest(conn->fd, state->builder, buf->object_id, + state->addr, state->port) >= 0); break; - case PLASMA_DATA: + case MessageType_PlasmaDataReply: LOG_DEBUG("Transferring object to manager"); if (conn->cursor == 0) { /* If the cursor is zero, we haven't sent any requests for this object - * yet, - * so send the initial PLASMA_DATA request. */ - manager_req.data_size = buf->data_size; - manager_req.metadata_size = buf->metadata_size; - CHECK(plasma_send_request(conn->fd, PLASMA_DATA, &manager_req) >= 0); + * yet, so send the initial data request. */ + CHECK(plasma_send_DataReply(conn->fd, state->builder, buf->object_id, + buf->data_size, buf->metadata_size) >= 0); } write_object_chunk(conn, buf); break; @@ -707,7 +701,7 @@ client_connection *get_manager_connection(plasma_manager_state *state, void process_transfer_request(event_loop *loop, object_id object_id, - uint8_t addr[4], + const char *addr, int port, client_connection *conn) { uint8_t *data; @@ -725,19 +719,15 @@ void process_transfer_request(event_loop *loop, &metadata_size, &metadata); assert(metadata == data + data_size); plasma_request_buffer *buf = malloc(sizeof(plasma_request_buffer)); - buf->type = PLASMA_DATA; + buf->type = MessageType_PlasmaDataReply; buf->object_id = object_id; buf->data = data; /* We treat this as a pointer to the concatenated data and metadata. */ buf->data_size = data_size; buf->metadata_size = metadata_size; - UT_string *ip_addr; - utstring_new(ip_addr); - utstring_printf(ip_addr, "%d.%d.%d.%d", addr[0], addr[1], addr[2], addr[3]); client_connection *manager_conn = - get_manager_connection(conn->manager_state, utstring_body(ip_addr), port); - utstring_free(ip_addr); + get_manager_connection(conn->manager_state, addr, port); if (manager_conn->transfer_queue == NULL) { /* If we already have a connection to this manager and its inactive, @@ -751,7 +741,7 @@ void process_transfer_request(event_loop *loop, /** * Receive object_id requested by this Plamsa Manager from the remote Plasma - * Manager identified by client_sock. The object_id is sent via the PLASMA_DATA + * Manager identified by client_sock. The object_id is sent via the data requst * message. * * @param loop The event data structure. @@ -838,7 +828,7 @@ void request_transfer_from(plasma_manager_state *manager_state, plasma_request_buffer *transfer_request = malloc(sizeof(plasma_request_buffer)); - transfer_request->type = PLASMA_TRANSFER; + transfer_request->type = MessageType_PlasmaDataRequest; transfer_request->object_id = fetch_req->object_id; if (manager_conn->transfer_queue == NULL) { @@ -957,7 +947,7 @@ void object_present_callback(object_id object_id, /* Update the in-progress remote wait requests. */ update_object_wait_requests(manager_state, object_id, PLASMA_QUERY_ANYWHERE, - PLASMA_OBJECT_REMOTE); + ObjectStatus_Remote); } /* This callback is used by both fetch and wait. Therefore, it may have to @@ -981,7 +971,7 @@ void object_table_subscribe_callback(object_id object_id, void process_fetch_requests(client_connection *client_conn, int num_object_ids, - object_request object_requests[]) { + object_id object_ids[]) { plasma_manager_state *manager_state = client_conn->manager_state; int num_object_ids_to_request = 0; @@ -990,7 +980,7 @@ void process_fetch_requests(client_connection *client_conn, object_id *object_ids_to_request = malloc(num_object_ids * sizeof(object_id)); for (int i = 0; i < num_object_ids; ++i) { - object_id obj_id = object_requests[i].object_id; + object_id obj_id = object_ids[i]; /* Check if this object is already present locally. If so, do nothing. */ if (is_object_local(manager_state, obj_id)) { @@ -1057,7 +1047,7 @@ void process_wait_request(client_connection *client_conn, for (int i = 0; i < num_object_requests; ++i) { wait_req->object_requests[i].object_id = object_requests[i].object_id; wait_req->object_requests[i].type = object_requests[i].type; - wait_req->object_requests[i].status = PLASMA_OBJECT_NONEXISTENT; + wait_req->object_requests[i].status = ObjectStatus_Nonexistent; } wait_req->num_objects_to_wait_for = num_ready_objects; wait_req->num_satisfied = 0; @@ -1074,7 +1064,7 @@ void process_wait_request(client_connection *client_conn, /* Check if this object is already present locally. If so, mark the object * as present. */ if (is_object_local(manager_state, obj_id)) { - wait_req->object_requests[i].status = PLASMA_OBJECT_LOCAL; + wait_req->object_requests[i].status = ObjectStatus_Local; wait_req->num_satisfied += 1; continue; } @@ -1130,9 +1120,9 @@ void process_wait_request(client_connection *client_conn, * @param object_id ID of the object whose status we require. * @param manager_cont Number of remote nodes object_id is stored at. If * manager_count > 0, then object_id exists on a remote node an its - * status is PLASMA_OBJECT_REMOTE. Otherwise, if manager_count == 0, the + * status is ObjectStatus_Remote. Otherwise, if manager_count == 0, the * object doesn't exist in the system and its status is - * PLASMA_OBJECT_NONEXISTENT. + * ObjectStatus_Nonexistent. * @param manager_vector Array containing the Plasma Managers running at the * nodes where object_id is stored. Not used; it will be eventually * deallocated. @@ -1146,9 +1136,9 @@ void request_status_done(object_id object_id, client_connection *client_conn = (client_connection *) context; int status = request_status(object_id, manager_count, manager_vector, context); - plasma_reply reply = plasma_make_reply(object_id); - reply.object_status = status; - CHECK(plasma_send_reply(client_conn->fd, &reply) >= 0); + CHECK(plasma_send_StatusReply(client_conn->fd, + client_conn->manager_state->builder, &object_id, + &status, 1) >= 0); } int request_status(object_id object_id, @@ -1159,39 +1149,39 @@ int request_status(object_id object_id, /* Return success immediately if we already have this object. */ if (is_object_local(client_conn->manager_state, object_id)) { - return PLASMA_OBJECT_LOCAL; + return ObjectStatus_Local; } /* Since object is not stored at the local locally, manager_count > 0 means * that the object is stored at another remote object. Otherwise, if * manager_count == 0, the object is not stored anywhere. */ - return (manager_count > 0 ? PLASMA_OBJECT_REMOTE : PLASMA_OBJECT_NONEXISTENT); + return (manager_count > 0 ? ObjectStatus_Remote : ObjectStatus_Nonexistent); } void object_table_lookup_fail_callback(object_id object_id, void *user_context, void *user_data) { - /* Fail for now. Later, we may want to send a PLASMA_OBJECT_NONEXISTENT to the + /* Fail for now. Later, we may want to send a ObjectStatus_Nonexistent to the * client. */ CHECK(0); } void process_status_request(client_connection *client_conn, object_id object_id) { - client_conn->wait_reply = NULL; - /* Return success immediately if we already have this object. */ if (is_object_local(client_conn->manager_state, object_id)) { - plasma_reply reply = plasma_make_reply(object_id); - reply.object_status = PLASMA_OBJECT_LOCAL; - CHECK(plasma_send_reply(client_conn->fd, &reply) >= 0); + int status = ObjectStatus_Local; + CHECK(plasma_send_StatusReply(client_conn->fd, + client_conn->manager_state->builder, + &object_id, &status, 1) >= 0); return; } if (client_conn->manager_state->db == NULL) { - plasma_reply reply = plasma_make_reply(object_id); - reply.object_status = PLASMA_OBJECT_NONEXISTENT; - CHECK(plasma_send_reply(client_conn->fd, &reply) >= 0); + int status = ObjectStatus_Nonexistent; + CHECK(plasma_send_StatusReply(client_conn->fd, + client_conn->manager_state->builder, + &object_id, &status, 1) >= 0); return; } @@ -1265,9 +1255,9 @@ void process_add_object_notification(plasma_manager_state *state, /* Update the in-progress local and remote wait requests. */ update_object_wait_requests(state, obj_id, PLASMA_QUERY_LOCAL, - PLASMA_OBJECT_LOCAL); + ObjectStatus_Local); update_object_wait_requests(state, obj_id, PLASMA_QUERY_ANYWHERE, - PLASMA_OBJECT_LOCAL); + ObjectStatus_Local); } void process_object_notification(event_loop *loop, @@ -1302,37 +1292,56 @@ void process_message(event_loop *loop, int events) { client_connection *conn = (client_connection *) context; + int64_t length; int64_t type; - plasma_request *req; - CHECK(plasma_receive_request(client_sock, &type, &req) >= 0); + uint8_t *data; + read_message(client_sock, &type, &length, &data); switch (type) { - case PLASMA_TRANSFER: - LOG_DEBUG("Processing plasma transfer request."); - DCHECK(req->num_object_ids == 1); - process_transfer_request(loop, req->object_requests[0].object_id, req->addr, - req->port, conn); - break; - case PLASMA_DATA: - LOG_DEBUG("Starting to stream data"); - DCHECK(req->num_object_ids == 1); - process_data_request(loop, client_sock, req->object_requests[0].object_id, - req->data_size, req->metadata_size, conn); - break; - case PLASMA_FETCH: + case MessageType_PlasmaDataRequest: { + LOG_DEBUG("Processing data request"); + object_id object_id; + char *address; + int port; + plasma_read_DataRequest(data, &object_id, &address, &port); + process_transfer_request(loop, object_id, address, port, conn); + free(address); + } break; + case MessageType_PlasmaDataReply: { + LOG_DEBUG("Processing data reply"); + object_id object_id; + int64_t object_size; + int64_t metadata_size; + plasma_read_DataReply(data, &object_id, &object_size, &metadata_size); + process_data_request(loop, client_sock, object_id, object_size, + metadata_size, conn); + } break; + case MessageType_PlasmaFetchRequest: { LOG_DEBUG("Processing fetch remote"); - process_fetch_requests(conn, req->num_object_ids, req->object_requests); - break; - case PLASMA_WAIT: + int64_t num_objects = plasma_read_FetchRequest_num_objects(data); + object_id object_ids_to_fetch[num_objects]; + plasma_read_FetchRequest(data, object_ids_to_fetch, num_objects); + process_fetch_requests(conn, num_objects, &object_ids_to_fetch[0]); + } break; + case MessageType_PlasmaWaitRequest: { LOG_DEBUG("Processing wait"); - process_wait_request(conn, req->num_object_ids, req->object_requests, - req->timeout, req->num_ready_objects); - break; - case PLASMA_STATUS: + int num_object_ids = plasma_read_WaitRequest_num_object_ids(data); + int64_t timeout_ms; + int num_ready_objects; + object_request object_requests[num_object_ids]; + plasma_read_WaitRequest(data, &object_requests[0], num_object_ids, + &timeout_ms, &num_ready_objects); + process_wait_request(conn, num_object_ids, &object_requests[0], timeout_ms, + num_ready_objects); + } break; + case MessageType_PlasmaStatusRequest: { LOG_DEBUG("Processing status"); - DCHECK(req->num_object_ids == 1); - process_status_request(conn, req->object_requests[0].object_id); - break; + object_id object_id; + int64_t num_objects = plasma_read_StatusRequest_num_objects(data); + CHECK(num_objects == 1); + plasma_read_StatusRequest(data, &object_id, 1); + process_status_request(conn, object_id); + } break; case DISCONNECT_CLIENT: { LOG_INFO("Disconnecting client on fd %d", client_sock); /* TODO(swang): Check if this connection was to a plasma manager. If so, @@ -1344,8 +1353,7 @@ void process_message(event_loop *loop, default: LOG_FATAL("invalid request %" PRId64, type); } - - free(req); + free(data); } /* TODO(pcm): Split this into two methods: new_worker_connection diff --git a/src/plasma/plasma_protocol.c b/src/plasma/plasma_protocol.c new file mode 100644 index 000000000..d15630569 --- /dev/null +++ b/src/plasma/plasma_protocol.c @@ -0,0 +1,719 @@ +#include + +#include "plasma_protocol.h" +#include "io.h" + +#include "format/plasma_builder.h" +#include "plasma.h" + +#define FLATBUFFER_BUILDER_DEFAULT_SIZE 1024 + +protocol_builder *make_protocol_builder(void) { + protocol_builder *builder = malloc(sizeof(protocol_builder)); + CHECK(builder); + flatcc_builder_init(builder); + return builder; +} + +void free_protocol_builder(protocol_builder *builder) { + flatcc_builder_clear(builder); + free(builder); +} + +/** + * Writes an array of object IDs into a flatbuffer buffer and return + * the resulting vector. + * + * @params B Pointer to the flatbuffer builder. + * @param object_ids Array of object IDs to be written. + * @param num_objects The number of elements in the array. + * @return Reference to the flatbuffer string vector. + */ +flatbuffers_string_vec_ref_t object_ids_to_flatbuffer(flatcc_builder_t *B, + object_id object_ids[], + int64_t num_objects) { + flatbuffers_string_vec_start(B); + for (int i = 0; i < num_objects; i++) { + flatbuffers_string_ref_t id = flatbuffers_string_create( + B, (const char *) &object_ids[i].id[0], sizeof(object_ids[i].id)); + flatbuffers_string_vec_push(B, id); + } + return flatbuffers_string_vec_end(B); +} + +/** + * Reads an array of object IDs from a flatbuffer vector. + * + * @param object_id_vector Flatbuffer vector containing object IDs. + * @param object_ids_ptr Pointer to array that will contain the object IDs. The + * array is allocated by this function and must be freed by the user. + * @param num_objects Pointer to the number of objects, will be written by this + * method. + * @return Void. + */ +void object_ids_from_flatbuffer(flatbuffers_string_vec_t object_id_vector, + object_id object_ids[], + int64_t num_objects) { + CHECK(flatbuffers_string_vec_len(object_id_vector) == num_objects); + for (int64_t i = 0; i < num_objects; ++i) { + memcpy(&object_ids[i].id[0], flatbuffers_string_vec_at(object_id_vector, i), + sizeof(object_ids[i].id)); + } +} + +/** + * Finalize the flatbuffers and write a message with the result to a + * file descriptor. + * + * @param B Pointer to the flatbuffer builder. + * @param fd File descriptor the message gets written to. + * @param message_type Type of the message that is written. + * @return Whether there was an error while writing. 0 corresponds to success + * and -1 corresponds to an error (errno will be set). + */ +int finalize_buffer_and_send(flatcc_builder_t *B, int fd, int message_type) { + size_t size; + void *buff = flatcc_builder_finalize_buffer(B, &size); + int r = write_message(fd, message_type, size, buff); + free(buff); + flatcc_builder_reset(B); + return r; +} + +uint8_t *plasma_receive(int sock, int64_t message_type) { + int64_t type; + int64_t length; + uint8_t *reply_data; + read_message(sock, &type, &length, &reply_data); + CHECKM(type == message_type, "type = %" PRId64 ", message_type = %" PRId64, + type, message_type); + return reply_data; +} + +int plasma_send_CreateRequest(int sock, + protocol_builder *B, + object_id object_id, + int64_t data_size, + int64_t metadata_size) { + PlasmaCreateRequest_start_as_root(B); + PlasmaCreateRequest_object_id_create(B, (const char *) &object_id.id[0], + sizeof(object_id.id)); + PlasmaCreateRequest_data_size_add(B, data_size); + PlasmaCreateRequest_metadata_size_add(B, metadata_size); + PlasmaCreateRequest_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaCreateRequest); +} + +void plasma_read_CreateRequest(uint8_t *data, + object_id *object_id, + int64_t *data_size, + int64_t *metadata_size) { + DCHECK(data); + PlasmaCreateRequest_table_t req = PlasmaCreateRequest_as_root(data); + *data_size = PlasmaCreateRequest_data_size(req); + *metadata_size = PlasmaCreateRequest_metadata_size(req); + flatbuffers_string_t id = PlasmaCreateRequest_object_id(req); + DCHECK(flatbuffers_string_len(id) == sizeof(object_id->id)); + memcpy(&object_id->id[0], id, sizeof(object_id->id)); +} + +int plasma_send_CreateReply(int sock, + protocol_builder *B, + object_id object_id, + plasma_object *object, + int error_code) { + PlasmaCreateReply_start_as_root(B); + PlasmaCreateReply_object_id_create(B, (const char *) &object_id.id[0], + sizeof(object_id.id)); + PlasmaCreateReply_plasma_object_create( + B, object->handle.store_fd, object->handle.mmap_size, object->data_offset, + object->data_size, object->metadata_offset, object->metadata_size); + PlasmaCreateReply_error_add(B, error_code); + PlasmaCreateReply_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaCreateReply); +} + +void plasma_read_CreateReply(uint8_t *data, + object_id *object_id, + plasma_object *object, + int *error_code) { + DCHECK(data); + PlasmaCreateReply_table_t rep = PlasmaCreateReply_as_root(data); + flatbuffers_string_t id = PlasmaCreateReply_object_id(rep); + CHECK(flatbuffers_string_len(id) == sizeof(object_id->id)); + memcpy(&object_id->id[0], id, sizeof(object_id->id)); + PlasmaObject_struct_t obj = PlasmaCreateReply_plasma_object(rep); + object->handle.store_fd = PlasmaObject_segment_index(obj); + object->handle.mmap_size = PlasmaObject_mmap_size(obj); + object->data_offset = PlasmaObject_data_offset(obj); + object->data_size = PlasmaObject_data_size(obj); + object->metadata_offset = PlasmaObject_metadata_offset(obj); + object->metadata_size = PlasmaObject_metadata_size(obj); + *error_code = PlasmaCreateReply_error(rep); +} + +#define DEFINE_SIMPLE_SEND_REQUEST(MESSAGE_NAME) \ + int plasma_send_##MESSAGE_NAME(int sock, protocol_builder *B, \ + object_id object_id) { \ + Plasma##MESSAGE_NAME##_start_as_root(B); \ + Plasma##MESSAGE_NAME##_object_id_create( \ + B, (const char *) &object_id.id[0], sizeof(object_id.id)); \ + Plasma##MESSAGE_NAME##_end_as_root(B); \ + return finalize_buffer_and_send(B, sock, \ + MessageType_Plasma##MESSAGE_NAME); \ + } + +#define DEFINE_SIMPLE_READ_REQUEST(MESSAGE_NAME) \ + void plasma_read_##MESSAGE_NAME(uint8_t *data, object_id *object_id) { \ + DCHECK(data); \ + Plasma##MESSAGE_NAME##_table_t req = Plasma##MESSAGE_NAME##_as_root(data); \ + flatbuffers_string_t id = Plasma##MESSAGE_NAME##_object_id(req); \ + CHECK(flatbuffers_string_len(id) == sizeof(object_id->id)); \ + memcpy(&object_id->id[0], id, sizeof(object_id->id)); \ + } + +#define DEFINE_SIMPLE_SEND_REPLY(MESSAGE_NAME) \ + int plasma_send_##MESSAGE_NAME(int sock, protocol_builder *B, \ + object_id object_id, int error) { \ + Plasma##MESSAGE_NAME##_start_as_root(B); \ + Plasma##MESSAGE_NAME##_object_id_create( \ + B, (const char *) &object_id.id[0], sizeof(object_id.id)); \ + Plasma##MESSAGE_NAME##_error_add(B, error); \ + Plasma##MESSAGE_NAME##_end_as_root(B); \ + return finalize_buffer_and_send(B, sock, \ + MessageType_Plasma##MESSAGE_NAME); \ + } + +#define DEFINE_SIMPLE_READ_REPLY(MESSAGE_NAME) \ + void plasma_read_##MESSAGE_NAME(uint8_t *data, object_id *object_id, \ + int *error) { \ + DCHECK(data); \ + Plasma##MESSAGE_NAME##_table_t req = Plasma##MESSAGE_NAME##_as_root(data); \ + flatbuffers_string_t id = Plasma##MESSAGE_NAME##_object_id(req); \ + CHECK(flatbuffers_string_len(id) == sizeof(object_id->id)); \ + memcpy(&object_id->id[0], id, sizeof(object_id->id)); \ + *error = Plasma##MESSAGE_NAME##_error(req); \ + } + +int plasma_send_SealRequest(int sock, + protocol_builder *B, + object_id object_id, + unsigned char *digest) { + PlasmaSealRequest_start_as_root(B); + PlasmaSealRequest_object_id_create(B, (const char *) &object_id.id[0], + sizeof(object_id.id)); + PlasmaSealRequest_digest_create(B, digest, DIGEST_SIZE); + PlasmaSealRequest_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaSealRequest); +} + +void plasma_read_SealRequest(uint8_t *data, + object_id *object_id, + unsigned char *digest) { + DCHECK(data); + PlasmaSealRequest_table_t req = PlasmaSealRequest_as_root(data); + flatbuffers_string_t id = PlasmaSealRequest_object_id(req); + CHECK(flatbuffers_string_len(id) == sizeof(object_id->id)); + memcpy(&object_id->id[0], id, sizeof(object_id->id)); + flatbuffers_uint8_vec_t d = PlasmaSealRequest_digest(req); + CHECK(flatbuffers_uint8_vec_len(d) == DIGEST_SIZE); + memcpy(digest, d, DIGEST_SIZE); +} + +DEFINE_SIMPLE_SEND_REPLY(SealReply); +DEFINE_SIMPLE_READ_REPLY(SealReply); + +DEFINE_SIMPLE_SEND_REQUEST(ReleaseRequest); +DEFINE_SIMPLE_READ_REQUEST(ReleaseRequest); +DEFINE_SIMPLE_SEND_REPLY(ReleaseReply); +DEFINE_SIMPLE_READ_REPLY(ReleaseReply); + +DEFINE_SIMPLE_SEND_REQUEST(DeleteRequest); +DEFINE_SIMPLE_READ_REQUEST(DeleteRequest); +DEFINE_SIMPLE_SEND_REPLY(DeleteReply); +DEFINE_SIMPLE_READ_REPLY(DeleteReply); + +/* Plasma status message. */ + +int plasma_send_StatusRequest(int sock, + protocol_builder *B, + object_id object_ids[], + int64_t num_objects) { + PlasmaStatusRequest_start_as_root(B); + PlasmaStatusRequest_object_ids_add( + B, object_ids_to_flatbuffer(B, object_ids, num_objects)); + PlasmaStatusRequest_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaStatusRequest); +} + +int64_t plasma_read_StatusRequest_num_objects(uint8_t *data) { + DCHECK(data); + PlasmaStatusRequest_table_t req = PlasmaStatusRequest_as_root(data); + return flatbuffers_string_vec_len(PlasmaStatusRequest_object_ids(req)); +} + +void plasma_read_StatusRequest(uint8_t *data, + object_id object_ids[], + int64_t num_objects) { + DCHECK(data); + PlasmaStatusRequest_table_t req = PlasmaStatusRequest_as_root(data); + object_ids_from_flatbuffer(PlasmaStatusRequest_object_ids(req), object_ids, + num_objects); +} + +int plasma_send_StatusReply(int sock, + protocol_builder *B, + object_id object_ids[], + int object_status[], + int64_t num_objects) { + PlasmaStatusReply_start_as_root(B); + PlasmaStatusReply_object_ids_add( + B, object_ids_to_flatbuffer(B, object_ids, num_objects)); + flatbuffers_int32_vec_start(B); + for (int64_t i = 0; i < num_objects; ++i) { + flatbuffers_int32_vec_push(B, &object_status[i]); + } + PlasmaStatusReply_status_add(B, flatbuffers_int32_vec_end(B)); + PlasmaStatusReply_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaStatusReply); +} + +int64_t plasma_read_StatusReply_num_objects(uint8_t *data) { + DCHECK(data); + PlasmaStatusReply_table_t req = PlasmaStatusReply_as_root(data); + return flatbuffers_string_vec_len(PlasmaStatusReply_object_ids(req)); +} + +void plasma_read_StatusReply(uint8_t *data, + object_id object_ids[], + int object_status[], + int64_t num_objects) { + DCHECK(data); + PlasmaStatusReply_table_t rep = PlasmaStatusReply_as_root(data); + object_ids_from_flatbuffer(PlasmaStatusReply_object_ids(rep), object_ids, + num_objects); + for (int64_t i = 0; i < num_objects; ++i) { + object_status[i] = + flatbuffers_int32_vec_at(PlasmaStatusReply_status(rep), i); + } +} + +/* Plasma contains message. */ + +int plasma_send_ContainsRequest(int sock, + protocol_builder *B, + object_id object_id) { + PlasmaContainsRequest_start_as_root(B); + PlasmaContainsRequest_object_id_create(B, (const char *) &object_id.id[0], + sizeof(object_id.id)); + PlasmaContainsRequest_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaContainsRequest); +} + +void plasma_read_ContainsRequest(uint8_t *data, object_id *object_id) { + DCHECK(data); + PlasmaContainsRequest_table_t req = PlasmaContainsRequest_as_root(data); + flatbuffers_string_t id = PlasmaContainsRequest_object_id(req); + CHECK(flatbuffers_string_len(id) == sizeof(object_id->id)); + memcpy(&object_id->id[0], id, sizeof(object_id->id)); +} + +int plasma_send_ContainsReply(int sock, + protocol_builder *B, + object_id object_id, + int has_object) { + PlasmaContainsReply_start_as_root(B); + PlasmaContainsReply_object_id_create(B, (const char *) &object_id.id[0], + sizeof(object_id.id)); + PlasmaContainsReply_has_object_add(B, has_object); + PlasmaContainsReply_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaContainsReply); +} + +void plasma_read_ContainsReply(uint8_t *data, + object_id *object_id, + int *has_object) { + DCHECK(data); + PlasmaContainsReply_table_t rep = PlasmaContainsReply_as_root(data); + flatbuffers_string_t id = PlasmaContainsReply_object_id(rep); + CHECK(flatbuffers_string_len(id) == sizeof(object_id->id)); + memcpy(&object_id->id[0], id, sizeof(object_id->id)); + *has_object = PlasmaContainsReply_has_object(rep); +} + +/* Plasma evict message. */ + +int plasma_send_EvictRequest(int sock, protocol_builder *B, int64_t num_bytes) { + PlasmaEvictRequest_start_as_root(B); + PlasmaEvictRequest_num_bytes_add(B, num_bytes); + PlasmaEvictRequest_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaEvictRequest); +} + +void plasma_read_EvictRequest(uint8_t *data, int64_t *num_bytes) { + DCHECK(data); + PlasmaEvictRequest_table_t req = PlasmaEvictRequest_as_root(data); + *num_bytes = PlasmaEvictRequest_num_bytes(req); +} + +int plasma_send_EvictReply(int sock, protocol_builder *B, int64_t num_bytes) { + PlasmaEvictReply_start_as_root(B); + PlasmaEvictReply_num_bytes_add(B, num_bytes); + PlasmaEvictReply_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaEvictReply); +} + +void plasma_read_EvictReply(uint8_t *data, int64_t *num_bytes) { + DCHECK(data); + PlasmaEvictReply_table_t req = PlasmaEvictReply_as_root(data); + *num_bytes = PlasmaEvictReply_num_bytes(req); +} + +/* Plasma Get message. */ + +int plasma_send_GetRequest(int sock, + protocol_builder *B, + object_id object_ids[], + int64_t num_objects) { + PlasmaGetRequest_start_as_root(B); + PlasmaGetRequest_object_ids_add( + B, object_ids_to_flatbuffer(B, object_ids, num_objects)); + PlasmaGetRequest_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaGetRequest); +} + +int64_t plasma_read_GetRequest_num_objects(uint8_t *data) { + DCHECK(data); + PlasmaGetRequest_table_t req = PlasmaGetRequest_as_root(data); + return flatbuffers_string_vec_len(PlasmaGetRequest_object_ids(req)); +} + +void plasma_read_GetRequest(uint8_t *data, + object_id object_ids[], + int64_t num_objects) { + DCHECK(data); + PlasmaGetRequest_table_t req = PlasmaGetRequest_as_root(data); + flatbuffers_string_vec_t object_id_vector = PlasmaGetRequest_object_ids(req); + object_ids_from_flatbuffer(object_id_vector, object_ids, num_objects); +} + +int plasma_send_GetReply(int sock, + protocol_builder *B, + object_id object_ids[], + plasma_object plasma_objects[], + int64_t num_objects) { + PlasmaGetReply_start_as_root(B); + + flatbuffers_string_vec_ref_t ids = + object_ids_to_flatbuffer(B, object_ids, num_objects); + PlasmaGetReply_object_ids_add(B, ids); + + PlasmaObject_vec_start(B); + for (int i = 0; i < num_objects; ++i) { + plasma_object obj = plasma_objects[i]; + PlasmaObject_t plasma_obj; + memset(&plasma_obj, 0, sizeof(PlasmaObject_t)); + plasma_obj.segment_index = obj.handle.store_fd; + plasma_obj.mmap_size = obj.handle.mmap_size; + plasma_obj.data_offset = obj.data_offset; + plasma_obj.data_size = obj.data_size; + plasma_obj.metadata_offset = obj.metadata_offset; + plasma_obj.metadata_size = obj.metadata_size; + PlasmaObject_vec_push(B, &plasma_obj); + } + PlasmaObject_vec_ref_t object_vec = PlasmaObject_vec_end(B); + PlasmaGetReply_plasma_objects_add(B, object_vec); + PlasmaGetReply_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaGetReply); +} + +void plasma_read_GetReply(uint8_t *data, + object_id object_ids[], + plasma_object plasma_objects[], + int64_t num_objects) { + CHECK(data); + PlasmaGetReply_table_t req = PlasmaGetReply_as_root(data); + flatbuffers_string_vec_t object_id_vector = PlasmaGetReply_object_ids(req); + object_ids_from_flatbuffer(object_id_vector, object_ids, num_objects); + + memset(plasma_objects, 0, sizeof(plasma_object) * num_objects); + PlasmaObject_vec_t plasma_objects_vector = PlasmaGetReply_plasma_objects(req); + + for (int i = 0; i < num_objects; ++i) { + PlasmaObject_struct_t obj = PlasmaObject_vec_at(plasma_objects_vector, i); + plasma_objects[i].handle.store_fd = PlasmaObject_segment_index(obj); + plasma_objects[i].handle.mmap_size = PlasmaObject_mmap_size(obj); + plasma_objects[i].data_offset = PlasmaObject_data_offset(obj); + plasma_objects[i].data_size = PlasmaObject_data_size(obj); + plasma_objects[i].metadata_offset = PlasmaObject_metadata_offset(obj); + plasma_objects[i].metadata_size = PlasmaObject_metadata_size(obj); + } +} + +/* Plasma get local messages. */ + +int plasma_send_GetLocalRequest(int sock, + protocol_builder *B, + object_id object_ids[], + int64_t num_objects) { + PlasmaGetLocalRequest_start_as_root(B); + PlasmaGetLocalRequest_object_ids_add( + B, object_ids_to_flatbuffer(B, object_ids, num_objects)); + PlasmaGetLocalRequest_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaGetLocalRequest); +} + +int64_t plasma_read_GetLocalRequest_num_objects(uint8_t *data) { + DCHECK(data); + PlasmaGetLocalRequest_table_t req = PlasmaGetLocalRequest_as_root(data); + return flatbuffers_string_vec_len(PlasmaGetLocalRequest_object_ids(req)); +} + +void plasma_read_GetLocalRequest(uint8_t *data, + object_id object_ids[], + int64_t num_objects) { + DCHECK(data); + PlasmaGetLocalRequest_table_t req = PlasmaGetLocalRequest_as_root(data); + flatbuffers_string_vec_t object_id_vector = + PlasmaGetLocalRequest_object_ids(req); + object_ids_from_flatbuffer(object_id_vector, object_ids, num_objects); +} + +int plasma_send_GetLocalReply(int sock, + protocol_builder *B, + object_id object_ids[], + plasma_object plasma_objects[], + int has_object[], + int64_t num_objects) { + PlasmaGetLocalReply_start_as_root(B); + + flatbuffers_string_vec_ref_t ids = + object_ids_to_flatbuffer(B, object_ids, num_objects); + PlasmaGetLocalReply_object_ids_add(B, ids); + + PlasmaObject_vec_start(B); + for (int i = 0; i < num_objects; ++i) { + plasma_object obj = plasma_objects[i]; + PlasmaObject_t plasma_obj; + memset(&plasma_obj, 0, sizeof(PlasmaObject_t)); + plasma_obj.segment_index = obj.handle.store_fd; + plasma_obj.mmap_size = obj.handle.mmap_size; + plasma_obj.data_offset = obj.data_offset; + plasma_obj.data_size = obj.data_size; + plasma_obj.metadata_offset = obj.metadata_offset; + plasma_obj.metadata_size = obj.metadata_size; + PlasmaObject_vec_push(B, &plasma_obj); + } + PlasmaObject_vec_ref_t object_vec = PlasmaObject_vec_end(B); + PlasmaGetLocalReply_plasma_objects_add(B, object_vec); + + flatbuffers_int32_vec_start(B); + for (int64_t i = 0; i < num_objects; ++i) { + flatbuffers_int32_vec_push(B, &has_object[i]); + } + PlasmaGetLocalReply_has_object_add(B, flatbuffers_int32_vec_end(B)); + PlasmaGetLocalReply_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaGetLocalReply); +} + +void plasma_read_GetLocalReply(uint8_t *data, + object_id object_ids[], + plasma_object plasma_objects[], + int has_object[], + int64_t num_objects) { + CHECK(data); + PlasmaGetLocalReply_table_t req = PlasmaGetLocalReply_as_root(data); + flatbuffers_string_vec_t object_id_vector = + PlasmaGetLocalReply_object_ids(req); + object_ids_from_flatbuffer(object_id_vector, object_ids, num_objects); + + memset(plasma_objects, 0, sizeof(plasma_object) * num_objects); + PlasmaObject_vec_t plasma_objects_vector = + PlasmaGetLocalReply_plasma_objects(req); + + for (int i = 0; i < num_objects; ++i) { + PlasmaObject_struct_t obj = PlasmaObject_vec_at(plasma_objects_vector, i); + plasma_objects[i].handle.store_fd = PlasmaObject_segment_index(obj); + plasma_objects[i].handle.mmap_size = PlasmaObject_mmap_size(obj); + plasma_objects[i].data_offset = PlasmaObject_data_offset(obj); + plasma_objects[i].data_size = PlasmaObject_data_size(obj); + plasma_objects[i].metadata_offset = PlasmaObject_metadata_offset(obj); + plasma_objects[i].metadata_size = PlasmaObject_metadata_size(obj); + has_object[i] = + flatbuffers_int32_vec_at(PlasmaGetLocalReply_has_object(req), i); + } +} + +/* Plasma fetch messages. */ + +int plasma_send_FetchRequest(int sock, + protocol_builder *B, + object_id object_ids[], + int64_t num_objects) { + PlasmaFetchRequest_start_as_root(B); + PlasmaFetchRequest_object_ids_add( + B, object_ids_to_flatbuffer(B, object_ids, num_objects)); + PlasmaFetchRequest_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaFetchRequest); +} + +int64_t plasma_read_FetchRequest_num_objects(uint8_t *data) { + DCHECK(data); + PlasmaFetchRequest_table_t req = PlasmaFetchRequest_as_root(data); + return ObjectRequest_vec_len(PlasmaFetchRequest_object_ids(req)); +} + +void plasma_read_FetchRequest(uint8_t *data, + object_id object_ids[], + int64_t num_objects) { + DCHECK(data); + PlasmaFetchRequest_table_t req = PlasmaFetchRequest_as_root(data); + flatbuffers_string_vec_t object_id_vector = + PlasmaFetchRequest_object_ids(req); + object_ids_from_flatbuffer(object_id_vector, object_ids, num_objects); +} + +/* Plasma wait messages. */ + +int plasma_send_WaitRequest(int sock, + protocol_builder *B, + object_request object_requests[], + int num_requests, + int num_ready_objects, + int64_t timeout_ms) { + PlasmaWaitRequest_start_as_root(B); + ObjectRequest_vec_start(B); + for (int i = 0; i < num_requests; ++i) { + flatbuffers_string_ref_t id = flatbuffers_string_create( + B, (const char *) &object_requests[i].object_id.id[0], + sizeof(object_requests[i].object_id.id)); + ObjectRequest_vec_push_create(B, id, (int32_t) object_requests[i].type); + } + ObjectRequest_vec_ref_t objreq_vec = ObjectRequest_vec_end(B); + PlasmaWaitRequest_object_requests_add(B, objreq_vec); + PlasmaWaitRequest_num_ready_objects_add(B, num_ready_objects); + PlasmaWaitRequest_timeout_add(B, timeout_ms); + PlasmaWaitRequest_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaWaitRequest); +} + +int plasma_read_WaitRequest_num_object_ids(uint8_t *data) { + DCHECK(data); + PlasmaWaitRequest_table_t req = PlasmaWaitRequest_as_root(data); + return ObjectRequest_vec_len(PlasmaWaitRequest_object_requests(req)); +} + +void plasma_read_WaitRequest(uint8_t *data, + object_request object_requests[], + int num_object_ids, + int64_t *timeout_ms, + int *num_ready_objects) { + DCHECK(data); + PlasmaWaitRequest_table_t req = PlasmaWaitRequest_as_root(data); + ObjectRequest_vec_t objreq_vec = PlasmaWaitRequest_object_requests(req); + CHECK(num_object_ids == ObjectRequest_vec_len(objreq_vec)); + for (int i = 0; i < num_object_ids; i++) { + ObjectRequest_table_t objreq = ObjectRequest_vec_at(objreq_vec, i); + memcpy(&object_requests[i].object_id.id[0], ObjectRequest_object_id(objreq), + sizeof(object_requests[i].object_id.id)); + object_requests[i].type = ObjectRequest_type(objreq); + } + *timeout_ms = PlasmaWaitRequest_timeout(req); + *num_ready_objects = PlasmaWaitRequest_num_ready_objects(req); +} + +int plasma_send_WaitReply(int sock, + protocol_builder *B, + object_request object_requests[], + int num_ready_objects) { + PlasmaWaitReply_start_as_root(B); + ObjectReply_vec_start(B); + for (int i = 0; i < num_ready_objects; ++i) { + flatbuffers_string_ref_t id = flatbuffers_string_create( + B, (const char *) &object_requests[i].object_id.id[0], + sizeof(object_requests[i].object_id.id)); + ObjectReply_vec_push_create(B, id, (int32_t) object_requests[i].status); + } + ObjectReply_vec_ref_t objreq_vec = ObjectReply_vec_end(B); + PlasmaWaitReply_object_requests_add(B, objreq_vec); + PlasmaWaitReply_num_ready_objects_add(B, num_ready_objects); + PlasmaWaitReply_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaWaitReply); +} + +void plasma_read_WaitReply(uint8_t *data, + object_request object_requests[], + int *num_ready_objects) { + DCHECK(data); + PlasmaWaitReply_table_t req = PlasmaWaitReply_as_root(data); + ObjectReply_vec_t objreq_vec = PlasmaWaitReply_object_requests(req); + /* TODO (ion): This is risky, maybe num_ready_objects should contain length of + * object_request object_requests? */ + *num_ready_objects = ObjectReply_vec_len(objreq_vec); + for (int i = 0; i < *num_ready_objects; i++) { + ObjectReply_table_t objreq = ObjectReply_vec_at(objreq_vec, i); + memcpy(&object_requests[i].object_id.id[0], ObjectReply_object_id(objreq), + sizeof(object_requests[i].object_id.id)); + object_requests[i].status = ObjectReply_status(objreq); + } +} + +int plasma_send_SubscribeRequest(int sock, protocol_builder *B) { + PlasmaSubscribeRequest_start_as_root(B); + PlasmaSubscribeRequest_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaSubscribeRequest); +} + +int plasma_send_DataRequest(int sock, + protocol_builder *B, + object_id object_id, + const char *address, + int port) { + PlasmaDataRequest_start_as_root(B); + PlasmaDataRequest_object_id_create(B, (const char *) &object_id.id[0], + sizeof(object_id.id)); + PlasmaDataRequest_address_create(B, address, strlen(address)); + PlasmaDataRequest_port_add(B, port); + PlasmaDataRequest_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaDataRequest); +} + +void plasma_read_DataRequest(uint8_t *data, + object_id *object_id, + char **address, + int *port) { + DCHECK(data); + PlasmaDataRequest_table_t req = PlasmaDataRequest_as_root(data); + flatbuffers_string_t id = PlasmaDataRequest_object_id(req); + DCHECK(flatbuffers_string_len(id) == sizeof(object_id->id)); + memcpy(&object_id->id[0], id, sizeof(object_id->id)); + *address = strdup(PlasmaDataRequest_address(req)); + *port = PlasmaDataRequest_port(req); +} + +int plasma_send_DataReply(int sock, + protocol_builder *B, + object_id object_id, + int64_t object_size, + int64_t metadata_size) { + PlasmaDataReply_start_as_root(B); + PlasmaDataReply_object_id_create(B, (const char *) &object_id.id[0], + sizeof(object_id.id)); + PlasmaDataReply_object_size_add(B, object_size); + PlasmaDataReply_metadata_size_add(B, metadata_size); + PlasmaDataRequest_end_as_root(B); + return finalize_buffer_and_send(B, sock, MessageType_PlasmaDataReply); +} + +void plasma_read_DataReply(uint8_t *data, + object_id *object_id, + int64_t *object_size, + int64_t *metadata_size) { + DCHECK(data); + PlasmaDataReply_table_t rep = PlasmaDataReply_as_root(data); + flatbuffers_string_t id = PlasmaDataReply_object_id(rep); + DCHECK(flatbuffers_string_len(id) == sizeof(object_id->id)); + memcpy(&object_id->id[0], id, sizeof(object_id->id)); + *object_size = PlasmaDataReply_object_size(rep); + *metadata_size = PlasmaDataReply_metadata_size(rep); +} diff --git a/src/plasma/plasma_protocol.h b/src/plasma/plasma_protocol.h new file mode 100644 index 000000000..b474c16a6 --- /dev/null +++ b/src/plasma/plasma_protocol.h @@ -0,0 +1,267 @@ +#ifndef PLASMA_PROTOCOL_H +#define PLASMA_PROTOCOL_H + +#include "common.h" +#include "plasma.h" + +#include "format/plasma_builder.h" +#include "format/plasma_reader.h" + +/* An argument to a function that a return value gets written to. */ +#define OUT + +/* Protocol builder. */ + +typedef flatcc_builder_t protocol_builder; + +protocol_builder *make_protocol_builder(void); + +void free_protocol_builder(protocol_builder *builder); + +/* Plasma receive message. */ + +uint8_t *plasma_receive(int sock, int64_t message_type); + +/* Plasma Create message functions. */ + +int plasma_send_CreateRequest(int sock, + protocol_builder *B, + object_id object_id, + int64_t data_size, + int64_t metadata_size); + +void plasma_read_CreateRequest(uint8_t *data, + object_id *object_id, + int64_t *data_size, + int64_t *metadata_size); + +int plasma_send_CreateReply(int sock, + protocol_builder *B, + object_id object_id, + plasma_object *object, + int error); + +void plasma_read_CreateReply(uint8_t *data, + object_id *object_id, + plasma_object *object, + int *error); + +/* Plasma Seal message functions. */ + +int plasma_send_SealRequest(int sock, + protocol_builder *B, + object_id object_id, + unsigned char *digest); + +void plasma_read_SealRequest(uint8_t *data, + OUT object_id *object_id, + OUT unsigned char *digest); + +int plasma_send_SealReply(int sock, + protocol_builder *B, + object_id object_id, + int error); + +void plasma_read_SealReply(uint8_t *data, object_id *object_id, int *error); + +/* Plasma Get message functions. */ + +int plasma_send_GetRequest(int sock, + protocol_builder *B, + object_id object_ids[], + int64_t num_objects); + +int64_t plasma_read_GetRequest_num_objects(uint8_t *data); + +void plasma_read_GetRequest(uint8_t *data, + object_id object_ids[], + int64_t num_objects); + +int plasma_send_GetReply(int sock, + protocol_builder *B, + object_id object_ids[], + plasma_object plasma_objects[], + int64_t num_objects); + +void plasma_read_GetReply(uint8_t *data, + object_id object_ids[], + plasma_object plasma_objects[], + int64_t num_objects); + +/* Plasma Get local message functions. */ + +int plasma_send_GetLocalRequest(int sock, + protocol_builder *B, + object_id object_ids[], + int64_t num_objects); + +int64_t plasma_read_GetLocalRequest_num_objects(uint8_t *data); + +void plasma_read_GetLocalRequest(uint8_t *data, + object_id object_ids[], + int64_t num_objects); + +int plasma_send_GetLocalReply(int sock, + protocol_builder *B, + object_id object_ids[], + plasma_object plasma_objects[], + int has_object[], + int64_t num_objects); + +void plasma_read_GetLocalReply(uint8_t *data, + object_id object_ids[], + plasma_object plasma_objects[], + int has_object[], + int64_t num_objects); + +/* Plasma Release message functions. */ + +int plasma_send_ReleaseRequest(int sock, + protocol_builder *B, + object_id object_id); + +void plasma_read_ReleaseRequest(uint8_t *data, object_id *object_id); + +int plasma_send_ReleaseReply(int sock, + protocol_builder *B, + object_id object_id, + int error); + +void plasma_read_ReleaseReply(uint8_t *data, object_id *object_id, int *error); + +/* Plasma Delete message functions. */ + +int plasma_send_DeleteRequest(int sock, + protocol_builder *B, + object_id object_id); + +void plasma_read_DeleteRequest(uint8_t *data, object_id *object_id); + +int plasma_send_DeleteReply(int sock, + protocol_builder *B, + object_id object_id, + int error); + +void plasma_read_DeleteReply(uint8_t *data, object_id *object_id, int *error); + +/* Plasma Status message functions. */ + +int plasma_send_StatusRequest(int sock, + protocol_builder *B, + object_id object_ids[], + int64_t num_objects); + +int64_t plasma_read_StatusRequest_num_objects(uint8_t *data); + +void plasma_read_StatusRequest(uint8_t *data, + object_id object_ids[], + int64_t num_objects); + +int plasma_send_StatusReply(int sock, + protocol_builder *B, + object_id object_ids[], + int object_status[], + int64_t num_objects); + +int64_t plasma_read_StatusReply_num_objects(uint8_t *data); + +void plasma_read_StatusReply(uint8_t *data, + object_id object_ids[], + int object_status[], + int64_t num_objects); + +/* Plasma Constains message functions. */ + +int plasma_send_ContainsRequest(int sock, + protocol_builder *B, + object_id object_id); + +void plasma_read_ContainsRequest(uint8_t *data, object_id *object_id); + +int plasma_send_ContainsReply(int sock, + protocol_builder *B, + object_id object_id, + int has_object); + +void plasma_read_ContainsReply(uint8_t *data, + object_id *object_id, + int *has_object); + +/* Plasma Evict message functions (no reply so far). */ + +int plasma_send_EvictRequest(int sock, protocol_builder *B, int64_t num_bytes); + +void plasma_read_EvictRequest(uint8_t *data, int64_t *num_bytes); + +int plasma_send_EvictReply(int sock, protocol_builder *B, int64_t num_bytes); + +void plasma_read_EvictReply(uint8_t *data, int64_t *num_bytes); + +/* Plasma Fetch Remote message functions. */ + +int plasma_send_FetchRequest(int sock, + protocol_builder *B, + object_id object_ids[], + int64_t num_objects); + +int64_t plasma_read_FetchRequest_num_objects(uint8_t *data); + +void plasma_read_FetchRequest(uint8_t *data, + object_id object_ids[], + int64_t num_objects); + +/* Plasma Wait message functions. */ + +int plasma_send_WaitRequest(int sock, + protocol_builder *B, + object_request object_requests[], + int num_requests, + int num_ready_objects, + int64_t timeout_ms); + +int plasma_read_WaitRequest_num_object_ids(uint8_t *data); + +void plasma_read_WaitRequest(uint8_t *data, + object_request object_requests[], + int num_object_ids, + int64_t *timeout_ms, + int *num_ready_objects); + +int plasma_send_WaitReply(int sock, + protocol_builder *B, + object_request object_requests[], + int num_ready_objects); + +void plasma_read_WaitReply(uint8_t *data, + object_request object_requests[], + int *num_ready_objects); + +/* Plasma Subscribe message functions. */ + +int plasma_send_SubscribeRequest(int sock, protocol_builder *B); + +/* Plasma Data message functions. */ + +int plasma_send_DataRequest(int sock, + protocol_builder *B, + object_id object_id, + const char *address, + int port); + +void plasma_read_DataRequest(uint8_t *data, + object_id *object_id, + char **address, + int *port); + +int plasma_send_DataReply(int sock, + protocol_builder *B, + object_id object_id, + int64_t object_size, + int64_t metadata_size); + +void plasma_read_DataReply(uint8_t *data, + object_id *object_id, + int64_t *object_size, + int64_t *metadata_size); + +#endif /* PLASMA_PROTOCOL */ diff --git a/src/plasma/plasma_store.c b/src/plasma/plasma_store.c index 7aab1e5cb..04e806606 100644 --- a/src/plasma/plasma_store.c +++ b/src/plasma/plasma_store.c @@ -31,6 +31,7 @@ #include "utarray.h" #include "fling.h" #include "malloc.h" +#include "plasma_protocol.h" #include "plasma_store.h" #include "plasma.h" @@ -85,6 +86,8 @@ struct plasma_store_state { /** Input buffer. This is allocated only once to avoid mallocs for every * call to process_message. */ UT_array *input_buffer; + /** Buffer that holds memory for serializing plasma protocol messages. */ + protocol_builder *builder; }; UT_icd byte_icd = {sizeof(uint8_t), NULL, NULL, NULL}; @@ -100,6 +103,7 @@ plasma_store_state *init_plasma_store(event_loop *loop, int64_t system_memory) { /* Initialize the eviction state. */ state->eviction_state = make_eviction_state(system_memory); utarray_new(state->input_buffer, &byte_icd); + state->builder = make_protocol_builder(); return state; } @@ -330,20 +334,20 @@ void seal_object(client *client_context, HASH_FIND(handle, plasma_state->objects_notify, &object_id, sizeof(object_id), notify_entry); if (notify_entry) { - 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; - 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; + plasma_object object; + object.handle.store_fd = entry->fd; + object.handle.mmap_size = entry->map_size; + object.data_offset = entry->offset; + object.metadata_offset = entry->offset + entry->info.data_size; + object.data_size = entry->info.data_size; + object.metadata_size = entry->info.metadata_size; HASH_DELETE(handle, plasma_state->objects_notify, notify_entry); /* 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); - CHECK(plasma_send_reply((*c)->sock, &reply) >= 0); - CHECK(send_fd((*c)->sock, reply.object.handle.store_fd) >= 0); + CHECK(plasma_send_GetReply((*c)->sock, plasma_state->builder, &object_id, + &object, 1) >= 0); + CHECK(send_fd((*c)->sock, object.handle.store_fd) >= 0); /* Record that the client is using this object. */ add_client_to_object_clients(entry, *c); } @@ -484,81 +488,93 @@ void process_message(event_loop *loop, plasma_store_state *state = client_context->plasma_state; int64_t type; read_buffer(client_sock, &type, state->input_buffer); - 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 = plasma_make_reply(NIL_OBJECT_ID); + uint8_t *input = (uint8_t *) utarray_front(state->input_buffer); + object_id object_ids[1]; + int64_t num_objects; + plasma_object objects[1]; + memset(&objects[0], 0, sizeof(objects)); + int error; + + flatcc_builder_reset(state->builder); + /* Process the different types of requests. */ switch (type) { - case PLASMA_CREATE: - DCHECK(req->num_object_ids == 1); - if (create_object(client_context, req->object_requests[0].object_id, - req->data_size, req->metadata_size, &reply.object)) { - reply.error_code = PLASMA_REPLY_OK; + case MessageType_PlasmaCreateRequest: { + int64_t data_size; + int64_t metadata_size; + plasma_read_CreateRequest(input, &object_ids[0], &data_size, + &metadata_size); + if (create_object(client_context, object_ids[0], data_size, metadata_size, + &objects[0])) { + error = PlasmaError_OK; } else { - reply.error_code = PLASMA_OBJECT_ALREADY_EXISTS; + error = PlasmaError_ObjectExists; } - CHECK(plasma_send_reply(client_sock, &reply) >= 0); - CHECK(send_fd(client_sock, reply.object.handle.store_fd) >= 0); - break; - case PLASMA_GET: - DCHECK(req->num_object_ids == 1); - if (get_object(client_context, client_sock, - req->object_requests[0].object_id, - &reply.object) == OBJECT_FOUND) { - CHECK(plasma_send_reply(client_sock, &reply) >= 0); - CHECK(send_fd(client_sock, reply.object.handle.store_fd) >= 0); + CHECK(plasma_send_CreateReply(client_sock, state->builder, object_ids[0], + &objects[0], error) >= 0); + if (error == PlasmaError_OK) { + CHECK(send_fd(client_sock, objects[0].handle.store_fd) >= 0); } - break; - case PLASMA_GET_LOCAL: - DCHECK(req->num_object_ids == 1); - if (get_object_local(client_context, client_sock, - req->object_requests[0].object_id, - &reply.object) == OBJECT_FOUND) { - reply.has_object = true; - CHECK(plasma_send_reply(client_sock, &reply) >= 0); - CHECK(send_fd(client_sock, reply.object.handle.store_fd) >= 0); - } else { - reply.has_object = false; - CHECK(plasma_send_reply(client_sock, &reply) >= 0); - CHECK(send_fd(client_sock, reply.object.handle.store_fd) >= 0); - } - break; - case PLASMA_RELEASE: - DCHECK(req->num_object_ids == 1); - release_object(client_context, req->object_requests[0].object_id); - break; - case PLASMA_CONTAINS: - DCHECK(req->num_object_ids == 1); - if (contains_object(client_context, req->object_requests[0].object_id) == + } break; + case MessageType_PlasmaGetRequest: { + plasma_read_GetRequest(input, object_ids, 1); + if (get_object(client_context, client_sock, object_ids[0], &objects[0]) == OBJECT_FOUND) { - reply.has_object = 1; + CHECK(plasma_send_GetReply(client_sock, state->builder, object_ids, + objects, 1) >= 0); + CHECK(send_fd(client_sock, objects[0].handle.store_fd) >= 0); } - CHECK(plasma_send_reply(client_sock, &reply) >= 0); + } break; + case MessageType_PlasmaGetLocalRequest: { + plasma_read_GetLocalRequest(input, &object_ids[0], 1); + if (get_object_local(client_context, client_sock, object_ids[0], + &objects[0]) == OBJECT_FOUND) { + int has_object = 1; + CHECK(plasma_send_GetLocalReply(client_sock, state->builder, object_ids, + objects, &has_object, 1) >= 0); + CHECK(send_fd(client_sock, objects[0].handle.store_fd) >= 0); + } else { + int has_object = 0; + CHECK(plasma_send_GetLocalReply(client_sock, state->builder, object_ids, + objects, &has_object, 1) >= 0); + } + } break; + case MessageType_PlasmaReleaseRequest: + plasma_read_ReleaseRequest(input, &object_ids[0]); + release_object(client_context, object_ids[0]); break; - case PLASMA_SEAL: - DCHECK(req->num_object_ids == 1); - seal_object(client_context, req->object_requests[0].object_id, req->digest); + case MessageType_PlasmaContainsRequest: + plasma_read_ContainsRequest(input, &object_ids[0]); + if (contains_object(client_context, object_ids[0]) == OBJECT_FOUND) { + CHECK(plasma_send_ContainsReply(client_sock, state->builder, + object_ids[0], 1) >= 0); + } else { + CHECK(plasma_send_ContainsReply(client_sock, state->builder, + object_ids[0], 0) >= 0); + } break; - case PLASMA_DELETE: - /* TODO(rkn): In the future, we can use this method to give hints to the - * eviction policy about when an object will no longer be needed. */ - break; - case PLASMA_EVICT: { + case MessageType_PlasmaSealRequest: { + unsigned char digest[DIGEST_SIZE]; + plasma_read_SealRequest(input, &object_ids[0], &digest[0]); + seal_object(client_context, object_ids[0], &digest[0]); + } break; + case MessageType_PlasmaEvictRequest: { /* This code path should only be used for testing. */ + int64_t num_bytes; + plasma_read_EvictRequest(input, &num_bytes); int64_t num_objects_to_evict; object_id *objects_to_evict; int64_t num_bytes_evicted = choose_objects_to_evict( client_context->plasma_state->eviction_state, - client_context->plasma_state->plasma_store_info, req->num_bytes, + client_context->plasma_state->plasma_store_info, num_bytes, &num_objects_to_evict, &objects_to_evict); remove_objects(client_context->plasma_state, num_objects_to_evict, objects_to_evict); - reply.num_bytes = num_bytes_evicted; - CHECK(plasma_send_reply(client_sock, &reply) >= 0); - break; - } - case PLASMA_SUBSCRIBE: + CHECK(plasma_send_EvictReply(client_sock, state->builder, + num_bytes_evicted) >= 0); + } break; + case MessageType_PlasmaSubscribeRequest: subscribe_to_updates(client_context, client_sock); break; case DISCONNECT_CLIENT: { diff --git a/src/plasma/test/client_tests.c b/src/plasma/test/client_tests.c index d0f5d869f..06a19744b 100644 --- a/src/plasma/test/client_tests.c +++ b/src/plasma/test/client_tests.c @@ -5,6 +5,7 @@ #include #include "plasma.h" +#include "plasma_protocol.h" #include "plasma_client.h" SUITE(plasma_client_tests); @@ -18,7 +19,7 @@ TEST plasma_status_tests(void) { /* Test for object non-existence. */ int status = plasma_status(plasma_conn1, oid1); - ASSERT(status == PLASMA_OBJECT_NONEXISTENT); + ASSERT(status == ObjectStatus_Nonexistent); /* Test for the object being in local Plasma store. */ /* First create object. */ @@ -32,11 +33,11 @@ TEST plasma_status_tests(void) { */ sleep(1); status = plasma_status(plasma_conn1, oid1); - ASSERT(status == PLASMA_OBJECT_LOCAL); + ASSERT(status == ObjectStatus_Local); /* Test for object being remote. */ status = plasma_status(plasma_conn2, oid1); - ASSERT(status == PLASMA_OBJECT_REMOTE); + ASSERT(status == ObjectStatus_Remote); plasma_disconnect(plasma_conn1); plasma_disconnect(plasma_conn2); @@ -56,7 +57,7 @@ TEST plasma_fetch_tests(void) { /* No object in the system */ status = plasma_status(plasma_conn1, oid1); - ASSERT(status == PLASMA_OBJECT_NONEXISTENT); + ASSERT(status == ObjectStatus_Nonexistent); /* Test for the object being in local Plasma store. */ /* First create object. */ @@ -73,24 +74,24 @@ TEST plasma_fetch_tests(void) { object_id oid_array1[1] = {oid1}; plasma_fetch(plasma_conn1, 1, oid_array1); status = plasma_status(plasma_conn1, oid1); - ASSERT((status == PLASMA_OBJECT_LOCAL) || - (status == PLASMA_OBJECT_NONEXISTENT)); + ASSERT((status == ObjectStatus_Local) || + (status == ObjectStatus_Nonexistent)); /* Sleep to make sure Plasma Manager got the notification. */ sleep(1); status = plasma_status(plasma_conn1, oid1); - ASSERT(status == PLASMA_OBJECT_LOCAL); + ASSERT(status == ObjectStatus_Local); /* Test for object being remote. */ status = plasma_status(plasma_conn2, oid1); - ASSERT(status == PLASMA_OBJECT_REMOTE); + ASSERT(status == ObjectStatus_Remote); /* Sleep to make sure the object has been fetched and it is now stored in the * local Plasma Store. */ plasma_fetch(plasma_conn2, 1, oid_array1); sleep(1); status = plasma_status(plasma_conn2, oid1); - ASSERT(status == PLASMA_OBJECT_LOCAL); + ASSERT(status == ObjectStatus_Local); sleep(1); plasma_disconnect(plasma_conn1); diff --git a/src/plasma/test/manager_tests.c b/src/plasma/test/manager_tests.c index 570f7c7e0..24439a97c 100644 --- a/src/plasma/test/manager_tests.c +++ b/src/plasma/test/manager_tests.c @@ -15,6 +15,7 @@ #include "plasma.h" #include "plasma_client.h" #include "plasma_manager.h" +#include "plasma_protocol.h" SUITE(plasma_manager_tests); @@ -113,8 +114,8 @@ void destroy_plasma_mock(plasma_mock *mock) { * - Buffer a transfer request for the remote manager. * - Start and stop the event loop to make sure that we send the buffered * request. - * - Expect to see a PLASMA_TRANSFER message on the remote manager with the - * correct object ID. + * - Expect to see a MessageType_PlasmaDataRequest message on the remote manager + * with the correct object ID. */ TEST request_transfer_test(void) { plasma_mock *local_mock = init_plasma_mock(NULL); @@ -129,17 +130,18 @@ TEST request_transfer_test(void) { event_loop_add_timer(local_mock->loop, MANAGER_TIMEOUT, test_done_handler, local_mock->state); event_loop_run(local_mock->loop); - int64_t type; - int64_t length; - plasma_request *req; int read_fd = get_client_sock(remote_mock->read_conn); - read_message(read_fd, &type, &length, (uint8_t **) &req); - ASSERT(type == PLASMA_TRANSFER); - ASSERT(req->num_object_ids == 1); - ASSERT(object_ids_equal(oid, req->object_requests[0].object_id)); + uint8_t *request_data = + plasma_receive(read_fd, MessageType_PlasmaDataRequest); + object_id oid2; + char *address; + int port; + plasma_read_DataRequest(request_data, &oid2, &address, &port); + ASSERT(object_ids_equal(oid, oid2)); + free(address); /* Clean up. */ utstring_free(addr); - free(req); + free(request_data); destroy_plasma_mock(remote_mock); destroy_plasma_mock(local_mock); PASS(); @@ -154,8 +156,8 @@ TEST request_transfer_test(void) { * - Buffer a transfer request for the remote managers. * - Start and stop the event loop after a timeout to make sure that we * trigger the timeout on the first manager. - * - Expect to see a PLASMA_TRANSFER message on the second remote manager - * with the correct object ID. + * - Expect to see a MessageType_PlasmaDataRequest message on the second remote + * manager with the correct object ID. */ TEST request_transfer_retry_test(void) { plasma_mock *local_mock = init_plasma_mock(NULL); @@ -176,18 +178,19 @@ TEST request_transfer_retry_test(void) { local_mock->state); event_loop_run(local_mock->loop); - int64_t type; - int64_t length; - plasma_request *req; int read_fd = get_client_sock(remote_mock2->read_conn); - read_message(read_fd, &type, &length, (uint8_t **) &req); - ASSERT(type == PLASMA_TRANSFER); - ASSERT(req->num_object_ids == 1); - ASSERT(object_ids_equal(oid, req->object_requests[0].object_id)); + uint8_t *request_data = + plasma_receive(read_fd, MessageType_PlasmaDataRequest); + object_id oid2; + char *address; + int port; + plasma_read_DataRequest(request_data, &oid2, &address, &port); + free(address); + ASSERT(object_ids_equal(oid, oid2)); /* Clean up. */ utstring_free(addr0); utstring_free(addr1); - free(req); + free(request_data); destroy_plasma_mock(remote_mock2); destroy_plasma_mock(remote_mock1); destroy_plasma_mock(local_mock); @@ -209,7 +212,7 @@ TEST read_write_object_chunk_test(void) { const int data_size = strlen(data) + 1; const int metadata_size = 0; plasma_request_buffer remote_buf = { - .type = PLASMA_DATA, + .type = MessageType_PlasmaDataReply, .object_id = oid, .data = (uint8_t *) data, .data_size = data_size, diff --git a/src/plasma/test/serialization_tests.c b/src/plasma/test/serialization_tests.c new file mode 100644 index 000000000..d6e47e465 --- /dev/null +++ b/src/plasma/test/serialization_tests.c @@ -0,0 +1,490 @@ +#include "greatest.h" + +#include +#include + +#include "plasma.h" +#include "plasma_protocol.h" +#include "common.h" +#include "io.h" +#include "../plasma.h" + +SUITE(plasma_serialization_tests); + +protocol_builder *g_B; + +/** + * Create a temporary file. Needs to be closed by the caller. + * + * @return File descriptor of the file. + */ +int create_temp_file(void) { + static char template[] = "/tmp/tempfileXXXXXX"; + char file_name[32]; + strncpy(file_name, template, 32); + return mkstemp(file_name); +} + +/** + * Seek to the beginning of a file and read a message from it. + * + * @param fd File descriptor of the file. + * @param message type Message type that we expect in the file. + * + * @return Pointer to the content of the message. Needs to be freed by the + * caller. + */ +uint8_t *read_message_from_file(int fd, int message_type) { + /* Go to the beginning of the file. */ + lseek(fd, 0, SEEK_SET); + int64_t type; + int64_t length; + uint8_t *data; + read_message(fd, &type, &length, &data); + CHECK(type == message_type); + return data; +} + +plasma_object random_plasma_object(void) { + int random = rand(); + plasma_object object; + memset(&object, 0, sizeof(object)); + object.handle.store_fd = random + 7; + object.handle.mmap_size = random + 42; + object.data_offset = random + 1; + object.metadata_offset = random + 2; + object.data_size = random + 3; + object.metadata_size = random + 4; + return object; +} + +TEST plasma_create_request_test(void) { + int fd = create_temp_file(); + object_id object_id1 = globally_unique_id(); + int64_t data_size1 = 42; + int64_t metadata_size1 = 11; + plasma_send_CreateRequest(fd, g_B, object_id1, data_size1, metadata_size1); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaCreateRequest); + object_id object_id2; + int64_t data_size2; + int64_t metadata_size2; + plasma_read_CreateRequest(data, &object_id2, &data_size2, &metadata_size2); + ASSERT_EQ(data_size1, data_size2); + ASSERT_EQ(metadata_size1, metadata_size2); + ASSERT(object_ids_equal(object_id1, object_id2)); + free(data); + close(fd); + PASS(); +} + +TEST plasma_create_reply_test(void) { + int fd = create_temp_file(); + object_id object_id1 = globally_unique_id(); + plasma_object object1 = random_plasma_object(); + plasma_send_CreateReply(fd, g_B, object_id1, &object1, 0); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaCreateReply); + object_id object_id2; + plasma_object object2; + memset(&object2, 0, sizeof(object2)); + int error_code; + plasma_read_CreateReply(data, &object_id2, &object2, &error_code); + ASSERT(object_ids_equal(object_id1, object_id2)); + ASSERT(memcmp(&object1, &object2, sizeof(object1)) == 0); + free(data); + close(fd); + PASS(); +} + +TEST plasma_seal_request_test(void) { + int fd = create_temp_file(); + object_id object_id1 = globally_unique_id(); + unsigned char digest1[DIGEST_SIZE]; + memset(&digest1[0], 7, DIGEST_SIZE); + plasma_send_SealRequest(fd, g_B, object_id1, &digest1[0]); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaSealRequest); + object_id object_id2; + unsigned char digest2[DIGEST_SIZE]; + plasma_read_SealRequest(data, &object_id2, &digest2[0]); + ASSERT(object_ids_equal(object_id1, object_id2)); + ASSERT(memcmp(&digest1[0], &digest2[0], DIGEST_SIZE) == 0); + free(data); + close(fd); + PASS(); +} + +TEST plasma_seal_reply_test(void) { + int fd = create_temp_file(); + object_id object_id1 = globally_unique_id(); + int error1 = 5; + plasma_send_SealReply(fd, g_B, object_id1, error1); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaSealReply); + object_id object_id2; + int error2; + plasma_read_SealReply(data, &object_id2, &error2); + ASSERT(object_ids_equal(object_id1, object_id2)); + ASSERT(error1 == error2); + free(data); + close(fd); + PASS(); +} + +TEST plasma_get_request_test(void) { + int fd = create_temp_file(); + object_id object_ids[2]; + object_ids[0] = globally_unique_id(); + object_ids[1] = globally_unique_id(); + plasma_send_GetRequest(fd, g_B, object_ids, 2); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaGetRequest); + int64_t num_objects; + object_id object_ids_return[2]; + plasma_read_GetRequest(data, &object_ids_return[0], 2); + ASSERT(object_ids_equal(object_ids[0], object_ids_return[0])); + ASSERT(object_ids_equal(object_ids[1], object_ids_return[1])); + free(data); + close(fd); + PASS(); +} + +TEST plasma_get_reply_test(void) { + int fd = create_temp_file(); + object_id object_ids[2]; + object_ids[0] = globally_unique_id(); + object_ids[1] = globally_unique_id(); + plasma_object plasma_objects[2]; + plasma_objects[0] = random_plasma_object(); + plasma_objects[1] = random_plasma_object(); + plasma_send_GetReply(fd, g_B, object_ids, plasma_objects, 2); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaGetReply); + int64_t num_objects = plasma_read_GetLocalRequest_num_objects(data); + object_id object_ids_return[num_objects]; + plasma_object plasma_objects_return[2]; + plasma_read_GetReply(data, object_ids_return, &plasma_objects_return[0], + num_objects); + ASSERT(object_ids_equal(object_ids[0], object_ids_return[0])); + ASSERT(object_ids_equal(object_ids[1], object_ids_return[1])); + ASSERT(memcmp(&plasma_objects[0], &plasma_objects_return[0], + sizeof(plasma_object)) == 0); + ASSERT(memcmp(&plasma_objects[1], &plasma_objects_return[1], + sizeof(plasma_object)) == 0); + free(data); + close(fd); + PASS(); +} + +TEST plasma_get_local_request_test(void) { + int fd = create_temp_file(); + object_id object_ids[2]; + object_ids[0] = globally_unique_id(); + object_ids[1] = globally_unique_id(); + plasma_send_GetLocalRequest(fd, g_B, object_ids, 2); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaGetLocalRequest); + int64_t num_objects = plasma_read_GetLocalRequest_num_objects(data); + object_id object_ids_return[num_objects]; + plasma_read_GetLocalRequest(data, object_ids_return, num_objects); + ASSERT(object_ids_equal(object_ids[0], object_ids_return[0])); + ASSERT(object_ids_equal(object_ids[1], object_ids_return[1])); + free(data); + close(fd); + PASS(); +} + +TEST plasma_get_local_reply_test(void) { + int fd = create_temp_file(); + object_id object_ids[2]; + object_ids[0] = globally_unique_id(); + object_ids[1] = globally_unique_id(); + plasma_object plasma_objects[2]; + plasma_objects[0] = random_plasma_object(); + plasma_objects[1] = random_plasma_object(); + int has_object[2] = {1, 0}; + plasma_send_GetLocalReply(fd, g_B, object_ids, plasma_objects, has_object, 2); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaGetLocalReply); + int64_t num_objects; + object_id object_ids_return[2]; + plasma_object plasma_objects_return[2]; + int has_object_return[2]; + plasma_read_GetLocalReply(data, &object_ids_return[0], + &plasma_objects_return[0], has_object_return, 2); + ASSERT(object_ids_equal(object_ids[0], object_ids_return[0])); + ASSERT(object_ids_equal(object_ids[1], object_ids_return[1])); + ASSERT(memcmp(&plasma_objects[0], &plasma_objects_return[0], + sizeof(plasma_object)) == 0); + ASSERT(memcmp(&plasma_objects[1], &plasma_objects_return[1], + sizeof(plasma_object)) == 0); + ASSERT(has_object_return[0] == has_object[0]); + ASSERT(has_object_return[1] == has_object[1]); + free(data); + close(fd); + PASS(); +} + +TEST plasma_release_request_test(void) { + int fd = create_temp_file(); + object_id object_id1 = globally_unique_id(); + plasma_send_ReleaseRequest(fd, g_B, object_id1); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaReleaseRequest); + object_id object_id2; + plasma_read_ReleaseRequest(data, &object_id2); + ASSERT(object_ids_equal(object_id1, object_id2)); + free(data); + close(fd); + PASS(); +} + +TEST plasma_release_reply_test(void) { + int fd = create_temp_file(); + object_id object_id1 = globally_unique_id(); + int error1 = 5; + plasma_send_ReleaseReply(fd, g_B, object_id1, error1); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaReleaseReply); + object_id object_id2; + int error2; + plasma_read_ReleaseReply(data, &object_id2, &error2); + ASSERT(object_ids_equal(object_id1, object_id2)); + ASSERT(error1 == error2); + free(data); + close(fd); + PASS(); +} + +TEST plasma_delete_request_test(void) { + int fd = create_temp_file(); + object_id object_id1 = globally_unique_id(); + plasma_send_DeleteRequest(fd, g_B, object_id1); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaDeleteRequest); + object_id object_id2; + plasma_read_DeleteRequest(data, &object_id2); + ASSERT(object_ids_equal(object_id1, object_id2)); + free(data); + close(fd); + PASS(); +} + +TEST plasma_delete_reply_test(void) { + int fd = create_temp_file(); + object_id object_id1 = globally_unique_id(); + int error1 = 5; + plasma_send_DeleteReply(fd, g_B, object_id1, error1); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaDeleteReply); + object_id object_id2; + int error2; + plasma_read_DeleteReply(data, &object_id2, &error2); + ASSERT(object_ids_equal(object_id1, object_id2)); + ASSERT(error1 == error2); + free(data); + close(fd); + PASS(); +} + +TEST plasma_status_request_test(void) { + int fd = create_temp_file(); + object_id object_ids[2]; + object_ids[0] = globally_unique_id(); + object_ids[1] = globally_unique_id(); + plasma_send_StatusRequest(fd, g_B, object_ids, 2); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaStatusRequest); + int64_t num_objects = plasma_read_StatusRequest_num_objects(data); + object_id object_ids_read[num_objects]; + plasma_read_StatusRequest(data, object_ids_read, num_objects); + ASSERT(object_ids_equal(object_ids[0], object_ids_read[0])); + ASSERT(object_ids_equal(object_ids[1], object_ids_read[1])); + free(data); + close(fd); + PASS(); +} + +TEST plasma_status_reply_test(void) { + int fd = create_temp_file(); + object_id object_ids[2]; + object_ids[0] = globally_unique_id(); + object_ids[1] = globally_unique_id(); + int object_statuses[2] = {42, 43}; + plasma_send_StatusReply(fd, g_B, object_ids, object_statuses, 2); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaStatusReply); + int64_t num_objects = plasma_read_StatusReply_num_objects(data); + object_id object_ids_read[num_objects]; + int object_statuses_read[num_objects]; + plasma_read_StatusReply(data, object_ids_read, object_statuses_read, + num_objects); + ASSERT(object_ids_equal(object_ids[0], object_ids_read[0])); + ASSERT(object_ids_equal(object_ids[1], object_ids_read[1])); + ASSERT_EQ(object_statuses[0], object_statuses_read[0]); + ASSERT_EQ(object_statuses[1], object_statuses_read[1]); + free(data); + close(fd); + PASS(); +} + +TEST plasma_evict_request_test(void) { + int fd = create_temp_file(); + int64_t num_bytes = 111; + object_id object_id1 = globally_unique_id(); + plasma_send_EvictRequest(fd, g_B, num_bytes); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaEvictRequest); + int64_t num_bytes_received; + plasma_read_EvictRequest(data, &num_bytes_received); + ASSERT_EQ(num_bytes, num_bytes_received); + free(data); + close(fd); + PASS(); +} + +TEST plasma_evict_reply_test(void) { + int fd = create_temp_file(); + int64_t num_bytes = 111; + object_id object_id1 = globally_unique_id(); + plasma_send_EvictReply(fd, g_B, num_bytes); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaEvictReply); + int64_t num_bytes_received; + plasma_read_EvictReply(data, &num_bytes_received); + ASSERT_EQ(num_bytes, num_bytes_received); + free(data); + close(fd); + PASS(); +} + +TEST plasma_fetch_request_test(void) { + int fd = create_temp_file(); + object_id object_ids[2]; + object_ids[0] = globally_unique_id(); + object_ids[1] = globally_unique_id(); + plasma_send_FetchRequest(fd, g_B, object_ids, 2); + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaFetchRequest); + object_id object_ids_read[2]; + plasma_read_FetchRequest(data, &object_ids_read[0], 2); + ASSERT(object_ids_equal(object_ids[0], object_ids_read[0])); + ASSERT(object_ids_equal(object_ids[1], object_ids_read[1])); + free(data); + close(fd); + PASS(); +} + +TEST plasma_wait_request_test(void) { + int fd = create_temp_file(); + object_request object_requests[2]; + object_requests[0].object_id = globally_unique_id(); + object_requests[0].type = PLASMA_QUERY_ANYWHERE; + object_requests[1].object_id = globally_unique_id(); + object_requests[1].type = PLASMA_QUERY_LOCAL; + int num_ready_objects = 1; + int64_t timeout_ms = 1000; + plasma_send_WaitRequest(fd, g_B, object_requests, 2, num_ready_objects, + timeout_ms); + /* Read message back. */ + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaWaitRequest); + object_request object_requests_read[2]; + int num_object_ids_read = plasma_read_WaitRequest_num_object_ids(data); + ASSERT_EQ(num_object_ids_read, 2); + int num_ready_objects_read; + int64_t timeout_ms_read; + plasma_read_WaitRequest(data, &object_requests_read[0], num_object_ids_read, + &timeout_ms_read, &num_ready_objects_read); + ASSERT(object_ids_equal(object_requests[0].object_id, + object_requests_read[0].object_id)); + ASSERT(object_ids_equal(object_requests[1].object_id, + object_requests_read[1].object_id)); + ASSERT(object_requests[0].type == object_requests_read[0].type); + ASSERT(object_requests[1].type == object_requests_read[1].type); + free(data); + close(fd); + PASS(); +} + +TEST plasma_wait_reply_test(void) { + int fd = create_temp_file(); + object_request object_replies1[2]; + object_replies1[0].object_id = globally_unique_id(); + object_replies1[0].status = ObjectStatus_Local; + object_replies1[1].object_id = globally_unique_id(); + object_replies1[1].status = ObjectStatus_Nonexistent; + int num_ready_objects1 = 2; + plasma_send_WaitReply(fd, g_B, object_replies1, num_ready_objects1); + /* Read message back. */ + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaWaitReply); + object_request object_replies2[2]; + int num_ready_objects_read2; + plasma_read_WaitReply(data, &object_replies2[0], &num_ready_objects_read2); + ASSERT(object_ids_equal(object_replies1[0].object_id, + object_replies2[0].object_id)); + ASSERT(object_ids_equal(object_replies1[1].object_id, + object_replies2[1].object_id)); + ASSERT(object_replies1[0].status == object_replies2[0].status); + ASSERT(object_replies1[1].status == object_replies2[1].status); + free(data); + close(fd); + PASS(); +} + +TEST plasma_data_request_test(void) { + int fd = create_temp_file(); + object_id object_id1 = globally_unique_id(); + const char *address1 = "address1"; + int port1 = 12345; + plasma_send_DataRequest(fd, g_B, object_id1, address1, port1); + /* Reading message back. */ + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaDataRequest); + object_id object_id2; + char *address2; + int port2; + plasma_read_DataRequest(data, &object_id2, &address2, &port2); + ASSERT(object_ids_equal(object_id1, object_id2)); + ASSERT(strcmp(address1, address2) == 0); + ASSERT(port1 == port2); + free(address2); + free(data); + close(fd); + PASS(); +} + +TEST plasma_data_reply_test(void) { + int fd = create_temp_file(); + object_id object_id1 = globally_unique_id(); + int64_t object_size1 = 146; + int64_t metadata_size1 = 198; + plasma_send_DataReply(fd, g_B, object_id1, object_size1, metadata_size1); + /* Reading message back. */ + uint8_t *data = read_message_from_file(fd, MessageType_PlasmaDataReply); + object_id object_id2; + int64_t object_size2; + int64_t metadata_size2; + plasma_read_DataReply(data, &object_id2, &object_size2, &metadata_size2); + ASSERT(object_ids_equal(object_id1, object_id2)); + ASSERT(object_size1 == object_size2); + ASSERT(metadata_size1 == metadata_size2); + free(data); + PASS(); +} + +SUITE(plasma_serialization_tests) { + RUN_TEST(plasma_create_request_test); + RUN_TEST(plasma_create_reply_test); + RUN_TEST(plasma_seal_request_test); + RUN_TEST(plasma_seal_reply_test); + RUN_TEST(plasma_get_request_test); + RUN_TEST(plasma_get_reply_test); + RUN_TEST(plasma_get_local_request_test); + RUN_TEST(plasma_get_local_reply_test); + RUN_TEST(plasma_release_request_test); + RUN_TEST(plasma_release_reply_test); + RUN_TEST(plasma_delete_request_test); + RUN_TEST(plasma_delete_reply_test); + RUN_TEST(plasma_status_request_test); + RUN_TEST(plasma_status_reply_test); + RUN_TEST(plasma_evict_request_test); + RUN_TEST(plasma_evict_reply_test); + RUN_TEST(plasma_fetch_request_test); + RUN_TEST(plasma_wait_request_test); + RUN_TEST(plasma_wait_reply_test); + RUN_TEST(plasma_data_request_test); + RUN_TEST(plasma_data_reply_test); +} + +GREATEST_MAIN_DEFS(); + +int main(int argc, char **argv) { + g_B = make_protocol_builder(); + GREATEST_MAIN_BEGIN(); + RUN_SUITE(plasma_serialization_tests); + GREATEST_MAIN_END(); + free_protocol_builder(g_B); +}