// instanceSelector retrieves the data instance given its complete string name and
// forwards the request to that instance's HTTP handler.
func instanceSelector(c *web.C, h http.Handler) http.Handler {
	fn := func(w http.ResponseWriter, r *http.Request) {
		var err error
		dataname := dvid.InstanceName(c.URLParams["dataname"])
		uuid, ok := c.Env["uuid"].(dvid.UUID)
		if !ok {
			msg := fmt.Sprintf("Bad format for UUID %q\n", c.Env["uuid"])
			BadRequest(w, r, msg)
			return
		}
		data, err := datastore.GetDataByUUID(uuid, dataname)
		if err != nil {
			BadRequest(w, r, err)
			return
		}
		v, err := datastore.VersionFromUUID(uuid)
		if err != nil {
			BadRequest(w, r, err)
		}
		ctx := datastore.NewVersionedCtx(data, v)

		// Handle DVID-wide query string commands like non-interactive call designations
		queryValues := r.URL.Query()

		// All HTTP requests are interactive so let server tally request.
		interactive := queryValues.Get("interactive")
		if interactive == "" || (interactive != "false" && interactive != "0") {
			GotInteractiveRequest()
		}

		// TODO: setup routing for data instances as well.
		data.ServeHTTP(uuid, ctx, w, r)
	}
	return http.HandlerFunc(fn)
}

func repoDeleteInstanceHandler(c web.C, w http.ResponseWriter, r *http.Request) {
	uuid := c.Env["uuid"].(dvid.UUID)
	queryValues := r.URL.Query()
	imsure := queryValues.Get("imsure")
	if imsure != "true" {
		BadRequest(w, r, "Cannot delete instance unless query string 'imsure=true' is present!")
		return
	}

	dataname, ok := c.URLParams["dataname"]
	if !ok {
		BadRequest(w, r, "Error retrieving data instance name from URL parameters")
		return
	}

	// Make sure this instance exists.
	_, err := datastore.GetDataByUUID(uuid, dvid.InstanceName(dataname))
	if err != nil {
		BadRequest(w, r, "Error trying to delete %q for UUID %s: %v", dataname, uuid, err)
		return
	}

	// Do the deletion.  Under hood, modifies metadata immediately and launches async k/v deletion.
	if err := datastore.DeleteDataByUUID(uuid, dvid.InstanceName(dataname)); err != nil {
		BadRequest(w, r, "Error deleting data instance %q: %v", dataname, err)
		return
	}

	// Just respond that deletion was successfully started
	w.Header().Set("Content-Type", "application/json")
	fmt.Fprintf(w, `{"result": "Started deletion of data instance %q from repo with root %s"}`,
		dataname, uuid)
}

// getBlankTileData returns zero 2d tile data with a given scaling and format.
func (d *Data) getBlankTileImage(uuid dvid.UUID, shape dvid.DataShape, scaling Scaling) (image.Image, error) {
	levelSpec, found := d.Levels[scaling]
	if !found {
		return nil, fmt.Errorf("Could not extract tiles for unspecified scale level %d", scaling)
	}
	tileW, tileH, err := shape.GetSize2D(levelSpec.TileSize)
	if err != nil {
		return nil, err
	}
	source, err := datastore.GetDataByUUID(uuid, d.Source)
	if err != nil {
		return nil, err
	}
	src, ok := source.(*imageblk.Data)
	if !ok {
		return nil, fmt.Errorf("Data instance %q for uuid %q is not imageblk.Data", d.Source, uuid)
	}
	bytesPerVoxel := src.Values.BytesPerElement()
	switch bytesPerVoxel {
	case 1, 2, 4, 8:
		numBytes := tileW * tileH * bytesPerVoxel
		data := make([]byte, numBytes, numBytes)
		return dvid.GoImageFromData(data, int(tileW), int(tileH))
	default:
		return nil, fmt.Errorf("Cannot construct blank tile for data %q with %d bytes/voxel",
			d.Source, src.Values.BytesPerElement())
	}
}

func TestDeleteInstance(t *testing.T) {
	tests.UseStore()
	defer tests.CloseStore()

	apiStr := fmt.Sprintf("%srepos", server.WebAPIPath)
	r := server.TestHTTP(t, "POST", apiStr, nil)
	var jsonResp map[string]interface{}

	if err := json.Unmarshal(r, &jsonResp); err != nil {
		t.Fatalf("Unable to unmarshal repo creation response: %s\n", string(r))
	}
	v, ok := jsonResp["root"]
	if !ok {
		t.Fatalf("No 'root' metadata returned: %s\n", string(r))
	}
	uuidStr, ok := v.(string)
	if !ok {
		t.Fatalf("Couldn't cast returned 'root' data (%v) into string.\n", v)
	}
	uuid := dvid.UUID(uuidStr)

	// Add a data instance.
	var config dvid.Config
	server.CreateTestInstance(t, uuid, "keyvalue", "foo", config)

	// Make sure it exists.
	_, err := datastore.GetDataByUUID(uuid, "foo")
	if err != nil {
		t.Errorf("Couldn't create data instance 'foo'\n")
	}

	// Shouldn't be able to delete instance without "imsure"
	delReq := fmt.Sprintf("%srepo/%s/%s", server.WebAPIPath, uuid, "foo")
	server.TestBadHTTP(t, "DELETE", delReq, nil)
	delReq = fmt.Sprintf("%srepo/%s/%s?imsure=true", server.WebAPIPath, uuid, "foo")
	server.TestHTTP(t, "DELETE", delReq, nil)

	// Make sure it no longer exists.
	_, err = datastore.GetDataByUUID(uuid, "foo")
	if err == nil {
		t.Errorf("Shouldn't be able to access a deleted data instance 'foo'\n")
	}
}

// GetByUUID returns a pointer to labelsurf data given a version (UUID) and data name.
func GetByUUID(uuid dvid.UUID, name dvid.InstanceName) (*Data, error) {
	source, err := datastore.GetDataByUUID(uuid, name)
	if err != nil {
		return nil, err
	}
	data, ok := source.(*Data)
	if !ok {
		return nil, fmt.Errorf("Instance '%s' is not a labelsurf datatype!", name)
	}
	return data, nil
}

func TestMultiscale2dRepoPersistence(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	// Make source
	uuid, _ := initTestRepo()
	makeGrayscale(uuid, t, "grayscale")

	// Make labels and set various properties
	config := dvid.NewConfig()
	config.Set("Placeholder", "true")
	config.Set("Format", "jpg")
	config.Set("Source", "grayscale")
	dataservice, err := datastore.NewData(uuid, mstype, "myimagetile", config)
	if err != nil {
		t.Errorf("Unable to create imagetile instance: %v\n", err)
	}
	msdata, ok := dataservice.(*Data)
	if !ok {
		t.Fatalf("Can't cast imagetile data service into imagetile.Data\n")
	}
	oldData := *msdata

	// Restart test datastore and see if datasets are still there.
	if err = datastore.SaveDataByUUID(uuid, msdata); err != nil {
		t.Fatalf("Unable to save repo during imagetile persistence test: %v\n", err)
	}
	datastore.CloseReopenTest()

	dataservice2, err := datastore.GetDataByUUID(uuid, "myimagetile")
	if err != nil {
		t.Fatalf("Can't get keyvalue instance from reloaded test db: %v\n", err)
	}
	msdata2, ok := dataservice2.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 2 is not imagetile.Data\n")
	}

	if !reflect.DeepEqual(oldData.Properties, msdata2.Properties) {
		t.Errorf("Expected properties %v, got %v\n", oldData.Properties, msdata2.Properties)
	}
}

func TestLabelblkRepoPersistence(t *testing.T) {
	tests.UseStore()
	defer tests.CloseStore()

	uuid, _ := initTestRepo()

	// Make labels and set various properties
	config := dvid.NewConfig()
	config.Set("BlockSize", "12,13,14")
	config.Set("VoxelSize", "1.1,2.8,11")
	config.Set("VoxelUnits", "microns,millimeters,nanometers")
	dataservice, err := datastore.NewData(uuid, labelsT, "mylabels", config)
	if err != nil {
		t.Errorf("Unable to create labels instance: %v\n", err)
	}
	labels, ok := dataservice.(*Data)
	if !ok {
		t.Errorf("Can't cast labels data service into Data\n")
	}
	oldData := *labels

	// Restart test datastore and see if datasets are still there.
	if err = datastore.SaveDataByUUID(uuid, labels); err != nil {
		t.Fatalf("Unable to save repo during labels persistence test: %v\n", err)
	}
	tests.CloseReopenStore()

	dataservice2, err := datastore.GetDataByUUID(uuid, "mylabels")
	if err != nil {
		t.Fatalf("Can't get labels instance from reloaded test db: %v\n", err)
	}
	labels2, ok := dataservice2.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 2 is not imageblk.Data\n")
	}
	if !oldData.Equals(labels2) {
		t.Errorf("Expected %v, got %v\n", oldData, *labels2)
	}
}

func TestGrayscaleRepoPersistence(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	uuid, _ := initTestRepo()

	// Make grayscale and set various properties
	config := dvid.NewConfig()
	config.Set("BlockSize", "12,13,14")
	config.Set("VoxelSize", "1.1,2.8,11")
	config.Set("VoxelUnits", "microns,millimeters,nanometers")
	dataservice, err := datastore.NewData(uuid, grayscaleT, "mygrayscale", config)
	if err != nil {
		t.Errorf("Unable to create grayscale instance: %s\n", err)
	}
	grayscale, ok := dataservice.(*Data)
	if !ok {
		t.Errorf("Can't cast uint8 data service into Data\n")
	}
	oldData := *grayscale

	// Restart test datastore and see if datasets are still there.
	if err = datastore.SaveDataByUUID(uuid, grayscale); err != nil {
		t.Fatalf("Unable to save repo during grayscale persistence test: %v\n", err)
	}
	datastore.CloseReopenTest()

	dataservice2, err := datastore.GetDataByUUID(uuid, "mygrayscale")
	if err != nil {
		t.Fatalf("Can't get grayscale instance from reloaded test db: %v\n", err)
	}
	grayscale2, ok := dataservice2.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 2 is not imageblk.Data\n")
	}
	if !oldData.Equals(grayscale2) {
		t.Errorf("Expected %v, got %v\n", oldData, *grayscale2)
	}
}

func TestMultichan16RepoPersistence(t *testing.T) {
	tests.UseStore()
	defer tests.CloseStore()

	uuid, _ := initTestRepo()

	// Make labels and set various properties
	config := dvid.NewConfig()
	dataservice, err := datastore.NewData(uuid, dtype, "mymultichan16", config)
	if err != nil {
		t.Errorf("Unable to create multichan16 instance: %v\n", err)
	}
	mcdata, ok := dataservice.(*Data)
	if !ok {
		t.Errorf("Can't cast multichan16 data service into multichan16.Data\n")
	}
	oldData := *mcdata

	// Restart test datastore and see if datasets are still there.
	if err = datastore.SaveDataByUUID(uuid, mcdata); err != nil {
		t.Fatalf("Unable to save repo during multichan16 persistence test: %v\n", err)
	}
	tests.CloseReopenStore()

	dataservice2, err := datastore.GetDataByUUID(uuid, "mymultichan16")
	if err != nil {
		t.Fatalf("Can't get multichan16 instance from reloaded test db: %v\n", err)
	}
	mcdata2, ok := dataservice2.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 2 is not multichan16.Data\n")
	}
	if !oldData.Equals(mcdata2) {
		t.Errorf("Expected %v, got %v\n", oldData, *mcdata2)
	}
}

func TestKeyvalueRepoPersistence(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	uuid, _ := initTestRepo()

	// Make labels and set various properties
	config := dvid.NewConfig()
	dataservice, err := datastore.NewData(uuid, kvtype, "mykv", config)
	if err != nil {
		t.Errorf("Unable to create keyvalue instance: %v\n", err)
	}
	kvdata, ok := dataservice.(*Data)
	if !ok {
		t.Errorf("Can't cast keyvalue data service into keyvalue.Data\n")
	}
	oldData := *kvdata

	// Restart test datastore and see if datasets are still there.
	if err = datastore.SaveDataByUUID(uuid, kvdata); err != nil {
		t.Fatalf("Unable to save repo during keyvalue persistence test: %v\n", err)
	}
	datastore.CloseReopenTest()

	dataservice2, err := datastore.GetDataByUUID(uuid, "mykv")
	if err != nil {
		t.Fatalf("Can't get keyvalue instance from reloaded test db: %v\n", err)
	}
	kvdata2, ok := dataservice2.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 2 is not keyvalue.Data\n")
	}
	if !oldData.Equals(kvdata2) {
		t.Errorf("Expected %v, got %v\n", oldData, *kvdata2)
	}
}

// ServeHTTP handles all incoming HTTP requests for this data.
func (d *Data) ServeHTTP(uuid dvid.UUID, ctx *datastore.VersionedCtx, w http.ResponseWriter, r *http.Request) {
	timedLog := dvid.NewTimeLog()

	action := strings.ToLower(r.Method)
	switch action {
	case "get", "post":
		// Acceptable
	default:
		server.BadRequest(w, r, "Can only handle GET or POST HTTP verbs")
		return
	}

	// Break URL request into arguments
	url := r.URL.Path[len(server.WebAPIPath):]
	parts := strings.Split(url, "/")
	if len(parts[len(parts)-1]) == 0 {
		parts = parts[:len(parts)-1]
	}
	if len(parts) < 4 {
		server.BadRequest(w, r, "incomplete API request")
		return
	}

	switch parts[3] {
	case "help":
		w.Header().Set("Content-Type", "text/plain")
		fmt.Fprintln(w, d.Help())

	case "info":
		jsonBytes, err := d.MarshalJSON()
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		fmt.Fprintf(w, string(jsonBytes))

	case "tile":
		if action == "post" {
			server.BadRequest(w, r, "DVID does not yet support POST of imagetile")
			return
		}
		if err := d.ServeTile(uuid, ctx, w, r, parts); err != nil {
			server.BadRequest(w, r, err)
			return
		}
		timedLog.Infof("HTTP %s: tile (%s)", r.Method, r.URL)

	case "raw", "isotropic":
		if action == "post" {
			server.BadRequest(w, r, "imagetile '%s' can only PUT tiles not images", d.DataName())
			return
		}
		if len(parts) < 7 {
			server.BadRequest(w, r, "%q must be followed by shape/size/offset", parts[3])
			return
		}
		shapeStr, sizeStr, offsetStr := parts[4], parts[5], parts[6]
		planeStr := dvid.DataShapeString(shapeStr)
		plane, err := planeStr.DataShape()
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
		if plane.ShapeDimensions() != 2 {
			server.BadRequest(w, r, "Quadtrees can only return 2d images not %s", plane)
			return
		}
		slice, err := dvid.NewSliceFromStrings(planeStr, offsetStr, sizeStr, "_")
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
		source, err := datastore.GetDataByUUID(uuid, d.Source)
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
		src, ok := source.(*imageblk.Data)
		if !ok {
			server.BadRequest(w, r, "Cannot construct imagetile for non-voxels data: %s", d.Source)
			return
		}
		img, err := d.GetImage(ctx, src, slice, parts[3] == "isotropic")
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
		var formatStr string
		if len(parts) >= 8 {
			formatStr = parts[7]
		}
		err = dvid.WriteImageHttp(w, img.Get(), formatStr)
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
		timedLog.Infof("HTTP %s: tile-accelerated %s %s (%s)", r.Method, planeStr, parts[3], r.URL)
	default:
//.........这里部分代码省略.........

// Returns the default tile spec that will fully cover the source extents and scaling 0
// uses the original voxel resolutions with each subsequent scale causing a 2x zoom out.
func (d *Data) DefaultTileSpec(uuidStr string) (TileSpec, error) {
	uuid, _, err := datastore.MatchingUUID(uuidStr)
	if err != nil {
		return nil, err
	}
	source, err := datastore.GetDataByUUID(uuid, d.Source)
	if err != nil {
		return nil, err
	}
	var ok bool
	var src *imageblk.Data
	src, ok = source.(*imageblk.Data)
	if !ok {
		return nil, fmt.Errorf("Cannot construct tile spec for non-voxels data: %s", d.Source)
	}

	// Set scaling 0 based on extents and resolution of source.
	//extents := src.Extents()
	resolution := src.Properties.Resolution.VoxelSize

	if len(resolution) != 3 {
		return nil, fmt.Errorf("Cannot construct tile spec for non-3d data: voxel is %d-d",
			len(resolution))
	}

	// Expand min and max points to coincide with full tile boundaries of highest resolution.
	minTileCoord := src.MinPoint.(dvid.Chunkable).Chunk(DefaultTileSize)
	maxTileCoord := src.MaxPoint.(dvid.Chunkable).Chunk(DefaultTileSize)
	minTiledPt := minTileCoord.MinPoint(DefaultTileSize)
	maxTiledPt := maxTileCoord.MaxPoint(DefaultTileSize)
	sizeVolume := maxTiledPt.Sub(minTiledPt).AddScalar(1)

	dvid.Infof("Creating default multiscale tile spec for volume of size %s\n", sizeVolume)

	// For each dimension, calculate the number of scaling levels necessary to cover extent,
	// assuming we use the raw resolution at scaling 0.
	numScales := make([]int, 3)
	var maxScales int
	var dim uint8
	for dim = 0; dim < 3; dim++ {
		numPixels := float64(sizeVolume.Value(dim))
		tileSize := float64(DefaultTileSize.Value(dim))
		if numPixels <= tileSize {
			numScales[dim] = 1
		} else {
			numScales[dim] = int(math.Ceil(math.Log2(numPixels/tileSize))) + 1
		}
		if numScales[dim] > maxScales {
			maxScales = numScales[dim]
		}
	}

	// Initialize the tile level specification
	specs := make(TileSpec, maxScales)
	curRes := resolution
	levelMag := dvid.Point3d{2, 2, 2}
	var scaling Scaling
	for scaling = 0; scaling < Scaling(maxScales); scaling++ {
		for dim = 0; dim < 3; dim++ {
			if scaling >= Scaling(numScales[dim]) {
				levelMag[dim] = 1
			}
		}
		specs[scaling] = TileScaleSpec{
			LevelSpec{curRes, DefaultTileSize},
			levelMag,
		}
		curRes = curRes.MultScalar(2.0)
	}
	return specs, nil
}

func (d *Data) ConstructTiles(uuidStr string, tileSpec TileSpec, request datastore.Request) error {
	config := request.Settings()
	uuid, versionID, err := datastore.MatchingUUID(uuidStr)
	if err != nil {
		return err
	}
	if err = datastore.AddToNodeLog(uuid, []string{request.Command.String()}); err != nil {
		return err
	}

	source, err := datastore.GetDataByUUID(uuid, d.Source)
	if err != nil {
		return err
	}
	src, ok := source.(*imageblk.Data)
	if !ok {
		return fmt.Errorf("Cannot construct imagetile for non-voxels data: %s", d.Source)
	}

	// Save the current tile specification
	d.Levels = tileSpec
	if err := datastore.SaveDataByUUID(uuid, d); err != nil {
		return err
	}

	// Get size of tile at lowest resolution.
	lastLevel := Scaling(len(tileSpec) - 1)
	loresSpec, found := tileSpec[lastLevel]
	if !found {
		return fmt.Errorf("Illegal tile spec.  Should have levels 0 to absent %d.", lastLevel)
	}
	var loresSize [3]float64
	for i := 0; i < 3; i++ {
		loresSize[i] = float64(loresSpec.Resolution[i]) * float64(DefaultTileSize[i])
	}
	loresMag := dvid.Point3d{1, 1, 1}
	for i := Scaling(0); i < lastLevel; i++ {
		levelMag := tileSpec[i].levelMag
		loresMag[0] *= levelMag[0]
		loresMag[1] *= levelMag[1]
		loresMag[2] *= levelMag[2]
	}

	// Get min and max points in terms of distance.
	var minPtDist, maxPtDist [3]float64
	for i := uint8(0); i < 3; i++ {
		minPtDist[i] = float64(src.MinPoint.Value(i)) * float64(src.VoxelSize[i])
		maxPtDist[i] = float64(src.MaxPoint.Value(i)) * float64(src.VoxelSize[i])
	}

	// Adjust min and max points for the tileable surface at lowest resolution.
	var minTiledPt, maxTiledPt dvid.Point3d
	for i := 0; i < 3; i++ {
		minInt, _ := math.Modf(minPtDist[i] / loresSize[i])
		maxInt, _ := math.Modf(maxPtDist[i] / loresSize[i])
		minTileCoord := int32(minInt)
		maxTileCoord := int32(maxInt)
		minTiledPt[i] = minTileCoord * DefaultTileSize[i] * loresMag[i]
		maxTiledPt[i] = (maxTileCoord+1)*DefaultTileSize[i]*loresMag[i] - 1
	}
	sizeVolume := maxTiledPt.Sub(minTiledPt).AddScalar(1)

	// Setup swappable ExtData buffers (the stitched slices) so we can be generating tiles
	// at same time we are reading and stitching them.
	var bufferLock [2]sync.Mutex
	var sliceBuffers [2]*imageblk.Voxels
	var bufferNum int

	// Get the planes we should tile.
	planes, err := config.GetShapes("planes", ";")
	if planes == nil {
		// If no planes are specified, construct imagetile for 3 orthogonal planes.
		planes = []dvid.DataShape{dvid.XY, dvid.XZ, dvid.YZ}
	}

	outF, err := d.putTileFunc(versionID)

	// sort the tile spec keys to iterate from highest to lowest resolution
	var sortedKeys []int
	for scaling, _ := range tileSpec {
		sortedKeys = append(sortedKeys, int(scaling))
	}
	sort.Ints(sortedKeys)

	for _, plane := range planes {
		timedLog := dvid.NewTimeLog()
		offset := minTiledPt.Duplicate()

		switch {

		case plane.Equals(dvid.XY):
			width, height, err := plane.GetSize2D(sizeVolume)
			if err != nil {
				return err
			}
			dvid.Debugf("Tiling XY image %d x %d pixels\n", width, height)
			for z := src.MinPoint.Value(2); z <= src.MaxPoint.Value(2); z++ {
				server.BlockOnInteractiveRequests("imagetile.ConstructTiles [xy]")

				sliceLog := dvid.NewTimeLog()
//.........这里部分代码省略.........

func repoResolveHandler(c web.C, w http.ResponseWriter, r *http.Request) {
	uuid, _, err := datastore.MatchingUUID(c.URLParams["uuid"])
	if err != nil {
		BadRequest(w, r, err)
		return
	}
	if r.Body == nil {
		BadRequest(w, r, "merge resolving requires JSON to be POSTed per API documentation")
		return
	}
	data, err := ioutil.ReadAll(r.Body)
	if err != nil {
		BadRequest(w, r, err)
		return
	}

	jsonData := struct {
		Data    []dvid.InstanceName `json:"data"`
		Note    string              `json:"note"`
		Parents []string            `json:"parents"`
	}{}
	if err := json.Unmarshal(data, &jsonData); err != nil {
		BadRequest(w, r, fmt.Sprintf("Malformed JSON request in body: %v", err))
		return
	}
	if len(jsonData.Data) == 0 {
		BadRequest(w, r, "Must specify at least one data instance using 'data' field")
		return
	}
	if len(jsonData.Parents) < 2 {
		BadRequest(w, r, "Must specify at least two parent UUIDs using 'parents' field")
		return
	}

	// Convert JSON of parents into []UUID and whether we need a child for them to add deletions.
	numParents := len(jsonData.Parents)
	oldParents := make([]dvid.UUID, numParents)
	newParents := make([]dvid.UUID, numParents, numParents) // UUID of any parent extension for deletions.
	for i, uuidFrag := range jsonData.Parents {
		uuid, _, err := datastore.MatchingUUID(uuidFrag)
		if err != nil {
			BadRequest(w, r, fmt.Sprintf("can't match parent %q: %v", uuidFrag, err))
			return
		}
		oldParents[i] = uuid
		newParents[i] = dvid.NilUUID
	}

	// Iterate through all k/v for given data instances, making sure we find any conflicts.
	// If any are found, remove them with first UUIDs taking priority.
	for _, name := range jsonData.Data {
		data, err := datastore.GetDataByUUID(uuid, name)
		if err != nil {
			BadRequest(w, r, err)
			return
		}

		if err := datastore.DeleteConflicts(uuid, data, oldParents, newParents); err != nil {
			BadRequest(w, r, fmt.Errorf("Conflict deletion error for data %q: %v", data.DataName(), err))
			return
		}
	}

	// If we have any new nodes to accomodate deletions, commit them.
	for i, oldUUID := range oldParents {
		if newParents[i] != oldUUID {
			err := datastore.Commit(newParents[i], "Version for deleting conflicts before merge", nil)
			if err != nil {
				BadRequest(w, r, "Error while creating new nodes to handle required deletions: %v", err)
				return
			}
		}
	}

	// Do the merge
	mt := datastore.MergeConflictFree
	newuuid, err := datastore.Merge(newParents, jsonData.Note, mt)
	if err != nil {
		BadRequest(w, r, err)
	} else {
		w.Header().Set("Content-Type", "application/json")
		fmt.Fprintf(w, "{%q: %q}", "child", newuuid)
	}
}

//.........这里部分代码省略.........
		if err != nil {
			return err
		}

		switch subcommand {
		case "new":
			var typename, dataname string
			cmd.CommandArgs(3, &typename, &dataname)

			// Get TypeService
			typeservice, err := datastore.TypeServiceByName(dvid.TypeString(typename))
			if err != nil {
				return err
			}

			// Create new data
			config := cmd.Settings()
			_, err = datastore.NewData(uuid, typeservice, dvid.InstanceName(dataname), config)
			if err != nil {
				return err
			}
			reply.Text = fmt.Sprintf("Data %q [%s] added to node %s\n", dataname, typename, uuid)
			datastore.AddToRepoLog(uuid, []string{cmd.String()})

		case "branch":
			cmd.CommandArgs(3, &uuidStr)

			var assign *dvid.UUID
			if uuidStr == "" {
				assign = nil
			} else {
				u := dvid.UUID(uuidStr)
				assign = &u
			}
			child, err := datastore.NewVersion(uuid, fmt.Sprintf("branch of %s", uuid), assign)
			if err != nil {
				return err
			}
			reply.Text = fmt.Sprintf("Branch %s added to node %s\n", child, uuid)
			datastore.AddToRepoLog(uuid, []string{cmd.String()})

		case "merge":
			uuids := cmd.CommandArgs(2)

			parents := make([]dvid.UUID, len(uuids)+1)
			parents[0] = dvid.UUID(uuid)
			i := 1
			for uuid := range uuids {
				parents[i] = dvid.UUID(uuid)
				i++
			}
			child, err := datastore.Merge(parents, fmt.Sprintf("merge of parents %v", parents), datastore.MergeConflictFree)
			if err != nil {
				return err
			}
			reply.Text = fmt.Sprintf("Parents %v merged into node %s\n", parents, child)
			datastore.AddToRepoLog(uuid, []string{cmd.String()})

		case "push":
			/*
				var target string
				cmd.CommandArgs(3, &target)
				config := cmd.Settings()
				if err = datastore.Push(repo, target, config); err != nil {
					return err
				}
				reply.Text = fmt.Sprintf("Repo %q pushed to %q\n", repo.RootUUID(), target)
			*/
			return fmt.Errorf("push command has been temporarily suspended")
		default:
			return fmt.Errorf("Unknown command: %q", cmd)
		}

	case "node":
		var uuidStr, descriptor string
		cmd.CommandArgs(1, &uuidStr, &descriptor)
		uuid, _, err := datastore.MatchingUUID(uuidStr)
		if err != nil {
			return err
		}

		// Get the DataService
		dataname := dvid.InstanceName(descriptor)
		var subcommand string
		cmd.CommandArgs(3, &subcommand)
		dataservice, err := datastore.GetDataByUUID(uuid, dataname)
		if err != nil {
			return err
		}
		if subcommand == "help" {
			reply.Text = dataservice.Help()
			return nil
		}
		return dataservice.DoRPC(cmd, reply)

	default:
		return fmt.Errorf("Unknown command: '%s'", cmd)
	}
	return nil
}

func TestROIRepoPersistence(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	uuid, _ := initTestRepo()

	// Add data
	config := dvid.NewConfig()
	dataservice1, err := datastore.NewData(uuid, roitype, "myroi", config)
	if err != nil {
		t.Errorf("Error creating new roi instance: %v\n", err)
	}
	roi1, ok := dataservice1.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 1 is not roi.Data\n")
	}
	if roi1.DataName() != "myroi" {
		t.Errorf("New roi data instance name set incorrectly: %q != %q\n",
			roi1.DataName(), "myroi")
	}

	config.Set("BlockSize", "15,16,17")
	dataservice2, err := datastore.NewData(uuid, roitype, "myroi2", config)
	if err != nil {
		t.Errorf("Error creating new roi instance: %v\n", err)
	}
	roi2, ok := dataservice2.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 2 is not roi.Data\n")
	}
	roi2.MinZ = 13
	roi2.MaxZ = 3098
	oldData := *roi2

	// Check instance IDs
	if roi1.InstanceID() == roi2.InstanceID() {
		t.Errorf("Instance IDs should be different: %d == %d\n",
			roi1.InstanceID(), roi2.InstanceID())
	}

	// Restart test datastore and see if datasets are still there.
	if err = datastore.SaveDataByUUID(uuid, dataservice1); err != nil {
		t.Fatalf("Unable to save data1 during ROI persistence test: %v\n", err)
	}
	if err = datastore.SaveDataByUUID(uuid, dataservice2); err != nil {
		t.Fatalf("Unable to save data2 during ROI persistence test: %v\n", err)
	}

	datastore.CloseReopenTest()

	dataservice3, err := datastore.GetDataByUUID(uuid, "myroi2")
	if err != nil {
		t.Fatalf("Can't get first ROI instance from reloaded test db: %v\n", err)
	}
	roi2new, ok := dataservice3.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 3 is not roi.Data\n")
	}
	if !oldData.Equals(roi2new) {
		t.Errorf("Expected %v, got %v\n", oldData, *roi2new)
	}
}

// CreateComposite creates a new rgba8 image by combining hash of labels + the grayscale
func (d *Data) CreateComposite(request datastore.Request, reply *datastore.Response) error {
	timedLog := dvid.NewTimeLog()

	// Parse the request
	var uuidStr, dataName, cmdStr, grayscaleName, destName string
	request.CommandArgs(1, &uuidStr, &dataName, &cmdStr, &grayscaleName, &destName)

	// Get the version
	uuid, v, err := datastore.MatchingUUID(uuidStr)
	if err != nil {
		return err
	}

	// Log request
	if err = datastore.AddToNodeLog(uuid, []string{request.Command.String()}); err != nil {
		return err
	}

	// Get the grayscale data.
	dataservice, err := datastore.GetDataByUUID(uuid, dvid.InstanceName(grayscaleName))
	if err != nil {
		return err
	}
	grayscale, ok := dataservice.(*imageblk.Data)
	if !ok {
		return fmt.Errorf("%s is not the name of uint8 data", grayscaleName)
	}

	// Create a new rgba8blk data.
	var compservice datastore.DataService
	compservice, err = datastore.GetDataByUUID(uuid, dvid.InstanceName(destName))
	if err == nil {
		return fmt.Errorf("Data instance with name %q already exists", destName)
	}
	typeService, err := datastore.TypeServiceByName("rgba8blk")
	if err != nil {
		return fmt.Errorf("Could not get rgba8 type service from DVID")
	}
	config := dvid.NewConfig()
	compservice, err = datastore.NewData(uuid, typeService, dvid.InstanceName(destName), config)
	if err != nil {
		return err
	}
	composite, ok := compservice.(*imageblk.Data)
	if !ok {
		return fmt.Errorf("Error: %s was unable to be set to rgba8 data", destName)
	}

	// Iterate through all labels and grayscale chunks incrementally in Z, a layer at a time.
	wg := new(sync.WaitGroup)
	op := &blockOp{grayscale, composite, v}
	chunkOp := &storage.ChunkOp{op, wg}

	store, err := storage.BigDataStore()
	if err != nil {
		return err
	}
	ctx := datastore.NewVersionedCtx(d, v)
	extents := d.Extents()
	blockBeg := imageblk.NewTKey(extents.MinIndex)
	blockEnd := imageblk.NewTKey(extents.MaxIndex)
	err = store.ProcessRange(ctx, blockBeg, blockEnd, chunkOp, storage.ChunkFunc(d.CreateCompositeChunk))
	wg.Wait()

	// Set new mapped data to same extents.
	composite.Properties.Extents = grayscale.Properties.Extents
	if err := datastore.SaveDataByUUID(uuid, composite); err != nil {
		dvid.Infof("Could not save new data '%s': %v\n", destName, err)
	}

	timedLog.Infof("Created composite of %s and %s", grayscaleName, destName)
	return nil
}