augurier
6 months ago
13 changed files with 443 additions and 539 deletions
Split View
Diff Options
-
+19 -6README.md
-
+5 -6cmd/main.go
-
+63 -42internal/client/client_node.go
-
+3 -79internal/nodes/init.go
-
+0 -208internal/nodes/node.go
-
+2 -1internal/nodes/node_storage.go
-
+214 -0internal/nodes/replica.go
-
+8 -4internal/nodes/server_node.go
-
+30 -13internal/nodes/vote.go
-
+0 -61scripts/run.sh
-
+29 -44test/restart_node_test.go
-
+2 -7test/server_client_test.go
+ 19
- 6
README.md
View File
+ 5
- 6
cmd/main.go
View File
+ 63
- 42
internal/client/client_node.go
View File
+ 3
- 79
internal/nodes/init.go
View File
+ 0
- 208
internal/nodes/node.go
View File
+ 2
- 1
internal/nodes/node_storage.go
View File
+ 214
- 0
internal/nodes/replica.go
View File
| @ -0,0 +1,214 @@ | |||
| package nodes | |||
| import ( | |||
| "math/rand" | |||
| "net/rpc" | |||
| "sort" | |||
| "strconv" | |||
| "time" | |||
| "go.uber.org/zap" | |||
| ) | |||
| type AppendEntriesArg struct { | |||
| Term int | |||
| LeaderId string | |||
| PrevLogIndex int | |||
| PrevLogTerm int | |||
| Entries []RaftLogEntry | |||
| LeaderCommit int | |||
| } | |||
| type AppendEntriesReply struct { | |||
| Term int | |||
| Success bool | |||
| } | |||
| // leader收到新内容要广播，以及心跳广播（同步自己的log) | |||
| func (node *Node) BroadCastKV(callMode CallMode) { | |||
| // 遍历所有节点 | |||
| for id := range node.nodes { | |||
| go func(id string, kv CallMode) { | |||
| node.sendKV(id, callMode) | |||
| }(id, callMode) | |||
| } | |||
| } | |||
| func (node *Node) sendKV(id string, callMode CallMode) { | |||
| switch callMode { | |||
| case Fail: | |||
| log.Info("模拟发送失败") | |||
| // 这么写向所有的node发送都失败，也可以随机数确定是否失败 | |||
| case Delay: | |||
| log.Info("模拟发送延迟") | |||
| // 随机延迟0-5ms | |||
| time.Sleep(time.Millisecond * time.Duration(rand.Intn(5))) | |||
| default: | |||
| } | |||
| client, err := rpc.DialHTTP("tcp", node.nodes[id].address) | |||
| if err != nil { | |||
| log.Error("dialing: ", zap.Error(err)) | |||
| return | |||
| } | |||
| defer func(client *rpc.Client) { | |||
| err := client.Close() | |||
| if err != nil { | |||
| log.Error("client close err: ", zap.Error(err)) | |||
| } | |||
| }(client) | |||
| node.mu.Lock() | |||
| defer node.mu.Unlock() | |||
| var appendReply AppendEntriesReply | |||
| appendReply.Success = false | |||
| nextIndex := node.nextIndex[id] | |||
| // log.Info("nextindex " + strconv.Itoa(nextIndex)) | |||
| for (!appendReply.Success) { | |||
| if nextIndex < 0 { | |||
| log.Fatal("assert >= 0 here") | |||
| } | |||
| sendEntries := node.log[nextIndex:] | |||
| arg := AppendEntriesArg{ | |||
| Term: node.currTerm, | |||
| PrevLogIndex: nextIndex - 1, | |||
| Entries: sendEntries, | |||
| LeaderCommit: node.commitIndex, | |||
| LeaderId: node.selfId, | |||
| } | |||
| if arg.PrevLogIndex >= 0 { | |||
| arg.PrevLogTerm = node.log[arg.PrevLogIndex].Term | |||
| } | |||
| callErr := client.Call("Node.AppendEntries", arg, &appendReply) // RPC | |||
| if callErr != nil { | |||
| log.Error("dialing node_"+ id +"fail: ", zap.Error(callErr)) | |||
| } | |||
| if appendReply.Term != node.currTerm { | |||
| log.Info("Leader[" + node.selfId + "]收到更高的 term=" + strconv.Itoa(appendReply.Term) + "，转换为 Follower") | |||
| node.currTerm = appendReply.Term | |||
| node.state = Follower | |||
| node.votedFor = "" | |||
| node.storage.SetTermAndVote(node.currTerm, node.votedFor) | |||
| node.resetElectionTimer() | |||
| return | |||
| } | |||
| nextIndex-- // 失败往前传一格 | |||
| } | |||
| // 不变成follower情况下 | |||
| node.nextIndex[id] = node.maxLogId + 1 | |||
| node.matchIndex[id] = node.maxLogId | |||
| node.updateCommitIndex() | |||
| } | |||
| func (node *Node) updateCommitIndex() { | |||
| totalNodes := len(node.nodes) | |||
| // 收集所有 matchIndex 并排序 | |||
| matchIndexes := make([]int, 0, totalNodes) | |||
| for _, index := range node.matchIndex { | |||
| matchIndexes = append(matchIndexes, index) | |||
| } | |||
| sort.Ints(matchIndexes) // 排序 | |||
| // 计算多数派 commitIndex | |||
| majorityIndex := matchIndexes[totalNodes/2] // 取 N/2 位置上的索引（多数派） | |||
| // 确保这个索引的日志条目属于当前 term，防止提交旧 term 的日志 | |||
| if majorityIndex > node.commitIndex && majorityIndex < len(node.log) && node.log[majorityIndex].Term == node.currTerm { | |||
| node.commitIndex = majorityIndex | |||
| log.Info("Leader[" + node.selfId + "]更新 commitIndex: " + strconv.Itoa(majorityIndex)) | |||
| // 应用日志到状态机 | |||
| node.applyCommittedLogs() | |||
| } | |||
| } | |||
| // 应用日志到状态机 | |||
| func (node *Node) applyCommittedLogs() { | |||
| for node.lastApplied < node.commitIndex { | |||
| node.lastApplied++ | |||
| logEntry := node.log[node.lastApplied] | |||
| log.Sugar().Infof("[%s]应用日志到状态机: " + logEntry.print(), node.selfId) | |||
| err := node.db.Put([]byte(logEntry.LogE.Key), []byte(logEntry.LogE.Value), nil) | |||
| if err != nil { | |||
| log.Error(node.selfId + "应用状态机失败: ", zap.Error(err)) | |||
| } | |||
| } | |||
| } | |||
| // RPC call | |||
| func (node *Node) AppendEntries(arg AppendEntriesArg, reply *AppendEntriesReply) error { | |||
| node.mu.Lock() | |||
| defer node.mu.Unlock() | |||
| // 如果 term 过期，拒绝接受日志 | |||
| if node.currTerm > arg.Term { | |||
| *reply = AppendEntriesReply{node.currTerm, false} | |||
| return nil | |||
| } | |||
| node.leaderId = arg.LeaderId // 记录Leader | |||
| // 如果term比自己高，或自己不是follower但收到相同term的心跳 | |||
| if node.currTerm < arg.Term || node.state != Follower { | |||
| log.Sugar().Infof("[%s]发现更高 term(%s)", node.selfId, strconv.Itoa(arg.Term)) | |||
| node.currTerm = arg.Term | |||
| node.state = Follower | |||
| node.votedFor = "" | |||
| node.storage.SetTermAndVote(node.currTerm, node.votedFor) | |||
| } | |||
| node.storage.SetTermAndVote(node.currTerm, node.votedFor) | |||
| // 检查 prevLogIndex 是否有效 | |||
| if arg.PrevLogIndex >= len(node.log) || (arg.PrevLogIndex >= 0 && node.log[arg.PrevLogIndex].Term != arg.PrevLogTerm) { | |||
| *reply = AppendEntriesReply{node.currTerm, false} | |||
| return nil | |||
| } | |||
| // 处理日志冲突（如果存在不同 term，则截断日志） | |||
| idx := arg.PrevLogIndex + 1 | |||
| for i := idx; i < len(node.log) && i-idx < len(arg.Entries); i++ { | |||
| if node.log[i].Term != arg.Entries[i-idx].Term { | |||
| node.log = node.log[:idx] | |||
| break | |||
| } | |||
| } | |||
| // log.Info(strconv.Itoa(idx) + strconv.Itoa(len(node.log))) | |||
| // 追加新的日志条目 | |||
| for _, raftLogEntry := range arg.Entries { | |||
| log.Sugar().Infof("[%s]写入：" + raftLogEntry.print(), node.selfId) | |||
| if idx < len(node.log) { | |||
| node.log[idx] = raftLogEntry | |||
| } else { | |||
| node.log = append(node.log, raftLogEntry) | |||
| } | |||
| idx++ | |||
| } | |||
| // 暴力持久化 | |||
| node.storage.WriteLog(node.log) | |||
| // 更新 maxLogId | |||
| node.maxLogId = len(node.log) - 1 | |||
| // 更新 commitIndex | |||
| if arg.LeaderCommit < node.maxLogId { | |||
| node.commitIndex = arg.LeaderCommit | |||
| } else { | |||
| node.commitIndex = node.maxLogId | |||
| } | |||
| // 提交已提交的日志 | |||
| node.applyCommittedLogs() | |||
| // 在成功接受日志或心跳后，重置选举超时 | |||
| node.resetElectionTimer() | |||
| *reply = AppendEntriesReply{node.currTerm, true} | |||
| return nil | |||
| } | |||
+ 8
- 4
internal/nodes/server_node.go
View File
+ 30
- 13
internal/nodes/vote.go
View File
+ 0
- 61
scripts/run.sh
View File
| @ -1,61 +0,0 @@ | |||
| #!/bin/bash | |||
| # 设置运行时间限制：s | |||
| RUN_TIME=10 | |||
| # 需要传递数据的管道 | |||
| PIPE_NAME="/tmp/input_pipe" | |||
| # 启动节点1 | |||
| echo "Starting Node 1..." | |||
| timeout $RUN_TIME ./main -id 1 -port ":9091" -cluster "127.0.0.1:9092,127.0.0.1:9093" -pipe "$PIPE_NAME" -isleader=true & | |||
| # 启动节点2 | |||
| echo "Starting Node 2..." | |||
| timeout $RUN_TIME ./main -id 2 -port ":9092" -cluster "127.0.0.1:9091,127.0.0.1:9093" -pipe "$PIPE_NAME" & | |||
| # 启动节点3 | |||
| echo "Starting Node 3..." | |||
| timeout $RUN_TIME ./main -id 3 -port ":9093" -cluster "127.0.0.1:9091,127.0.0.1:9092" -pipe "$PIPE_NAME"& | |||
| echo "All nodes started successfully!" | |||
| # 创建一个管道用于进程间通信 | |||
| if [[ ! -p "$PIPE_NAME" ]]; then | |||
| mkfifo "$PIPE_NAME" | |||
| fi | |||
| # 捕获终端输入并通过管道传递给三个节点 | |||
| echo "Enter input to send to nodes:" | |||
| start_time=$(date +%s) | |||
| while true; do | |||
| # 从终端读取用户输入 | |||
| read -r user_input | |||
| current_time=$(date +%s) | |||
| elapsed_time=$((current_time - start_time)) | |||
| # 如果运行时间大于限制时间，就退出 | |||
| if [ $elapsed_time -ge $RUN_TIME ]; then | |||
| echo 'Timeout reached, normal exit now' | |||
| break | |||
| fi | |||
| # 如果输入为空，跳过 | |||
| if [[ -z "$user_input" ]]; then | |||
| continue | |||
| fi | |||
| # 将用户输入发送到管道 | |||
| echo "$user_input" > "$PIPE_NAME" | |||
| # 如果输入 "exit"，结束脚本 | |||
| if [[ "$user_input" == "exit" ]]; then | |||
| break | |||
| fi | |||
| done | |||
| # 删除管道 | |||
| rm "$PIPE_NAME" | |||
| # 等待所有节点完成启动 | |||
| wait | |||