Method QueueThread

tensorflow/compiler/xrt/client/xrt_tf_client.cc:153–203

}

void XrtTfContext::QueueThread() {
  auto should_flush_queue = [this]() {
    mu_.AssertHeld();  // For annotalysis.
    return enqueue_request_->queue_size() > options_.max_queue_size ||
           flush_requested_ || shutting_down_;
  };
  while (true) {
    auto request = absl::make_unique<eager::EnqueueRequest>();
    {
      absl::MutexLock lock(&mu_);
      // To keep the connection alive, make sure we send an EnqueueRequest
      // regularly, currently every 5 seconds.
      mu_.AwaitWithTimeout(absl::Condition(&should_flush_queue),
                           absl::Seconds(5));
      if (shutting_down_) break;
      std::swap(request, enqueue_request_);
      flush_requested_ = false;
    }

    std::vector<OperationId> op_ids;
    for (const auto& item : request->queue()) {
      if (item.has_operation()) {
        op_ids.push_back(item.operation().id());
      }
    }
    request->set_context_id(context_id_);

    VLOG(10) << "Enqueue:\n" << request->DebugString();
    eager::EnqueueResponse response;
    Status status;
    absl::Notification done;
    eager_client_->EnqueueAsync(request.get(), &response, [&](Status s) {
      status = s;
      done.Notify();
    });

    done.WaitForNotification();

    VLOG(10) << "EnqueueResponse: " << status << "\n" << response.DebugString();
    {
      absl::MutexLock lock(&mu_);
      if (status.ok()) {
        for (OperationId op_id : op_ids) {
          DeleteOperation(op_id);
        }
      } else {
        ReportError(op_ids, status);
      }
    }
  }
}

void XrtTfContext::ReportError(absl::Span<const OperationId> op_ids,
                               Status status) {