diff --git a/build.sh b/build.sh index 7214c10a3..d5ad48ea9 100755 --- a/build.sh +++ b/build.sh @@ -25,7 +25,7 @@ bash "$ROOT_DIR/src/numbuf/thirdparty/build_thirdparty.sh" # Now build everything. pushd "$ROOT_DIR/python/core" - cmake ../.. + cmake -DCMAKE_BUILD_TYPE=Release ../.. make clean make popd diff --git a/src/common/CMakeLists.txt b/src/common/CMakeLists.txt index f6a27f653..90552266f 100644 --- a/src/common/CMakeLists.txt +++ b/src/common/CMakeLists.txt @@ -34,7 +34,7 @@ target_link_libraries(common "${CMAKE_CURRENT_LIST_DIR}/thirdparty/hiredis/libhi function(define_test test_name library) add_executable(${test_name} test/${test_name}.c ${ARGN}) - add_dependencies(${test_name} hiredis flatcc) + add_dependencies(${test_name} hiredis flatbuffers_ep) target_link_libraries(${test_name} common ${FLATBUFFERS_STATIC_LIB} ${library}) target_compile_options(${test_name} PUBLIC "-DPLASMA_TEST -DLOCAL_SCHEDULER_TEST -DCOMMON_TEST -DRAY_COMMON_LOG_LEVEL=4 -DRAY_TIMEOUT=50") endfunction() diff --git a/src/common/build/.gitkeep b/src/common/build/.gitkeep deleted file mode 100644 index e69de29bb..000000000 diff --git a/src/common/cmake/Common.cmake b/src/common/cmake/Common.cmake index a85d228c2..261ff524c 100644 --- a/src/common/cmake/Common.cmake +++ b/src/common/cmake/Common.cmake @@ -1,21 +1,30 @@ # Code for compiling flatbuffers include(ExternalProject) +include(CMakeParseArguments) -set(FLATCC_PREFIX "${CMAKE_BINARY_DIR}/src/common/flatcc-prefix/src/flatcc") +set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11") -if (NOT TARGET 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_VERSION "1.3.0") + +set(FLATBUFFERS_PREFIX "${CMAKE_BINARY_DIR}/flatbuffers_ep-prefix/src/flatbuffers_ep-install") +if (NOT TARGET flatbuffers_ep) + ExternalProject_Add(flatbuffers_ep + URL "https://github.com/google/flatbuffers/archive/v${FLATBUFFERS_VERSION}.tar.gz" + CMAKE_ARGS + "-DCMAKE_CXX_FLAGS=-fPIC" + "-DCMAKE_INSTALL_PREFIX:PATH=${FLATBUFFERS_PREFIX}" + "-DFLATBUFFERS_BUILD_TESTS=OFF") endif() -set(FLATBUFFERS_INCLUDE_DIR "${FLATCC_PREFIX}/include") -set(FLATBUFFERS_STATIC_LIB "${FLATCC_PREFIX}/lib/libflatcc.a") -set(FLATBUFFERS_COMPILER "${FLATCC_PREFIX}/bin/flatcc") +set(FLATBUFFERS_INCLUDE_DIR "${FLATBUFFERS_PREFIX}/include") +set(FLATBUFFERS_STATIC_LIB "${FLATBUFFERS_PREFIX}/lib/libflatbuffers.a") +set(FLATBUFFERS_COMPILER "${FLATBUFFERS_PREFIX}/bin/flatc") -include_directories("${FLATBUFFERS_INCLUDE_DIR}") +message(STATUS "Flatbuffers include dir: ${FLATBUFFERS_INCLUDE_DIR}") +message(STATUS "Flatbuffers static library: ${FLATBUFFERS_STATIC_LIB}") +message(STATUS "Flatbuffers compiler: ${FLATBUFFERS_COMPILER}") +include_directories(SYSTEM ${FLATBUFFERS_INCLUDE_DIR}) # Custom CFLAGS diff --git a/src/global_scheduler/build/.gitkeep b/src/global_scheduler/build/.gitkeep deleted file mode 100644 index e69de29bb..000000000 diff --git a/src/numbuf/CMakeLists.txt b/src/numbuf/CMakeLists.txt index 43cbc23ab..bfef07f42 100644 --- a/src/numbuf/CMakeLists.txt +++ b/src/numbuf/CMakeLists.txt @@ -47,7 +47,6 @@ if(HAS_PLASMA) include_directories("${CMAKE_CURRENT_LIST_DIR}/../plasma") include_directories("${CMAKE_CURRENT_LIST_DIR}/../common") include_directories("${CMAKE_CURRENT_LIST_DIR}/../common/thirdparty") - include_directories("${CMAKE_CURRENT_LIST_DIR}/../common/build/flatcc-prefix/src/flatcc/include") set(COMMON_EXTENSION ../common/lib/python/common_extension.c) endif() diff --git a/src/numbuf/python/src/pynumbuf/numbuf.cc b/src/numbuf/python/src/pynumbuf/numbuf.cc index d610e6ba3..63b052b3d 100644 --- a/src/numbuf/python/src/pynumbuf/numbuf.cc +++ b/src/numbuf/python/src/pynumbuf/numbuf.cc @@ -16,8 +16,8 @@ #ifdef HAS_PLASMA extern "C" { -#include "format/plasma_reader.h" #include "plasma_client.h" +#include "plasma_protocol.h" } PyObject* NumbufPlasmaOutOfMemoryError; diff --git a/src/plasma/CMakeLists.txt b/src/plasma/CMakeLists.txt index 11fe1fc73..080fb5524 100644 --- a/src/plasma/CMakeLists.txt +++ b/src/plasma/CMakeLists.txt @@ -5,6 +5,8 @@ project(plasma) # Recursively include common include(${CMAKE_CURRENT_LIST_DIR}/../common/cmake/Common.cmake) +set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11") + if(APPLE) SET(CMAKE_SHARED_LIBRARY_SUFFIX ".so") endif(APPLE) @@ -22,10 +24,14 @@ 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} + COMMAND ${FLATBUFFERS_COMPILER} -c -o ${OUTPUT_DIR} ${PLASMA_FBS_SRC} + DEPENDS ${PLASMA_FBS_SRC} COMMENT "Running flatc compiler on ${PLASMA_FBS_SRC}" - VERBATIM) + VERBATIM +) + +add_custom_target(protocol_fbs) +add_dependencies(protocol_fbs flatbuffers_ep) if(UNIX AND NOT APPLE) link_libraries(rt) @@ -38,7 +44,7 @@ add_library(plasma SHARED plasma.c plasma_extension.c ../common/lib/python/common_extension.c - plasma_protocol.c + plasma_protocol.cc plasma_client.c thirdparty/xxhash.c fling.c) @@ -51,14 +57,22 @@ if(APPLE) 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} ${FLATBUFFERS_STATIC_LIB}) +if(APPLE) + target_link_libraries(plasma -Wl,-force_load,${FLATBUFFERS_STATIC_LIB} ${COMMON_LIB} ${PYTHON_LIBRARIES} ${FLATBUFFERS_STATIC_LIB}) +else(APPLE) + target_link_libraries(plasma -Wl,--whole-archive ${FLATBUFFERS_STATIC_LIB} -Wl,--no-whole-archive ${COMMON_LIB} ${PYTHON_LIBRARIES} ${FLATBUFFERS_STATIC_LIB}) +endif(APPLE) + +include_directories("${FLATBUFFERS_INCLUDE_DIR}") + +set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fPIC") set_source_files_properties(thirdparty/dlmalloc.c PROPERTIES COMPILE_FLAGS -Wno-all) add_executable(plasma_store plasma_store.c plasma.c - plasma_protocol.c + plasma_protocol.cc eviction_policy.c fling.c malloc.c) @@ -70,19 +84,22 @@ target_link_libraries(plasma_store common ${FLATBUFFERS_STATIC_LIB}) add_library(plasma_lib STATIC plasma_client.c plasma.c - plasma_protocol.c + plasma_protocol.cc fling.c thirdparty/xxhash.c) target_link_libraries(plasma_lib common ${FLATBUFFERS_STATIC_LIB}) add_dependencies(plasma_lib gen_plasma_fbs) +add_dependencies(plasma protocol_fbs) + add_executable(plasma_manager plasma_manager.c) target_link_libraries(plasma_manager common plasma_lib ${FLATBUFFERS_STATIC_LIB}) add_library(plasma_client SHARED plasma_client.c) +target_link_libraries(plasma_client ${FLATBUFFERS_STATIC_LIB}) target_link_libraries(plasma_client common plasma_lib ${FLATBUFFERS_STATIC_LIB}) diff --git a/src/plasma/format/plasma.fbs b/src/plasma/format/plasma.fbs index 30abc8c07..6edda8507 100644 --- a/src/plasma/format/plasma.fbs +++ b/src/plasma/format/plasma.fbs @@ -93,7 +93,7 @@ table PlasmaSealRequest { // ID of the object to be sealed. object_id: string; // Hash of the object data. - digest: [ubyte]; + digest: string; } table PlasmaSealReply { @@ -213,7 +213,7 @@ table PlasmaFetchRequest { object_ids: [string]; } -table ObjectRequest { +table ObjectRequestSpec { // ID of the object. object_id: string; // The type of the object. This specifies whether we @@ -224,7 +224,7 @@ table ObjectRequest { table PlasmaWaitRequest { // Array of object requests whose status we are asking for. - object_requests: [ObjectRequest]; + object_requests: [ObjectRequestSpec]; // Number of objects expected to be returned, if available. num_ready_objects: int; // timeout diff --git a/src/plasma/plasma_client.c b/src/plasma/plasma_client.c index b9411bc48..8c57488f4 100644 --- a/src/plasma/plasma_client.c +++ b/src/plasma/plasma_client.c @@ -276,8 +276,6 @@ void plasma_get(PlasmaConnection *conn, /* Make a note to ourselves that the object is not present. */ object_buffers[i].data_size = -1; } else { - /* */ - PlasmaObject object_data; PlasmaObject *object; /* NOTE: If the object is still unsealed, we will deadlock, since we must * have been the one who created it. */ diff --git a/src/plasma/plasma_manager.c b/src/plasma/plasma_manager.c index 76f3819dc..8873b78f6 100644 --- a/src/plasma/plasma_manager.c +++ b/src/plasma/plasma_manager.c @@ -746,10 +746,6 @@ void process_transfer_request(event_loop *loop, } /* Allocate and append the request to the transfer queue. */ - uint8_t *data; - int64_t data_size; - uint8_t *metadata; - int64_t metadata_size; /* TODO(swang): A non-blocking plasma_get, or else we could block here * forever if we don't end up sealing this object. */ /* The corresponding call to plasma_release will happen in diff --git a/src/plasma/plasma_protocol.c b/src/plasma/plasma_protocol.c deleted file mode 100644 index 700b23b59..000000000 --- a/src/plasma/plasma_protocol.c +++ /dev/null @@ -1,659 +0,0 @@ -#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, - ObjectID 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, - ObjectID 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, - ObjectID 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, - ObjectID *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, - ObjectID object_id, - PlasmaObject *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, - ObjectID *object_id, - PlasmaObject *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)); - PlasmaObjectSpec_struct_t obj = PlasmaCreateReply_plasma_object(rep); - object->handle.store_fd = PlasmaObjectSpec_segment_index(obj); - object->handle.mmap_size = PlasmaObjectSpec_mmap_size(obj); - object->data_offset = PlasmaObjectSpec_data_offset(obj); - object->data_size = PlasmaObjectSpec_data_size(obj); - object->metadata_offset = PlasmaObjectSpec_metadata_offset(obj); - object->metadata_size = PlasmaObjectSpec_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, \ - ObjectID 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, ObjectID *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, \ - ObjectID 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, ObjectID *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, - ObjectID 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, - ObjectID *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, - ObjectID 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, - ObjectID 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, - ObjectID 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, - ObjectID 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, - ObjectID 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, ObjectID *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, - ObjectID 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, - ObjectID *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 connect message. */ - -int plasma_send_ConnectRequest(int sock, protocol_builder *B) { - PlasmaConnectRequest_start_as_root(B); - PlasmaConnectRequest_end_as_root(B); - return finalize_buffer_and_send(B, sock, MessageType_PlasmaConnectRequest); -} - -void plasma_read_ConnectRequest(uint8_t *data) { - DCHECK(data); - PlasmaConnectRequest_table_t req = PlasmaConnectRequest_as_root(data); -} - -int plasma_send_ConnectReply(int sock, - protocol_builder *B, - int64_t memory_capacity) { - PlasmaConnectReply_start_as_root(B); - PlasmaConnectReply_memory_capacity_add(B, memory_capacity); - PlasmaConnectReply_end_as_root(B); - return finalize_buffer_and_send(B, sock, MessageType_PlasmaConnectReply); -} - -void plasma_read_ConnectReply(uint8_t *data, int64_t *memory_capacity) { - DCHECK(data); - PlasmaConnectReply_table_t rep = PlasmaConnectReply_as_root(data); - *memory_capacity = PlasmaConnectReply_memory_capacity(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, - ObjectID object_ids[], - int64_t num_objects, - int64_t timeout_ms) { - PlasmaGetRequest_start_as_root(B); - PlasmaGetRequest_object_ids_add( - B, object_ids_to_flatbuffer(B, object_ids, num_objects)); - PlasmaGetRequest_timeout_ms_add(B, timeout_ms); - 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, - ObjectID object_ids[], - int64_t *timeout_ms, - 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); - *timeout_ms = PlasmaGetRequest_timeout_ms(req); -} - -int plasma_send_GetReply(int sock, - protocol_builder *B, - ObjectID object_ids[], - PlasmaObject 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); - - PlasmaObjectSpec_vec_start(B); - for (int i = 0; i < num_objects; ++i) { - PlasmaObject obj = plasma_objects[i]; - PlasmaObjectSpec_t plasma_obj; - memset(&plasma_obj, 0, sizeof(PlasmaObjectSpec_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; - PlasmaObjectSpec_vec_push(B, &plasma_obj); - } - PlasmaObjectSpec_vec_ref_t object_vec = PlasmaObjectSpec_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, - ObjectID object_ids[], - PlasmaObject 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(PlasmaObject) * num_objects); - PlasmaObjectSpec_vec_t plasma_objects_vector = - PlasmaGetReply_plasma_objects(req); - - for (int i = 0; i < num_objects; ++i) { - PlasmaObjectSpec_struct_t obj = - PlasmaObjectSpec_vec_at(plasma_objects_vector, i); - plasma_objects[i].handle.store_fd = PlasmaObjectSpec_segment_index(obj); - plasma_objects[i].handle.mmap_size = PlasmaObjectSpec_mmap_size(obj); - plasma_objects[i].data_offset = PlasmaObjectSpec_data_offset(obj); - plasma_objects[i].data_size = PlasmaObjectSpec_data_size(obj); - plasma_objects[i].metadata_offset = PlasmaObjectSpec_metadata_offset(obj); - plasma_objects[i].metadata_size = PlasmaObjectSpec_metadata_size(obj); - } -} - -/* Plasma fetch messages. */ - -int plasma_send_FetchRequest(int sock, - protocol_builder *B, - ObjectID 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, - ObjectID 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, - ObjectRequest 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, - ObjectRequest 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, - ObjectRequest 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, - ObjectRequest 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, - ObjectID 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, - ObjectID *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, - ObjectID 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, - ObjectID *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.cc b/src/plasma/plasma_protocol.cc new file mode 100644 index 000000000..fe2e5b8c1 --- /dev/null +++ b/src/plasma/plasma_protocol.cc @@ -0,0 +1,656 @@ +#include "flatbuffers/flatbuffers.h" +#include "format/plasma_generated.h" + +extern "C" { + +#include "plasma_protocol.h" +#include "io.h" +} + +/** + * Convert an object ID to a flatbuffer string. + * + * @param fbb Reference to the flatbuffer builder. + * @param object_id The object ID to be converted. + * @return The flatbuffer string contining the object ID. + */ +flatbuffers::Offset to_flat( + flatbuffers::FlatBufferBuilder &fbb, + ObjectID object_id) { + return fbb.CreateString((char *) &object_id.id[0], sizeof(object_id.id)); +} + +/** + * Convert a flatbuffer string to an object ID. + * + * @param string The flatbuffer string. + * @return The object ID. + */ +ObjectID from_flat(const flatbuffers::String *string) { + ObjectID object_id; + CHECK(string->size() == sizeof(object_id.id)); + memcpy(&object_id.id[0], string->data(), sizeof(object_id.id)); + return object_id; +} + +flatbuffers::Offset< + flatbuffers::Vector>> +to_flat(flatbuffers::FlatBufferBuilder &fbb, + ObjectID object_ids[], + int64_t num_objects) { + std::vector> results; + for (size_t i = 0; i < num_objects; i++) { + results.push_back(to_flat(fbb, object_ids[i])); + } + return fbb.CreateVector(results); +} + +extern "C" { + +#define FLATBUFFER_BUILDER_DEFAULT_SIZE 1024 + +protocol_builder *make_protocol_builder(void) { + return NULL; +} + +void free_protocol_builder(protocol_builder *builder) {} + +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; +} + +/* Create messages. */ + +int plasma_send_CreateRequest(int sock, + protocol_builder *B, + ObjectID object_id, + int64_t data_size, + int64_t metadata_size) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = CreatePlasmaCreateRequest(fbb, to_flat(fbb, object_id), + data_size, metadata_size); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaCreateRequest, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_CreateRequest(uint8_t *data, + ObjectID *object_id, + int64_t *data_size, + int64_t *metadata_size) { + CHECK(data); + auto message = flatbuffers::GetRoot(data); + *data_size = message->data_size(); + *metadata_size = message->metadata_size(); + *object_id = from_flat(message->object_id()); +} + +int plasma_send_CreateReply(int sock, + protocol_builder *B, + ObjectID object_id, + PlasmaObject *object, + int error_code) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + PlasmaObjectSpec plasma_object( + object->handle.store_fd, object->handle.mmap_size, object->data_offset, + object->data_size, object->metadata_offset, object->metadata_size); + auto message = CreatePlasmaCreateReply( + fbb, to_flat(fbb, object_id), &plasma_object, (PlasmaError) error_code); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaCreateReply, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_CreateReply(uint8_t *data, + ObjectID *object_id, + PlasmaObject *object, + int *error_code) { + CHECK(data); + auto message = flatbuffers::GetRoot(data); + *object_id = from_flat(message->object_id()); + object->handle.store_fd = message->plasma_object()->segment_index(); + object->handle.mmap_size = message->plasma_object()->mmap_size(); + object->data_offset = message->plasma_object()->data_offset(); + object->data_size = message->plasma_object()->data_size(); + object->metadata_offset = message->plasma_object()->metadata_offset(); + object->metadata_size = message->plasma_object()->metadata_size(); + *error_code = message->error(); +} + +/* Seal messages. */ + +int plasma_send_SealRequest(int sock, + protocol_builder *B, + ObjectID object_id, + unsigned char *digest) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto digest_string = fbb.CreateString((char *) digest, DIGEST_SIZE); + auto message = + CreatePlasmaSealRequest(fbb, to_flat(fbb, object_id), digest_string); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaSealRequest, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_SealRequest(uint8_t *data, + ObjectID *object_id, + unsigned char *digest) { + CHECK(data); + auto message = flatbuffers::GetRoot(data); + *object_id = from_flat(message->object_id()); + CHECK(message->digest()->size() == DIGEST_SIZE); + memcpy(digest, message->digest()->data(), DIGEST_SIZE); +} + +int plasma_send_SealReply(int sock, + protocol_builder *B, + ObjectID object_id, + int error) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = + CreatePlasmaSealReply(fbb, to_flat(fbb, object_id), (PlasmaError) error); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaSealReply, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_SealReply(uint8_t *data, ObjectID *object_id, int *error) { + CHECK(data); + auto message = flatbuffers::GetRoot(data); + *object_id = from_flat(message->object_id()); + *error = message->error(); +} + +/* Release messages. */ + +int plasma_send_ReleaseRequest(int sock, + protocol_builder *B, + ObjectID object_id) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = CreatePlasmaSealRequest(fbb, to_flat(fbb, object_id)); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaReleaseRequest, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_ReleaseRequest(uint8_t *data, ObjectID *object_id) { + CHECK(data); + auto message = flatbuffers::GetRoot(data); + *object_id = from_flat(message->object_id()); +} + +int plasma_send_ReleaseReply(int sock, + protocol_builder *B, + ObjectID object_id, + int error) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = CreatePlasmaReleaseReply(fbb, to_flat(fbb, object_id), + (PlasmaError) error); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaReleaseReply, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_ReleaseReply(uint8_t *data, ObjectID *object_id, int *error) { + CHECK(data); + auto message = flatbuffers::GetRoot(data); + *object_id = from_flat(message->object_id()); + *error = message->error(); +} + +/* Delete messages. */ + +int plasma_send_DeleteRequest(int sock, + protocol_builder *B, + ObjectID object_id) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = CreatePlasmaDeleteRequest(fbb, to_flat(fbb, object_id)); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaDeleteRequest, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_DeleteRequest(uint8_t *data, ObjectID *object_id) { + CHECK(data); + auto message = flatbuffers::GetRoot(data); + *object_id = from_flat(message->object_id()); +} + +int plasma_send_DeleteReply(int sock, + protocol_builder *B, + ObjectID object_id, + int error) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = CreatePlasmaDeleteReply(fbb, to_flat(fbb, object_id), + (PlasmaError) error); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaDeleteReply, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_DeleteReply(uint8_t *data, ObjectID *object_id, int *error) { + CHECK(data); + auto message = flatbuffers::GetRoot(data); + *object_id = from_flat(message->object_id()); + *error = message->error(); +} + +/* Satus messages. */ + +int plasma_send_StatusRequest(int sock, + protocol_builder *B, + ObjectID object_ids[], + int64_t num_objects) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = + CreatePlasmaStatusRequest(fbb, to_flat(fbb, object_ids, num_objects)); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaStatusRequest, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +int64_t plasma_read_StatusRequest_num_objects(uint8_t *data) { + DCHECK(data); + auto message = flatbuffers::GetRoot(data); + return message->object_ids()->size(); +} + +void plasma_read_StatusRequest(uint8_t *data, + ObjectID object_ids[], + int64_t num_objects) { + DCHECK(data); + auto message = flatbuffers::GetRoot(data); + for (int64_t i = 0; i < num_objects; ++i) { + object_ids[i] = from_flat(message->object_ids()->Get(i)); + } +} + +int plasma_send_StatusReply(int sock, + protocol_builder *B, + ObjectID object_ids[], + int object_status[], + int64_t num_objects) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = + CreatePlasmaStatusReply(fbb, to_flat(fbb, object_ids, num_objects), + fbb.CreateVector(object_status, num_objects)); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaStatusReply, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +int64_t plasma_read_StatusReply_num_objects(uint8_t *data) { + DCHECK(data); + auto message = flatbuffers::GetRoot(data); + return message->object_ids()->size(); +} + +void plasma_read_StatusReply(uint8_t *data, + ObjectID object_ids[], + int object_status[], + int64_t num_objects) { + DCHECK(data); + auto message = flatbuffers::GetRoot(data); + for (int64_t i = 0; i < num_objects; ++i) { + object_ids[i] = from_flat(message->object_ids()->Get(i)); + } + for (int64_t i = 0; i < num_objects; ++i) { + object_status[i] = message->status()->data()[i]; + } +} + +/* Contains messages. */ + +int plasma_send_ContainsRequest(int sock, + protocol_builder *B, + ObjectID object_id) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = CreatePlasmaContainsRequest(fbb, to_flat(fbb, object_id)); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaContainsRequest, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_ContainsRequest(uint8_t *data, ObjectID *object_id) { + CHECK(data); + auto message = flatbuffers::GetRoot(data); + *object_id = from_flat(message->object_id()); +} + +int plasma_send_ContainsReply(int sock, + protocol_builder *B, + ObjectID object_id, + int has_object) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = + CreatePlasmaContainsReply(fbb, to_flat(fbb, object_id), has_object); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaContainsReply, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_ContainsReply(uint8_t *data, + ObjectID *object_id, + int *has_object) { + CHECK(data); + auto message = flatbuffers::GetRoot(data); + *object_id = from_flat(message->object_id()); + *has_object = message->has_object(); +} + +/* Connect messages. */ + +int plasma_send_ConnectRequest(int sock, protocol_builder *B) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = CreatePlasmaConnectRequest(fbb); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaConnectRequest, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_ConnectRequest(uint8_t *data) {} + +int plasma_send_ConnectReply(int sock, + protocol_builder *B, + int64_t memory_capacity) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = CreatePlasmaConnectReply(fbb, memory_capacity); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaConnectReply, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_ConnectReply(uint8_t *data, int64_t *memory_capacity) { + DCHECK(data); + auto message = flatbuffers::GetRoot(data); + *memory_capacity = message->memory_capacity(); +} + +/* Evict messages. */ + +int plasma_send_EvictRequest(int sock, protocol_builder *B, int64_t num_bytes) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = CreatePlasmaEvictRequest(fbb, num_bytes); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaEvictRequest, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_EvictRequest(uint8_t *data, int64_t *num_bytes) { + CHECK(data); + auto message = flatbuffers::GetRoot(data); + *num_bytes = message->num_bytes(); +} + +int plasma_send_EvictReply(int sock, protocol_builder *B, int64_t num_bytes) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = CreatePlasmaEvictReply(fbb, num_bytes); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaEvictReply, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_EvictReply(uint8_t *data, int64_t *num_bytes) { + DCHECK(data); + auto message = flatbuffers::GetRoot(data); + *num_bytes = message->num_bytes(); +} + +/* Get messages. */ + +int plasma_send_GetRequest(int sock, + protocol_builder *B, + ObjectID object_ids[], + int64_t num_objects, + int64_t timeout_ms) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = CreatePlasmaGetRequest( + fbb, to_flat(fbb, object_ids, num_objects), timeout_ms); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaGetRequest, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +int64_t plasma_read_GetRequest_num_objects(uint8_t *data) { + DCHECK(data); + auto message = flatbuffers::GetRoot(data); + return message->object_ids()->size(); +} + +void plasma_read_GetRequest(uint8_t *data, + ObjectID object_ids[], + int64_t *timeout_ms, + int64_t num_objects) { + DCHECK(data); + auto message = flatbuffers::GetRoot(data); + for (int64_t i = 0; i < num_objects; ++i) { + object_ids[i] = from_flat(message->object_ids()->Get(i)); + } + *timeout_ms = message->timeout_ms(); +} + +int plasma_send_GetReply(int sock, + protocol_builder *B, + ObjectID object_ids[], + PlasmaObject plasma_objects[], + int64_t num_objects) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + std::vector objects; + + for (int i = 0; i < num_objects; ++i) { + PlasmaObject *object = &plasma_objects[i]; + objects.push_back(PlasmaObjectSpec( + object->handle.store_fd, object->handle.mmap_size, object->data_offset, + object->data_size, object->metadata_offset, object->metadata_size)); + } + auto message = CreatePlasmaGetReply( + fbb, to_flat(fbb, object_ids, num_objects), + fbb.CreateVectorOfStructs(objects.data(), num_objects)); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaGetReply, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_GetReply(uint8_t *data, + ObjectID object_ids[], + PlasmaObject plasma_objects[], + int64_t num_objects) { + CHECK(data); + auto message = flatbuffers::GetRoot(data); + for (int64_t i = 0; i < num_objects; ++i) { + object_ids[i] = from_flat(message->object_ids()->Get(i)); + } + for (int64_t i = 0; i < num_objects; ++i) { + const PlasmaObjectSpec *object = message->plasma_objects()->Get(i); + plasma_objects[i].handle.store_fd = object->segment_index(); + plasma_objects[i].handle.mmap_size = object->mmap_size(); + plasma_objects[i].data_offset = object->data_offset(); + plasma_objects[i].data_size = object->data_size(); + plasma_objects[i].metadata_offset = object->metadata_offset(); + plasma_objects[i].metadata_size = object->metadata_size(); + } +} + +/* Fetch messages. */ + +int plasma_send_FetchRequest(int sock, + protocol_builder *B, + ObjectID object_ids[], + int64_t num_objects) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = + CreatePlasmaFetchRequest(fbb, to_flat(fbb, object_ids, num_objects)); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaFetchRequest, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +int64_t plasma_read_FetchRequest_num_objects(uint8_t *data) { + DCHECK(data); + auto message = flatbuffers::GetRoot(data); + return message->object_ids()->size(); +} + +void plasma_read_FetchRequest(uint8_t *data, + ObjectID object_ids[], + int64_t num_objects) { + CHECK(data); + auto message = flatbuffers::GetRoot(data); + for (int64_t i = 0; i < num_objects; ++i) { + object_ids[i] = from_flat(message->object_ids()->Get(i)); + } +} + +/* Wait messages. */ + +int plasma_send_WaitRequest(int sock, + protocol_builder *B, + ObjectRequest object_requests[], + int num_requests, + int num_ready_objects, + int64_t timeout_ms) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + + std::vector> object_request_specs; + for (int i = 0; i < num_requests; i++) { + object_request_specs.push_back( + CreateObjectRequestSpec(fbb, to_flat(fbb, object_requests[i].object_id), + object_requests[i].type)); + } + + auto message = + CreatePlasmaWaitRequest(fbb, fbb.CreateVector(object_request_specs), + num_ready_objects, timeout_ms); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaWaitRequest, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +int plasma_read_WaitRequest_num_object_ids(uint8_t *data) { + DCHECK(data); + auto message = flatbuffers::GetRoot(data); + return message->object_requests()->size(); +} + +void plasma_read_WaitRequest(uint8_t *data, + ObjectRequest object_requests[], + int num_object_ids, + int64_t *timeout_ms, + int *num_ready_objects) { + DCHECK(data); + auto message = flatbuffers::GetRoot(data); + *num_ready_objects = message->num_ready_objects(); + *timeout_ms = message->timeout(); + + CHECK(num_object_ids == message->object_requests()->size()); + for (int i = 0; i < num_object_ids; i++) { + object_requests[i].object_id = + from_flat(message->object_requests()->Get(i)->object_id()); + object_requests[i].type = message->object_requests()->Get(i)->type(); + } +} + +int plasma_send_WaitReply(int sock, + protocol_builder *B, + ObjectRequest object_requests[], + int num_ready_objects) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + + std::vector> object_replies; + for (int i = 0; i < num_ready_objects; i++) { + object_replies.push_back( + CreateObjectReply(fbb, to_flat(fbb, object_requests[i].object_id), + object_requests[i].status)); + } + + auto message = CreatePlasmaWaitReply( + fbb, fbb.CreateVector(object_replies.data(), num_ready_objects), + num_ready_objects); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaWaitReply, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_WaitReply(uint8_t *data, + ObjectRequest object_requests[], + int *num_ready_objects) { + DCHECK(data); + + auto message = flatbuffers::GetRoot(data); + *num_ready_objects = message->num_ready_objects(); + for (int i = 0; i < *num_ready_objects; i++) { + object_requests[i].object_id = + from_flat(message->object_requests()->Get(i)->object_id()); + object_requests[i].status = message->object_requests()->Get(i)->status(); + } +} + +/* Subscribe messages. */ + +int plasma_send_SubscribeRequest(int sock, protocol_builder *B) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = CreatePlasmaSubscribeRequest(fbb); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaSubscribeRequest, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +/* Data messages. */ + +int plasma_send_DataRequest(int sock, + protocol_builder *B, + ObjectID object_id, + const char *address, + int port) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto addr = fbb.CreateString((char *) address, strlen(address)); + auto message = + CreatePlasmaDataRequest(fbb, to_flat(fbb, object_id), addr, port); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaDataRequest, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_DataRequest(uint8_t *data, + ObjectID *object_id, + char **address, + int *port) { + DCHECK(data); + auto message = flatbuffers::GetRoot(data); + DCHECK(message->object_id()->size() == sizeof(object_id->id)); + *object_id = from_flat(message->object_id()); + *address = strdup(message->address()->c_str()); + *port = message->port(); +} + +int plasma_send_DataReply(int sock, + protocol_builder *B, + ObjectID object_id, + int64_t object_size, + int64_t metadata_size) { + flatbuffers::FlatBufferBuilder fbb(FLATBUFFER_BUILDER_DEFAULT_SIZE); + auto message = CreatePlasmaDataReply(fbb, to_flat(fbb, object_id), + object_size, metadata_size); + fbb.Finish(message); + return write_message(sock, MessageType_PlasmaDataReply, fbb.GetSize(), + fbb.GetBufferPointer()); +} + +void plasma_read_DataReply(uint8_t *data, + ObjectID *object_id, + int64_t *object_size, + int64_t *metadata_size) { + DCHECK(data); + auto message = flatbuffers::GetRoot(data); + *object_id = from_flat(message->object_id()); + *object_size = (int64_t) message->object_size(); + *metadata_size = (int64_t) message->metadata_size(); +} + +} /* extern "C" */ diff --git a/src/plasma/plasma_protocol.h b/src/plasma/plasma_protocol.h index 297cba008..ee4533fa0 100644 --- a/src/plasma/plasma_protocol.h +++ b/src/plasma/plasma_protocol.h @@ -4,16 +4,55 @@ #include "common.h" #include "plasma.h" -#include "format/plasma_builder.h" -#include "format/plasma_reader.h" +/* This is temporary to facilitate the transition between flatcc and google + * flatbuffers. */ + +typedef int32_t MessageType_enum_t; +#define MessageType_PlasmaCreateRequest ((MessageType_enum_t) 1L) +#define MessageType_PlasmaCreateReply ((MessageType_enum_t) 2L) +#define MessageType_PlasmaSealRequest ((MessageType_enum_t) 3L) +#define MessageType_PlasmaSealReply ((MessageType_enum_t) 4L) +#define MessageType_PlasmaGetRequest ((MessageType_enum_t) 5L) +#define MessageType_PlasmaGetReply ((MessageType_enum_t) 6L) +#define MessageType_PlasmaReleaseRequest ((MessageType_enum_t) 7L) +#define MessageType_PlasmaReleaseReply ((MessageType_enum_t) 8L) +#define MessageType_PlasmaDeleteRequest ((MessageType_enum_t) 9L) +#define MessageType_PlasmaDeleteReply ((MessageType_enum_t) 10L) +#define MessageType_PlasmaStatusRequest ((MessageType_enum_t) 11L) +#define MessageType_PlasmaStatusReply ((MessageType_enum_t) 12L) +#define MessageType_PlasmaContainsRequest ((MessageType_enum_t) 13L) +#define MessageType_PlasmaContainsReply ((MessageType_enum_t) 14L) +#define MessageType_PlasmaConnectRequest ((MessageType_enum_t) 15L) +#define MessageType_PlasmaConnectReply ((MessageType_enum_t) 16L) +#define MessageType_PlasmaEvictRequest ((MessageType_enum_t) 17L) +#define MessageType_PlasmaEvictReply ((MessageType_enum_t) 18L) +#define MessageType_PlasmaFetchRequest ((MessageType_enum_t) 19L) +#define MessageType_PlasmaWaitRequest ((MessageType_enum_t) 20L) +#define MessageType_PlasmaWaitReply ((MessageType_enum_t) 21L) +#define MessageType_PlasmaSubscribeRequest ((MessageType_enum_t) 22L) +#define MessageType_PlasmaUnsubscribeRequest ((MessageType_enum_t) 23L) +#define MessageType_PlasmaDataRequest ((MessageType_enum_t) 24L) +#define MessageType_PlasmaDataReply ((MessageType_enum_t) 25L) + +typedef int32_t PlasmaError_enum_t; +#define PlasmaError_OK ((PlasmaError_enum_t) 0L) +#define PlasmaError_ObjectExists ((PlasmaError_enum_t) 1L) +#define PlasmaError_ObjectNonexistent ((PlasmaError_enum_t) 2L) +#define PlasmaError_OutOfMemory ((PlasmaError_enum_t) 3L) + +typedef int32_t ObjectStatus_enum_t; +#define ObjectStatus_Local ((ObjectStatus_enum_t) 1L) +#define ObjectStatus_Remote ((ObjectStatus_enum_t) 2L) +#define ObjectStatus_Nonexistent ((ObjectStatus_enum_t) 3L) +#define ObjectStatus_Transfer ((ObjectStatus_enum_t) 4L) + +typedef void protocol_builder; + +/* End temporary. */ /* 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); diff --git a/src/plasma/plasma_store.c b/src/plasma/plasma_store.c index ad3b61066..7ca6dd178 100644 --- a/src/plasma/plasma_store.c +++ b/src/plasma/plasma_store.c @@ -689,9 +689,6 @@ void process_message(event_loop *loop, int64_t num_objects; PlasmaObject objects[1]; memset(&objects[0], 0, sizeof(objects)); - int error; - - flatcc_builder_reset(state->builder); /* Process the different types of requests. */ switch (type) { diff --git a/src/plasma/test/serialization_tests.c b/src/plasma/test/serialization_tests.c index f4a2562e7..bdab9461f 100644 --- a/src/plasma/test/serialization_tests.c +++ b/src/plasma/test/serialization_tests.c @@ -136,7 +136,6 @@ TEST plasma_get_request_test(void) { int64_t timeout_ms = 1234; plasma_send_GetRequest(fd, g_B, object_ids, 2, timeout_ms); uint8_t *data = read_message_from_file(fd, MessageType_PlasmaGetRequest); - int64_t num_objects; ObjectID object_ids_return[2]; int64_t timeout_ms_return; plasma_read_GetRequest(data, &object_ids_return[0], &timeout_ms_return, 2); @@ -161,6 +160,7 @@ TEST plasma_get_reply_test(void) { int64_t num_objects = plasma_read_GetRequest_num_objects(data); ObjectID object_ids_return[num_objects]; PlasmaObject plasma_objects_return[2]; + memset(&plasma_objects_return, 0, sizeof(plasma_objects_return)); plasma_read_GetReply(data, object_ids_return, &plasma_objects_return[0], num_objects); ASSERT(ObjectID_equal(object_ids[0], object_ids_return[0])); @@ -274,7 +274,6 @@ TEST plasma_status_reply_test(void) { TEST plasma_evict_request_test(void) { int fd = create_temp_file(); int64_t num_bytes = 111; - ObjectID 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; @@ -288,7 +287,6 @@ TEST plasma_evict_request_test(void) { TEST plasma_evict_reply_test(void) { int fd = create_temp_file(); int64_t num_bytes = 111; - ObjectID 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;