You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by pc...@apache.org on 2018/06/27 20:34:54 UTC
[arrow] branch master updated: ARROW-2758: [Plasma] Use Scope enum
in Plasma
This is an automated email from the ASF dual-hosted git repository.
pcmoritz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/master by this push:
new d38e862 ARROW-2758: [Plasma] Use Scope enum in Plasma
d38e862 is described below
commit d38e86253668773005511b91e35b87aa7d36c29f
Author: Yuhong Guo <yu...@antfin.com>
AuthorDate: Wed Jun 27 13:34:46 2018 -0700
ARROW-2758: [Plasma] Use Scope enum in Plasma
In this PR there are following changes:
1. add option "--scoped-enum" to Flat Buffer Compiler.
2. change the old-styled c++ enum to c++11 style.
This change will unify the enum style in both Plasma and Ray. See Ray [Issue2121](https://github.com/ray-project/ray/issues/2121) and Ray [PR2194](https://github.com/ray-project/ray/pull/2194).
Author: Yuhong Guo <yu...@antfin.com>
Closes #2179 from guoyuhong/scopeEnum and squashes the following commits:
fbe70105 <Yuhong Guo> Fix Python and Java Building failure.
397e8b1b <Yuhong Guo> Fix flatc failure.
e96c99b3 <Yuhong Guo> Fix the comment mismatch.
8ab8716d <Yuhong Guo> Change old c++ enum style to c++11 style and fix lint
87cc5918 <Yuhong Guo> Add --scoped-enums to FlatBuffer Compiler
---
cpp/src/plasma/CMakeLists.txt | 2 +-
cpp/src/plasma/client.cc | 36 ++++-----
cpp/src/plasma/client.h | 6 +-
cpp/src/plasma/common.cc | 16 ++--
cpp/src/plasma/common.h | 34 ++++----
cpp/src/plasma/format/plasma.fbs | 4 +-
cpp/src/plasma/io.cc | 12 +--
cpp/src/plasma/io.h | 7 +-
.../org_apache_arrow_plasma_PlasmaClientJNI.cc | 2 +-
cpp/src/plasma/plasma.h | 4 +-
cpp/src/plasma/protocol.cc | 90 +++++++++++-----------
cpp/src/plasma/protocol.h | 12 +--
cpp/src/plasma/store.cc | 82 ++++++++++----------
cpp/src/plasma/store.h | 21 ++---
cpp/src/plasma/test/serialization_tests.cc | 65 +++++++++-------
python/pyarrow/_plasma.pyx | 4 +-
16 files changed, 208 insertions(+), 189 deletions(-)
diff --git a/cpp/src/plasma/CMakeLists.txt b/cpp/src/plasma/CMakeLists.txt
index 8797d96..744f9ad 100644
--- a/cpp/src/plasma/CMakeLists.txt
+++ b/cpp/src/plasma/CMakeLists.txt
@@ -57,7 +57,7 @@ add_custom_command(
# flatbuffers message Message, which can be used to store deserialized
# messages in data structures. This is currently used for ObjectInfo for
# example.
- COMMAND ${FLATBUFFERS_COMPILER} -c -o ${OUTPUT_DIR} ${PLASMA_FBS_SRC} --gen-object-api
+ COMMAND ${FLATBUFFERS_COMPILER} -c -o ${OUTPUT_DIR} ${PLASMA_FBS_SRC} --gen-object-api --scoped-enums
DEPENDS ${PLASMA_FBS_SRC}
COMMENT "Running flatc compiler on ${PLASMA_FBS_SRC}"
VERBATIM)
diff --git a/cpp/src/plasma/client.cc b/cpp/src/plasma/client.cc
index 30d2e43..8e66cf4 100644
--- a/cpp/src/plasma/client.cc
+++ b/cpp/src/plasma/client.cc
@@ -366,7 +366,7 @@ Status PlasmaClient::Impl::Create(const ObjectID& object_id, int64_t data_size,
RETURN_NOT_OK(
SendCreateRequest(store_conn_, object_id, data_size, metadata_size, device_num));
std::vector<uint8_t> buffer;
- RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType_PlasmaCreateReply, &buffer));
+ RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaCreateReply, &buffer));
ObjectID id;
PlasmaObject object;
int store_fd;
@@ -475,7 +475,7 @@ Status PlasmaClient::Impl::GetBuffers(
// 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));
std::vector<uint8_t> buffer;
- RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType_PlasmaGetReply, &buffer));
+ 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;
@@ -677,7 +677,7 @@ Status PlasmaClient::Impl::Contains(const ObjectID& object_id, bool* has_object)
// to see if we have the object.
RETURN_NOT_OK(SendContainsRequest(store_conn_, object_id));
std::vector<uint8_t> buffer;
- RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType_PlasmaContainsReply, &buffer));
+ RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaContainsReply, &buffer));
ObjectID object_id2;
DCHECK_GT(buffer.size(), 0);
RETURN_NOT_OK(
@@ -803,7 +803,7 @@ Status PlasmaClient::Impl::Abort(const ObjectID& object_id) {
std::vector<uint8_t> buffer;
ObjectID id;
- int64_t type;
+ MessageType type;
RETURN_NOT_OK(ReadMessage(store_conn_, &type, &buffer));
return ReadAbortReply(buffer.data(), buffer.size(), &id);
}
@@ -817,7 +817,7 @@ Status PlasmaClient::Impl::Delete(const ObjectID& object_id) {
// If we don't already have a reference to the object, we can try to remove the object
RETURN_NOT_OK(SendDeleteRequest(store_conn_, object_id));
std::vector<uint8_t> buffer;
- RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType_PlasmaDeleteReply, &buffer));
+ RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaDeleteReply, &buffer));
ObjectID object_id2;
DCHECK_GT(buffer.size(), 0);
RETURN_NOT_OK(ReadDeleteReply(buffer.data(), buffer.size(), &object_id2));
@@ -830,7 +830,7 @@ Status PlasmaClient::Impl::Evict(int64_t num_bytes, int64_t& num_bytes_evicted)
RETURN_NOT_OK(SendEvictRequest(store_conn_, num_bytes));
// Wait for a response with the number of bytes actually evicted.
std::vector<uint8_t> buffer;
- int64_t type;
+ MessageType type;
RETURN_NOT_OK(ReadMessage(store_conn_, &type, &buffer));
return ReadEvictReply(buffer.data(), buffer.size(), num_bytes_evicted);
}
@@ -904,7 +904,7 @@ Status PlasmaClient::Impl::Connect(const std::string& store_socket_name,
// Send a ConnectRequest to the store to get its memory capacity.
RETURN_NOT_OK(SendConnectRequest(store_conn_));
std::vector<uint8_t> buffer;
- RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType_PlasmaConnectReply, &buffer));
+ RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaConnectReply, &buffer));
RETURN_NOT_OK(ReadConnectReply(buffer.data(), buffer.size(), &store_capacity_));
return Status::OK();
}
@@ -942,7 +942,7 @@ Status PlasmaClient::Impl::Info(const ObjectID& object_id, int* object_status) {
RETURN_NOT_OK(SendStatusRequest(manager_conn_, &object_id, 1));
std::vector<uint8_t> buffer;
- RETURN_NOT_OK(PlasmaReceive(manager_conn_, MessageType_PlasmaStatusReply, &buffer));
+ RETURN_NOT_OK(PlasmaReceive(manager_conn_, MessageType::PlasmaStatusReply, &buffer));
ObjectID id;
RETURN_NOT_OK(ReadStatusReply(buffer.data(), buffer.size(), &id, object_status, 1));
ARROW_CHECK(object_id == id);
@@ -958,32 +958,32 @@ Status PlasmaClient::Impl::Wait(int64_t num_object_requests,
ARROW_CHECK(num_ready_objects <= num_object_requests);
for (int i = 0; i < num_object_requests; ++i) {
- ARROW_CHECK(object_requests[i].type == PLASMA_QUERY_LOCAL ||
- object_requests[i].type == PLASMA_QUERY_ANYWHERE);
+ ARROW_CHECK(object_requests[i].type == ObjectRequestType::PLASMA_QUERY_LOCAL ||
+ object_requests[i].type == ObjectRequestType::PLASMA_QUERY_ANYWHERE);
}
RETURN_NOT_OK(SendWaitRequest(manager_conn_, object_requests, num_object_requests,
num_ready_objects, timeout_ms));
std::vector<uint8_t> buffer;
- RETURN_NOT_OK(PlasmaReceive(manager_conn_, MessageType_PlasmaWaitReply, &buffer));
+ RETURN_NOT_OK(PlasmaReceive(manager_conn_, MessageType::PlasmaWaitReply, &buffer));
RETURN_NOT_OK(
ReadWaitReply(buffer.data(), buffer.size(), object_requests, &num_ready_objects));
*num_objects_ready = 0;
for (int i = 0; i < num_object_requests; ++i) {
- int type = object_requests[i].type;
- int status = object_requests[i].status;
+ ObjectRequestType type = object_requests[i].type;
+ ObjectStatus status = object_requests[i].status;
switch (type) {
- case PLASMA_QUERY_LOCAL:
- if (status == ObjectStatus_Local) {
+ case ObjectRequestType::PLASMA_QUERY_LOCAL:
+ if (status == ObjectStatus::Local) {
*num_objects_ready += 1;
}
break;
- case PLASMA_QUERY_ANYWHERE:
- if (status == ObjectStatus_Local || status == ObjectStatus_Remote) {
+ case ObjectRequestType::PLASMA_QUERY_ANYWHERE:
+ if (status == ObjectStatus::Local || status == ObjectStatus::Remote) {
*num_objects_ready += 1;
} else {
- ARROW_CHECK(status == ObjectStatus_Nonexistent);
+ ARROW_CHECK(status == ObjectStatus::Nonexistent);
}
break;
default:
diff --git a/cpp/src/plasma/client.h b/cpp/src/plasma/client.h
index c3b5548..5501488 100644
--- a/cpp/src/plasma/client.h
+++ b/cpp/src/plasma/client.h
@@ -259,15 +259,15 @@ class ARROW_EXPORT PlasmaClient {
/// "type" field.
/// - A PLASMA_QUERY_LOCAL request is satisfied when object_id becomes
/// available in the local Plasma Store. In this case, this function
- /// sets the "status" field to ObjectStatus_Local. Note, if the
+ /// sets the "status" field to ObjectStatus::Local. Note, if the
/// status
- /// is not ObjectStatus_Local, it will be ObjectStatus_Nonexistent,
+ /// is not ObjectStatus::Local, it will be ObjectStatus::Nonexistent,
/// but it may exist elsewhere in the system.
/// - A PLASMA_QUERY_ANYWHERE request is satisfied when object_id
/// becomes
/// available either at the local Plasma Store or on a remote Plasma
/// Store. In this case, the functions sets the "status" field to
- /// ObjectStatus_Local or ObjectStatus_Remote.
+ /// ObjectStatus::Local or ObjectStatus::Remote.
/// \param num_ready_objects The number of requests in object_requests array
/// that
/// must be satisfied before the function returns, unless it timeouts.
diff --git a/cpp/src/plasma/common.cc b/cpp/src/plasma/common.cc
index 7ac5413..2e3899b 100644
--- a/cpp/src/plasma/common.cc
+++ b/cpp/src/plasma/common.cc
@@ -78,24 +78,24 @@ bool UniqueID::operator==(const UniqueID& rhs) const {
return std::memcmp(data(), rhs.data(), kUniqueIDSize) == 0;
}
-Status plasma_error_status(int plasma_error) {
+Status plasma_error_status(PlasmaError plasma_error) {
switch (plasma_error) {
- case PlasmaError_OK:
+ case PlasmaError::OK:
return Status::OK();
- case PlasmaError_ObjectExists:
+ case PlasmaError::ObjectExists:
return Status::PlasmaObjectExists("object already exists in the plasma store");
- case PlasmaError_ObjectNonexistent:
+ case PlasmaError::ObjectNonexistent:
return Status::PlasmaObjectNonexistent("object does not exist in the plasma store");
- case PlasmaError_OutOfMemory:
+ case PlasmaError::OutOfMemory:
return Status::PlasmaStoreFull("object does not fit in the plasma store");
default:
- ARROW_LOG(FATAL) << "unknown plasma error code " << plasma_error;
+ ARROW_LOG(FATAL) << "unknown plasma error code " << static_cast<int>(plasma_error);
}
return Status::OK();
}
-ARROW_EXPORT int ObjectStatusLocal = ObjectStatus_Local;
-ARROW_EXPORT int ObjectStatusRemote = ObjectStatus_Remote;
+ARROW_EXPORT ObjectStatus ObjectStatusLocal = ObjectStatus::Local;
+ARROW_EXPORT ObjectStatus ObjectStatusRemote = ObjectStatus::Remote;
const PlasmaStoreInfo* plasma_config;
diff --git a/cpp/src/plasma/common.h b/cpp/src/plasma/common.h
index 7dbcf80..90cd6a0 100644
--- a/cpp/src/plasma/common.h
+++ b/cpp/src/plasma/common.h
@@ -31,6 +31,10 @@
#include "arrow/status.h"
#include "arrow/util/logging.h"
+// Forward declaration outside the namespace, which is defined in plasma_generated.h.
+enum class PlasmaError : int32_t;
+enum class ObjectStatus : int32_t;
+
namespace plasma {
constexpr int64_t kUniqueIDSize = 20;
@@ -54,11 +58,18 @@ static_assert(std::is_pod<UniqueID>::value, "UniqueID must be plain old data");
typedef UniqueID ObjectID;
-arrow::Status plasma_error_status(int plasma_error);
+arrow::Status plasma_error_status(PlasmaError plasma_error);
/// Size of object hash digests.
constexpr int64_t kDigestSize = sizeof(uint64_t);
+enum class ObjectRequestType : int {
+ /// Query for object in the local plasma store.
+ PLASMA_QUERY_LOCAL = 1,
+ /// Query for object in the local plasma store or in a remote plasma store.
+ PLASMA_QUERY_ANYWHERE
+};
+
/// Object request data structure. Used for Wait.
struct ObjectRequest {
/// The ID of the requested object. If ID_NIL request any object.
@@ -68,26 +79,19 @@ struct ObjectRequest {
/// local Plasma Store.
/// - PLASMA_QUERY_ANYWHERE: return if or when the object is available in
/// the system (i.e., either in the local or a remote Plasma Store).
- int type;
+ ObjectRequestType type;
/// Object status. Same as the status returned by plasma_status() function
/// call. This is filled in by plasma_wait_for_objects1():
- /// - ObjectStatus_Local: object is ready at the local Plasma Store.
- /// - ObjectStatus_Remote: object is ready at a remote Plasma Store.
- /// - ObjectStatus_Nonexistent: object does not exist in the system.
+ /// - ObjectStatus::Local: object is ready at the local Plasma Store.
+ /// - ObjectStatus::Remote: object is ready at a remote Plasma Store.
+ /// - ObjectStatus::Nonexistent: object does not exist in the system.
/// - PLASMA_CLIENT_IN_TRANSFER, if the object is currently being scheduled
/// for being transferred or it is transferring.
- int status;
-};
-
-enum ObjectRequestType {
- /// Query for object in the local plasma store.
- PLASMA_QUERY_LOCAL = 1,
- /// Query for object in the local plasma store or in a remote plasma store.
- PLASMA_QUERY_ANYWHERE
+ ObjectStatus status;
};
-extern int ObjectStatusLocal;
-extern int ObjectStatusRemote;
+extern ObjectStatus ObjectStatusLocal;
+extern ObjectStatus ObjectStatusRemote;
/// Globally accessible reference to plasma store configuration.
/// TODO(pcm): This can be avoided with some refactoring of existing code
diff --git a/cpp/src/plasma/format/plasma.fbs b/cpp/src/plasma/format/plasma.fbs
index 6a58fb0..0bc7a07 100644
--- a/cpp/src/plasma/format/plasma.fbs
+++ b/cpp/src/plasma/format/plasma.fbs
@@ -17,7 +17,7 @@
// Plasma protocol specification
-enum MessageType:int {
+enum MessageType:long {
// Message that gets send when a client hangs up.
PlasmaDisconnectClient = 0,
// Create a new object.
@@ -301,7 +301,7 @@ table ObjectReply {
// ID of the object.
object_id: string;
// The object status. This specifies where the object is stored.
- status: int;
+ status: ObjectStatus;
}
table PlasmaWaitReply {
diff --git a/cpp/src/plasma/io.cc b/cpp/src/plasma/io.cc
index aefd297..d9e805c 100644
--- a/cpp/src/plasma/io.cc
+++ b/cpp/src/plasma/io.cc
@@ -59,7 +59,7 @@ Status WriteBytes(int fd, uint8_t* cursor, size_t length) {
return Status::OK();
}
-Status WriteMessage(int fd, int64_t type, int64_t length, uint8_t* bytes) {
+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)));
@@ -90,24 +90,24 @@ Status ReadBytes(int fd, uint8_t* cursor, size_t length) {
return Status::OK();
}
-Status ReadMessage(int fd, int64_t* type, std::vector<uint8_t>* buffer) {
+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)),
- *type = MessageType_PlasmaDisconnectClient);
+ *type = MessageType::PlasmaDisconnectClient);
ARROW_CHECK(version == kPlasmaProtocolVersion) << "version = " << version;
RETURN_NOT_OK_ELSE(ReadBytes(fd, reinterpret_cast<uint8_t*>(type), sizeof(*type)),
- *type = MessageType_PlasmaDisconnectClient);
+ *type = MessageType::PlasmaDisconnectClient);
int64_t length_temp;
RETURN_NOT_OK_ELSE(
ReadBytes(fd, reinterpret_cast<uint8_t*>(&length_temp), sizeof(length_temp)),
- *type = MessageType_PlasmaDisconnectClient);
+ *type = MessageType::PlasmaDisconnectClient);
// The length must be read as an int64_t, but it should be used as a size_t.
size_t length = static_cast<size_t>(length_temp);
if (length > buffer->size()) {
buffer->resize(length);
}
RETURN_NOT_OK_ELSE(ReadBytes(fd, buffer->data(), length),
- *type = MessageType_PlasmaDisconnectClient);
+ *type = MessageType::PlasmaDisconnectClient);
return Status::OK();
}
diff --git a/cpp/src/plasma/io.h b/cpp/src/plasma/io.h
index f0d57c0..9fc2d1c 100644
--- a/cpp/src/plasma/io.h
+++ b/cpp/src/plasma/io.h
@@ -30,6 +30,9 @@
#include "arrow/status.h"
#include "plasma/compat.h"
+// Forward declaration outside the namespace, which is defined in plasma_generated.h.
+enum class MessageType : int64_t;
+
namespace plasma {
// TODO(pcm): Replace our own custom message header (message type,
@@ -41,11 +44,11 @@ using arrow::Status;
Status WriteBytes(int fd, uint8_t* cursor, size_t length);
-Status WriteMessage(int fd, int64_t type, int64_t length, uint8_t* bytes);
+Status WriteMessage(int fd, MessageType type, int64_t length, uint8_t* bytes);
Status ReadBytes(int fd, uint8_t* cursor, size_t length);
-Status ReadMessage(int fd, int64_t* type, std::vector<uint8_t>* buffer);
+Status ReadMessage(int fd, MessageType* type, std::vector<uint8_t>* buffer);
int bind_ipc_sock(const std::string& pathname, bool shall_listen);
diff --git a/cpp/src/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc b/cpp/src/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc
index 8ddfeeb..270e696 100644
--- a/cpp/src/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc
+++ b/cpp/src/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc
@@ -252,7 +252,7 @@ JNIEXPORT jobjectArray JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_wait
jbyteArray_to_object_id(
env, reinterpret_cast<jbyteArray>(env->GetObjectArrayElement(object_ids, i)),
&oreqs[i].object_id);
- oreqs[i].type = plasma::PLASMA_QUERY_ANYWHERE;
+ oreqs[i].type = plasma::ObjectRequestType::PLASMA_QUERY_ANYWHERE;
}
int num_return_objects;
diff --git a/cpp/src/plasma/plasma.h b/cpp/src/plasma/plasma.h
index 43a7496..8cc7cac 100644
--- a/cpp/src/plasma/plasma.h
+++ b/cpp/src/plasma/plasma.h
@@ -95,14 +95,14 @@ struct PlasmaObject {
int device_num;
};
-enum object_state {
+enum class object_state : int {
/// Object was created but not sealed in the local Plasma Store.
PLASMA_CREATED = 1,
/// Object is sealed and stored in the local Plasma Store.
PLASMA_SEALED
};
-enum object_status {
+enum class object_status : int {
/// The object was not found.
OBJECT_NOT_FOUND = 0,
/// The object was found.
diff --git a/cpp/src/plasma/protocol.cc b/cpp/src/plasma/protocol.cc
index e5cce75..0503e10 100644
--- a/cpp/src/plasma/protocol.cc
+++ b/cpp/src/plasma/protocol.cc
@@ -41,16 +41,17 @@ to_flatbuffer(flatbuffers::FlatBufferBuilder* fbb, const ObjectID* object_ids,
return fbb->CreateVector(results);
}
-Status PlasmaReceive(int sock, int64_t message_type, std::vector<uint8_t>* buffer) {
- int64_t type;
+Status PlasmaReceive(int sock, MessageType message_type, std::vector<uint8_t>* buffer) {
+ MessageType type;
RETURN_NOT_OK(ReadMessage(sock, &type, buffer));
ARROW_CHECK(type == message_type)
- << "type = " << type << ", message_type = " << message_type;
+ << "type = " << static_cast<int64_t>(type)
+ << ", message_type = " << static_cast<int64_t>(message_type);
return Status::OK();
}
template <typename Message>
-Status PlasmaSend(int sock, int64_t message_type, flatbuffers::FlatBufferBuilder* fbb,
+Status PlasmaSend(int sock, MessageType message_type, flatbuffers::FlatBufferBuilder* fbb,
const Message& message) {
fbb->Finish(message);
return WriteMessage(sock, message_type, fbb->GetSize(), fbb->GetBufferPointer());
@@ -63,7 +64,7 @@ Status SendCreateRequest(int sock, ObjectID object_id, int64_t data_size,
flatbuffers::FlatBufferBuilder fbb;
auto message = CreatePlasmaCreateRequest(fbb, fbb.CreateString(object_id.binary()),
data_size, metadata_size, device_num);
- return PlasmaSend(sock, MessageType_PlasmaCreateRequest, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaCreateRequest, &fbb, message);
}
Status ReadCreateRequest(uint8_t* data, size_t size, ObjectID* object_id,
@@ -78,8 +79,8 @@ Status ReadCreateRequest(uint8_t* data, size_t size, ObjectID* object_id,
return Status::OK();
}
-Status SendCreateReply(int sock, ObjectID object_id, PlasmaObject* object, int error_code,
- int64_t mmap_size) {
+Status SendCreateReply(int sock, ObjectID object_id, PlasmaObject* object,
+ PlasmaError error_code, int64_t mmap_size) {
flatbuffers::FlatBufferBuilder fbb;
PlasmaObjectSpec plasma_object(object->store_fd, object->data_offset, object->data_size,
object->metadata_offset, object->metadata_size,
@@ -107,7 +108,7 @@ Status SendCreateReply(int sock, ObjectID object_id, PlasmaObject* object, int e
#endif
}
auto message = crb.Finish();
- return PlasmaSend(sock, MessageType_PlasmaCreateReply, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaCreateReply, &fbb, message);
}
Status ReadCreateReply(uint8_t* data, size_t size, ObjectID* object_id,
@@ -138,7 +139,7 @@ Status ReadCreateReply(uint8_t* data, size_t size, ObjectID* object_id,
Status SendAbortRequest(int sock, ObjectID object_id) {
flatbuffers::FlatBufferBuilder fbb;
auto message = CreatePlasmaAbortRequest(fbb, fbb.CreateString(object_id.binary()));
- return PlasmaSend(sock, MessageType_PlasmaAbortRequest, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaAbortRequest, &fbb, message);
}
Status ReadAbortRequest(uint8_t* data, size_t size, ObjectID* object_id) {
@@ -152,7 +153,7 @@ Status ReadAbortRequest(uint8_t* data, size_t size, ObjectID* object_id) {
Status SendAbortReply(int sock, ObjectID object_id) {
flatbuffers::FlatBufferBuilder fbb;
auto message = CreatePlasmaAbortReply(fbb, fbb.CreateString(object_id.binary()));
- return PlasmaSend(sock, MessageType_PlasmaAbortReply, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaAbortReply, &fbb, message);
}
Status ReadAbortReply(uint8_t* data, size_t size, ObjectID* object_id) {
@@ -170,7 +171,7 @@ Status SendSealRequest(int sock, ObjectID object_id, unsigned char* digest) {
auto digest_string = fbb.CreateString(reinterpret_cast<char*>(digest), kDigestSize);
auto message =
CreatePlasmaSealRequest(fbb, fbb.CreateString(object_id.binary()), digest_string);
- return PlasmaSend(sock, MessageType_PlasmaSealRequest, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaSealRequest, &fbb, message);
}
Status ReadSealRequest(uint8_t* data, size_t size, ObjectID* object_id,
@@ -184,11 +185,10 @@ Status ReadSealRequest(uint8_t* data, size_t size, ObjectID* object_id,
return Status::OK();
}
-Status SendSealReply(int sock, ObjectID object_id, int error) {
+Status SendSealReply(int sock, ObjectID object_id, PlasmaError error) {
flatbuffers::FlatBufferBuilder fbb;
- auto message = CreatePlasmaSealReply(fbb, fbb.CreateString(object_id.binary()),
- static_cast<PlasmaError>(error));
- return PlasmaSend(sock, MessageType_PlasmaSealReply, &fbb, message);
+ auto message = CreatePlasmaSealReply(fbb, fbb.CreateString(object_id.binary()), error);
+ return PlasmaSend(sock, MessageType::PlasmaSealReply, &fbb, message);
}
Status ReadSealReply(uint8_t* data, size_t size, ObjectID* object_id) {
@@ -204,7 +204,7 @@ Status ReadSealReply(uint8_t* data, size_t size, ObjectID* object_id) {
Status SendReleaseRequest(int sock, ObjectID object_id) {
flatbuffers::FlatBufferBuilder fbb;
auto message = CreatePlasmaReleaseRequest(fbb, fbb.CreateString(object_id.binary()));
- return PlasmaSend(sock, MessageType_PlasmaReleaseRequest, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaReleaseRequest, &fbb, message);
}
Status ReadReleaseRequest(uint8_t* data, size_t size, ObjectID* object_id) {
@@ -215,11 +215,11 @@ Status ReadReleaseRequest(uint8_t* data, size_t size, ObjectID* object_id) {
return Status::OK();
}
-Status SendReleaseReply(int sock, ObjectID object_id, int error) {
+Status SendReleaseReply(int sock, ObjectID object_id, PlasmaError error) {
flatbuffers::FlatBufferBuilder fbb;
- auto message = CreatePlasmaReleaseReply(fbb, fbb.CreateString(object_id.binary()),
- static_cast<PlasmaError>(error));
- return PlasmaSend(sock, MessageType_PlasmaReleaseReply, &fbb, message);
+ auto message =
+ CreatePlasmaReleaseReply(fbb, fbb.CreateString(object_id.binary()), error);
+ return PlasmaSend(sock, MessageType::PlasmaReleaseReply, &fbb, message);
}
Status ReadReleaseReply(uint8_t* data, size_t size, ObjectID* object_id) {
@@ -235,7 +235,7 @@ Status ReadReleaseReply(uint8_t* data, size_t size, ObjectID* object_id) {
Status SendDeleteRequest(int sock, ObjectID object_id) {
flatbuffers::FlatBufferBuilder fbb;
auto message = CreatePlasmaDeleteRequest(fbb, fbb.CreateString(object_id.binary()));
- return PlasmaSend(sock, MessageType_PlasmaDeleteRequest, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaDeleteRequest, &fbb, message);
}
Status ReadDeleteRequest(uint8_t* data, size_t size, ObjectID* object_id) {
@@ -246,11 +246,11 @@ Status ReadDeleteRequest(uint8_t* data, size_t size, ObjectID* object_id) {
return Status::OK();
}
-Status SendDeleteReply(int sock, ObjectID object_id, int error) {
+Status SendDeleteReply(int sock, ObjectID object_id, PlasmaError error) {
flatbuffers::FlatBufferBuilder fbb;
- auto message = CreatePlasmaDeleteReply(fbb, fbb.CreateString(object_id.binary()),
- static_cast<PlasmaError>(error));
- return PlasmaSend(sock, MessageType_PlasmaDeleteReply, &fbb, message);
+ auto message =
+ CreatePlasmaDeleteReply(fbb, fbb.CreateString(object_id.binary()), error);
+ return PlasmaSend(sock, MessageType::PlasmaDeleteReply, &fbb, message);
}
Status ReadDeleteReply(uint8_t* data, size_t size, ObjectID* object_id) {
@@ -267,7 +267,7 @@ Status SendStatusRequest(int sock, const ObjectID* object_ids, int64_t num_objec
flatbuffers::FlatBufferBuilder fbb;
auto message =
CreatePlasmaStatusRequest(fbb, to_flatbuffer(&fbb, object_ids, num_objects));
- return PlasmaSend(sock, MessageType_PlasmaStatusRequest, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaStatusRequest, &fbb, message);
}
Status ReadStatusRequest(uint8_t* data, size_t size, ObjectID object_ids[],
@@ -287,7 +287,7 @@ Status SendStatusReply(int sock, ObjectID object_ids[], int object_status[],
auto message =
CreatePlasmaStatusReply(fbb, to_flatbuffer(&fbb, object_ids, num_objects),
fbb.CreateVector(object_status, num_objects));
- return PlasmaSend(sock, MessageType_PlasmaStatusReply, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaStatusReply, &fbb, message);
}
int64_t ReadStatusReply_num_objects(uint8_t* data, size_t size) {
@@ -316,7 +316,7 @@ Status ReadStatusReply(uint8_t* data, size_t size, ObjectID object_ids[],
Status SendContainsRequest(int sock, ObjectID object_id) {
flatbuffers::FlatBufferBuilder fbb;
auto message = CreatePlasmaContainsRequest(fbb, fbb.CreateString(object_id.binary()));
- return PlasmaSend(sock, MessageType_PlasmaContainsRequest, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaContainsRequest, &fbb, message);
}
Status ReadContainsRequest(uint8_t* data, size_t size, ObjectID* object_id) {
@@ -331,7 +331,7 @@ Status SendContainsReply(int sock, ObjectID object_id, bool has_object) {
flatbuffers::FlatBufferBuilder fbb;
auto message =
CreatePlasmaContainsReply(fbb, fbb.CreateString(object_id.binary()), has_object);
- return PlasmaSend(sock, MessageType_PlasmaContainsReply, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaContainsReply, &fbb, message);
}
Status ReadContainsReply(uint8_t* data, size_t size, ObjectID* object_id,
@@ -349,7 +349,7 @@ Status ReadContainsReply(uint8_t* data, size_t size, ObjectID* object_id,
Status SendConnectRequest(int sock) {
flatbuffers::FlatBufferBuilder fbb;
auto message = CreatePlasmaConnectRequest(fbb);
- return PlasmaSend(sock, MessageType_PlasmaConnectRequest, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaConnectRequest, &fbb, message);
}
Status ReadConnectRequest(uint8_t* data) { return Status::OK(); }
@@ -357,7 +357,7 @@ Status ReadConnectRequest(uint8_t* data) { return Status::OK(); }
Status SendConnectReply(int sock, int64_t memory_capacity) {
flatbuffers::FlatBufferBuilder fbb;
auto message = CreatePlasmaConnectReply(fbb, memory_capacity);
- return PlasmaSend(sock, MessageType_PlasmaConnectReply, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaConnectReply, &fbb, message);
}
Status ReadConnectReply(uint8_t* data, size_t size, int64_t* memory_capacity) {
@@ -373,7 +373,7 @@ Status ReadConnectReply(uint8_t* data, size_t size, int64_t* memory_capacity) {
Status SendEvictRequest(int sock, int64_t num_bytes) {
flatbuffers::FlatBufferBuilder fbb;
auto message = CreatePlasmaEvictRequest(fbb, num_bytes);
- return PlasmaSend(sock, MessageType_PlasmaEvictRequest, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaEvictRequest, &fbb, message);
}
Status ReadEvictRequest(uint8_t* data, size_t size, int64_t* num_bytes) {
@@ -387,7 +387,7 @@ Status ReadEvictRequest(uint8_t* data, size_t size, int64_t* num_bytes) {
Status SendEvictReply(int sock, int64_t num_bytes) {
flatbuffers::FlatBufferBuilder fbb;
auto message = CreatePlasmaEvictReply(fbb, num_bytes);
- return PlasmaSend(sock, MessageType_PlasmaEvictReply, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaEvictReply, &fbb, message);
}
Status ReadEvictReply(uint8_t* data, size_t size, int64_t& num_bytes) {
@@ -405,7 +405,7 @@ Status SendGetRequest(int sock, const ObjectID* object_ids, int64_t num_objects,
flatbuffers::FlatBufferBuilder fbb;
auto message = CreatePlasmaGetRequest(fbb, to_flatbuffer(&fbb, object_ids, num_objects),
timeout_ms);
- return PlasmaSend(sock, MessageType_PlasmaGetRequest, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaGetRequest, &fbb, message);
}
Status ReadGetRequest(uint8_t* data, size_t size, std::vector<ObjectID>& object_ids,
@@ -447,7 +447,7 @@ Status SendGetReply(int sock, ObjectID object_ids[],
fbb, to_flatbuffer(&fbb, object_ids, num_objects),
fbb.CreateVectorOfStructs(objects.data(), num_objects), fbb.CreateVector(store_fds),
fbb.CreateVector(mmap_sizes), fbb.CreateVector(handles));
- return PlasmaSend(sock, MessageType_PlasmaGetReply, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaGetReply, &fbb, message);
}
Status ReadGetReply(uint8_t* data, size_t size, ObjectID object_ids[],
@@ -491,7 +491,7 @@ Status SendFetchRequest(int sock, const ObjectID* object_ids, int64_t num_object
flatbuffers::FlatBufferBuilder fbb;
auto message =
CreatePlasmaFetchRequest(fbb, to_flatbuffer(&fbb, object_ids, num_objects));
- return PlasmaSend(sock, MessageType_PlasmaFetchRequest, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaFetchRequest, &fbb, message);
}
Status ReadFetchRequest(uint8_t* data, size_t size, std::vector<ObjectID>& object_ids) {
@@ -514,12 +514,12 @@ Status SendWaitRequest(int sock, ObjectRequest object_requests[], int64_t num_re
for (int i = 0; i < num_requests; i++) {
object_request_specs.push_back(CreateObjectRequestSpec(
fbb, fbb.CreateString(object_requests[i].object_id.binary()),
- object_requests[i].type));
+ static_cast<int>(object_requests[i].type)));
}
auto message = CreatePlasmaWaitRequest(fbb, fbb.CreateVector(object_request_specs),
num_ready_objects, timeout_ms);
- return PlasmaSend(sock, MessageType_PlasmaWaitRequest, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaWaitRequest, &fbb, message);
}
Status ReadWaitRequest(uint8_t* data, size_t size, ObjectRequestMap& object_requests,
@@ -533,8 +533,10 @@ Status ReadWaitRequest(uint8_t* data, size_t size, ObjectRequestMap& object_requ
for (uoffset_t i = 0; i < message->object_requests()->size(); i++) {
ObjectID object_id =
ObjectID::from_binary(message->object_requests()->Get(i)->object_id()->str());
- ObjectRequest object_request({object_id, message->object_requests()->Get(i)->type(),
- ObjectStatus_Nonexistent});
+ ObjectRequest object_request(
+ {object_id,
+ static_cast<ObjectRequestType>(message->object_requests()->Get(i)->type()),
+ ObjectStatus::Nonexistent});
object_requests[object_id] = object_request;
}
return Status::OK();
@@ -553,7 +555,7 @@ Status SendWaitReply(int sock, const ObjectRequestMap& object_requests,
auto message = CreatePlasmaWaitReply(
fbb, fbb.CreateVector(object_replies.data(), num_ready_objects), num_ready_objects);
- return PlasmaSend(sock, MessageType_PlasmaWaitReply, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaWaitReply, &fbb, message);
}
Status ReadWaitReply(uint8_t* data, size_t size, ObjectRequest object_requests[],
@@ -576,7 +578,7 @@ Status ReadWaitReply(uint8_t* data, size_t size, ObjectRequest object_requests[]
Status SendSubscribeRequest(int sock) {
flatbuffers::FlatBufferBuilder fbb;
auto message = CreatePlasmaSubscribeRequest(fbb);
- return PlasmaSend(sock, MessageType_PlasmaSubscribeRequest, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaSubscribeRequest, &fbb, message);
}
// Data messages.
@@ -586,7 +588,7 @@ Status SendDataRequest(int sock, ObjectID object_id, const char* address, int po
auto addr = fbb.CreateString(address, strlen(address));
auto message =
CreatePlasmaDataRequest(fbb, fbb.CreateString(object_id.binary()), addr, port);
- return PlasmaSend(sock, MessageType_PlasmaDataRequest, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaDataRequest, &fbb, message);
}
Status ReadDataRequest(uint8_t* data, size_t size, ObjectID* object_id, char** address,
@@ -606,7 +608,7 @@ Status SendDataReply(int sock, ObjectID object_id, int64_t object_size,
flatbuffers::FlatBufferBuilder fbb;
auto message = CreatePlasmaDataReply(fbb, fbb.CreateString(object_id.binary()),
object_size, metadata_size);
- return PlasmaSend(sock, MessageType_PlasmaDataReply, &fbb, message);
+ return PlasmaSend(sock, MessageType::PlasmaDataReply, &fbb, message);
}
Status ReadDataReply(uint8_t* data, size_t size, ObjectID* object_id,
diff --git a/cpp/src/plasma/protocol.h b/cpp/src/plasma/protocol.h
index 7028132..2b477a8 100644
--- a/cpp/src/plasma/protocol.h
+++ b/cpp/src/plasma/protocol.h
@@ -38,7 +38,7 @@ bool verify_flatbuffer(T* object, uint8_t* data, size_t size) {
/* Plasma receive message. */
-Status PlasmaReceive(int sock, int64_t message_type, std::vector<uint8_t>* buffer);
+Status PlasmaReceive(int sock, MessageType message_type, std::vector<uint8_t>* buffer);
/* Plasma Create message functions. */
@@ -48,8 +48,8 @@ Status SendCreateRequest(int sock, ObjectID object_id, int64_t data_size,
Status ReadCreateRequest(uint8_t* data, size_t size, ObjectID* object_id,
int64_t* data_size, int64_t* metadata_size, int* device_num);
-Status SendCreateReply(int sock, ObjectID object_id, PlasmaObject* object, int error,
- int64_t mmap_size);
+Status SendCreateReply(int sock, ObjectID object_id, PlasmaObject* object,
+ PlasmaError error, int64_t mmap_size);
Status ReadCreateReply(uint8_t* data, size_t size, ObjectID* object_id,
PlasmaObject* object, int* store_fd, int64_t* mmap_size);
@@ -69,7 +69,7 @@ Status SendSealRequest(int sock, ObjectID object_id, unsigned char* digest);
Status ReadSealRequest(uint8_t* data, size_t size, ObjectID* object_id,
unsigned char* digest);
-Status SendSealReply(int sock, ObjectID object_id, int error);
+Status SendSealReply(int sock, ObjectID object_id, PlasmaError error);
Status ReadSealReply(uint8_t* data, size_t size, ObjectID* object_id);
@@ -96,7 +96,7 @@ Status SendReleaseRequest(int sock, ObjectID object_id);
Status ReadReleaseRequest(uint8_t* data, size_t size, ObjectID* object_id);
-Status SendReleaseReply(int sock, ObjectID object_id, int error);
+Status SendReleaseReply(int sock, ObjectID object_id, PlasmaError error);
Status ReadReleaseReply(uint8_t* data, size_t size, ObjectID* object_id);
@@ -106,7 +106,7 @@ Status SendDeleteRequest(int sock, ObjectID object_id);
Status ReadDeleteRequest(uint8_t* data, size_t size, ObjectID* object_id);
-Status SendDeleteReply(int sock, ObjectID object_id, int error);
+Status SendDeleteReply(int sock, ObjectID object_id, PlasmaError error);
Status ReadDeleteReply(uint8_t* data, size_t size, ObjectID* object_id);
diff --git a/cpp/src/plasma/store.cc b/cpp/src/plasma/store.cc
index 69a02dc..e86db21 100644
--- a/cpp/src/plasma/store.cc
+++ b/cpp/src/plasma/store.cc
@@ -144,14 +144,14 @@ void PlasmaStore::add_to_client_object_ids(ObjectTableEntry* entry, Client* clie
}
// Create a new object buffer in the hash table.
-int PlasmaStore::create_object(const ObjectID& object_id, int64_t data_size,
- int64_t metadata_size, int device_num, Client* client,
- PlasmaObject* result) {
+PlasmaError PlasmaStore::create_object(const ObjectID& object_id, int64_t data_size,
+ int64_t metadata_size, int device_num,
+ Client* client, PlasmaObject* result) {
ARROW_LOG(DEBUG) << "creating object " << object_id.hex();
if (store_info_.objects.count(object_id) != 0) {
// There is already an object with the same ID in the Plasma Store, so
// ignore this requst.
- return PlasmaError_ObjectExists;
+ return PlasmaError::ObjectExists;
}
// Try to evict objects until there is enough space.
uint8_t* pointer;
@@ -182,7 +182,7 @@ int PlasmaStore::create_object(const ObjectID& object_id, int64_t data_size,
// Return an error to the client if not enough space could be freed to
// create the object.
if (!success) {
- return PlasmaError_OutOfMemory;
+ return PlasmaError::OutOfMemory;
}
} else {
break;
@@ -211,7 +211,7 @@ int PlasmaStore::create_object(const ObjectID& object_id, int64_t data_size,
entry->fd = fd;
entry->map_size = map_size;
entry->offset = offset;
- entry->state = PLASMA_CREATED;
+ entry->state = object_state::PLASMA_CREATED;
entry->device_num = device_num;
#ifdef PLASMA_GPU
if (device_num != 0) {
@@ -232,13 +232,13 @@ int PlasmaStore::create_object(const ObjectID& object_id, int64_t data_size,
eviction_policy_.object_created(object_id);
// Record that this client is using this object.
add_to_client_object_ids(store_info_.objects[object_id].get(), client);
- return PlasmaError_OK;
+ return PlasmaError::OK;
}
void PlasmaObject_init(PlasmaObject* object, ObjectTableEntry* entry) {
DCHECK(object != NULL);
DCHECK(entry != NULL);
- DCHECK(entry->state == PLASMA_SEALED);
+ DCHECK(entry->state == object_state::PLASMA_SEALED);
#ifdef PLASMA_GPU
if (entry->device_num != 0) {
object->ipc_handle = entry->ipc_handle;
@@ -357,7 +357,7 @@ void PlasmaStore::process_get_request(Client* client,
// Check if this object is already present locally. If so, record that the
// object is being used and mark it as accounted for.
auto entry = get_object_table_entry(&store_info_, object_id);
- if (entry && entry->state == PLASMA_SEALED) {
+ if (entry && entry->state == object_state::PLASMA_SEALED) {
// Update the get request to take into account the present object.
PlasmaObject_init(&get_req->objects[object_id], entry);
get_req->num_satisfied += 1;
@@ -419,9 +419,11 @@ void PlasmaStore::release_object(const ObjectID& object_id, Client* client) {
}
// Check if an object is present.
-int PlasmaStore::contains_object(const ObjectID& object_id) {
+object_status PlasmaStore::contains_object(const ObjectID& object_id) {
auto entry = get_object_table_entry(&store_info_, object_id);
- return entry && (entry->state == PLASMA_SEALED) ? OBJECT_FOUND : OBJECT_NOT_FOUND;
+ return entry && (entry->state == object_state::PLASMA_SEALED)
+ ? object_status::OBJECT_FOUND
+ : object_status::OBJECT_NOT_FOUND;
}
// Seal an object that has been created in the hash table.
@@ -429,9 +431,9 @@ void PlasmaStore::seal_object(const ObjectID& object_id, unsigned char digest[])
ARROW_LOG(DEBUG) << "sealing object " << object_id.hex();
auto entry = get_object_table_entry(&store_info_, object_id);
ARROW_CHECK(entry != NULL);
- ARROW_CHECK(entry->state == PLASMA_CREATED);
+ ARROW_CHECK(entry->state == object_state::PLASMA_CREATED);
// Set the state of object to SEALED.
- entry->state = PLASMA_SEALED;
+ entry->state = object_state::PLASMA_SEALED;
// Set the object digest.
entry->info.digest = std::string(reinterpret_cast<char*>(&digest[0]), kDigestSize);
// Inform all subscribers that a new object has been sealed.
@@ -444,7 +446,7 @@ void PlasmaStore::seal_object(const ObjectID& object_id, unsigned char digest[])
int PlasmaStore::abort_object(const ObjectID& object_id, Client* client) {
auto entry = get_object_table_entry(&store_info_, object_id);
ARROW_CHECK(entry != NULL) << "To abort an object it must be in the object table.";
- ARROW_CHECK(entry->state != PLASMA_SEALED)
+ ARROW_CHECK(entry->state != object_state::PLASMA_SEALED)
<< "To abort an object it must not have been sealed.";
auto it = client->object_ids.find(object_id);
if (it == client->object_ids.end()) {
@@ -458,24 +460,24 @@ int PlasmaStore::abort_object(const ObjectID& object_id, Client* client) {
}
}
-int PlasmaStore::delete_object(ObjectID& object_id) {
+PlasmaError PlasmaStore::delete_object(ObjectID& object_id) {
auto entry = get_object_table_entry(&store_info_, object_id);
// TODO(rkn): This should probably not fail, but should instead throw an
// error. Maybe we should also support deleting objects that have been
// created but not sealed.
if (entry == NULL) {
// To delete an object it must be in the object table.
- return PlasmaError_ObjectNonexistent;
+ return PlasmaError::ObjectNonexistent;
}
- if (entry->state != PLASMA_SEALED) {
+ if (entry->state != object_state::PLASMA_SEALED) {
// To delete an object it must have been sealed.
- return PlasmaError_ObjectNotSealed;
+ return PlasmaError::ObjectNotSealed;
}
if (entry->ref_count != 0) {
// To delete an object, there must be no clients currently using it.
- return PlasmaError_ObjectInUse;
+ return PlasmaError::ObjectInUse;
}
eviction_policy_.remove_object(object_id);
@@ -487,7 +489,7 @@ int PlasmaStore::delete_object(ObjectID& object_id) {
notification.is_deletion = true;
push_notification(¬ification);
- return PlasmaError_OK;
+ return PlasmaError::OK;
}
void PlasmaStore::delete_objects(const std::vector<ObjectID>& object_ids) {
@@ -498,7 +500,7 @@ void PlasmaStore::delete_objects(const std::vector<ObjectID>& object_ids) {
// error. Maybe we should also support deleting objects that have been
// created but not sealed.
ARROW_CHECK(entry != NULL) << "To delete an object it must be in the object table.";
- ARROW_CHECK(entry->state == PLASMA_SEALED)
+ ARROW_CHECK(entry->state == object_state::PLASMA_SEALED)
<< "To delete an object it must have been sealed.";
ARROW_CHECK(entry->ref_count == 0)
<< "To delete an object, there must be no clients currently using it.";
@@ -545,7 +547,7 @@ void PlasmaStore::disconnect_client(int client_fd) {
continue;
}
- if (it->second->state == PLASMA_SEALED) {
+ if (it->second->state == object_state::PLASMA_SEALED) {
// Add sealed objects to a temporary list of object IDs. Do not perform
// the remove here, since it potentially modifies the object_ids table.
sealed_objects.push_back(it->second.get());
@@ -683,14 +685,14 @@ void PlasmaStore::subscribe_to_updates(Client* client) {
// Push notifications to the new subscriber about existing sealed objects.
for (const auto& entry : store_info_.objects) {
- if (entry.second->state == PLASMA_SEALED) {
+ if (entry.second->state == object_state::PLASMA_SEALED) {
push_notification(&entry.second->info, fd);
}
}
}
Status PlasmaStore::process_message(Client* client) {
- int64_t type;
+ MessageType type;
Status s = ReadMessage(client->fd, &type, &input_buffer_);
ARROW_CHECK(s.ok() || s.IsIOError());
@@ -703,61 +705,61 @@ Status PlasmaStore::process_message(Client* client) {
// Process the different types of requests.
switch (type) {
- case MessageType_PlasmaCreateRequest: {
+ case MessageType::PlasmaCreateRequest: {
int64_t data_size;
int64_t metadata_size;
int device_num;
RETURN_NOT_OK(ReadCreateRequest(input, input_size, &object_id, &data_size,
&metadata_size, &device_num));
- int error_code =
+ PlasmaError error_code =
create_object(object_id, data_size, metadata_size, device_num, client, &object);
int64_t mmap_size = 0;
- if (error_code == PlasmaError_OK && device_num == 0) {
+ if (error_code == PlasmaError::OK && device_num == 0) {
mmap_size = get_mmap_size(object.store_fd);
}
HANDLE_SIGPIPE(
SendCreateReply(client->fd, object_id, &object, error_code, mmap_size),
client->fd);
- if (error_code == PlasmaError_OK && device_num == 0) {
+ if (error_code == PlasmaError::OK && device_num == 0) {
warn_if_sigpipe(send_fd(client->fd, object.store_fd), client->fd);
}
} break;
- case MessageType_PlasmaAbortRequest: {
+ case MessageType::PlasmaAbortRequest: {
RETURN_NOT_OK(ReadAbortRequest(input, input_size, &object_id));
ARROW_CHECK(abort_object(object_id, client) == 1) << "To abort an object, the only "
"client currently using it "
"must be the creator.";
HANDLE_SIGPIPE(SendAbortReply(client->fd, object_id), client->fd);
} break;
- case MessageType_PlasmaGetRequest: {
+ case 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));
process_get_request(client, object_ids_to_get, timeout_ms);
} break;
- case MessageType_PlasmaReleaseRequest: {
+ case MessageType::PlasmaReleaseRequest: {
RETURN_NOT_OK(ReadReleaseRequest(input, input_size, &object_id));
release_object(object_id, client);
} break;
- case MessageType_PlasmaDeleteRequest: {
+ case MessageType::PlasmaDeleteRequest: {
RETURN_NOT_OK(ReadDeleteRequest(input, input_size, &object_id));
- int error_code = delete_object(object_id);
+ PlasmaError error_code = delete_object(object_id);
HANDLE_SIGPIPE(SendDeleteReply(client->fd, object_id, error_code), client->fd);
} break;
- case MessageType_PlasmaContainsRequest: {
+ case MessageType::PlasmaContainsRequest: {
RETURN_NOT_OK(ReadContainsRequest(input, input_size, &object_id));
- if (contains_object(object_id) == OBJECT_FOUND) {
+ if (contains_object(object_id) == object_status::OBJECT_FOUND) {
HANDLE_SIGPIPE(SendContainsReply(client->fd, object_id, 1), client->fd);
} else {
HANDLE_SIGPIPE(SendContainsReply(client->fd, object_id, 0), client->fd);
}
} break;
- case MessageType_PlasmaSealRequest: {
+ case MessageType::PlasmaSealRequest: {
unsigned char digest[kDigestSize];
RETURN_NOT_OK(ReadSealRequest(input, input_size, &object_id, &digest[0]));
seal_object(object_id, &digest[0]);
} break;
- case MessageType_PlasmaEvictRequest: {
+ case MessageType::PlasmaEvictRequest: {
// This code path should only be used for testing.
int64_t num_bytes;
RETURN_NOT_OK(ReadEvictRequest(input, input_size, &num_bytes));
@@ -767,14 +769,14 @@ Status PlasmaStore::process_message(Client* client) {
delete_objects(objects_to_evict);
HANDLE_SIGPIPE(SendEvictReply(client->fd, num_bytes_evicted), client->fd);
} break;
- case MessageType_PlasmaSubscribeRequest:
+ case MessageType::PlasmaSubscribeRequest:
subscribe_to_updates(client);
break;
- case MessageType_PlasmaConnectRequest: {
+ case MessageType::PlasmaConnectRequest: {
HANDLE_SIGPIPE(SendConnectReply(client->fd, store_info_.memory_capacity),
client->fd);
} break;
- case MessageType_PlasmaDisconnectClient:
+ case MessageType::PlasmaDisconnectClient:
ARROW_LOG(DEBUG) << "Disconnecting client on fd " << client->fd;
disconnect_client(client->fd);
break;
diff --git a/cpp/src/plasma/store.h b/cpp/src/plasma/store.h
index e5ef917..e40f040 100644
--- a/cpp/src/plasma/store.h
+++ b/cpp/src/plasma/store.h
@@ -83,15 +83,16 @@ class PlasmaStore {
/// @param client The client that created the object.
/// @param result The object that has been created.
/// @return One of the following error codes:
- /// - PlasmaError_OK, if the object was created successfully.
- /// - PlasmaError_ObjectExists, if an object with this ID is already
+ /// - PlasmaError::OK, if the object was created successfully.
+ /// - PlasmaError::ObjectExists, if an object with this ID is already
/// present in the store. In this case, the client should not call
/// plasma_release.
- /// - PlasmaError_OutOfMemory, if the store is out of memory and
+ /// - PlasmaError::OutOfMemory, if the store is out of memory and
/// cannot create the object. In this case, the client should not call
/// plasma_release.
- int create_object(const ObjectID& object_id, int64_t data_size, int64_t metadata_size,
- int device_num, Client* client, PlasmaObject* result);
+ PlasmaError create_object(const ObjectID& object_id, int64_t data_size,
+ int64_t metadata_size, int device_num, Client* client,
+ PlasmaObject* result);
/// Abort a created but unsealed object. If the client is not the
/// creator, then the abort will fail.
@@ -106,10 +107,10 @@ class PlasmaStore {
///
/// @param object_id Object ID of the object to be deleted.
/// @return One of the following error codes:
- /// - PlasmaError_OK, if the object was delete successfully.
- /// - PlasmaError_ObjectNonexistent, if ths object isn't existed.
- /// - PlasmaError_ObjectInUse, if the object is in use.
- int delete_object(ObjectID& object_id);
+ /// - PlasmaError::OK, if the object was delete successfully.
+ /// - PlasmaError::ObjectNonexistent, if ths object isn't existed.
+ /// - PlasmaError::ObjectInUse, if the object is in use.
+ PlasmaError delete_object(ObjectID& object_id);
/// Delete objects that have been created in the hash table. This should only
/// be called on objects that are returned by the eviction policy to evict.
@@ -144,7 +145,7 @@ class PlasmaStore {
/// @param object_id Object ID that will be checked.
/// @return OBJECT_FOUND if the object is in the store, OBJECT_NOT_FOUND if
/// not
- int contains_object(const ObjectID& object_id);
+ object_status contains_object(const ObjectID& object_id);
/// Record the fact that a particular client is no longer using an object.
///
diff --git a/cpp/src/plasma/test/serialization_tests.cc b/cpp/src/plasma/test/serialization_tests.cc
index 006fa74..e05c1b4 100644
--- a/cpp/src/plasma/test/serialization_tests.cc
+++ b/cpp/src/plasma/test/serialization_tests.cc
@@ -48,10 +48,10 @@ int create_temp_file(void) {
* @return Pointer to the content of the message. Needs to be freed by the
* caller.
*/
-std::vector<uint8_t> read_message_from_file(int fd, int message_type) {
+std::vector<uint8_t> read_message_from_file(int fd, MessageType message_type) {
/* Go to the beginning of the file. */
lseek(fd, 0, SEEK_SET);
- int64_t type;
+ MessageType type;
std::vector<uint8_t> data;
ARROW_CHECK_OK(ReadMessage(fd, &type, &data));
ARROW_CHECK(type == message_type);
@@ -80,7 +80,8 @@ TEST(PlasmaSerialization, CreateRequest) {
int device_num1 = 0;
ARROW_CHECK_OK(
SendCreateRequest(fd, object_id1, data_size1, metadata_size1, device_num1));
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaCreateRequest);
+ std::vector<uint8_t> data =
+ read_message_from_file(fd, MessageType::PlasmaCreateRequest);
ObjectID object_id2;
int64_t data_size2;
int64_t metadata_size2;
@@ -99,8 +100,8 @@ TEST(PlasmaSerialization, CreateReply) {
ObjectID object_id1 = ObjectID::from_random();
PlasmaObject object1 = random_plasma_object();
int64_t mmap_size1 = 1000000;
- ARROW_CHECK_OK(SendCreateReply(fd, object_id1, &object1, 0, mmap_size1));
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaCreateReply);
+ ARROW_CHECK_OK(SendCreateReply(fd, object_id1, &object1, PlasmaError::OK, mmap_size1));
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaCreateReply);
ObjectID object_id2;
PlasmaObject object2;
memset(&object2, 0, sizeof(object2));
@@ -121,7 +122,7 @@ TEST(PlasmaSerialization, SealRequest) {
unsigned char digest1[kDigestSize];
memset(&digest1[0], 7, kDigestSize);
ARROW_CHECK_OK(SendSealRequest(fd, object_id1, &digest1[0]));
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaSealRequest);
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaSealRequest);
ObjectID object_id2;
unsigned char digest2[kDigestSize];
ARROW_CHECK_OK(ReadSealRequest(data.data(), data.size(), &object_id2, &digest2[0]));
@@ -133,8 +134,8 @@ TEST(PlasmaSerialization, SealRequest) {
TEST(PlasmaSerialization, SealReply) {
int fd = create_temp_file();
ObjectID object_id1 = ObjectID::from_random();
- ARROW_CHECK_OK(SendSealReply(fd, object_id1, PlasmaError_ObjectExists));
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaSealReply);
+ ARROW_CHECK_OK(SendSealReply(fd, object_id1, PlasmaError::ObjectExists));
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaSealReply);
ObjectID object_id2;
Status s = ReadSealReply(data.data(), data.size(), &object_id2);
ASSERT_EQ(object_id1, object_id2);
@@ -149,7 +150,7 @@ TEST(PlasmaSerialization, GetRequest) {
object_ids[1] = ObjectID::from_random();
int64_t timeout_ms = 1234;
ARROW_CHECK_OK(SendGetRequest(fd, object_ids, 2, timeout_ms));
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaGetRequest);
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaGetRequest);
std::vector<ObjectID> object_ids_return;
int64_t timeout_ms_return;
ARROW_CHECK_OK(
@@ -172,7 +173,7 @@ TEST(PlasmaSerialization, GetReply) {
std::vector<int64_t> mmap_sizes = {100, 200, 300};
ARROW_CHECK_OK(SendGetReply(fd, object_ids, plasma_objects, 2, store_fds, mmap_sizes));
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaGetReply);
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaGetReply);
ObjectID object_ids_return[2];
PlasmaObject plasma_objects_return[2];
std::vector<int> store_fds_return;
@@ -200,7 +201,7 @@ TEST(PlasmaSerialization, ReleaseRequest) {
ObjectID object_id1 = ObjectID::from_random();
ARROW_CHECK_OK(SendReleaseRequest(fd, object_id1));
std::vector<uint8_t> data =
- read_message_from_file(fd, MessageType_PlasmaReleaseRequest);
+ read_message_from_file(fd, MessageType::PlasmaReleaseRequest);
ObjectID object_id2;
ARROW_CHECK_OK(ReadReleaseRequest(data.data(), data.size(), &object_id2));
ASSERT_EQ(object_id1, object_id2);
@@ -210,8 +211,8 @@ TEST(PlasmaSerialization, ReleaseRequest) {
TEST(PlasmaSerialization, ReleaseReply) {
int fd = create_temp_file();
ObjectID object_id1 = ObjectID::from_random();
- ARROW_CHECK_OK(SendReleaseReply(fd, object_id1, PlasmaError_ObjectExists));
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaReleaseReply);
+ ARROW_CHECK_OK(SendReleaseReply(fd, object_id1, PlasmaError::ObjectExists));
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaReleaseReply);
ObjectID object_id2;
Status s = ReadReleaseReply(data.data(), data.size(), &object_id2);
ASSERT_EQ(object_id1, object_id2);
@@ -223,7 +224,8 @@ TEST(PlasmaSerialization, DeleteRequest) {
int fd = create_temp_file();
ObjectID object_id1 = ObjectID::from_random();
ARROW_CHECK_OK(SendDeleteRequest(fd, object_id1));
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaDeleteRequest);
+ std::vector<uint8_t> data =
+ read_message_from_file(fd, MessageType::PlasmaDeleteRequest);
ObjectID object_id2;
ARROW_CHECK_OK(ReadDeleteRequest(data.data(), data.size(), &object_id2));
ASSERT_EQ(object_id1, object_id2);
@@ -233,9 +235,9 @@ TEST(PlasmaSerialization, DeleteRequest) {
TEST(PlasmaSerialization, DeleteReply) {
int fd = create_temp_file();
ObjectID object_id1 = ObjectID::from_random();
- int error1 = PlasmaError_ObjectExists;
+ PlasmaError error1 = PlasmaError::ObjectExists;
ARROW_CHECK_OK(SendDeleteReply(fd, object_id1, error1));
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaDeleteReply);
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaDeleteReply);
ObjectID object_id2;
Status s = ReadDeleteReply(data.data(), data.size(), &object_id2);
ASSERT_EQ(object_id1, object_id2);
@@ -250,7 +252,8 @@ TEST(PlasmaSerialization, StatusRequest) {
object_ids[0] = ObjectID::from_random();
object_ids[1] = ObjectID::from_random();
ARROW_CHECK_OK(SendStatusRequest(fd, object_ids, num_objects));
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaStatusRequest);
+ std::vector<uint8_t> data =
+ read_message_from_file(fd, MessageType::PlasmaStatusRequest);
ObjectID object_ids_read[num_objects];
ARROW_CHECK_OK(
ReadStatusRequest(data.data(), data.size(), object_ids_read, num_objects));
@@ -266,7 +269,7 @@ TEST(PlasmaSerialization, StatusReply) {
object_ids[1] = ObjectID::from_random();
int object_statuses[2] = {42, 43};
ARROW_CHECK_OK(SendStatusReply(fd, object_ids, object_statuses, 2));
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaStatusReply);
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaStatusReply);
int64_t num_objects = ReadStatusReply_num_objects(data.data(), data.size());
std::vector<ObjectID> object_ids_read(num_objects);
@@ -284,7 +287,7 @@ TEST(PlasmaSerialization, EvictRequest) {
int fd = create_temp_file();
int64_t num_bytes = 111;
ARROW_CHECK_OK(SendEvictRequest(fd, num_bytes));
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaEvictRequest);
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaEvictRequest);
int64_t num_bytes_received;
ARROW_CHECK_OK(ReadEvictRequest(data.data(), data.size(), &num_bytes_received));
ASSERT_EQ(num_bytes, num_bytes_received);
@@ -295,7 +298,7 @@ TEST(PlasmaSerialization, EvictReply) {
int fd = create_temp_file();
int64_t num_bytes = 111;
ARROW_CHECK_OK(SendEvictReply(fd, num_bytes));
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaEvictReply);
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaEvictReply);
int64_t num_bytes_received;
ARROW_CHECK_OK(ReadEvictReply(data.data(), data.size(), num_bytes_received));
ASSERT_EQ(num_bytes, num_bytes_received);
@@ -308,7 +311,7 @@ TEST(PlasmaSerialization, FetchRequest) {
object_ids[0] = ObjectID::from_random();
object_ids[1] = ObjectID::from_random();
ARROW_CHECK_OK(SendFetchRequest(fd, object_ids, 2));
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaFetchRequest);
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaFetchRequest);
std::vector<ObjectID> object_ids_read;
ARROW_CHECK_OK(ReadFetchRequest(data.data(), data.size(), object_ids_read));
ASSERT_EQ(object_ids[0], object_ids_read[0]);
@@ -320,15 +323,17 @@ TEST(PlasmaSerialization, WaitRequest) {
int fd = create_temp_file();
const int num_objects_in = 2;
ObjectRequest object_requests_in[num_objects_in] = {
- ObjectRequest({ObjectID::from_random(), PLASMA_QUERY_ANYWHERE, 0}),
- ObjectRequest({ObjectID::from_random(), PLASMA_QUERY_LOCAL, 0})};
+ ObjectRequest({ObjectID::from_random(), ObjectRequestType::PLASMA_QUERY_ANYWHERE,
+ ObjectStatus::Local}),
+ ObjectRequest({ObjectID::from_random(), ObjectRequestType::PLASMA_QUERY_LOCAL,
+ ObjectStatus::Local})};
const int num_ready_objects_in = 1;
int64_t timeout_ms = 1000;
ARROW_CHECK_OK(SendWaitRequest(fd, &object_requests_in[0], num_objects_in,
num_ready_objects_in, timeout_ms));
/* Read message back. */
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaWaitRequest);
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaWaitRequest);
int num_ready_objects_out;
int64_t timeout_ms_read;
ObjectRequestMap object_requests_out;
@@ -353,13 +358,15 @@ TEST(PlasmaSerialization, WaitReply) {
/* Create a map with two ObjectRequests in it. */
ObjectRequestMap objects_in(num_objects_in);
ObjectID id1 = ObjectID::from_random();
- objects_in[id1] = ObjectRequest({id1, 0, ObjectStatus_Local});
+ objects_in[id1] =
+ ObjectRequest({id1, ObjectRequestType::PLASMA_QUERY_LOCAL, ObjectStatus::Local});
ObjectID id2 = ObjectID::from_random();
- objects_in[id2] = ObjectRequest({id2, 0, ObjectStatus_Nonexistent});
+ objects_in[id2] = ObjectRequest(
+ {id2, ObjectRequestType::PLASMA_QUERY_LOCAL, ObjectStatus::Nonexistent});
ARROW_CHECK_OK(SendWaitReply(fd, objects_in, num_objects_in));
/* Read message back. */
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaWaitReply);
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaWaitReply);
ObjectRequest objects_out[2];
int num_objects_out;
ARROW_CHECK_OK(
@@ -383,7 +390,7 @@ TEST(PlasmaSerialization, DataRequest) {
int port1 = 12345;
ARROW_CHECK_OK(SendDataRequest(fd, object_id1, address1, port1));
/* Reading message back. */
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaDataRequest);
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaDataRequest);
ObjectID object_id2;
char* address2;
int port2;
@@ -403,7 +410,7 @@ TEST(PlasmaSerialization, DataReply) {
int64_t metadata_size1 = 198;
ARROW_CHECK_OK(SendDataReply(fd, object_id1, object_size1, metadata_size1));
/* Reading message back. */
- std::vector<uint8_t> data = read_message_from_file(fd, MessageType_PlasmaDataReply);
+ std::vector<uint8_t> data = read_message_from_file(fd, MessageType::PlasmaDataReply);
ObjectID object_id2;
int64_t object_size2;
int64_t metadata_size2;
diff --git a/python/pyarrow/_plasma.pyx b/python/pyarrow/_plasma.pyx
index 8845b87..1e2cb66 100644
--- a/python/pyarrow/_plasma.pyx
+++ b/python/pyarrow/_plasma.pyx
@@ -63,8 +63,8 @@ cdef extern from "plasma/common.h":
cdef int64_t kDigestSize" plasma::kDigestSize"
cdef enum ObjectRequestType:
- PLASMA_QUERY_LOCAL"plasma::PLASMA_QUERY_LOCAL",
- PLASMA_QUERY_ANYWHERE"plasma::PLASMA_QUERY_ANYWHERE"
+ PLASMA_QUERY_LOCAL"plasma::ObjectRequestType::PLASMA_QUERY_LOCAL",
+ PLASMA_QUERY_ANYWHERE"plasma::ObjectRequestType::PLASMA_QUERY_ANYWHERE"
cdef int ObjectStatusLocal"plasma::ObjectStatusLocal"
cdef int ObjectStatusRemote"plasma::ObjectStatusRemote"