| 353 | }; |
| 354 | |
| 355 | void Master::CreateSession(const CreateSessionRequest* req, |
| 356 | CreateSessionResponse* resp, MyClosure done) { |
| 357 | SchedClosure([this, req, resp, done]() { |
| 358 | Status status; |
| 359 | WorkerCacheFactoryOptions worker_cache_factory_options; |
| 360 | string grpc_protocol("grpc"); |
| 361 | worker_cache_factory_options.protocol = &grpc_protocol; |
| 362 | auto call_done = gtl::MakeCleanup([&status, &done] { done(status); }); |
| 363 | status = ValidateExternalGraphDefSyntax(req->graph_def()); |
| 364 | if (!status.ok()) return; |
| 365 | |
| 366 | // The following 4 variables are set differently, depending on whether this |
| 367 | // session uses a client-provided clusterspec or not. |
| 368 | WorkerCacheInterface* worker_cache = nullptr; |
| 369 | // Note: worker_cache_ptr will be null except if this session is using a |
| 370 | // client-supplied ClusterDef (ClusterSpec propagation). |
| 371 | std::unique_ptr<WorkerCacheInterface> worker_cache_ptr; |
| 372 | std::unique_ptr<DeviceSet> device_set; |
| 373 | // TODO(saeta): Convert to std::make_unique when available. |
| 374 | std::unique_ptr<std::vector<std::unique_ptr<Device>>> remote_devices( |
| 375 | new std::vector<std::unique_ptr<Device>>()); |
| 376 | |
| 377 | // set tensor fuse. |
| 378 | env_->tensor_fuse = req->config().tensor_fuse(); |
| 379 | |
| 380 | if (req->config().has_cluster_def()) { |
| 381 | worker_cache_factory_options.cluster_def = &req->config().cluster_def(); |
| 382 | |
| 383 | // Set the server_def's job_name and task_index fields. |
| 384 | string normalized_string; |
| 385 | string grpc_protocol(kGrpcProtocol); |
| 386 | if (req->target().compare(0, grpc_protocol.length(), grpc_protocol) == |
| 387 | 0) { |
| 388 | normalized_string = |
| 389 | req->target().substr(grpc_protocol.length(), string::npos); |
| 390 | } else { |
| 391 | normalized_string = req->target(); |
| 392 | } |
| 393 | for (auto&& job : req->config().cluster_def().job()) { |
| 394 | for (auto&& task : job.tasks()) { |
| 395 | if (task.second == normalized_string) { |
| 396 | if (worker_cache_factory_options.job_name != nullptr) { |
| 397 | status = errors::InvalidArgument( |
| 398 | "Found multiple matching tasks that correspond to " |
| 399 | "to the master. Master target: '", |
| 400 | req->target(), "'. ClusterDef: ", |
| 401 | req->config().cluster_def().ShortDebugString()); |
| 402 | LOG(ERROR) << status; |
| 403 | return; |
| 404 | } |
| 405 | if (env_->local_devices[0]->parsed_name().job == job.name() && |
| 406 | env_->local_devices[0]->parsed_name().task == task.first) { |
| 407 | // TODO(b/37868888): Remove this limitation when resolved |
| 408 | status = errors::InvalidArgument( |
| 409 | "The ClusterSpec names the job and task index to be the same " |
| 410 | "names that were provided when the server booted. This is " |
| 411 | "currently not allowed. Job: ", |
| 412 | job.name(), ", task index: ", task.first); |