// deserializeVertex deserializes a dvid.GraphVertex (compression turned off for now)
func (db *GraphKeyValueDB) deserializeVertex(vertexdata []byte) (dvid.GraphVertex, error) {
	// create vertex to be returned
	vert := dvid.GraphVertex{GraphElement: &dvid.GraphElement{}}

	// if vertexdata is empty return an error
	if vertexdata == nil || len(vertexdata) == 0 {
		return vert, fmt.Errorf("Vertex data empty")
	}

	// boilerplate deserialization from DVID
	data, _, err := dvid.DeserializeData(vertexdata, true)
	if err != nil {
		return vert, err
	}

	// load data from vertex (vertex id, vertex weight, num vertices,
	// vertex array, num properties, property array

	// load vertex id
	start := 0
	vert.Id = dvid.VertexID(binary.LittleEndian.Uint64(data[start:]))
	start += 8

	// load vertex weight
	floatbits := binary.LittleEndian.Uint64(data[start:])
	vert.Weight = math.Float64frombits(floatbits)
	start += 8

	// number of vertices
	count := binary.LittleEndian.Uint64(data[start:])
	start += 8

	vert.Vertices = make([]dvid.VertexID, count, count)
	// load vertices
	for i := uint64(0); i < count; i++ {
		vert.Vertices[i] = dvid.VertexID(binary.LittleEndian.Uint64(data[start:]))
		start += 8
	}

	// number of properties
	vert.Properties = make(dvid.ElementProperties)
	count = binary.LittleEndian.Uint64(data[start:])
	start += 8

	// create property strings
	for i := uint64(0); i < count; i++ {
		propertyname := string("")
		// null separated strings
		for data[start] != byte(0) {
			propertyname += string(data[start])
			start += 1
		}
		vert.Properties[propertyname] = struct{}{}

		// increment beyond null
		start += 1
	}

	return vert, nil
}

// Loads blocks with old data if they exist.
func (d *Data) loadOldBlocks(v dvid.VersionID, vox *Voxels, blocks storage.TKeyValues) error {
	store, err := storage.MutableStore()
	if err != nil {
		return fmt.Errorf("Data type imageblk had error initializing store: %v\n", err)
	}

	ctx := datastore.NewVersionedCtx(d, v)

	// Create a map of old blocks indexed by the index
	oldBlocks := map[dvid.IZYXString]([]byte){}

	// Iterate through index space for this data using ZYX ordering.
	blockSize := d.BlockSize()
	blockNum := 0
	for it, err := vox.IndexIterator(blockSize); err == nil && it.Valid(); it.NextSpan() {
		indexBeg, indexEnd, err := it.IndexSpan()
		if err != nil {
			return err
		}
		begTKey := NewTKey(indexBeg)
		endTKey := NewTKey(indexEnd)

		// Get previous data.
		keyvalues, err := store.GetRange(ctx, begTKey, endTKey)
		if err != nil {
			return err
		}
		for _, kv := range keyvalues {
			indexZYX, err := DecodeTKey(kv.K)
			if err != nil {
				return err
			}
			block, _, err := dvid.DeserializeData(kv.V, true)
			if err != nil {
				return fmt.Errorf("Unable to deserialize block, %s: %v", ctx, err)
			}
			oldBlocks[indexZYX.ToIZYXString()] = block
		}

		// Load previous data into blocks
		ptBeg := indexBeg.Duplicate().(dvid.ChunkIndexer)
		ptEnd := indexEnd.Duplicate().(dvid.ChunkIndexer)
		begX := ptBeg.Value(0)
		endX := ptEnd.Value(0)
		c := dvid.ChunkPoint3d{begX, ptBeg.Value(1), ptBeg.Value(2)}
		for x := begX; x <= endX; x++ {
			c[0] = x
			curIndex := dvid.IndexZYX(c)
			curTKey := NewTKey(&curIndex)
			blocks[blockNum].K = curTKey
			block, ok := oldBlocks[curIndex.ToIZYXString()]
			if ok {
				copy(blocks[blockNum].V, block)
			}
			blockNum++
		}
	}
	return nil
}

// Goroutine that handles splits across a lot of blocks for one label.
func (d *Data) splitBlock(ctx *datastore.VersionedCtx, op splitOp) {
	defer d.MutDone(op.mutID)

	store, err := d.GetOrderedKeyValueDB()
	if err != nil {
		dvid.Errorf("Data type labelblk had error initializing store: %v\n", err)
		return
	}

	// Read the block.
	tk := NewTKeyByCoord(op.block)
	data, err := store.Get(ctx, tk)
	if err != nil {
		dvid.Errorf("Error on GET of labelblk with coord string %v\n", []byte(op.block))
		return
	}
	if data == nil {
		dvid.Errorf("nil label block where split was done, coord %v\n", []byte(op.block))
		return
	}
	blockData, _, err := dvid.DeserializeData(data, true)
	if err != nil {
		dvid.Criticalf("unable to deserialize label block in '%s' key %v: %v\n", d.DataName(), []byte(op.block), err)
		return
	}
	blockBytes := int(d.BlockSize().Prod() * 8)
	if len(blockData) != blockBytes {
		dvid.Criticalf("splitBlock: coord %v got back %d bytes, expected %d bytes\n", []byte(op.block), len(blockData), blockBytes)
		return
	}

	// Modify the block using either voxel-level changes or coarser block-level mods.
	if op.rles != nil {
		if err := d.storeRLEs(blockData, op.newLabel, op.block, op.rles); err != nil {
			dvid.Errorf("can't store label %d RLEs into block %s: %v\n", op.newLabel, op.block, err)
			return
		}
	} else {
		// We are doing coarse split and will replace all
		if err := d.replaceLabel(blockData, op.oldLabel, op.newLabel); err != nil {
			dvid.Errorf("can't replace label %d with %d in block %s: %v\n", op.oldLabel, op.newLabel, op.block, err)
			return
		}
	}

	// Write the modified block.
	serialization, err := dvid.SerializeData(blockData, d.Compression(), d.Checksum())
	if err != nil {
		dvid.Criticalf("Unable to serialize block %s in %q: %v\n", op.block, d.DataName(), err)
		return
	}
	if err := store.Put(ctx, tk, serialization); err != nil {
		dvid.Errorf("Error in putting key %v: %v\n", tk, err)
	}

	// Notify any downstream downres instance.
	d.publishBlockChange(ctx.VersionID(), op.mutID, op.block, blockData)
}

// 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 int) ([]byte, error) {
	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)
	end := start
	end[0] += int32(span - 1)
	indexEnd := dvid.IndexZYX(end)
	begTKey := NewTKey(&indexBeg)
	endTKey := NewTKey(&indexEnd)

	ctx := datastore.NewVersionedCtx(d, v)

	iv := dvid.InstanceVersion{d.DataName(), v}
	mapping := labels.MergeCache.LabelMap(iv)

	keyvalues, err := store.GetRange(ctx, begTKey, endTKey)
	if err != nil {
		return nil, err
	}

	var buf bytes.Buffer

	// Save the # of keyvalues actually obtained.
	numkv := len(keyvalues)
	binary.Write(&buf, binary.LittleEndian, int32(numkv))

	// Write the block indices in XYZ little-endian format + the size of each block
	uncompress := true
	for _, kv := range keyvalues {
		block, _, err := dvid.DeserializeData(kv.V, uncompress)
		if err != nil {
			return nil, fmt.Errorf("Unable to deserialize block, %s (%v): %v", ctx, kv.K, err)
		}
		if mapping != nil {
			n := len(block) / 8
			for i := 0; i < n; i++ {
				orig := binary.LittleEndian.Uint64(block[i*8 : i*8+8])
				mapped, found := mapping.FinalLabel(orig)
				if !found {
					mapped = orig
				}
				binary.LittleEndian.PutUint64(block[i*8:i*8+8], mapped)
			}
		}

		_, err = buf.Write(block)
		if err != nil {
			return nil, err
		}
	}

	return buf.Bytes(), nil
}

// Goroutine that handles relabeling of blocks during a merge operation.
// Since the same block coordinate always gets mapped to the same goroutine, we handle
// concurrency by serializing GET/PUT for a particular block coordinate.
func (d *Data) mergeBlock(in <-chan mergeOp) {
	store, err := storage.MutableStore()
	if err != nil {
		dvid.Errorf("Data type labelblk had error initializing store: %v\n", err)
		return
	}
	blockBytes := int(d.BlockSize().Prod() * 8)

	for op := range in {
		tk := NewTKeyByCoord(op.izyx)
		data, err := store.Get(op.ctx, tk)
		if err != nil {
			dvid.Errorf("Error on GET of labelblk with coord string %q\n", op.izyx)
			op.wg.Done()
			continue
		}
		if data == nil {
			dvid.Errorf("nil label block where merge was done!\n")
			op.wg.Done()
			continue
		}

		blockData, _, err := dvid.DeserializeData(data, true)
		if err != nil {
			dvid.Criticalf("unable to deserialize label block in '%s': %v\n", d.DataName(), err)
			op.wg.Done()
			continue
		}
		if len(blockData) != blockBytes {
			dvid.Criticalf("After labelblk deserialization got back %d bytes, expected %d bytes\n", len(blockData), blockBytes)
			op.wg.Done()
			continue
		}

		// Iterate through this block of labels and relabel if label in merge.
		for i := 0; i < blockBytes; i += 8 {
			label := binary.LittleEndian.Uint64(blockData[i : i+8])
			if _, merged := op.Merged[label]; merged {
				binary.LittleEndian.PutUint64(blockData[i:i+8], op.Target)
			}
		}

		// Store this block.
		serialization, err := dvid.SerializeData(blockData, d.Compression(), d.Checksum())
		if err != nil {
			dvid.Criticalf("Unable to serialize block in %q: %v\n", d.DataName(), err)
			op.wg.Done()
			continue
		}
		if err := store.Put(op.ctx, tk, serialization); err != nil {
			dvid.Errorf("Error in putting key %v: %v\n", tk, err)
		}
		op.wg.Done()
	}
}

// deserializeEdge deserializes a dvid.GraphEdge (compression turned off for now)
func (db *GraphKeyValueDB) deserializeEdge(edgedata []byte) (dvid.GraphEdge, error) {
	// create edge to be returned
	edge := dvid.GraphEdge{GraphElement: &dvid.GraphElement{}}

	// if edgedata is empty return an error
	if edgedata == nil || len(edgedata) == 0 {
		return edge, fmt.Errorf("Edge data empty")
	}

	// boilerplate deserialization from DVID
	data, _, err := dvid.DeserializeData(edgedata, true)
	if err != nil {
		return edge, err
	}

	// load data from edge (vertex1 id, vertex2 id, edge weight,
	// num properties, property array

	// load vertex1 id
	start := 0
	edge.Vertexpair.Vertex1 = dvid.VertexID(binary.LittleEndian.Uint64(data[start:]))
	start += 8

	// load vertex2 id
	edge.Vertexpair.Vertex2 = dvid.VertexID(binary.LittleEndian.Uint64(data[start:]))
	start += 8

	// load edge weight
	floatbits := binary.LittleEndian.Uint64(data[start:])
	edge.Weight = math.Float64frombits(floatbits)
	start += 8

	// number of properties
	count := binary.LittleEndian.Uint64(data[start:])
	start += 8

	// create property strings
	edge.Properties = make(dvid.ElementProperties)
	for i := uint64(0); i < count; i++ {
		propertyname := string("")
		// null separated strings
		for data[start] != 0 {
			propertyname += string(data[start])
			start += 1
		}
		edge.Properties[propertyname] = struct{}{}

		// increment beyond null
		start += 1
	}

	return edge, nil
}

// handles relabeling of blocks during a merge operation.
func (d *Data) mergeBlock(ctx *datastore.VersionedCtx, op mergeOp) {
	defer d.MutDone(op.mutID)

	store, err := d.GetKeyValueDB()
	if err != nil {
		dvid.Errorf("Data type labelblk had error initializing store: %v\n", err)
		return
	}

	tk := NewTKeyByCoord(op.block)
	data, err := store.Get(ctx, tk)
	if err != nil {
		dvid.Errorf("Error on GET of labelblk with coord string %q\n", op.block)
		return
	}
	if data == nil {
		dvid.Errorf("nil label block where merge was done!\n")
		return
	}

	blockData, _, err := dvid.DeserializeData(data, true)
	if err != nil {
		dvid.Criticalf("unable to deserialize label block in '%s': %v\n", d.DataName(), err)
		return
	}
	blockBytes := int(d.BlockSize().Prod() * 8)
	if len(blockData) != blockBytes {
		dvid.Criticalf("After labelblk deserialization got back %d bytes, expected %d bytes\n", len(blockData), blockBytes)
		return
	}

	// Iterate through this block of labels and relabel if label in merge.
	for i := 0; i < blockBytes; i += 8 {
		label := binary.LittleEndian.Uint64(blockData[i : i+8])
		if _, merged := op.Merged[label]; merged {
			binary.LittleEndian.PutUint64(blockData[i:i+8], op.Target)
		}
	}

	// Store this block.
	serialization, err := dvid.SerializeData(blockData, d.Compression(), d.Checksum())
	if err != nil {
		dvid.Criticalf("Unable to serialize block in %q: %v\n", d.DataName(), err)
		return
	}
	if err := store.Put(ctx, tk, serialization); err != nil {
		dvid.Errorf("Error in putting key %v: %v\n", tk, err)
	}

	// Notify any downstream downres instance.
	d.publishBlockChange(ctx.VersionID(), op.mutID, op.block, blockData)
}

func xferBlock(buf []byte, chunk *storage.Chunk, wg *sync.WaitGroup) {
	defer wg.Done()

	kv := chunk.TKeyValue
	uncompress := true
	block, _, err := dvid.DeserializeData(kv.V, uncompress)
	if err != nil {
		dvid.Errorf("Unable to deserialize block (%v): %v", kv.K, err)
		return
	}
	if len(block) != len(buf) {
		dvid.Errorf("Deserialized block length (%d) != allocated block length (%d)", len(block), len(buf))
		return
	}
	copy(buf, block)
}

// load block of data from storage
func (d *Data) loadOldBlock(v dvid.VersionID, k storage.TKey) ([]byte, error) {
	store, err := d.GetOrderedKeyValueDB()
	if err != nil {
		return nil, fmt.Errorf("Data type imageblk had error initializing store: %v\n", err)
	}

	ctx := datastore.NewVersionedCtx(d, v)
	serialization, err := store.Get(ctx, k)
	if err != nil {
		return nil, err
	}
	data, _, err := dvid.DeserializeData(serialization, true)
	if err != nil {
		return nil, fmt.Errorf("Unable to deserialize block, %s: %v", ctx, err)
	}
	return data, err
}

// GetSurface returns a gzipped byte array with # voxels and float32 arrays for vertices and
// normals.
func GetSurface(ctx storage.Context, label uint64) ([]byte, bool, error) {
	bigdata, err := storage.BigDataStore()
	if err != nil {
		return nil, false, fmt.Errorf("Cannot get datastore that handles big data: %v\n", err)
	}

	// Retrieve the precomputed surface or that it's not available.
	data, err := bigdata.Get(ctx, NewLabelSurfaceIndex(label))
	if err != nil {
		return nil, false, fmt.Errorf("Error in retrieving surface for label %d: %v", label, err)
	}
	if data == nil {
		return []byte{}, false, nil
	}
	uncompress := false
	surfaceBytes, _, err := dvid.DeserializeData(data, uncompress)
	if err != nil {
		return nil, false, fmt.Errorf("Unable to deserialize surface for label %d: %v\n", label, err)
	}
	return surfaceBytes, true, nil
}

// GetData gets a value using a key
func (d *Data) GetData(ctx storage.Context, keyStr string) ([]byte, bool, error) {
	db, err := storage.MutableStore()
	if err != nil {
		return nil, false, err
	}
	tk, err := NewTKey(keyStr)
	if err != nil {
		return nil, false, err
	}
	data, err := db.Get(ctx, tk)
	if err != nil {
		return nil, false, fmt.Errorf("Error in retrieving key '%s': %v", keyStr, err)
	}
	if data == nil {
		return nil, false, nil
	}
	uncompress := true
	value, _, err := dvid.DeserializeData(data, uncompress)
	if err != nil {
		return nil, false, fmt.Errorf("Unable to deserialize data for key '%s': %v\n", keyStr, err)
	}
	return value, true, nil
}

// GetLabelBytesAtPoint returns the 8 byte slice corresponding to a 64-bit label at a point.
func (d *Data) GetLabelBytesAtPoint(v dvid.VersionID, pt dvid.Point) ([]byte, error) {
	store, err := storage.MutableStore()
	if err != nil {
		return nil, err
	}

	// Compute the block key that contains the given point.
	coord, ok := pt.(dvid.Chunkable)
	if !ok {
		return nil, fmt.Errorf("Can't determine block of point %s", pt)
	}
	blockSize := d.BlockSize()
	blockCoord := coord.Chunk(blockSize).(dvid.ChunkPoint3d)
	index := dvid.IndexZYX(blockCoord)

	// Retrieve the block of labels
	ctx := datastore.NewVersionedCtx(d, v)
	serialization, err := store.Get(ctx, NewTKey(&index))
	if err != nil {
		return nil, fmt.Errorf("Error getting '%s' block for index %s\n", d.DataName(), blockCoord)
	}
	if serialization == nil {
		return zeroLabelBytes, nil
	}
	labelData, _, err := dvid.DeserializeData(serialization, true)
	if err != nil {
		return nil, fmt.Errorf("Unable to deserialize block %s in '%s': %v\n", blockCoord, d.DataName(), err)
	}

	// Retrieve the particular label within the block.
	ptInBlock := coord.PointInChunk(blockSize)
	nx := int64(blockSize.Value(0))
	nxy := nx * int64(blockSize.Value(1))
	i := (int64(ptInBlock.Value(0)) + int64(ptInBlock.Value(1))*nx + int64(ptInBlock.Value(2))*nxy) * 8
	return labelData[i : i+8], nil
}

//.........这里部分代码省略.........

			if data_begin == data_end {
				continue
			}

			// check if post is possible
			if _, ok := transactions.locked_ids[id]; !ok {
				continue
			}
			if edgemode {
				if _, ok := transactions.locked_ids[id2]; !ok {
					continue
				}
			}

			// execute post
			serialization, err := dvid.SerializeData(data[data_begin:data_end], d.Compression(), d.Checksum())
			if err != nil {
				return fmt.Errorf("Unable to serialize data: %v\n", err)
			}
			if edgemode {
				err = db.SetEdgeProperty(ctx, id, id2, propertyname, serialization)
			} else {
				err = db.SetVertexProperty(ctx, id, propertyname, serialization)
			}
			if err != nil {
				return fmt.Errorf("Failed to add property %s: %v\n", propertyname, err)
			}
		}
	} else {
		num_properties := binary.LittleEndian.Uint64(data[start:])
		start += 8
		num_properties_loc := len(returned_data)
		longbuf := make([]byte, 8, 8)
		binary.LittleEndian.PutUint64(longbuf, 0)
		returned_data = append(returned_data, longbuf...)
		num_executed_transactions := uint64(0)

		// read the vertex or edge properties desired
		for i := uint64(0); i < num_properties; i++ {
			temp := binary.LittleEndian.Uint64(data[start:])
			id := dvid.VertexID(temp)
			var id2 dvid.VertexID
			start += 8
			if edgemode {
				temp := binary.LittleEndian.Uint64(data[start:])
				id2 = dvid.VertexID(temp)
				start += 8
			}

			// check if post is possible
			if _, ok := transactions.locked_ids[id]; !ok {
				continue
			}
			if edgemode {
				if _, ok := transactions.locked_ids[id2]; !ok {
					continue
				}
			}

			// execute get command
			var dataout []byte
			if edgemode {
				dataout, err = db.GetEdgeProperty(ctx, id, id2, propertyname)
			} else {
				dataout, err = db.GetVertexProperty(ctx, id, propertyname)
			}

			// serialize return data only if there is return data and no error;
			// otherwise return just return the id and size of 0
			var data_serialized []byte
			if (err == nil) && len(dataout) > 0 {
				uncompress := true
				data_serialized, _, err = dvid.DeserializeData(dataout, uncompress)
				if err != nil {
					return fmt.Errorf("Unable to deserialize data for property '%s': %v\n", propertyname, err)
				}
			}

			// save transaction
			num_executed_transactions += 1
			binary.LittleEndian.PutUint64(longbuf, uint64(id))
			returned_data = append(returned_data, longbuf...)
			if edgemode {
				binary.LittleEndian.PutUint64(longbuf, uint64(id2))
				returned_data = append(returned_data, longbuf...)
			}
			binary.LittleEndian.PutUint64(longbuf, uint64(len(data_serialized)))
			returned_data = append(returned_data, longbuf...)
			returned_data = append(returned_data, data_serialized...)
		}

		// update the number of transactions
		binary.LittleEndian.PutUint64(returned_data[num_properties_loc:], num_executed_transactions)
	}

	w.Header().Set("Content-Type", "application/octet-stream")
	_, err = w.Write(returned_data)
	return err
}

// Goroutine that handles splits across a lot of blocks for one label.
func (d *Data) splitBlock(in <-chan splitOp) {
	store, err := storage.MutableStore()
	if err != nil {
		dvid.Errorf("Data type labelblk had error initializing store: %v\n", err)
		return
	}
	batcher, ok := store.(storage.KeyValueBatcher)
	if !ok {
		err = fmt.Errorf("Data type labelblk requires batch-enabled store, which %q is not\n", store)
		return
	}
	blockBytes := int(d.BlockSize().Prod() * 8)

	for op := range in {
		// Iterate through all the modified blocks, inserting the new label using the RLEs for that block.
		timedLog := dvid.NewTimeLog()
		splitCache.Incr(op.ctx.InstanceVersion(), op.OldLabel)
		batch := batcher.NewBatch(&op.ctx)
		for _, zyxStr := range op.SortedBlocks {
			// Read the block.
			tk := NewTKeyByCoord(zyxStr)
			data, err := store.Get(&op.ctx, tk)
			if err != nil {
				dvid.Errorf("Error on GET of labelblk with coord string %v\n", []byte(zyxStr))
				continue
			}
			if data == nil {
				dvid.Errorf("nil label block where split was done, coord %v\n", []byte(zyxStr))
				continue
			}
			bdata, _, err := dvid.DeserializeData(data, true)
			if err != nil {
				dvid.Criticalf("unable to deserialize label block in '%s' key %v: %v\n", d.DataName(), []byte(zyxStr), err)
				continue
			}
			if len(bdata) != blockBytes {
				dvid.Criticalf("splitBlock: coord %v got back %d bytes, expected %d bytes\n", []byte(zyxStr), len(bdata), blockBytes)
				continue
			}

			// Modify the block using either voxel-level changes or coarser block-level mods.
			if op.Split != nil {
				rles, found := op.Split[zyxStr]
				if !found {
					dvid.Errorf("split block %s not present in block RLEs\n", zyxStr.Print())
					continue
				}
				if err := d.storeRLEs(bdata, op.NewLabel, zyxStr, rles); err != nil {
					dvid.Errorf("can't store label %d RLEs into block %s: %v\n", op.NewLabel, zyxStr.Print(), err)
					continue
				}
			} else {
				// We are doing coarse split and will replace all
				if err := d.replaceLabel(bdata, op.OldLabel, op.NewLabel); err != nil {
					dvid.Errorf("can't replace label %d with %d in block %s: %v\n", op.OldLabel, op.NewLabel, zyxStr.Print(), err)
					continue
				}
			}

			// Write the modified block.
			serialization, err := dvid.SerializeData(bdata, d.Compression(), d.Checksum())
			if err != nil {
				dvid.Criticalf("Unable to serialize block %s in %q: %v\n", zyxStr.Print(), d.DataName(), err)
				continue
			}
			batch.Put(tk, serialization)
		}
		if err := batch.Commit(); err != nil {
			dvid.Errorf("Batch PUT during %q block split of %d: %v\n", d.DataName(), op.OldLabel, err)
		}
		splitCache.Decr(op.ctx.InstanceVersion(), op.OldLabel)
		timedLog.Debugf("labelblk sync complete for split of %d -> %d", op.OldLabel, op.NewLabel)
	}
}

func (d *Data) createCompositeChunk(chunk *storage.Chunk) {
	defer func() {
		// After processing a chunk, return the token.
		server.HandlerToken <- 1

		// Notify the requestor that this chunk is done.
		if chunk.Wg != nil {
			chunk.Wg.Done()
		}
	}()

	op := chunk.Op.(*compositeOp)

	// Get the spatial index associated with this chunk.
	zyx, err := imageblk.DecodeTKey(chunk.K)
	if err != nil {
		dvid.Errorf("Error in %s.ChunkApplyMap(): %v", d.Data.DataName(), err)
		return
	}

	// Initialize the label buffers.  For voxels, this data needs to be uncompressed and deserialized.
	curZMutex.Lock()
	if zyx[2] > curZ {
		curZ = zyx[2]
		minZ := zyx.MinPoint(d.BlockSize()).Value(2)
		maxZ := zyx.MaxPoint(d.BlockSize()).Value(2)
		dvid.Debugf("Now creating composite blocks for Z %d to %d\n", minZ, maxZ)
	}
	curZMutex.Unlock()

	labelData, _, err := dvid.DeserializeData(chunk.V, true)
	if err != nil {
		dvid.Infof("Unable to deserialize block in '%s': %v\n", d.DataName(), err)
		return
	}
	blockBytes := len(labelData)
	if blockBytes%8 != 0 {
		dvid.Infof("Retrieved, deserialized block is wrong size: %d bytes\n", blockBytes)
		return
	}

	// Get the corresponding grayscale block.
	store, err := op.grayscale.GetOrderedKeyValueDB()
	if err != nil {
		dvid.Errorf("Unable to retrieve store for %q: %v\n", op.grayscale.DataName(), err)
		return
	}
	grayscaleCtx := datastore.NewVersionedCtx(op.grayscale, op.versionID)
	blockData, err := store.Get(grayscaleCtx, chunk.K)
	if err != nil {
		dvid.Errorf("Error getting grayscale block for index %s\n", zyx)
		return
	}
	grayscaleData, _, err := dvid.DeserializeData(blockData, true)
	if err != nil {
		dvid.Errorf("Unable to deserialize block in '%s': %v\n", op.grayscale.DataName(), err)
		return
	}

	// Compute the composite block.
	// TODO -- Exploit run lengths, use cache of hash?
	compositeBytes := blockBytes / 2
	compositeData := make([]byte, compositeBytes, compositeBytes)
	compositeI := 0
	labelI := 0
	hashBuf := make([]byte, 4, 4)
	for _, grayscale := range grayscaleData {
		//murmurhash3(labelData[labelI:labelI+8], hashBuf)
		//hashBuf[3] = grayscale
		writePseudoColor(grayscale, labelData[labelI:labelI+8], hashBuf)
		copy(compositeData[compositeI:compositeI+4], hashBuf)
		compositeI += 4
		labelI += 8
	}

	// Store the composite block into the rgba8 data.
	serialization, err := dvid.SerializeData(compositeData, d.Compression(), d.Checksum())
	if err != nil {
		dvid.Errorf("Unable to serialize composite block %s: %v\n", zyx, err)
		return
	}
	compositeCtx := datastore.NewVersionedCtx(op.composite, op.versionID)
	err = store.Put(compositeCtx, chunk.K, serialization)
	if err != nil {
		dvid.Errorf("Unable to PUT composite block %s: %v\n", zyx, err)
		return
	}
}

// Calculates value of a 3d real world point in space defined by underlying data resolution.
func (d *Data) computeValue(pt dvid.Vector3d, ctx storage.Context, keyF KeyFunc, cache *ValueCache) ([]byte, error) {
	db, err := storage.BigDataStore()
	if err != nil {
		return nil, err
	}

	valuesPerElement := d.Properties.Values.ValuesPerElement()
	bytesPerValue, err := d.Properties.Values.BytesPerValue()
	if err != nil {
		return nil, err
	}
	bytesPerVoxel := valuesPerElement * bytesPerValue

	// Allocate an empty block.
	blockSize, ok := d.BlockSize().(dvid.Point3d)
	if !ok {
		return nil, fmt.Errorf("Data %q does not have a 3d block size", d.DataName())
	}
	nx := blockSize[0]
	nxy := nx * blockSize[1]
	emptyBlock := d.BackgroundBlock()

	populateF := func(key []byte) ([]byte, error) {
		serializedData, err := db.Get(ctx, key)
		if err != nil {
			return nil, err
		}
		var deserializedData []byte
		if serializedData == nil || len(serializedData) == 0 {
			deserializedData = emptyBlock
		} else {
			deserializedData, _, err = dvid.DeserializeData(serializedData, true)
			if err != nil {
				return nil, fmt.Errorf("Unable to deserialize block: %v", err)
			}
		}
		return deserializedData, nil
	}

	// For the given point, compute surrounding lattice points and retrieve values.
	neighbors := d.neighborhood(pt)
	var valuesI int32
	for _, voxelCoord := range neighbors.coords {
		deserializedData, _, err := cache.Get(keyF(voxelCoord), populateF)
		if err != nil {
			return nil, err
		}
		blockPt := voxelCoord.PointInChunk(blockSize).(dvid.Point3d)
		blockI := blockPt[2]*nxy + blockPt[1]*nx + blockPt[0]
		//fmt.Printf("Block %s (%d) len %d -> Neighbor %s (buffer %d, len %d)\n",
		//	blockPt, blockI, len(blockData), voxelCoord, valuesI, len(neighbors.values))
		copy(neighbors.values[valuesI:valuesI+bytesPerVoxel], deserializedData[blockI:blockI+bytesPerVoxel])
		valuesI += bytesPerVoxel
	}

	// Perform trilinear interpolation on the underlying data values.
	unsupported := func() error {
		return fmt.Errorf("DVID cannot retrieve images with arbitrary orientation using %d channels and %d bytes/channel",
			valuesPerElement, bytesPerValue)
	}
	var value []byte
	switch valuesPerElement {
	case 1:
		switch bytesPerValue {
		case 1:
			if d.Interpolable {
				interpValue := trilinearInterpUint8(neighbors.xd, neighbors.yd, neighbors.zd, []uint8(neighbors.values))
				value = []byte{byte(interpValue)}
			} else {
				value = []byte{nearestNeighborUint8(neighbors.xd, neighbors.yd, neighbors.zd, []uint8(neighbors.values))}
			}
		case 2:
			fallthrough
		case 4:
			fallthrough
		case 8:
			fallthrough
		default:
			return nil, unsupported()
		}
	case 4:
		switch bytesPerValue {
		case 1:
			value = make([]byte, 4, 4)
			for c := 0; c < 4; c++ {
				channelValues := make([]uint8, 8, 8)
				for i := 0; i < 8; i++ {
					channelValues[i] = uint8(neighbors.values[i*4+c])
				}
				if d.Interpolable {
					interpValue := trilinearInterpUint8(neighbors.xd, neighbors.yd, neighbors.zd, channelValues)
					value[c] = byte(interpValue)
				} else {
					value[c] = byte(nearestNeighborUint8(neighbors.xd, neighbors.yd, neighbors.zd, channelValues))
				}
			}
		case 2:
			fallthrough
		default:
//.........这里部分代码省略.........

func (d *Data) readChunk(chunk *storage.Chunk) {
	defer func() {
		// After processing a chunk, return the token.
		server.HandlerToken <- 1

		// Notify the requestor that this chunk is done.
		if chunk.Wg != nil {
			chunk.Wg.Done()
		}
	}()

	op, ok := chunk.Op.(*getOperation)
	if !ok {
		log.Fatalf("Illegal operation passed to readChunk() for data %s\n", d.DataName())
	}

	// Make sure our received chunk is valid.
	if chunk == nil {
		dvid.Errorf("Received nil chunk in readChunk.  Ignoring chunk.\n")
		return
	}
	if chunk.K == nil {
		dvid.Errorf("Received nil chunk key in readChunk.  Ignoring chunk.\n")
		return
	}

	// If there's an ROI, if outside ROI, use blank buffer or allow scaling via attenuation.
	var zeroOut bool
	var attenuation uint8
	indexZYX, err := DecodeTKey(chunk.K)
	if err != nil {
		dvid.Errorf("Error processing voxel block: %v\n", err)
		return
	}
	if op.blocksInROI != nil {
		indexString := string(indexZYX.Bytes())
		_, insideROI := op.blocksInROI[indexString]
		if !insideROI {
			if op.attenuation == 0 {
				zeroOut = true
			}
			attenuation = op.attenuation
		}
	}

	// Initialize the block buffer using the chunk of data.  For voxels, this chunk of
	// data needs to be uncompressed and deserialized.
	var blockData []byte
	if zeroOut || chunk.V == nil {
		blockData = d.BackgroundBlock()
	} else {
		blockData, _, err = dvid.DeserializeData(chunk.V, true)
		if err != nil {
			dvid.Errorf("Unable to deserialize block in '%s': %v\n", d.DataName(), err)
			return
		}
	}

	// Perform the operation.
	block := &storage.TKeyValue{chunk.K, blockData}
	if err = op.voxels.ReadBlock(block, d.BlockSize(), attenuation); err != nil {
		dvid.Errorf("Unable to ReadFromBlock() in %q: %v\n", d.DataName(), 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
}

func (d *Data) DeleteBlocks(ctx *datastore.VersionedCtx, start dvid.ChunkPoint3d, span int) error {
	store, err := storage.MutableStore()
	if err != nil {
		return fmt.Errorf("Data type labelblk had error initializing store: %v\n", err)
	}
	batcher, ok := store.(storage.KeyValueBatcher)
	if !ok {
		return fmt.Errorf("Data type labelblk requires batch-enabled store, which %q is not\n", store)
	}

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

	iv := dvid.InstanceVersion{d.DataName(), ctx.VersionID()}
	mapping := labels.MergeCache.LabelMap(iv)

	kvs, err := store.GetRange(ctx, begTKey, endTKey)
	if err != nil {
		return err
	}

	batch := batcher.NewBatch(ctx)
	uncompress := true
	for _, kv := range kvs {
		izyx, err := DecodeTKey(kv.K)
		if err != nil {
			return err
		}

		// Delete the labelblk (really tombstones it)
		batch.Delete(kv.K)

		// Send data to delete associated labelvol for labels in this block
		block, _, err := dvid.DeserializeData(kv.V, uncompress)
		if err != nil {
			return fmt.Errorf("Unable to deserialize block, %s (%v): %v", ctx, kv.K, err)
		}
		if mapping != nil {
			n := len(block) / 8
			for i := 0; i < n; i++ {
				orig := binary.LittleEndian.Uint64(block[i*8 : i*8+8])
				mapped, found := mapping.FinalLabel(orig)
				if !found {
					mapped = orig
				}
				binary.LittleEndian.PutUint64(block[i*8:i*8+8], mapped)
			}
		}

		// Notify any subscribers that we've deleted this block.
		evt := datastore.SyncEvent{d.DataName(), labels.DeleteBlockEvent}
		msg := datastore.SyncMessage{ctx.VersionID(), labels.DeleteBlock{izyx, block}}
		if err := datastore.NotifySubscribers(evt, msg); err != nil {
			return err
		}

	}
	return batch.Commit()
}

// Deserialize returns a type key-value pair where the value has been deserialized.
func (kv TKeyValue) Deserialize(uncompress bool) (TKeyValue, error) {
	value, _, err := dvid.DeserializeData(kv.V, uncompress)
	return TKeyValue{kv.K, value}, err
}