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)
}

// 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 repoDeleteHandler(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 in retrieving data instance name from URL parameters")
		return
	}

	// Do the deletion asynchronously since they can take a very long time.
	go func() {
		if err := datastore.DeleteDataByUUID(uuid, dvid.InstanceName(dataname)); err != nil {
			dvid.Errorf("Error in deleting data instance %q: %v", dataname, err)
		}
	}()

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

// DataBySpec returns a ROI Data based on a string specification of the form
// "<roiname>,<uuid>". If the given string is not parsable, the "found" return value is false.
func DataBySpec(spec string) (d *Data, v dvid.VersionID, found bool, err error) {
	roispec := strings.Split(spec, ",")
	if len(roispec) != 2 {
		err = fmt.Errorf("Expect ROI filters to have format %q, but got %q", "roi:<roiname>,<uuid>", spec)
		return
	}
	roiName := dvid.InstanceName(roispec[0])
	_, v, err = datastore.MatchingUUID(roispec[1])
	if err != nil {
		return
	}
	var data datastore.DataService
	data, err = datastore.GetDataByVersionName(v, roiName)
	if err != nil {
		return
	}
	var ok bool
	d, ok = data.(*Data)
	if !ok {
		err = fmt.Errorf("Data instance %q is not ROI instance", roiName)
		return
	}
	found = true
	return
}

func repoNewDataHandler(c web.C, w http.ResponseWriter, r *http.Request) {
	uuid := c.Env["uuid"].(dvid.UUID)
	config := dvid.NewConfig()
	if err := config.SetByJSON(r.Body); err != nil {
		BadRequest(w, r, fmt.Sprintf("Error decoding POSTed JSON config for 'new': %v", err))
		return
	}

	// Make sure that the passed configuration has data type and instance name.
	typename, found, err := config.GetString("typename")
	if !found || err != nil {
		BadRequest(w, r, "POST on repo endpoint requires specification of valid 'typename'")
		return
	}
	dataname, found, err := config.GetString("dataname")
	if !found || err != nil {
		BadRequest(w, r, "POST on repo endpoint requires specification of valid 'dataname'")
		return
	}
	typeservice, err := datastore.TypeServiceByName(dvid.TypeString(typename))
	if err != nil {
		BadRequest(w, r, err)
		return
	}
	_, err = datastore.NewData(uuid, typeservice, dvid.InstanceName(dataname), config)
	if err != nil {
		BadRequest(w, r, err)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	fmt.Fprintf(w, "{%q: 'Added %s [%s] to node %s'}", "result", dataname, typename, uuid)
}

// ForegroundROI creates a new ROI by determining all non-background blocks.
func (d *Data) ForegroundROI(req datastore.Request, reply *datastore.Response) error {
	if d.Values.BytesPerElement() != 1 {
		return fmt.Errorf("Foreground ROI command only implemented for 1 byte/voxel data!")
	}

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

	// Get the version and repo
	uuid, versionID, err := datastore.MatchingUUID(uuidStr)
	if err != nil {
		return err
	}
	if err = datastore.AddToNodeLog(uuid, []string{req.Command.String()}); err != nil {
		return err
	}

	// Use existing destination data or a new ROI data.
	var dest *roi.Data
	dest, err = roi.GetByUUID(uuid, dvid.InstanceName(destName))
	if err != nil {
		config := dvid.NewConfig()
		typeservice, err := datastore.TypeServiceByName("roi")
		if err != nil {
			return err
		}
		dataservice, err := datastore.NewData(uuid, typeservice, dvid.InstanceName(destName), config)
		if err != nil {
			return err
		}
		var ok bool
		dest, ok = dataservice.(*roi.Data)
		if !ok {
			return fmt.Errorf("Could not create ROI data instance")
		}
	}

	// Asynchronously process the voxels.
	background, err := dvid.StringToPointNd(backgroundStr, ",")
	if err != nil {
		return err
	}
	go d.foregroundROI(versionID, dest, background)

	return nil
}

func GetTestDataContext(uuid dvid.UUID, name string, instanceID dvid.InstanceID) *DataContext {
	versionID, found := testUUIDToVersion[uuid]
	if !found {
		return nil
	}
	data := &testData{uuid, dvid.InstanceName(name), instanceID}
	return NewDataContext(data, versionID)
}

// ServeTile returns a tile with appropriate Content-Type set.
func (d *Data) ServeTile(uuid dvid.UUID, ctx storage.Context, w http.ResponseWriter,
	r *http.Request, parts []string) error {

	if d.Levels == nil || len(d.Levels) == 0 {
		return ErrNoMetadataSet
	}
	tileReq, err := d.parseTileReq(r, parts)

	queryValues := r.URL.Query()
	noblanksStr := dvid.InstanceName(queryValues.Get("noblanks"))
	var noblanks bool
	if noblanksStr == "true" {
		noblanks = true
	}
	var formatStr string
	if len(parts) >= 8 {
		formatStr = parts[7]
	}

	data, err := d.getTileData(ctx, tileReq)
	if err != nil {
		server.BadRequest(w, r, err)
		return err
	}
	if data == nil {
		if noblanks {
			http.NotFound(w, r)
			return nil
		}
		img, err := d.getBlankTileImage(uuid, tileReq)
		if err != nil {
			return err
		}
		return dvid.WriteImageHttp(w, img, formatStr)
	}

	switch d.Encoding {
	case LZ4:
		var img dvid.Image
		if err := img.Deserialize(data); err != nil {
			return err
		}
		data, err = img.GetPNG()
		w.Header().Set("Content-type", "image/png")
	case PNG:
		w.Header().Set("Content-type", "image/png")
	case JPG:
		w.Header().Set("Content-type", "image/jpeg")
	}
	if err != nil {
		server.BadRequest(w, r, err)
		return err
	}
	if _, err = w.Write(data); err != nil {
		return err
	}
	return nil
}

func makeGrayscale(uuid dvid.UUID, t *testing.T, name string) *Data {
	config := dvid.NewConfig()
	dataservice, err := datastore.NewData(uuid, grayscaleT, dvid.InstanceName(name), config)
	if err != nil {
		t.Errorf("Unable to create grayscale instance %q: %v\n", name, err)
	}
	grayscale, ok := dataservice.(*Data)
	if !ok {
		t.Errorf("Can't cast uint8blk data service into Data\n")
	}
	return grayscale
}

func setupGroupcache(config GroupcacheConfig) error {
	if config.GB == 0 {
		return nil
	}
	var cacheBytes int64
	cacheBytes = int64(config.GB) << 30

	pool := groupcache.NewHTTPPool(config.Host)
	if pool != nil {
		dvid.Infof("Initializing groupcache with %d GB at %s...\n", config.GB, config.Host)
		manager.gcache.cache = groupcache.NewGroup("immutable", cacheBytes, groupcache.GetterFunc(
			func(c groupcache.Context, key string, dest groupcache.Sink) error {
				// Use KeyValueDB defined as context.
				gctx, ok := c.(GroupcacheCtx)
				if !ok {
					return fmt.Errorf("bad groupcache context: expected GroupcacheCtx, got %v", c)
				}

				// First four bytes of key is instance ID to isolate groupcache collisions.
				tk := TKey(key[4:])
				data, err := gctx.KeyValueDB.Get(gctx.Context, tk)
				if err != nil {
					return err
				}
				return dest.SetBytes(data)
			}))
		manager.gcache.supported = make(map[dvid.DataSpecifier]struct{})
		for _, dataspec := range config.Instances {
			name := strings.Trim(dataspec, "\"")
			parts := strings.Split(name, ":")
			switch len(parts) {
			case 2:
				dataid := dvid.GetDataSpecifier(dvid.InstanceName(parts[0]), dvid.UUID(parts[1]))
				manager.gcache.supported[dataid] = struct{}{}
			default:
				dvid.Errorf("bad data instance specification %q given for groupcache support in config file\n", dataspec)
			}
		}

		// If we have additional peers, add them and start a listener via the HTTP port.
		if len(config.Peers) > 0 {
			peers := []string{config.Host}
			peers = append(peers, config.Peers...)
			pool.Set(peers...)
			dvid.Infof("Groupcache configuration has %d peers in addition to local host.\n", len(config.Peers))
			dvid.Infof("Starting groupcache HTTP server on %s\n", config.Host)
			http.ListenAndServe(config.Host, http.HandlerFunc(pool.ServeHTTP))
		}
	}
	return nil
}

// ParseFilterSpec returns the specified ROI instance name and version within a FilterSpec.
// Currently, only one ROI can be specified in a FilterSpec.  Multiple ROIs should use a
// different FilterSpec like "intersect" instead of "roi".
func ParseFilterSpec(spec storage.FilterSpec) (name dvid.InstanceName, v dvid.VersionID, found bool, err error) {
	var filterval string
	filterval, found = spec.GetFilterSpec("roi")
	if !found {
		return
	}
	roispec := strings.Split(filterval, ",")
	if len(roispec) != 2 {
		err = fmt.Errorf("bad ROI spec: %s", filterval)
		return
	}
	name = dvid.InstanceName(roispec[0])
	_, v, err = datastore.MatchingUUID(roispec[1])
	return
}

// Make a copy of a repository, customizing it via config.
// TODO -- modify data instance properties based on filters.
func (r *repoT) customize(v dvid.VersionID, config dvid.Config) (*repoT, rpc.Transmit, error) {
	// Since we can have names separated by commas, split them
	namesStr, found, err := config.GetString("data")
	if err != nil {
		return nil, rpc.TransmitUnknown, err
	}
	var datanames dvid.InstanceNames
	if found {
		for _, name := range strings.Split(namesStr, ",") {
			datanames = append(datanames, dvid.InstanceName(strings.TrimSpace(name)))
		}
	}

	// Check the transmission behavior: all, flatten, or deltas.
	var versions map[dvid.VersionID]struct{}

	transmitStr, found, err := config.GetString("transmit")
	if err != nil {
		return nil, rpc.TransmitUnknown, err
	}
	if !found {
		transmitStr = "all"
	}
	var transmit rpc.Transmit
	switch transmitStr {
	case "flatten":
		transmit = rpc.TransmitFlatten
		versions = map[dvid.VersionID]struct{}{
			v: struct{}{},
		}
	case "all":
		transmit = rpc.TransmitAll
		versions = r.versionSet()
	case "branch":
		transmit = rpc.TransmitBranch
		versions = r.versionSet()
	default:
		return nil, rpc.TransmitUnknown, fmt.Errorf("unknown transmit %s", transmitStr)
	}

	// Make a copy filtering by allowed data instances.
	r.RLock()
	defer r.RUnlock()

	dup, err := r.duplicate(versions, datanames)
	return dup, transmit, err
}

// SetSyncByJSON takes a JSON object of sync names and UUID, and creates the sync graph
// and sets the data instance's sync.  If replace is false (default), the new sync
// is appended to the current syncs.
func SetSyncByJSON(d dvid.Data, uuid dvid.UUID, replace bool, in io.ReadCloser) error {
	if manager == nil {
		return ErrManagerNotInitialized
	}
	jsonData := make(map[string]string)
	decoder := json.NewDecoder(in)
	if err := decoder.Decode(&jsonData); err != nil && err != io.EOF {
		return fmt.Errorf("Malformed JSON request in sync request: %v", err)
	}
	syncedCSV, ok := jsonData["sync"]
	if !ok {
		return fmt.Errorf("Could not find 'sync' value in POSTed JSON to sync request.")
	}

	syncedNames := strings.Split(syncedCSV, ",")
	if len(syncedNames) == 0 || (len(syncedNames) == 1 && syncedNames[0] == "") {
		syncedNames = []string{}
	}

	if len(syncedNames) == 0 && !replace {
		dvid.Infof("Ignored attempt to append no syncs to instance %q.\n", d.DataName())
		return nil
	}

	// Make sure all synced names currently exist under this UUID, then transform to data UUIDs.
	syncs := make(dvid.UUIDSet)
	for _, name := range syncedNames {
		data, err := GetDataByUUIDName(uuid, dvid.InstanceName(name))
		if err != nil {
			return err
		}
		syncs[data.DataUUID()] = struct{}{}
	}

	if err := SetSyncData(d, syncs, replace); err != nil {
		return err
	}
	return nil
}

// NewData returns a pointer to labelvol data.
func NewData(uuid dvid.UUID, id dvid.InstanceID, name dvid.InstanceName, c dvid.Config) (*Data, error) {
	// Make sure we have a valid labelvol source
	s, found, err := c.GetString("Source")
	if err != nil {
		return nil, err
	}
	if !found {
		return nil, fmt.Errorf("Cannot make labelsz data without valid 'Source' specifying an associated labelvol.")
	}
	srcname := dvid.InstanceName(s)
	if _, err = labelvol.GetByUUID(uuid, srcname); err != nil {
		return nil, err
	}
	c.Set("sync", s) // This will set base data sync list

	// Initialize the Data for this data type
	basedata, err := datastore.NewDataService(dtype, uuid, id, name, c)
	if err != nil {
		return nil, err
	}
	return &Data{basedata, srcname}, 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) {
	// --- Don't time labelgraph ops because they are very small and frequent.
	// --- TODO -- Implement monitoring system that aggregates logged ops instead of
	// ----------- printing out each one.
	// timedLog := dvid.NewTimeLog()

	db, err := storage.GraphStore()
	if err != nil {
		server.BadRequest(w, r, err)
		return
	}

	// make sure transaction log is created
	d.initializeLog()

	// Break URL request into arguments
	url := r.URL.Path[len(server.WebAPIPath):]
	parts := strings.Split(url, "/")
	if len(parts) < 4 {
		server.BadRequest(w, r, "No resource specified in URI")
		return
	}
	method := strings.ToLower(r.Method)
	if method == "put" {
		server.BadRequest(w, r, "PUT requests not supported")
		return
	}

	// Process help and info.
	switch parts[3] {
	case "help":
		w.Header().Set("Content-Type", "text/plain")
		fmt.Fprintln(w, d.Help())
		return
	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 "subgraph":
		// disable json schema validation (will speedup POST command)
		queryValues := r.URL.Query()
		disableSchemaT := dvid.InstanceName(queryValues.Get("unsafe"))
		disableSchema := false
		if len(disableSchemaT) != 0 && disableSchemaT == "true" {
			disableSchema = true
		}
		labelgraph, err := d.ExtractGraph(r, disableSchema)
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
		err = d.handleSubgraphBulk(ctx, db, w, labelgraph, method)
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
	case "neighbors":
		if method != "get" {
			server.BadRequest(w, r, "Only supports GETs")
			return
		}
		err := d.handleNeighbors(ctx, db, w, parts[4:])
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
	case "merge":
		if method != "post" {
			server.BadRequest(w, r, "Only supports POSTs")
			return
		}
		labelgraph, err := d.ExtractGraph(r, false)
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
		err = d.handleMerge(ctx, db, w, labelgraph)
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
	case "weight":
		if method != "post" {
			server.BadRequest(w, r, "Only supports POSTs")
			return
		}
		labelgraph, err := d.ExtractGraph(r, false)
		if err != nil {
			server.BadRequest(w, r, err)
			return
		}
		err = d.handleWeightUpdate(ctx, db, w, labelgraph)
		if err != nil {
			server.BadRequest(w, r, err)
			return
//.........这里部分代码省略.........

// Do acts as a switchboard for remote command execution
func (c *RPCConnection) Do(cmd datastore.Request, reply *datastore.Response) error {
	if reply == nil {
		dvid.Debugf("reply is nil coming in!\n")
		return nil
	}
	if cmd.Name() == "" {
		return fmt.Errorf("Server error: got empty command!")
	}

	switch cmd.Name() {

	case "help":
		reply.Text = fmt.Sprintf(RPCHelpMessage, config.RPCAddress(), config.HTTPAddress())

	case "shutdown":
		Shutdown()
		// Make this process shutdown in a second to allow time for RPC to finish.
		// TODO -- Better way to do this?
		log.Printf("DVID server halted due to 'shutdown' command.")
		reply.Text = fmt.Sprintf("DVID server at %s has been halted.\n", config.RPCAddress())
		go func() {
			time.Sleep(1 * time.Second)
			os.Exit(0)
		}()

	case "types":
		if len(cmd.Command) == 1 {
			text := "\nData Types within this DVID Server\n"
			text += "----------------------------------\n"
			mapTypes, err := datastore.Types()
			if err != nil {
				return fmt.Errorf("Error trying to retrieve data types within this DVID server!")
			}
			for url, typeservice := range mapTypes {
				text += fmt.Sprintf("%-20s %s\n", typeservice.GetTypeName(), url)
			}
			reply.Text = text
		} else {
			if len(cmd.Command) != 3 || cmd.Command[2] != "help" {
				return fmt.Errorf("Unknown types command: %q", cmd.Command)
			}
			var typename string
			cmd.CommandArgs(1, &typename)
			typeservice, err := datastore.TypeServiceByName(dvid.TypeString(typename))
			if err != nil {
				return err
			}
			reply.Text = typeservice.Help()
		}

	case "repos":
		var subcommand, alias, description, uuidStr string
		cmd.CommandArgs(1, &subcommand, &alias, &description, &uuidStr)
		switch subcommand {
		case "new":
			var assign *dvid.UUID
			if uuidStr == "" {
				assign = nil
			} else {
				u := dvid.UUID(uuidStr)
				assign = &u
			}
			root, err := datastore.NewRepo(alias, description, assign)
			if err != nil {
				return err
			}
			if err := datastore.SetRepoAlias(root, alias); err != nil {
				return err
			}
			if err := datastore.SetRepoDescription(root, description); err != nil {
				return err
			}
			reply.Text = fmt.Sprintf("New repo %q created with head node %s\n", alias, root)
		default:
			return fmt.Errorf("Unknown repos command: %q", subcommand)
		}

	case "repo":
		var uuidStr, subcommand string
		cmd.CommandArgs(1, &uuidStr, &subcommand)
		uuid, _, err := datastore.MatchingUUID(uuidStr)
		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)
//.........这里部分代码省略.........

// 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) {
		if httpUnavailable(w) {
			return
		}

		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.GetDataByUUIDName(uuid, dataname)
		if err != nil {
			BadRequest(w, r, err)
			return
		}
		v, err := datastore.VersionFromUUID(uuid)
		if err != nil {
			BadRequest(w, r, err)
			return
		}

		if data.Versioned() {
			// Make sure we aren't trying mutable methods on committed nodes.
			locked, err := datastore.LockedUUID(uuid)
			if err != nil {
				BadRequest(w, r, err)
				return
			}
			if locked && data.IsMutationRequest(r.Method, c.URLParams["keyword"]) {
				BadRequest(w, r, "Cannot do %s on endpoint %q of locked node %s", r.Method, c.URLParams["keyword"], uuid)
				return
			}
		} else {
			// Map everything to root version.
			v, err = datastore.GetRepoRootVersion(v)
			if err != nil {
				BadRequest(w, r, err)
				return
			}
		}
		ctx := datastore.NewVersionedCtx(data, v)

		// Also set the web request information in case logging needs it downstream.
		ctx.SetRequestID(middleware.GetReqID(*c))

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

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

		// TODO: setup routing for data instances as well.
		if config != nil && config.AllowTiming() {
			w.Header().Set("Timing-Allow-Origin", "*")
		}
		data.ServeHTTP(uuid, ctx, w, r)
	}
	return http.HandlerFunc(fn)
}

// handleTileReq returns a tile with appropriate Content-Type set.
func (d *Data) handleTileReq(w http.ResponseWriter, r *http.Request, parts []string) error {

	if len(parts) < 7 {
		return fmt.Errorf("'tile' request must be following by plane, scale level, and tile coordinate")
	}
	planeStr, scalingStr, coordStr := parts[4], parts[5], parts[6]
	queryStrings := r.URL.Query()

	var noblanks bool
	noblanksStr := dvid.InstanceName(queryStrings.Get("noblanks"))
	if noblanksStr == "true" {
		noblanks = true
	}

	var tilesize int32 = DefaultTileSize
	tileSizeStr := queryStrings.Get("tilesize")
	if tileSizeStr != "" {
		tilesizeInt, err := strconv.Atoi(tileSizeStr)
		if err != nil {
			return err
		}
		tilesize = int32(tilesizeInt)
	}
	size := dvid.Point2d{tilesize, tilesize}

	var formatStr string
	if len(parts) >= 8 {
		formatStr = parts[7]
	}
	if formatStr == "" {
		formatStr = DefaultTileFormat
	}

	// Parse the tile specification
	plane := dvid.DataShapeString(planeStr)
	shape, err := plane.DataShape()
	if err != nil {
		err = fmt.Errorf("Illegal tile plane: %s (%v)", planeStr, err)
		server.BadRequest(w, r, err)
		return err
	}
	scale, err := strconv.ParseUint(scalingStr, 10, 8)
	if err != nil {
		err = fmt.Errorf("Illegal tile scale: %s (%v)", scalingStr, err)
		server.BadRequest(w, r, err)
		return err
	}
	tileCoord, err := dvid.StringToPoint(coordStr, "_")
	if err != nil {
		err = fmt.Errorf("Illegal tile coordinate: %s (%v)", coordStr, err)
		server.BadRequest(w, r, err)
		return err
	}

	// Convert tile coordinate to offset.
	var ox, oy, oz int32
	switch {
	case shape.Equals(dvid.XY):
		ox = tileCoord.Value(0) * tilesize
		oy = tileCoord.Value(1) * tilesize
		oz = tileCoord.Value(2)
	case shape.Equals(dvid.XZ):
		ox = tileCoord.Value(0) * tilesize
		oy = tileCoord.Value(1)
		oz = tileCoord.Value(2) * tilesize
	case shape.Equals(dvid.YZ):
		ox = tileCoord.Value(0)
		oy = tileCoord.Value(1) * tilesize
		oz = tileCoord.Value(2) * tilesize
	default:
		return fmt.Errorf("Unknown tile orientation: %s", shape)
	}

	// Determine how this request sits in the available scaled volumes.
	geom, err := d.GetGoogleSubvolGeom(Scaling(scale), shape, dvid.Point3d{ox, oy, oz}, size)
	if err != nil {
		server.BadRequest(w, r, err)
		return err
	}

	// Send the tile.
	return d.serveTile(w, r, geom, formatStr, noblanks)
}

// ServeTile returns a tile with appropriate Content-Type set.
func (d *Data) ServeTile(uuid dvid.UUID, ctx storage.Context, w http.ResponseWriter,
	r *http.Request, parts []string) error {

	if len(parts) < 7 {
		return fmt.Errorf("'tile' request must be following by plane, scale level, and tile coordinate")
	}
	planeStr, scalingStr, coordStr := parts[4], parts[5], parts[6]
	queryValues := r.URL.Query()
	noblanksStr := dvid.InstanceName(queryValues.Get("noblanks"))
	var noblanks bool
	if noblanksStr == "true" {
		noblanks = true
	}
	var formatStr string
	if len(parts) >= 8 {
		formatStr = parts[7]
	}

	// Construct the index for this tile
	plane := dvid.DataShapeString(planeStr)
	shape, err := plane.DataShape()
	if err != nil {
		err = fmt.Errorf("Illegal tile plane: %s (%v)", planeStr, err)
		server.BadRequest(w, r, err)
		return err
	}
	scaling, err := strconv.ParseUint(scalingStr, 10, 8)
	if err != nil {
		err = fmt.Errorf("Illegal tile scale: %s (%v)", scalingStr, err)
		server.BadRequest(w, r, err)
		return err
	}
	tileCoord, err := dvid.StringToPoint(coordStr, "_")
	if err != nil {
		err = fmt.Errorf("Illegal tile coordinate: %s (%v)", coordStr, err)
		server.BadRequest(w, r, err)
		return err
	}
	if tileCoord.NumDims() != 3 {
		err = fmt.Errorf("Expected 3d tile coordinate, got %s", coordStr)
		server.BadRequest(w, r, err)
		return err
	}
	indexZYX := dvid.IndexZYX{tileCoord.Value(0), tileCoord.Value(1), tileCoord.Value(2)}
	data, err := d.getTileData(ctx, shape, Scaling(scaling), indexZYX)
	if err != nil {
		server.BadRequest(w, r, err)
		return err
	}
	if data == nil {
		if noblanks {
			http.NotFound(w, r)
			return nil
		}
		img, err := d.getBlankTileImage(uuid, shape, Scaling(scaling))
		if err != nil {
			return err
		}
		return dvid.WriteImageHttp(w, img, formatStr)
	}

	switch d.Encoding {
	case LZ4:
		var img dvid.Image
		if err := img.Deserialize(data); err != nil {
			return err
		}
		data, err = img.GetPNG()
		w.Header().Set("Content-type", "image/png")
	case PNG:
		w.Header().Set("Content-type", "image/png")
	case JPG:
		w.Header().Set("Content-type", "image/jpeg")
	}
	if err != nil {
		server.BadRequest(w, r, err)
		return err
	}
	if _, err = w.Write(data); err != nil {
		return err
	}
	return nil
}

// NewData returns a pointer to new tile data with default values.
func (dtype *Type) NewDataService(uuid dvid.UUID, id dvid.InstanceID, name dvid.InstanceName, c dvid.Config) (datastore.DataService, error) {
	// Make sure we have a valid DataService source
	sourcename, found, err := c.GetString("Source")
	if err != nil {
		return nil, err
	}
	if !found {
		return nil, fmt.Errorf("Cannot make imagetile data without valid 'Source' setting.")
	}

	// See if we want placeholder imagetile.
	placeholder, found, err := c.GetBool("Placeholder")
	if err != nil {
		return nil, err
	}

	// Determine encoding for tile storage and this dictates what kind of compression we use.
	encoding, found, err := c.GetString("Format")
	if err != nil {
		return nil, err
	}
	format := PNG
	if found {
		switch strings.ToLower(encoding) {
		case "lz4":
			format = LZ4
		case "png":
			format = PNG
		case "jpg":
			format = JPG
		default:
			return nil, fmt.Errorf("Unknown encoding specified: '%s' (should be 'lz4', 'png', or 'jpg'", encoding)
		}
	}

	// Compression is determined by encoding.  Inform user if there's a discrepancy.
	var compression string
	switch format {
	case LZ4:
		compression = "lz4"
	case PNG:
		compression = "none"
	case JPG:
		compression = "none"
	}
	compressConfig, found, err := c.GetString("Compression")
	if err != nil {
		return nil, err
	}
	if found && strings.ToLower(compressConfig) != compression {
		return nil, fmt.Errorf("Conflict between specified compression '%s' and format '%s'.  Suggest not dictating compression.",
			compressConfig, encoding)
	}
	c.Set("Compression", compression)

	// Initialize the imagetile data
	basedata, err := datastore.NewDataService(dtype, uuid, id, name, c)
	if err != nil {
		return nil, err
	}
	data := &Data{
		Data: basedata,
		Properties: Properties{
			Source:      dvid.InstanceName(sourcename),
			Placeholder: placeholder,
			Encoding:    format,
		},
	}
	return data, nil
}