From b35ce5dbf1139a1b14800e94ca0ace5c15a39ad2 Mon Sep 17 00:00:00 2001 From: Yuhong Guo Date: Thu, 26 Jul 2018 05:28:33 +0800 Subject: [PATCH] Update Arrow Package with breaking changes (#2440) * Merge the breaking change of Arrow Package. * Fix typo * Fix lint. * put forward declarations into header * fix * add protocol.h * fix linting --- src/plasma/CMakeLists.txt | 31 ++ src/plasma/format/plasma.fbs | 274 ------------------ src/plasma/plasma_manager.cc | 64 ++-- src/plasma/plasma_manager.h | 5 +- src/plasma/protocol.h | 77 +++++ src/plasma/test/client_tests.cc | 20 +- src/plasma/test/manager_tests.cc | 29 +- src/ray/object_manager/object_buffer_pool.h | 1 - src/ray/object_manager/object_manager.h | 1 - .../object_store_notification_manager.h | 1 - thirdparty/scripts/build_arrow.sh | 4 +- 11 files changed, 171 insertions(+), 336 deletions(-) delete mode 100644 src/plasma/format/plasma.fbs create mode 100644 src/plasma/protocol.h diff --git a/src/plasma/CMakeLists.txt b/src/plasma/CMakeLists.txt index 9332fbeb0..6f20b1f7a 100644 --- a/src/plasma/CMakeLists.txt +++ b/src/plasma/CMakeLists.txt @@ -15,15 +15,46 @@ endif() include_directories("${ARROW_DIR}/cpp/src/") # include_directories("${CMAKE_CURRENT_LIST_DIR}/../") +set(PLASMA_FBS_SRC "${CMAKE_CURRENT_LIST_DIR}/format/plasma.fbs" "${CMAKE_CURRENT_LIST_DIR}/format/common.fbs") +set(OUTPUT_DIR ${CMAKE_CURRENT_LIST_DIR}/format/) + +set(PLASMA_FBS_OUTPUT_FILES + "${OUTPUT_DIR}/plasma_generated.h" + "${OUTPUT_DIR}/common_generated.h") + +add_custom_target(gen_plasma_fbs DEPENDS ${PLASMA_FBS_OUTPUT_FILES}) +add_dependencies(gen_plasma_fbs flatbuffers_ep) + +# Copy the fbs files from Arrow project to local directory. +add_custom_command( + OUTPUT ${PLASMA_FBS_SRC} + COMMAND cp -rf ${CMAKE_CURRENT_LIST_DIR}/../../thirdparty/build/arrow/cpp/src/plasma/format/ ${CMAKE_CURRENT_LIST_DIR}/format/ + COMMENT "Copying ${PLASMA_FBS_SRC} to local" + VERBATIM) + +# Compile flatbuffers +add_custom_command( + OUTPUT ${PLASMA_FBS_OUTPUT_FILES} + # The --gen-object-api flag generates a C++ class MessageT for each + # flatbuffers message Message, which can be used to store deserialized + # messages in data structures. This is currently used for ObjectInfo for + # example. + COMMAND ${FLATBUFFERS_COMPILER} -c -o ${OUTPUT_DIR} ${PLASMA_FBS_SRC} --gen-object-api --scoped-enums + DEPENDS ${PLASMA_FBS_SRC} + COMMENT "Running flatc compiler on ${PLASMA_FBS_SRC}" + VERBATIM) + include_directories("${FLATBUFFERS_INCLUDE_DIR}") set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fPIC") add_executable(plasma_manager plasma_manager.cc) +add_dependencies(plasma_manager gen_plasma_fbs) target_link_libraries(plasma_manager common ${PLASMA_STATIC_LIB} ray_static ${ARROW_STATIC_LIB} -lpthread ${Boost_SYSTEM_LIBRARY}) define_test(client_tests "") define_test(manager_tests "" plasma_manager.cc) target_link_libraries(manager_tests ${Boost_SYSTEM_LIBRARY}) +add_dependencies(manager_tests gen_plasma_fbs) diff --git a/src/plasma/format/plasma.fbs b/src/plasma/format/plasma.fbs deleted file mode 100644 index 19324089c..000000000 --- a/src/plasma/format/plasma.fbs +++ /dev/null @@ -1,274 +0,0 @@ -// Plasma protocol specification - -enum MessageType:int { - // Create a new object. - PlasmaCreateRequest = 1, - PlasmaCreateReply, - // Seal an object. - PlasmaSealRequest, - PlasmaSealReply, - // Get an object that is stored on the local Plasma store. - PlasmaGetRequest, - PlasmaGetReply, - // Release an object. - PlasmaReleaseRequest, - PlasmaReleaseReply, - // Delete an object. - PlasmaDeleteRequest, - PlasmaDeleteReply, - // Get status of an object. - PlasmaStatusRequest, - PlasmaStatusReply, - // See if the store contains an object (will be deprecated). - PlasmaContainsRequest, - PlasmaContainsReply, - // Get information for a newly connecting client. - PlasmaConnectRequest, - PlasmaConnectReply, - // Make room for new objects in the plasma store. - PlasmaEvictRequest, - PlasmaEvictReply, - // Fetch objects from remote Plasma stores. - PlasmaFetchRequest, - // Wait for objects to be ready either from local or remote Plasma stores. - PlasmaWaitRequest, - PlasmaWaitReply, - // Subscribe to a list of objects or to all objects. - PlasmaSubscribeRequest, - // Unsubscribe. - PlasmaUnsubscribeRequest, - // Sending and receiving data. - // PlasmaDataRequest initiates sending the data, there will be one - // such message per data transfer. - PlasmaDataRequest, - // PlasmaDataReply contains the actual data and is sent back to the - // object store that requested the data. For each transfer, multiple - // reply messages get sent. Each one contains a fixed number of bytes. - PlasmaDataReply, - // Object notifications. - PlasmaNotification -} - -enum PlasmaError:int { - // Operation was successful. - OK, - // Trying to create an object that already exists. - ObjectExists, - // Trying to access an object that doesn't exist. - ObjectNonexistent, - // Trying to create an object but there isn't enough space in the store. - OutOfMemory -} - -// Plasma store messages - -struct PlasmaObjectSpec { - // Index of the memory segment (= memory mapped file) that - // this object is allocated in. - segment_index: int; - // Size in bytes of this segment (needed to call mmap). - mmap_size: ulong; - // The offset in bytes in the memory mapped file of the data. - data_offset: ulong; - // The size in bytes of the data. - data_size: ulong; - // The offset in bytes in the memory mapped file of the metadata. - metadata_offset: ulong; - // The size in bytes of the metadata. - metadata_size: ulong; -} - -table PlasmaCreateRequest { - // ID of the object to be created. - object_id: string; - // The size of the object's data in bytes. - data_size: ulong; - // The size of the object's metadata in bytes. - metadata_size: ulong; -} - -table PlasmaCreateReply { - // ID of the object that was created. - object_id: string; - // The object that is returned with this reply. - plasma_object: PlasmaObjectSpec; - // Error that occurred for this call. - error: PlasmaError; -} - -table PlasmaSealRequest { - // ID of the object to be sealed. - object_id: string; - // Hash of the object data. - digest: string; -} - -table PlasmaSealReply { - // ID of the object that was sealed. - object_id: string; - // Error code. - error: PlasmaError; -} - -table PlasmaGetRequest { - // IDs of the objects stored at local Plasma store we are getting. - object_ids: [string]; - // The number of milliseconds before the request should timeout. - timeout_ms: long; -} - -table PlasmaGetReply { - // IDs of the objects being returned. - // This number can be smaller than the number of requested - // objects if not all requested objects are stored and sealed - // in the local Plasma store. - object_ids: [string]; - // Plasma object information, in the same order as their IDs. - plasma_objects: [PlasmaObjectSpec]; - // The number of elements in both object_ids and plasma_objects arrays must agree. -} - -table PlasmaReleaseRequest { - // ID of the object to be released. - object_id: string; -} - -table PlasmaReleaseReply { - // ID of the object that was released. - object_id: string; - // Error code. - error: PlasmaError; -} - -table PlasmaDeleteRequest { - // ID of the object to be deleted. - object_id: string; -} - -table PlasmaDeleteReply { - // ID of the object that was deleted. - object_id: string; - // Error code. - error: PlasmaError; -} - -table PlasmaStatusRequest { - // IDs of the objects stored at local Plasma store we request the status of. - object_ids: [string]; -} - -enum ObjectStatus:int { - // Object is stored in the local Plasma Store. - Local = 1, - // Object is stored on a remote Plasma store, and it is not stored on the - // local Plasma Store. - Remote, - // Object is not stored in the system. - Nonexistent, - // Object is currently transferred from a remote Plasma store the the local - // Plasma Store. - Transfer -} - -table PlasmaStatusReply { - // IDs of the objects being returned. - object_ids: [string]; - // Status of the object. - status: [ObjectStatus]; -} - -// PlasmaContains is a subset of PlasmaStatus which does not -// involve the plasma manager, only the store. We should consider -// unifying them in the future and deprecating PlasmaContains. - -table PlasmaContainsRequest { - // ID of the object we are querying. - object_id: string; -} - -table PlasmaContainsReply { - // ID of the object we are querying. - object_id: string; - // 1 if the object is in the store and 0 otherwise. - has_object: int; -} - -// PlasmaConnect is used by a plasma client the first time it connects with the -// store. This is not really necessary, but is used to get some information -// about the store such as its memory capacity. - -table PlasmaConnectRequest { -} - -table PlasmaConnectReply { - // The memory capacity of the store. - memory_capacity: long; -} - -table PlasmaEvictRequest { - // Number of bytes that shall be freed. - num_bytes: ulong; -} - -table PlasmaEvictReply { - // Number of bytes that have been freed. - num_bytes: ulong; -} - -table PlasmaFetchRequest { - // IDs of objects to be gotten. - object_ids: [string]; -} - -table ObjectRequestSpec { - // ID of the object. - object_id: string; - // The type of the object. This specifies whether we - // will be waiting for an object store in the local or - // global Plasma store. - type: int; -} - -table PlasmaWaitRequest { - // Array of object requests whose status we are asking for. - object_requests: [ObjectRequestSpec]; - // Number of objects expected to be returned, if available. - num_ready_objects: int; - // timeout - timeout: long; -} - -table ObjectReply { - // ID of the object. - object_id: string; - // The object status. This specifies where the object is stored. - status: int; -} - -table PlasmaWaitReply { - // Array of object requests being returned. - object_requests: [ObjectReply]; - // Number of objects expected to be returned, if available. - num_ready_objects: int; -} - -table PlasmaSubscribeRequest { -} - -table PlasmaDataRequest { - // ID of the object that is requested. - object_id: string; - // The host address where the data shall be sent to. - address: string; - // The port of the manager the data shall be sent to. - port: int; -} - -table PlasmaDataReply { - // ID of the object that will be sent. - object_id: string; - // Size of the object data in bytes. - object_size: ulong; - // Size of the metadata in bytes. - metadata_size: ulong; -} diff --git a/src/plasma/plasma_manager.cc b/src/plasma/plasma_manager.cc index f4eaa27f2..95c8c08aa 100644 --- a/src/plasma/plasma_manager.cc +++ b/src/plasma/plasma_manager.cc @@ -26,22 +26,24 @@ #include #include +#include "common.h" #include "common_protocol.h" +#include "event_loop.h" +#include "format/plasma_generated.h" #include "io.h" #include "net.h" -#include "event_loop.h" -#include "common.h" -#include "plasma/plasma.h" -#include "plasma/events.h" -#include "plasma/protocol.h" #include "plasma/client.h" +#include "plasma/events.h" #include "plasma_manager.h" -#include "state/db.h" -#include "state/object_table.h" -#include "state/error_table.h" -#include "state/task_table.h" -#include "state/db_client_table.h" #include "ray/gcs/client.h" +#include "state/db.h" +#include "state/db_client_table.h" +#include "state/error_table.h" +#include "state/object_table.h" +#include "state/task_table.h" + +using plasma::ObjectLocation; +using plasma::flatbuf::MessageType; int handle_sigpipe(plasma::Status s, int fd) { if (s.ok()) { @@ -101,9 +103,9 @@ void process_status_request(ClientConnection *client_conn, ObjectID object_id); * @param context Client connection. * @return Status of object_id as defined in plasma.h */ -ObjectStatus request_status(ObjectID object_id, - const std::vector &manager_vector, - void *context); +ObjectLocation request_status(ObjectID object_id, + const std::vector &manager_vector, + void *context); /** * Send requested object_id back to the Plasma Manager identified @@ -394,7 +396,7 @@ void return_from_wait(PlasmaManagerState *manager_state, void update_object_wait_requests(PlasmaManagerState *manager_state, ObjectID obj_id, plasma::ObjectRequestType type, - ObjectStatus status) { + ObjectLocation status) { auto &object_wait_requests = object_wait_requests_from_type(manager_state, type); /* Update the in-progress wait requests in the specified table. */ @@ -418,9 +420,11 @@ void update_object_wait_requests(PlasmaManagerState *manager_state, /* Check that we found the object. */ RAY_CHECK(object_request != wait_req->object_requests.end()); /* Check that the object found was not previously known to us. */ - RAY_CHECK(object_request->second.status == ObjectStatus::Nonexistent); + RAY_CHECK(object_request->second.location == + plasma::ObjectLocation::Nonexistent); /* Update the found object's status to a known status. */ - object_request->second.status = status; + object_request->second.location = + static_cast(status); /* If this wait request is done, reply to the client. */ if (wait_req->num_satisfied == wait_req->num_objects_to_wait_for) { @@ -1087,7 +1091,7 @@ void object_table_subscribe_callback(ObjectID object_id, /* Run the callback for wait requests. */ update_object_wait_requests(manager_state, object_id, plasma::ObjectRequestType::PLASMA_QUERY_ANYWHERE, - ObjectStatus::Remote); + ObjectLocation::Remote); } void process_fetch_requests(ClientConnection *client_conn, @@ -1168,7 +1172,7 @@ void process_wait_request(ClientConnection *client_conn, /* Check if this object is already present locally. If so, mark the object * as present. */ if (is_object_local(manager_state, obj_id)) { - object_request.status = ObjectStatus::Local; + object_request.location = plasma::ObjectLocation::Local; wait_req->num_satisfied += 1; continue; } @@ -1238,28 +1242,28 @@ void request_status_done(ObjectID object_id, client_conn->fd); } -ObjectStatus request_status(ObjectID object_id, - const std::vector &manager_vector, - void *context) { +ObjectLocation request_status(ObjectID object_id, + const std::vector &manager_vector, + void *context) { ClientConnection *client_conn = (ClientConnection *) context; /* Return success immediately if we already have this object. */ if (is_object_local(client_conn->manager_state, object_id)) { - return ObjectStatus::Local; + return ObjectLocation::Local; } /* Since object is not stored at the local locally, manager_vector.size() > 0 * means that the object is stored at another remote object. Otherwise, if * manager_vector.size() == 0, the object is not stored anywhere. */ - return manager_vector.size() > 0 ? ObjectStatus::Remote - : ObjectStatus::Nonexistent; + return manager_vector.size() > 0 ? ObjectLocation::Remote + : ObjectLocation::Nonexistent; } void object_table_lookup_fail_callback(ObjectID object_id, void *user_context, void *user_data) { - /* Fail for now. Later, we may want to send a ObjectStatus::Nonexistent to the - * client. */ + /* Fail for now. Later, we may want to send a ObjectLocation::Nonexistent to + * the client. */ RAY_CHECK(0); } @@ -1267,7 +1271,7 @@ void process_status_request(ClientConnection *client_conn, plasma::ObjectID object_id) { /* Return success immediately if we already have this object. */ if (is_object_local(client_conn->manager_state, object_id)) { - int status = static_cast(ObjectStatus::Local); + int status = static_cast(ObjectLocation::Local); handle_sigpipe( plasma::SendStatusReply(client_conn->fd, &object_id, &status, 1), client_conn->fd); @@ -1275,7 +1279,7 @@ void process_status_request(ClientConnection *client_conn, } if (client_conn->manager_state->db == NULL) { - auto status = static_cast(ObjectStatus::Nonexistent); + auto status = static_cast(ObjectLocation::Nonexistent); handle_sigpipe( plasma::SendStatusReply(client_conn->fd, &object_id, &status, 1), client_conn->fd); @@ -1374,10 +1378,10 @@ void process_add_object_notification(PlasmaManagerState *state, /* Update the in-progress local and remote wait requests. */ update_object_wait_requests(state, object_id, plasma::ObjectRequestType::PLASMA_QUERY_LOCAL, - ObjectStatus::Local); + ObjectLocation::Local); update_object_wait_requests(state, object_id, plasma::ObjectRequestType::PLASMA_QUERY_ANYWHERE, - ObjectStatus::Local); + ObjectLocation::Local); } void process_object_notification(event_loop *loop, diff --git a/src/plasma/plasma_manager.h b/src/plasma/plasma_manager.h index 527f19b8e..5b9b0bcc7 100644 --- a/src/plasma/plasma_manager.h +++ b/src/plasma/plasma_manager.h @@ -1,6 +1,8 @@ #ifndef PLASMA_MANAGER_H #define PLASMA_MANAGER_H +#include "protocol.h" + #ifndef RAY_NUM_RETRIES #define NUM_RETRIES 5 #else @@ -9,7 +11,6 @@ typedef struct PlasmaManagerState PlasmaManagerState; typedef struct ClientConnection ClientConnection; -enum class MessageType : int64_t; /** * Initializes the plasma manager state. This connects the manager to the local @@ -155,7 +156,7 @@ ClientConnection *ClientConnection_listen(event_loop *loop, /* Buffer for requests between plasma managers. */ typedef struct PlasmaRequestBuffer { - MessageType type; + plasma::flatbuf::MessageType type; ray::ObjectID object_id; uint8_t *data; int64_t data_size; diff --git a/src/plasma/protocol.h b/src/plasma/protocol.h new file mode 100644 index 000000000..89f1256b3 --- /dev/null +++ b/src/plasma/protocol.h @@ -0,0 +1,77 @@ +#ifndef PLASMA_PROTOCOL_H +#define PLASMA_PROTOCOL_H + +#include "./format/common_generated.h" +#include "./format/plasma_generated.h" + +namespace plasma { + +namespace flatbuf { +enum class MessageType : int64_t; +}; + +using arrow::Status; + +typedef std::unordered_map ObjectRequestMap; + +Status PlasmaReceive(int sock, + flatbuf::MessageType message_type, + std::vector *buffer); + +Status SendWaitReply(int sock, + const ObjectRequestMap &object_requests, + int num_ready_objects); + +Status SendStatusReply(int sock, + ObjectID object_ids[], + int object_status[], + int64_t num_objects); + +Status SendDataRequest(int sock, + ObjectID object_id, + const char *address, + int port); + +Status SendDataReply(int sock, + ObjectID object_id, + int64_t object_size, + int64_t metadata_size); + +Status ReadDataRequest(uint8_t *data, + size_t size, + ObjectID *object_id, + char **address, + int *port); + +Status ReadDataReply(uint8_t *data, + size_t size, + ObjectID *object_id, + int64_t *object_size, + int64_t *metadata_size); + +Status ReadFetchRequest(uint8_t *data, + size_t size, + std::vector &object_ids); + +Status ReadStatusRequest(uint8_t *data, + size_t size, + ObjectID object_ids[], + int64_t num_objects); + +Status ReadWaitRequest(uint8_t *data, + size_t size, + ObjectRequestMap &object_requests, + int64_t *timeout_ms, + int *num_ready_objects); + +Status ReadStatusRequest(uint8_t *data, + size_t size, + ObjectID object_ids[], + int64_t num_objects); + +std::unique_ptr CreateObjectInfoBuffer( + flatbuf::ObjectInfoT *object_info); + +} // namespace plasma + +#endif diff --git a/src/plasma/test/client_tests.cc b/src/plasma/test/client_tests.cc index 023515a6a..5d55336f9 100644 --- a/src/plasma/test/client_tests.cc +++ b/src/plasma/test/client_tests.cc @@ -5,8 +5,6 @@ #include #include "plasma/common.h" -#include "plasma/plasma.h" -#include "plasma/protocol.h" #include "plasma/client.h" using namespace plasma; @@ -25,7 +23,7 @@ TEST plasma_status_tests(void) { /* Test for object non-existence. */ int status; ARROW_CHECK_OK(client1.Info(oid1, &status)); - ASSERT(status == static_cast(ObjectStatus::Nonexistent)); + ASSERT(status == static_cast(ObjectLocation::Nonexistent)); /* Test for the object being in local Plasma store. */ /* First create object. */ @@ -40,11 +38,11 @@ TEST plasma_status_tests(void) { */ sleep(1); ARROW_CHECK_OK(client1.Info(oid1, &status)); - ASSERT(status == static_cast(ObjectStatus::Local)); + ASSERT(status == static_cast(ObjectLocation::Local)); /* Test for object being remote. */ ARROW_CHECK_OK(client2.Info(oid1, &status)); - ASSERT(status == static_cast(ObjectStatus::Remote)); + ASSERT(status == static_cast(ObjectLocation::Remote)); ARROW_CHECK_OK(client1.Disconnect()); ARROW_CHECK_OK(client2.Disconnect()); @@ -66,7 +64,7 @@ TEST plasma_fetch_tests(void) { /* No object in the system */ ARROW_CHECK_OK(client1.Info(oid1, &status)); - ASSERT(status == static_cast(ObjectStatus::Nonexistent)); + ASSERT(status == static_cast(ObjectLocation::Nonexistent)); /* Test for the object being in local Plasma store. */ /* First create object. */ @@ -84,24 +82,24 @@ TEST plasma_fetch_tests(void) { ObjectID oid_array1[1] = {oid1}; ARROW_CHECK_OK(client1.Fetch(1, oid_array1)); ARROW_CHECK_OK(client1.Info(oid1, &status)); - ASSERT(status == static_cast(ObjectStatus::Local) || - status == static_cast(ObjectStatus::Nonexistent)); + ASSERT(status == static_cast(ObjectLocation::Local) || + status == static_cast(ObjectLocation::Nonexistent)); /* Sleep to make sure Plasma Manager got the notification. */ sleep(1); ARROW_CHECK_OK(client1.Info(oid1, &status)); - ASSERT(status == static_cast(ObjectStatus::Local)); + ASSERT(status == static_cast(ObjectLocation::Local)); /* Test for object being remote. */ ARROW_CHECK_OK(client2.Info(oid1, &status)); - ASSERT(status == static_cast(ObjectStatus::Remote)); + ASSERT(status == static_cast(ObjectLocation::Remote)); /* Sleep to make sure the object has been fetched and it is now stored in the * local Plasma Store. */ ARROW_CHECK_OK(client2.Fetch(1, oid_array1)); sleep(1); ARROW_CHECK_OK(client2.Info(oid1, &status)); - ASSERT(status == static_cast(ObjectStatus::Local)); + ASSERT(status == static_cast(ObjectLocation::Local)); sleep(1); ARROW_CHECK_OK(client1.Disconnect()); diff --git a/src/plasma/test/manager_tests.cc b/src/plasma/test/manager_tests.cc index 4bb1eacec..a2679eb8d 100644 --- a/src/plasma/test/manager_tests.cc +++ b/src/plasma/test/manager_tests.cc @@ -14,10 +14,11 @@ #include "event_loop.h" #include "io.h" -#include "plasma/plasma.h" -#include "plasma/client.h" #include "../plasma_manager.h" -#include "plasma/protocol.h" +#include "plasma/client.h" +#include "../protocol.h" + +namespace fb = plasma::flatbuf; SUITE(plasma_manager_tests); @@ -112,7 +113,7 @@ void destroy_plasma_mock(plasma_mock *mock) { * - Buffer a transfer request for the remote manager. * - Start and stop the event loop to make sure that we send the buffered * request. - * - Expect to see a MessageType::PlasmaDataRequest message on the remote + * - Expect to see a fb::MessageType::PlasmaDataRequest message on the remote * manager with the correct object ID. */ TEST request_transfer_test(void) { @@ -128,8 +129,8 @@ TEST request_transfer_test(void) { event_loop_run(local_mock->loop); int read_fd = get_client_sock(remote_mock->read_conn); std::vector request_data; - ARROW_CHECK_OK(plasma::PlasmaReceive(read_fd, MessageType::PlasmaDataRequest, - &request_data)); + ARROW_CHECK_OK(plasma::PlasmaReceive( + read_fd, fb::MessageType::PlasmaDataRequest, &request_data)); plasma::ObjectID object_id2; char *address; int port; @@ -152,8 +153,8 @@ TEST request_transfer_test(void) { * - Buffer a transfer request for the remote managers. * - Start and stop the event loop after a timeout to make sure that we * trigger the timeout on the first manager. - * - Expect to see a MessageType::PlasmaDataRequest message on the second remote - * manager with the correct object ID. + * - Expect to see a fb::MessageType::PlasmaDataRequest message on the second + * remote manager with the correct object ID. */ TEST request_transfer_retry_test(void) { plasma_mock *local_mock = init_plasma_mock(NULL); @@ -180,8 +181,8 @@ TEST request_transfer_retry_test(void) { int read_fd = get_client_sock(remote_mock2->read_conn); std::vector request_data; - ARROW_CHECK_OK(plasma::PlasmaReceive(read_fd, MessageType::PlasmaDataRequest, - &request_data)); + ARROW_CHECK_OK(plasma::PlasmaReceive( + read_fd, fb::MessageType::PlasmaDataRequest, &request_data)); plasma::ObjectID object_id2; char *address; int port; @@ -211,7 +212,7 @@ TEST read_write_object_chunk_test(void) { const int data_size = strlen(data) + 1; const int metadata_size = 0; PlasmaRequestBuffer remote_buf; - remote_buf.type = MessageType::PlasmaDataReply; + remote_buf.type = fb::MessageType::PlasmaDataReply; remote_buf.object_id = object_id; remote_buf.data = (uint8_t *) data; remote_buf.data_size = data_size; @@ -255,7 +256,7 @@ TEST object_notifications_test(void) { RAY_CHECK(fcntl(fd[1], F_SETFL, flags | O_NONBLOCK) == 0); ObjectID object_id = ObjectID::from_random(); - ObjectInfoT info; + fb::ObjectInfoT info; info.object_id = object_id.binary(); info.data_size = 10; info.metadata_size = 1; @@ -269,7 +270,7 @@ TEST object_notifications_test(void) { ASSERT(!is_local); /* Check that the object is local after receiving an object notification. */ - auto notification = plasma::create_object_info_buffer(&info); + auto notification = plasma::CreateObjectInfoBuffer(&info); int64_t size = *((int64_t *) notification.get()); send(fd[1], notification.get(), sizeof(int64_t) + size, 0); process_object_notification(local_mock->loop, fd[0], local_mock->state, 0); @@ -279,7 +280,7 @@ TEST object_notifications_test(void) { /* Check that the object is not local after receiving a notification about * the object deletion. */ info.is_deletion = true; - notification = plasma::create_object_info_buffer(&info); + notification = plasma::CreateObjectInfoBuffer(&info); size = *((int64_t *) notification.get()); send(fd[1], notification.get(), sizeof(int64_t) + size, 0); process_object_notification(local_mock->loop, fd[0], local_mock->state, 0); diff --git a/src/ray/object_manager/object_buffer_pool.h b/src/ray/object_manager/object_buffer_pool.h index 2b8fcfb67..2a1e02a39 100644 --- a/src/ray/object_manager/object_buffer_pool.h +++ b/src/ray/object_manager/object_buffer_pool.h @@ -12,7 +12,6 @@ #include "plasma/client.h" #include "plasma/events.h" -#include "plasma/plasma.h" #include "ray/id.h" #include "ray/status.h" diff --git a/src/ray/object_manager/object_manager.h b/src/ray/object_manager/object_manager.h index 8bb68fcc3..c58b6bd85 100644 --- a/src/ray/object_manager/object_manager.h +++ b/src/ray/object_manager/object_manager.h @@ -14,7 +14,6 @@ #include "plasma/client.h" #include "plasma/events.h" -#include "plasma/plasma.h" #include "ray/common/client_connection.h" #include "ray/id.h" diff --git a/src/ray/object_manager/object_store_notification_manager.h b/src/ray/object_manager/object_store_notification_manager.h index 59e0dcd6f..592486e84 100644 --- a/src/ray/object_manager/object_store_notification_manager.h +++ b/src/ray/object_manager/object_store_notification_manager.h @@ -11,7 +11,6 @@ #include "plasma/client.h" #include "plasma/events.h" -#include "plasma/plasma.h" #include "ray/id.h" #include "ray/status.h" diff --git a/thirdparty/scripts/build_arrow.sh b/thirdparty/scripts/build_arrow.sh index b70e08028..7ecaff81e 100755 --- a/thirdparty/scripts/build_arrow.sh +++ b/thirdparty/scripts/build_arrow.sh @@ -40,10 +40,10 @@ else exit 1 fi -# The PR for this commit is https://github.com/apache/arrow/pull/2104. We +# The PR for this commit is https://github.com/apache/arrow/pull/2282. We # include the link here to make it easier to find the right commit because # Arrow often rewrites git history and invalidates certain commits. -TARGET_COMMIT_ID=cecbcf7de00d2bf255bdba97cee1d37130e5bb79 +TARGET_COMMIT_ID=35ef303ad4f5a1f7a7e156e94ef331b7f9586ca5 build_arrow() { echo "building arrow"