package participant

import (
	"github.com/nusdbsystem/PRAC/constants"
	"github.com/nusdbsystem/PRAC/remote"
	"github.com/nusdbsystem/PRAC/rlsm"
	"github.com/nusdbsystem/PRAC/utils"
	"testing"
	"time"
)

var address = []string{"127.0.0.1:6001", "127.0.0.1:6002", "127.0.0.1:6003"}

func makeLocal() {
	utils.SetLocal()
	constants.BasicWaitTime = 20
	constants.SetR(1)
	constants.SetMsgDelay4RAC(1)
	constants.SetConcurrency(1)
	constants.SetServerTimeOut(28)
	constants.SetDown(0)
	constants.SetNF(-1)
	constants.SetMinLevel(1)
}

func ParticipantsTestKit() []*ParticipantStmt {
	stmts := make([]*ParticipantStmt, 3)
	ch := make(chan bool)

	var Args1 = []string{"*", "*", address[0], "5"}
	stmts[0] = &ParticipantStmt{}
	go begin(stmts[0], ch, Args1[2])
	<-ch

	var Args2 = []string{"*", "*", address[1], "5"}
	stmts[1] = &ParticipantStmt{}
	go begin(stmts[1], ch, Args2[2])
	<-ch

	var Args3 = []string{"*", "*", address[2], "5"}
	stmts[2] = &ParticipantStmt{}
	go begin(stmts[2], ch, Args3[2])
	<-ch

	for i := 0; i < 3; i++ {
		stmts[i].Participant.Kv.SetDDL(time.Millisecond * 200)
		for j := 0; j < 5; j++ {
			stmts[i].Participant.Kv.Update(j, j)
		}
	}
	return stmts
}

func StopServers(stmt []*ParticipantStmt) {
	for _, v := range stmt {
		v.Stop()
	}
	time.Sleep(constants.OptEps * 10)
}

func CheckPreRead(coh *ParticipantManager, txn *remote.PRACTransaction, ans []int, shardID string) {
	if len(txn.OptList) > 0 {
		txn.OptList = txn.OptList[:0]
	}
	for i := 0; i < 5; i++ {
		txn.AddRead(shardID, i)
	}
	ok, val := coh.forTestPreRead(txn)
	utils.Assert(ok && len(val) == len(ans), "PreRead Failed or With different length")
	for i := 0; i < len(ans); i++ {
		utils.Assert(ans[i] == val[utils.Hash(coh.stmt.participantID, i)], "PreRead Failed with different value")
	}
}

func TestBuildTestCases(t *testing.T) {
	makeLocal()
	stmts := ParticipantsTestKit()
	CheckVal(stmts[0].Participant, []int{0, 1, 2, 3, 4})
	CheckVal(stmts[1].Participant, []int{0, 1, 2, 3, 4})
	CheckVal(stmts[2].Participant, []int{0, 1, 2, 3, 4})
	StopServers(stmts)
}

func TestCohortsPreRead(t *testing.T) {
	makeLocal()
	stmts := ParticipantsTestKit()
	txn1 := remote.NewPRACTransaction(1, stmts[0].participantID, rlsm.CFNF, address)
	txn2 := remote.NewPRACTransaction(2, stmts[1].participantID, rlsm.CFNF, address)

	c := make(chan bool)

	go func() {
		CheckPreRead(stmts[1].Participant, txn2, []int{0, 1, 2, 3, 4}, stmts[1].participantID)
		c <- true
	}()
	go func() {
		CheckPreRead(stmts[0].Participant, txn1, []int{0, 1, 2, 3, 4}, stmts[0].participantID)
		c <- true
	}()
	<-c
	<-c
	StopServers(stmts)
}

func TestCohortsPreWrite(t *testing.T) {
	makeLocal()
	stmts := ParticipantsTestKit()
	txn1 := remote.NewPRACTransaction(1, stmts[0].participantID, rlsm.CFNF, address)
	CheckPreRead(stmts[0].Participant, txn1, []int{0, 1, 2, 3, 4}, stmts[0].participantID)

	txn1.OptList = txn1.OptList[:0]
	for i := 0; i < 5; i++ {
		txn1.AddUpdate(stmts[0].participantID, i, i+1)
	}
	stmts[0].Participant.PreWrite(txn1)
	stmts[0].Participant.Commit(txn1)
	CheckVal(stmts[0].Participant, []int{1, 2, 3, 4, 5})
	StopServers(stmts)
}

func Test2PCCommit(t *testing.T) {
	makeLocal()
	stmts := ParticipantsTestKit()
	txn1 := remote.NewPRACTransaction(1, stmts[0].participantID, rlsm.CFNF, address)
	CheckPreRead(stmts[0].Participant, txn1, []int{0, 1, 2, 3, 4}, stmts[0].participantID)
	txn2 := remote.NewPRACTransaction(1, stmts[1].participantID, rlsm.CFNF, address)
	CheckPreRead(stmts[1].Participant, txn2, []int{0, 1, 2, 3, 4}, stmts[1].participantID)
	txn1.OptList = txn1.OptList[:0]
	txn2.OptList = txn2.OptList[:0]
	for i := 0; i < 5; i++ {
		txn1.AddUpdate(stmts[0].participantID, i, i+1)
		txn2.AddUpdate(stmts[1].participantID, i, i+2)
	}

	c := make(chan bool)

	go func() {
		ok := stmts[0].Participant.PreWrite(txn1)
		c <- utils.Assert(ok, "PreWrite Made Failed")
	}()
	go func() {
		ok := stmts[1].Participant.PreWrite(txn2)
		c <- utils.Assert(ok, "PreWrite Made Failed")
	}()
	<-c
	<-c

	go func() {
		ok := stmts[0].Participant.Commit(txn1)
		c <- utils.Assert(ok, "Commit Made Failed")
	}()
	go func() {
		ok := stmts[1].Participant.Commit(txn2)
		c <- utils.Assert(ok, "Commit Made Failed")
	}()
	<-c
	<-c

	CheckPreRead(stmts[0].Participant, txn1, []int{1, 2, 3, 4, 5}, stmts[0].participantID)
	CheckVal(stmts[0].Participant, []int{1, 2, 3, 4, 5})
	CheckVal(stmts[1].Participant, []int{2, 3, 4, 5, 6})
	StopServers(stmts)
}

func Test2PCAbort(t *testing.T) {
	makeLocal()
	stmts := ParticipantsTestKit()
	txn1 := remote.NewPRACTransaction(1, stmts[0].participantID, rlsm.CFNF, address)
	CheckPreRead(stmts[0].Participant, txn1, []int{0, 1, 2, 3, 4}, stmts[0].participantID)
	txn2 := remote.NewPRACTransaction(1, stmts[1].participantID, rlsm.CFNF, address)
	CheckPreRead(stmts[1].Participant, txn2, []int{0, 1, 2, 3, 4}, stmts[1].participantID)
	txn1.OptList = txn1.OptList[:0]
	txn2.OptList = txn2.OptList[:0]
	for i := 0; i < 5; i++ {
		txn1.AddUpdate(stmts[0].participantID, i, i+1)
		txn2.AddUpdate(stmts[1].participantID, i, i+2)
	}
	stmts[0].Participant.Kv.SetDDL(0)

	ok := stmts[0].Participant.PreWrite(txn1)
	ok = ok && stmts[1].Participant.PreWrite(txn2)
	utils.Assert(!ok, "PreWrite Fail Made Failed")

	ok = stmts[0].Participant.Abort(txn1)
	ok = ok && stmts[1].Participant.Abort(txn2) 
	utils.Assert(ok, "Abort Made Failed")

	CheckPreRead(stmts[0].Participant, txn1, []int{0, 1, 2, 3, 4}, stmts[0].participantID)
	CheckPreRead(stmts[1].Participant, txn2, []int{0, 1, 2, 3, 4}, stmts[1].participantID)
	StopServers(stmts)
}
package participant

import (
	"github.com/nusdbsystem/PRAC/constants"
	"github.com/nusdbsystem/PRAC/remote"
	"github.com/nusdbsystem/PRAC/rlsm"
	"github.com/nusdbsystem/PRAC/utils"
	"testing"
	"time"
)

var address = []string{"127.0.0.1:6001", "127.0.0.1:6002", "127.0.0.1:6003"}

func makeLocal() {
	utils.SetLocal()
	constants.BasicWaitTime = 20
	constants.SetR(1)
	constants.SetMsgDelay4RAC(1)
	constants.SetConcurrency(1)
	constants.SetServerTimeOut(28)
	constants.SetDown(0)
	constants.SetNF(-1)
	constants.SetMinLevel(1)
}

func ParticipantsTestKit() []*ParticipantStmt {
	stmts := make([]*ParticipantStmt, 3)
	ch := make(chan bool)

	var Args1 = []string{"*", "*", address[0], "5"}
	stmts[0] = &ParticipantStmt{}
	go begin(stmts[0], ch, Args1[2])
	<-ch

	var Args2 = []string{"*", "*", address[1], "5"}
	stmts[1] = &ParticipantStmt{}
	go begin(stmts[1], ch, Args2[2])
	<-ch

	var Args3 = []string{"*", "*", address[2], "5"}
	stmts[2] = &ParticipantStmt{}
	go begin(stmts[2], ch, Args3[2])
	<-ch

	for i := 0; i < 3; i++ {
		stmts[i].Participant.Kv.SetDDL(time.Millisecond * 200)
		for j := 0; j < 5; j++ {
			stmts[i].Participant.Kv.Update(j, j)
		}
	}
	return stmts
}

func StopServers(stmt []*ParticipantStmt) {
	for _, v := range stmt {
		v.Stop()
	}
	time.Sleep(constants.OptEps * 10)
}

func CheckPreRead(coh *ParticipantManager, txn *remote.PRACTransaction, ans []int, shardID string) {
	if len(txn.OptList) > 0 {
		txn.OptList = txn.OptList[:0]
	}
	for i := 0; i < 5; i++ {
		txn.AddRead(shardID, i)
	}
	ok, val := coh.forTestPreRead(txn)
	utils.Assert(ok && len(val) == len(ans), "PreRead Failed or With different length")
	for i := 0; i < len(ans); i++ {
		utils.Assert(ans[i] == val[utils.Hash(coh.stmt.participantID, i)], "PreRead Failed with different value")
	}
}

func TestBuildTestCases(t *testing.T) {
	makeLocal()
	stmts := ParticipantsTestKit()
	CheckVal(stmts[0].Participant, []int{0, 1, 2, 3, 4})
	CheckVal(stmts[1].Participant, []int{0, 1, 2, 3, 4})
	CheckVal(stmts[2].Participant, []int{0, 1, 2, 3, 4})
	StopServers(stmts)
}

func TestCohortsPreRead(t *testing.T) {
	makeLocal()
	stmts := ParticipantsTestKit()
	txn1 := remote.NewPRACTransaction(1, stmts[0].participantID, rlsm.CFNF, address)
	txn2 := remote.NewPRACTransaction(2, stmts[1].participantID, rlsm.CFNF, address)

	c := make(chan bool)

	go func() {
		CheckPreRead(stmts[1].Participant, txn2, []int{0, 1, 2, 3, 4}, stmts[1].participantID)
		c <- true
	}()
	go func() {
		CheckPreRead(stmts[0].Participant, txn1, []int{0, 1, 2, 3, 4}, stmts[0].participantID)
		c <- true
	}()
	<-c
	<-c
	StopServers(stmts)
}

func TestCohortsPreWrite(t *testing.T) {
	makeLocal()
	stmts := ParticipantsTestKit()
	txn1 := remote.NewPRACTransaction(1, stmts[0].participantID, rlsm.CFNF, address)
	CheckPreRead(stmts[0].Participant, txn1, []int{0, 1, 2, 3, 4}, stmts[0].participantID)

	txn1.OptList = txn1.OptList[:0]
	for i := 0; i < 5; i++ {
		txn1.AddUpdate(stmts[0].participantID, i, i+1)
	}
	stmts[0].Participant.PreWrite(txn1)
	stmts[0].Participant.Commit(txn1)
	CheckVal(stmts[0].Participant, []int{1, 2, 3, 4, 5})
	StopServers(stmts)
}

func Test2PCCommit(t *testing.T) {
	makeLocal()
	stmts := ParticipantsTestKit()
	txn1 := remote.NewPRACTransaction(1, stmts[0].participantID, rlsm.CFNF, address)
	CheckPreRead(stmts[0].Participant, txn1, []int{0, 1, 2, 3, 4}, stmts[0].participantID)
	txn2 := remote.NewPRACTransaction(1, stmts[1].participantID, rlsm.CFNF, address)
	CheckPreRead(stmts[1].Participant, txn2, []int{0, 1, 2, 3, 4}, stmts[1].participantID)
	txn1.OptList = txn1.OptList[:0]
	txn2.OptList = txn2.OptList[:0]
	for i := 0; i < 5; i++ {
		txn1.AddUpdate(stmts[0].participantID, i, i+1)
		txn2.AddUpdate(stmts[1].participantID, i, i+2)
	}

	c := make(chan bool)

	go func() {
		ok := stmts[0].Participant.PreWrite(txn1)
		c <- utils.Assert(ok, "PreWrite Made Failed")
	}()
	go func() {
		ok := stmts[1].Participant.PreWrite(txn2)
		c <- utils.Assert(ok, "PreWrite Made Failed")
	}()
	<-c
	<-c

	go func() {
		ok := stmts[0].Participant.Commit(txn1)
		c <- utils.Assert(ok, "Commit Made Failed")
	}()
	go func() {
		ok := stmts[1].Participant.Commit(txn2)
		c <- utils.Assert(ok, "Commit Made Failed")
	}()
	<-c
	<-c

	CheckPreRead(stmts[0].Participant, txn1, []int{1, 2, 3, 4, 5}, stmts[0].participantID)
	CheckVal(stmts[0].Participant, []int{1, 2, 3, 4, 5})
	CheckVal(stmts[1].Participant, []int{2, 3, 4, 5, 6})
	StopServers(stmts)
}

func Test2PCAbort(t *testing.T) {
	makeLocal()
	stmts := ParticipantsTestKit()
	txn1 := remote.NewPRACTransaction(1, stmts[0].participantID, rlsm.CFNF, address)
	CheckPreRead(stmts[0].Participant, txn1, []int{0, 1, 2, 3, 4}, stmts[0].participantID)
	txn2 := remote.NewPRACTransaction(1, stmts[1].participantID, rlsm.CFNF, address)
	CheckPreRead(stmts[1].Participant, txn2, []int{0, 1, 2, 3, 4}, stmts[1].participantID)
	txn1.OptList = txn1.OptList[:0]
	txn2.OptList = txn2.OptList[:0]
	for i := 0; i < 5; i++ {
		txn1.AddUpdate(stmts[0].participantID, i, i+1)
		txn2.AddUpdate(stmts[1].participantID, i, i+2)
	}
	stmts[0].Participant.Kv.SetDDL(0)

	ok := stmts[0].Participant.PreWrite(txn1)
	ok = ok && stmts[1].Participant.PreWrite(txn2)
	utils.Assert(!ok, "PreWrite Fail Made Failed")

	ok = stmts[0].Participant.Abort(txn1)
	ok = ok && stmts[1].Participant.Abort(txn2) // if the first term is false, the commit will not be executed.
	utils.Assert(ok, "Abort Made Failed")

	CheckPreRead(stmts[0].Participant, txn1, []int{0, 1, 2, 3, 4}, stmts[0].participantID)
	CheckPreRead(stmts[1].Participant, txn2, []int{0, 1, 2, 3, 4}, stmts[1].participantID)
	StopServers(stmts)
}
