Browse Source
Merge pull request 'ld' (#1) from ld into master
Reviewed-on: https://gitea.shuishan.net.cn/10225501448/simple-kv-store/pulls/1part1
15 changed files with 558 additions and 91 deletions
Split View
Diff Options
-
+2 -1.gitignore
-
+3 -1README.md
-
+49 -5cmd/main.go
-
+2 -0go.mod
-
+18 -0go.sum
-
+94 -0internal/client/client_node.go
-
+42 -30internal/nodes/init.go
-
+16 -0internal/nodes/log.go
-
+43 -49internal/nodes/node.go
-
+42 -0internal/nodes/server_node.go
-
+1 -1scripts/build.sh
-
+4 -4scripts/run.sh
-
+45 -0test/common.go
-
+112 -0test/restart_follower_test.go
-
+85 -0test/server_client_test.go
+ 2
- 1
.gitignore
View File
+ 3
- 1
README.md
View File
+ 49
- 5
cmd/main.go
View File
+ 2
- 0
go.mod
View File
+ 18
- 0
go.sum
View File
+ 94
- 0
internal/client/client_node.go
View File
| @ -0,0 +1,94 @@ | |||
| package clientPkg | |||
| import ( | |||
| "net/rpc" | |||
| "simple-kv-store/internal/logprovider" | |||
| "simple-kv-store/internal/nodes" | |||
| "go.uber.org/zap" | |||
| ) | |||
| var log, _ = logprovider.CreateDefaultZapLogger(zap.InfoLevel) | |||
| type Client struct { | |||
| // 连接的server端节点（node1） | |||
| ServerId string | |||
| Address string | |||
| } | |||
| type Status = uint8 | |||
| const ( | |||
| Ok Status = iota + 1 | |||
| NotFound | |||
| Fail | |||
| ) | |||
| func (client *Client) Write(kvCall nodes.LogEntryCall) Status { | |||
| log.Info("client write request key :" + kvCall.LogE.Key) | |||
| c, err := rpc.DialHTTP("tcp", client.Address) | |||
| if err != nil { | |||
| log.Error("dialing: ", zap.Error(err)) | |||
| return Fail | |||
| } | |||
| defer func(server *rpc.Client) { | |||
| err := c.Close() | |||
| if err != nil { | |||
| log.Error("client close err: ", zap.Error(err)) | |||
| } | |||
| }(c) | |||
| var reply nodes.ServerReply | |||
| callErr := c.Call("Node.WriteKV", kvCall, &reply) // RPC | |||
| if callErr != nil { | |||
| log.Error("dialing: ", zap.Error(callErr)) | |||
| return Fail | |||
| } | |||
| if reply.Isconnect { // 发送成功 | |||
| return Ok | |||
| } else { // 失败 | |||
| return Fail | |||
| } | |||
| } | |||
| func (client *Client) Read(key string, value *string) Status { // 查不到value为空 | |||
| log.Info("client read request key :" + key) | |||
| if value == nil { | |||
| return Fail | |||
| } | |||
| c, err := rpc.DialHTTP("tcp", client.Address) | |||
| if err != nil { | |||
| log.Error("dialing: ", zap.Error(err)) | |||
| return Fail | |||
| } | |||
| defer func(server *rpc.Client) { | |||
| err := c.Close() | |||
| if err != nil { | |||
| log.Error("client close err: ", zap.Error(err)) | |||
| } | |||
| }(c) | |||
| var reply nodes.ServerReply | |||
| callErr := c.Call("Node.ReadKey", key, &reply) // RPC | |||
| if callErr != nil { | |||
| log.Error("dialing: ", zap.Error(callErr)) | |||
| return Fail | |||
| } | |||
| if reply.Isconnect { // 发送成功 | |||
| if reply.HaveValue { | |||
| *value = reply.Value | |||
| return Ok | |||
| } else { | |||
| return NotFound | |||
| } | |||
| } else { // 失败 | |||
| return Fail | |||
| } | |||
| } | |||
+ 42
- 30
internal/nodes/init.go
View File
+ 16
- 0
internal/nodes/log.go
View File
| @ -1,10 +1,26 @@ | |||
| package nodes | |||
| const ( | |||
| Normal State = iota + 1 | |||
| Delay | |||
| Fail | |||
| ) | |||
| type LogEntry struct { | |||
| Key string | |||
| Value string | |||
| } | |||
| type LogEntryCall struct { | |||
| LogE LogEntry | |||
| CallState State | |||
| } | |||
| type KVReply struct { | |||
| Reply bool | |||
| } | |||
| type LogIdAndEntry struct { | |||
| LogId int | |||
| Entry LogEntry | |||
| } | |||
+ 43
- 49
internal/nodes/node.go
View File
+ 42
- 0
internal/nodes/server_node.go
View File
| @ -0,0 +1,42 @@ | |||
| package nodes | |||
| import ( | |||
| "strconv" | |||
| "github.com/syndtr/goleveldb/leveldb" | |||
| ) | |||
| // leader node作为server为client注册的方法 | |||
| type ServerReply struct{ | |||
| Isconnect bool | |||
| HaveValue bool | |||
| Value string | |||
| } | |||
| // RPC call | |||
| func (node *Node) WriteKV(kvCall LogEntryCall, reply *ServerReply) error { | |||
| logId := node.maxLogId | |||
| node.maxLogId++ | |||
| node.log[logId] = kvCall.LogE | |||
| node.db.Put([]byte(kvCall.LogE.Key), []byte(kvCall.LogE.Value), nil) | |||
| log.Info("server write : logId = " + strconv.Itoa(logId) + ", key = " + kvCall.LogE.Key) | |||
| // 广播给其它节点 | |||
| node.BroadCastKV(logId, kvCall) | |||
| reply.Isconnect = true | |||
| return nil | |||
| } | |||
| // RPC call | |||
| func (node *Node) ReadKey(key string, reply *ServerReply) error { | |||
| log.Info("server read : " + key) | |||
| // 先只读leader自己 | |||
| value, err := node.db.Get([]byte(key), nil) | |||
| if err == leveldb.ErrNotFound { | |||
| reply.HaveValue = false | |||
| } else { | |||
| reply.HaveValue = true | |||
| reply.Value = string(value) | |||
| } | |||
| reply.Isconnect = true | |||
| return nil | |||
| } | |||
+ 1
- 1
scripts/build.sh
View File
| @ -1 +1 @@ | |||
| go build -o raftnode ./cmd/main.go | |||
| go build -o main ./cmd/main.go | |||
+ 4
- 4
scripts/run.sh
View File
+ 45
- 0
test/common.go
View File
| @ -0,0 +1,45 @@ | |||
| package test | |||
| import ( | |||
| "fmt" | |||
| "os" | |||
| "os/exec" | |||
| "strconv" | |||
| "strings" | |||
| ) | |||
| func ExecuteNodeI(i int, isLeader bool, isNewDb bool, clusters []string) *exec.Cmd { | |||
| tmpClusters := append(clusters[:i], clusters[i+1:]...) | |||
| port := fmt.Sprintf(":%d", uint16(9090)+uint16(i)) | |||
| var isleader string | |||
| if isLeader { | |||
| isleader = "true" | |||
| } else { | |||
| isleader = "false" | |||
| } | |||
| var isnewdb string | |||
| if isNewDb { | |||
| isnewdb = "true" | |||
| } else { | |||
| isnewdb = "false" | |||
| } | |||
| cmd := exec.Command( | |||
| "../main", | |||
| "-id", strconv.Itoa(i + 1), | |||
| "-port", port, | |||
| "-cluster", strings.Join(tmpClusters, ","), | |||
| "-isleader=" + isleader, | |||
| "-isNewDb=" + isnewdb, | |||
| ) | |||
| cmd.Stdout = os.Stdout | |||
| cmd.Stderr = os.Stderr | |||
| // 执行命令 | |||
| err := cmd.Start() | |||
| if err != nil { | |||
| fmt.Println("启动进程出错:", err) | |||
| return nil | |||
| } | |||
| return cmd | |||
| } | |||
+ 112
- 0
test/restart_follower_test.go
View File
| @ -0,0 +1,112 @@ | |||
| package test | |||
| import ( | |||
| "fmt" | |||
| "os/exec" | |||
| "simple-kv-store/internal/client" | |||
| "simple-kv-store/internal/nodes" | |||
| "strconv" | |||
| "syscall" | |||
| "testing" | |||
| "time" | |||
| ) | |||
| func TestFollowerRestart(t *testing.T) { | |||
| // 登记结点信息 | |||
| n := 3 | |||
| var clusters []string | |||
| for i := 0; i < n; i++ { | |||
| port := fmt.Sprintf("%d", uint16(9090)+uint16(i)) | |||
| addr := "127.0.0.1:" + port | |||
| clusters = append(clusters, addr) | |||
| } | |||
| // 结点启动 | |||
| var cmds []*exec.Cmd | |||
| for i := 0; i < n; i++ { | |||
| var cmd *exec.Cmd | |||
| if i == 0 { | |||
| cmd = ExecuteNodeI(i, true, true, clusters) | |||
| } else { | |||
| cmd = ExecuteNodeI(i, false, true, clusters) | |||
| } | |||
| if cmd == nil { | |||
| return | |||
| } else { | |||
| cmds = append(cmds, cmd) | |||
| } | |||
| } | |||
| time.Sleep(time.Second) // 等待启动完毕 | |||
| // client启动, 连接leader | |||
| cWrite := clientPkg.Client{Address: clusters[0], ServerId: "1"} | |||
| // 写入 | |||
| var s clientPkg.Status | |||
| for i := 0; i < 5; i++ { | |||
| key := strconv.Itoa(i) | |||
| newlog := nodes.LogEntry{Key: key, Value: "hello"} | |||
| s := cWrite.Write(nodes.LogEntryCall{LogE: newlog, CallState: nodes.Normal}) | |||
| if s != clientPkg.Ok { | |||
| t.Errorf("write test fail") | |||
| } | |||
| } | |||
| time.Sleep(time.Second) // 等待写入完毕 | |||
| // 模拟最后一个结点崩溃 | |||
| err := cmds[n - 1].Process.Signal(syscall.SIGTERM) | |||
| if err != nil { | |||
| fmt.Println("Error sending signal:", err) | |||
| return | |||
| } | |||
| // 继续写入 | |||
| for i := 5; i < 10; i++ { | |||
| key := strconv.Itoa(i) | |||
| newlog := nodes.LogEntry{Key: key, Value: "hello"} | |||
| s := cWrite.Write(nodes.LogEntryCall{LogE: newlog, CallState: nodes.Normal}) | |||
| if s != clientPkg.Ok { | |||
| t.Errorf("write test fail") | |||
| } | |||
| } | |||
| // 恢复结点 | |||
| cmd := ExecuteNodeI(n - 1, false, false, clusters) | |||
| if cmd == nil { | |||
| t.Errorf("recover test1 fail") | |||
| return | |||
| } else { | |||
| cmds[n - 1] = cmd | |||
| } | |||
| time.Sleep(time.Second) // 等待启动完毕 | |||
| // client启动, 连接节点n-1(去读它的数据) | |||
| cRead := clientPkg.Client{Address: clusters[n - 1], ServerId: "n"} | |||
| // 读崩溃前写入数据 | |||
| 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") | |||
| } | |||
| } | |||
| // 读未写入数据 | |||
| for i := 5; i < 15; i++ { | |||
| key := strconv.Itoa(i) | |||
| var value string | |||
| s = cRead.Read(key, &value) | |||
| if s != clientPkg.NotFound { | |||
| t.Errorf("Read test2 fail") | |||
| } | |||
| } | |||
| // 通知进程结束 | |||
| for _, cmd := range cmds { | |||
| err := cmd.Process.Signal(syscall.SIGTERM) | |||
| if err != nil { | |||
| fmt.Println("Error sending signal:", err) | |||
| return | |||
| } | |||
| } | |||
| } | |||
+ 85
- 0
test/server_client_test.go
View File
| @ -0,0 +1,85 @@ | |||
| package test | |||
| import ( | |||
| "fmt" | |||
| "os/exec" | |||
| "simple-kv-store/internal/client" | |||
| "simple-kv-store/internal/nodes" | |||
| "strconv" | |||
| "syscall" | |||
| "testing" | |||
| "time" | |||
| ) | |||
| func TestServerClient(t *testing.T) { | |||
| // 登记结点信息 | |||
| n := 5 | |||
| var clusters []string | |||
| for i := 0; i < n; i++ { | |||
| port := fmt.Sprintf("%d", uint16(9090)+uint16(i)) | |||
| addr := "127.0.0.1:" + port | |||
| clusters = append(clusters, addr) | |||
| } | |||
| // 结点启动 | |||
| var cmds []*exec.Cmd | |||
| for i := 0; i < n; i++ { | |||
| var cmd *exec.Cmd | |||
| if i == 0 { | |||
| cmd = ExecuteNodeI(i, true, true, clusters) | |||
| } else { | |||
| cmd = ExecuteNodeI(i, false, true, clusters) | |||
| } | |||
| if cmd == nil { | |||
| return | |||
| } else { | |||
| cmds = append(cmds, cmd) | |||
| } | |||
| } | |||
| time.Sleep(time.Second) // 等待启动完毕 | |||
| // client启动 | |||
| c := clientPkg.Client{Address: "127.0.0.1:9090", ServerId: "1"} | |||
| // 写入 | |||
| var s clientPkg.Status | |||
| for i := 0; i < 10; i++ { | |||
| key := strconv.Itoa(i) | |||
| newlog := nodes.LogEntry{Key: key, Value: "hello"} | |||
| s := c.Write(nodes.LogEntryCall{LogE: newlog, CallState: nodes.Normal}) | |||
| if s != clientPkg.Ok { | |||
| t.Errorf("write test fail") | |||
| } | |||
| } | |||
| // 读写入数据 | |||
| for i := 0; i < 10; i++ { | |||
| key := strconv.Itoa(i) | |||
| var value string | |||
| s = c.Read(key, &value) | |||
| if s != clientPkg.Ok && value != "hello" + key { | |||
| t.Errorf("Read test1 fail") | |||
| } | |||
| } | |||
| // 读未写入数据 | |||
| for i := 10; i < 15; i++ { | |||
| key := strconv.Itoa(i) | |||
| var value string | |||
| s = c.Read(key, &value) | |||
| if s != clientPkg.NotFound { | |||
| t.Errorf("Read test2 fail") | |||
| } | |||
| } | |||
| // 通知进程结束 | |||
| for _, cmd := range cmds { | |||
| err := cmd.Process.Signal(syscall.SIGTERM) | |||
| if err != nil { | |||
| fmt.Println("Error sending signal:", err) | |||
| return | |||
| } | |||
| } | |||
| } | |||