func (r *Replica) bcastPrepare(replica int32, instance int32, ballot int32) {
	defer func() {
		if err := recover(); err != nil {
			dlog.Println("Prepare bcast failed:", err)
		}
	}()
	args := &gpaxosproto.Prepare{r.Id, instance, ballot}

	n := r.N - 1
	//TODO: fix quorum size
	if r.Thrifty {
		n = r.N >> 1
	}
	q := r.Id
	var w *bufio.Writer
	for sent := 0; sent < n; {
		q = (q + 1) % int32(r.N)
		if q == r.Id {
			dlog.Println("Not enough replicas alive!")
			break
		}
		if !r.Alive[q] {
			continue
		}
		sent++
		w = r.PeerWriters[q]
		w.WriteByte(gpaxosproto.PREPARE)
		args.Marshal(w)
		w.Flush()
	}
}

func (r *Replica) handle1b(msg *gpaxosproto.M_1b) {
	if msg.Balnum != r.crtBalnum {
		log.Println("1b from a different ballot")
		return
	}

	crtbal := r.ballotArray[r.crtBalnum]

	if crtbal.status != PHASE1 {
		//delayed 1b
		return
	}

	dlog.Println("msg.Cstruct: ", msg.Cstruct)
	crtbal.lb.cstructs = append(crtbal.lb.cstructs, msg.Cstruct)
	count := len(crtbal.lb.cstructs)

	//is it sufficient to have the same initial quorum size for both fast and slow rounds?
	if (r.fastRound && count == r.fastQSize) ||
		(!r.fastRound && count == r.N/2+1) {
		_, _, crtbal.cstruct = r.learn(true)
		dlog.Println("LUB:", crtbal.cstruct)
		r.bcast2a(r.crtBalnum, crtbal.cstruct, r.fastRound)
		crtbal.lb.cstructs = make([][]int32, r.N)
		crtbal.status = PHASE2
		if r.fastRound && ALL_TO_ALL {
			r.bcast2b(&gpaxosproto.M_2b{r.Id, r.crtBalnum, crtbal.cstruct, crtbal.cstruct})
		}
	}
}

func (r *Replica) bcastPrepare(replica int32, instance int32, ballot int32) {
	defer func() {
		if err := recover(); err != nil {
			dlog.Println("Prepare bcast failed:", err)
		}
	}()
	args := &epaxosproto.Prepare{r.Id, replica, instance, ballot}

	n := r.N - 1
	if r.Thrifty {
		n = r.N / 2
	}
	q := r.Id
	for sent := 0; sent < n; {
		q = (q + 1) % int32(r.N)
		if q == r.Id {
			dlog.Println("Not enough replicas alive!")
			break
		}
		if !r.Alive[q] {
			continue
		}
		r.SendMsg(q, r.prepareRPC, args)
		sent++
	}
}

func (r *Replica) tryToLearn() {
	var glb []int32
	var conflict bool
	crtbal := r.ballotArray[r.crtBalnum]

	if conflict, glb, _ = r.learn(false); conflict {
		log.Println("Conflict")
		if r.isLeader {
			r.startHigherBallot()
		}
	} else if glb != nil {
		dlog.Println("Got GLB:", glb)
		for _, cid := range glb {
			dlog.Println("Committing command ", cid)
			r.committed[cid] = true
			crtbal.lb.committed++
			if prop, present := r.commandReplies[cid]; present {
				r.ReplyProposeTS(&genericsmrproto.ProposeReplyTS{TRUE, cid, state.NIL, prop.Timestamp}, prop.Reply)
				delete(r.commandReplies, cid)
			}
		}
		if r.isLeader && crtbal.lb.committed >= CMDS_PER_BALLOT {
			r.startHigherBallot()
		}
	}
}

func (r *Replica) bcastCommit(cstruct []int32) {
	defer func() {
		if err := recover(); err != nil {
			dlog.Println("Commit bcast failed:", err)
		}
	}()
	args := &gpaxosproto.Commit{cstruct}

	n := r.N - 1
	q := r.Id
	var w *bufio.Writer
	for sent := 0; sent < n; {
		q = (q + 1) % int32(r.N)
		if q == r.Id {
			break
		}
		if !r.Alive[q] {
			continue
		}
		sent++
		w = r.PeerWriters[q]
		w.WriteByte(gpaxosproto.COMMIT)
		args.Marshal(w)
		w.Flush()
	}
}

func (r *Replica) bcast2a(balnum int32, cstruct []int32, fast bool) {
	defer func() {
		if err := recover(); err != nil {
			dlog.Println("1a bcast failed:", err)
		}
	}()
	args := &gpaxosproto.M_2a{r.Id, balnum, cstruct}

	n := r.N - 1
	if r.Thrifty {
		if fast {
			n = r.fastQSize - 1
		} else {
			n = r.N >> 1
		}
	}
	q := r.Id
	var w *bufio.Writer
	for sent := 0; sent < n; {
		q = (q + 1) % int32(r.N)
		if q == r.Id {
			break
		}
		if !r.Alive[q] {
			continue
		}
		sent++
		w = r.PeerWriters[q]
		w.WriteByte(gpaxosproto.M2A)
		args.Marshal(w)
		w.Flush()
	}
}

func (r *Replica) bcastSkip(startInstance int32, endInstance int32, exceptReplica int32) {
	defer func() {
		if err := recover(); err != nil {
			dlog.Println("Skip bcast failed:", err)
		}
	}()
	sk.LeaderId = r.Id
	sk.StartInstance = startInstance
	sk.EndInstance = endInstance
	args := &sk
	//args := &menciusproto.Skip{r.Id, startInstance, endInstance}

	n := r.N - 1
	q := r.Id

	for sent := 0; sent < n; {
		q = (q + 1) % int32(r.N)
		if q == r.Id {
			break
		}
		if !r.Alive[q] || q == exceptReplica {
			continue
		}
		sent++
		r.SendMsgNoFlush(q, r.skipRPC, args)
	}
}

func (r *Replica) bcastAccept(replica int32, instance int32, ballot int32, count int32, seq int32, deps [DS]int32) {
	defer func() {
		if err := recover(); err != nil {
			dlog.Println("Accept bcast failed:", err)
		}
	}()

	ea.LeaderId = r.Id
	ea.Replica = replica
	ea.Instance = instance
	ea.Ballot = ballot
	ea.Count = count
	ea.Seq = seq
	ea.Deps = deps
	args := &ea

	n := r.N - 1
	if r.Thrifty {
		n = r.N / 2
	}

	sent := 0
	for q := 0; q < r.N-1; q++ {
		if !r.Alive[r.PreferredPeerOrder[q]] {
			continue
		}
		r.SendMsg(r.PreferredPeerOrder[q], r.acceptRPC, args)
		sent++
		if sent >= n {
			break
		}
	}
}

func (r *Replica) bcastCommit(replica int32, instance int32, cmds []state.Command, seq int32, deps [DS]int32) {
	defer func() {
		if err := recover(); err != nil {
			dlog.Println("Commit bcast failed:", err)
		}
	}()
	ec.LeaderId = r.Id
	ec.Replica = replica
	ec.Instance = instance
	ec.Command = cmds
	ec.Seq = seq
	ec.Deps = deps
	args := &ec
	ecs.LeaderId = r.Id
	ecs.Replica = replica
	ecs.Instance = instance
	ecs.Count = int32(len(cmds))
	ecs.Seq = seq
	ecs.Deps = deps
	argsShort := &ecs

	sent := 0
	for q := 0; q < r.N-1; q++ {
		if !r.Alive[r.PreferredPeerOrder[q]] {
			continue
		}
		if r.Thrifty && sent >= r.N/2 {
			r.SendMsg(r.PreferredPeerOrder[q], r.commitRPC, args)
		} else {
			r.SendMsg(r.PreferredPeerOrder[q], r.commitShortRPC, argsShort)
			sent++
		}
	}
}

func (r *Replica) bcastPreAccept(replica int32, instance int32, ballot int32, cmds []state.Command, seq int32, deps [DS]int32) {
	defer func() {
		if err := recover(); err != nil {
			dlog.Println("PreAccept bcast failed:", err)
		}
	}()
	pa.LeaderId = r.Id
	pa.Replica = replica
	pa.Instance = instance
	pa.Ballot = ballot
	pa.Command = cmds
	pa.Seq = seq
	pa.Deps = deps
	args := &pa

	n := r.N - 1
	if r.Thrifty {
		n = r.N / 2
	}

	sent := 0
	for q := 0; q < r.N-1; q++ {
		if !r.Alive[r.PreferredPeerOrder[q]] {
			continue
		}
		r.SendMsg(r.PreferredPeerOrder[q], r.preAcceptRPC, args)
		sent++
		if sent >= n {
			break
		}
	}
}

func (r *Replica) bcastCommit(instance int32, skip uint8, nbInstToSkip int32, command state.Command) {
	defer func() {
		if err := recover(); err != nil {
			dlog.Println("Commit bcast failed:", err)
		}
	}()
	mc.LeaderId = r.Id
	mc.Instance = instance
	mc.Skip = skip
	mc.NbInstancesToSkip = nbInstToSkip
	//mc.Command = command
	//args := &menciusproto.Commit{r.Id, instance, skip, nbInstToSkip, command}
	args := &mc

	n := r.N - 1
	q := r.Id

	for sent := 0; sent < n; {
		q = (q + 1) % int32(r.N)
		if q == r.Id {
			break
		}
		if !r.Alive[q] {
			continue
		}
		sent++
		r.SendMsg(q, r.commitRPC, args)
	}
}

func (r *Replica) bcastTryPreAccept(replica int32, instance int32, ballot int32, cmds []state.Command, seq int32, deps [DS]int32) {
	defer func() {
		if err := recover(); err != nil {
			dlog.Println("PreAccept bcast failed:", err)
		}
	}()
	tpa.LeaderId = r.Id
	tpa.Replica = replica
	tpa.Instance = instance
	tpa.Ballot = ballot
	tpa.Command = cmds
	tpa.Seq = seq
	tpa.Deps = deps
	args := &pa

	for q := int32(0); q < int32(r.N); q++ {
		if q == r.Id {
			continue
		}
		if !r.Alive[q] {
			continue
		}
		r.SendMsg(q, r.tryPreAcceptRPC, args)
	}
}

func (r *Replica) bcastPrepare(instance int32, ballot int32) {
	defer func() {
		if err := recover(); err != nil {
			dlog.Println("Prepare bcast failed:", err)
		}
	}()
	args := &menciusproto.Prepare{r.Id, instance, ballot}

	n := r.N - 1
	if r.Thrifty {
		n = r.N >> 1
	}
	q := r.Id

	for sent := 0; sent < n; {
		q = (q + 1) % int32(r.N)
		if q == r.Id {
			break
		}
		if !r.Alive[q] {
			continue
		}
		sent++
		r.SendMsg(q, r.prepareRPC, args)
	}
}

func (r *Replica) handle2b(msg *gpaxosproto.M_2b) {
	if msg.Balnum != r.crtBalnum {
		dlog.Println("2b from a different ballot")
		return
	}

	crtbal := r.ballotArray[r.crtBalnum]

	if r.isLeader && crtbal.status != PHASE2 {
		log.Println("2b before its time")
		return
	}

	crtbal.lb.cstructs[msg.ReplicaId] = msg.Cstruct
	dlog.Printf("Replica %d 2b msg.Cstruct: ", msg.ReplicaId)
	dlog.Println(msg.Cstruct)
	dlog.Println("my cstruct:", crtbal.cstruct)

	crtbal.lb.cstructs[r.Id] = crtbal.cstruct

	r.tryToLearn()
}

func (r *Replica) handle2a(msg *gpaxosproto.M_2a) {
	if r.isLeader {
		log.Println("Received 2a even though I am the leader")
		return
	}

	if r.leaderId != msg.LeaderId {
		log.Println("Received 2a from unrecognized leader")
		return
	}

	if r.crtBalnum != msg.Balnum {
		log.Println("Received 2a for different ballot: ", msg.Balnum)
		return
	}

	crtbal := r.ballotArray[r.crtBalnum]
	crtbal.received2a = true
	crtbal.status = PHASE2

	dlog.Println("old cstruct", crtbal.cstruct)

	cids := make([]int32, 0)

	if crtbal.cstruct == nil || len(crtbal.cstruct) == 0 {
		crtbal.cstruct = msg.Cstruct
	} else {
		for _, ocid := range crtbal.cstruct {
			present := false
			for _, cid := range msg.Cstruct {
				if cid == ocid {
					present = true
					break
				}
			}
			if !present {
				msg.Cstruct = append(msg.Cstruct, ocid)
				cids = append(cids, ocid)
			}
		}
		crtbal.cstruct = msg.Cstruct
	}

	if ALL_TO_ALL {
		r.bcast2b(&gpaxosproto.M_2b{r.Id, r.crtBalnum, crtbal.cstruct, cids})
		r.tryToLearn()
	} else {
		r.send2b(&gpaxosproto.M_2b{r.Id, r.crtBalnum, crtbal.cstruct, cids}, r.PeerWriters[r.leaderId])
	}
}

func (r *Replica) bcast2b(msg *gpaxosproto.M_2b) {
	defer func() {
		if err := recover(); err != nil {
			dlog.Println("Commit bcast failed:", err)
		}
	}()

	for rid, w := range r.PeerWriters {
		if int32(rid) == r.Id {
			continue
		}
		w.WriteByte(gpaxosproto.M2B)
		msg.Marshal(w)
		for _, cid := range msg.Cids {
			cmd := r.commands[cid]
			cmd.Marshal(w)
		}
		w.Flush()
	}
}

func (r *Replica) handlePrepare(prepare *menciusproto.Prepare) {
	inst := r.instanceSpace[prepare.Instance]

	if inst == nil {
		dlog.Println("Replying OK to null-instance Prepare")
		r.replyPrepare(prepare.LeaderId, &menciusproto.PrepareReply{prepare.Instance,
			TRUE,
			-1,
			FALSE,
			0,
			state.Command{state.NONE, 0, 0}})

		r.instanceSpace[prepare.Instance] = &Instance{false,
			0,
			nil,
			prepare.Ballot,
			PREPARING,
			nil}
	} else {
		ok := TRUE
		if prepare.Ballot < inst.ballot {
			ok = FALSE
		}
		if inst.command == nil {
			inst.command = &state.Command{state.NONE, 0, 0}
		}
		skipped := FALSE
		if inst.skipped {
			skipped = TRUE
		}
		r.replyPrepare(prepare.LeaderId, &menciusproto.PrepareReply{prepare.Instance,
			ok,
			inst.ballot,
			skipped,
			int32(inst.nbInstSkipped),
			*inst.command})
	}
}

func (r *Replica) learn(getLub bool) (conflict bool, glb []int32, lub []int32) {
	if r.crtBalnum < 0 {
		return false, nil, nil
	}

	crtbal := r.ballotArray[r.crtBalnum]

	if len(crtbal.lb.cstructs) == 0 {
		return false, nil, nil
	}

	idToNode := make(map[int32]*node, 2*len(crtbal.cstruct))

	// build directed graph

	dlog.Println(crtbal.lb.cstructs)
	for i := 0; i < len(crtbal.lb.cstructs); i++ {
		cs := crtbal.lb.cstructs[i]
		for idx, cid := range cs {
			var n *node
			var present bool
			crtCmd := r.commands[cid]
			if _, present = r.committed[cid]; present {
				continue
			}
			if n, present = idToNode[cid]; !present {
				n = &node{0, make(map[int32]int, 2), WHITE}
				idToNode[cid] = n
			}
			n.count++
			for j := 0; j < idx; j++ {
				if _, present = r.committed[cs[j]]; present {
					continue
				}
				if crtCmd == nil {
					log.Println("crtCmd is nil")
					return false, nil, nil
				}
				if r.commands[cs[j]] == nil {
					log.Println("cs[j] is nil")
					return false, nil, nil
				}
				if !state.Conflict(crtCmd, r.commands[cs[j]]) {
					continue
				}
				n.outEdges[cs[j]] = n.outEdges[cs[j]] + 1
			}
		}
	}

	// hack

	/*    conflict = false
	      glb = make([]int32, 0)
	      lub = make([]int32, 0)

	      for cid, n := range idToNode {
	          if n.count >= r.fastQSize {
	              glb = append(glb, cid)
	          }
	          lub = append(lub, cid)
	      }

	      return false, glb, lub*/

	// depth-first search

	for cid, n := range idToNode {
		if n.color == WHITE {
			var conf bool
			conf, glb, lub = r.dfs(cid, n, glb, lub, idToNode)
			conflict = conflict || conf
		}
	}
	/*
	   if getLub && conflict {
	       //sort out lub
	       done := false
	       for !done {
	           done = true
	           for i := 1; i < len(lub) - 1; i++ {
	               for j := i + 1; j < len(lub); j++ {
	                   u := lub[i]
	                   v := lub[j]
	                   cu := r.commands[u]
	                   cv := r.commands[v]
	                   if !state.Conflict(cu, cv) {
	                       continue
	                   }
	                   nu := idToNode[u]
	                   nv := idToNode[v]
	                   if nv.count - nv.outEdges[u] > nu.count - nu.outEdges[v] {
	                       lub[i] = v
	                       lub[j] = u
	                       done = false
	                   }
	               }
	           }
	       }
	   }
//.........这里部分代码省略.........

func (r *Replica) run() {
	r.ConnectToPeers()

	dlog.Println("Waiting for client connections")

	go r.WaitForClientConnections()

	if r.Exec {
		go r.executeCommands()
	}

	go r.clock()

	for !r.Shutdown {

		select {

		case propose := <-r.ProposeChan:
			//got a Propose from a client
			dlog.Printf("Proposal with id %d\n", propose.CommandId)
			r.handlePropose(propose)
			break

		case skipS := <-r.skipChan:
			skip := skipS.(*menciusproto.Skip)
			//got a Skip from another replica
			dlog.Printf("Skip for instances %d-%d\n", skip.StartInstance, skip.EndInstance)
			r.handleSkip(skip)

		case prepareS := <-r.prepareChan:
			prepare := prepareS.(*menciusproto.Prepare)
			//got a Prepare message
			dlog.Printf("Received Prepare from replica %d, for instance %d\n", prepare.LeaderId, prepare.Instance)
			r.handlePrepare(prepare)
			break

		case acceptS := <-r.acceptChan:
			accept := acceptS.(*menciusproto.Accept)
			//got an Accept message
			dlog.Printf("Received Accept from replica %d, for instance %d\n", accept.LeaderId, accept.Instance)
			r.handleAccept(accept)
			break

		case commitS := <-r.commitChan:
			commit := commitS.(*menciusproto.Commit)
			//got a Commit message
			dlog.Printf("Received Commit from replica %d, for instance %d\n", commit.LeaderId, commit.Instance)
			r.handleCommit(commit)
			break

		case prepareReplyS := <-r.prepareReplyChan:
			prepareReply := prepareReplyS.(*menciusproto.PrepareReply)
			//got a Prepare reply
			dlog.Printf("Received PrepareReply for instance %d\n", prepareReply.Instance)
			r.handlePrepareReply(prepareReply)
			break

		case acceptReplyS := <-r.acceptReplyChan:
			acceptReply := acceptReplyS.(*menciusproto.AcceptReply)
			//got an Accept reply
			dlog.Printf("Received AcceptReply for instance %d\n", acceptReply.Instance)
			r.handleAcceptReply(acceptReply)
			break

		case delayedSkip := <-r.delayedSkipChan:
			r.handleDelayedSkip(delayedSkip)
			break

		case <-r.clockChan:
			if lastSeenInstance == r.blockingInstance {
				r.noCommitFor++
			} else {
				r.noCommitFor = 0
				lastSeenInstance = r.blockingInstance
			}
			if r.noCommitFor >= 50+int(r.Id) && r.crtInstance >= r.blockingInstance+int32(r.N) {
				r.noCommitFor = 0
				dlog.Printf("Doing force commit\n")
				r.forceCommit()
			}
			break
		}
	}
}

func (r *Replica) run() {
	if r.Id == 0 {
		r.isLeader = true
	}

	r.ConnectToPeersNoListeners()

	for rid, peerReader := range r.PeerReaders {
		if int32(rid) == r.Id {
			continue
		}
		go r.handleReplicaConnection(rid, peerReader)
	}

	dlog.Println("Waiting for client connections")

	go r.WaitForClientConnections()

	/*if r.Exec {
	    go r.executeCommands()
	}*/

	clockChan = make(chan bool, 1)
	go r.clock()

	if r.isLeader {
		r.crtBalnum = 0
		r.fastRound = true
		r.ballotArray[0] = &Ballot{nil, 0, PHASE1, false, &LeaderBookkeeping{cstructs: make([][]int32, 0)}}
		r.ballotArray[0].lb.cstructs = append(r.ballotArray[0].lb.cstructs, make([]int32, 0))
		r.bcast1a(0, true)
	}

	for !r.Shutdown {

		if r.crtBalnum >= 0 && len(r.ballotArray[r.crtBalnum].cstruct) >= CMDS_PER_BALLOT {

			select {

			case prepare := <-r.prepareChan:
				//got a Prepare message
				dlog.Printf("Received Prepare for balnum %d\n", prepare.Balnum)
				r.commandsMutex.Lock()
				r.handlePrepare(prepare)
				r.commandsMutex.Unlock()
				break

			case msg := <-r.m1aChan:
				dlog.Printf("Received 1a for balnum %d @ replica %d\n", msg.Balnum, r.Id)
				r.commandsMutex.Lock()
				r.handle1a(msg)
				r.commandsMutex.Unlock()
				break

			case msg := <-r.m1bChan:
				dlog.Printf("Received 1b for balnum %d @ replica %d\n", msg.Balnum, r.Id)
				r.commandsMutex.Lock()
				r.handle1b(msg)
				r.commandsMutex.Unlock()
				break

			case msg := <-r.m2aChan:
				dlog.Printf("Received 2a for balnum %d @ replica %d\n", msg.Balnum, r.Id)
				r.commandsMutex.Lock()
				r.handle2a(msg)
				r.commandsMutex.Unlock()
				break

			case msg := <-r.m2bChan:
				dlog.Printf("Received 2b for balnum %d @ replica %d\n", msg.Balnum, r.Id)
				r.commandsMutex.Lock()
				r.handle2b(msg)
				r.commandsMutex.Unlock()
				break

			case <-clockChan:
				//way out of deadlock

				select {

				case propose := <-r.ProposeChan:
					//got a Propose from a client
					dlog.Printf("Proposal with id %d @ replica %d\n", propose.CommandId, r.Id)
					r.commandsMutex.Lock()
					r.handlePropose(propose)
					r.commandsMutex.Unlock()
					break

				default:
					break
				}
			}

		} else {

			select {

			case prepare := <-r.prepareChan:
				//got a Prepare message
				dlog.Printf("Received Prepare for balnum %d\n", prepare.Balnum)
//.........这里部分代码省略.........