func NewRegCluster(name string, id *xi.NodeID, attrs uint64,
	maxSize, epCount uint32) (rc *RegCluster, err error) {

	var m ha.HAMT

	if name == "" {
		name = "xlCluster"
	}
	nameMap := make(map[string]*ClientInfo)
	if epCount < 1 {
		err = ClusterMembersMustHaveEndPoint
	}
	if err == nil && maxSize < 1 {
		//err = ClusterMustHaveTwo
		err = ClusterMustHaveMember
	} else {
		t := uint(xm.NextExp2_32(maxSize))
		m, err = ha.NewHAMT(DEFAULT_W, t)
	}
	if err == nil {
		rc = &RegCluster{
			Attrs:         attrs,
			Name:          name,
			ID:            id.Value(),
			epCount:       epCount,
			maxSize:       maxSize,
			MembersByName: nameMap,
			MembersByID:   m,
		}
	}
	return
}

func (reg *Registry) ContainsID(n *xi.NodeID) (found bool, err error) {
	found, _, err = reg.idFilter.IsMember(n.Value())

	// DEBUG
	reg.Logger.Printf("Registry.ContainsID(%x) returning %v, %v\n",
		n.Value(), found, err)
	// END
	return
}

// Add a file to the collection.  This operation must be idempotent.
func (mc *MemCache) Add(id *xi.NodeID, b []byte) (err error) {

	key := id.Value()

	// XXX POSSIBLE DEADLOCK
	mc.mu.Lock()
	defer mc.mu.Unlock()

	value, err := mc.idMap.Find(key)
	if err == nil {
		if value == nil {
			mc.idMap.Insert(key, b)
			mc.itemCount++
			mc.byteCount += uint64(len(b))
		}
	}
	return
}

func (reg *Registry) InsertID(n *xi.NodeID) (err error) {
	b := n.Value()
	// DEBUG
	reg.Logger.Printf("Registry.InsertID(%x)\n", b)
	// END
	found, _, err := reg.idFilter.IsMember(b)
	if err == nil {
		if found {
			// DEBUG
			reg.Logger.Printf("  id is already registered\n")
			// END
			err = IDAlreadyInUse
		} else {
			err = reg.idFilter.Insert(b)
			// DEBUG
			reg.Logger.Printf("  added %x to registry; err is %v\n", b, err)
			// END
		}
	}
	return
}

// Datum is declared a NodeID to restrict its value to certain byteslice
// lengths.
func NewChunk(datum *xi.NodeID, ndx uint32, data []byte) (
	ch *Chunk, err error) {

	if datum == nil {
		err = NilDatum
	} else if data == nil {
		err = NilData
	} else if len(data) == 0 {
		err = ZeroLengthChunk
	} else if len(data) > MAX_CHUNK_BYTES {
		err = ChunkTooLong
	} else {
		msgHash := datum.Value()
		realLen := len(data)
		adjLen := ((realLen + WORD_BYTES - 1) / WORD_BYTES) * WORD_BYTES
		paddingBytes := adjLen - realLen
		packet := make([]byte, DATUM_OFFSET)
		datumPadding := make([]byte, DATUM_PADDING)
		ch = &Chunk{packet: packet}
		ch.setLength(uint32(realLen)) // length of the data part
		ch.setIndex(ndx)              // index of this chunk in overall message
		ch.packet = append(ch.packet, msgHash...)
		ch.packet = append(ch.packet, datumPadding...)
		ch.packet = append(ch.packet, data...)
		if paddingBytes > 0 {
			padding := make([]byte, paddingBytes)
			ch.packet = append(ch.packet, padding...)
		}
		// calculate the SHA1 hash of the chunk
		d := sha1.New()
		d.Write(ch.packet)
		chunkHash := d.Sum(nil)

		// append that to the packet
		ch.packet = append(ch.packet, chunkHash...)
	}
	return
}

func doClientMsg(h *InHandler) {
	var err error
	defer func() {
		h.errOut = err
	}()

	// DEBUG
	regName := h.reg.GetName()
	h.reg.Logger.Printf("doClientMsg: regName is %s\n", regName)
	// END

	// Examine incoming message -------------------------------------
	var (
		name   string
		attrs  uint64
		nodeID *xi.NodeID
		ck, sk *rsa.PublicKey
		myEnds []string
		hash   []byte
		cm     *ClientInfo
	)

	// XXX We should accept EITHER clientName + token OR clientID
	// This implementation only accepts a token.

	clientMsg := h.msgIn
	clientSpecs := clientMsg.GetMemberSpecs()
	name = clientSpecs.GetName()
	attrs = clientSpecs.GetAttrs()
	ckBytes := clientSpecs.GetCommsKey()
	skBytes := clientSpecs.GetSigKey()
	digSig := clientSpecs.GetDigSig()

	ck, err = xc.RSAPubKeyFromWire(ckBytes)
	if err == nil {
		sk, err = xc.RSAPubKeyFromWire(skBytes)
		if err == nil {
			myEnds = clientSpecs.GetMyEnds() // a string array
		}
	}
	if err == nil {
		// calculate hash over fields in canonical order XXX EXCLUDING ID
		aBytes := make([]byte, 8)
		binary.LittleEndian.PutUint64(aBytes, attrs)
		d := sha1.New()
		d.Write([]byte(name))
		d.Write(aBytes)
		d.Write(ckBytes)
		d.Write(skBytes)
		for i := 0; i < len(myEnds); i++ {
			d.Write([]byte(myEnds[i]))
		}
		hash = d.Sum(nil)
		// verify the digital signature
		err = rsa.VerifyPKCS1v15(sk, crypto.SHA1, hash, digSig)
	}
	if err == nil {
		id := clientSpecs.GetID()
		// DEBUG
		if id == nil {
			h.reg.Logger.Println("  doClientMsg: id from Specs is NIL")
		} else {
			h.reg.Logger.Printf("  doClientMsg: id from Specs is %x\n", id)
		}
		// END
		if id == nil {
			nodeID, err = h.reg.InsertUniqueNodeID()
			// DEBUG
			h.reg.Logger.Printf("  doClientMsg: inserting %x returned %v\n", id, err)
			// END
			if err == nil {
				id := nodeID.Value()
				// this is echoed to the console
				h.reg.Logger.Printf("doClientMsg: assigning new MemberID %xi, user %s",
					id, name)
			}
		} else {
			// must be known to the registry
			nodeID, err = xi.New(id)
			if err == nil {
				var found bool
				found, err = h.reg.ContainsID(nodeID)
				if err == nil && !found {
					err = h.reg.InsertID(nodeID)
				}
			}
		}
	}
	// Take appropriate action --------------------------------------
	if err == nil || err == IDAlreadyInUse {
		// The appropriate action is to hang a token for this client off
		// the InHandler.
		cm, err = NewClientInfo(name, nodeID, ck, sk, attrs, myEnds)
		if err == nil {
			h.thisClient = cm
		}
	}
	if err == nil {
		// Prepare reply to client --------------------------------------
		// In this implementation We simply accept the client's proposed
//.........这里部分代码省略.........

func doCreateMsg(h *InHandler) {
	var err error
	defer func() {
		h.errOut = err
	}()
	// Examine incoming message -------------------------------------
	var (
		clusterID *xi.NodeID
		index     int
	)

	createMsg := h.msgIn
	clusterName := createMsg.GetClusterName()
	clusterAttrs := createMsg.GetClusterAttrs()
	clusterMaxSize := createMsg.GetClusterMaxSize()
	endPointCount := createMsg.GetEndPointCount()

	// Take appropriate action --------------------------------------

	// Determine whether the cluster exists.  If it does, we will just
	// use its existing properties.

	h.reg.mu.RLock()
	cluster, exists := h.reg.ClustersByName[clusterName]
	h.reg.mu.RUnlock()

	if exists {
		h.reg.Logger.Printf("doCreateMsg: cluster %s already exists\n", clusterName)
		clusterMaxSize = cluster.maxSize
		clusterAttrs = cluster.Attrs
		endPointCount = cluster.epCount
		if cluster.ID == nil {
			h.reg.Logger.Println("  no ID for cluster %s\n", clusterName)
			clusterID, _ = xi.New(nil)
		} else {
			clusterID, _ = xi.New(cluster.ID)
		}
		// XXX index not assigned
	} else {
		h.reg.Logger.Printf("doCreateMsg: new cluster %s\n", clusterName)
		attrs := uint64(0)
		if clusterMaxSize < MIN_CLUSTER_SIZE {
			clusterMaxSize = MIN_CLUSTER_SIZE
		} else if clusterMaxSize > MAX_CLUSTER_SIZE {
			clusterMaxSize = MAX_CLUSTER_SIZE
		}
		// Assign a quasi-random cluster ID, adding it to the registry
		clusterID, err = h.reg.InsertUniqueNodeID()
		if err == nil {
			cluster, err = NewRegCluster(clusterName, clusterID, attrs,
				clusterMaxSize, endPointCount)
			h.reg.Logger.Printf("cluster %s assigning ID %x\n",
				clusterName, clusterID.Value())
		}
		if err == nil {
			h.cluster = cluster
			index, err = h.reg.AddCluster(cluster)
			// XXX index not used
		}
	}
	_ = index // INDEX IS NOT BEING USED

	if err == nil {
		// Prepare reply to client --------------------------------------
		op := XLRegMsg_CreateReply
		id := clusterID.Value()
		h.msgOut = &XLRegMsg{
			Op:             &op,
			ClusterID:      id,
			ClusterMaxSize: &clusterMaxSize,
			ClusterAttrs:   &clusterAttrs,
			EndPointCount:  &endPointCount,
		}
		// Set exit state -----------------------------------------------
		h.exitState = CREATE_REQUEST_RCVD
	}
}