// call KVAutobus key_range API
func (db *KVAutobus) getKeyRange(kStart, kEnd storage.Key) (Ks, error) {
	b64key1 := encodeKey(kStart)
	b64key2 := encodeKey(kEnd)
	url := fmt.Sprintf("%s/kvautobus/api/key_range/%s/%s/%s/", db.host, db.collection, b64key1, b64key2)

	timedLog := dvid.NewTimeLog()
	resp, err := db.client.Get(url)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusNotFound {
		return nil, nil // Handle no keys found.
	}

	r := msgp.NewReader(bufio.NewReader(resp.Body))
	var mks Ks
	if err := mks.DecodeMsg(r); err != nil {
		return nil, err
	}
	for _, mk := range mks {
		storage.StoreKeyBytesRead <- len(mk)
	}
	timedLog.Infof("PROXY key_range to %s returned %d (%d keys)\n", db.host, resp.StatusCode, len(mks))
	return mks, nil
}

// check if any metadata has been written into this store.
func (db *KVAutobus) metadataExists() (bool, error) {
	var ctx storage.MetadataContext
	kStart, kEnd := ctx.KeyRange()
	b64key1 := encodeKey(kStart)
	b64key2 := encodeKey(kEnd)
	url := fmt.Sprintf("%s/kvautobus/api/key_range/%s/%s/%s/", db.host, db.collection, b64key1, b64key2)
	dvid.Infof("metdataExists: doing GET on %s\n", url)

	timedLog := dvid.NewTimeLog()
	resp, err := db.client.Get(url)
	if err != nil {
		return false, err
	}
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusNotFound {
		return true, nil // Handle no keys found.
	}

	r := msgp.NewReader(bufio.NewReader(resp.Body))
	var mks Ks
	if err := mks.DecodeMsg(r); err != nil {
		return false, err
	}
	timedLog.Infof("PROXY key_range metadata to %s returned %d (%d keys)\n", db.host, resp.StatusCode, len(mks))
	if len(mks) == 0 {
		return false, nil
	}
	return true, nil
}

// call KVAutobus keyvalue_range API
func (db *KVAutobus) getKVRange(kStart, kEnd storage.Key) (KVs, error) {
	b64key1 := encodeKey(kStart)
	b64key2 := encodeKey(kEnd)
	url := fmt.Sprintf("%s/kvautobus/api/keyvalue_range/%s/%s/", db.host, b64key1, b64key2)

	timedLog := dvid.NewTimeLog()
	resp, err := http.Get(url)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusNotFound {
		return nil, nil // Handle no keys found.
	}

	r := msgp.NewReader(bufio.NewReader(resp.Body))
	var mkvs KVs
	if err := mkvs.DecodeMsg(r); err != nil {
		dvid.Errorf("Couldn't decode getKVRange return\n")
		return nil, err
	}

	timedLog.Infof("PROXY keyvalue_range to %s returned %d (%d kv pairs)\n", db.host, resp.StatusCode, len(mkvs))
	return mkvs, nil
}

// GetVoxels copies voxels from the storage engine to Voxels, a requested subvolume or 2d image.
func (d *Data) GetVoxels(v dvid.VersionID, vox *Voxels, r *ROI) error {
	timedLog := dvid.NewTimeLog()
	defer timedLog.Infof("GetVoxels %s", vox)

	store, err := storage.MutableStore()
	if err != nil {
		return fmt.Errorf("Data type imageblk had error initializing store: %v\n", err)
	}

	// Only do one request at a time, although each request can start many goroutines.
	server.SpawnGoroutineMutex.Lock()
	defer server.SpawnGoroutineMutex.Unlock()

	ctx := datastore.NewVersionedCtx(d, v)

	wg := new(sync.WaitGroup)
	for it, err := vox.IndexIterator(d.BlockSize()); err == nil && it.Valid(); it.NextSpan() {
		indexBeg, indexEnd, err := it.IndexSpan()
		if err != nil {
			return err
		}
		begTKey := NewTKey(indexBeg)
		endTKey := NewTKey(indexEnd)

		// Get set of blocks in ROI if ROI provided
		var chunkOp *storage.ChunkOp
		if r != nil && r.Iter != nil {
			ptBeg := indexBeg.Duplicate().(dvid.ChunkIndexer)
			ptEnd := indexEnd.Duplicate().(dvid.ChunkIndexer)
			begX := ptBeg.Value(0)
			endX := ptEnd.Value(0)

			blocksInROI := make(map[string]bool, (endX - begX + 1))
			c := dvid.ChunkPoint3d{begX, ptBeg.Value(1), ptBeg.Value(2)}
			for x := begX; x <= endX; x++ {
				c[0] = x
				curIndex := dvid.IndexZYX(c)
				if r.Iter.InsideFast(curIndex) {
					indexString := string(curIndex.Bytes())
					blocksInROI[indexString] = true
				}
			}
			chunkOp = &storage.ChunkOp{&getOperation{vox, blocksInROI, r.attenuation}, wg}
		} else {
			chunkOp = &storage.ChunkOp{&getOperation{vox, nil, 0}, wg}
		}

		// Send the entire range of key-value pairs to chunk processor
		err = store.ProcessRange(ctx, begTKey, endTKey, chunkOp, storage.ChunkFunc(d.ReadChunk))
		if err != nil {
			return fmt.Errorf("Unable to GET data %s: %v", ctx, err)
		}
	}
	if err != nil {
		return err
	}
	wg.Wait()
	return nil
}

// 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":
		// Acceptable
	default:
		server.BadRequest(w, r, "googlevoxels can only handle GET HTTP verbs at this time")
		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 err := d.ServeTile(w, r, parts); err != nil {
			server.BadRequest(w, r, err)
			return
		}
		timedLog.Infof("HTTP %s: tile (%s)", r.Method, r.URL)

	case "raw":
		if err := d.ServeImage(w, r, parts); err != nil {
			server.BadRequest(w, r, err)
			return
		}
		timedLog.Infof("HTTP %s: image (%s)", r.Method, r.URL)
	default:
		server.BadRequest(w, r, "Illegal request for googlevoxels data.  See 'help' for REST API")
	}
}

func (d *Data) processSplit(v dvid.VersionID, delta labels.DeltaSplit) {
	timedLog := dvid.NewTimeLog()

	mutID := d.NewMutationID()
	splitOpStart := labels.DeltaSplitStart{delta.OldLabel, delta.NewLabel}
	splitOpEnd := labels.DeltaSplitEnd{delta.OldLabel, delta.NewLabel}
	iv := dvid.InstanceVersion{d.DataUUID(), v}
	labels.SplitStart(iv, splitOpStart)

	d.StartUpdate()
	if delta.Split == nil {
		// Coarse Split
		for _, izyxStr := range delta.SortedBlocks {
			n := izyxStr.Hash(numBlockHandlers)
			d.MutAdd(mutID)
			op := splitOp{
				mutID:    mutID,
				oldLabel: delta.OldLabel,
				newLabel: delta.NewLabel,
				block:    izyxStr,
			}
			d.procCh[n] <- procMsg{op: op, v: v}
		}
	} else {
		// Fine Split
		for izyxStr, blockRLEs := range delta.Split {
			n := izyxStr.Hash(numBlockHandlers)
			d.MutAdd(mutID)
			op := splitOp{
				mutID:    mutID,
				oldLabel: delta.OldLabel,
				newLabel: delta.NewLabel,
				rles:     blockRLEs,
				block:    izyxStr,
			}
			d.procCh[n] <- procMsg{op: op, v: v}
		}
	}
	// Wait for all blocks to be split then mark end of split op.
	go func() {
		d.MutWait(mutID)
		d.MutDelete(mutID)
		labels.SplitStop(iv, splitOpEnd)
		timedLog.Debugf("labelblk sync complete for split of %d -> %d", delta.OldLabel, delta.NewLabel)
		d.StopUpdate()
		d.publishDownresCommit(v, mutID)
	}()
}

func (db *KVAutobus) deleteRange(kStart, kEnd storage.Key) error {
	b64key1 := encodeKey(kStart)
	b64key2 := encodeKey(kEnd)
	url := fmt.Sprintf("%s/kvautobus/api/keyvalue_range/%s/%s/%s/", db.host, db.collection, b64key1, b64key2)

	timedLog := dvid.NewTimeLog()
	req, err := http.NewRequest("DELETE", url, nil)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	resp, err := db.client.Do(req)
	if err != nil {
		return err
	}
	timedLog.Infof("PROXY delete keyvalue_range to %s returned %d\n", db.host, resp.StatusCode)
	return nil
}

// LoadImages bulk loads images using different techniques if it is a multidimensional
// file like HDF5 or a sequence of PNG/JPG/TIF images.
func (d *Data) LoadImages(v dvid.VersionID, offset dvid.Point, filenames []string) error {
	if len(filenames) == 0 {
		return nil
	}
	timedLog := dvid.NewTimeLog()

	// We only want one PUT on given version for given data to prevent interleaved
	// chunk PUTs that could potentially overwrite slice modifications.
	ctx := storage.NewDataContext(d, v)
	loadMutex := ctx.Mutex()
	loadMutex.Lock()

	// Handle cleanup given multiple goroutines still writing data.
	load := &bulkLoadInfo{filenames: filenames, versionID: v, offset: offset}
	defer func() {
		loadMutex.Unlock()

		if load.extentChanged.Value() {
			err := datastore.SaveDataByVersion(v, d)
			if err != nil {
				dvid.Errorf("Error in trying to save repo for voxel extent change: %v\n", err)
			}
		}
	}()

	// Use different loading techniques if we have a potentially multidimensional HDF5 file
	// or many 2d images.
	var err error
	if dvid.Filename(filenames[0]).HasExtensionPrefix("hdf", "h5") {
		err = d.loadHDF(load)
	} else {
		err = d.loadXYImages(load)
	}

	if err != nil {
		timedLog.Infof("RPC load of %d files had error: %v\n", err)
	} else {
		timedLog.Infof("RPC load of %d files completed.\n", len(filenames))
	}
	return err
}

func (d *Data) processMerge(v dvid.VersionID, delta labels.DeltaMerge) {
	timedLog := dvid.NewTimeLog()
	d.StartUpdate()

	mutID := d.NewMutationID()
	for izyxStr := range delta.Blocks {
		n := izyxStr.Hash(numBlockHandlers)
		d.MutAdd(mutID)
		op := mergeOp{mutID: mutID, MergeOp: delta.MergeOp, block: izyxStr}
		d.procCh[n] <- procMsg{op: op, v: v}
	}
	// When we've processed all the delta blocks, we can remove this merge op
	// from the merge cache since all labels will have completed.
	go func() {
		d.MutWait(mutID)
		d.MutDelete(mutID)
		timedLog.Debugf("labelblk sync complete for merge (%d blocks) of %s -> %d", len(delta.Blocks), delta.MergeOp.Merged, delta.MergeOp.Target)
		d.StopUpdate()
		d.publishDownresCommit(v, mutID)
	}()
}

// call KVAutobus keyvalue_range API
func (db *KVAutobus) getKVRange(ctx storage.Context, kStart, kEnd storage.Key) (KVs, error) {
	// Get any request context and pass to kvautobus for tracking.
	reqctx, ok := ctx.(storage.RequestCtx)
	var reqID string
	if ok {
		parts := strings.Split(reqctx.GetRequestID(), "/")
		if len(parts) > 0 {
			reqID = parts[len(parts)-1]
		}
	}

	// Construct the KVAutobus URL
	b64key1 := encodeKey(kStart)
	b64key2 := encodeKey(kEnd)
	url := fmt.Sprintf("%s/kvautobus/api/keyvalue_range/%s/%s/%s/?=%s", db.host, db.collection, b64key1, b64key2, reqID)

	timedLog := dvid.NewTimeLog()
	resp, err := db.client.Get(url)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusNotFound {
		return nil, nil // Handle no keys found.
	}

	r := msgp.NewReader(bufio.NewReader(resp.Body))
	var mkvs KVs
	if err := mkvs.DecodeMsg(r); err != nil {
		dvid.Errorf("Couldn't decode getKVRange return\n")
		return nil, err
	}
	for _, mkv := range mkvs {
		storage.StoreKeyBytesRead <- len(mkv[0])
		storage.StoreValueBytesRead <- len(mkv[1])
	}

	timedLog.Infof("[%s] PROXY keyvalue_range to %s returned %d (%d kv pairs)\n", reqID, db.host, resp.StatusCode, len(mkvs))
	return mkvs, nil
}

func (db *KVAutobus) RawGet(key storage.Key) ([]byte, error) {
	b64key := encodeKey(key)
	url := fmt.Sprintf("%s/kvautobus/api/value/%s/", db.host, b64key)

	timedLog := dvid.NewTimeLog()
	resp, err := http.Get(url)
	if err != nil {
		return nil, err
	}
	timedLog.Infof("PROXY get to %s returned %d\n", db.host, resp.StatusCode)
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusNotFound {
		return nil, nil // Handle no key found.
	}

	r := msgp.NewReader(bufio.NewReader(resp.Body))
	var bin Binary
	if err := bin.DecodeMsg(r); err != nil {
		return nil, err
	}
	return []byte(bin), nil
}

func (d *Data) serveVolume(w http.ResponseWriter, r *http.Request, geom *GoogleSubvolGeom, noblanks bool) error {
	// If it's outside, write blank tile unless user wants no blanks.
	if geom.outside {
		if noblanks {
			http.NotFound(w, r)
			return fmt.Errorf("Requested subvolume is outside of available volume.")
		}
		return nil
	}

	// If we are within volume, get data from Google.
	url, err := geom.GetURL(d.VolumeID, "")
	if err != nil {
		return err
	}

	timedLog := dvid.NewTimeLog()
	client, err := d.GetClient()
	if err != nil {
		dvid.Errorf("Can't get OAuth2 connection to Google: %v\n", err)
		return err
	}
	resp, err := client.Get(url)
	if err != nil {
		return err
	}
	timedLog.Infof("PROXY HTTP to Google: %s, returned response %d", url, resp.StatusCode)
	defer resp.Body.Close()

	// If it's on edge, we need to pad the subvolume to the requested size.
	if geom.edge {
		return fmt.Errorf("Googlevoxels subvolume GET does not pad data on edge at this time")
	}

	// If we aren't on edge or outside, our return status should be OK.
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("Unexpected status code %d on volume request (%q, volume id %q)", resp.StatusCode, d.DataName(), d.VolumeID)
	}

	w.Header().Set("Content-type", "application/octet-stream")

	queryStrings := r.URL.Query()
	compression := queryStrings.Get("compression")

	switch compression {
	case "lz4":
		// Decompress snappy
		sdata, err := ioutil.ReadAll(resp.Body)
		timedLog.Infof("Got snappy-encoded subvolume from Google, %d bytes\n", len(sdata))
		if err != nil {
			return err
		}
		data, err := snappy.Decode(nil, sdata)
		if err != nil {
			return err
		}
		// Recompress and transmit as lz4
		lz4data := make([]byte, lz4.CompressBound(data))
		outSize, err := lz4.Compress(data, lz4data)
		if err != nil {
			return err
		}
		if _, err := w.Write(lz4data[:outSize]); err != nil {
			return err
		}
		timedLog.Infof("Sent lz4-encoded subvolume from DVID, %d bytes\n", outSize)

	default: // "snappy"
		// Just stream data from Google
		respBytes := 0
		const BufferSize = 32 * 1024
		buf := make([]byte, BufferSize)
		for {
			n, err := resp.Body.Read(buf)
			respBytes += n
			eof := (err == io.EOF)
			if err != nil && !eof {
				return err
			}
			if _, err = w.Write(buf[:n]); err != nil {
				return err
			}
			if f, ok := w.(http.Flusher); ok {
				f.Flush()
			}
			if eof {
				break
			}
		}
		timedLog.Infof("Proxied snappy-encoded subvolume from Google, %d bytes\n", respBytes)
	}

	return nil
}

// 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:
//.........这里部分代码省略.........

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()
//.........这里部分代码省略.........

// GetBlocks returns a slice of bytes corresponding to all the blocks along a span in X
func (d *Data) GetBlocks(v dvid.VersionID, start dvid.ChunkPoint3d, span int32) ([]byte, error) {
	timedLog := dvid.NewTimeLog()
	defer timedLog.Infof("GetBlocks %s, span %d", start, span)

	store, err := storage.MutableStore()
	if err != nil {
		return nil, fmt.Errorf("Data type imageblk had error initializing store: %v\n", err)
	}

	indexBeg := dvid.IndexZYX(start)
	sx, sy, sz := indexBeg.Unpack()

	end := start
	end[0] += int32(span - 1)
	indexEnd := dvid.IndexZYX(end)
	keyBeg := NewTKey(&indexBeg)
	keyEnd := NewTKey(&indexEnd)

	// Allocate one uncompressed-sized slice with background values.
	blockBytes := int32(d.BlockSize().Prod()) * d.Values.BytesPerElement()
	numBytes := blockBytes * span

	buf := make([]byte, numBytes, numBytes)
	if d.Background != 0 {
		for i := range buf {
			buf[i] = byte(d.Background)
		}
	}

	// Write the blocks that we can get concurrently on this byte slice.
	ctx := datastore.NewVersionedCtx(d, v)

	var wg sync.WaitGroup
	err = store.ProcessRange(ctx, keyBeg, keyEnd, &storage.ChunkOp{}, func(c *storage.Chunk) error {
		if c == nil || c.TKeyValue == nil {
			return nil
		}
		kv := c.TKeyValue
		if kv.V == nil {
			return nil
		}

		// Determine which block this is.
		indexZYX, err := DecodeTKey(kv.K)
		if err != nil {
			return err
		}
		x, y, z := indexZYX.Unpack()
		if z != sz || y != sy || x < sx || x >= sx+int32(span) {
			return fmt.Errorf("Received key-value for %s, not supposed to be within span range %s, length %d", *indexZYX, start, span)
		}
		n := x - sx
		i := n * blockBytes
		j := i + blockBytes

		// Spawn goroutine to transfer data
		wg.Add(1)
		go xferBlock(buf[i:j], c, &wg)
		return nil
	})
	if err != nil {
		return nil, err
	}
	wg.Wait()
	return buf, nil
}

// 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 "metadata":
		switch action {
		case "post":
			jsonBytes, err := ioutil.ReadAll(r.Body)
			if err != nil {
				server.BadRequest(w, r, err)
				return
			}
			if err := d.SetMetadata(uuid, jsonBytes); err != nil {
				server.BadRequest(w, r, err)
				return
			}

		case "get":
			if d.Levels == nil || len(d.Levels) == 0 {
				server.BadRequest(w, r, "tile metadata for imagetile %q was not set\n", d.DataName())
				return
			}
			metadata := struct {
				MinTileCoord dvid.Point3d
				MaxTileCoord dvid.Point3d
				Levels       TileSpec
			}{
				d.MinTileCoord,
				d.MaxTileCoord,
				d.Levels,
			}
			jsonBytes, err := json.Marshal(metadata)
			if err != nil {
				server.BadRequest(w, r, err)
				return
			}
			w.Header().Set("Content-Type", "application/json")
			fmt.Fprintf(w, string(jsonBytes))
		}
		timedLog.Infof("HTTP %s: metadata (%s)", r.Method, r.URL)

	case "tile":
		switch action {
		case "post":
			err := d.PostTile(ctx, w, r, parts)
			if err != nil {
				server.BadRequest(w, r, "Error in posting tile with URL %q: %v\n", url, err)
				return
			}
		case "get":
			if err := d.ServeTile(ctx, w, r, parts); err != nil {
				server.BadRequest(w, r, err)
				return
			}
		}
		timedLog.Infof("HTTP %s: tile (%s)", r.Method, r.URL)

	case "tilekey":
		switch action {
		case "get":
			var err error
			var hexkey string
			if hexkey, err = d.GetTileKey(ctx, w, r, parts); err != nil {
				server.BadRequest(w, r, err)
				return
			}
//.........这里部分代码省略.........

func (d *Data) foregroundROI(v dvid.VersionID, dest *roi.Data, background dvid.PointNd) {
	dest.Ready = false

	store, err := storage.MutableStore()
	if err != nil {
		dvid.Criticalf("Data type imageblk had error initializing store: %v\n", err)
		return
	}

	timedLog := dvid.NewTimeLog()
	timedLog.Infof("Starting foreground ROI %q for %s", dest.DataName(), d.DataName())

	// Iterate through all voxel blocks, loading and then checking blocks
	// for any foreground voxels.
	ctx := datastore.NewVersionedCtx(d, v)

	backgroundBytes := make([]byte, len(background))
	for i, b := range background {
		backgroundBytes[i] = byte(b)
	}

	const BATCH_SIZE = 1000
	var numBatches int
	var span *dvid.Span
	spans := []dvid.Span{}

	var f storage.ChunkFunc = func(chunk *storage.Chunk) error {
		if chunk == nil || chunk.V == nil {
			return nil
		}
		data, _, err := dvid.DeserializeData(chunk.V, true)
		if err != nil {
			return fmt.Errorf("Error decoding block: %v\n", err)
		}
		numVoxels := d.BlockSize().Prod()
		var foreground bool
		for i := int64(0); i < numVoxels; i++ {
			isBackground := false
			for _, b := range backgroundBytes {
				if data[i] == b {
					isBackground = true
					break
				}
			}
			if !isBackground {
				foreground = true
				break
			}
		}
		if foreground {
			indexZYX, err := DecodeTKey(chunk.K)
			if err != nil {
				return fmt.Errorf("Error decoding voxel block key: %v\n", err)
			}
			x, y, z := indexZYX.Unpack()
			if span == nil {
				span = &dvid.Span{z, y, x, x}
			} else if !span.Extends(x, y, z) {
				spans = append(spans, *span)
				if len(spans) >= BATCH_SIZE {
					init := (numBatches == 0)
					numBatches++
					go func(spans []dvid.Span) {
						if err := dest.PutSpans(v, spans, init); err != nil {
							dvid.Errorf("Error in storing ROI: %v\n", err)
						} else {
							timedLog.Debugf("-- Wrote batch %d of spans for foreground ROI %q", numBatches, dest.DataName())
						}
					}(spans)
					spans = []dvid.Span{}
				}
				span = &dvid.Span{z, y, x, x}
			}
		}
		server.BlockOnInteractiveRequests("voxels [compute foreground ROI]")
		return nil
	}

	minTKey := storage.MinTKey(keyImageBlock)
	maxTKey := storage.MaxTKey(keyImageBlock)

	err = store.ProcessRange(ctx, minTKey, maxTKey, &storage.ChunkOp{}, f)
	if err != nil {
		dvid.Errorf("Error in processing chunks in ROI: %v\n", err)
		return
	}
	if span != nil {
		spans = append(spans, *span)
	}

	// Save new ROI
	if len(spans) > 0 {
		if err := dest.PutSpans(v, spans, numBatches == 0); err != nil {
			dvid.Errorf("Error in storing ROI: %v\n", err)
			return
		}
	}
	timedLog.Infof("Created foreground ROI %q for %s", dest.DataName(), d.DataName())
	dest.Ready = true
}

// 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()

	// Get the action (GET, POST)
	action := strings.ToLower(r.Method)

	// 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]
	}

	// Handle POST on data -> setting of configuration
	if len(parts) == 3 && action == "put" {
		config, err := server.DecodeJSON(r)
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
		if err := d.ModifyConfig(config); err != nil {
			server.BadRequest(w, r, err)
			return
		}
		if err := datastore.SaveDataByUUID(uuid, d); err != nil {
			server.BadRequest(w, r, err)
			return
		}
		fmt.Fprintf(w, "Changed '%s' based on received configuration:\n%s\n", d.DataName(), config)
		return
	}

	if len(parts) < 4 {
		server.BadRequest(w, r, "Incomplete API request")
		return
	}

	// Process help and info.
	switch parts[3] {
	case "help":
		w.Header().Set("Content-Type", "text/plain")
		fmt.Fprintln(w, dtype.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 "sync":
		if action != "post" {
			server.BadRequest(w, r, "Only POST allowed to sync endpoint")
			return
		}
		replace := r.URL.Query().Get("replace") == "true"
		if err := datastore.SetSyncByJSON(d, uuid, replace, r.Body); err != nil {
			server.BadRequest(w, r, err)
			return
		}

	case "count":
		if action != "get" {
			server.BadRequest(w, r, "Only GET action is available on 'count' endpoint.")
			return
		}
		if len(parts) < 6 {
			server.BadRequest(w, r, "Must include label and element type after 'count' endpoint.")
			return
		}
		label, err := strconv.ParseUint(parts[4], 10, 64)
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
		i := StringToIndexType(parts[5])
		if i == UnknownIndex {
			server.BadRequest(w, r, fmt.Errorf("unknown index type specified (%q)", parts[5]))
			return
		}
		count, err := d.GetCountElementType(ctx, label, i)
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
		w.Header().Set("Content-type", "application/json")
		jsonStr := fmt.Sprintf(`{"Label":%d,%q:%d}`, label, i, count)
		if _, err := io.WriteString(w, jsonStr); err != nil {
			server.BadRequest(w, r, err)
			return
		}
		timedLog.Infof("HTTP %s: get count for label %d, index type %s: %s", r.Method, label, i, r.URL)

	case "top":
		if action != "get" {
			server.BadRequest(w, r, "Only GET action is available on 'top' endpoint.")
			return
//.........这里部分代码省略.........

// Optimized bulk loading of XY images by loading all slices for a block before processing.
// Trades off memory for speed.
func (d *Data) loadXYImages(load *bulkLoadInfo) error {
	// Load first slice, get dimensions, allocate blocks for whole slice.
	// Note: We don't need to lock the block slices because goroutines do NOT
	// access the same elements of a slice.
	const numLayers = 2
	var numBlocks int
	var blocks [numLayers]storage.TKeyValues
	var layerTransferred, layerWritten [numLayers]sync.WaitGroup
	var waitForWrites sync.WaitGroup

	curBlocks := 0
	blockSize := d.BlockSize()
	blockBytes := blockSize.Prod() * int64(d.Values.BytesPerElement())

	// Iterate through XY slices batched into the Z length of blocks.
	fileNum := 1
	for _, filename := range load.filenames {
		server.BlockOnInteractiveRequests("imageblk.loadXYImages")

		timedLog := dvid.NewTimeLog()

		zInBlock := load.offset.Value(2) % blockSize.Value(2)
		firstSlice := fileNum == 1
		lastSlice := fileNum == len(load.filenames)
		firstSliceInBlock := firstSlice || zInBlock == 0
		lastSliceInBlock := lastSlice || zInBlock == blockSize.Value(2)-1
		lastBlocks := fileNum+int(blockSize.Value(2)) > len(load.filenames)

		// Load images synchronously
		vox, err := d.loadXYImage(filename, load.offset)
		if err != nil {
			return err
		}

		// Allocate blocks and/or load old block data if first/last XY blocks.
		// Note: Slices are only zeroed out on first and last slice with assumption
		// that ExtData is packed in XY footprint (values cover full extent).
		// If that is NOT the case, we need to zero out blocks for each block layer.
		if fileNum == 1 || (lastBlocks && firstSliceInBlock) {
			numBlocks = dvid.GetNumBlocks(vox, blockSize)
			if fileNum == 1 {
				for layer := 0; layer < numLayers; layer++ {
					blocks[layer] = make(storage.TKeyValues, numBlocks, numBlocks)
					for b := 0; b < numBlocks; b++ {
						blocks[layer][b].V = d.BackgroundBlock()
					}
				}
				var bufSize uint64 = uint64(blockBytes) * uint64(numBlocks) * uint64(numLayers) / 1000000
				dvid.Debugf("Allocated %d MB for buffers.\n", bufSize)
			} else {
				blocks[curBlocks] = make(storage.TKeyValues, numBlocks, numBlocks)
				for b := 0; b < numBlocks; b++ {
					blocks[curBlocks][b].V = d.BackgroundBlock()
				}
			}
			err = d.loadOldBlocks(load.versionID, vox, blocks[curBlocks])
			if err != nil {
				return err
			}
		}

		// Transfer data between external<->internal blocks asynchronously
		layerTransferred[curBlocks].Add(1)
		go func(vox *Voxels, curBlocks int) {
			// Track point extents
			if d.Extents().AdjustPoints(vox.StartPoint(), vox.EndPoint()) {
				load.extentChanged.SetTrue()
			}

			// Process an XY image (slice).
			changed, err := d.writeXYImage(load.versionID, vox, blocks[curBlocks])
			if err != nil {
				dvid.Infof("Error writing XY image: %v\n", err)
			}
			if changed {
				load.extentChanged.SetTrue()
			}
			layerTransferred[curBlocks].Done()
		}(vox, curBlocks)

		// If this is the end of a block (or filenames), wait until all goroutines complete,
		// then asynchronously write blocks.
		if lastSliceInBlock {
			waitForWrites.Add(1)
			layerWritten[curBlocks].Add(1)
			go func(curBlocks int) {
				layerTransferred[curBlocks].Wait()
				dvid.Debugf("Writing block buffer %d using %s and %s...\n",
					curBlocks, d.Compression(), d.Checksum())
				err := d.writeBlocks(load.versionID, blocks[curBlocks], &layerWritten[curBlocks], &waitForWrites)
				if err != nil {
					dvid.Errorf("Error in async write of voxel blocks: %v", err)
				}
			}(curBlocks)
			// We can't move to buffer X until all blocks from buffer X have already been written.
			curBlocks = (curBlocks + 1) % numLayers
			dvid.Debugf("Waiting for layer %d to be written before reusing layer %d blocks\n",
				curBlocks, curBlocks)
//.........这里部分代码省略.........

// LoadLocal adds image data to a version node.  See HelpMessage for example of
// command-line use of "load local".
func (d *Data) LoadLocal(request datastore.Request, reply *datastore.Response) error {
	timedLog := dvid.NewTimeLog()

	// Parse the request
	var uuidStr, dataName, cmdStr, sourceStr, filename string
	_ = request.CommandArgs(1, &uuidStr, &dataName, &cmdStr, &sourceStr, &filename)

	// Get the uuid from a uniquely identifiable string
	uuid, versionID, err := datastore.MatchingUUID(uuidStr)
	if err != nil {
		return fmt.Errorf("Could not find node with UUID %s: %v", uuidStr, err)
	}

	// Load the V3D Raw file.
	ext := filepath.Ext(filename)
	switch ext {
	case ".raw", ".v3draw":
	default:
		return fmt.Errorf("Unknown extension '%s' when expected V3D Raw file", ext)
	}
	file, err := os.Open(filename)
	if err != nil {
		return err
	}
	unmarshaler := V3DRawMarshaler{}
	channels, err := unmarshaler.UnmarshalV3DRaw(file)
	if err != nil {
		return err
	}

	// Store the metadata
	d.NumChannels = len(channels)
	d.Properties.Values = make(dvid.DataValues, d.NumChannels)
	if d.NumChannels > 0 {
		reply.Text = fmt.Sprintf("Loaded %s into data '%s': found %d channels\n",
			d.DataName(), filename, d.NumChannels)
		reply.Text +