mirrors/arrow
0
0 Fork 0
mirror of https://github.com/apache/arrow.git synced 2026-10-28 16:47:19 +08:00
Code Activity

ARROW-4015: [Plasma] remove unused interfaces for plasma manager

Browse Source
This commit is contained in:
Zhijun Fu committed 2018-12-13 14:55:10 +08:00
1 parent 45940410e6
commit 583cd97c40
11 files changed
+3 -782
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

cpp/src/plasma/client.cc
+1 -110
View File
@@ -210,17 +210,6 @@ class PlasmaClient::Impl : public std::enable_shared_from_this<PlasmaClient::Imp
Status Disconnect();
Status Fetch(int num_object_ids, const ObjectID* object_ids);
Status Wait(int64_t num_object_requests, ObjectRequest* object_requests,
int num_ready_objects, int64_t timeout_ms, int* num_objects_ready);
Status Transfer(const char* addr, int port, const ObjectID& object_id);
Status Info(const ObjectID& object_id, int* object_status);
int get_manager_fd() const;
Status FlushReleaseHistory();
bool IsInUse(const ObjectID& object_id);
@@ -258,8 +247,6 @@ class PlasmaClient::Impl : public std::enable_shared_from_this<PlasmaClient::Imp
/// File descriptor of the Unix domain socket that connects to the store.
int store_conn_;
/// File descriptor of the Unix domain socket that connects to the manager.
int manager_conn_;
/// Table of dlmalloc buffer files that have been memory mapped so far. This
/// is a hash table mapping a file descriptor to a struct containing the
/// address of the corresponding memory-mapped file.
@@ -976,10 +963,7 @@ Status PlasmaClient::Impl::Connect(const std::string& store_socket_name,
int release_delay, int num_retries) {
RETURN_NOT_OK(ConnectIpcSocketRetry(store_socket_name, num_retries, -1, &store_conn_));
if (manager_socket_name != "") {
RETURN_NOT_OK(
ConnectIpcSocketRetry(manager_socket_name, num_retries, -1, &manager_conn_));
} else {
manager_conn_ = -1;
return Status::NotImplemented("plasma manager is no longer supported");
}
config_.release_delay = release_delay;
in_use_object_bytes_ = 0;
@@ -1000,78 +984,6 @@ Status PlasmaClient::Impl::Disconnect() {
// that were in use by us when handling the SIGPIPE.
close(store_conn_);
store_conn_ = -1;
if (manager_conn_ >= 0) {
close(manager_conn_);
manager_conn_ = -1;
}
return Status::OK();
}
Status PlasmaClient::Impl::Transfer(const char* address, int port,
const ObjectID& object_id) {
return SendDataRequest(manager_conn_, object_id, address, port);
}
Status PlasmaClient::Impl::Fetch(int num_object_ids, const ObjectID* object_ids) {
ARROW_CHECK(manager_conn_ >= 0);
return SendFetchRequest(manager_conn_, object_ids, num_object_ids);
}
int PlasmaClient::Impl::get_manager_fd() const { return manager_conn_; }
Status PlasmaClient::Impl::Info(const ObjectID& object_id, int* object_status) {
ARROW_CHECK(manager_conn_ >= 0);
RETURN_NOT_OK(SendStatusRequest(manager_conn_, &object_id, 1));
std::vector<uint8_t> 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);
return Status::OK();
}
Status PlasmaClient::Impl::Wait(int64_t num_object_requests,
ObjectRequest* object_requests, int num_ready_objects,
int64_t timeout_ms, int* num_objects_ready) {
ARROW_CHECK(manager_conn_ >= 0);
ARROW_CHECK(num_object_requests > 0);
ARROW_CHECK(num_ready_objects > 0);
ARROW_CHECK(num_ready_objects <= num_object_requests);
for (int i = 0; i < num_object_requests; ++i) {
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(
ReadWaitReply(buffer.data(), buffer.size(), object_requests, &num_ready_objects));
*num_objects_ready = 0;
for (int i = 0; i < num_object_requests; ++i) {
ObjectRequestType type = object_requests[i].type;
auto status = static_cast<fb::ObjectStatus>(object_requests[i].location);
switch (type) {
case ObjectRequestType::PLASMA_QUERY_LOCAL:
if (status == fb::ObjectStatus::Local) {
*num_objects_ready += 1;
}
break;
case ObjectRequestType::PLASMA_QUERY_ANYWHERE:
if (status == fb::ObjectStatus::Local || status == fb::ObjectStatus::Remote) {
*num_objects_ready += 1;
} else {
ARROW_CHECK(status == fb::ObjectStatus::Nonexistent);
}
break;
default:
ARROW_LOG(FATAL) << "This code should be unreachable.";
}
}
return Status::OK();
}
@@ -1154,27 +1066,6 @@ Status PlasmaClient::DecodeNotification(const uint8_t* buffer, ObjectID* object_
Status PlasmaClient::Disconnect() { return impl_->Disconnect(); }
Status PlasmaClient::Fetch(int num_object_ids, const ObjectID* object_ids) {
return impl_->Fetch(num_object_ids, object_ids);
}
Status PlasmaClient::Wait(int64_t num_object_requests, ObjectRequest* object_requests,
int num_ready_objects, int64_t timeout_ms,
int* num_objects_ready) {
return impl_->Wait(num_object_requests, object_requests, num_ready_objects, timeout_ms,
num_objects_ready);
}
Status PlasmaClient::Transfer(const char* addr, int port, const ObjectID& object_id) {
return impl_->Transfer(addr, port, object_id);
}
Status PlasmaClient::Info(const ObjectID& object_id, int* object_status) {
return impl_->Info(object_id, object_status);
}
int PlasmaClient::get_manager_fd() const { return impl_->get_manager_fd(); }
Status PlasmaClient::FlushReleaseHistory() { return impl_->FlushReleaseHistory(); }
bool PlasmaClient::IsInUse(const ObjectID& object_id) {
cpp/src/plasma/client.h
+2 -93
View File
@@ -62,6 +62,8 @@ class ARROW_EXPORT PlasmaClient {
/// \param manager_socket_name The name of the UNIX domain socket to use to
/// connect to the local Plasma manager. If this is "", then this
/// function will not connect to a manager.
/// Note that plasma manager is no longer supported, this function
/// will return failure if this is not "".
/// \param release_delay Number of released objects that are kept around
/// and not evicted to avoid too many munmaps.
/// \param num_retries number of attempts to connect to IPC socket, default 50
@@ -255,99 +257,6 @@ class ARROW_EXPORT PlasmaClient {
/// \return The return status.
Status Disconnect();
/// Attempt to initiate the transfer of some objects from remote Plasma
/// Stores.
/// This method does not guarantee that the fetched objects will arrive
/// locally.
///
/// For an object that is available in the local Plasma Store, this method
/// will
/// not do anything. For an object that is not available locally, it will
/// check
/// if the object are already being fetched. If so, it will not do anything.
/// If
/// not, it will query the object table for a list of Plasma Managers that
/// have
/// the object. The object table will return a non-empty list, and this Plasma
/// Manager will attempt to initiate transfers from one of those Plasma
/// Managers.
///
/// This function is non-blocking.
///
/// This method is idempotent in the sense that it is ok to call it multiple
/// times.
///
/// \param num_object_ids The number of object IDs fetch is being called on.
/// \param object_ids The IDs of the objects that fetch is being called on.
/// \return The return status.
Status Fetch(int num_object_ids, const ObjectID* object_ids);
/// Wait for (1) a specified number of objects to be available (sealed) in the
/// local Plasma Store or in a remote Plasma Store, or (2) for a timeout to
/// expire. This is a blocking call.
///
/// \param num_object_requests Size of the object_requests array.
/// \param object_requests Object event array. Each element contains a request
/// for a particular object_id. The type of request is specified in the
/// "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
/// status
/// 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.
/// \param num_ready_objects The number of requests in object_requests array
/// that
/// must be satisfied before the function returns, unless it timeouts.
/// The num_ready_objects should be no larger than num_object_requests.
/// \param timeout_ms Timeout value in milliseconds. If this timeout expires
/// before min_num_ready_objects of requests are satisfied, the
/// function
/// returns.
/// \param num_objects_ready Out parameter for number of satisfied requests in
/// the object_requests list. If the returned number is less than
/// min_num_ready_objects this means that timeout expired.
/// \return The return status.
Status Wait(int64_t num_object_requests, ObjectRequest* object_requests,
int num_ready_objects, int64_t timeout_ms, int* num_objects_ready);
/// Transfer local object to a different plasma manager.
///
/// \param addr IP address of the plasma manager we are transfering to.
/// \param port Port of the plasma manager we are transfering to.
/// \param object_id ObjectID of the object we are transfering.
/// \return The return status.
Status Transfer(const char* addr, int port, const ObjectID& object_id);
/// Return the status of a given object. This method may query the object
/// table.
///
/// \param object_id The ID of the object whose status we query.
/// \param object_status Out parameter for object status. Can take the
/// following values.
/// - PLASMA_CLIENT_LOCAL, if object is stored in the local Plasma
/// Store.
/// has been already scheduled by the Plasma Manager.
/// - PLASMA_CLIENT_TRANSFER, if the object is either currently being
/// transferred or just scheduled.
/// - PLASMA_CLIENT_REMOTE, if the object is stored at a remote
/// Plasma Store.
/// - PLASMA_CLIENT_DOES_NOT_EXIST, if the object doesn’t exist in the
/// system.
/// \return The return status.
Status Info(const ObjectID& object_id, int* object_status);
/// Get the file descriptor for the socket connection to the plasma manager.
///
/// \return The file descriptor for the manager connection. If there is no
/// connection to the manager, this is -1.
int get_manager_fd() const;
private:
friend class PlasmaBuffer;
FRIEND_TEST(TestPlasmaStore, GetTest);
cpp/src/plasma/common.cc
-3
View File
@@ -107,9 +107,6 @@ bool UniqueID::operator==(const UniqueID& rhs) const {
return std::memcmp(data(), rhs.data(), kUniqueIDSize) == 0;
}
ARROW_EXPORT fb::ObjectStatus ObjectStatusLocal = fb::ObjectStatus::Local;
ARROW_EXPORT fb::ObjectStatus ObjectStatusRemote = fb::ObjectStatus::Remote;
const PlasmaStoreInfo* plasma_config;
} // namespace plasma
cpp/src/plasma/common.h
-24
View File
@@ -66,30 +66,6 @@ typedef UniqueID ObjectID;
/// 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.
ObjectID object_id;
/// Request associated to the object. It can take one of the following values:
/// - PLASMA_QUERY_LOCAL: return if or when the object is available in the
/// 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).
ObjectRequestType type;
/// Object location. This can be
/// - ObjectLocation::Local: object is ready at the local Plasma Store.
/// - ObjectLocation::Remote: object is ready at a remote Plasma Store.
/// - ObjectLocation::Nonexistent: object does not exist in the system.
ObjectLocation location;
};
enum class ObjectState : int {
/// Object was created but not sealed in the local Plasma Store.
PLASMA_CREATED = 1,
cpp/src/plasma/format/plasma.fbs
-70
View File
@@ -42,9 +42,6 @@ enum MessageType:long {
// Delete an object.
PlasmaDeleteRequest,
PlasmaDeleteReply,
// Get status of an object.
PlasmaStatusRequest,
PlasmaStatusReply,
// See if the store contains an object (will be deprecated).
PlasmaContainsRequest,
PlasmaContainsReply,
@@ -57,11 +54,6 @@ enum MessageType:long {
// Make room for new objects in the plasma store.
PlasmaEvictRequest,
PlasmaEvictReply,
// Fetch objects from remote Plasma stores.
PlasmaFetchRequest,
// Wait for objects to be ready either from local or remote Plasma stores.
PlasmaWaitRequest,
PlasmaWaitReply,
// Subscribe to a list of objects or to all objects.
PlasmaSubscribeRequest,
// Unsubscribe.
@@ -239,31 +231,6 @@ table PlasmaDeleteReply {
errors: [PlasmaError];
}
table PlasmaStatusRequest {
// IDs of the objects stored at local Plasma store we request the status of.
object_ids: [string];
}
enum ObjectStatus:int {
// Object is stored in the local Plasma Store.
Local,
// Object is stored on a remote Plasma store, and it is not stored on the
// local Plasma Store.
Remote,
// Object is not stored in the system.
Nonexistent,
// Object is currently transferred from a remote Plasma store the local
// Plasma Store.
Transfer
}
table PlasmaStatusReply {
// IDs of the objects being returned.
object_ids: [string];
// Status of the object.
status: [ObjectStatus];
}
// PlasmaContains is a subset of PlasmaStatus which does not
// involve the plasma manager, only the store. We should consider
// unifying them in the future and deprecating PlasmaContains.
@@ -309,43 +276,6 @@ table PlasmaEvictReply {
num_bytes: ulong;
}
table PlasmaFetchRequest {
// IDs of objects to be gotten.
object_ids: [string];
}
table ObjectRequestSpec {
// ID of the object.
object_id: string;
// The type of the object. This specifies whether we
// will be waiting for an object store in the local or
// global Plasma store.
type: int;
}
table PlasmaWaitRequest {
// Array of object requests whose status we are asking for.
object_requests: [ObjectRequestSpec];
// Number of objects expected to be returned, if available.
num_ready_objects: int;
// timeout
timeout: long;
}
table ObjectReply {
// ID of the object.
object_id: string;
// The object status. This specifies where the object is stored.
status: ObjectStatus;
}
table PlasmaWaitReply {
// Array of object requests being returned.
object_requests: [ObjectReply];
// Number of objects expected to be returned, if available.
num_ready_objects: int;
}
table PlasmaSubscribeRequest {
}
cpp/src/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc
-73
View File
@@ -220,79 +220,6 @@ JNIEXPORT jboolean JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_contains
return has_object;
}
JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_fetch(
JNIEnv* env, jclass cls, jlong conn, jobjectArray object_ids) {
plasma::PlasmaClient* client = reinterpret_cast<plasma::PlasmaClient*>(conn);
jsize num_oids = env->GetArrayLength(object_ids);
std::vector<plasma::ObjectID> oids(num_oids);
for (int i = 0; i < num_oids; ++i) {
jbyteArray_to_object_id(
env, reinterpret_cast<jbyteArray>(env->GetObjectArrayElement(object_ids, i)),
&oids[i]);
}
ARROW_CHECK_OK(client->Fetch(static_cast<int>(num_oids), oids.data()));
return;
}
JNIEXPORT jobjectArray JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_wait(
JNIEnv* env, jclass cls, jlong conn, jobjectArray object_ids, jint timeout_ms,
jint num_returns) {
plasma::PlasmaClient* client = reinterpret_cast<plasma::PlasmaClient*>(conn);
jsize num_oids = env->GetArrayLength(object_ids);
if (num_returns < 0) {
jclass Exception = env->FindClass("java/lang/RuntimeException");
env->ThrowNew(Exception, "The argument num_returns cannot be less than zero.");
return nullptr;
}
if (num_returns > num_oids) {
jclass Exception = env->FindClass("java/lang/RuntimeException");
env->ThrowNew(Exception,
"The argument num_returns cannot be greater than len(object_ids).");
return nullptr;
}
std::vector<plasma::ObjectRequest> oreqs(num_oids);
for (int i = 0; i < num_oids; ++i) {
jbyteArray_to_object_id(
env, reinterpret_cast<jbyteArray>(env->GetObjectArrayElement(object_ids, i)),
&oreqs[i].object_id);
oreqs[i].type = plasma::ObjectRequestType::PLASMA_QUERY_ANYWHERE;
}
int num_return_objects;
// TODO: may be blocked. consider to add the thread support
ARROW_CHECK_OK(client->Wait(static_cast<int>(num_oids), oreqs.data(), num_returns,
static_cast<uint64_t>(timeout_ms), &num_return_objects));
int num_to_return = std::min(num_return_objects, num_returns);
jclass clsByteArray = env->FindClass("[B");
jobjectArray ret = env->NewObjectArray(num_to_return, clsByteArray, nullptr);
int num_returned = 0;
jbyteArray oid = nullptr;
for (int i = 0; i < num_oids; ++i) {
if (num_returned >= num_to_return) {
break;
}
if (oreqs[i].location == plasma::ObjectLocation::Local ||
oreqs[i].location == plasma::ObjectLocation::Remote) {
oid = env->NewByteArray(OBJECT_ID_SIZE);
object_id_to_jbyteArray(env, oid, &oreqs[i].object_id);
env->SetObjectArrayElement(ret, num_returned, oid);
num_returned++;
}
}
ARROW_CHECK(num_returned == num_to_return);
return ret;
}
JNIEXPORT jlong JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_evict(
JNIEnv* env, jclass cls, jlong conn, jlong num_bytes) {
plasma::PlasmaClient* client = reinterpret_cast<plasma::PlasmaClient*>(conn);
cpp/src/plasma/plasma.h
-3
View File
@@ -68,9 +68,6 @@ constexpr int64_t kBlockSize = 64;
struct Client;
/// Mapping from object IDs to type and status of the request.
typedef std::unordered_map<ObjectID, ObjectRequest> ObjectRequestMap;
// TODO(pcm): Replace this by the flatbuffers message PlasmaObjectSpec.
struct PlasmaObject {
#ifdef PLASMA_CUDA
cpp/src/plasma/protocol.cc
-143
View File
@@ -42,10 +42,6 @@ using flatbuffers::uoffset_t;
#define PLASMA_CHECK_ENUM(x, y) \
static_assert(static_cast<int>(x) == static_cast<int>(y), "protocol mismatch")
PLASMA_CHECK_ENUM(ObjectLocation::Local, fb::ObjectStatus::Local);
PLASMA_CHECK_ENUM(ObjectLocation::Remote, fb::ObjectStatus::Remote);
PLASMA_CHECK_ENUM(ObjectLocation::Nonexistent, fb::ObjectStatus::Nonexistent);
flatbuffers::Offset<flatbuffers::Vector<flatbuffers::Offset<flatbuffers::String>>>
ToFlatbuffer(flatbuffers::FlatBufferBuilder* fbb, const ObjectID* object_ids,
int64_t num_objects) {
@@ -367,56 +363,6 @@ Status ReadDeleteReply(uint8_t* data, size_t size, std::vector<ObjectID>* object
return Status::OK();
}
// Satus messages.
Status SendStatusRequest(int sock, const ObjectID* object_ids, int64_t num_objects) {
flatbuffers::FlatBufferBuilder fbb;
auto message =
fb::CreatePlasmaStatusRequest(fbb, ToFlatbuffer(&fbb, object_ids, num_objects));
return PlasmaSend(sock, MessageType::PlasmaStatusRequest, &fbb, message);
}
Status ReadStatusRequest(uint8_t* data, size_t size, ObjectID object_ids[],
int64_t num_objects) {
DCHECK(data);
auto message = flatbuffers::GetRoot<fb::PlasmaStatusRequest>(data);
DCHECK(VerifyFlatbuffer(message, data, size));
for (uoffset_t i = 0; i < num_objects; ++i) {
object_ids[i] = ObjectID::from_binary(message->object_ids()->Get(i)->str());
}
return Status::OK();
}
Status SendStatusReply(int sock, ObjectID object_ids[], int object_status[],
int64_t num_objects) {
flatbuffers::FlatBufferBuilder fbb;
auto message =
fb::CreatePlasmaStatusReply(fbb, ToFlatbuffer(&fbb, object_ids, num_objects),
fbb.CreateVector(object_status, num_objects));
return PlasmaSend(sock, MessageType::PlasmaStatusReply, &fbb, message);
}
int64_t ReadStatusReply_num_objects(uint8_t* data, size_t size) {
DCHECK(data);
auto message = flatbuffers::GetRoot<fb::PlasmaStatusReply>(data);
DCHECK(VerifyFlatbuffer(message, data, size));
return message->object_ids()->size();
}
Status ReadStatusReply(uint8_t* data, size_t size, ObjectID object_ids[],
int object_status[], int64_t num_objects) {
DCHECK(data);
auto message = flatbuffers::GetRoot<fb::PlasmaStatusReply>(data);
DCHECK(VerifyFlatbuffer(message, data, size));
for (uoffset_t i = 0; i < num_objects; ++i) {
object_ids[i] = ObjectID::from_binary(message->object_ids()->Get(i)->str());
}
for (uoffset_t i = 0; i < num_objects; ++i) {
object_status[i] = message->status()->data()[i];
}
return Status::OK();
}
// Contains messages.
Status SendContainsRequest(int sock, ObjectID object_id) {
@@ -640,95 +586,6 @@ Status ReadGetReply(uint8_t* data, size_t size, ObjectID object_ids[],
}
return Status::OK();
}
// Fetch messages.
Status SendFetchRequest(int sock, const ObjectID* object_ids, int64_t num_objects) {
flatbuffers::FlatBufferBuilder fbb;
auto message =
fb::CreatePlasmaFetchRequest(fbb, ToFlatbuffer(&fbb, object_ids, num_objects));
return PlasmaSend(sock, MessageType::PlasmaFetchRequest, &fbb, message);
}
Status ReadFetchRequest(uint8_t* data, size_t size, std::vector<ObjectID>& object_ids) {
DCHECK(data);
auto message = flatbuffers::GetRoot<fb::PlasmaFetchRequest>(data);
DCHECK(VerifyFlatbuffer(message, data, size));
for (uoffset_t i = 0; i < message->object_ids()->size(); ++i) {
object_ids.push_back(ObjectID::from_binary(message->object_ids()->Get(i)->str()));
}
return Status::OK();
}
// Wait messages.
Status SendWaitRequest(int sock, ObjectRequest object_requests[], int64_t num_requests,
int num_ready_objects, int64_t timeout_ms) {
flatbuffers::FlatBufferBuilder fbb;
std::vector<flatbuffers::Offset<fb::ObjectRequestSpec>> object_request_specs;
for (int i = 0; i < num_requests; i++) {
object_request_specs.push_back(fb::CreateObjectRequestSpec(
fbb, fbb.CreateString(object_requests[i].object_id.binary()),
static_cast<int>(object_requests[i].type)));
}
auto message = fb::CreatePlasmaWaitRequest(fbb, fbb.CreateVector(object_request_specs),
num_ready_objects, timeout_ms);
return PlasmaSend(sock, MessageType::PlasmaWaitRequest, &fbb, message);
}
Status ReadWaitRequest(uint8_t* data, size_t size, ObjectRequestMap& object_requests,
int64_t* timeout_ms, int* num_ready_objects) {
DCHECK(data);
auto message = flatbuffers::GetRoot<fb::PlasmaWaitRequest>(data);
DCHECK(VerifyFlatbuffer(message, data, size));
*num_ready_objects = message->num_ready_objects();
*timeout_ms = message->timeout();
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,
static_cast<ObjectRequestType>(message->object_requests()->Get(i)->type()),
ObjectLocation::Nonexistent});
object_requests[object_id] = object_request;
}
return Status::OK();
}
Status SendWaitReply(int sock, const ObjectRequestMap& object_requests,
int num_ready_objects) {
flatbuffers::FlatBufferBuilder fbb;
std::vector<flatbuffers::Offset<fb::ObjectReply>> object_replies;
for (const auto& entry : object_requests) {
const auto& object_request = entry.second;
object_replies.push_back(
fb::CreateObjectReply(fbb, fbb.CreateString(object_request.object_id.binary()),
static_cast<fb::ObjectStatus>(object_request.location)));
}
auto message = fb::CreatePlasmaWaitReply(
fbb, fbb.CreateVector(object_replies.data(), num_ready_objects), num_ready_objects);
return PlasmaSend(sock, MessageType::PlasmaWaitReply, &fbb, message);
}
Status ReadWaitReply(uint8_t* data, size_t size, ObjectRequest object_requests[],
int* num_ready_objects) {
DCHECK(data);
auto message = flatbuffers::GetRoot<fb::PlasmaWaitReply>(data);
DCHECK(VerifyFlatbuffer(message, data, size));
*num_ready_objects = message->num_ready_objects();
for (int i = 0; i < *num_ready_objects; i++) {
object_requests[i].object_id =
ObjectID::from_binary(message->object_requests()->Get(i)->object_id()->str());
object_requests[i].location =
static_cast<ObjectLocation>(message->object_requests()->Get(i)->status());
}
return Status::OK();
}
// Subscribe messages.
cpp/src/plasma/protocol.h
-35
View File
@@ -128,21 +128,6 @@ Status SendDeleteReply(int sock, const std::vector<ObjectID>& object_ids,
Status ReadDeleteReply(uint8_t* data, size_t size, std::vector<ObjectID>* object_ids,
std::vector<PlasmaError>* errors);
/* Satus messages. */
Status SendStatusRequest(int sock, const ObjectID* object_ids, int64_t num_objects);
Status ReadStatusRequest(uint8_t* data, size_t size, ObjectID object_ids[],
int64_t num_objects);
Status SendStatusReply(int sock, ObjectID object_ids[], int object_status[],
int64_t num_objects);
int64_t ReadStatusReply_num_objects(uint8_t* data, size_t size);
Status ReadStatusReply(uint8_t* data, size_t size, ObjectID object_ids[],
int object_status[], int64_t num_objects);
/* Plasma Constains message functions. */
Status SendContainsRequest(int sock, ObjectID object_id);
@@ -184,26 +169,6 @@ Status SendEvictReply(int sock, int64_t num_bytes);
Status ReadEvictReply(uint8_t* data, size_t size, int64_t& num_bytes);
/* Plasma Fetch Remote message functions. */
Status SendFetchRequest(int sock, const ObjectID* object_ids, int64_t num_objects);
Status ReadFetchRequest(uint8_t* data, size_t size, std::vector<ObjectID>& object_ids);
/* Plasma Wait message functions. */
Status SendWaitRequest(int sock, ObjectRequest object_requests[], int64_t num_requests,
int num_ready_objects, int64_t timeout_ms);
Status ReadWaitRequest(uint8_t* data, size_t size, ObjectRequestMap& object_requests,
int64_t* timeout_ms, int* num_ready_objects);
Status SendWaitReply(int sock, const ObjectRequestMap& object_requests,
int num_ready_objects);
Status ReadWaitReply(uint8_t* data, size_t size, ObjectRequest object_requests[],
int* num_ready_objects);
/* Plasma Subscribe message functions. */
Status SendSubscribeRequest(int sock);
cpp/src/plasma/test/serialization_tests.cc
-116
View File
@@ -254,44 +254,6 @@ TEST(PlasmaSerialization, DeleteReply) {
close(fd);
}
TEST(PlasmaSerialization, StatusRequest) {
int fd = create_temp_file();
constexpr int64_t num_objects = 2;
ObjectID object_ids[num_objects];
object_ids[0] = random_object_id();
object_ids[1] = random_object_id();
ARROW_CHECK_OK(SendStatusRequest(fd, object_ids, num_objects));
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));
ASSERT_EQ(object_ids[0], object_ids_read[0]);
ASSERT_EQ(object_ids[1], object_ids_read[1]);
close(fd);
}
TEST(PlasmaSerialization, StatusReply) {
int fd = create_temp_file();
ObjectID object_ids[2];
object_ids[0] = random_object_id();
object_ids[1] = random_object_id();
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);
int64_t num_objects = ReadStatusReply_num_objects(data.data(), data.size());
std::vector<ObjectID> object_ids_read(num_objects);
std::vector<int> object_statuses_read(num_objects);
ARROW_CHECK_OK(ReadStatusReply(data.data(), data.size(), object_ids_read.data(),
object_statuses_read.data(), num_objects));
ASSERT_EQ(object_ids[0], object_ids_read[0]);
ASSERT_EQ(object_ids[1], object_ids_read[1]);
ASSERT_EQ(object_statuses[0], object_statuses_read[0]);
ASSERT_EQ(object_statuses[1], object_statuses_read[1]);
close(fd);
}
TEST(PlasmaSerialization, EvictRequest) {
int fd = create_temp_file();
int64_t num_bytes = 111;
@@ -314,84 +276,6 @@ TEST(PlasmaSerialization, EvictReply) {
close(fd);
}
TEST(PlasmaSerialization, FetchRequest) {
int fd = create_temp_file();
ObjectID object_ids[2];
object_ids[0] = random_object_id();
object_ids[1] = random_object_id();
ARROW_CHECK_OK(SendFetchRequest(fd, object_ids, 2));
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]);
ASSERT_EQ(object_ids[1], object_ids_read[1]);
close(fd);
}
TEST(PlasmaSerialization, WaitRequest) {
int fd = create_temp_file();
const int num_objects_in = 2;
ObjectRequest object_requests_in[num_objects_in] = {
ObjectRequest({random_object_id(), ObjectRequestType::PLASMA_QUERY_ANYWHERE,
ObjectLocation::Local}),
ObjectRequest({random_object_id(), ObjectRequestType::PLASMA_QUERY_LOCAL,
ObjectLocation::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);
int num_ready_objects_out;
int64_t timeout_ms_read;
ObjectRequestMap object_requests_out;
ARROW_CHECK_OK(ReadWaitRequest(data.data(), data.size(), object_requests_out,
&timeout_ms_read, &num_ready_objects_out));
ASSERT_EQ(num_objects_in, object_requests_out.size());
ASSERT_EQ(num_ready_objects_out, num_ready_objects_in);
for (int i = 0; i < num_objects_in; i++) {
const ObjectID& object_id = object_requests_in[i].object_id;
ASSERT_EQ(1, object_requests_out.count(object_id));
const auto& entry = object_requests_out.find(object_id);
ASSERT_TRUE(entry != object_requests_out.end());
ASSERT_EQ(entry->second.object_id, object_requests_in[i].object_id);
ASSERT_EQ(entry->second.type, object_requests_in[i].type);
}
close(fd);
}
TEST(PlasmaSerialization, WaitReply) {
int fd = create_temp_file();
const int num_objects_in = 2;
/* Create a map with two ObjectRequests in it. */
ObjectRequestMap objects_in(num_objects_in);
ObjectID id1 = random_object_id();
objects_in[id1] =
ObjectRequest({id1, ObjectRequestType::PLASMA_QUERY_LOCAL, ObjectLocation::Local});
ObjectID id2 = random_object_id();
objects_in[id2] = ObjectRequest(
{id2, ObjectRequestType::PLASMA_QUERY_LOCAL, ObjectLocation::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);
ObjectRequest objects_out[2];
int num_objects_out;
ARROW_CHECK_OK(
ReadWaitReply(data.data(), data.size(), &objects_out[0], &num_objects_out));
ASSERT_EQ(num_objects_in, num_objects_out);
for (int i = 0; i < num_objects_out; i++) {
/* Each object request must appear exactly once. */
ASSERT_EQ(objects_in.count(objects_out[i].object_id), 1);
const auto& entry = objects_in.find(objects_out[i].object_id);
ASSERT_TRUE(entry != objects_in.end());
ASSERT_EQ(entry->second.object_id, objects_out[i].object_id);
ASSERT_EQ(entry->second.location, objects_out[i].location);
}
close(fd);
}
TEST(PlasmaSerialization, DataRequest) {
int fd = create_temp_file();
ObjectID object_id1 = random_object_id();
python/pyarrow/_plasma.pyx
-112
View File
@@ -62,11 +62,6 @@ cdef extern from "plasma/common.h" nogil:
@staticmethod
int64_t size()
cdef struct CObjectRequest" plasma::ObjectRequest":
CUniqueID object_id
int type
int location
cdef enum CObjectState" plasma::ObjectState":
PLASMA_CREATED" plasma::ObjectState::PLASMA_CREATED"
PLASMA_SEALED" plasma::ObjectState::PLASMA_SEALED"
@@ -91,14 +86,6 @@ cdef extern from "plasma/common.h" nogil:
cdef extern from "plasma/common.h":
cdef int64_t kDigestSize" plasma::kDigestSize"
cdef enum ObjectRequestType:
PLASMA_QUERY_LOCAL"plasma::ObjectRequestType::PLASMA_QUERY_LOCAL",
PLASMA_QUERY_ANYWHERE"plasma::ObjectRequestType::PLASMA_QUERY_ANYWHERE"
cdef enum ObjectLocation:
ObjectStatusLocal"plasma::ObjectLocation::Local"
ObjectStatusRemote"plasma::ObjectLocation::Remote"
cdef extern from "plasma/client.h" nogil:
cdef cppclass CPlasmaClient" plasma::PlasmaClient":
@@ -142,16 +129,6 @@ cdef extern from "plasma/client.h" nogil:
CStatus Disconnect()
CStatus Fetch(int num_object_ids, const CUniqueID* object_ids)
CStatus Wait(int64_t num_object_requests,
CObjectRequest* object_requests,
int num_ready_objects, int64_t timeout_ms,
int* num_objects_ready)
CStatus Transfer(const char* addr, int port,
const CUniqueID& object_id)
CStatus Delete(const c_vector[CUniqueID] object_ids)
cdef extern from "plasma/client.h" nogil:
@@ -641,95 +618,6 @@ cdef class PlasmaClient:
check_status(self.client.get().Evict(num_bytes, num_bytes_evicted))
return num_bytes_evicted
def transfer(self, address, int port, ObjectID object_id):
"""
Transfer local object with id object_id to another plasma instance
Parameters
----------
addr : str
IPv4 address of the plasma instance the object is sent to.
port : int
Port number of the plasma instance the object is sent to.
object_id : str
A string used to identify an object.
"""
cdef c_string addr = address.encode()
with nogil:
check_status(self.client.get()
.Transfer(addr.c_str(), port, object_id.data))
def fetch(self, object_ids):
"""
Fetch the objects with the given IDs from other plasma managers.
Parameters
----------
object_ids : list
A list of strings used to identify the objects.
"""
cdef c_vector[CUniqueID] ids
cdef ObjectID object_id
for object_id in object_ids:
ids.push_back(object_id.data)
with nogil:
check_status(self.client.get().Fetch(ids.size(), ids.data()))
def wait(self, object_ids, int64_t timeout=PLASMA_WAIT_TIMEOUT,
int num_returns=1):
"""
Wait until num_returns objects in object_ids are ready.
Currently, the object ID arguments to wait must be unique.
Parameters
----------
object_ids : list
List of object IDs to wait for.
timeout :int
Return to the caller after timeout milliseconds.
num_returns : int
We are waiting for this number of objects to be ready.
Returns
-------
list
List of object IDs that are ready.
list
List of object IDs we might still wait on.
"""
# Check that the object ID arguments are unique. The plasma manager
# currently crashes if given duplicate object IDs.
if len(object_ids) != len(set(object_ids)):
raise Exception("Wait requires a list of unique object IDs.")
cdef int64_t num_object_requests = len(object_ids)
cdef c_vector[CObjectRequest] object_requests = (
c_vector[CObjectRequest](num_object_requests))
cdef int num_objects_ready = 0
cdef ObjectID object_id
for i, object_id in enumerate(object_ids):
object_requests[i].object_id = object_id.data
object_requests[i].type = PLASMA_QUERY_ANYWHERE
with nogil:
check_status(self.client.get().Wait(num_object_requests,
object_requests.data(),
num_returns, timeout,
&num_objects_ready))
cdef int num_to_return = min(num_objects_ready, num_returns)
ready_ids = []
waiting_ids = set(object_ids)
cdef int num_returned = 0
for i in range(len(object_ids)):
if num_returned == num_to_return:
break
if (object_requests[i].location == ObjectStatusLocal or
object_requests[i].location == ObjectStatusRemote):
ready_ids.append(
ObjectID(object_requests[i].object_id.binary()))
waiting_ids.discard(
ObjectID(object_requests[i].object_id.binary()))
num_returned += 1
return ready_ids, list(waiting_ids)
def subscribe(self):
"""Subscribe to notifications about sealed objects."""
with nogil: