wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-17 11:32:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

Adding flatbuffers and migrating flatcc to flatbuffers for plasma (#325)

Browse Source 
* adding flatbuffers and migrating flatcc to flatbuffers for plasma

* variable name changes in plasma_protocol and plasma flatbuffers schema

* quick fix

* cleanups and remove flatcc

* more cleanup

* add doc

* linting

* fix linting

* fix mac os x build

* linting

* cleanup

* c++ fix for plasma flatbuffers

* Remove flatcc from CMakeLists.txt.

* linting; trigger travis
This commit is contained in:
Alexey Tumanov
2017-02-28 18:47:40 -08:00
committed by Robert Nishihara
parent 1a997ed279
commit b91d9cba45
16 changed files with 751 additions and 701 deletions
Download Patch File Download Diff File Expand all files Collapse all files
build.sh
+1 -1
View File
@@ -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
src/common/CMakeLists.txt
+1 -1
View File
@@ -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()
src/common/build/.gitkeep
View File
src/common/cmake/Common.cmake
+19 -10
View File
@@ -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
src/global_scheduler/build/.gitkeep
View File
src/numbuf/CMakeLists.txt
-1
View File
@@ -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()
src/numbuf/python/src/pynumbuf/numbuf.cc
+1 -1
View File
@@ -16,8 +16,8 @@
#ifdef HAS_PLASMA
extern "C" {
#include "format/plasma_reader.h"
#include "plasma_client.h"
#include "plasma_protocol.h"
}
PyObject* NumbufPlasmaOutOfMemoryError;
src/plasma/CMakeLists.txt
+24 -7
View File
@@ -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})
src/plasma/format/plasma.fbs
+3 -3
View File
@@ -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
src/plasma/plasma_client.c
-2
View File
@@ -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. */
src/plasma/plasma_manager.c
-4
View File
@@ -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
src/plasma/plasma_protocol.c
-659
View File
@@ -1,659 +0,0 @@
#include <assert.h>
#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);
}
src/plasma/plasma_protocol.cc
+656
View File
@@ -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<flatbuffers::String> 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<flatbuffers::Offset<flatbuffers::String>>>
to_flat(flatbuffers::FlatBufferBuilder &fbb,
ObjectID object_ids[],
int64_t num_objects) {
std::vector<flatbuffers::Offset<flatbuffers::String>> 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<PlasmaCreateRequest>(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<PlasmaCreateReply>(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<PlasmaSealRequest>(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<PlasmaSealReply>(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<PlasmaReleaseRequest>(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<PlasmaReleaseReply>(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<PlasmaReleaseReply>(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<PlasmaDeleteReply>(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<PlasmaStatusRequest>(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<PlasmaStatusRequest>(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<PlasmaStatusReply>(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<PlasmaStatusReply>(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<PlasmaContainsRequest>(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<PlasmaContainsReply>(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<PlasmaConnectReply>(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<PlasmaEvictRequest>(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<PlasmaEvictReply>(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<PlasmaGetRequest>(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<PlasmaGetRequest>(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<PlasmaObjectSpec> 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<PlasmaGetReply>(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<PlasmaFetchRequest>(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<PlasmaFetchRequest>(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<flatbuffers::Offset<ObjectRequestSpec>> 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<PlasmaWaitRequest>(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<PlasmaWaitRequest>(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<flatbuffers::Offset<ObjectReply>> 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<PlasmaWaitReply>(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<PlasmaDataRequest>(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<PlasmaDataReply>(data);
*object_id = from_flat(message->object_id());
*object_size = (int64_t) message->object_size();
*metadata_size = (int64_t) message->metadata_size();
}
} /* extern "C" */
src/plasma/plasma_protocol.h
+45 -6
View File
@@ -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);
src/plasma/plasma_store.c
-3
View File
@@ -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) {
src/plasma/test/serialization_tests.c
+1 -3
View File
@@ -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;