| 169 | } |
| 170 | |
| 171 | Status StarServerBase::Init() { |
| 172 | mutex_lock l(mu_); |
| 173 | CHECK_EQ(state_, NEW); |
| 174 | master_env_.env = env_; |
| 175 | worker_env_.env = env_; |
| 176 | |
| 177 | SessionOptions sess_opts; |
| 178 | ConfigProto config = server_def_.default_session_config(); |
| 179 | sess_opts.config = config; |
| 180 | master_env_.run_graph_mode = GetRunGraphModeFlag(config); |
| 181 | master_env_.run_graph_mode_lite = GetRunGraphModeFlagLite(config); |
| 182 | if (master_env_.run_graph_mode || |
| 183 | master_env_.run_graph_mode_lite) { |
| 184 | master_env_.run_graph_mode_with_zero_copy = true; |
| 185 | } |
| 186 | |
| 187 | string name_prefix = |
| 188 | strings::StrCat("/job:", server_def_.job_name(), "/replica:0", |
| 189 | "/task:", server_def_.task_index()); |
| 190 | |
| 191 | std::vector<std::unique_ptr<Device>> devices; |
| 192 | TF_RETURN_IF_ERROR( |
| 193 | DeviceFactory::AddDevices(sess_opts, name_prefix, &devices)); |
| 194 | worker_env_.device_mgr = new DeviceMgr(std::move(devices)); |
| 195 | master_env_.local_devices = worker_env_.device_mgr->ListDevices(); |
| 196 | worker_env_.local_devices = worker_env_.device_mgr->ListDevices(); |
| 197 | worker_env_.rendezvous_mgr = CreateRendezvousMgr(&worker_env_); |
| 198 | |
| 199 | string unused; |
| 200 | string default_worker_name; |
| 201 | if (!DeviceNameUtils::SplitDeviceName(master_env_.local_devices[0]->name(), |
| 202 | &default_worker_name, &unused)) { |
| 203 | return errors::Internal( |
| 204 | "[Distributed] Could not parse worker name."); |
| 205 | } |
| 206 | |
| 207 | int requested_port = -1; |
| 208 | for (const auto& job : server_def_.cluster().job()) { |
| 209 | if (job.name() == server_def_.job_name()) { |
| 210 | auto iter = job.tasks().find(server_def_.task_index()); |
| 211 | if (iter == job.tasks().end()) { |
| 212 | return errors::InvalidArgument( |
| 213 | "[Distributed] Task ", server_def_.task_index(), |
| 214 | " was not defined in job \"", |
| 215 | server_def_.job_name(), "\""); |
| 216 | } |
| 217 | |
| 218 | const auto& hostname_port = str_util::Split(iter->second, ':'); |
| 219 | |
| 220 | if (hostname_port.size() != 2) { |
| 221 | return errors::InvalidArgument( |
| 222 | "[Distributed] Could not parse port for local server from \"", |
| 223 | iter->second, "\""); |
| 224 | } |
| 225 | |
| 226 | if (!strings::safe_strto32(hostname_port[1], &requested_port)) { |
| 227 | return errors::InvalidArgument( |
| 228 | "[Distributed] Could not parse port for local server from \"", |
no test coverage detected