func (kv *KVPaxos) paxosAgreement(op Op) bool {
	//initiates a paxos agreement for this operation
	glog.Infoln(kv.me, "Initiating Paxos agreement for ", op)
	//Start with a sequence number and increment it if it fails
	//if agreement reached check if it was really for the value I suggested or not

	agreementReached := false
	index := kv.px.Max() + 1

	to := 100 * time.Millisecond
	for !agreementReached {
		glog.Infoln(kv.me, "[paxosAgreement]Index to try for paxos agreement", index)
		kv.px.Start(index, op)
		time.Sleep(to)
		agreementReached = kv.waitForPaxosAgreement(index, op)
		if agreementReached {
			//agreement has been reached interpret value from last time around keeping into account xid
			kv.process(index)
			break
		} else {
			if to < 10*time.Second {
				to = to + 50
			}
			glog.Infoln(kv.me, " SPINNING, no decision for seq number ", index, " trying higher number")
			index++
		}
		// if agreement reached then act on it, evaluate state uptill last
		//wait till status turns to agreed
	}
	return false
}

//
// shared by Put and Append.
//
func (ck *Clerk) PutAppend(key string, value string, op string) {
	ck.mu.Lock()
	defer ck.mu.Unlock()

	xid := nrand()
	putReq := &PutAppendArgs{}
	putRep := &PutAppendReply{}
	putReq.Xid = xid
	putReq.Value = value
	putReq.Op = op
	putReq.Key = key
	tries := 0

	for {
		time.Sleep(100 * time.Millisecond)
		tries = tries + 1
		glog.Infoln(" Loop count", tries)
		for index, _ := range ck.servers {
			glog.Infoln(ck.uuid, "PutAppend Calling server ", index, " for request ", putReq.Op)
			ret := call(ck.servers[index], "KVPaxos.PutAppend", putReq, putRep)
			if ret {
				//we got a response check if invalid response in which case try again
				if putRep.Err != OK {
					continue
				}
				return
			} else {
				glog.Warningln(ck.uuid, "RPC broke, retrying for a different server ", ck.servers)
				continue
			}
		}
	}

}

func (kv *KVPaxos) waitForPaxosAgreement(seq int, op Op) bool {

	status, val := kv.px.Status(seq)

	if status == paxos.Decided {
		//check if value was truly what we sent originally or some morphed value
		// check if val was what was trying to be inserted

		if val.(Op).Xid == op.Xid {
			glog.Infoln(kv.me, "===>> Agreement reached at seq with value", seq, " with value:", val)
			return true
		} else {
			glog.Infoln(kv.me, "++++Agreement reached for seq by other Server#:", val.(Op).Server, " for seq ", seq, " processed till now:", kv.toprocess, " Min/Max", kv.px.Min(), "/", kv.px.Max())
			return false
		}
	}
	if status == paxos.Forgotten {
		glog.Infoln("WARNING...!!!")
		//this has already been applied by the application cannot agree on this
		return false

	}
	//in pending we wait
	return false
}

//--------------------------------------------------
func (px *Paxos) accept(seq int, state *State) bool {
	//send accept message with state in
	acceptReq := &AcceptReq{}
	acceptRep := &AcceptReply{}
	acceptReq.Proposal = state.proposal
	acceptReq.V = state.v
	acceptReq.Seq = seq
	numAccepts := 0

	glog.Infoln(px.me, "[Accept] hadnler with seq", seq)
	for indx, server := range px.peers {
		ret := false
		if indx == px.me {
			ret = (px.AcceptHandler(acceptReq, acceptRep) == nil)
		} else {
			ret = call(server, "Paxos.AcceptHandler", acceptReq, acceptRep)
		}
		//update highestProposalInReply and value in highestProposaled Reply
		if ret {
			//RPC or method call successul
			if acceptRep.Status == Accept {
				numAccepts++
			}
		}
	}
	if 2*numAccepts > len(px.peers) {
		glog.Infoln(px.me, "[Accept] majority accepted proposal decision reached will send decide soon ", seq)
		//do not update internal state yet
		return true
	} else {
		glog.Infoln(px.me, "[Accept]Peers rejected accept proposal having #", state.proposal, " will go to next round for slot#", seq)
	}
	return false
}

//
// the application on this machine is done with
// all instances <= seq.
//
// see the comments for Min() for more explanation.
//
func (px *Paxos) Done(seq int) {
	glog.Infoln(px.me, "[Done] called with ", seq)
	px.mu.Lock()
	if seq > px.lastDoneSignalled {
		px.lastDoneSignalled = seq
		glog.Infoln(px.me, "[Done] updating done#", seq, " value of done ", px.lastDoneSignalled)
	}

	//clear everything below maxCanDisregard
	px.mu.Unlock()

}

func (px *Paxos) PromiseHandler(req *PrepareReq, rep *PrepareRep) error {
	//handles promise messages following is the pseudo code
	//if this is itself the proposer for same instance and incoming seq numner is highr it should back off
	//if this proposal is encountered for the first time for instance it should be accepted and acked yes
	px.mu.Lock()
	rep.LastDoneSignalled = px.lastDoneSignalled
	if val, ok := px.stateMap[req.Seq]; ok {
		//we have already seen this paxos seq but we might have a different proposal number for it
		if val.proposal > req.Proposal {
			//reject this proposal
			//	glog.Infoln(px.me, "[PromisHandler] Proposal #", req.Proposal, " rejected since own proposalNum#", val.proposal, " was higher")
			rep.Status = Reject
			rep.V = nil
			//set proposal number in return msg
			rep.Proposal = -1
			px.mu.Unlock()
			return nil
		} else {
			//make sure I have not already decided
			//proposal will be accepted and value accepted returned if not nil
			rep.Status = Accept
			rep.ProposalAccepted = val.highestproposalaccepted
			rep.V = val.v
			//return back locally accepted value if any
			//update highest proposal seen
			val.proposal = req.Proposal
			//	glog.Infoln(px.me, "[PromisHandler] Proposal #", req.Proposal, " accepted, since", "incoming propsal is higher, printing v", val, " sending rep", rep)
			glog.Infoln(px.me, "[PromiseHandler] printing my state map", px.stateMap)
			px.stateMap[req.Seq] = val
			px.mu.Unlock()
			return nil
		}

	} else {
		//no record exists for this paxos seq
		//proposal will be accepted and value accepted returned if not nil
		val := State{}
		rep.Status = Accept
		//return back locally accepted value if any
		rep.V = nil
		//update highest proposal seen
		val.proposal = req.Proposal
		val.highestproposalaccepted = -1
		rep.Proposal = -1
		rep.ProposalAccepted = -1
		val.decision = Pending
		px.stateMap[req.Seq] = val
		glog.Infoln(px.me, "[PromisHandler] Proposal #", req.Proposal, " accepted")
		px.mu.Unlock()
		return nil
	}

}

func (kv *KVPaxos) Get(args *GetArgs, reply *GetReply) error {
	glog.Infoln(kv.me, "[GET]Got client request -->", args)
	//this server is a client for the paxos library
	//initiate a paxos agreement for this operation
	op := Op{}
	op.Optype = Get
	op.Xid = args.Xid
	op.Key = args.Key
	op.Server = kv.me
	kv.paxosAgreement(op)
	reply.Value = kv.kvData[args.Key]
	reply.Err = OK
	glog.Infoln(kv.me, "[Get] returned ", reply.Value)
	return nil
}

//
// Min() should return one more than the minimum among z_i,
// where z_i is the highest number ever passed
// to Done() on peer i. A peers z_i is -1 if it has
// never called Done().
//
// Paxos is required to have forgotten all information
// about any instances it knows that are < Min().
// The point is to free up memory in long-running
// Paxos-based servers.
//
// Paxos peers need to exchange their highest Done()
// arguments in order to implement Min(). These
// exchanges can be piggybacked on ordinary Paxos
// agreement protocol messages, so it is OK if one
// peers Min does not reflect another Peers Done()
// until after the next instance is agreed to.
//
// The fact that Min() is defined as a minimum over
// *all* Paxos peers means that Min() cannot increase until
// all peers have been heard from. So if a peer is dead
// or unreachable, other peers Min()s will not increase
// even if all reachable peers call Done. The reason for
// this is that when the unreachable peer comes back to
// life, it will need to catch up on instances that it
// missed -- the other peers therefor cannot forget these
// instances.
//
func (px *Paxos) Min() int {
	min := 100000
	lastDoneReq := &LastDoneMsg{}
	lastDoneRep := &LastDoneReply{}
	for indx, server := range px.peers {
		if indx == px.me {
			px.LastDoneHandler(lastDoneReq, lastDoneRep)
		} else {
			call(server, "Paxos.LastDoneHandler", lastDoneReq, lastDoneRep)
		}
		if lastDoneRep.Done < min {
			min = lastDoneRep.Done
			glog.Infoln(px.me, "Updating min to ", min)
		}
		//update highestProposalInReply and value in highestProposaled Reply
		//we don't care about return code here we just update intrnal state
	}
	px.mu.Lock()
	if min > px.maxCanDisregard {

		px.maxCanDisregard = min
	}
	px.mu.Unlock()
	return min + 1
}

func main() {

	role := flag.String("role", "master", "Escort role")
	flag.Set("alsologtostderr", "true")
	flag.Set("v", "5")
	flag.Parse()

	if role == nil || len(*role) == 0 {
		Usage()
		return
	}

	glog.Infoln(*role)

	switch *role {
	case "master":
		// 主服务
		server.Work()
	case "slave":
		// 备份服务
		client.Work()
	default:
		Usage()
	}
}

func connectServer() {
	//接通
	conn, err := net.Dial("tcp", "localhost:8260")
	defer func() {
		conn.Close()
	}()

	if err != nil {
		glog.Errorln("Escort slave can't connect to master!")
		netErrCh <- 1
		//return new(NetConnectErr)
	} else {
		glog.Infoln("Escort slave connected successfully!")
		// 定时发送ping
		go func() {
			timer := time.NewTicker(10 * time.Second)
			for {
				select {
				case <-timer.C:
					if err := ping(conn); err != nil {
						glog.Errorf("[ERROR] Escort slave send ping err: %s", err.Error())
						// ping发生错误，切换
						doSwitch()
					}
				}
			}
		}()

		handler(conn)
	}
}

func ping(conn net.Conn) error {
	select {
	case pingAtCh <- 1:
		glog.Infoln("Escort slave send ping!")
	default:
		// 没有收到上一次ping的pong，超时
		glog.Warningln("Escort slave doesn't get pong!")
		return new(PingTimeOutErr)
	}

	pingcmd := new(Packet)
	pingcmd.Version = V1
	pingcmd.Flags = FLAG_REQUEST
	pingcmd.Length = 0
	pingcmd.Sequence = 0

	data, err := pingcmd.Encode()
	_, err = conn.Write(data)
	if err != nil {
		glog.Errorf("[ERROR] %s", err.Error())
		return err
	}

	return nil
}

func (kv *KVPaxos) PutAppend(args *PutAppendArgs, reply *PutAppendReply) error {

	glog.Infoln(kv.me, "[PUTAPPEND]Got client request -->", args)
	op := Op{}
	if args.Op == "Put" {
		op.Optype = Put
	} else {
		op.Optype = Append
	}
	op.Xid = args.Xid
	op.Key = args.Key
	op.Value = args.Value
	op.Server = kv.me
	kv.paxosAgreement(op)
	reply.Err = OK
	glog.Infoln(kv.me, "[Put] returned")
	return nil
}

func (kv *KVPaxos) process(indexOfAgreement int) {
	//processing value in op
	// first time we are seeing this xid need to process this
	glog.Infoln(kv.me, "In process with index of agreement", indexOfAgreement, "with current kvtprocess", kv.toprocess)
	kv.applyBackLogOps(indexOfAgreement) //these are both inclusive
	//only signal done till applyBackLogOps
	//increment kv.toprocess
	//finally call done here signalling we are done processing this

}

func doService(conn net.Conn) {
	glog.Infoln("Escort server is connected!")
	defer func() {
		conn.Close()
	}()

	for {
		buf, err := ReadFromConn(conn)
		if err != nil {
			glog.Errorf("[ERROR] %s", err.Error())
			break
		}

		p := new(Packet)
		p.Decode(buf)

		if p.Flags == FLAG_REQUEST && p.Length == 0 { // ping命令
			glog.Infoln("Escort server get client ping cmd!")

			pong := new(Packet)
			pong.Version = V1
			pong.Flags = FLAG_RESPONSE
			pong.Sequence = 0
			pong.Length = 0

			var data []byte
			var err error
			data, err = pong.Encode()
			if err != nil {
				glog.Errorf("[ERROR] %s", err.Error())
				continue
			}
			_, err = conn.Write(data)
			if err == nil {
				glog.Infoln("Escort server send pong successfully!")
			} else {
				glog.Errorf("[ERROR] %s", err.Error())
			}
		}
	}

}

//
// the application wants to know the
// highest instance sequence known to
// this peer.
//
func (px *Paxos) Max() int {
	max := 0
	px.mu.Lock()
	for key, _ := range px.stateMap {
		if key > max {
			max = key
		}

	}
	px.mu.Unlock()
	glog.Infoln("[Max] called with value ", max)
	return max
}

func (px *Paxos) AcceptHandler(req *AcceptReq, rep *AcceptReply) error {
	//accept request handler will need to toggle global state
	px.mu.Lock()
	instance := px.stateMap[req.Seq]
	if instance.proposal > req.Proposal {
		//send reject
		rep.Status = Reject
		glog.Infoln(px.me, "[ACCEPTHANDLEr]Peer rejected accept message with proposal#", req.Proposal, "since it has seen higher proposal ", instance.proposal)
	} else {
		//accept this request covers equality case
		rep.Status = Accept
		instance.highestproposalaccepted = req.Proposal
		instance.v = req.V
		//this should not be necessary here since accept messages should have same proposal number
		instance.proposal = req.Proposal
		instance.decision = Pending
		px.stateMap[req.Seq] = instance
		glog.Infoln(px.me, " [AcceptHandler]Peer accepted accept message with proposal#", req.Proposal, "and value ", req.V, " for seq", req.Seq)

	}
	px.mu.Unlock()
	return nil
}

//
// fetch the current value for a key.
// returns "" if the key does not exist.
// keeps trying forever in the face of all other errors.
//
func (ck *Clerk) Get(key string) string {
	ck.mu.Lock()
	defer ck.mu.Unlock()
	//generate a unique xid
	xid := nrand()
	retVal := ""
	//contact every server for the operation in round robin fashion break if you got a response else move on
	getReq := &GetArgs{}
	getReq.Xid = xid
	getReq.Key = key
	getRep := &GetReply{}
	tries := 0

	for {
		tries = 1 + tries
		glog.Infoln(ck.uuid, " client trying for ", tries)
		time.Sleep(100 * time.Millisecond)
		index := 0
		for index < len(ck.servers) {
			glog.Infoln(ck.uuid, "Get Calling server ", index, " for request ", ck.servers)
			ret := call(ck.servers[index], "KVPaxos.Get", getReq, getRep)
			if ret {
				//we got a response check if invalid response in which case try again
				if getRep.Err != OK {
					index++
					continue
				}
				retVal = getRep.Value
				return retVal
			} else {
				glog.Infoln(ck.uuid, "RPC broke retrying get request with another server", ck.servers)
				index++
			}
		}
	}
	return retVal
}

//--------------------------------------------------
func (px *Paxos) decide(seq int, state *State) bool {
	//decide handler updates self state and sends decide message to peers
	//proposal number is inconsequential here, we can possibly clean state here
	decideReq := &DecideReq{}
	decideRep := &DecideReply{}
	decideReq.V = state.v
	decideReq.Seq = seq

	glog.Infoln(px.me, "[Decide] handler with seq", seq)
	for indx, server := range px.peers {
		if indx == px.me {
			px.DecideHandler(decideReq, decideRep)
		} else {
			ret := call(server, "Paxos.DecideHandler", decideReq, decideRep)
			if !ret {
				glog.Infoln(px.me, "Decide message failed for index", indx)
			}
		}
		//update highestProposalInReply and value in highestProposaled Reply
		//we don't care about return code here we just update intrnal state
	}

	return true
}

func (px *Paxos) DecideHandler(req *DecideReq, rep *DecideReply) error {
	//updates internal state here
	glog.Infoln(px.me, "[DecideHandler] decision reached with value ", req.V)
	px.mu.Lock()

	if instance, ok := px.stateMap[req.Seq]; ok {
		//instance has to be present
		instance.decision = Decided
		instance.v = req.V
		px.stateMap[req.Seq] = instance
		glog.Infoln(px.me, "[DecideHandler]Priting my stateMap", px.stateMap, " for seq", req.Seq)

	} else {
		instance := State{}
		instance.decision = Decided
		instance.v = req.V
		px.stateMap[req.Seq] = instance
		glog.Infoln(px.me, "[DecideHandler] WARNING!! decide message but no state reached, possible crash recovery")

	}

	px.mu.Unlock()
	return nil
}

//
// call() sends an RPC to the rpcname handler on server srv
// with arguments args, waits for the reply, and leaves the
// reply in reply. the reply argument should be a pointer
// to a reply structure.
//
// the return value is true if the server responded, and false
// if call() was not able to contact the server. in particular,
// the reply's contents are only valid if call() returned true.
//
//
//
func call(srv string, rpcname string,
	args interface{}, reply interface{}) bool {
	c, errx := rpc.Dial("unix", srv)
	if errx != nil {
		return false
	}
	defer c.Close()

	err := c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	glog.Infoln(err)
	return false
}