补充测试

internal/nodes/init.go

@@ -37,6 +37,7 @@ func InitRPCNode(SelfId string, port string, nodeAddr map[string]string, db *lev
 		Db:          db,
 		Storage:     rstorage,
 		Transport:   &HTTPTransport{NodeMap:  nodeAddr},
+		RTTable: NewRTTable(),
 		SeenRequests: make(map[LogEntryCallId]bool),
 	}
 	node.initLeaderState()
@@ -91,6 +92,7 @@ func InitThreadNode(SelfId string, peerIds []string, db *leveldb.DB, rstorage *R
 		Db:          db,
 		Storage:     rstorage,
 		Transport:   threadTransport,
+		RTTable: NewRTTable(),
 		SeenRequests: make(map[LogEntryCallId]bool),
 	}
 	node.initLeaderState()

internal/nodes/node.go

@@ -60,6 +60,9 @@ type Node struct {
 	// 通信方式
 	Transport Transport
 
+	// 系统的随机时间
+	RTTable *RandomTimeTable
+
 	// 已经处理过的客户端请求
 	SeenRequests map[LogEntryCallId]bool
 }

internal/nodes/random_timetable.go

@@ -0,0 +1,38 @@
+package nodes
+
+import (
+	"math/rand"
+	"time"
+)
+
+type RandomTimeTable struct {
+	electionTimeOut time.Duration
+	israndom bool
+	// heartbeat 50ms
+	// rpcTimeout 100ms
+	// follower变candidate 500ms
+	// 等待选举成功时间 300ms
+}
+
+func NewRTTable() *RandomTimeTable {
+	return &RandomTimeTable{
+		israndom: true,
+	}
+}
+
+func (rttable *RandomTimeTable) GetElectionTimeout() time.Duration {
+	if rttable.israndom {
+		return time.Duration(500+rand.Intn(500)) * time.Millisecond
+	} else {
+		return time.Duration(rttable.electionTimeOut)
+	}
+}
+
+func (rttable *RandomTimeTable) SetElectionTimeout(t time.Duration) {
+	rttable.israndom = false
+	rttable.electionTimeOut = t
+}
+
+func (rttable *RandomTimeTable) ResetElectionTimeout() {
+	rttable.israndom = true
+}

internal/nodes/vote.go

@@ -175,10 +175,10 @@ func (n *Node) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) error
 	return nil
 }
 
-// follower 500-1000ms内没收到appendentries心跳，就变成candidate发起选举
+// follower 150-300ms内没收到appendentries心跳，就变成candidate发起选举
 func (node *Node) ResetElectionTimer() {
 	if node.ElectionTimer == nil {
-		node.ElectionTimer = time.NewTimer(time.Duration(500+rand.Intn(500)) * time.Millisecond)
+		node.ElectionTimer = time.NewTimer(node.RTTable.GetElectionTimeout())
 		go func() {
 			for {
 				<-node.ElectionTimer.C

threadTest/common.go

@@ -3,7 +3,9 @@ package threadTest
 import (
 	"fmt"
 	"os"
+	"simple-kv-store/internal/client"
 	"simple-kv-store/internal/nodes"
+	"strconv"
 	"testing"
 	"time"
 
@@ -102,6 +104,28 @@ func SendKvCall(kvCall *nodes.LogEntryCall, node *nodes.Node) {
 	node.BroadCastKV()
 }
 
+func ClientWriteLog(t *testing.T, startLogid int, endLogid int, cWrite *clientPkg.Client) {
+	var s clientPkg.Status
+	for i := startLogid; i < endLogid; i++ {
+		key := strconv.Itoa(i)
+		newlog := nodes.LogEntry{Key: key, Value: "hello"}
+		s = cWrite.Write(newlog)
+		if s != clientPkg.Ok {
+			t.Errorf("write test fail")
+		}		
+	}
+}
+
+func FindLeader(t *testing.T, nodeCollections []* nodes.Node) (i int) {
+	for i, node := range nodeCollections {
+		if node.State == nodes.Leader {
+			return i
+		}
+	}
+	t.Errorf("系统目前没有leader")
+	return 0
+}
+
 func CheckOneLeader(t *testing.T, nodeCollections []* nodes.Node) {
 	cnt := 0
 	for _, node := range nodeCollections {
@@ -114,6 +138,18 @@ func CheckOneLeader(t *testing.T, nodeCollections []* nodes.Node) {
 	}
 }
 
+func CheckNoLeader(t *testing.T, nodeCollections []* nodes.Node) {
+	cnt := 0
+	for _, node := range nodeCollections {
+		if node.State == nodes.Leader {
+			cnt++
+		}
+	}
+	if cnt != 0 {
+		t.Errorf("实际有%d个leader(!=0)", cnt)
+	}
+}
+
 func CheckZeroOrOneLeader(t *testing.T, nodeCollections []* nodes.Node) {
 	cnt := 0
 	for _, node := range nodeCollections {

threadTest/election_test.go

@@ -198,6 +198,18 @@ func TestRepeatVoteRpc(t *testing.T) {
 	CheckOneLeader(t, nodeCollections)
 	CheckIsLeader(t, nodeCollections[0])
 	CheckTerm(t, nodeCollections[0], 2)
+
+	for i := 0; i < n; i++ {
+		ctx.SetBehavior("1", nodeCollections[i].SelfId, nodes.RetryRpc, 0, 2)
+		ctx.SetBehavior("2", nodeCollections[i].SelfId, nodes.RetryRpc, 0, 2)
+	}
+
+	go nodeCollections[0].StartElection()
+	go nodeCollections[1].StartElection()
+	time.Sleep(time.Second)
+
+	CheckOneLeader(t, nodeCollections)
+	CheckTerm(t, nodeCollections[0], 3)
 }
 
 func TestFailVoteRpc(t *testing.T) {
@@ -237,4 +249,65 @@ func TestFailVoteRpc(t *testing.T) {
 	CheckOneLeader(t, nodeCollections)
 	CheckIsLeader(t, nodeCollections[0])
 	CheckTerm(t, nodeCollections[0], 2)
-}
+
+	ctx.SetBehavior("1", "3", nodes.FailRpc, 0, 0)
+	ctx.SetBehavior("1", "4", nodes.FailRpc, 0, 0)
+	nodeCollections[0].StartElection()
+	time.Sleep(time.Second)
+
+	CheckNoLeader(t, nodeCollections)
+	CheckTerm(t, nodeCollections[0], 3)
+}
+
+func TestDelayVoteRpc(t *testing.T) {
+	n := 5
+	var peerIds []string
+	for i := 0; i < n; i++ {
+		peerIds = append(peerIds, strconv.Itoa(i + 1))
+	}
+
+	// 结点启动
+	var quitCollections []chan struct{}
+	var nodeCollections []*nodes.Node
+	ctx := nodes.NewCtx()
+	threadTransport := nodes.NewThreadTransport(ctx)
+	for i := 0; i < n; i++ {
+		n, quitChan := ExecuteStaticNodeI(strconv.Itoa(i + 1), false, peerIds, threadTransport)
+		quitCollections = append(quitCollections, quitChan)
+		nodeCollections = append(nodeCollections, n)
+	}
+	StopElectionReset(nodeCollections, quitCollections)
+
+	// 通知所有node结束
+	defer func(){
+		for _, quitChan := range quitCollections {
+			close(quitChan)
+		}
+	}()
+
+	for i := 0; i < n; i++ {
+		nodeCollections[i].State = nodes.Follower
+		ctx.SetBehavior("1", nodeCollections[i].SelfId, nodes.DelayRpc, time.Second, 0)
+	}
+
+	nodeCollections[0].StartElection()
+	time.Sleep(2 * time.Second)
+
+	CheckNoLeader(t, nodeCollections)
+	for i := 0; i < n; i++ {
+		CheckTerm(t, nodeCollections[i], 2)
+	}
+
+	for i := 0; i < n; i++ {
+		nodeCollections[i].State = nodes.Follower
+		ctx.SetBehavior("1", nodeCollections[i].SelfId, nodes.DelayRpc, 50 * time.Millisecond, 0)
+	}
+
+	nodeCollections[0].StartElection()
+	time.Sleep(time.Second)
+
+	CheckOneLeader(t, nodeCollections)
+	for i := 0; i < n; i++ {
+		CheckTerm(t, nodeCollections[i], 3)
+	}
+}

threadTest/log_replication_test.go

@@ -155,3 +155,170 @@ func TestFollowerLagging(t *testing.T) {
 		CheckLogNum(t, nodeCollections[i], 10)
 	}
 }
+
+func TestFailLogAppendRpc(t *testing.T) {
+	n := 5
+	var peerIds []string
+	for i := 0; i < n; i++ {
+		peerIds = append(peerIds, strconv.Itoa(i + 1))
+	}
+
+	// 结点启动
+	var quitCollections []chan struct{}
+	var nodeCollections []*nodes.Node
+	ctx := nodes.NewCtx()
+	threadTransport := nodes.NewThreadTransport(ctx)
+	for i := 0; i < n; i++ {
+		n, quitChan := ExecuteStaticNodeI(strconv.Itoa(i + 1), false, peerIds, threadTransport)
+		quitCollections = append(quitCollections, quitChan)
+		nodeCollections = append(nodeCollections, n)
+	}
+	StopElectionReset(nodeCollections, quitCollections)
+
+	// 通知所有node结束
+	defer func(){
+		for _, quitChan := range quitCollections {
+			close(quitChan)
+		}
+	}()
+
+	for i := 0; i < n; i++ {
+		nodeCollections[i].State = nodes.Follower
+	}
+
+	nodeCollections[0].StartElection()
+	time.Sleep(time.Second)
+	CheckOneLeader(t, nodeCollections)
+	CheckIsLeader(t, nodeCollections[0])
+	CheckTerm(t, nodeCollections[0], 2)
+	
+	for i := 0; i < n; i++ {
+		ctx.SetBehavior("1", nodeCollections[i].SelfId, nodes.FailRpc, 0, 0)
+	}
+	
+	for i := 0; i < 10; i++ {
+		key := strconv.Itoa(i)
+		newlog := nodes.LogEntry{Key: key, Value: "hello"}
+		go SendKvCall(&nodes.LogEntryCall{LogE: newlog}, nodeCollections[0])
+	}
+
+	time.Sleep(time.Second)
+	for i := 1; i < n; i++ {
+		CheckLogNum(t, nodeCollections[i], 0)
+	}
+}
+
+func TestRepeatLogAppendRpc(t *testing.T) {
+	n := 5
+	var peerIds []string
+	for i := 0; i < n; i++ {
+		peerIds = append(peerIds, strconv.Itoa(i + 1))
+	}
+
+	// 结点启动
+	var quitCollections []chan struct{}
+	var nodeCollections []*nodes.Node
+	ctx := nodes.NewCtx()
+	threadTransport := nodes.NewThreadTransport(ctx)
+	for i := 0; i < n; i++ {
+		n, quitChan := ExecuteStaticNodeI(strconv.Itoa(i + 1), false, peerIds, threadTransport)
+		quitCollections = append(quitCollections, quitChan)
+		nodeCollections = append(nodeCollections, n)
+	}
+	StopElectionReset(nodeCollections, quitCollections)
+
+	// 通知所有node结束
+	defer func(){
+		for _, quitChan := range quitCollections {
+			close(quitChan)
+		}
+	}()
+
+	for i := 0; i < n; i++ {
+		nodeCollections[i].State = nodes.Follower
+	}
+
+	nodeCollections[0].StartElection()
+	time.Sleep(time.Second)
+	CheckOneLeader(t, nodeCollections)
+	CheckIsLeader(t, nodeCollections[0])
+	CheckTerm(t, nodeCollections[0], 2)
+	
+	for i := 0; i < n; i++ {
+		ctx.SetBehavior("1", nodeCollections[i].SelfId, nodes.RetryRpc, 0, 2)
+	}
+	
+	for i := 0; i < 10; i++ {
+		key := strconv.Itoa(i)
+		newlog := nodes.LogEntry{Key: key, Value: "hello"}
+		go SendKvCall(&nodes.LogEntryCall{LogE: newlog}, nodeCollections[0])
+	}
+
+	time.Sleep(time.Second)
+	for i := 0; i < n; i++ {
+		CheckLogNum(t, nodeCollections[i], 10)
+	}
+}
+
+func TestDelayLogAppendRpc(t *testing.T) {
+	n := 5
+	var peerIds []string
+	for i := 0; i < n; i++ {
+		peerIds = append(peerIds, strconv.Itoa(i + 1))
+	}
+
+	// 结点启动
+	var quitCollections []chan struct{}
+	var nodeCollections []*nodes.Node
+	ctx := nodes.NewCtx()
+	threadTransport := nodes.NewThreadTransport(ctx)
+	for i := 0; i < n; i++ {
+		n, quitChan := ExecuteStaticNodeI(strconv.Itoa(i + 1), false, peerIds, threadTransport)
+		quitCollections = append(quitCollections, quitChan)
+		nodeCollections = append(nodeCollections, n)
+	}
+	StopElectionReset(nodeCollections, quitCollections)
+
+	// 通知所有node结束
+	defer func(){
+		for _, quitChan := range quitCollections {
+			close(quitChan)
+		}
+	}()
+
+	for i := 0; i < n; i++ {
+		nodeCollections[i].State = nodes.Follower
+	}
+
+	nodeCollections[0].StartElection()
+	time.Sleep(time.Second)
+	CheckOneLeader(t, nodeCollections)
+	CheckIsLeader(t, nodeCollections[0])
+	CheckTerm(t, nodeCollections[0], 2)
+	
+	for i := 0; i < n; i++ {
+		ctx.SetBehavior("1", nodeCollections[i].SelfId, nodes.DelayRpc, time.Second, 0)
+	}
+	
+	for i := 0; i < 5; i++ {
+		key := strconv.Itoa(i)
+		newlog := nodes.LogEntry{Key: key, Value: "hello"}
+		go SendKvCall(&nodes.LogEntryCall{LogE: newlog}, nodeCollections[0])
+	}
+
+	time.Sleep(time.Millisecond * 100)
+
+	for i := 0; i < n; i++ {
+		ctx.SetBehavior("1", nodeCollections[i].SelfId, nodes.NormalRpc, 0, 0)
+	}
+	for i := 5; i < 10; i++ {
+		key := strconv.Itoa(i)
+		newlog := nodes.LogEntry{Key: key, Value: "hello"}
+		go SendKvCall(&nodes.LogEntryCall{LogE: newlog}, nodeCollections[0])
+	}
+
+	time.Sleep(time.Second * 2)
+	for i := 0; i < n; i++ {
+		CheckLogNum(t, nodeCollections[i], 10)
+	}
+}

threadTest/restart_node_test.go

@@ -35,15 +35,7 @@ func TestNodeRestart(t *testing.T) {
 	// client启动, 连接任意节点
 	cWrite := clientPkg.NewClient("0", peerIds, threadTransport)
 	// 写入
-	var s clientPkg.Status
-	for i := 0; i < 5; i++ {
-		key := strconv.Itoa(i)
-		newlog := nodes.LogEntry{Key: key, Value: "hello"}
-		s = cWrite.Write(newlog)
-		if s != clientPkg.Ok {
-			t.Errorf("write test fail")
-		}		
-	}
+	ClientWriteLog(t, 0, 5, cWrite)
 	time.Sleep(time.Second) // 等待写入完毕
 
 	// 模拟结点轮流崩溃
@@ -63,7 +55,7 @@ func TestNodeRestart(t *testing.T) {
 	for i := 0; i < 5; i++ {
 		key := strconv.Itoa(i)
 		var value string
-		s = cRead.Read(key, &value)
+		s := cRead.Read(key, &value)
 		if s != clientPkg.Ok {
 			t.Errorf("Read test1 fail")
 		}
@@ -73,9 +65,65 @@ func TestNodeRestart(t *testing.T) {
 	for i := 5; i < 15; i++ {
 		key := strconv.Itoa(i)
 		var value string
-		s = cRead.Read(key, &value)
+		s := cRead.Read(key, &value)
 		if s != clientPkg.NotFound {
 			t.Errorf("Read test2 fail")
 		}
 	}
 }
+
+
+func TestRestartWhileWriting(t *testing.T) {
+	// 登记结点信息
+	n := 5
+	var peerIds []string
+	for i := 0; i < n; i++ {
+		peerIds = append(peerIds, strconv.Itoa(i + 1))
+	}
+
+	// 结点启动
+	var quitCollections []chan struct{}
+	var nodeCollections []*nodes.Node
+	threadTransport := nodes.NewThreadTransport(nodes.NewCtx())
+	for i := 0; i < n; i++ {
+		n, quitChan := ExecuteNodeI(strconv.Itoa(i + 1), false, peerIds, threadTransport)
+		quitCollections = append(quitCollections, quitChan)
+		nodeCollections = append(nodeCollections, n)
+	}
+
+	// 通知所有node结束
+	defer func(){
+		for _, quitChan := range quitCollections {
+			close(quitChan)
+		}
+	}()
+
+	time.Sleep(time.Second) // 等待启动完毕
+	leaderIdx := FindLeader(t, nodeCollections)
+	// client启动, 连接任意节点
+	cWrite := clientPkg.NewClient("0", peerIds, threadTransport)
+	// 写入
+	go ClientWriteLog(t, 0, 5, cWrite)
+
+	go func() {
+		close(quitCollections[leaderIdx])
+
+		n, quitChan := ExecuteNodeI(strconv.Itoa(leaderIdx + 1), true, peerIds, threadTransport)
+		quitCollections[leaderIdx] = quitChan
+		nodeCollections[leaderIdx] = n
+	}()
+
+	time.Sleep(time.Second) // 等待启动完毕	
+	// client启动
+	cRead := clientPkg.NewClient("0", peerIds, threadTransport)
+	// 读写入数据
+	for i := 0; i < 5; i++ {
+		key := strconv.Itoa(i)
+		var value string
+		s := cRead.Read(key, &value)
+		if s != clientPkg.Ok {
+			t.Errorf("Read test1 fail")
+		}
+	}
+	CheckLogNum(t, nodeCollections[leaderIdx], 5)
+}

threadTest/server_client_test.go

@@ -137,3 +137,56 @@ func TestRepeatClientReq(t *testing.T) {
 		CheckLogNum(t, nodeCollections[i], 10)
 	}
 }
+
+func TestParallelClientReq(t *testing.T) {
+	// 登记结点信息
+	n := 5
+	var peerIds []string
+	for i := 0; i < n; i++ {
+		peerIds = append(peerIds, strconv.Itoa(i + 1))
+	}
+
+	// 结点启动
+	var quitCollections []chan struct{}
+	var nodeCollections []*nodes.Node
+	ctx := nodes.NewCtx()
+	threadTransport := nodes.NewThreadTransport(ctx)
+	for i := 0; i < n; i++ {
+		n, quitChan := ExecuteNodeI(strconv.Itoa(i + 1), false, peerIds, threadTransport)
+		quitCollections = append(quitCollections, quitChan)
+		nodeCollections = append(nodeCollections, n)
+	}
+
+	// 通知所有node结束
+	defer func(){
+		for _, quitChan := range quitCollections {
+			close(quitChan)
+		}
+	}()
+
+	time.Sleep(time.Second) // 等待启动完毕
+	// client启动
+	c1 := clientPkg.NewClient("0", peerIds, threadTransport)
+	c2 := clientPkg.NewClient("1", peerIds, threadTransport)
+
+
+
+	// 写入
+	go ClientWriteLog(t, 0, 10, c1)
+	go ClientWriteLog(t, 0, 10, c2)
+
+	time.Sleep(time.Second) // 等待写入完毕
+	// 读写入数据
+	for i := 0; i < 10; i++ {
+		key := strconv.Itoa(i)
+		var value string
+		s := c1.Read(key, &value)
+		if s != clientPkg.Ok || value != "hello" {
+			t.Errorf("Read test1 fail")
+		}
+	}
+
+	for i := 0; i < n; i++ {
+		CheckLogNum(t, nodeCollections[i], 20)
+	}
+}