wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-08-05 13:21:03 +08:00
Code Issues Packages Projects Releases Wiki Activity

Replace arrow::Status with ray::Status in Plasma (#9154)

Browse Source 
* add object store status

* replace arrow status with ray status

* cleanup

* remove plasma common.cc
This commit is contained in:
Siyuan (Ryans) Zhuang
2020-06-26 14:06:34 -07:00
committed by GitHub
parent b7cc1e78d7
commit d7549d6184
24 changed files with 215 additions and 344 deletions
Download Patch File Download Diff File Expand all files Collapse all files
BUILD.bazel
-1
View File
@@ -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",
src/ray/common/status.cc
+11 -5
View File
@@ -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;
src/ray/common/status.h
+34 -33
View File
@@ -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.
src/ray/core_worker/store_provider/plasma_store_provider.cc
+10 -10
View File
@@ -37,7 +37,7 @@ CoreWorkerPlasmaStoreProvider::CoreWorkerPlasmaStoreProvider(
get_current_call_site_ = []() { return "<no callsite callback>"; };
}
buffer_tracker_ = std::make_shared<BufferTracker>();
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<std::mutex> 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<Buffer> &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> arrow_buffer;
{
std::lock_guard<std::mutex> guard(store_client_mutex_);
@@ -99,7 +99,7 @@ Status CoreWorkerPlasmaStoreProvider::Create(const std::shared_ptr<Buffer> &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<Buffer> &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>(PlasmaBuffer(arrow_buffer));
status = Status::OK();
}
@@ -138,7 +138,7 @@ Status CoreWorkerPlasmaStoreProvider::Create(const std::shared_ptr<Buffer> &meta
Status CoreWorkerPlasmaStoreProvider::Seal(const ObjectID &object_id) {
{
std::lock_guard<std::mutex> 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<std::mutex> 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::ObjectBuffer> plasma_results;
{
std::lock_guard<std::mutex> 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<std::mutex> 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();
}
src/ray/object_manager/notification/object_store_notification_manager_ipc.cc
+4 -4
View File
@@ -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() {
src/ray/object_manager/object_buffer_pool.cc
+11 -12
View File
@@ -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<const ObjectBufferPool::ChunkInfo &, ray::Status> ObjectBufferPool::Ge
std::lock_guard<std::mutex> 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<const ObjectBufferPool::ChunkInfo &, ray::Status>(
@@ -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<std::mutex> 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<const ObjectBufferPool::ChunkInfo &, ray::Status> ObjectBufferPool::Cr
int64_t object_size = data_size - metadata_size;
// Try to create shared buffer.
std::shared_ptr<Buffer> 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<boost::asio::mutable_buffer> 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::ChunkInfo> ObjectBufferPool::BuildChunks(
void ObjectBufferPool::FreeObjects(const std::vector<ObjectID> &object_ids) {
std::lock_guard<std::mutex> 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 {
src/ray/object_manager/plasma/client.cc
+56 -46
View File
@@ -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<uint8_t> 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<uint8_t> 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<ObjectID>& 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<uint8_t> 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<uint8_t> buffer;
RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaGetReply, &buffer));
RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaGetReply, &buffer));
std::vector<ObjectID> received_object_ids(num_objects);
std::vector<PlasmaObject> object_data(num_objects);
PlasmaObject* object;
std::vector<int> store_fds;
std::vector<int64_t> 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<uint8_t> 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<uint8_t> 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<uint8_t> 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<ObjectID>& 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<uint8_t> 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<PlasmaError> error_codes;
not_in_use_ids.clear();
RETURN_NOT_OK(
RAY_RETURN_NOT_OK(
ReadDeleteReply(buffer.data(), buffer.size(), &not_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<std::recursive_mutex> 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<uint8_t> 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<ObjectID>& object_ids) {
std::lock_guard<std::recursive_mutex> guard(client_mutex_);
RETURN_NOT_OK(SendRefreshLRURequest(store_conn_, object_ids));
RAY_RETURN_NOT_OK(SendRefreshLRURequest(store_conn_, object_ids));
std::vector<uint8_t> 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<ObjectBuffer> 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<ObjectID> object_ids;
std::vector<int64_t> data_sizes;
std::vector<int64_t> 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<std::recursive_mutex> 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<uint8_t> 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<std::recursive_mutex> 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<uint8_t> 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());
}
src/ray/object_manager/plasma/client.h
+3 -2
View File
@@ -23,16 +23,17 @@
#include <vector>
#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.
src/ray/object_manager/plasma/common.cc
-115
View File
@@ -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 <limits>
#include <utility>
#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<PlasmaStatusDetail*>(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<PlasmaStatusDetail>(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
src/ray/object_manager/plasma/common.h
-25
View File
@@ -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<ObjectID, std::unique_ptr<ObjectTableEntry>> 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
src/ray/object_manager/plasma/dlmalloc.cc
+2
View File
@@ -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
src/ray/object_manager/plasma/io.cc
+8 -12
View File
@@ -21,8 +21,6 @@
#include <memory>
#include <sstream>
#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 <netinet/in.h>
#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<uint8_t*>(&version), sizeof(version)));
RETURN_NOT_OK(WriteBytes(fd, reinterpret_cast<uint8_t*>(&type), sizeof(type)));
RETURN_NOT_OK(WriteBytes(fd, reinterpret_cast<uint8_t*>(&length), sizeof(length)));
RAY_RETURN_NOT_OK(WriteBytes(fd, reinterpret_cast<uint8_t*>(&version), sizeof(version)));
RAY_RETURN_NOT_OK(WriteBytes(fd, reinterpret_cast<uint8_t*>(&type), sizeof(type)));
RAY_RETURN_NOT_OK(WriteBytes(fd, reinterpret_cast<uint8_t*>(&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<uint8_t>* buffer) {
int64_t version;
RETURN_NOT_OK_ELSE(ReadBytes(fd, reinterpret_cast<uint8_t*>(&version), sizeof(version)),
RAY_RETURN_NOT_OK_ELSE(ReadBytes(fd, reinterpret_cast<uint8_t*>(&version), sizeof(version)),
*type = MessageType::PlasmaDisconnectClient);
RAY_CHECK(version == kPlasmaProtocolVersion) << "version = " << version;
RETURN_NOT_OK_ELSE(ReadBytes(fd, reinterpret_cast<uint8_t*>(type), sizeof(*type)),
RAY_RETURN_NOT_OK_ELSE(ReadBytes(fd, reinterpret_cast<uint8_t*>(type), sizeof(*type)),
*type = MessageType::PlasmaDisconnectClient);
int64_t length_temp;
RETURN_NOT_OK_ELSE(
RAY_RETURN_NOT_OK_ELSE(
ReadBytes(fd, reinterpret_cast<uint8_t*>(&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<uint8_t>* 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();
src/ray/object_manager/plasma/io.h
+3 -3
View File
@@ -26,12 +26,14 @@
#include <string>
#include <vector>
#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);
src/ray/object_manager/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc
+17 -13
View File
@@ -28,19 +28,23 @@
#include <string>
#include <vector>
#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<jbyte*>(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<jbyte*>(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<plasma::PlasmaClient*>(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<Buffer> 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<plasma::PlasmaClient*>(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<plasma::PlasmaClient*>(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<plasma::PlasmaClient*>(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<plasma::PlasmaClient*>(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<plasma::PlasmaClient*>(conn);
jsize num_oids = env->GetArrayLength(object_ids);
std::vector<plasma::ObjectID> oids(num_oids);
std::vector<ObjectID> oids(num_oids);
std::vector<plasma::ObjectBuffer> 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<plasma::PlasmaClient*>(conn);
plasma::ObjectID oid;
ObjectID oid;
jbyteArray_to_object_id(env, object_id, &oid);
bool has_object;
src/ray/object_manager/plasma/plasma.h
+5 -1
View File
@@ -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<uint8_t[]> CreateObjectInfoBuffer(ObjectInfoT* object_info);
std::unique_ptr<uint8_t[]> CreatePlasmaNotificationBuffer(
const std::vector<ObjectInfoT>& object_info);
/// Globally accessible reference to plasma store configuration.
extern const PlasmaStoreInfo* plasma_config;
} // namespace plasma
src/ray/object_manager/plasma/protocol.cc
+4 -7
View File
@@ -71,7 +71,7 @@ flatbuffers::Offset<flatbuffers::Vector<int64_t>> ToFlatbuffer(
Status PlasmaReceive(int sock, MessageType message_type, std::vector<uint8_t>* 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<int64_t>(type)
<< ", message_type = " << static_cast<int64_t>(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<int>(plasma_error);
}
src/ray/object_manager/plasma/protocol.h
+2 -2
View File
@@ -22,13 +22,13 @@
#include <unordered_map>
#include <vector>
#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;
src/ray/object_manager/plasma/store.cc
+14 -16
View File
@@ -41,8 +41,6 @@
#include <utility>
#include <vector>
#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<ObjectID>& 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<std::string> data;
std::vector<std::string> 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<ObjectID> 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<ObjectID> object_ids;
std::vector<PlasmaError> 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<ObjectID> 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<ObjectID> 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);
src/ray/object_manager/plasma/store.h
+4 -5
View File
@@ -24,6 +24,7 @@
#include <unordered_set>
#include <vector>
#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<ray::ObjectStoreNotificationManager> &notification_listener) {
src/ray/object_manager/plasma/store_runner.cc
+2 -2
View File
@@ -84,14 +84,14 @@ void PlasmaStoreRunner::Start() {
std::shared_ptr<plasma::ExternalStore> 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_;
src/ray/object_manager/test/object_manager_stress_test.cc
+8 -10
View File
@@ -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<arrow::Buffer> 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;
}
src/ray/object_manager/test/object_manager_test.cc
+6 -7
View File
@@ -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<arrow::Buffer> 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;
}
src/ray/raylet/node_manager.cc
+4 -3
View File
@@ -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<int>(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;
}
src/ray/raylet/object_manager_integration_test.cc
+7 -10
View File
@@ -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<Buffer> 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;
}