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

[xray] Corrects Error Handling During Push and Pull. (#2059)

Browse Source 
* Makes bad status during Pull non-fatal.
Makes a bad status during Push fatal.

* pretty logs

* Stephanie's feedback.
This commit is contained in:
Melih Elibol
2018-05-17 17:51:55 -07:00
committed by Philipp Moritz
parent 6ca122f723
commit 3c245f66d4
1 changed files with 15 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/object_manager/object_manager.cc
+15 -15
View File
@@ -133,7 +133,8 @@ void ObjectManager::GetLocationsSuccess(const std::vector<ray::ClientID> &client
const ray::ObjectID &object_id) {
RAY_CHECK(!client_ids.empty());
ClientID client_id = client_ids.front();
ray::Status status_code = Pull(object_id, client_id);
ray::Status status = Pull(object_id, client_id);
RAY_CHECK_OK(status);
}
void ObjectManager::GetLocationsFailed(const ObjectID &object_id) {
@@ -157,13 +158,11 @@ ray::Status ObjectManager::PullEstablishConnection(const ObjectID &object_id,
// TODO(hme): There is no cap on the number of pull request connections.
status = connection_pool_.GetSender(ConnectionPool::ConnectionType::MESSAGE, client_id,
&conn);
if (!status.ok()) {
// TODO(hme): Keep track of retries,
// and only retry on object not local
// for now.
SchedulePull(object_id, config_.pull_timeout_ms);
return status;
}
// Currently, acquiring a connection should not fail.
// No status from GetSender is returned which can be
// handled without failing.
RAY_CHECK_OK(status);
if (conn == nullptr) {
status = object_directory_->GetInformation(
client_id,
@@ -172,13 +171,14 @@ ray::Status ObjectManager::PullEstablishConnection(const ObjectID &object_id,
ConnectionPool::ConnectionType::MESSAGE, connection_info);
connection_pool_.RegisterSender(ConnectionPool::ConnectionType::MESSAGE,
client_id, async_conn);
RAY_CHECK_OK(PullSendRequest(object_id, async_conn));
Status pull_send_status = PullSendRequest(object_id, async_conn);
RAY_CHECK_OK(pull_send_status);
},
[this, object_id](const Status &status) {
SchedulePull(object_id, config_.pull_timeout_ms);
});
} else {
RAY_CHECK_OK(PullSendRequest(object_id, conn));
status = PullSendRequest(object_id, conn);
}
return status;
}
@@ -256,10 +256,9 @@ ray::Status ObjectManager::SendObjectHeaders(const ObjectID &object_id,
buffer_pool_.GetChunk(object_id, data_size, metadata_size, chunk_index);
ObjectBufferPool::ChunkInfo chunk_info = chunk_status.first;
// If status is not okay, then return immediately because
// plasma_client.Get failed.
// No reference is acquired for this chunk, so no need to release the chunk.
RAY_RETURN_NOT_OK(chunk_status.second);
// Fail on status not okay. The object is local, and there is
// no other anticipated error here.
RAY_CHECK_OK(chunk_status.second);
// Create buffer.
flatbuffers::FlatBufferBuilder fbb;
@@ -427,7 +426,8 @@ void ObjectManager::ExecuteReceiveObject(const ClientID &client_id,
// TODO(hme): This chunk failed, so create a pull request for this chunk.
}
} else {
RAY_LOG(ERROR) << "Buffer Create Failed: " << chunk_status.second.message();
RAY_LOG(ERROR) << "Create Chunk Failed index = " << chunk_index << ": "
<< chunk_status.second.message();
// Read object into empty buffer.
uint64_t buffer_length = buffer_pool_.GetBufferLength(chunk_index, data_size);
std::vector<uint8_t> mutable_vec;