| 268 | } |
| 269 | |
| 270 | Status EagerServiceImpl::Enqueue(const EnqueueRequest* request, |
| 271 | EnqueueResponse* response) { |
| 272 | profiler::TraceMe activity( |
| 273 | [&] { |
| 274 | return absl::StrCat("EagerService:Enqueue:", request->DebugString()); |
| 275 | }, |
| 276 | profiler::TraceMeLevel::kInfo); |
| 277 | ServerContext* context = nullptr; |
| 278 | TF_RETURN_IF_ERROR(GetServerContext(request->context_id(), &context)); |
| 279 | core::ScopedUnref context_unref(context); |
| 280 | |
| 281 | auto executor = context->Context()->Executor(); |
| 282 | for (const auto& item : request->queue()) { |
| 283 | auto* queue_response = response->add_queue_response(); |
| 284 | if (item.has_operation()) { |
| 285 | TF_RETURN_IF_ERROR( |
| 286 | ExecuteOp(item.operation(), context->Context(), queue_response)); |
| 287 | } else if (item.has_handle_to_decref()) { |
| 288 | auto handle_to_decref = absl::make_unique<RemoteTensorHandleInternal>( |
| 289 | item.handle_to_decref()); |
| 290 | auto node = absl::make_unique<ClientTensorHandleDeleteNode>( |
| 291 | context, std::move(handle_to_decref)); |
| 292 | TF_RETURN_IF_ERROR( |
| 293 | executor->Async() |
| 294 | ? context->Context()->Executor()->Add(std::move(node)) |
| 295 | : node->Run()); |
| 296 | } else { |
| 297 | TF_RETURN_IF_ERROR(SendTensor(item.send_tensor(), context->Context())); |
| 298 | } |
| 299 | } |
| 300 | |
| 301 | return Status::OK(); |
| 302 | } |
| 303 | |
| 304 | Status EagerServiceImpl::WaitQueueDone(const WaitQueueDoneRequest* request, |
| 305 | WaitQueueDoneResponse* response) { |