func newHCA(s subspace.Subspace) highContentionAllocator {
	var hca highContentionAllocator

	hca.counters = s.Sub(0)
	hca.recent = s.Sub(1)

	return hca
}

// NewDirectoryLayer returns a new root directory (as a Directory). The
// subspaces nodeSS and contentSS control where the directory metadata and
// contents are stored. The default root directory has a nodeSS of
// subspace.FromBytes([]byte{0xFE}) and a contentSS of
// subspace.AllKeys(). Specifying more restrictive values for nodeSS and
// contentSS will allow using the directory layer alongside other content in a
// database.
//
// If allowManualPrefixes is false, all calls to CreatePrefix on the returned
// Directory (or any subdirectories) will fail, and all directory prefixes will
// be automatically allocated. The default root directory does not allow manual
// prefixes.
func NewDirectoryLayer(nodeSS, contentSS subspace.Subspace, allowManualPrefixes bool) Directory {
	var dl directoryLayer

	dl.nodeSS = subspace.FromBytes(nodeSS.Bytes())
	dl.contentSS = subspace.FromBytes(contentSS.Bytes())

	dl.allowManualPrefixes = allowManualPrefixes

	dl.rootNode = dl.nodeSS.Sub(dl.nodeSS.Bytes())
	dl.allocator = newHCA(dl.rootNode.Sub([]byte("hca")))

	return dl
}

func (dl directoryLayer) subdirNodes(tr fdb.Transaction, node subspace.Subspace) []subspace.Subspace {
	sd := node.Sub(_SUBDIRS)

	rr := tr.GetRange(sd, fdb.RangeOptions{})
	ri := rr.Iterator()

	var ret []subspace.Subspace

	for ri.Advance() {
		kv := ri.MustGet()

		ret = append(ret, dl.nodeWithPrefix(kv.Value))
	}

	return ret
}

func (hca highContentionAllocator) allocate(tr fdb.Transaction, s subspace.Subspace) (subspace.Subspace, error) {
	rr := tr.Snapshot().GetRange(hca.counters, fdb.RangeOptions{Limit: 1, Reverse: true})
	kvs := rr.GetSliceOrPanic()

	var start, count int64

	if len(kvs) == 1 {
		t, e := hca.counters.Unpack(kvs[0].Key)
		if e != nil {
			return nil, e
		}
		start = t[0].(int64)

		e = binary.Read(bytes.NewBuffer(kvs[0].Value), binary.LittleEndian, &count)
		if e != nil {
			return nil, e
		}
	}

	window := windowSize(start)

	if (count+1)*2 >= window {
		// Advance the window
		tr.ClearRange(fdb.KeyRange{hca.counters, append(hca.counters.Sub(start).FDBKey(), 0x00)})
		start += window
		tr.ClearRange(fdb.KeyRange{hca.recent, hca.recent.Sub(start)})
		window = windowSize(start)
	}

	// Increment the allocation count for the current window
	tr.Add(hca.counters.Sub(start), oneBytes)

	for {
		// As of the snapshot being read from, the window is less than half
		// full, so this should be expected to take 2 tries.  Under high
		// contention (and when the window advances), there is an additional
		// subsequent risk of conflict for this transaction.
		candidate := rand.Int63n(window) + start
		key := hca.recent.Sub(candidate)
		if tr.Get(key).MustGet() == nil {
			tr.Set(key, []byte(""))
			return s.Sub(candidate), nil
		}
	}
}

func (dl directoryLayer) subdirNames(rtr fdb.ReadTransaction, node subspace.Subspace) ([]string, error) {
	sd := node.Sub(_SUBDIRS)

	rr := rtr.GetRange(sd, fdb.RangeOptions{})
	ri := rr.Iterator()

	var ret []string

	for ri.Advance() {
		kv := ri.MustGet()

		p, e := sd.Unpack(kv.Key)
		if e != nil {
			return nil, e
		}

		ret = append(ret, p[0].(string))
	}

	return ret, nil
}

func GetLastKeyFuture(tr KeyReader, space subspace.Subspace) *LastKeyFuture {
	_, end := space.FDBRangeKeys()
	key := tr.GetKey(fdb.LastLessThan(end))

	return &LastKeyFuture{key, space}
}

func (dl directoryLayer) createOrOpen(rtr fdb.ReadTransaction, tr *fdb.Transaction, path []string, layer []byte, prefix []byte, allowCreate, allowOpen bool) (DirectorySubspace, error) {
	if e := dl.checkVersion(rtr, nil); e != nil {
		return nil, e
	}

	if prefix != nil && !dl.allowManualPrefixes {
		if len(dl.path) == 0 {
			return nil, errors.New("cannot specify a prefix unless manual prefixes are enabled")
		} else {
			return nil, errors.New("cannot specify a prefix in a partition")
		}
	}

	if len(path) == 0 {
		return nil, errors.New("the root directory cannot be opened")
	}

	existingNode := dl.find(rtr, path).prefetchMetadata(rtr)
	if existingNode.exists() {
		if existingNode.isInPartition(nil, false) {
			subpath := existingNode.getPartitionSubpath()
			enc, e := existingNode.getContents(dl, nil)
			if e != nil {
				return nil, e
			}
			return enc.(directoryPartition).createOrOpen(rtr, tr, subpath, layer, prefix, allowCreate, allowOpen)
		}

		if !allowOpen {
			return nil, errors.New("the directory already exists")
		}

		if layer != nil && bytes.Compare(existingNode._layer.MustGet(), layer) != 0 {
			return nil, errors.New("the directory was created with an incompatible layer")
		}

		return existingNode.getContents(dl, nil)
	}

	if !allowCreate {
		return nil, errors.New("the directory does not exist")
	}

	if e := dl.checkVersion(rtr, tr); e != nil {
		return nil, e
	}

	if prefix == nil {
		newss, e := dl.allocator.allocate(*tr, dl.contentSS)
		if e != nil {
			return nil, fmt.Errorf("unable to allocate new directory prefix (%s)", e.Error())
		}

		if !isRangeEmpty(rtr, newss) {
			return nil, fmt.Errorf("the database has keys stored at the prefix chosen by the automatic prefix allocator: %v", prefix)
		}

		prefix = newss.Bytes()

		pf, e := dl.isPrefixFree(rtr.Snapshot(), prefix)
		if e != nil {
			return nil, e
		}
		if !pf {
			return nil, errors.New("the directory layer has manually allocated prefixes that conflict with the automatic prefix allocator")
		}
	} else {
		pf, e := dl.isPrefixFree(rtr, prefix)
		if e != nil {
			return nil, e
		}
		if !pf {
			return nil, errors.New("the given prefix is already in use")
		}
	}

	var parentNode subspace.Subspace

	if len(path) > 1 {
		pd, e := dl.createOrOpen(rtr, tr, path[:len(path)-1], nil, nil, true, true)
		if e != nil {
			return nil, e
		}
		parentNode = dl.nodeWithPrefix(pd.Bytes())
	} else {
		parentNode = dl.rootNode
	}

	if parentNode == nil {
		return nil, errors.New("the parent directory does not exist")
	}

	node := dl.nodeWithPrefix(prefix)
	tr.Set(parentNode.Sub(_SUBDIRS, path[len(path)-1]), prefix)

	if layer == nil {
		layer = []byte{}
	}

	tr.Set(node.Sub([]byte("layer")), layer)
//.........这里部分代码省略.........