// Make a copy so that we can mutate it.
    FunctionDef fdef_to_load = fdef;
    if (mutate_proto_func) {
      (*mutate_proto_func)(&fdef_to_load);
    }
    VLOG(1) << "Adding func to graph: " << fdef_to_load.DebugString();
    std::vector<char> binary_proto_buf(fdef_to_load.ByteSizeLong());
    fdef_to_load.SerializeToArray(binary_proto_buf.data(),
                                  binary_proto_buf.size());

  CheckTFE_TensorHandleHasFloats(value_handle, {2});
  TFE_DeleteTensorHandle(value_handle);

  // Start a new worker to replace task:1
  ReplaceTaskInServerDef(&server_def, 1);
  server_def.set_task_index(1);
  // TODO(b/136478427): Figure out how to correctly shut the server down.
  worker_server.release();
  ASSERT_TRUE(tensorflow::GrpcServer::Create(

} PosixFile;

static void Cleanup(TF_RandomAccessFile* file) {
  auto posix_file = static_cast<PosixFile*>(file->plugin_file);
  close(posix_file->fd);
  // This would be safe to free using `free` directly as it is only opaque.
  // However, it is better to be consistent everywhere.
  plugin_memory_free(const_cast<char*>(posix_file->filename));
  delete posix_file;
}

        immediate_execution_ctx_.get(), 2.0f, &y_raw);
    ASSERT_EQ(errors::OK, status_.code()) << status_.message();
    y.reset(y_raw);
  }

  // TODO(srbs): Rename ops::Add to ops::AddV2 and AddRegister to
  // AddV2Registerer.
  status_ = registry_.Register("AddV2", AddRegisterer);
  ASSERT_EQ(errors::OK, status_.code()) << status_.message();

  ASSERT_NO_FATAL_FAILURE(CompareNumericalAndAutodiffGradients(

  // Note that future collectives with the same context do not work at the
  // moment; once canceled, the collective executor requires the program to be
  // restarted / context to be reset.
}

TEST(PARALLEL_DEVICE_LIB, TestDifferentShapes) {
  std::unique_ptr<TF_Status, decltype(&TF_DeleteStatus)> status(
      TF_NewStatus(), TF_DeleteStatus);

  return result;
}

// Used as an argument to TFE_NewCustomDeviceTensorHandle, indicating how
// ParallelTensors wrapped in TFE_TensorHandles should be cleaned up once their
// reference counts drop to zero.
void ParallelTensorDeallocator(void* data) {
  delete reinterpret_cast<ParallelTensor*>(data);
}

// Used as an argument to TFE_NewCustomDeviceTensorHandle, for computing the

 public:
  void operator()(TFE_Op* to_delete) const { TFE_DeleteOp(to_delete); }
};

using OpPtr = std::unique_ptr<TFE_Op, OpDeleter>;

class StatusDeleter {
 public:
  void operator()(TF_Status* to_delete) const { TF_DeleteStatus(to_delete); }
};

using StatusPtr = std::unique_ptr<TF_Status, StatusDeleter>;

class ExecutorDeleter {
 public:
  void operator()(TFE_Executor* to_delete) const {

  // Create a variable handle (uninitialized to start) placed on the parallel
  // device.
  std::function<void(Variable*)> variable_deleter = [&](Variable* to_delete) {
    to_delete->Destroy(context, status.get());
    ASSERT_TRUE(TF_GetCode(status.get()) == TF_OK) << TF_Message(status.get());
    delete to_delete;
  };
  std::unique_ptr<Variable, decltype(variable_deleter)> variable(

    control_output_nodes.push_back(&control_outputs[i]->node);
  }

  // Do the actual function creation.
  DCHECK(append_hash_to_fn_name <= 1);
  tensorflow::FunctionDef fdef;
  status->status = tensorflow::GraphToFunctionDef(
      fn_body->graph, fn_name, append_hash_to_fn_name != 0,
      /*set_stateful_from_nodes=*/true,
      /*copy_placeholder_attrs_from_nodes=*/true, body_nodes, input_tensors,

 private:
  bool first_call_ = true;
};

// After the graph pass is registered, it takes effect globally and can affect
// other test cases. Define a static variable to switch it on and off.
bool GraphErrorInjectionPass::enabled_ = false;

// Test to ensure that a registered graph optimization pass is only executed
// once (i.e., on the main function side) in running distributed functions.