diff --git a/src/ray/object_manager/plasma/client.cc b/src/ray/object_manager/plasma/client.cc index 08dbd30fa..30f7e9a61 100644 --- a/src/ray/object_manager/plasma/client.cc +++ b/src/ray/object_manager/plasma/client.cc @@ -253,14 +253,6 @@ class PlasmaClient::Impl : public std::enable_shared_from_this* 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& object_ids, - const std::vector& data, - const std::vector& metadata, - bool evict_if_full = true); - Status Get(const std::vector& object_ids, int64_t timeout_ms, std::vector* 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 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 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& object_ids, - const std::vector& data, - const std::vector& metadata, - bool evict_if_full) { - std::lock_guard 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 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( @@ -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& object_ids, - const std::vector& data, - const std::vector& metadata, - bool evict_if_full) { - return impl_->CreateAndSealBatch(object_ids, data, metadata, evict_if_full); -} - Status PlasmaClient::Get(const std::vector& object_ids, int64_t timeout_ms, std::vector* object_buffers) { return impl_->Get(object_ids, timeout_ms, object_buffers); diff --git a/src/ray/object_manager/plasma/client.h b/src/ray/object_manager/plasma/client.h index e0b1fdbff..6ef994380 100644 --- a/src/ray/object_manager/plasma/client.h +++ b/src/ray/object_manager/plasma/client.h @@ -101,33 +101,6 @@ class RAY_EXPORT PlasmaClient { int64_t metadata_size, std::shared_ptr* 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& object_ids, - const std::vector& data, - const std::vector& 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. diff --git a/src/ray/object_manager/plasma/plasma.fbs b/src/ray/object_manager/plasma/plasma.fbs index 41de16556..856b6d5b7 100644 --- a/src/ray/object_manager/plasma/plasma.fbs +++ b/src/ray/object_manager/plasma/plasma.fbs @@ -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; diff --git a/src/ray/object_manager/plasma/protocol.cc b/src/ray/object_manager/plasma/protocol.cc index bcc9e553e..fea821001 100644 --- a/src/ray/object_manager/plasma/protocol.cc +++ b/src/ray/object_manager/plasma/protocol.cc @@ -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 &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(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 &store_conn, const std::vector& object_ids, - bool evict_if_full, - const std::vector& data, - const std::vector& 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* object_ids, - bool* evict_if_full, - std::vector* object_data, - std::vector* metadata) { - RAY_DCHECK(data); - auto message = flatbuffers::GetRoot(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, PlasmaError error) { - flatbuffers::FlatBufferBuilder fbb; - auto message = fb::CreatePlasmaCreateAndSealReply(fbb, static_cast(error)); - return PlasmaSend(client, MessageType::PlasmaCreateAndSealReply, &fbb, message); -} - -Status ReadCreateAndSealReply(uint8_t* data, size_t size) { - RAY_DCHECK(data); - auto message = flatbuffers::GetRoot(data); - RAY_DCHECK(VerifyFlatbuffer(message, data, size)); - return PlasmaErrorStatus(message->error()); -} - -Status SendCreateAndSealBatchReply(const std::shared_ptr &client, PlasmaError error) { - flatbuffers::FlatBufferBuilder fbb; - auto message = - fb::CreatePlasmaCreateAndSealBatchReply(fbb, static_cast(error)); - return PlasmaSend(client, MessageType::PlasmaCreateAndSealBatchReply, &fbb, message); -} - -Status ReadCreateAndSealBatchReply(uint8_t* data, size_t size) { - RAY_DCHECK(data); - auto message = flatbuffers::GetRoot(data); - RAY_DCHECK(VerifyFlatbuffer(message, data, size)); - return PlasmaErrorStatus(message->error()); -} - Status SendAbortRequest(const std::shared_ptr &store_conn, ObjectID object_id) { flatbuffers::FlatBufferBuilder fbb; auto message = fb::CreatePlasmaAbortRequest(fbb, fbb.CreateString(object_id.Binary())); diff --git a/src/ray/object_manager/plasma/protocol.h b/src/ray/object_manager/plasma/protocol.h index 2289d26bb..070f41b11 100644 --- a/src/ray/object_manager/plasma/protocol.h +++ b/src/ray/object_manager/plasma/protocol.h @@ -89,32 +89,6 @@ Status SendCreateReply(const std::shared_ptr &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 &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 &store_conn, const std::vector& object_ids, - bool evict_if_full, - const std::vector& data, - const std::vector& metadata); - -Status ReadCreateAndSealBatchRequest(uint8_t* data, size_t size, - std::vector* object_id, - bool* evict_if_full, - std::vector* object_data, - std::vector* metadata); - -Status SendCreateAndSealReply(const std::shared_ptr &client, PlasmaError error); - -Status ReadCreateAndSealReply(uint8_t* data, size_t size); - -Status SendCreateAndSealBatchReply(const std::shared_ptr &client, PlasmaError error); - -Status ReadCreateAndSealBatchReply(uint8_t* data, size_t size); - Status SendAbortRequest(const std::shared_ptr &store_conn, ObjectID object_id); Status ReadAbortRequest(uint8_t* data, size_t size, ObjectID* object_id); diff --git a/src/ray/object_manager/plasma/store.cc b/src/ray/object_manager/plasma/store.cc index a8a9c286b..6aa983294 100644 --- a/src/ray/object_manager/plasma/store.cc +++ b/src/ray/object_manager/plasma/store.cc @@ -927,87 +927,6 @@ Status PlasmaStore::ProcessMessage(const std::shared_ptr &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 object_ids; - std::vector data; - std::vector 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 " diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 7927d384d..d67090730 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -2237,7 +2237,14 @@ void NodeManager::MarkObjectsAsFailed(const ErrorType &error_type, const JobID &job_id) { const std::string meta = std::to_string(static_cast(error_type)); for (const auto &object_id : objects_to_fail) { - Status status = store_client_.CreateAndSeal(object_id, "", meta); + std::shared_ptr data; + Status status; + status = store_client_.Create(object_id, 0, + reinterpret_cast(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.