| 67 | } |
| 68 | |
| 69 | void NewRemoteDevices(Env* env, WorkerCacheInterface* worker_cache, |
| 70 | const string& worker_name, NewRemoteDevicesDone done) { |
| 71 | WorkerInterface* wi = worker_cache->GetOrCreateWorker(worker_name); |
| 72 | if (wi == nullptr) { |
| 73 | std::vector<Device*> empty; |
| 74 | done(errors::NotFound("Device ", worker_name, " is not found."), &empty); |
| 75 | return; |
| 76 | } |
| 77 | struct Call { |
| 78 | GetStatusRequest req; |
| 79 | GetStatusResponse resp; |
| 80 | }; |
| 81 | Call* call = new Call; |
| 82 | CallOptions* call_opts = new CallOptions; |
| 83 | StatusCallback* cb = new StatusCallback(); |
| 84 | *cb = [env, worker_cache, worker_name, done, wi, |
| 85 | call, call_opts, cb](const Status& status) { |
| 86 | Status s = status; |
| 87 | std::vector<Device*> remote_devices; |
| 88 | auto cleanup = gtl::MakeCleanup( |
| 89 | [&worker_cache, &worker_name, &wi, &done, |
| 90 | &remote_devices, &s, call, call_opts, cb] { |
| 91 | worker_cache->ReleaseWorker(worker_name, wi); |
| 92 | done(s, &remote_devices); |
| 93 | delete call; |
| 94 | delete call_opts; |
| 95 | delete cb; |
| 96 | }); |
| 97 | if (s.ok()) { |
| 98 | DeviceNameUtils::ParsedName worker_name_parsed; |
| 99 | if (!DeviceNameUtils::ParseFullName(worker_name, &worker_name_parsed) || |
| 100 | !worker_name_parsed.has_job || !worker_name_parsed.has_replica || |
| 101 | !worker_name_parsed.has_task) { |
| 102 | s = errors::InvalidArgument("Could not parse worker name: ", |
| 103 | worker_name); |
| 104 | LOG(WARNING) << s; |
| 105 | return; |
| 106 | } |
| 107 | remote_devices.reserve(call->resp.device_attributes_size()); |
| 108 | for (const DeviceAttributes& da : call->resp.device_attributes()) { |
| 109 | DeviceNameUtils::ParsedName device_name_parsed; |
| 110 | CHECK(DeviceNameUtils::ParseFullName(da.name(), &device_name_parsed)) |
| 111 | << "Device attribute name '" << da.name() << "' could not be " |
| 112 | << "parsed. Device Attribute: " << da.DebugString(); |
| 113 | // Preserve the exact name, if possible. |
| 114 | // TODO(b/37868888): Simplify when legacy device name formats removed. |
| 115 | if (device_name_parsed.job == worker_name_parsed.job && |
| 116 | device_name_parsed.replica == worker_name_parsed.replica && |
| 117 | device_name_parsed.task == worker_name_parsed.task) { |
| 118 | auto d = new RemoteDevice(env, da); |
| 119 | remote_devices.push_back(d); |
| 120 | } else { |
| 121 | DeviceAttributes da_rewritten = da; |
| 122 | da_rewritten.set_name(DeviceNameUtils::FullName( |
| 123 | worker_name_parsed.job, worker_name_parsed.replica, |
| 124 | worker_name_parsed.task, device_name_parsed.type, |
| 125 | device_name_parsed.id)); |
| 126 | auto d = new RemoteDevice(env, da_rewritten); |