From d7549d61842041aa8ad39cbebd1ca822771f742c Mon Sep 17 00:00:00 2001 From: "Siyuan (Ryans) Zhuang" Date: Fri, 26 Jun 2020 14:06:34 -0700 Subject: [PATCH] Replace arrow::Status with ray::Status in Plasma (#9154) * add object store status * replace arrow status with ray status * cleanup * remove plasma common.cc --- BUILD.bazel | 1 - src/ray/common/status.cc | 16 ++- src/ray/common/status.h | 67 +++++----- .../store_provider/plasma_store_provider.cc | 20 +-- .../object_store_notification_manager_ipc.cc | 8 +- src/ray/object_manager/object_buffer_pool.cc | 23 ++-- src/ray/object_manager/plasma/client.cc | 102 +++++++++------- src/ray/object_manager/plasma/client.h | 5 +- src/ray/object_manager/plasma/common.cc | 115 ------------------ src/ray/object_manager/plasma/common.h | 25 ---- src/ray/object_manager/plasma/dlmalloc.cc | 2 + src/ray/object_manager/plasma/io.cc | 20 ++- src/ray/object_manager/plasma/io.h | 6 +- ...org_apache_arrow_plasma_PlasmaClientJNI.cc | 30 +++-- src/ray/object_manager/plasma/plasma.h | 6 +- src/ray/object_manager/plasma/protocol.cc | 11 +- src/ray/object_manager/plasma/protocol.h | 4 +- src/ray/object_manager/plasma/store.cc | 30 +++-- src/ray/object_manager/plasma/store.h | 9 +- src/ray/object_manager/plasma/store_runner.cc | 4 +- .../test/object_manager_stress_test.cc | 18 ++- .../test/object_manager_test.cc | 13 +- src/ray/raylet/node_manager.cc | 7 +- .../raylet/object_manager_integration_test.cc | 17 ++- 24 files changed, 215 insertions(+), 344 deletions(-) delete mode 100644 src/ray/object_manager/plasma/common.cc diff --git a/BUILD.bazel b/BUILD.bazel index 04ba84223..f84528082 100644 --- a/BUILD.bazel +++ b/BUILD.bazel @@ -270,7 +270,6 @@ cc_library( name = "plasma_client", srcs = [ "src/ray/object_manager/plasma/client.cc", - "src/ray/object_manager/plasma/common.cc", "src/ray/object_manager/plasma/fling.cc", "src/ray/object_manager/plasma/io.cc", "src/ray/object_manager/plasma/malloc.cc", diff --git a/src/ray/common/status.cc b/src/ray/common/status.cc index cba252966..43e6cbbd3 100644 --- a/src/ray/common/status.cc +++ b/src/ray/common/status.cc @@ -41,8 +41,6 @@ namespace ray { #define STATUS_CODE_TYPE_ERROR "Type error" #define STATUS_CODE_INVALID "Invalid" #define STATUS_CODE_IO_ERROR "IOError" -#define STATUS_CODE_OBJECT_EXISTS "ObjectExists" -#define STATUS_CODE_OBJECT_STORE_FULL "ObjectStoreFull" #define STATUS_CODE_UNKNOWN_ERROR "Unknown error" #define STATUS_CODE_NOT_IMPLEMENTED "NotImplemented" #define STATUS_CODE_REDIS_ERROR "RedisError" @@ -52,6 +50,11 @@ namespace ray { #define STATUS_CODE_UNEXPECTED_SYSTEM_EXIT "UnexpectedSystemExit" #define STATUS_CODE_UNKNOWN "Unknown" #define STATUS_CODE_NOT_FOUND "NotFound" +// object store status +#define STATUS_CODE_OBJECT_EXISTS "ObjectExists" +#define STATUS_CODE_OBJECT_NOT_FOUND "ObjectNotFound" +#define STATUS_CODE_OBJECT_STORE_ALREADY_SEALED "ObjectAlreadySealed" +#define STATUS_CODE_OBJECT_STORE_FULL "ObjectStoreFull" Status::Status(StatusCode code, const std::string &msg) { assert(code != StatusCode::OK); @@ -81,8 +84,6 @@ std::string Status::CodeAsString() const { {StatusCode::TypeError, STATUS_CODE_TYPE_ERROR}, {StatusCode::Invalid, STATUS_CODE_INVALID}, {StatusCode::IOError, STATUS_CODE_IO_ERROR}, - {StatusCode::ObjectExists, STATUS_CODE_OBJECT_EXISTS}, - {StatusCode::ObjectStoreFull, STATUS_CODE_OBJECT_STORE_FULL}, {StatusCode::UnknownError, STATUS_CODE_UNKNOWN_ERROR}, {StatusCode::NotImplemented, STATUS_CODE_NOT_IMPLEMENTED}, {StatusCode::RedisError, STATUS_CODE_REDIS_ERROR}, @@ -90,7 +91,12 @@ std::string Status::CodeAsString() const { {StatusCode::Interrupted, STATUS_CODE_INTERRUPTED}, {StatusCode::IntentionalSystemExit, STATUS_CODE_INTENTIONAL_SYSTEM_EXIT}, {StatusCode::UnexpectedSystemExit, STATUS_CODE_UNEXPECTED_SYSTEM_EXIT}, - {StatusCode::NotFound, STATUS_CODE_NOT_FOUND}}; + {StatusCode::NotFound, STATUS_CODE_NOT_FOUND}, + {StatusCode::ObjectExists, STATUS_CODE_OBJECT_EXISTS}, + {StatusCode::ObjectNotFound, STATUS_CODE_OBJECT_NOT_FOUND}, + {StatusCode::ObjectAlreadySealed, STATUS_CODE_OBJECT_STORE_ALREADY_SEALED}, + {StatusCode::ObjectStoreFull, STATUS_CODE_OBJECT_STORE_FULL}, + }; if (!code_to_str.count(code())) { return STATUS_CODE_UNKNOWN; diff --git a/src/ray/common/status.h b/src/ray/common/status.h index 63a094084..885ad2bb8 100644 --- a/src/ray/common/status.h +++ b/src/ray/common/status.h @@ -55,6 +55,15 @@ class error_code; } \ } while (0) +#define RAY_RETURN_NOT_OK_ELSE(s, else_) \ + do { \ + ::ray::Status _s = (s); \ + if (!_s.ok()) { \ + else_; \ + return _s; \ + } \ + } while (0) + // If 'to_call' returns a bad status, CHECK immediately with a logged message // of 'msg' followed by the status. #define RAY_CHECK_OK_PREPEND(to_call, msg) \ @@ -67,27 +76,6 @@ class error_code; // logged message. #define RAY_CHECK_OK(s) RAY_CHECK_OK_PREPEND(s, "Bad status") -// This macro is used to replace the "ARROW_CHECK_OK_PREPEND" macro. -#define RAY_ARROW_CHECK_OK_PREPEND(to_call, msg) \ - do { \ - ::arrow::Status _s = (to_call); \ - RAY_CHECK(_s.ok()) << (msg) << ": " << _s.ToString(); \ - } while (0) - -// This macro is used to replace the "ARROW_CHECK_OK" macro. -#define RAY_ARROW_CHECK_OK(s) RAY_ARROW_CHECK_OK_PREPEND(s, "Bad status") - -// If arrow status is not ok, return a ray IOError status -// with the error message. -#define RAY_ARROW_RETURN_NOT_OK(s) \ - do { \ - ::arrow::Status _s = (s); \ - if (RAY_PREDICT_FALSE(!_s.ok())) { \ - return ray::Status::IOError(_s.message()); \ - ; \ - } \ - } while (0) - namespace ray { enum class StatusCode : char { @@ -97,8 +85,6 @@ enum class StatusCode : char { TypeError = 3, Invalid = 4, IOError = 5, - ObjectExists = 6, - ObjectStoreFull = 7, UnknownError = 9, NotImplemented = 10, RedisError = 11, @@ -107,6 +93,11 @@ enum class StatusCode : char { IntentionalSystemExit = 14, UnexpectedSystemExit = 15, NotFound = 16, + // object store status + ObjectExists = 21, + ObjectNotFound = 22, + ObjectAlreadySealed = 23, + ObjectStoreFull = 24, }; #if defined(__clang__) @@ -158,14 +149,6 @@ class RAY_EXPORT Status { return Status(StatusCode::IOError, msg); } - static Status ObjectExists(const std::string &msg) { - return Status(StatusCode::ObjectExists, msg); - } - - static Status ObjectStoreFull(const std::string &msg) { - return Status(StatusCode::ObjectStoreFull, msg); - } - static Status RedisError(const std::string &msg) { return Status(StatusCode::RedisError, msg); } @@ -190,6 +173,22 @@ class RAY_EXPORT Status { return Status(StatusCode::NotFound, msg); } + static Status ObjectExists(const std::string &msg) { + return Status(StatusCode::ObjectExists, msg); + } + + static Status ObjectNotFound(const std::string &msg) { + return Status(StatusCode::ObjectNotFound, msg); + } + + static Status ObjectAlreadySealed(const std::string &msg) { + return Status(StatusCode::ObjectAlreadySealed, msg); + } + + static Status ObjectStoreFull(const std::string &msg) { + return Status(StatusCode::ObjectStoreFull, msg); + } + // Returns true iff the status indicates success. bool ok() const { return (state_ == NULL); } @@ -197,8 +196,6 @@ class RAY_EXPORT Status { bool IsKeyError() const { return code() == StatusCode::KeyError; } bool IsInvalid() const { return code() == StatusCode::Invalid; } bool IsIOError() const { return code() == StatusCode::IOError; } - bool IsObjectExists() const { return code() == StatusCode::ObjectExists; } - bool IsObjectStoreFull() const { return code() == StatusCode::ObjectStoreFull; } bool IsTypeError() const { return code() == StatusCode::TypeError; } bool IsUnknownError() const { return code() == StatusCode::UnknownError; } bool IsNotImplemented() const { return code() == StatusCode::NotImplemented; } @@ -213,6 +210,10 @@ class RAY_EXPORT Status { return code() == StatusCode::IntentionalSystemExit; } bool IsNotFound() const { return code() == StatusCode::NotFound; } + bool IsObjectExists() const { return code() == StatusCode::ObjectExists; } + bool IsObjectNotFound() const { return code() == StatusCode::ObjectNotFound; } + bool IsObjectAlreadySealed() const { return code() == StatusCode::ObjectAlreadySealed; } + bool IsObjectStoreFull() const { return code() == StatusCode::ObjectStoreFull; } // Return a string representation of this status suitable for printing. // Returns the string "OK" for success. diff --git a/src/ray/core_worker/store_provider/plasma_store_provider.cc b/src/ray/core_worker/store_provider/plasma_store_provider.cc index 23a1cc045..fb7e4d967 100644 --- a/src/ray/core_worker/store_provider/plasma_store_provider.cc +++ b/src/ray/core_worker/store_provider/plasma_store_provider.cc @@ -37,7 +37,7 @@ CoreWorkerPlasmaStoreProvider::CoreWorkerPlasmaStoreProvider( get_current_call_site_ = []() { return ""; }; } buffer_tracker_ = std::make_shared(); - RAY_ARROW_CHECK_OK(store_client_.Connect(store_socket)); + RAY_CHECK_OK(store_client_.Connect(store_socket)); } CoreWorkerPlasmaStoreProvider::~CoreWorkerPlasmaStoreProvider() { @@ -47,7 +47,7 @@ CoreWorkerPlasmaStoreProvider::~CoreWorkerPlasmaStoreProvider() { Status CoreWorkerPlasmaStoreProvider::SetClientOptions(std::string name, int64_t limit_bytes) { std::lock_guard guard(store_client_mutex_); - RAY_ARROW_RETURN_NOT_OK(store_client_.SetClientOptions(name, limit_bytes)); + RAY_RETURN_NOT_OK(store_client_.SetClientOptions(name, limit_bytes)); return Status::OK(); } @@ -88,7 +88,7 @@ Status CoreWorkerPlasmaStoreProvider::Create(const std::shared_ptr &meta bool evict_if_full = max_retries == 0 ? true : evict_if_full_; while (should_retry) { should_retry = false; - arrow::Status plasma_status; + Status plasma_status; std::shared_ptr arrow_buffer; { std::lock_guard guard(store_client_mutex_); @@ -99,7 +99,7 @@ Status CoreWorkerPlasmaStoreProvider::Create(const std::shared_ptr &meta // Always try to evict after the first attempt. evict_if_full = true; } - if (plasma::IsPlasmaStoreFull(plasma_status)) { + if (plasma_status.IsObjectStoreFull()) { std::ostringstream message; message << "Failed to put object " << object_id << " in object store because it " << "is full. Object size is " << data_size << " bytes."; @@ -122,12 +122,12 @@ Status CoreWorkerPlasmaStoreProvider::Create(const std::shared_ptr &meta "in the cluster." << "\n---\n"; } - } else if (plasma::IsPlasmaObjectExists(plasma_status)) { + } else if (plasma_status.IsObjectExists()) { RAY_LOG(WARNING) << "Trying to put an object that already existed in plasma: " << object_id << "."; status = Status::OK(); } else { - RAY_ARROW_RETURN_NOT_OK(plasma_status); + RAY_RETURN_NOT_OK(plasma_status); *data = std::make_shared(PlasmaBuffer(arrow_buffer)); status = Status::OK(); } @@ -138,7 +138,7 @@ Status CoreWorkerPlasmaStoreProvider::Create(const std::shared_ptr &meta Status CoreWorkerPlasmaStoreProvider::Seal(const ObjectID &object_id) { { std::lock_guard guard(store_client_mutex_); - RAY_ARROW_RETURN_NOT_OK(store_client_.Seal(object_id)); + RAY_RETURN_NOT_OK(store_client_.Seal(object_id)); } return Status::OK(); } @@ -146,7 +146,7 @@ Status CoreWorkerPlasmaStoreProvider::Seal(const ObjectID &object_id) { Status CoreWorkerPlasmaStoreProvider::Release(const ObjectID &object_id) { { std::lock_guard guard(store_client_mutex_); - RAY_ARROW_RETURN_NOT_OK(store_client_.Release(object_id)); + RAY_RETURN_NOT_OK(store_client_.Release(object_id)); } return Status::OK(); } @@ -162,7 +162,7 @@ Status CoreWorkerPlasmaStoreProvider::FetchAndGetFromPlasmaStore( std::vector plasma_results; { std::lock_guard guard(store_client_mutex_); - RAY_ARROW_RETURN_NOT_OK(store_client_.Get(batch_ids, timeout_ms, &plasma_results)); + RAY_RETURN_NOT_OK(store_client_.Get(batch_ids, timeout_ms, &plasma_results)); } // Add successfully retrieved objects to the result map and remove them from @@ -311,7 +311,7 @@ Status CoreWorkerPlasmaStoreProvider::Get( Status CoreWorkerPlasmaStoreProvider::Contains(const ObjectID &object_id, bool *has_object) { std::lock_guard guard(store_client_mutex_); - RAY_ARROW_RETURN_NOT_OK(store_client_.Contains(object_id, has_object)); + RAY_RETURN_NOT_OK(store_client_.Contains(object_id, has_object)); return Status::OK(); } diff --git a/src/ray/object_manager/notification/object_store_notification_manager_ipc.cc b/src/ray/object_manager/notification/object_store_notification_manager_ipc.cc index 5419ee289..dbd96ce9f 100644 --- a/src/ray/object_manager/notification/object_store_notification_manager_ipc.cc +++ b/src/ray/object_manager/notification/object_store_notification_manager_ipc.cc @@ -36,10 +36,10 @@ ObjectStoreNotificationManagerIPC::ObjectStoreNotificationManagerIPC( length_(0), socket_(io_service), exit_on_error_(exit_on_error) { - RAY_ARROW_CHECK_OK(store_client_.Connect(store_socket_name.c_str(), "", 0, 300)); + RAY_CHECK_OK(store_client_.Connect(store_socket_name.c_str(), "", 0, 300)); int fd; - RAY_ARROW_CHECK_OK(store_client_.Subscribe(&fd)); + RAY_CHECK_OK(store_client_.Subscribe(&fd)); boost::system::error_code ec; #ifdef _WIN32 boost::asio::detail::socket_type c_socket = fh_release(fd); @@ -71,11 +71,11 @@ ObjectStoreNotificationManagerIPC::ObjectStoreNotificationManagerIPC( } ObjectStoreNotificationManagerIPC::~ObjectStoreNotificationManagerIPC() { - RAY_ARROW_CHECK_OK(store_client_.Disconnect()); + RAY_CHECK_OK(store_client_.Disconnect()); } void ObjectStoreNotificationManagerIPC::Shutdown() { - RAY_ARROW_CHECK_OK(store_client_.Disconnect()); + RAY_CHECK_OK(store_client_.Disconnect()); } void ObjectStoreNotificationManagerIPC::NotificationWait() { diff --git a/src/ray/object_manager/object_buffer_pool.cc b/src/ray/object_manager/object_buffer_pool.cc index 717cca08a..22805bbf8 100644 --- a/src/ray/object_manager/object_buffer_pool.cc +++ b/src/ray/object_manager/object_buffer_pool.cc @@ -23,7 +23,7 @@ ObjectBufferPool::ObjectBufferPool(const std::string &store_socket_name, uint64_t chunk_size) : default_chunk_size_(chunk_size) { store_socket_name_ = store_socket_name; - RAY_ARROW_CHECK_OK(store_client_.Connect(store_socket_name_.c_str(), "", 0, 300)); + RAY_CHECK_OK(store_client_.Connect(store_socket_name_.c_str(), "", 0, 300)); } ObjectBufferPool::~ObjectBufferPool() { @@ -38,7 +38,7 @@ ObjectBufferPool::~ObjectBufferPool() { } RAY_CHECK(get_buffer_state_.empty()); RAY_CHECK(create_buffer_state_.empty()); - RAY_ARROW_CHECK_OK(store_client_.Disconnect()); + RAY_CHECK_OK(store_client_.Disconnect()); } uint64_t ObjectBufferPool::GetNumChunks(uint64_t data_size) { @@ -57,7 +57,7 @@ std::pair ObjectBufferPool::Ge std::lock_guard lock(pool_mutex_); if (get_buffer_state_.count(object_id) == 0) { plasma::ObjectBuffer object_buffer; - RAY_ARROW_CHECK_OK(store_client_.Get(&object_id, 1, 0, &object_buffer)); + RAY_CHECK_OK(store_client_.Get(&object_id, 1, 0, &object_buffer)); if (object_buffer.data == nullptr) { RAY_LOG(ERROR) << "Failed to get object"; return std::pair( @@ -85,14 +85,14 @@ void ObjectBufferPool::ReleaseGetChunk(const ObjectID &object_id, uint64_t chunk GetBufferState &buffer_state = get_buffer_state_[object_id]; buffer_state.references--; if (buffer_state.references == 0) { - RAY_ARROW_CHECK_OK(store_client_.Release(object_id)); + RAY_CHECK_OK(store_client_.Release(object_id)); get_buffer_state_.erase(object_id); } } void ObjectBufferPool::AbortGet(const ObjectID &object_id) { std::lock_guard lock(pool_mutex_); - RAY_ARROW_CHECK_OK(store_client_.Release(object_id)); + RAY_CHECK_OK(store_client_.Release(object_id)); get_buffer_state_.erase(object_id); } @@ -104,8 +104,7 @@ std::pair ObjectBufferPool::Cr int64_t object_size = data_size - metadata_size; // Try to create shared buffer. std::shared_ptr data; - arrow::Status s = - store_client_.Create(object_id, object_size, NULL, metadata_size, &data); + Status s = store_client_.Create(object_id, object_size, NULL, metadata_size, &data); std::vector buffer; if (!s.ok()) { // Create failed. The object may already exist locally. If something else went @@ -165,8 +164,8 @@ void ObjectBufferPool::SealChunk(const ObjectID &object_id, const uint64_t chunk create_buffer_state_[object_id].chunk_state[chunk_index] = CreateChunkState::SEALED; create_buffer_state_[object_id].num_seals_remaining--; if (create_buffer_state_[object_id].num_seals_remaining == 0) { - RAY_ARROW_CHECK_OK(store_client_.Seal(object_id)); - RAY_ARROW_CHECK_OK(store_client_.Release(object_id)); + RAY_CHECK_OK(store_client_.Seal(object_id)); + RAY_CHECK_OK(store_client_.Release(object_id)); create_buffer_state_.erase(object_id); RAY_LOG(DEBUG) << "Have received all chunks for object " << object_id << ", last chunk index: " << chunk_index; @@ -174,8 +173,8 @@ void ObjectBufferPool::SealChunk(const ObjectID &object_id, const uint64_t chunk } void ObjectBufferPool::AbortCreate(const ObjectID &object_id) { - RAY_ARROW_CHECK_OK(store_client_.Release(object_id)); - RAY_ARROW_CHECK_OK(store_client_.Abort(object_id)); + RAY_CHECK_OK(store_client_.Release(object_id)); + RAY_CHECK_OK(store_client_.Abort(object_id)); create_buffer_state_.erase(object_id); } @@ -199,7 +198,7 @@ std::vector ObjectBufferPool::BuildChunks( void ObjectBufferPool::FreeObjects(const std::vector &object_ids) { std::lock_guard lock(pool_mutex_); - RAY_ARROW_CHECK_OK(store_client_.Delete(object_ids)); + RAY_CHECK_OK(store_client_.Delete(object_ids)); } std::string ObjectBufferPool::DebugString() const { diff --git a/src/ray/object_manager/plasma/client.cc b/src/ray/object_manager/plasma/client.cc index edbed32ae..fcf3fa3bc 100644 --- a/src/ray/object_manager/plasma/client.cc +++ b/src/ray/object_manager/plasma/client.cc @@ -56,6 +56,16 @@ #include "ray/object_manager/plasma/plasma.h" #include "ray/object_manager/plasma/protocol.h" +// This macro is used to replace the "ARROW_CHECK_OK_PREPEND" macro. +#define RAY_ARROW_CHECK_OK_PREPEND(to_call, msg) \ + do { \ + ::arrow::Status _s = (to_call); \ + RAY_CHECK(_s.ok()) << (msg) << ": " << _s.ToString(); \ + } while (0) + +// This macro is used to replace the "ARROW_CHECK_OK" macro. +#define RAY_ARROW_CHECK_OK(s) RAY_ARROW_CHECK_OK_PREPEND(s, "Bad status") + #ifdef PLASMA_CUDA #include "arrow/gpu/cuda_api.h" @@ -437,15 +447,15 @@ Status PlasmaClient::Impl::Create(const ObjectID& object_id, int64_t data_size, RAY_LOG(DEBUG) << "called plasma_create on conn " << store_conn_ << " with size " << data_size << " and metadata size " << metadata_size; - RETURN_NOT_OK(SendCreateRequest(store_conn_, object_id, evict_if_full, data_size, + RAY_RETURN_NOT_OK(SendCreateRequest(store_conn_, object_id, evict_if_full, data_size, metadata_size, device_num)); std::vector buffer; - RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaCreateReply, &buffer)); + RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaCreateReply, &buffer)); ObjectID id; PlasmaObject object; int store_fd; int64_t mmap_size; - RETURN_NOT_OK( + RAY_RETURN_NOT_OK( ReadCreateReply(buffer.data(), buffer.size(), &id, &object, &store_fd, &mmap_size)); // If the CreateReply included an error, then the store will not send a file // descriptor. @@ -479,7 +489,7 @@ Status PlasmaClient::Impl::Create(const ObjectID& object_id, int64_t data_size, if (metadata != NULL) { // Copy the metadata to the buffer. CudaBufferWriter writer(handle->ptr); - RETURN_NOT_OK(writer.WriteAt(object.data_size, metadata, metadata_size)); + RAY_RETURN_NOT_OK(writer.WriteAt(object.data_size, metadata, metadata_size)); } *data = MakeBufferFromGpuProcessHandle(handle); #else @@ -507,12 +517,12 @@ Status PlasmaClient::Impl::CreateAndSeal(const ObjectID& object_id, RAY_LOG(DEBUG) << "called CreateAndSeal on conn " << store_conn_; - RETURN_NOT_OK(SendCreateAndSealRequest(store_conn_, object_id, evict_if_full, data, + RAY_RETURN_NOT_OK(SendCreateAndSealRequest(store_conn_, object_id, evict_if_full, data, metadata)); std::vector buffer; - RETURN_NOT_OK( + RAY_RETURN_NOT_OK( PlasmaReceive(store_conn_, MessageType::PlasmaCreateAndSealReply, &buffer)); - RETURN_NOT_OK(ReadCreateAndSealReply(buffer.data(), buffer.size())); + RAY_RETURN_NOT_OK(ReadCreateAndSealReply(buffer.data(), buffer.size())); return Status::OK(); } @@ -524,12 +534,12 @@ Status PlasmaClient::Impl::CreateAndSealBatch(const std::vector& objec RAY_LOG(DEBUG) << "called CreateAndSealBatch on conn " << store_conn_; - RETURN_NOT_OK(SendCreateAndSealBatchRequest(store_conn_, object_ids, evict_if_full, + RAY_RETURN_NOT_OK(SendCreateAndSealBatchRequest(store_conn_, object_ids, evict_if_full, data, metadata)); std::vector buffer; - RETURN_NOT_OK( + RAY_RETURN_NOT_OK( PlasmaReceive(store_conn_, MessageType::PlasmaCreateAndSealBatchReply, &buffer)); - RETURN_NOT_OK(ReadCreateAndSealBatchReply(buffer.data(), buffer.size())); + RAY_RETURN_NOT_OK(ReadCreateAndSealBatchReply(buffer.data(), buffer.size())); return Status::OK(); } @@ -592,15 +602,15 @@ Status PlasmaClient::Impl::GetBuffers( // If we get here, then the objects aren't all currently in use by this // client, so we need to send a request to the plasma store. - RETURN_NOT_OK(SendGetRequest(store_conn_, &object_ids[0], num_objects, timeout_ms)); + RAY_RETURN_NOT_OK(SendGetRequest(store_conn_, &object_ids[0], num_objects, timeout_ms)); std::vector buffer; - RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaGetReply, &buffer)); + RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaGetReply, &buffer)); std::vector received_object_ids(num_objects); std::vector object_data(num_objects); PlasmaObject* object; std::vector store_fds; std::vector mmap_sizes; - RETURN_NOT_OK(ReadGetReply(buffer.data(), buffer.size(), received_object_ids.data(), + RAY_RETURN_NOT_OK(ReadGetReply(buffer.data(), buffer.size(), received_object_ids.data(), object_data.data(), num_objects, store_fds, mmap_sizes)); // We mmap all of the file descriptors here so that we can avoid look them up @@ -729,12 +739,12 @@ Status PlasmaClient::Impl::Release(const ObjectID& object_id) { // Check if the client is no longer using this object. if (object_entry->second->count == 0) { // Tell the store that the client no longer needs the object. - RETURN_NOT_OK(MarkObjectUnused(object_id)); - RETURN_NOT_OK(SendReleaseRequest(store_conn_, object_id)); + RAY_RETURN_NOT_OK(MarkObjectUnused(object_id)); + RAY_RETURN_NOT_OK(SendReleaseRequest(store_conn_, object_id)); auto iter = deletion_cache_.find(object_id); if (iter != deletion_cache_.end()) { deletion_cache_.erase(object_id); - RETURN_NOT_OK(Delete({object_id})); + RAY_RETURN_NOT_OK(Delete({object_id})); } } return Status::OK(); @@ -750,12 +760,12 @@ Status PlasmaClient::Impl::Contains(const ObjectID& object_id, bool* has_object) } else { // If we don't already have a reference to the object, check with the store // to see if we have the object. - RETURN_NOT_OK(SendContainsRequest(store_conn_, object_id)); + RAY_RETURN_NOT_OK(SendContainsRequest(store_conn_, object_id)); std::vector buffer; - RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaContainsReply, &buffer)); + RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaContainsReply, &buffer)); ObjectID object_id2; RAY_DCHECK(buffer.size() > 0); - RETURN_NOT_OK( + RAY_RETURN_NOT_OK( ReadContainsReply(buffer.data(), buffer.size(), &object_id2, has_object)); } return Status::OK(); @@ -839,21 +849,21 @@ Status PlasmaClient::Impl::Seal(const ObjectID& object_id) { auto object_entry = objects_in_use_.find(object_id); if (object_entry == objects_in_use_.end()) { - return MakePlasmaError(PlasmaErrorCode::PlasmaObjectNonexistent, - "Seal() called on an object without a reference to it"); + return Status::ObjectNotFound( + "Seal() called on an object without a reference to it"); } if (object_entry->second->is_sealed) { - return MakePlasmaError(PlasmaErrorCode::PlasmaObjectAlreadySealed, - "Seal() called on an already sealed object"); + return Status::ObjectAlreadySealed( + "Seal() called on an already sealed object"); } object_entry->second->is_sealed = true; /// Send the seal request to Plasma. - RETURN_NOT_OK(SendSealRequest(store_conn_, object_id)); + RAY_RETURN_NOT_OK(SendSealRequest(store_conn_, object_id)); std::vector buffer; - RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaSealReply, &buffer)); + RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaSealReply, &buffer)); ObjectID sealed_id; - RETURN_NOT_OK(ReadSealReply(buffer.data(), buffer.size(), &sealed_id)); + RAY_RETURN_NOT_OK(ReadSealReply(buffer.data(), buffer.size(), &sealed_id)); RAY_CHECK(sealed_id == object_id); // We call PlasmaClient::Release to decrement the number of instances of this // object @@ -890,15 +900,15 @@ Status PlasmaClient::Impl::Abort(const ObjectID& object_id) { #endif // Send the abort request. - RETURN_NOT_OK(SendAbortRequest(store_conn_, object_id)); + RAY_RETURN_NOT_OK(SendAbortRequest(store_conn_, object_id)); // Decrease the reference count to zero, then remove the object. object_entry->second->count--; - RETURN_NOT_OK(MarkObjectUnused(object_id)); + RAY_RETURN_NOT_OK(MarkObjectUnused(object_id)); std::vector buffer; ObjectID id; MessageType type; - RETURN_NOT_OK(ReadMessage(store_conn_, &type, &buffer)); + RAY_RETURN_NOT_OK(ReadMessage(store_conn_, &type, &buffer)); return ReadAbortReply(buffer.data(), buffer.size(), &id); } @@ -915,13 +925,13 @@ Status PlasmaClient::Impl::Delete(const std::vector& object_ids) { } } if (not_in_use_ids.size() > 0) { - RETURN_NOT_OK(SendDeleteRequest(store_conn_, not_in_use_ids)); + RAY_RETURN_NOT_OK(SendDeleteRequest(store_conn_, not_in_use_ids)); std::vector buffer; - RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaDeleteReply, &buffer)); + RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaDeleteReply, &buffer)); RAY_DCHECK(buffer.size() > 0); std::vector error_codes; not_in_use_ids.clear(); - RETURN_NOT_OK( + RAY_RETURN_NOT_OK( ReadDeleteReply(buffer.data(), buffer.size(), ¬_in_use_ids, &error_codes)); } return Status::OK(); @@ -931,21 +941,21 @@ Status PlasmaClient::Impl::Evict(int64_t num_bytes, int64_t& num_bytes_evicted) std::lock_guard guard(client_mutex_); // Send a request to the store to evict objects. - RETURN_NOT_OK(SendEvictRequest(store_conn_, num_bytes)); + RAY_RETURN_NOT_OK(SendEvictRequest(store_conn_, num_bytes)); // Wait for a response with the number of bytes actually evicted. std::vector buffer; MessageType type; - RETURN_NOT_OK(ReadMessage(store_conn_, &type, &buffer)); + RAY_RETURN_NOT_OK(ReadMessage(store_conn_, &type, &buffer)); return ReadEvictReply(buffer.data(), buffer.size(), num_bytes_evicted); } Status PlasmaClient::Impl::Refresh(const std::vector& object_ids) { std::lock_guard guard(client_mutex_); - RETURN_NOT_OK(SendRefreshLRURequest(store_conn_, object_ids)); + RAY_RETURN_NOT_OK(SendRefreshLRURequest(store_conn_, object_ids)); std::vector buffer; MessageType type; - RETURN_NOT_OK(ReadMessage(store_conn_, &type, &buffer)); + RAY_RETURN_NOT_OK(ReadMessage(store_conn_, &type, &buffer)); return ReadRefreshLRUReply(buffer.data(), buffer.size()); } @@ -955,10 +965,10 @@ Status PlasmaClient::Impl::Hash(const ObjectID& object_id, uint8_t* digest) { // Get the plasma object data. We pass in a timeout of 0 to indicate that // the operation should timeout immediately. std::vector object_buffers; - RETURN_NOT_OK(Get({object_id}, 0, &object_buffers)); + RAY_RETURN_NOT_OK(Get({object_id}, 0, &object_buffers)); // If the object was not retrieved, return false. if (!object_buffers[0].data) { - return MakePlasmaError(PlasmaErrorCode::PlasmaObjectNonexistent, "Object not found"); + return Status::ObjectNotFound("Object not found"); } // Compute the hash. uint64_t hash = ComputeObjectHash(object_buffers[0]); @@ -989,7 +999,7 @@ Status PlasmaClient::Impl::Subscribe(int* fd) { RAY_CHECK(fcntl(sock[1], F_SETFL, flags | O_NONBLOCK) == 0); #endif // Tell the Plasma store about the subscription. - RETURN_NOT_OK(SendSubscribeRequest(store_conn_)); + RAY_RETURN_NOT_OK(SendSubscribeRequest(store_conn_)); // Send the file descriptor that the Plasma store should use to push // notifications about sealed objects to this client. RAY_CHECK(send_fd(store_conn_, sock[1]) >= 0); @@ -1013,7 +1023,7 @@ Status PlasmaClient::Impl::GetNotification(int fd, ObjectID* object_id, std::vector object_ids; std::vector data_sizes; std::vector metadata_sizes; - RETURN_NOT_OK( + RAY_RETURN_NOT_OK( DecodeNotifications(message.get(), &object_ids, &data_sizes, &metadata_sizes)); for (size_t i = 0; i < object_ids.size(); ++i) { pending_notification_.emplace_back(object_ids[i], data_sizes[i], metadata_sizes[i]); @@ -1058,7 +1068,7 @@ Status PlasmaClient::Impl::Connect(const std::string& store_socket_name, int release_delay, int num_retries) { std::lock_guard guard(client_mutex_); - RETURN_NOT_OK(ConnectIpcSocketRetry(store_socket_name, num_retries, -1, &store_conn_)); + RAY_RETURN_NOT_OK(ConnectIpcSocketRetry(store_socket_name, num_retries, -1, &store_conn_)); if (manager_socket_name != "") { return Status::NotImplemented("plasma manager is no longer supported"); } @@ -1067,19 +1077,19 @@ Status PlasmaClient::Impl::Connect(const std::string& store_socket_name, << "is deprecated"; } // Send a ConnectRequest to the store to get its memory capacity. - RETURN_NOT_OK(SendConnectRequest(store_conn_)); + RAY_RETURN_NOT_OK(SendConnectRequest(store_conn_)); std::vector buffer; - RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaConnectReply, &buffer)); - RETURN_NOT_OK(ReadConnectReply(buffer.data(), buffer.size(), &store_capacity_)); + RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaConnectReply, &buffer)); + RAY_RETURN_NOT_OK(ReadConnectReply(buffer.data(), buffer.size(), &store_capacity_)); return Status::OK(); } Status PlasmaClient::Impl::SetClientOptions(const std::string& client_name, int64_t output_memory_quota) { std::lock_guard guard(client_mutex_); - RETURN_NOT_OK(SendSetOptionsRequest(store_conn_, client_name, output_memory_quota)); + RAY_RETURN_NOT_OK(SendSetOptionsRequest(store_conn_, client_name, output_memory_quota)); std::vector buffer; - RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaSetOptionsReply, &buffer)); + RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaSetOptionsReply, &buffer)); return ReadSetOptionsReply(buffer.data(), buffer.size()); } diff --git a/src/ray/object_manager/plasma/client.h b/src/ray/object_manager/plasma/client.h index d13a29d8e..e0b1fdbff 100644 --- a/src/ray/object_manager/plasma/client.h +++ b/src/ray/object_manager/plasma/client.h @@ -23,16 +23,17 @@ #include #include "arrow/buffer.h" -#include "arrow/status.h" +#include "ray/common/status.h" #include "ray/object_manager/plasma/common.h" #include "ray/util/visibility.h" using arrow::Buffer; -using arrow::Status; namespace plasma { +using ray::Status; + /// Object buffer data structure. struct ObjectBuffer { /// The data buffer. diff --git a/src/ray/object_manager/plasma/common.cc b/src/ray/object_manager/plasma/common.cc deleted file mode 100644 index 84602507f..000000000 --- a/src/ray/object_manager/plasma/common.cc +++ /dev/null @@ -1,115 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -#include "ray/object_manager/plasma/common.h" - -#include -#include - -#include "arrow/util/ubsan.h" - -#include "ray/object_manager/plasma/plasma_generated.h" - -namespace fb = plasma::flatbuf; - -namespace plasma { - -namespace { - -const char kErrorDetailTypeId[] = "plasma::PlasmaStatusDetail"; - -class PlasmaStatusDetail : public arrow::StatusDetail { - public: - explicit PlasmaStatusDetail(PlasmaErrorCode code) : code_(code) {} - const char* type_id() const override { return kErrorDetailTypeId; } - std::string ToString() const override { - const char* type; - switch (code()) { - case PlasmaErrorCode::PlasmaObjectExists: - type = "Plasma object exists"; - break; - case PlasmaErrorCode::PlasmaObjectNonexistent: - type = "Plasma object is nonexistent"; - break; - case PlasmaErrorCode::PlasmaStoreFull: - type = "Plasma store is full"; - break; - case PlasmaErrorCode::PlasmaObjectAlreadySealed: - type = "Plasma object is already sealed"; - break; - default: - type = "Unknown plasma error"; - break; - } - return std::string(type); - } - PlasmaErrorCode code() const { return code_; } - - private: - PlasmaErrorCode code_; -}; - -bool IsPlasmaStatus(const arrow::Status& status, PlasmaErrorCode code) { - if (status.ok()) { - return false; - } - auto* detail = status.detail().get(); - return detail != nullptr && detail->type_id() == kErrorDetailTypeId && - static_cast(detail)->code() == code; -} - -} // namespace - -using arrow::Status; - -arrow::Status MakePlasmaError(PlasmaErrorCode code, std::string message) { - arrow::StatusCode arrow_code = arrow::StatusCode::UnknownError; - switch (code) { - case PlasmaErrorCode::PlasmaObjectExists: - arrow_code = arrow::StatusCode::AlreadyExists; - break; - case PlasmaErrorCode::PlasmaObjectNonexistent: - arrow_code = arrow::StatusCode::KeyError; - break; - case PlasmaErrorCode::PlasmaStoreFull: - arrow_code = arrow::StatusCode::CapacityError; - break; - case PlasmaErrorCode::PlasmaObjectAlreadySealed: - // Maybe a stretch? - arrow_code = arrow::StatusCode::TypeError; - break; - } - return arrow::Status(arrow_code, std::move(message), - std::make_shared(code)); -} - -bool IsPlasmaObjectExists(const arrow::Status& status) { - return IsPlasmaStatus(status, PlasmaErrorCode::PlasmaObjectExists); -} -bool IsPlasmaObjectNonexistent(const arrow::Status& status) { - return IsPlasmaStatus(status, PlasmaErrorCode::PlasmaObjectNonexistent); -} -bool IsPlasmaObjectAlreadySealed(const arrow::Status& status) { - return IsPlasmaStatus(status, PlasmaErrorCode::PlasmaObjectAlreadySealed); -} -bool IsPlasmaStoreFull(const arrow::Status& status) { - return IsPlasmaStatus(status, PlasmaErrorCode::PlasmaStoreFull); -} - -const PlasmaStoreInfo* plasma_config; - -} // namespace plasma diff --git a/src/ray/object_manager/plasma/common.h b/src/ray/object_manager/plasma/common.h index 0eea660dc..591fa0e2a 100644 --- a/src/ray/object_manager/plasma/common.h +++ b/src/ray/object_manager/plasma/common.h @@ -26,7 +26,6 @@ #include "ray/common/id.h" #include "ray/object_manager/plasma/compat.h" -#include "arrow/status.h" #ifdef PLASMA_CUDA #include "arrow/gpu/cuda_api.h" #endif @@ -37,23 +36,6 @@ using ray::ObjectID; enum class ObjectLocation : int32_t { Local, Remote, Nonexistent }; -enum class PlasmaErrorCode : int8_t { - PlasmaObjectExists = 1, - PlasmaObjectNonexistent = 2, - PlasmaStoreFull = 3, - PlasmaObjectAlreadySealed = 4, -}; - -RAY_EXPORT arrow::Status MakePlasmaError(PlasmaErrorCode code, std::string message); -/// Return true iff the status indicates an already existing Plasma object. -RAY_EXPORT bool IsPlasmaObjectExists(const arrow::Status& status); -/// Return true iff the status indicates a non-existent Plasma object. -RAY_EXPORT bool IsPlasmaObjectNonexistent(const arrow::Status& status); -/// Return true iff the status indicates an already sealed Plasma object. -RAY_EXPORT bool IsPlasmaObjectAlreadySealed(const arrow::Status& status); -/// Return true iff the status indicates the Plasma store reached its capacity limit. -RAY_EXPORT bool IsPlasmaStoreFull(const arrow::Status& status); - /// Size of object hash digests. constexpr int64_t kDigestSize = sizeof(uint64_t); @@ -102,8 +84,6 @@ struct ObjectTableEntry { /// The state of the object, e.g., whether it is open or sealed. ObjectState state; - /// The digest of the object. Used to see if two objects are the same. - unsigned char digest[kDigestSize]; #ifdef PLASMA_CUDA /// IPC GPU handle to share with clients. @@ -116,9 +96,4 @@ struct ObjectTableEntry { /// Mapping from ObjectIDs to information about the object. typedef std::unordered_map> ObjectTable; -/// Globally accessible reference to plasma store configuration. -/// TODO(pcm): This can be avoided with some refactoring of existing code -/// by making it possible to pass a context object through dlmalloc. -struct PlasmaStoreInfo; -extern const PlasmaStoreInfo* plasma_config; } // namespace plasma diff --git a/src/ray/object_manager/plasma/dlmalloc.cc b/src/ray/object_manager/plasma/dlmalloc.cc index 822cac0e4..772a33c23 100644 --- a/src/ray/object_manager/plasma/dlmalloc.cc +++ b/src/ray/object_manager/plasma/dlmalloc.cc @@ -183,4 +183,6 @@ int fake_munmap(void* addr, int64_t size) { void SetMallocGranularity(int value) { change_mparam(M_GRANULARITY, value); } +const PlasmaStoreInfo* plasma_config; + } // namespace plasma diff --git a/src/ray/object_manager/plasma/io.cc b/src/ray/object_manager/plasma/io.cc index 86179e831..3d160e022 100644 --- a/src/ray/object_manager/plasma/io.cc +++ b/src/ray/object_manager/plasma/io.cc @@ -21,8 +21,6 @@ #include #include -#include "arrow/status.h" - #include "ray/object_manager/plasma/common.h" #include "ray/object_manager/plasma/plasma_generated.h" #ifndef _WIN32 @@ -30,8 +28,6 @@ #include #endif -using arrow::Status; - /// Number of times we try connecting to a socket. constexpr int64_t kNumConnectAttempts = 80; /// Time to wait between connection attempts to a socket. @@ -67,9 +63,9 @@ Status WriteBytes(int fd, uint8_t* cursor, size_t length) { Status WriteMessage(int fd, MessageType type, int64_t length, uint8_t* bytes) { int64_t version = kPlasmaProtocolVersion; - RETURN_NOT_OK(WriteBytes(fd, reinterpret_cast(&version), sizeof(version))); - RETURN_NOT_OK(WriteBytes(fd, reinterpret_cast(&type), sizeof(type))); - RETURN_NOT_OK(WriteBytes(fd, reinterpret_cast(&length), sizeof(length))); + RAY_RETURN_NOT_OK(WriteBytes(fd, reinterpret_cast(&version), sizeof(version))); + RAY_RETURN_NOT_OK(WriteBytes(fd, reinterpret_cast(&type), sizeof(type))); + RAY_RETURN_NOT_OK(WriteBytes(fd, reinterpret_cast(&length), sizeof(length))); return WriteBytes(fd, bytes, length * sizeof(char)); } @@ -98,13 +94,13 @@ Status ReadBytes(int fd, uint8_t* cursor, size_t length) { Status ReadMessage(int fd, MessageType* type, std::vector* buffer) { int64_t version; - RETURN_NOT_OK_ELSE(ReadBytes(fd, reinterpret_cast(&version), sizeof(version)), + RAY_RETURN_NOT_OK_ELSE(ReadBytes(fd, reinterpret_cast(&version), sizeof(version)), *type = MessageType::PlasmaDisconnectClient); RAY_CHECK(version == kPlasmaProtocolVersion) << "version = " << version; - RETURN_NOT_OK_ELSE(ReadBytes(fd, reinterpret_cast(type), sizeof(*type)), + RAY_RETURN_NOT_OK_ELSE(ReadBytes(fd, reinterpret_cast(type), sizeof(*type)), *type = MessageType::PlasmaDisconnectClient); int64_t length_temp; - RETURN_NOT_OK_ELSE( + RAY_RETURN_NOT_OK_ELSE( ReadBytes(fd, reinterpret_cast(&length_temp), sizeof(length_temp)), *type = MessageType::PlasmaDisconnectClient); // The length must be read as an int64_t, but it should be used as a size_t. @@ -112,7 +108,7 @@ Status ReadMessage(int fd, MessageType* type, std::vector* buffer) { if (length > buffer->size()) { buffer->resize(length); } - RETURN_NOT_OK_ELSE(ReadBytes(fd, buffer->data(), length), + RAY_RETURN_NOT_OK_ELSE(ReadBytes(fd, buffer->data(), length), *type = MessageType::PlasmaDisconnectClient); return Status::OK(); } @@ -218,7 +214,7 @@ Status ConnectIpcSocketRetry(const std::string& pathname, int num_retries, // If we could not connect to the socket, exit. if (*fd == -1) { - return Status::IOError("Could not connect to socket ", pathname); + return Status::IOError("Could not connect to socket " + pathname); } return Status::OK(); diff --git a/src/ray/object_manager/plasma/io.h b/src/ray/object_manager/plasma/io.h index 4a0844f7c..906d30bb2 100644 --- a/src/ray/object_manager/plasma/io.h +++ b/src/ray/object_manager/plasma/io.h @@ -26,12 +26,14 @@ #include #include -#include "arrow/status.h" +#include "ray/common/status.h" #include "ray/object_manager/plasma/common.h" #include "ray/object_manager/plasma/compat.h" namespace plasma { +using ray::Status; + namespace flatbuf { // Forward declaration outside the namespace, which is defined in plasma_generated.h. @@ -44,8 +46,6 @@ enum class MessageType : int64_t; // using flatbuffers. constexpr int64_t kPlasmaProtocolVersion = 0x0000000000000000; -using arrow::Status; - Status WriteBytes(int fd, uint8_t* cursor, size_t length); Status WriteMessage(int fd, flatbuf::MessageType type, int64_t length, uint8_t* bytes); diff --git a/src/ray/object_manager/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc b/src/ray/object_manager/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc index 2748fca4a..d2ddd8ed7 100644 --- a/src/ray/object_manager/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc +++ b/src/ray/object_manager/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc @@ -28,19 +28,23 @@ #include #include +#include "ray/common/status.h" #include "ray/object_manager/plasma/client.h" -constexpr jsize OBJECT_ID_SIZE = sizeof(plasma::ObjectID) / sizeof(jbyte); +using ray::ObjectID; +using ray::Status; -inline void jbyteArray_to_object_id(JNIEnv* env, jbyteArray a, plasma::ObjectID* oid) { +constexpr jsize OBJECT_ID_SIZE = sizeof(ObjectID) / sizeof(jbyte); + +inline void jbyteArray_to_object_id(JNIEnv* env, jbyteArray a, ObjectID* oid) { env->GetByteArrayRegion(a, 0, OBJECT_ID_SIZE, reinterpret_cast(oid)); } -inline void object_id_to_jbyteArray(JNIEnv* env, jbyteArray a, plasma::ObjectID* oid) { +inline void object_id_to_jbyteArray(JNIEnv* env, jbyteArray a, ObjectID* oid) { env->SetByteArrayRegion(a, 0, OBJECT_ID_SIZE, reinterpret_cast(oid)); } -inline void throw_exception_if_not_OK(JNIEnv* env, const arrow::Status& status) { +inline void throw_exception_if_not_OK(JNIEnv* env, const Status& status) { if (!status.ok()) { jclass Exception = env->FindClass("org/apache/arrow/plasma/exceptions/PlasmaClientException"); @@ -93,7 +97,7 @@ JNIEXPORT jobject JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_create( JNIEnv* env, jclass cls, jlong conn, jbyteArray object_id, jint size, jbyteArray metadata) { plasma::PlasmaClient* client = reinterpret_cast(conn); - plasma::ObjectID oid; + ObjectID oid; jbyteArray_to_object_id(env, object_id, &oid); // prepare metadata buffer @@ -109,13 +113,13 @@ JNIEXPORT jobject JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_create( std::shared_ptr data; Status s = client->Create(oid, size, md, md_size, &data); - if (plasma::IsPlasmaObjectExists(s)) { + if (s.IsObjectExists()) { jclass exceptionClass = env->FindClass("org/apache/arrow/plasma/exceptions/DuplicateObjectException"); env->ThrowNew(exceptionClass, oid.Hex().c_str()); return nullptr; } - if (plasma::IsPlasmaStoreFull(s)) { + if (s.IsObjectStoreFull()) { jclass exceptionClass = env->FindClass("org/apache/arrow/plasma/exceptions/PlasmaOutOfMemoryException"); env->ThrowNew(exceptionClass, ""); @@ -129,7 +133,7 @@ JNIEXPORT jobject JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_create( JNIEXPORT jbyteArray JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_hash( JNIEnv* env, jclass cls, jlong conn, jbyteArray object_id) { plasma::PlasmaClient* client = reinterpret_cast(conn); - plasma::ObjectID oid; + ObjectID oid; jbyteArray_to_object_id(env, object_id, &oid); unsigned char digest[plasma::kDigestSize]; @@ -148,7 +152,7 @@ JNIEXPORT jbyteArray JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_hash( JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_seal( JNIEnv* env, jclass cls, jlong conn, jbyteArray object_id) { plasma::PlasmaClient* client = reinterpret_cast(conn); - plasma::ObjectID oid; + ObjectID oid; jbyteArray_to_object_id(env, object_id, &oid); throw_exception_if_not_OK(env, client->Seal(oid)); @@ -157,7 +161,7 @@ JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_seal( JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_release( JNIEnv* env, jclass cls, jlong conn, jbyteArray object_id) { plasma::PlasmaClient* client = reinterpret_cast(conn); - plasma::ObjectID oid; + ObjectID oid; jbyteArray_to_object_id(env, object_id, &oid); throw_exception_if_not_OK(env, client->Release(oid)); @@ -166,7 +170,7 @@ JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_release( JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_delete( JNIEnv* env, jclass cls, jlong conn, jbyteArray object_id) { plasma::PlasmaClient* client = reinterpret_cast(conn); - plasma::ObjectID oid; + ObjectID oid; jbyteArray_to_object_id(env, object_id, &oid); throw_exception_if_not_OK(env, client->Delete(oid)); @@ -177,7 +181,7 @@ JNIEXPORT jobjectArray JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_get( plasma::PlasmaClient* client = reinterpret_cast(conn); jsize num_oids = env->GetArrayLength(object_ids); - std::vector oids(num_oids); + std::vector oids(num_oids); std::vector obufs(num_oids); for (int i = 0; i < num_oids; ++i) { jbyteArray_to_object_id( @@ -220,7 +224,7 @@ JNIEXPORT jobjectArray JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_get( JNIEXPORT jboolean JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_contains( JNIEnv* env, jclass cls, jlong conn, jbyteArray object_id) { plasma::PlasmaClient* client = reinterpret_cast(conn); - plasma::ObjectID oid; + ObjectID oid; jbyteArray_to_object_id(env, object_id, &oid); bool has_object; diff --git a/src/ray/object_manager/plasma/plasma.h b/src/ray/object_manager/plasma/plasma.h index c9e83a91b..c2c9f4e4b 100644 --- a/src/ray/object_manager/plasma/plasma.h +++ b/src/ray/object_manager/plasma/plasma.h @@ -34,7 +34,7 @@ #include "ray/common/status.h" #include "ray/object_manager/plasma/compat.h" -#include "arrow/status.h" +#include "ray/common/status.h" #include "ray/object_manager/format/object_manager_generated.h" #include "ray/object_manager/plasma/common.h" #include "ray/util/logging.h" @@ -45,6 +45,7 @@ using arrow::cuda::CudaIpcMemHandle; namespace plasma { +using ray::Status; using ray::object_manager::protocol::ObjectInfoT; #define HANDLE_SIGPIPE(s, fd_) \ @@ -170,4 +171,7 @@ std::unique_ptr CreateObjectInfoBuffer(ObjectInfoT* object_info); std::unique_ptr CreatePlasmaNotificationBuffer( const std::vector& object_info); +/// Globally accessible reference to plasma store configuration. +extern const PlasmaStoreInfo* plasma_config; + } // namespace plasma diff --git a/src/ray/object_manager/plasma/protocol.cc b/src/ray/object_manager/plasma/protocol.cc index 8c2872cbe..1bf9cf97a 100644 --- a/src/ray/object_manager/plasma/protocol.cc +++ b/src/ray/object_manager/plasma/protocol.cc @@ -71,7 +71,7 @@ flatbuffers::Offset> ToFlatbuffer( Status PlasmaReceive(int sock, MessageType message_type, std::vector* buffer) { MessageType type; - RETURN_NOT_OK(ReadMessage(sock, &type, buffer)); + RAY_RETURN_NOT_OK(ReadMessage(sock, &type, buffer)); RAY_CHECK(type == message_type) << "type = " << static_cast(type) << ", message_type = " << static_cast(message_type); @@ -112,14 +112,11 @@ Status PlasmaErrorStatus(fb::PlasmaError plasma_error) { case fb::PlasmaError::OK: return Status::OK(); case fb::PlasmaError::ObjectExists: - return MakePlasmaError(PlasmaErrorCode::PlasmaObjectExists, - "object already exists in the plasma store"); + return Status::ObjectExists("object already exists in the plasma store"); case fb::PlasmaError::ObjectNonexistent: - return MakePlasmaError(PlasmaErrorCode::PlasmaObjectNonexistent, - "object does not exist in the plasma store"); + return Status::ObjectNotFound("object does not exist in the plasma store"); case fb::PlasmaError::OutOfMemory: - return MakePlasmaError(PlasmaErrorCode::PlasmaStoreFull, - "object does not fit in the plasma store"); + return Status::ObjectStoreFull("object does not fit in the plasma store"); default: RAY_LOG(FATAL) << "unknown plasma error code " << static_cast(plasma_error); } diff --git a/src/ray/object_manager/plasma/protocol.h b/src/ray/object_manager/plasma/protocol.h index fd010144f..a2924f88f 100644 --- a/src/ray/object_manager/plasma/protocol.h +++ b/src/ray/object_manager/plasma/protocol.h @@ -22,13 +22,13 @@ #include #include -#include "arrow/status.h" +#include "ray/common/status.h" #include "ray/object_manager/plasma/plasma.h" #include "ray/object_manager/plasma/plasma_generated.h" namespace plasma { -using arrow::Status; +using ray::Status; using flatbuf::MessageType; using flatbuf::PlasmaError; diff --git a/src/ray/object_manager/plasma/store.cc b/src/ray/object_manager/plasma/store.cc index 87827245b..6eb384db7 100644 --- a/src/ray/object_manager/plasma/store.cc +++ b/src/ray/object_manager/plasma/store.cc @@ -41,8 +41,6 @@ #include #include -#include "arrow/status.h" - #include "ray/object_manager/format/object_manager_generated.h" #include "ray/object_manager/plasma/common.h" #include "ray/object_manager/plasma/fling.h" @@ -199,7 +197,7 @@ Status PlasmaStore::AllocateCudaMemory( Status PlasmaStore::FreeCudaMemory(int device_num, int64_t size, uint8_t* pointer) { ARROW_ASSIGN_OR_RAISE(auto context, manager_->GetContext(device_num - 1)); - RETURN_NOT_OK(context->Free(pointer, size)); + RAY_RETURN_NOT_OK(context->Free(pointer, size)); return Status::OK(); } #endif @@ -691,7 +689,7 @@ void PlasmaStore::EvictObjects(const std::vector& object_ids) { } if (external_store_ && !object_ids.empty()) { - RAY_ARROW_CHECK_OK(external_store_->Put(object_ids, evicted_object_data)); + RAY_CHECK_OK(external_store_->Put(object_ids, evicted_object_data)); for (auto entry : evicted_entries) { PlasmaAllocator::Free(entry->pointer, entry->data_size + entry->metadata_size); entry->pointer = nullptr; @@ -918,7 +916,7 @@ Status PlasmaStore::ProcessMessage(Client* client) { int64_t data_size; int64_t metadata_size; int device_num; - RETURN_NOT_OK(ReadCreateRequest(input, input_size, &object_id, &evict_if_full, + RAY_RETURN_NOT_OK(ReadCreateRequest(input, input_size, &object_id, &evict_if_full, &data_size, &metadata_size, &device_num)); PlasmaError error_code = CreateObject(object_id, evict_if_full, data_size, metadata_size, device_num, client, &object); @@ -941,7 +939,7 @@ Status PlasmaStore::ProcessMessage(Client* client) { bool evict_if_full; std::string data; std::string metadata; - RETURN_NOT_OK(ReadCreateAndSealRequest(input, input_size, &object_id, + RAY_RETURN_NOT_OK(ReadCreateAndSealRequest(input, input_size, &object_id, &evict_if_full, &data, &metadata)); // CreateAndSeal currently only supports device_num = 0, which corresponds // to the host. @@ -973,7 +971,7 @@ Status PlasmaStore::ProcessMessage(Client* client) { std::vector data; std::vector metadata; - RETURN_NOT_OK(ReadCreateAndSealBatchRequest( + RAY_RETURN_NOT_OK(ReadCreateAndSealBatchRequest( input, input_size, &object_ids, &evict_if_full, &data, &metadata)); // CreateAndSeal currently only supports device_num = 0, which corresponds @@ -1019,7 +1017,7 @@ Status PlasmaStore::ProcessMessage(Client* client) { HANDLE_SIGPIPE(SendCreateAndSealBatchReply(client->fd, error_code), client->fd); } break; case fb::MessageType::PlasmaAbortRequest: { - RETURN_NOT_OK(ReadAbortRequest(input, input_size, &object_id)); + RAY_RETURN_NOT_OK(ReadAbortRequest(input, input_size, &object_id)); RAY_CHECK(AbortObject(object_id, client) == 1) << "To abort an object, the only " "client currently using it " "must be the creator."; @@ -1028,17 +1026,17 @@ Status PlasmaStore::ProcessMessage(Client* client) { case fb::MessageType::PlasmaGetRequest: { std::vector object_ids_to_get; int64_t timeout_ms; - RETURN_NOT_OK(ReadGetRequest(input, input_size, object_ids_to_get, &timeout_ms)); + RAY_RETURN_NOT_OK(ReadGetRequest(input, input_size, object_ids_to_get, &timeout_ms)); ProcessGetRequest(client, object_ids_to_get, timeout_ms); } break; case fb::MessageType::PlasmaReleaseRequest: { - RETURN_NOT_OK(ReadReleaseRequest(input, input_size, &object_id)); + RAY_RETURN_NOT_OK(ReadReleaseRequest(input, input_size, &object_id)); ReleaseObject(object_id, client); } break; case fb::MessageType::PlasmaDeleteRequest: { std::vector object_ids; std::vector error_codes; - RETURN_NOT_OK(ReadDeleteRequest(input, input_size, &object_ids)); + RAY_RETURN_NOT_OK(ReadDeleteRequest(input, input_size, &object_ids)); error_codes.reserve(object_ids.size()); for (auto& object_id : object_ids) { error_codes.push_back(DeleteObject(object_id)); @@ -1046,7 +1044,7 @@ Status PlasmaStore::ProcessMessage(Client* client) { HANDLE_SIGPIPE(SendDeleteReply(client->fd, object_ids, error_codes), client->fd); } break; case fb::MessageType::PlasmaContainsRequest: { - RETURN_NOT_OK(ReadContainsRequest(input, input_size, &object_id)); + RAY_RETURN_NOT_OK(ReadContainsRequest(input, input_size, &object_id)); if (ContainsObject(object_id) == ObjectStatus::OBJECT_FOUND) { HANDLE_SIGPIPE(SendContainsReply(client->fd, object_id, 1), client->fd); } else { @@ -1054,14 +1052,14 @@ Status PlasmaStore::ProcessMessage(Client* client) { } } break; case fb::MessageType::PlasmaSealRequest: { - RETURN_NOT_OK(ReadSealRequest(input, input_size, &object_id)); + RAY_RETURN_NOT_OK(ReadSealRequest(input, input_size, &object_id)); SealObjects({object_id}); HANDLE_SIGPIPE(SendSealReply(client->fd, object_id, PlasmaError::OK), client->fd); } break; case fb::MessageType::PlasmaEvictRequest: { // This code path should only be used for testing. int64_t num_bytes; - RETURN_NOT_OK(ReadEvictRequest(input, input_size, &num_bytes)); + RAY_RETURN_NOT_OK(ReadEvictRequest(input, input_size, &num_bytes)); std::vector objects_to_evict; int64_t num_bytes_evicted = eviction_policy_.ChooseObjectsToEvict(num_bytes, &objects_to_evict); @@ -1070,7 +1068,7 @@ Status PlasmaStore::ProcessMessage(Client* client) { } break; case fb::MessageType::PlasmaRefreshLRURequest: { std::vector object_ids; - RETURN_NOT_OK(ReadRefreshLRURequest(input, input_size, &object_ids)); + RAY_RETURN_NOT_OK(ReadRefreshLRURequest(input, input_size, &object_ids)); eviction_policy_.RefreshObjects(object_ids); HANDLE_SIGPIPE(SendRefreshLRUReply(client->fd), client->fd); } break; @@ -1088,7 +1086,7 @@ Status PlasmaStore::ProcessMessage(Client* client) { case fb::MessageType::PlasmaSetOptionsRequest: { std::string client_name; int64_t output_memory_quota; - RETURN_NOT_OK( + RAY_RETURN_NOT_OK( ReadSetOptionsRequest(input, input_size, &client_name, &output_memory_quota)); client->name = client_name; bool success = eviction_policy_.SetClientQuota(client, output_memory_quota); diff --git a/src/ray/object_manager/plasma/store.h b/src/ray/object_manager/plasma/store.h index bc06a78a6..128af541d 100644 --- a/src/ray/object_manager/plasma/store.h +++ b/src/ray/object_manager/plasma/store.h @@ -24,6 +24,7 @@ #include #include +#include "ray/common/status.h" #include "ray/object_manager/format/object_manager_generated.h" #include "ray/object_manager/notification/object_store_notification_manager.h" #include "ray/object_manager/plasma/common.h" @@ -33,12 +34,10 @@ #include "ray/object_manager/plasma/protocol.h" #include "ray/object_manager/plasma/quota_aware_policy.h" -namespace arrow { -class Status; -} // namespace arrow - namespace plasma { +using ray::Status; + namespace flatbuf { enum class PlasmaError; } // namespace flatbuf @@ -170,7 +169,7 @@ class PlasmaStore { NotificationMap::iterator SendNotifications(NotificationMap::iterator it); - arrow::Status ProcessMessage(Client* client); + Status ProcessMessage(Client* client); void SetNotificationListener( const std::shared_ptr ¬ification_listener) { diff --git a/src/ray/object_manager/plasma/store_runner.cc b/src/ray/object_manager/plasma/store_runner.cc index 1d369e525..fa10b57c6 100644 --- a/src/ray/object_manager/plasma/store_runner.cc +++ b/src/ray/object_manager/plasma/store_runner.cc @@ -84,14 +84,14 @@ void PlasmaStoreRunner::Start() { std::shared_ptr external_store{nullptr}; if (!external_store_endpoint_.empty()) { std::string name; - RAY_ARROW_CHECK_OK( + RAY_CHECK_OK( plasma::ExternalStores::ExtractStoreName(external_store_endpoint_, &name)); external_store = plasma::ExternalStores::GetStore(name); if (external_store == nullptr) { RAY_LOG(FATAL) << "No such external store \"" << name << "\""; } RAY_LOG(DEBUG) << "connecting to external store..."; - RAY_ARROW_CHECK_OK(external_store->Connect(external_store_endpoint_)); + RAY_CHECK_OK(external_store->Connect(external_store_endpoint_)); } RAY_LOG(DEBUG) << "starting server listening on " << socket_name_; diff --git a/src/ray/object_manager/test/object_manager_stress_test.cc b/src/ray/object_manager/test/object_manager_stress_test.cc index 172db930b..9d0723bc3 100644 --- a/src/ray/object_manager/test/object_manager_stress_test.cc +++ b/src/ray/object_manager/test/object_manager_stress_test.cc @@ -118,13 +118,13 @@ class TestObjectManagerBase : public ::testing::Test { server2.reset(new MockServer(main_service, om_config_2, gcs_client_2)); // connect to stores. - RAY_ARROW_CHECK_OK(client1.Connect(socket_name_1)); - RAY_ARROW_CHECK_OK(client2.Connect(socket_name_2)); + RAY_CHECK_OK(client1.Connect(socket_name_1)); + RAY_CHECK_OK(client2.Connect(socket_name_2)); } void TearDown() { - arrow::Status client1_status = client1.Disconnect(); - arrow::Status client2_status = client2.Disconnect(); + Status client1_status = client1.Disconnect(); + Status client2_status = client2.Disconnect(); ASSERT_TRUE(client1_status.ok() && client2_status.ok()); gcs_client_1->Disconnect(); @@ -143,9 +143,8 @@ class TestObjectManagerBase : public ::testing::Test { uint8_t metadata[] = {5}; int64_t metadata_size = sizeof(metadata); std::shared_ptr data; - RAY_ARROW_CHECK_OK( - client.Create(object_id, data_size, metadata, metadata_size, &data)); - RAY_ARROW_CHECK_OK(client.Seal(object_id)); + RAY_CHECK_OK(client.Create(object_id, data_size, metadata, metadata_size, &data)); + RAY_CHECK_OK(client.Seal(object_id)); return object_id; } @@ -264,15 +263,14 @@ class StressTestObjectManager : public TestObjectManagerBase { plasma::ObjectBuffer GetObject(plasma::PlasmaClient &client, ObjectID &object_id) { plasma::ObjectBuffer object_buffer; - plasma::ObjectID plasma_id = object_id; - RAY_ARROW_CHECK_OK(client.Get(&plasma_id, 1, 0, &object_buffer)); + RAY_CHECK_OK(client.Get(&object_id, 1, 0, &object_buffer)); return object_buffer; } static unsigned char *GetDigest(plasma::PlasmaClient &client, ObjectID &object_id) { const int64_t size = sizeof(uint64_t); static unsigned char digest_1[size]; - RAY_ARROW_CHECK_OK(client.Hash(object_id, &digest_1[0])); + RAY_CHECK_OK(client.Hash(object_id, &digest_1[0])); return digest_1; } diff --git a/src/ray/object_manager/test/object_manager_test.cc b/src/ray/object_manager/test/object_manager_test.cc index 9fcefba4f..f82ae85fc 100644 --- a/src/ray/object_manager/test/object_manager_test.cc +++ b/src/ray/object_manager/test/object_manager_test.cc @@ -113,13 +113,13 @@ class TestObjectManagerBase : public ::testing::Test { server2.reset(new MockServer(main_service, om_config_2, gcs_client_2)); // connect to stores. - RAY_ARROW_CHECK_OK(client1.Connect(socket_name_1)); - RAY_ARROW_CHECK_OK(client2.Connect(socket_name_2)); + RAY_CHECK_OK(client1.Connect(socket_name_1)); + RAY_CHECK_OK(client2.Connect(socket_name_2)); } void TearDown() { - arrow::Status client1_status = client1.Disconnect(); - arrow::Status client2_status = client2.Disconnect(); + Status client1_status = client1.Disconnect(); + Status client2_status = client2.Disconnect(); ASSERT_TRUE(client1_status.ok() && client2_status.ok()); gcs_client_1->Disconnect(); @@ -142,9 +142,8 @@ class TestObjectManagerBase : public ::testing::Test { uint8_t metadata[] = {5}; int64_t metadata_size = sizeof(metadata); std::shared_ptr data; - RAY_ARROW_CHECK_OK( - client.Create(object_id, data_size, metadata, metadata_size, &data)); - RAY_ARROW_CHECK_OK(client.Seal(object_id)); + RAY_CHECK_OK(client.Create(object_id, data_size, metadata, metadata_size, &data)); + RAY_CHECK_OK(client.Seal(object_id)); return object_id; } diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 522e34425..d2c77844b 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -185,7 +185,7 @@ NodeManager::NodeManager(boost::asio::io_service &io_service, local_resources.GetTotalResources().GetResourceMap())); } - RAY_ARROW_CHECK_OK(store_client_.Connect(config.store_socket_name.c_str())); + RAY_CHECK_OK(store_client_.Connect(config.store_socket_name.c_str())); // Run the node manger rpc server. node_manager_server_.RegisterService(node_manager_service_); node_manager_server_.Run(); @@ -2103,8 +2103,8 @@ void NodeManager::MarkObjectsAsFailed(const ErrorType &error_type, const JobID &job_id) { const std::string meta = std::to_string(static_cast(error_type)); for (const auto &object_id : objects_to_fail) { - arrow::Status status = store_client_.CreateAndSeal(object_id, "", meta); - if (!status.ok() && !plasma::IsPlasmaObjectExists(status)) { + Status status = store_client_.CreateAndSeal(object_id, "", meta); + if (!status.ok() && !status.IsObjectExists()) { // If we failed to save the error code, log a warning and push an error message // to the driver. std::ostringstream stream; @@ -3404,6 +3404,7 @@ void NodeManager::HandlePinObjectIDs(const rpc::PinObjectIDsRequest &request, // an `AsyncGet` instead. if (!store_client_.Get(object_ids, /*timeout_ms=*/0, &plasma_results).ok()) { RAY_LOG(WARNING) << "Failed to get objects to be pinned from object store."; + // TODO(suquark): Maybe "Status::ObjectNotFound" is more accurate here. send_reply_callback(Status::Invalid("Failed to get objects."), nullptr, nullptr); return; } diff --git a/src/ray/raylet/object_manager_integration_test.cc b/src/ray/raylet/object_manager_integration_test.cc index 441000577..2b6d0711f 100644 --- a/src/ray/raylet/object_manager_integration_test.cc +++ b/src/ray/raylet/object_manager_integration_test.cc @@ -30,9 +30,7 @@ std::string test_executable; // TODO(hme): Get this working once the dust settles. class TestObjectManagerBase : public ::testing::Test { public: - TestObjectManagerBase() { - RAY_LOG(INFO) << "TestObjectManagerBase: started."; - } + TestObjectManagerBase() { RAY_LOG(INFO) << "TestObjectManagerBase: started."; } NodeManagerConfig GetNodeManagerConfig(std::string raylet_socket_name, std::string store_socket_name) { @@ -80,13 +78,13 @@ class TestObjectManagerBase : public ::testing::Test { GetNodeManagerConfig("raylet_2", store_sock_2), om_config_2, gcs_client_2)); // connect to stores. - RAY_ARROW_CHECK_OK(client1.Connect(store_sock_1)); - RAY_ARROW_CHECK_OK(client2.Connect(store_sock_2)); + RAY_CHECK_OK(client1.Connect(store_sock_1)); + RAY_CHECK_OK(client2.Connect(store_sock_2)); } void TearDown() { - arrow::Status client1_status = client1.Disconnect(); - arrow::Status client2_status = client2.Disconnect(); + Status client1_status = client1.Disconnect(); + Status client2_status = client2.Disconnect(); ASSERT_TRUE(client1_status.ok() && client2_status.ok()); this->server1.reset(); @@ -105,9 +103,8 @@ class TestObjectManagerBase : public ::testing::Test { uint8_t metadata[] = {5}; int64_t metadata_size = sizeof(metadata); std::shared_ptr data; - RAY_ARROW_CHECK_OK( - client.Create(object_id, data_size, metadata, metadata_size, &data)); - RAY_ARROW_CHECK_OK(client.Seal(object_id)); + RAY_CHECK_OK(client.Create(object_id, data_size, metadata, metadata_size, &data)); + RAY_CHECK_OK(client.Seal(object_id)); return object_id; }