wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-27 11:26:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

[Core] remove create_and_seal and create_and_seal_batch (#9457)

Browse Source
This commit is contained in:
Zhuohan Li
2020-07-14 14:37:13 -07:00
committed by GitHub
parent ed6157c257
commit 003518619f
7 changed files with 8 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/object_manager/plasma/client.cc
-55
View File
@@ -253,14 +253,6 @@ class PlasmaClient::Impl : public std::enable_shared_from_this<PlasmaClient::Imp
int64_t metadata_size, std::shared_ptr<Buffer>* data, int device_num = 0,
bool evict_if_full = true);
Status CreateAndSeal(const ObjectID& object_id, const std::string& data,
const std::string& metadata, bool evict_if_full = true);
Status CreateAndSealBatch(const std::vector<ObjectID>& object_ids,
const std::vector<std::string>& data,
const std::vector<std::string>& metadata,
bool evict_if_full = true);
Status Get(const std::vector<ObjectID>& object_ids, int64_t timeout_ms,
std::vector<ObjectBuffer>* object_buffers);
@@ -509,41 +501,6 @@ Status PlasmaClient::Impl::Create(const ObjectID& object_id, int64_t data_size,
return Status::OK();
}
Status PlasmaClient::Impl::CreateAndSeal(const ObjectID& object_id,
const std::string& data,
const std::string& metadata,
bool evict_if_full) {
std::lock_guard<std::recursive_mutex> guard(client_mutex_);
RAY_LOG(DEBUG) << "called CreateAndSeal on conn " << store_conn_;
RAY_RETURN_NOT_OK(SendCreateAndSealRequest(store_conn_, object_id, evict_if_full, data,
metadata));
std::vector<uint8_t> buffer;
RAY_RETURN_NOT_OK(
PlasmaReceive(store_conn_, MessageType::PlasmaCreateAndSealReply, &buffer));
RAY_RETURN_NOT_OK(ReadCreateAndSealReply(buffer.data(), buffer.size()));
return Status::OK();
}
Status PlasmaClient::Impl::CreateAndSealBatch(const std::vector<ObjectID>& object_ids,
const std::vector<std::string>& data,
const std::vector<std::string>& metadata,
bool evict_if_full) {
std::lock_guard<std::recursive_mutex> guard(client_mutex_);
RAY_LOG(DEBUG) << "called CreateAndSealBatch on conn " << store_conn_;
RAY_RETURN_NOT_OK(SendCreateAndSealBatchRequest(store_conn_, object_ids, evict_if_full,
data, metadata));
std::vector<uint8_t> buffer;
RAY_RETURN_NOT_OK(
PlasmaReceive(store_conn_, MessageType::PlasmaCreateAndSealBatchReply, &buffer));
RAY_RETURN_NOT_OK(ReadCreateAndSealBatchReply(buffer.data(), buffer.size()));
return Status::OK();
}
Status PlasmaClient::Impl::GetBuffers(
const ObjectID* object_ids, int64_t num_objects, int64_t timeout_ms,
const std::function<std::shared_ptr<Buffer>(
@@ -1148,18 +1105,6 @@ Status PlasmaClient::Create(const ObjectID& object_id, int64_t data_size,
evict_if_full);
}
Status PlasmaClient::CreateAndSeal(const ObjectID& object_id, const std::string& data,
const std::string& metadata, bool evict_if_full) {
return impl_->CreateAndSeal(object_id, data, metadata, evict_if_full);
}
Status PlasmaClient::CreateAndSealBatch(const std::vector<ObjectID>& object_ids,
const std::vector<std::string>& data,
const std::vector<std::string>& metadata,
bool evict_if_full) {
return impl_->CreateAndSealBatch(object_ids, data, metadata, evict_if_full);
}
Status PlasmaClient::Get(const std::vector<ObjectID>& object_ids, int64_t timeout_ms,
std::vector<ObjectBuffer>* object_buffers) {
return impl_->Get(object_ids, timeout_ms, object_buffers);
src/ray/object_manager/plasma/client.h
-27
View File
@@ -101,33 +101,6 @@ class RAY_EXPORT PlasmaClient {
int64_t metadata_size, std::shared_ptr<Buffer>* data, int device_num = 0,
bool evict_if_full = true);
/// Create and seal an object in the object store. This is an optimization
/// which allows small objects to be created quickly with fewer messages to
/// the store.
///
/// \param object_id The ID of the object to create.
/// \param data The data for the object to create.
/// \param metadata The metadata for the object to create.
/// \param evict_if_full Whether to evict other objects to make space for
/// this object.
/// \return The return status.
Status CreateAndSeal(const ObjectID& object_id, const std::string& data,
const std::string& metadata, bool evict_if_full = true);
/// Create and seal multiple objects in the object store. This is an optimization
/// of CreateAndSeal to eliminate the cost of IPC per object.
///
/// \param object_ids The vector of IDs of the objects to create.
/// \param data The vector of data for the objects to create.
/// \param metadata The vector of metadata for the objects to create.
/// \param evict_if_full Whether to evict other objects to make space for
/// these objects.
/// \return The return status.
Status CreateAndSealBatch(const std::vector<ObjectID>& object_ids,
const std::vector<std::string>& data,
const std::vector<std::string>& metadata,
bool evict_if_full = true);
/// Get some objects from the Plasma Store. This function will block until the
/// objects have all been created and sealed in the Plasma Store or the
/// timeout expires.
src/ray/object_manager/plasma/plasma.fbs
-35
View File
@@ -24,8 +24,6 @@ enum MessageType:long {
// Create a new object.
PlasmaCreateRequest,
PlasmaCreateReply,
PlasmaCreateAndSealRequest,
PlasmaCreateAndSealReply,
PlasmaAbortRequest,
PlasmaAbortReply,
// Seal an object.
@@ -69,10 +67,6 @@ enum MessageType:long {
// Get debugging information from the store.
PlasmaGetDebugStringRequest,
PlasmaGetDebugStringReply,
// Create and seal a batch of objects. This should be used to save
// IPC for creating many small objects.
PlasmaCreateAndSealBatchRequest,
PlasmaCreateAndSealBatchReply,
// Touch a number of objects to bump their position in the LRU cache.
PlasmaRefreshLRURequest,
PlasmaRefreshLRUReply,
@@ -165,35 +159,6 @@ table PlasmaCreateReply {
ipc_handle: CudaHandle;
}
table PlasmaCreateAndSealRequest {
// ID of the object to be created.
object_id: string;
// Whether to evict other objects to make room for this one.
evict_if_full: bool;
// The object's data.
data: string;
// The object's metadata.
metadata: string;
}
table PlasmaCreateAndSealReply {
// Error that occurred for this call.
error: PlasmaError;
}
table PlasmaCreateAndSealBatchRequest {
object_ids: [string];
// Whether to evict other objects to make room for these objects.
evict_if_full: bool;
data: [string];
metadata: [string];
}
table PlasmaCreateAndSealBatchReply {
// Error that occurred for this call.
error: PlasmaError;
}
table PlasmaAbortRequest {
// ID of the object to be aborted.
object_id: string;
src/ray/object_manager/plasma/protocol.cc
-87
View File
@@ -278,93 +278,6 @@ Status ReadCreateReply(uint8_t* data, size_t size, ObjectID* object_id,
return PlasmaErrorStatus(message->error());
}
Status SendCreateAndSealRequest(const std::shared_ptr<StoreConn> &store_conn, const ObjectID& object_id, bool evict_if_full,
const std::string& data, const std::string& metadata) {
flatbuffers::FlatBufferBuilder fbb;
auto message = fb::CreatePlasmaCreateAndSealRequest(
fbb, fbb.CreateString(object_id.Binary()), evict_if_full, fbb.CreateString(data),
fbb.CreateString(metadata));
return PlasmaSend(store_conn, MessageType::PlasmaCreateAndSealRequest, &fbb, message);
}
Status ReadCreateAndSealRequest(uint8_t* data, size_t size, ObjectID* object_id,
bool* evict_if_full, std::string* object_data,
std::string* metadata) {
RAY_DCHECK(data);
auto message = flatbuffers::GetRoot<fb::PlasmaCreateAndSealRequest>(data);
RAY_DCHECK(VerifyFlatbuffer(message, data, size));
*object_id = ObjectID::FromBinary(message->object_id()->str());
*evict_if_full = message->evict_if_full();
*object_data = message->data()->str();
*metadata = message->metadata()->str();
return Status::OK();
}
Status SendCreateAndSealBatchRequest(const std::shared_ptr<StoreConn> &store_conn, const std::vector<ObjectID>& object_ids,
bool evict_if_full,
const std::vector<std::string>& data,
const std::vector<std::string>& metadata) {
flatbuffers::FlatBufferBuilder fbb;
auto message = fb::CreatePlasmaCreateAndSealBatchRequest(
fbb, ToFlatbuffer(&fbb, object_ids.data(), object_ids.size()), evict_if_full,
ToFlatbuffer(&fbb, data), ToFlatbuffer(&fbb, metadata));
return PlasmaSend(store_conn, MessageType::PlasmaCreateAndSealBatchRequest, &fbb, message);
}
Status ReadCreateAndSealBatchRequest(uint8_t* data, size_t size,
std::vector<ObjectID>* object_ids,
bool* evict_if_full,
std::vector<std::string>* object_data,
std::vector<std::string>* metadata) {
RAY_DCHECK(data);
auto message = flatbuffers::GetRoot<fb::PlasmaCreateAndSealBatchRequest>(data);
RAY_DCHECK(VerifyFlatbuffer(message, data, size));
*evict_if_full = message->evict_if_full();
ConvertToVector(message->object_ids(), object_ids,
[](const flatbuffers::String& element) {
return ObjectID::FromBinary(element.str());
});
ConvertToVector(message->data(), object_data,
[](const flatbuffers::String& element) { return element.str(); });
ConvertToVector(message->metadata(), metadata,
[](const flatbuffers::String& element) { return element.str(); });
return Status::OK();
}
Status SendCreateAndSealReply(const std::shared_ptr<Client> &client, PlasmaError error) {
flatbuffers::FlatBufferBuilder fbb;
auto message = fb::CreatePlasmaCreateAndSealReply(fbb, static_cast<PlasmaError>(error));
return PlasmaSend(client, MessageType::PlasmaCreateAndSealReply, &fbb, message);
}
Status ReadCreateAndSealReply(uint8_t* data, size_t size) {
RAY_DCHECK(data);
auto message = flatbuffers::GetRoot<fb::PlasmaCreateAndSealReply>(data);
RAY_DCHECK(VerifyFlatbuffer(message, data, size));
return PlasmaErrorStatus(message->error());
}
Status SendCreateAndSealBatchReply(const std::shared_ptr<Client> &client, PlasmaError error) {
flatbuffers::FlatBufferBuilder fbb;
auto message =
fb::CreatePlasmaCreateAndSealBatchReply(fbb, static_cast<PlasmaError>(error));
return PlasmaSend(client, MessageType::PlasmaCreateAndSealBatchReply, &fbb, message);
}
Status ReadCreateAndSealBatchReply(uint8_t* data, size_t size) {
RAY_DCHECK(data);
auto message = flatbuffers::GetRoot<fb::PlasmaCreateAndSealBatchReply>(data);
RAY_DCHECK(VerifyFlatbuffer(message, data, size));
return PlasmaErrorStatus(message->error());
}
Status SendAbortRequest(const std::shared_ptr<StoreConn> &store_conn, ObjectID object_id) {
flatbuffers::FlatBufferBuilder fbb;
auto message = fb::CreatePlasmaAbortRequest(fbb, fbb.CreateString(object_id.Binary()));
src/ray/object_manager/plasma/protocol.h
-26
View File
@@ -89,32 +89,6 @@ Status SendCreateReply(const std::shared_ptr<Client> &client, ObjectID object_id
Status ReadCreateReply(uint8_t* data, size_t size, ObjectID* object_id,
PlasmaObject* object, int* store_fd, int64_t* mmap_size);
Status SendCreateAndSealRequest(const std::shared_ptr<StoreConn> &store_conn, const ObjectID& object_id, bool evict_if_full,
const std::string& data, const std::string& metadata);
Status ReadCreateAndSealRequest(uint8_t* data, size_t size, ObjectID* object_id,
bool* evict_if_full, std::string* object_data,
std::string* metadata);
Status SendCreateAndSealBatchRequest(const std::shared_ptr<StoreConn> &store_conn, const std::vector<ObjectID>& object_ids,
bool evict_if_full,
const std::vector<std::string>& data,
const std::vector<std::string>& metadata);
Status ReadCreateAndSealBatchRequest(uint8_t* data, size_t size,
std::vector<ObjectID>* object_id,
bool* evict_if_full,
std::vector<std::string>* object_data,
std::vector<std::string>* metadata);
Status SendCreateAndSealReply(const std::shared_ptr<Client> &client, PlasmaError error);
Status ReadCreateAndSealReply(uint8_t* data, size_t size);
Status SendCreateAndSealBatchReply(const std::shared_ptr<Client> &client, PlasmaError error);
Status ReadCreateAndSealBatchReply(uint8_t* data, size_t size);
Status SendAbortRequest(const std::shared_ptr<StoreConn> &store_conn, ObjectID object_id);
Status ReadAbortRequest(uint8_t* data, size_t size, ObjectID* object_id);
src/ray/object_manager/plasma/store.cc
-81
View File
@@ -927,87 +927,6 @@ Status PlasmaStore::ProcessMessage(const std::shared_ptr<Client> &client) {
client->used_fds.insert(object.store_fd);
}
} break;
case fb::MessageType::PlasmaCreateAndSealRequest: {
bool evict_if_full;
std::string data;
std::string metadata;
RAY_RETURN_NOT_OK(ReadCreateAndSealRequest(input, input_size, &object_id,
&evict_if_full, &data, &metadata));
// CreateAndSeal currently only supports device_num = 0, which corresponds
// to the host.
int device_num = 0;
PlasmaError error_code = CreateObject(object_id, evict_if_full, data.size(),
metadata.size(), device_num, client, &object);
// If the object was successfully created, fill out the object data and seal it.
if (error_code == PlasmaError::OK) {
auto entry = GetObjectTableEntry(&store_info_, object_id);
RAY_CHECK(entry != nullptr);
// Write the inlined data and metadata into the allocated object.
std::memcpy(entry->pointer, data.data(), data.size());
std::memcpy(entry->pointer + data.size(), metadata.data(), metadata.size());
SealObjects({object_id});
// Remove the client from the object's array of clients because the
// object is not being used by any client. The client was added to the
// object's array of clients in CreateObject. This is analogous to the
// Release call that happens in the client's Seal method.
RAY_CHECK(RemoveFromClientObjectIds(object_id, entry, client) == 1);
}
// Reply to the client.
HANDLE_SIGPIPE(SendCreateAndSealReply(client, error_code), client->fd);
} break;
case fb::MessageType::PlasmaCreateAndSealBatchRequest: {
bool evict_if_full;
std::vector<ObjectID> object_ids;
std::vector<std::string> data;
std::vector<std::string> metadata;
RAY_RETURN_NOT_OK(ReadCreateAndSealBatchRequest(
input, input_size, &object_ids, &evict_if_full, &data, &metadata));
// CreateAndSeal currently only supports device_num = 0, which corresponds
// to the host.
int device_num = 0;
size_t i = 0;
PlasmaError error_code = PlasmaError::OK;
for (i = 0; i < object_ids.size(); i++) {
error_code = CreateObject(object_ids[i], evict_if_full, data[i].size(),
metadata[i].size(), device_num, client, &object);
if (error_code != PlasmaError::OK) {
break;
}
}
// if OK, seal all the objects,
// if error, abort the previous i objects immediately
if (error_code == PlasmaError::OK) {
for (i = 0; i < object_ids.size(); i++) {
auto entry = GetObjectTableEntry(&store_info_, object_ids[i]);
RAY_CHECK(entry != nullptr);
// Write the inlined data and metadata into the allocated object.
std::memcpy(entry->pointer, data[i].data(), data[i].size());
std::memcpy(entry->pointer + data[i].size(), metadata[i].data(),
metadata[i].size());
}
SealObjects(object_ids);
// Remove the client from the object's array of clients because the
// object is not being used by any client. The client was added to the
// object's array of clients in CreateObject. This is analogous to the
// Release call that happens in the client's Seal method.
for (i = 0; i < object_ids.size(); i++) {
auto entry = GetObjectTableEntry(&store_info_, object_ids[i]);
RAY_CHECK(RemoveFromClientObjectIds(object_ids[i], entry, client) == 1);
}
} else {
for (size_t j = 0; j < i; j++) {
AbortObject(object_ids[j], client);
}
}
HANDLE_SIGPIPE(SendCreateAndSealBatchReply(client, error_code), client->fd);
} break;
case fb::MessageType::PlasmaAbortRequest: {
RAY_RETURN_NOT_OK(ReadAbortRequest(input, input_size, &object_id));
RAY_CHECK(AbortObject(object_id, client) == 1) << "To abort an object, the only "
src/ray/raylet/node_manager.cc
+8 -1
View File
@@ -2237,7 +2237,14 @@ void NodeManager::MarkObjectsAsFailed(const ErrorType &error_type,
const JobID &job_id) {
const std::string meta = std::to_string(static_cast<int>(error_type));
for (const auto &object_id : objects_to_fail) {
Status status = store_client_.CreateAndSeal(object_id, "", meta);
std::shared_ptr<arrow::Buffer> data;
Status status;
status = store_client_.Create(object_id, 0,
reinterpret_cast<const uint8_t *>(meta.c_str()),
meta.length(), &data);
if (status.ok()) {
status = store_client_.Seal(object_id);
}
if (!status.ok() && !status.IsObjectExists()) {
// If we failed to save the error code, log a warning and push an error message
// to the driver.