BasicTestsForTwoDevices(context.get(),
                          "/job:worker/replica:0/task:1/device:CPU:0",
                          "/job:worker/replica:0/task:2/device:CPU:0");

  worker_server1.release();
  worker_server2.release();
}

TEST(PARALLEL_DEVICE, TestAsyncCopyOff) {
  std::unique_ptr<TFE_ContextOptions, decltype(&TFE_DeleteContextOptions)> opts(
      TFE_NewContextOptions(), TFE_DeleteContextOptions);

  return wrap(t);
}

void TF_DeleteExecutionContext(TF_ExecutionContext* c) { unwrap(c)->Release(); }

TF_AbstractOp* TF_NewAbstractOp(TF_ExecutionContext* c) {
  return wrap((unwrap(c)->CreateOperation()));
}

void TF_DeleteAbstractOp(TF_AbstractOp* op) { unwrap(op)->Release(); }

void TF_DeleteAbstractTensor(TF_AbstractTensor* t) { unwrap(t)->Unref(); }

    status->status = MessageToBuffer(bundle.meta_graph_def, meta_graph_def);
    if (!status->status.ok()) return nullptr;
  }

  TF_Session* session = new TF_Session(bundle.session.release(), graph);

  graph->sessions[session] = "";
  session->last_num_graph_nodes = graph->graph.num_node_ids();
  return session;
#endif  // defined(IS_MOBILE_PLATFORM) || defined(IS_SLIM_BUILD)
}

  TFE_ExecutorWaitForAllPendingNodes(executor, status);
  ASSERT_EQ(TF_OK, TF_GetCode(status)) << TF_Message(status);

  // TODO(b/136478427): Figure out how to correctly shut the server down.
  worker_server.release();

  // Update the server def with a new set of names (worker instead of
  // localhost).
  tensorflow::ServerDef updated_server_def = GetServerDef("worker", 2);
  serialized = updated_server_def.SerializeAsString();

  TF_RETURN_IF_ERROR(ops::Exp(ctx, inputs[0], &exp_output, "Exp"));
  std::unique_ptr<GradientFunction> gradient_function(
      new PassThroughGradientFunction);
  tape.RecordOperation(inputs, {exp_output}, gradient_function.release());
  TF_RETURN_IF_ERROR(tape.ComputeGradient(ctx,
                                          /*targets*/ {exp_output},
                                          /*sources=*/inputs,

  TFE_DeleteExecutor(executor);
  TFE_DeleteContext(ctx);

  TF_DeleteStatus(status);

  // TODO(b/136478427): Figure out how to correctly shut the server down.
  worker_server.release();
}

TEST(CAPI, RemoteExecute) { TestRemoteExecute(false); }
TEST(CAPI, RemoteExecuteAsync) { TestRemoteExecute(true); }

void TestRemoteExecuteSilentCopiesOp(bool async, bool remote,

  ::tensorflow::Status s =
      ::tensorflow::Env::Default()->NewWritableFile(filename, &f);
  ::tensorflow::Set_TF_Status_from_Status(status, s);

  if (s.ok()) {
    *handle = reinterpret_cast<TF_WritableFileHandle*>(f.release());
  }
}

void TF_CloseWritableFile(TF_WritableFileHandle* handle, TF_Status* status) {
  auto* cc_file = reinterpret_cast<::tensorflow::WritableFile*>(handle);
  TF_SetStatus(status, TF_OK, "");

  forward_op_->attrs.BuildNodeDef();
  std::unique_ptr<GradientFunction> gradient_fn;
  TF_RETURN_IF_ERROR(registry.Lookup(*forward_op_, &gradient_fn));
  tape->RecordOperation(forward_op_->inputs, retvals, gradient_fn.release(),
                        op_->Name());
  return absl::OkStatus();
}
}  // namespace internal

}  // namespace gradients

    if (forward_input) {
      // Since the input was forwarded, we released the input handle right away
      // and hence expect the input to be forwarded to the return tensor.
      EXPECT_EQ(orig_ptr, TF_TensorData(t));
    } else {
      // In async mode we expect forwarding to work without releasing the input
      // handle since by the time the kernel is executed we have released the
      // handle in the client code.

      parent_op_(parent_op),
      tape_(tape),
      registry_(registry) {
  // TODO(b/172003047): Consider making AbstractOperation RefCounted.
  // parent_op_->Ref();
}
void TapeOperation::Release() {
  // TODO(srbs): Change to Unref().
  delete this;
}
TapeOperation::~TapeOperation() {
  // TODO(b/172003047): Consider making AbstractOperation RefCounted.
  // parent_op->Unref();
}