// This is the main function of the propose part. The whole process of a
// successful propose consists of prepare, accept and commit. It will return
// once receiving Reject reply.
func (pn *paxosNode) Propose(args *paxosrpc.ProposeArgs, reply *paxosrpc.ProposeReply) error {
	// build arguments
	timeStamp := time.After(time.Second * 15)
	prepareArgs := &paxosrpc.PrepareArgs{args.Key, args.N}
	acceptArgs := &paxosrpc.AcceptArgs{Key: args.Key, N: args.N, V: args.V}
	commitArgs := &paxosrpc.CommitArgs{Key: args.Key}

	// prepare
	go pn.DoPrepare(prepareArgs)

	/* wait for the result of prepare, accept and commit
	 * of one failure, then the proposal failure
	 */
	for {
		select {
		case prepareData := <-pn.prepareDoneChan:
			if prepareData.status {
				if prepareData.va != nil {
					acceptArgs.V = prepareData.va
				}
				go pn.DoAccept(acceptArgs)
			} else {
				return nil
			}
		case status := <-pn.acceptDoneChan:
			if status {
				commitArgs.V = acceptArgs.V
				go pn.DoCommit(commitArgs)
			} else {
				return nil
			}
		case status := <-pn.commitDoneChan:
			if status {
				reply.V = acceptArgs.V
			}
			return nil
		case <-timeStamp:
			return errors.New("Time out error")
		}
	}
}

func (pn *paxosNode) Propose(args *paxosrpc.ProposeArgs, reply *paxosrpc.ProposeReply) error {
	doneChan := make(chan interface{})
	// Everything in a goroutine for timeout checking
	go func() {
		// Setup for prepare phase
		pArgs := &paxosrpc.PrepareArgs{Key: args.Key, N: args.N}
		var client *rpc.Client
		acceptChan := make(chan Na_va, pn.numNodes)

		// Send out prepare messages
		for i := 0; i < pn.numNodes; i++ {
			client = pn.allNodes[i]
			go sendProposal(pn, client, pArgs, acceptChan)
		}

		// Receive prepare messages
		highestna := 0
		var highestva interface{}
		for j := 0; j <= pn.numNodes/2; j++ {
			nava := <-acceptChan
			if nava.na >= highestna {
				highestna = nava.na
				highestva = nava.va
			}

		}

		// Determine if a higher proposal number was seen and adjust value
		ourValue := args.V

		if highestna > 0 {
			ourValue = highestva
		}

		// Setup accept phase with possibly new value
		aArgs := &paxosrpc.AcceptArgs{Key: args.Key, N: args.N, V: ourValue}
		replies := make(chan int, pn.numNodes)

		// Send out accept messages
		for i := 0; i < pn.numNodes; i++ {
			client = pn.allNodes[i]
			go sendAccept(pn, client, replies, aArgs)
		}

		// Receive accept messages
		for k := 0; k <= pn.numNodes/2; k++ {
			_ = <-replies
		}

		// Send commit messages
		cArgs := &paxosrpc.CommitArgs{Key: args.Key, V: ourValue}
		for i := 0; i < pn.numNodes; i++ {
			sendCommit(pn, i, cArgs)
		}
		// Notify main routine that we finished and commited ourValue
		doneChan <- ourValue
	}()

	// Run the main function and wait for up to 15 seconds
	select {
	case r := <-doneChan:
		// Propse completed successfully
		reply.V = r
		return nil
	case <-time.After(15 * time.Second):
		return nil
	}
}

func (pn *paxosNode) Propose(args *paxosrpc.ProposeArgs, reply *paxosrpc.ProposeReply) error {
	timeOutChan := time.After(time.Duration(15) * time.Second)
	proposeSuccess := make(signalChan, pn.numNodes)
	acceptSuccess := make(signalChan, pn.numNodes)

	// Set arguments of proposal
	proposalArgs := &LeadProposalArgs{}
	proposalArgs.value = args.V
	proposalArgs.highestNum = 0
	proposalArgs.lock = new(sync.Mutex)

	// Send prepare message to each node
	pn.connList.connLock.Lock()
	for _, conn := range pn.connList.conn {
		client := conn
		go func() {
			prepareArgs := &paxosrpc.PrepareArgs{Key: args.Key, N: args.N}
			var prepare_reply paxosrpc.PrepareReply
			client.Call("PaxosNode.RecvPrepare", prepareArgs, &prepare_reply)
			if prepare_reply.Status == paxosrpc.OK {
				proposeSuccess <- struct{}{}
				// Pick the accepted value with highest seq num
				proposalArgs.lock.Lock()
				if prepare_reply.V_a != nil && prepare_reply.N_a > proposalArgs.highestNum {
					// fmt.Printf("[PROPOSE] [ID %d] N_a: %d, V_a %d\n", pn.srvID, prepare_reply.N_a, prepare_reply.V_a)
					proposalArgs.value = prepare_reply.V_a
					proposalArgs.highestNum = prepare_reply.N_a
				}
				proposalArgs.lock.Unlock()
			}
		}()
	}
	pn.connList.connLock.Unlock()

	// Count the number of accept messages
	// If the number exceeds the half of the number of nodes
	// Move to ACCEPT Phase
	propose_counter := 0
Propose:
	for {
		select {
		case <-timeOutChan:
			fmt.Println("[Timeout]")
			reply.V = nil
			return nil
		case <-proposeSuccess:
			propose_counter += 1
			if propose_counter > pn.majority_num {
				break Propose
			}
		}
	}

	// Send ACCEPT message to each node
	pn.connList.connLock.Lock()
	for _, conn := range pn.connList.conn {
		client := conn
		go func() {
			accept_args := &paxosrpc.AcceptArgs{Key: args.Key, N: args.N, V: proposalArgs.value}
			var acc_reply paxosrpc.AcceptReply
			client.Call("PaxosNode.RecvAccept", accept_args, &acc_reply)
			if acc_reply.Status == paxosrpc.OK {
				acceptSuccess <- struct{}{}
			}
		}()
	}
	pn.connList.connLock.Unlock()

	// Count the number of accept messages
	// If the number exceeds the half of the number of nodes
	// Move to COMMIT Phase
	accept_counter := 0
Accept:
	for {
		select {
		case <-timeOutChan:
			fmt.Println("[Timeout]")
			reply.V = nil
			return nil
		case <-acceptSuccess:
			accept_counter += 1
			if accept_counter > pn.majority_num {
				break Accept
			}
		}
	}

	// Send COMMIT message to each node
	pn.connList.connLock.Lock()
	for _, conn := range pn.connList.conn {
		client := conn
		replyChan := make(chan bool)
		go func() {
			commit_args := &paxosrpc.CommitArgs{Key: args.Key, V: proposalArgs.value}
			var commit_reply paxosrpc.CommitReply
			client.Call("PaxosNode.RecvCommit", commit_args, &commit_reply)
			replyChan <- true
		}()
		select {
		case <-replyChan:
//.........这里部分代码省略.........

func (pn *paxosNode) Propose(args *paxosrpc.ProposeArgs, reply *paxosrpc.ProposeReply) error {
	timeoutChan := make(chan int, 100)
	go asyncTimeout(&timeoutChan)

	key := args.Key
	N := args.N
	V := args.V

	fmt.Println("\nPROPOSE:", key, N, V)

	// cant end until a commit has been made
	//pn.commitMutexes[key] = &sync.Mutex{}
	//pn.commitMutexes[key].Lock()

	// PREPARE //
	fmt.Println("PREPARE")
	// ask each node in the ring to prepare
	prepArgs := paxosrpc.PrepareArgs{key, N}
	prep_good_replies := make(chan *proposal, pn.numNodes)
	prep_bad_replies := make(chan int, pn.numNodes)
	for i, cli := range pn.nodes {
		fmt.Println("asyncp id, i, key, N:", pn.id, i, prepArgs.Key, prepArgs.N)
		go asyncCallRecvPrepare(cli,
			&prepArgs,
			&prep_good_replies,
			&prep_bad_replies)
	}
	// collect replies from nodes
	fmt.Println("Waiting for Prepare replies")
	max_n := -1
	good_count := 0
L:
	for good_count <= pn.numNodes/2+1 {
		select {
		case p := <-prep_good_replies:
			fmt.Println("Received good prepare-reply")
			N_a := p.N
			V_a := p.V
			if N_a != -1 && max_n < N_a {
				max_n = N_a
				V = V_a
			}
			good_count += 1
		case <-prep_bad_replies:
			fmt.Println("Received bad prepare-reply")
			break
		case <-timeoutChan:
			fmt.Println("timed out")
			break L
		}
	}
	// the proposing node fails to be the leader
	// (this code should be blocking until a commit is made, and then return that value)
	if good_count <= (pn.numNodes/2 + 1) {
		//	pn.commitMutexes[key].Lock()
		reply.V = pn.storage[key]
		//	pn.commitMutexes[key].Unlock()
	}
	////////////////////////////////////////////////////////////////////////////////////////

	// ACCEPT //
	fmt.Println("ACCEPT")
	// ask each node in the ring to accept
	acceptArgs := paxosrpc.AcceptArgs{key, N, V}
	acc_good_replies := make(chan int, pn.numNodes)
	acc_bad_replies := make(chan int, pn.numNodes)
	for i, cli := range pn.nodes {
		fmt.Println("asynca id, i, key, N, V:", pn.id, i, acceptArgs.Key, acceptArgs.N, acceptArgs.V)
		go asyncCallRecvAccept(cli,
			&acceptArgs,
			&acc_good_replies,
			&acc_bad_replies)
	}
	// collect replies from nodes
	good_count = 0
L1:
	for good_count <= (pn.numNodes/2 + 1) {
		select {
		case <-acc_good_replies:
			good_count += 1
			fmt.Println("Received good accept-reply")
		case <-acc_bad_replies:
			fmt.Println("Received bad accept-reply")
			break
		case <-timeoutChan:
			break L1
		}
	}
	// the proposing node fails to be the leader
	// (this code should be blocking until a commit is made, and then reply with that value)
	if good_count <= (pn.numNodes/2 + 1) {
		//	pn.commitMutexes[key].Lock()
		reply.V = pn.storage[key]
		//	pn.commitMutexes[key].Unlock()
		return nil
	}
	////////////////////////////////////////////////////////////////////////////////////////

	// COMMIT //
	fmt.Println("COMMIT")
//.........这里部分代码省略.........

// Propose initializes proposing a value for a key, and replies with the
// value that was committed for that key. Propose should not return until
// a value has been committed, or 15 seconds have passed. Returns an error if 15 seconds have passed
func (pn *paxosNode) Propose(args *paxosrpc.ProposeArgs, reply *paxosrpc.ProposeReply) error {
	key := args.Key
	n_proposal := args.N
	v := args.V
	pkd := pn.getInstance(key)
	//we lock on this key on this node because we only have one instance of paxos per key at one time
	pkd.proposeLock.Lock()
	defer pkd.proposeLock.Unlock()
	//PROPOSE PHASE
	// A node decides to be leader (and propose)
	// Leader chooses Myn > Nh
	// Leader sends <prepare, Myn> to all nodes
	// ask for majority here
	// should make it parallel, but here we use a loop
	// suppose there are not so much paxosNode in the test program
	vote := 0
	chanRet := make(chan bool)
	timeoutTime := time.Now().UnixNano() + DLPropose

	go func() {
		for {
			if timeoutTime < time.Now().UnixNano() {
				//fmt.println("BYE BYE")
				close(chanRet)
				return
			}
			time.Sleep(100 * time.Millisecond)
		}
	}()
	var wg sync.WaitGroup
	v_prime := v
	highest_n_a := -1
	for _, conn := range pn.ss {
		wg.Add(1)
		myargs := &paxosrpc.PrepareArgs{
			Key: key,
			N:   n_proposal,
		}
		//we call RecvPrepare on each server asynchronously
		go func(conn *rpc.Client) {
			defer wg.Done()

			chanTime := make(chan bool)
			chanRPC := make(chan bool)
			var myreply paxosrpc.PrepareReply

			// actual query
			go func() {
				conn.Call("PaxosNode.RecvPrepare", myargs, &myreply)
				if myreply.Status == paxosrpc.OK {
					vote++
					if myreply.N_a > highest_n_a {
						highest_n_a = myreply.N_a
						v_prime = myreply.V_a
					}

				}
				chanRPC <- true
			}()
			expiry := time.Now().UnixNano() + DL
			// deadline of DL seconds for RPC to return
			go func() {
				for {
					if expiry < time.Now().UnixNano() {
						chanTime <- true
						return
					}
					time.Sleep(100 * time.Millisecond)
				}
			}()
			//revoking using channels
			select {
			case <-chanRPC:
				// rpc returned
				return
			case <-chanTime:
				// RecvPrepare timeout
				return
			case <-chanRet:
				// Revoking because of 15 second timeout of propose

				return

			}
		}(conn)

	}
	// use of wait to ensure all the RPC calls to recvPrepare have either timed out or returned
	wg.Wait()
	pxi := pn.getInstance(key)

	// we hsould now update the highest_n_a for this paxos instance
	pxi.mu.Lock()
	if highest_n_a > pxi.Nh {
		pxi.Nh = highest_n_a

	}
//.........这里部分代码省略.........

func (pn *paxosNode) Propose(args *paxosrpc.ProposeArgs, reply *paxosrpc.ProposeReply) error {
	fmt.Printf("[node %d enter propose]\n", pn.id)
	endTime := time.Now().Add(time.Duration(15) * time.Second)
	timeoutChan := make(chan bool)
	go pn.timing(args.Key, endTime, timeoutChan)

	pn.nodeMutex.Lock()
	if _, ok := pn.keyMutex[args.Key]; !ok {
		pn.keyMutex[args.Key] = &sync.Mutex{}
	}
	pn.nodeMutex.Unlock()

	pn.keyMutex[args.Key].Lock()
	defer pn.keyMutex[args.Key].Unlock()

	// prepare: send <prepare, myn> to all the nodes
	fmt.Printf("----------------------- ndoe %d prepare value %d n=%d\n", pn.id, args.V, args.N)
	pRequestChan := make(chan voteRequest)
	pResponseChan := make(chan voteResponse)
	go pn.voter(args.Key, pRequestChan, pResponseChan, timeoutChan)
	for _, node := range pn.idMap {
		prepareArgs := &paxosrpc.PrepareArgs{args.Key, args.N}
		var prepareReply paxosrpc.PrepareReply
		go pn.prepare(node.client, prepareArgs, &prepareReply, pRequestChan)
	}

	voteRes := <-pResponseChan
	var value interface{}
	n := args.N
	// if not recieve majority prepare-ok from the paxos nodes, terminate propose
	if !voteRes.pass {
		fmt.Println("not recieve prepare-ok from a majority")
		return nil
	} else if voteRes.n_a > args.N && voteRes.v_a != nil {
		fmt.Printf("@@@@@@@@@@@@@@@@@@ args.N=%d\n", args.N)
		fmt.Printf("value is set to %d\n", voteRes.v_a)
		value = voteRes.v_a
		n = voteRes.n_a
	} else {
		value = args.V
	}

	// accept: send <accept, myn, V> to all the nodes
	fmt.Printf("----------------------- node %d accept value %d n=%d\n", pn.id, value, n)
	aRequestChan := make(chan voteRequest)
	aResponseChan := make(chan voteResponse)
	go pn.voter(args.Key, aRequestChan, aResponseChan, timeoutChan)
	for _, node := range pn.idMap {
		acceptArgs := &paxosrpc.AcceptArgs{args.Key, n, value}
		var acceptReply paxosrpc.AcceptReply
		go pn.accept(node.client, acceptArgs, &acceptReply, aRequestChan)
	}
	voteRes = <-aResponseChan
	if !voteRes.pass {
		fmt.Println("not recieve accept-ok from a majority")
		return nil
	}

	// commit: send <commit, va> to all the nodes
	fmt.Printf("----------------------- node %d commit value %d n=%d\n", pn.id, value, n)
	cRequestChan := make(chan voteRequest)
	cResponseChan := make(chan voteResponse)
	go pn.voter(args.Key, cRequestChan, cResponseChan, timeoutChan)
	for _, node := range pn.idMap {
		commitArgs := &paxosrpc.CommitArgs{args.Key, value}
		var commitReply paxosrpc.CommitReply
		go pn.commit(node.client, commitArgs, &commitReply, cRequestChan)
	}
	<-cResponseChan
	reply.V = value
	return nil
}

func (pn *paxosNode) Propose(args *paxosrpc.ProposeArgs, reply *paxosrpc.ProposeReply) error {
	// time.Sleep(time.Duration(rand.Int()%100) * time.Millisecond)
	fmt.Println("propose by node", pn.nodeID)
	defer fmt.Println("propose by node return", pn.nodeID)

	doneCh := make(chan int, 1)
	// fmt.Println("Propose begins, numNodes:", pn.numNodes, args.Key, pn.nodeID)
	go func(doneCh chan int) {
		// for {
		pn.mutex.Lock()
		if pn.keyValueMutex[args.Key] == nil {
			pn.keyValueMutex[args.Key] = &sync.Mutex{}
		}
		pn.mutex.Unlock()
		// keyValueMutex: make(map[string]*sync.Mutex),
		pn.keyValueMutex[args.Key].Lock()
		tempValue, ok := pn.keyValue[args.Key]
		if !ok {
			pn.keyValue[args.Key] = &value{
				Key:   args.Key,
				Value: args.V,
				N_a:   args.N,
				N_h:   args.N,
				my_n:  args.N,
			}
		}
		myProposalNumber := pn.keyValue[args.Key].N_h
		myProposalNumber = (myProposalNumber+pn.numNodes-pn.nodeID)/pn.numNodes*pn.numNodes + pn.nodeID
		tempValue = pn.keyValue[args.Key]
		tempValue.N_h = myProposalNumber
		pn.keyValue[args.Key] = tempValue
		pn.keyValueMutex[args.Key].Unlock()

		promiseCount := 1
		if pn.savedkeyValue[args.Key] == nil {
			pn.savedkeyValue[args.Key] = &value{
				Key:   args.Key,
				Value: args.V,
				N_a:   args.N,
				N_h:   args.N,
				my_n:  args.N,
			}
		}
		i := 0
		var temp int
		retChan := make(chan int, 100)
		// fmt.Println("Prepare phase", myProposalNumber)
		args.N = myProposalNumber
		for nodeID, hostPort := range pn.hostMap {
			go pn.ProposeByNode(nodeID, hostPort, args, retChan)
			i++
			// fmt.Println("Prepare phase,Sending", myProposalNumber, "nodeid", nodeID)

			// temp = <-retChan
			// promiseCount = promiseCount + temp
		}
		// fmt.Println("Prepare phase,Sending done", myProposalNumber)

		for ; i > 0; i-- {
			temp = <-retChan
			promiseCount = promiseCount + temp
			// fmt.Println("Prepare phase,counting", myProposalNumber)

		}
		// fmt.Println("Prepare phase,receive channel done", myProposalNumber)

		if promiseCount < pn.numNodes/2+1 {
			// Not get majority vote
			// time.Sleep(time.Duration(rand.Int()%100) * time.Millisecond)
			// fmt.Println("Prepare phase,reject", myProposalNumber)
			time.Sleep(time.Second)
			doneCh <- 1
			return
		}

		// Accept phase
		// fmt.Println("Accept phase", myProposalNumber)
		i = 0
		acceptCount := 1
		retChan = make(chan int, 100)
		for nodeID, hostPort := range pn.hostMap {
			go pn.AcceptByNode(nodeID, hostPort, args, retChan)
			i++
			// temp := <-retChan
			// acceptCount = acceptCount + temp
		}
		// fmt.Println("Accept phase, sending loop done", myProposalNumber)

		for ; i > 0; i-- {
			temp := <-retChan
			acceptCount = acceptCount + temp
		}
		// fmt.Println("Accept phase, receive channel done", myProposalNumber)

		if acceptCount < pn.numNodes/2+1 {
			// Not get majority vote
			// fmt.Println("Accept phase, reject", myProposalNumber)
			time.Sleep(time.Second)
			doneCh <- 1
			return
//.........这里部分代码省略.........